Create new transformer hieharchy

transformers/action_mapper.py

@@ -0,0 +1,45 @@
+"""Action mapping transformer.
+
+This module defines a transformer responsible for mapping action values
+between vendor-specific representations and the universal data model.
+"""
+
+from typing import Any
+from typing import Dict
+
+from .base_transformer import BaseTransformer
+
+
+class ActionMapper(BaseTransformer):
+    """Map action values between vendor and universal models.
+
+    This transformer replaces the ``action`` field of an item using a
+    predefined mapping dictionary. If the action is not found in the
+    mapping, it is left unchanged.
+    """
+
+    def __init__(self, action_map: Dict[str, str]) -> None:
+        """Initialize the ActionMapper.
+
+        Args:
+            action_map: A dictionary mapping source action values to
+                destination action values.
+        """
+        self.action_map = action_map
+
+    def transform(self, item: Dict[str, Any]) -> Dict[str, Any]:
+        """Transform an item's action field using the action mapping.
+
+        Args:
+            item: A dictionary representing a single rule or configuration
+                entry containing an ``action`` field.
+
+        Returns:
+            The transformed item with its ``action`` field mapped according
+            to the configured action map.
+        """
+        action = item.get("action")
+        if action in self.action_map:
+            item["action"] = self.action_map[action]
+
+        return item

transformers/base_transformer.py

@@ -0,0 +1,32 @@
+"""Base transformer definition.
+
+This module defines the abstract base class used by all transformers
+in the transformation pipeline.
+"""
+
+from abc import ABC
+from abc import abstractmethod
+from typing import Any
+from typing import Dict
+
+
+class BaseTransformer(ABC):
+    """Define the interface for all transformers.
+
+    All concrete transformers must implement the ``transform`` method,
+    which takes a single dictionary item and returns a transformed
+    dictionary.
+    """
+
+    @abstractmethod
+    def transform(self, item: Dict[str, Any]) -> Dict[str, Any]:
+        """Transform a single dictionary item.
+
+        Args:
+            item: A dictionary representing a single configuration or
+                URL entry.
+
+        Returns:
+            A transformed dictionary.
+        """
+        raise NotImplementedError

transformers/category_mapper.py

@@ -0,0 +1,46 @@
+"""Category mapping transformer.
+
+This module defines a transformer responsible for mapping category
+identifiers between vendor-specific representations and the universal
+data model.
+"""
+
+from typing import Any
+from typing import Dict
+
+from .base_transformer import BaseTransformer
+
+
+class CategoryMapper(BaseTransformer):   
+    """Map category identifiers between vendor and universal models.
+
+    This transformer replaces the ``category_id`` field of an item using
+    a predefined mapping dictionary. If the category is not found in the
+    mapping, it is left unchanged.
+    """
+
+    def __init__(self, category_map: Dict[str, str]) -> None:  
+        """Initialize the CategoryMapper.
+
+        Args:
+            category_map: A dictionary mapping source category identifiers
+                to destination category identifiers.
+        """
+        self.category_map = category_map
+
+    def transform(self, item: Dict[str, Any]) -> Dict[str, Any]: 
+        """Transform an item's category identifier using the category map.
+
+        Args:
+            item: A dictionary representing a single rule or configuration
+                entry containing a ``category_id`` field.
+
+        Returns:
+            The transformed item with its ``category_id`` field mapped
+            according to the configured category map.
+        """
+        category_id = item.get("category_id")
+        if category_id in self.category_map:
+            item["category_id"] = self.category_map[category_id]
+
+        return item

transformers/metadata_enricher.py

@@ -0,0 +1,45 @@
+"""Metadata enrichment transformer.
+
+This module defines a transformer that adds vendor information and
+metadata timestamps to each item in the transformation pipeline.
+"""
+
+from datetime import datetime
+from typing import Any
+from typing import Dict
+
+from .base_transformer import BaseTransformer
+
+
+class MetadataEnricher(BaseTransformer):
+    """Enrich items with vendor and metadata information.
+
+    This transformer adds a ``vendor`` field and a ``metadata`` dictionary
+    containing a ``processed_at`` timestamp to each item.
+    """
+
+    def __init__(self, vendor: str) -> None:        
+        """Initialize the MetadataEnricher.
+
+        Args:
+            vendor: The vendor name to attach to each item.
+        """      
+        self.vendor = vendor
+
+    def transform(self, item: Dict[str, Any]) -> Dict[str, Any]:        
+        """Add vendor and metadata information to an item.
+
+        Args:
+            item: A dictionary representing a single configuration or URL entry.
+
+        Returns:
+            The transformed dictionary containing the ``vendor`` field and
+            a ``metadata.processed_at`` timestamp.
+        """        
+        item["vendor"] = self.vendor
+        if "metadata" not in item:
+            item["metadata"] = {}
+
+        item["metadata"]["processed_at"] = datetime.utcnow().isoformat()
+
+        return item

transformers/pattern_normalizer.py

