feat: Checkpoint & Forti NAT support

roles/common/files/fwo-api-calls/report/getManagementForLatestNormalizedConfig.graphql

@@ -123,6 +123,8 @@ fragment ruleFragment on rule {
   rule_custom_fields
   rule_implied
   nat_rule
+  access_rule
+  xlate_rule
   uiuser {
     uiuser_username
   }
@@ -133,6 +135,9 @@ fragment ruleFragment on rule {
     rule_last_hit
   }
   rule_comment
+  ruleByXlateRule {
+    rule_uid
+  }
   section_header: rule_head_text
 }
 

roles/common/files/fwo-api-calls/report/getManagementForNormalizedConfig.graphql

@@ -178,6 +178,8 @@ fragment ruleFragment on rule {
   rule_custom_fields
   rule_implied
   nat_rule
+  access_rule
+  xlate_rule
   uiuser {
     uiuser_username
   }
@@ -188,6 +190,9 @@ fragment ruleFragment on rule {
     rule_last_hit
   }
   rule_comment
+  ruleByXlateRule {
+    rule_uid
+  }
   section_header: rule_head_text
 }
 

roles/database/files/sql/creation/fworch-fill-stm.sql

@@ -580,6 +580,7 @@ insert into stm_link_type (id, name) VALUES (2, 'ordered');
 insert into stm_link_type (id, name) VALUES (3, 'inline');
 insert into stm_link_type (id, name) VALUES (4, 'concatenated');
 insert into stm_link_type (id, name) VALUES (5, 'domain');
+insert into stm_link_type (id, name) VALUES (6, 'nat');
 
 -- insert into compliance.assessability_issue_type (type_id, type_name) VALUES (1, 'empty group');
 -- insert into compliance.assessability_issue_type (type_id, type_name) VALUES (2, 'broadcast address');

roles/database/files/upgrade/9.1.11.sql

@@ -0,0 +1 @@
+insert into stm_link_type (id, name) VALUES (6, 'nat') ON CONFLICT DO NOTHING;

roles/importer/files/importer/fw_modules/checkpointR8x/cp_getter.py

