Make dns and snmpv3 authorized hosts region-based (#12)

Author: mamullen13316

audit_settings.yaml.example

@@ -329,18 +329,26 @@ login_security:
   DefaultConfigurationLanguage: English
 
 # DNS Servers (Configure > Network > DNS)
+# Nest under the region. Region should be defined in inventory
 dns_servers:
-  - 4.2.2.1
-  - 4.2.2.2
+  na:
+    - 4.2.2.1
+    - 4.2.2.2
+  emea:
+    - 1.1.1.1
 
 # SNMPv3 (System > Administration > SNMP)
 snmpv3:
   Username: snmpv3_user
   AcceptQueries: Enable
   SendTraps: Enable
-  AuthorizedHosts: 
-    - 10.1.100.101
-    - 10.1.100.102
+  AuthorizedHosts:
+    na:  
+      - 10.1.100.101
+      - 10.1.100.102
+    emea:
+      - 10.2.100.101
+      - 10.2.100.102
 
 # Time Settings (System > Administration > Time)
 time:

firewalls.yaml.example

@@ -1,2 +1,3 @@
 - hostname: example-host.network.sophos.com
   port: 4444
+  region: na

nautobot_query/all_devices_query.gql

@@ -1,5 +1,12 @@
 query {
   devices (manufacturer: "Sophos", tags: "SFOS", tags__n: "Auxillary", status: "Active") {
     name
+    location {
+      parent {
+        parent {
+          name
+        }
+      }
+    }
   }
 }

nautobot_query/location_query.gql

@@ -1,5 +1,12 @@
 query {
   devices(location: ["Abingdon", "Vancouver"], manufacturer:"Sophos", tags: "SFOS", tags__n:"Auxillary", status:"Active") {
     name
+    location {
+      parent {
+        parent {
+          name
+        }
+      }
+    }
   }
 }

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "sophos-firewall-audit"
-version = "1.0.15"
+version = "1.0.16"
 description = "Sophos Firewall Audit"
 authors = ["Matt Mullen <matt.mullen@sophos.com>"]
 readme = "README.md"

sophos_firewall_audit/rules/dnsservers.py

@@ -27,7 +27,7 @@ def eval_dns_servers(fw_obj: SophosFirewall,
         dict: Audit results and output table(s)
     """
 
-    expected_servers = sorted(settings["dns_servers"])
+    expected_servers = sorted(settings["dns_servers"].get(fw_obj.region))
 
     for i in range(1,3):
         try:

sophos_firewall_audit/rules/snmpv3.py

@@ -30,7 +30,7 @@ def eval_snmpv3(fw_obj: SophosFirewall,
 
     for i in range(1,3):
         try:
-            result = fw_obj.get_tag_with_filter(xml_tag="SNMPv3User", key="Username", value=expected["Username"])
+            result = fw_obj.get_tag_with_filter(xml_tag="SNMPv3User", key="Username", value=expected["Username"], operator="=")
         except SophosFirewallZeroRecords:
             result = None
             break
@@ -61,21 +61,25 @@ def eval_snmpv3(fw_obj: SophosFirewall,
 
     # Changes for v22
     if result:
-        if result["Response"]["@APIVersion"][:2] >= "22":
+        if int(result["Response"]["@APIVersion"][:2]) >= 22:
             # Remove Name key since it did not exist pre-v22
             actual.pop("Name")
-            # Rename "AuthorizedHosts" to "AuthorizedHostsIpv4" in expected
-            expected["AuthorizedHostsIpv4"] = expected.pop("AuthorizedHosts")
+            # Rename  "AuthorizedHostsIpv4" to "AuthorizedHosts" to in actual
+            actual["AuthorizedHosts"] = actual.pop("AuthorizedHostsIpv4")
             # Convert SendTraps and AcceptQueries from "Enabled/Disabled" to True/False
             expected["SendTraps"] = "true" if expected["SendTraps"] == "Enabled" else "false"
             expected["AcceptQueries"] = "true" if expected["AcceptQueries"] == "Enabled" else "false"
 
-    
     output = []
     for key in expected:
         status = "AUDIT_PASS"
         if key in actual:
-            if not expected[key] == actual[key]:
+            if key == "AuthorizedHosts":
+                if not expected[key].get(fw_obj.region) == actual[key]:
+                    status = "AUDIT_FAIL"
+                    result_dict["audit_result"] = "FAIL"
+                    result_dict["fail_ct"] += 1
+            elif not expected[key] == actual[key]:
                 status = "AUDIT_FAIL"
                 result_dict["audit_result"] = "FAIL"
                 result_dict["fail_ct"] += 1
@@ -88,7 +92,7 @@ def eval_snmpv3(fw_obj: SophosFirewall,
             result_dict["fail_ct"] += 1
 
         if key == "AuthorizedHosts" and not actual.get(key) == "None" and status == "AUDIT_FAIL":
-            actual_output = '\n'.join(format_diff(unified_diff(sorted(expected[key]), sorted(actual.get(key)), n=1000000000)))
+            actual_output = '\n'.join(format_diff(unified_diff(sorted(expected[key].get(fw_obj.region)), sorted(actual.get(key)), n=1000000000)))
         elif key == "AuthorizedHosts" and not actual.get(key) == "None" and status == "AUDIT_PASS":
             actual_output = '\n'.join(actual.get(key))
         else:
@@ -98,7 +102,7 @@ def eval_snmpv3(fw_obj: SophosFirewall,
                 "SNMPv3",
                 "System > Administration > SNMP",
                 key,
-                '\n'.join(expected[key]) if key == "AuthorizedHosts" else expected[key],
+                '\n'.join(expected[key].get(fw_obj.region)) if key == "AuthorizedHosts" else expected[key],
                 actual_output,
                 html_status(status)
             ])

sophos_firewall_audit/sophosfirewallaudit.py

@@ -58,7 +58,11 @@ def nb_graphql_query(query, verify):
     token = os.environ.get("NAUTOBOT_TOKEN")
     nautobot = api(url=url, token=token, verify=verify)
     graphql_response = nautobot.graphql.query(query=query)
-    return [{"hostname": firewall["name"], "port": "4444"} for firewall in graphql_response.json["data"]["devices"]]
+    return [{"hostname": firewall["name"], "port": "4444",
+             "region": firewall["location"]["parent"]["parent"]["name"].lower()} 
+               for firewall in graphql_response.json["data"]["devices"]
+               if firewall["location"]["parent"].get("parent")
+               ]
 
 def device_query(environ, devices):
     """Generate GraphQL query
@@ -182,6 +186,7 @@ def main():
             port=firewall['port'],
             verify=False
         )
+        fw.region = firewall.get("region")
         try:
             fw.login()
         except Exception as Error: