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,54 @@ 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. Common TCP ports are ports that will never trigger a finding. A good example is 443. Same counts for common udp ports.
+SA (system administrator) ports will trigger a medium finding that a system administrator port is open, for example, port 22 is usually is SA port. Lastly, DB (database) ports trigger a more severe finding when a database port is open. As an of the configuration 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"}
+    }
+
+Defaults are:
+
+.. code-block:: python
+
+    COMMON_TCP_PORTS = [
+        25,  # SMTP
+        53,  # DNS
+        80,  # HTTP
+        110,  # POP3
+        143,  # IMAP
+        443,  # HTTPS
+        465,  # SMTPS
+        587,  # SMTP (message submmission)
+        993,  # IMAPS
+        995,  # POP3S
+    ]
+
+    COMMON_UDP_PORTS = [
+        53,  # DNS
+    ]
+
+    SA_TCP_PORTS = [
+        21,  # FTP
+        22,  # SSH
+        23,  # Telnet
+        3389,  # Remote Desktop
+        5900,  # VNC
+    ]
+    DB_TCP_PORTS = [
+        1433,  # MS SQL Server
+        1434,  # MS SQL Server
+        3050,  # Interbase/Firebase
+        3306,  # MySQL
+        5432,  # PostgreSQL
+    ]
+
+You can set the ports of SA and DB to an empty string to disable the check.

octopoes/bits/port_classification_ip/port_classification_ip.py

@@ -37,13 +37,26 @@
 ]
 
 
+def get_ports_from_config(config, config_key, default):
+    ports = config.get(config_key, None)
+    if ports is None:
+        return default
+    return list(map(int, ports.split(","))) if ports else []
+
+
 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 +67,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 +78,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)