@@ -348,6 +348,7 @@ def get_rulebases(
     native_config_domain: dict[str, Any] | None,
     device_config: dict[str, Any] | None,
     policy_rulebases_uid_list: list[str],
+    policy_structure: dict[str, Any],
     is_global: bool = False,
     access_type: str = "access",
     rulebase_uid: str | None = None,
@@ -407,6 +408,7 @@ def get_rulebases(
         show_params_rules,
         is_global,
         policy_rulebases_uid_list,
+        policy_structure=policy_structure,
     )
 
 
@@ -530,6 +532,7 @@ def get_inline_layers_recursively(
     show_params_rules: dict[str, Any],
     is_global: bool,
     policy_rulebases_uid_list: list[str],
+    policy_structure: dict[str, Any],
 ) -> list[str]:
     """
     Takes current_rulebase, splits sections into sub-rulebases and searches for layerguards to fetch
@@ -569,6 +572,7 @@ def get_inline_layers_recursively(
                             is_global=is_global,
                             access_type="access",
                             rulebase_uid=rule["inline-layer"],
+                            policy_structure=policy_structure,
                         )
 
     return policy_rulebases_uid_list
@@ -978,3 +982,30 @@ def get_object_details_from_api(uid_missing_obj: str, sid: str = "", apiurl: str
         return obj
     FWOLogger.warning(f"missing nw obj of unexpected type '{obj_type}': {uid_missing_obj}")
     return {}
+
+
+def get_gateways_and_servers(sid: str = "", apiurl: str = "") -> list[dict[str, Any]]:
+    """Fetch gateways and servers from the API."""
+    current = 0
+    total = 1
+
+    gateways_and_servers: list[dict[str, Any]] = []
+
+    while current < total:
+        try:
+            result = cp_api_call(apiurl, "show-gateways-and-servers", {"details-level": "full", "offset": current}, sid)
+        except Exception as e:
+            raise FwoImporterError(f"error while trying to get gateways and servers: {e}")
+
+        if result is None or "objects" not in result:
+            raise FwoImporterError("no objects received while trying to get gateways and servers")
+
+        gateways_and_servers.extend(result["objects"])
+
+        if "total" not in result or "to" not in result:
+            raise FwoImporterError("result does not contain total or to field while trying to get gateways and servers")
+
+        total = result["total"]
+        current = result["to"]
+
+    return gateways_and_servers

roles/importer/files/importer/fw_modules/checkpointR8x/cp_nat.py

@@ -0,0 +1,264 @@
+from typing import Any
+
+from fw_modules.checkpointR8x.cp_rule import parse_single_rule
+from fwo_log import FWOLogger
+from models.import_state import ImportState
+from models.rulebase import Rulebase
+
+
+def normalize_nat_rules(
+    native_config: dict[str, Any],
+    import_state: ImportState,
+    normalized_config: dict[str, Any],
+):
+    native_nat_rulebases = native_config.get("nat_rulebases", [])
+    if not native_nat_rulebases:
+        return
+
+    for gateway in native_config["gateways"]:
+        parse_native_nat_rulebases(gateway, native_nat_rulebases, import_state, normalized_config, native_config)
+
+
+def get_initial_nat_rulebase_link(gateway: dict[str, Any], normalized_config: dict[str, Any]) -> dict[str, Any] | None:
+    normalized_gateway = next((gw for gw in normalized_config["gateways"] if gw["Uid"] == gateway["uid"]), None)
+
+    if normalized_gateway is None:
+        FWOLogger.warning("Could not find normalized gateway for initial NAT rulebase link: " + str(gateway["uid"]))
+        return None
+
+    initial_gateway_link = next(
+        (
+            link
+            for link in normalized_gateway["RulebaseLinks"]
+            if link.get("is_initial") and link.get("link_type") == "ordered"
+        ),
+        None,
+    )
+
+    if initial_gateway_link is None:
+        FWOLogger.warning("Could not find initial gateway rulebase link for NAT rulebase: " + str(gateway["uid"]))
+        return None
+
+    return initial_gateway_link
+
+
+def parse_native_nat_rulebases(
+    gateway: dict[str, Any],
+    native_nat_rulebases: list[dict[str, Any]],
+    import_state: ImportState,
+    normalized_config: dict[str, Any],
+    native_config: dict[str, Any],
+):
+    for nat_rulebase in native_nat_rulebases:
+        if "nat_rule_chunks" not in nat_rulebase:
+            continue
+
+        normalized_nat_rulebase = insert_parent_nat_rulebase(gateway, import_state, normalized_config)
+        normalized_gateway = next((gw for gw in normalized_config["gateways"] if gw["Uid"] == gateway["uid"]), None)
+
+        if normalized_gateway is None:
+            FWOLogger.warning("Could not find normalized gateway for NAT rulebase, skipping: " + str(gateway["uid"]))
+            continue
+
+        initial_gateway_link = get_initial_nat_rulebase_link(gateway, normalized_config)
+
+        if initial_gateway_link is None:
+            continue
+
+        initial_to_rulebase_uid = initial_gateway_link.get("to_rulebase_uid")
+        if not initial_to_rulebase_uid:
+            FWOLogger.warning(
+                "Initial gateway rulebase link is missing to_rulebase_uid for NAT rulebase, skipping: "
+                + str(gateway["uid"])
+            )
+            continue
+
+        insert_rulebase_link(
+            from_rulebase_uid=initial_to_rulebase_uid,
+            to_rulebase_uid=normalized_nat_rulebase.uid,
+            link_type="nat",
+            normalized_gateway=normalized_gateway,
+        )
+
+        for chunk in nat_rulebase["nat_rule_chunks"]:
+            parse_nat_rule_chunk(
+                chunk,
+                normalized_nat_rulebase,
+                gateway,
+                native_config,
+                import_state,
+                normalized_config,
+                normalized_gateway,
+            )
+
+
+def insert_parent_nat_rulebase(
+    gateway: dict[str, Any],
+    import_state: ImportState,
+    normalized_config: dict[str, Any],
+) -> Rulebase:
+    nat_rulebase_uid = "nat-rulebase-" + gateway["uid"]
+    existing_nat_rulebase = next((rb for rb in normalized_config["policies"] if rb.uid == nat_rulebase_uid), None)
+
+    if existing_nat_rulebase is not None:
+        return existing_nat_rulebase
+
+    normalized_nat_rulebase = Rulebase(
+        uid=nat_rulebase_uid,
+        mgm_uid=import_state.mgm_details.uid,
+        name="NAT",
+        rules={},
+    )
+
+    normalized_config["policies"].append(normalized_nat_rulebase)
+
+    return normalized_nat_rulebase
+
+
+def insert_rulebase_link(
+    from_rulebase_uid: str,
+    to_rulebase_uid: str,
+    link_type: str,
+    normalized_gateway: dict[str, Any],
+) -> None:
+    if not any(
+        link
+        for link in normalized_gateway["RulebaseLinks"]
+        if link["to_rulebase_uid"] == to_rulebase_uid
+        and link["link_type"] == link_type
+        and link["from_rulebase_uid"] == from_rulebase_uid
+    ):
+        normalized_gateway["RulebaseLinks"].append(
+            {
+                "from_rulebase_uid": from_rulebase_uid,
+                "to_rulebase_uid": to_rulebase_uid,
+                "link_type": link_type,
+                "is_initial": False,
+                "is_global": False,
+                "is_section": False,
+            }
+        )
+
+
+def parse_nat_rulebase(
+    src_rulebase: dict[str, Any],
+    normalized_nat_rulebase: Rulebase,
+    gateway: dict[str, Any],
+    native_config: dict[str, Any],
+    import_state: ImportState,
+    normalized_config: dict[str, Any],
+    normalized_gateway: dict[str, Any],
+):
+    section_rulebase = Rulebase(
+        uid=src_rulebase["uid"],
+        mgm_uid=import_state.mgm_details.uid,
+        name=src_rulebase["name"],
+        rules={},
+    )
+
+    if not any(rb for rb in normalized_config["policies"] if rb.uid == section_rulebase.uid):
+        normalized_config["policies"].append(section_rulebase)
+
+    insert_rulebase_link(
+        from_rulebase_uid=normalized_nat_rulebase.uid,
+        to_rulebase_uid=section_rulebase.uid,
+        link_type="nat",
+        normalized_gateway=normalized_gateway,
+    )
+
+    for rule in src_rulebase["rulebase"]:
+        parse_nat_rule(rule, section_rulebase, gateway, native_config)
+
+
+def parse_nat_rule(
+    src_rulebase: dict[str, Any],
+    rulebase: Rulebase,
+    gateway: dict[str, Any],
+    native_config: dict[str, Any],
+):
+    (rule_match, rule_xlate) = parse_nat_rule_transform(src_rulebase)
+    parse_single_rule(
+        rule_match,
+        rulebase,
+        rulebase.name,
+        rulebase.uid,
+        gateway,
+        native_config["policies"],
+    )
+    parse_single_rule(  # do not increase rule_num here (xlate rules do not count)
+        rule_xlate,
+        rulebase,
+        rulebase.name,
+        rulebase.uid,
+        gateway,
+        native_config["policies"],
+    )
+
+
+def parse_nat_rule_chunk(
+    chunk: dict[str, Any],
+    normalized_nat_rulebase: Rulebase,
+    gateway: dict[str, Any],
+    native_config: dict[str, Any],
+    import_state: ImportState,
+    normalized_config: dict[str, Any],
+    normalized_gateway: dict[str, Any],
+):
+    if "rulebase" not in chunk:
+        return
+
+    for src_rulebase in chunk["rulebase"]:
+        if "rulebase" in src_rulebase:
+            parse_nat_rulebase(
+                src_rulebase,
+                normalized_nat_rulebase,
+                gateway,
+                native_config,
+                import_state,
+                normalized_config,
+                normalized_gateway,
+            )
+        if "rule-number" in src_rulebase:  # rulebase is just a single rule (xlate rules do not count)
+            parse_nat_rule(src_rulebase, normalized_nat_rulebase, gateway, native_config)
+
+
+def parse_nat_rule_transform(nat_rule: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
+    nat_in_rule = {
+        "uid": nat_rule["uid"],
+        "source": [nat_rule["original-source"]],
+        "destination": [nat_rule["original-destination"]],
+        "service": [nat_rule["original-service"]],
+        "action": [{"name": "accept", "type": "nat-action", "uid": nat_rule["uid"] + "_original-action"}],
+        "track": [{"type": "nat", "name": "None", "uid": nat_rule["uid"]}],
+        "type": "nat",
+        "rule-number": 0,
+        "source-negate": False,
+        "destination-negate": False,
+        "service-negate": False,
+        "install-on": nat_rule["install-on"],
+        "time": nat_rule.get("time", "time"),
+        "enabled": nat_rule["enabled"],
+        "comments": nat_rule["comments"],
+        "nat_rule": True,
+        "xlate_rule_uid": nat_rule["uid"] + "_translated",
+        "access_rule": False,
+    }
+    nat_out_rule = {
+        "uid": nat_rule["uid"] + "_translated",
+        "source": [nat_rule["translated-source"]],
+        "destination": [nat_rule["translated-destination"]],
+        "service": [nat_rule["translated-service"]],
+        "action": [{"name": "accept", "type": "nat-action", "uid": nat_rule["uid"] + "_translated-action"}],
+        "track": [{"type": "nat", "name": "None", "uid": nat_rule["uid"] + "_translated"}],
+        "type": "nat",
+        "rule-number": 0,
+        "enabled": True,
+        "source-negate": False,
+        "destination-negate": False,
+        "service-negate": False,
+        "install-on": nat_rule["install-on"],
+        "time": "time",
+        "nat_rule": True,
+        "access_rule": False,
+    }
+    return (nat_in_rule, nat_out_rule)