make port classification configurable

docs/source/manual/usermanual.rst

@@ -420,10 +420,10 @@ You can currently configure the ``max-age`` before HSTS headers will be consider
         "config": {"max-age": "4153600"}
     }
 
-Aggregate findings
-------------------
+Port classification
+-------------------
 
-Setting this to ``True`` will aggregate all findings of the same type into one finding,
+Setting aggregate_findings to ``True`` will aggregate all findings of the same type into one finding,
 resulting in cleaner finding reports (both in the web UI and in PDF's). For example, ``KAT-UNCOMMON-OPEN-PORT``
 will be aggregated into one finding, instead of one separate finding per port.
 
@@ -435,3 +435,15 @@ will be aggregated into one finding, instead of one separate finding per port.
         "bit-id": "port-classification-ip",
         "config": {"aggregate_findings": "True"}
     }
+
+Also you can configure which open ports should create findings and which ports should not. This is done by settings
+common_tcp_ports, common_udp_ports, sa_tcp_ports and/or db_tcp_ports. As an example:
+
+.. code-block:: json
+
+    {
+        "object_type": "Config",
+        "ooi": "Network|internet",
+        "bit-id": "port-classification-ip",
+        "config": {"common_tcp_ports": "1,2,3", "sa_tcp_ports": "4,5,6"}
+    }

octopoes/bits/port_classification_ip/port_classification_ip.py

@@ -37,13 +37,24 @@
 ]
 
 
+def get_ports_from_config(config, config_key, default):
+    ports = config.get(config_key, "")
+    return list(map(int, ports.split(","))) if ports else default
+
+
 def run(input_ooi: IPPort, additional_oois: List, config: Dict[str, str]) -> Iterator[OOI]:
     aggregate_findings = config.get("aggregate_findings", "False").lower() == "true" if config else False
     open_ports = []
+
+    common_tcp_ports = get_ports_from_config(config, "common_tcp_ports", COMMON_TCP_PORTS)
+    common_udp_ports = get_ports_from_config(config, "common_udp_ports", COMMON_UDP_PORTS)
+    sa_tcp_ports = get_ports_from_config(config, "sa_tcp_ports", SA_TCP_PORTS)
+    db_tcp_ports = get_ports_from_config(config, "db_tcp_ports", DB_TCP_PORTS)
+
     for ip_port in additional_oois:
         port = ip_port.port
         protocol = ip_port.protocol
-        if protocol == Protocol.TCP and port in SA_TCP_PORTS:
+        if protocol == Protocol.TCP and port in sa_tcp_ports:
             open_sa_port = KATFindingType(id="KAT-OPEN-SYSADMIN-PORT")
             if aggregate_findings:
                 open_ports.append(ip_port.port)
@@ -54,7 +65,7 @@ def run(input_ooi: IPPort, additional_oois: List, config: Dict[str, str]) -> Ite
                     ooi=ip_port.reference,
                     description=f"Port {port}/{protocol.value} is a system administrator port and should not be open.",
                 )
-        elif protocol == Protocol.TCP and port in DB_TCP_PORTS:
+        elif protocol == Protocol.TCP and port in db_tcp_ports:
             ft = KATFindingType(id="KAT-OPEN-DATABASE-PORT")
             if aggregate_findings:
                 open_ports.append(ip_port.port)
@@ -65,8 +76,8 @@ def run(input_ooi: IPPort, additional_oois: List, config: Dict[str, str]) -> Ite
                     ooi=ip_port.reference,
                     description=f"Port {port}/{protocol.value} is a database port and should not be open.",
                 )
-        elif (protocol == Protocol.TCP and port not in COMMON_TCP_PORTS) or (
-            protocol == Protocol.UDP and port not in COMMON_UDP_PORTS
+        elif (protocol == Protocol.TCP and port not in common_tcp_ports) or (
+            protocol == Protocol.UDP and port not in common_udp_ports
         ):
             kat = KATFindingType(id="KAT-UNCOMMON-OPEN-PORT")
             if aggregate_findings:

octopoes/tests/test_bit_ports.py

@@ -54,6 +54,17 @@ def test_port_classification_tcp_12345():
     assert finding.description == "Port 12345/tcp is not a common port and should possibly not be open."
 
 
+def test_port_classification_tcp_3306_with_config():
+    address = IPAddressV4(address="8.8.8.8", network="fake")
+    port = IPPort(address=address.reference, protocol="tcp", port=3306)
+    results = list(run_port_classification(address, [port], {"db_tcp_ports": "1234"}))
+
+    assert len(results) == 2
+    finding = results[-1]
+    assert isinstance(finding, Finding)
+    assert finding.description == "Port 3306/tcp is not a common port and should possibly not be open."
+
+
 def test_port_classification_udp_80():
     address = IPAddressV4(address="8.8.8.8", network="fake")
     port = IPPort(address=address.reference, protocol="udp", port=80)