@@ -0,0 +1,30 @@
+"""Pattern normalization transformer.
+
+This module defines a generic pattern normalizer that ensures each item
+has a ``pattern`` field. Currently, this transformer acts as a pass-through.
+"""
+
+from typing import Any
+from typing import Dict
+
+from .base_transformer import BaseTransformer
+
+
+class PatternNormalizer(BaseTransformer):    
+    """Normalize or enforce the presence of a pattern field in items.
+
+    This transformer guarantees that each dictionary item contains a
+    ``pattern`` key. If the key is missing, it is initialized to an empty string.
+    """
+
+    def transform(self, item: Dict[str, Any]) -> Dict[str, Any]:        
+        """Ensure the item has a pattern field.
+
+        Args:
+            item: A dictionary representing a single configuration or URL entry.
+
+        Returns:
+            The same dictionary with a ``pattern`` key ensured.
+        """       
+        item["pattern"] = item.get("pattern", "")
+        return item

transformers/pipelines.py

@@ -0,0 +1,39 @@
+"""Pipeline helper functions for URL transformations.
+
+This module provides utilities to apply a sequence of transformers
+to vendor configuration items, producing universal model dictionaries.
+"""
+
+from typing import Any
+from typing import Dict
+from typing import List
+
+from transformers.base_transformer import BaseTransformer
+
+
+def apply_transformers(
+    items: List[Dict[str, Any]],
+    transformers: List[BaseTransformer]
+) -> List[Dict[str, Any]]:   
+    """Apply a sequence of transformers to a list of items.
+
+    Each item in the input list is processed sequentially by all
+    transformers in the given order.
+
+    Args:
+        items: A list of dictionaries representing vendor configuration
+            entries.
+        transformers: An ordered list of transformer instances that
+            implement the `transform` method.
+
+    Returns:
+        A list of transformed dictionaries.
+    """    
+    result: List[Dict[str, Any]] = []
+
+    for item in items:
+        for transformer in transformers:
+            item = transformer.transform(item)
+        result.append(item)
+
+    return result

transformers/type_mapper.py

@@ -0,0 +1,46 @@
+"""Type mapping transformer.
+
+This module defines a transformer that maps item types (e.g., literal,
+wildcard, regex, substring) between vendor-specific representations
+and the universal data model.
+"""
+
+from typing import Any
+from typing import Dict
+
+from .base_transformer import BaseTransformer
+
+
+class TypeMapper(BaseTransformer):    
+    """Map type values between vendor and universal models.
+
+    This transformer replaces the ``type`` field of an item using a
+    predefined mapping dictionary. If the type is not found in the
+    mapping, it is left unchanged.
+    """
+
+    def __init__(self, type_map: Dict[str, str]) -> None:        
+        """Initialize the TypeMapper.
+
+        Args:
+            type_map: A dictionary mapping source type values to
+                destination type values.
+        """    
+        self.type_map = type_map
+
+    def transform(self, item: Dict[str, Any]) -> Dict[str, Any]:      
+        """Transform an item's type field using the type mapping.
+
+        Args:
+            item: A dictionary representing a single rule or configuration
+                entry containing a ``type`` field.
+
+        Returns:
+            The transformed item with its ``type`` field mapped according
+            to the configured type map.
+        """     
+        item_type = item.get("type")
+        if item_type in self.type_map:
+            item["type"] = self.type_map[item_type]
+
+        return item

transformers/vendors/fortinet_transformer.py

@@ -0,0 +1,54 @@
+"""Fortinet transformation pipelines.
+
+This module defines action, category, and type mappings for Fortinet
+URL filtering configurations, as well as transformation pipelines
+to convert between Fortinet-specific and universal data models.
+"""
+
+from transformers.action_mapper import ActionMapper
+from transformers.base_transformer import BaseTransformer
+from transformers.category_mapper import CategoryMapper
+from transformers.metadata_enricher import MetadataEnricher
+from transformers.pattern_normalizer import PatternNormalizer
+from transformers.type_mapper import TypeMapper
+
+# ---------------- FORTINET MAPPINGS ----------------
+
+FORTINET_ACTION_MAP = {
+    "block": "block",
+    "allow": "allow",
+    "monitor": "monitor",
+}
+
+FORTINET_CATEGORY_MAP = {
+    "3": "malware",
+    "4": "phishing",
+    "5": "gambling",
+    "default": "uncategorized",
+}
+
+FORTINET_TYPE_MAP = {
+    "simple": "literal",
+    "wildcard": "wildcard",
+    "regex": "regex",
+    "substring": "substring",
+}
+
+
+# ---------------- FORTINET PIPELINES ----------------
+
+VENDOR_TO_UNIVERSAL_PIPELINES = [
+    ActionMapper(FORTINET_ACTION_MAP),
+    PatternNormalizer(),
+    TypeMapper(FORTINET_TYPE_MAP),
+    CategoryMapper(FORTINET_CATEGORY_MAP),
+    MetadataEnricher("fortinet"),
+]
+
+UNIVERSAL_TO_VENDOR_PIPELINES = [
+    ActionMapper({value: key for key, value in FORTINET_ACTION_MAP.items()}),
+    PatternNormalizer(),
+    TypeMapper({value: key for key, value in FORTINET_TYPE_MAP.items()}),
+    CategoryMapper({value: key for key, value in FORTINET_CATEGORY_MAP.items()}),
+    MetadataEnricher("fortinet"),
+]