feat: ✨ add Shodan InternetDB boefje (#2615)

Co-authored-by: Jan Klopper <[email protected]>
Co-authored-by: originalsouth <[email protected]>
Co-authored-by: Ammar <[email protected]>

boefjes/boefjes/plugins/helpers.py

@@ -66,3 +66,12 @@ def get_file_from_container(container: docker.models.containers.Container, path:
         return None
 
     return extracted_file.read()
+
+
+def cpe_to_name_version(cpe: str) -> tuple[str | None, str | None]:
+    """Fetch the software name and version from a CPE string."""
+    cpe_split = cpe.split(":")
+    cpe_split_len = len(cpe_split)
+    name = None if cpe_split_len < 4 else cpe_split[3]
+    version = None if cpe_split_len < 5 else cpe_split[4]
+    return name, version

boefjes/boefjes/plugins/kat_binaryedge/http_web/normalize.py

@@ -3,7 +3,7 @@
 from collections.abc import Iterable
 
 from boefjes.job_models import NormalizerOutput
-from boefjes.plugins.kat_binaryedge.services.normalize import get_name_from_cpe
+from boefjes.plugins.helpers import cpe_to_name_version
 from octopoes.models import Reference
 from octopoes.models.ooi.network import IPAddressV4, IPAddressV6, IPPort, Network, PortState, Protocol
 from octopoes.models.ooi.software import Software, SoftwareInstance
@@ -50,7 +50,8 @@ def run(input_ooi: dict, raw: bytes) -> Iterable[NormalizerOutput]:
 
             for app in response.get("apps", {}):
                 if "cpe" in app:
-                    software_ooi = Software(name=get_name_from_cpe(app["cpe"]), cpe=app["cpe"])
+                    name, version = cpe_to_name_version(cpe=app["cpe"])
+                    software_ooi = Software(name=name, version=version, cpe=app["cpe"])
                     yield software_ooi
                     yield SoftwareInstance(ooi=ip_port_ooi.reference, software=software_ooi.reference)
                 else:
@@ -74,9 +75,8 @@ def run(input_ooi: dict, raw: bytes) -> Iterable[NormalizerOutput]:
             for potential_software in data:
                 # Check all values for 'cpe'
                 if isinstance(potential_software, dict) and "cpe" in potential_software:
-                    software_ooi = Software(
-                        name=get_name_from_cpe(potential_software["cpe"]), cpe=potential_software["cpe"]
-                    )
+                    name, version = cpe_to_name_version(cpe=potential_software["cpe"])
+                    software_ooi = Software(name=name, version=version, cpe=potential_software["cpe"])
                     yield software_ooi
                     yield SoftwareInstance(ooi=ip_port_ooi.reference, software=software_ooi.reference)
 

boefjes/boefjes/plugins/kat_binaryedge/service_identification/normalize.py

@@ -3,7 +3,7 @@
 from collections.abc import Iterable
 
 from boefjes.job_models import NormalizerOutput
-from boefjes.plugins.kat_binaryedge.services.normalize import get_name_from_cpe
+from boefjes.plugins.helpers import cpe_to_name_version
 from octopoes.models import Reference
 from octopoes.models.ooi.findings import Finding, KATFindingType
 from octopoes.models.ooi.network import IPAddressV4, IPAddressV6, IPPort, Network, PortState, Protocol
@@ -51,7 +51,8 @@ def run(input_ooi: dict, raw: bytes) -> Iterable[NormalizerOutput]:
 
             if "cpe" in service:
                 for cpe in service["cpe"]:
-                    software_ooi = Software(name=get_name_from_cpe(cpe), cpe=cpe)
+                    name, version = cpe_to_name_version(cpe=cpe)
+                    software_ooi = Software(name=name, version=version, cpe=cpe)
                     yield software_ooi
                     software_instance_ooi = SoftwareInstance(
                         ooi=ip_service_ooi.reference, software=software_ooi.reference

boefjes/boefjes/plugins/kat_binaryedge/services/normalize.py

@@ -3,24 +3,12 @@
 from collections.abc import Iterable
 
 from boefjes.job_models import NormalizerOutput
+from boefjes.plugins.helpers import cpe_to_name_version
 from octopoes.models import Reference
 from octopoes.models.ooi.network import IPAddressV4, IPAddressV6, IPPort, Network, PortState, Protocol
 from octopoes.models.ooi.software import Software, SoftwareInstance
 
 
-def get_name_from_cpe(cpe: str) -> str:
-    split = []
-    if cpe[0:5] == "cpe:/":
-        split = cpe[5:].split(":")
-    elif cpe[0:8] == "cpe:2.3:":
-        split = cpe[8:].split(":")
-
-    if len(split) > 3:
-        return split[2]
-    else:
-        return cpe
-
-
 def run(input_ooi: dict, raw: bytes) -> Iterable[NormalizerOutput]:
     results = json.loads(raw)
     pk_ooi = Reference.from_str(input_ooi["primary_key"])
@@ -73,7 +61,8 @@ def run(input_ooi: dict, raw: bytes) -> Iterable[NormalizerOutput]:
                 yield software_ooi
                 yield SoftwareInstance(ooi=ip_port_ooi.reference, software=software_ooi.reference)
             for cpe in scan.get("result", {}).get("data", {}).get("cpe", []):
-                software_ooi = Software(name=get_name_from_cpe(cpe), cpe=cpe)
+                name, version = cpe_to_name_version(cpe=cpe)
+                software_ooi = Software(name=name, version=version, cpe=cpe)
                 yield software_ooi
                 yield SoftwareInstance(ooi=ip_port_ooi.reference, software=software_ooi.reference)
 

boefjes/boefjes/plugins/kat_shodan_internetdb/boefje.json

@@ -0,0 +1,10 @@
+{
+  "id": "shodan_internetdb",
+  "name": "'Shodan InternetDB",
+  "description": "Use Shodan InternetDB to find open ports with vulnerabilities that are found on an IP.",
+  "consumes": [
+    "IPAddressV4",
+    "IPAddressV6"
+  ],
+  "scan_level": 1
+}

boefjes/boefjes/plugins/kat_shodan_internetdb/description.md

@@ -0,0 +1,30 @@
+# Shodan InternetDB
+
+Fast IP Lookups for Open Ports and Vulnerabilities. Only free for non-commercial use. The API gets updated once a week.
+
+See: https://internetdb.shodan.io/, https://internetdb.shodan.io/docs
+
+## Return Schema:
+
+```
+{
+  "cpes": [
+    "string"
+  ],
+  "hostnames": [
+    "string"
+  ],
+  "ip": "string",
+  "ports": [
+    0
+  ],
+  "tags": [
+    "string"
+  ],
+  "vulns": [
+    "string"
+  ]
+}
+```
+
+Tags are discarded in the normalizer.

boefjes/boefjes/plugins/kat_shodan_internetdb/main.py

@@ -0,0 +1,18 @@
+from ipaddress import ip_address
+
+import requests
+
+from boefjes.job_models import BoefjeMeta
+
+REQUEST_TIMEOUT = 60
+
+
+def run(boefje_meta: BoefjeMeta) -> list[tuple[set, bytes | str]]:
+    """Make request to InternetDB."""
+    ip = boefje_meta.arguments["input"]["address"]
+    if ip_address(ip).is_private:
+        return [({"info/boefje"}, "Skipping private IP address")]
+    response = requests.get(f"https://internetdb.shodan.io/{ip}", timeout=REQUEST_TIMEOUT)
+    response.raise_for_status()
+
+    return [(set(), response.content)]

boefjes/boefjes/plugins/kat_shodan_internetdb/normalize.py

@@ -0,0 +1,54 @@
+import json
+import logging
+from collections.abc import Iterable
+
+from boefjes.plugins.helpers import cpe_to_name_version
+from octopoes.models import OOI, Reference
+from octopoes.models.ooi.dns.records import DNSPTRRecord
+from octopoes.models.ooi.dns.zone import Hostname
+from octopoes.models.ooi.findings import CVEFindingType, Finding
+from octopoes.models.ooi.network import Network
+from octopoes.models.ooi.software import Software, SoftwareInstance
+
+DNS_PTR_STR = ".in-addr.arpa"
+
+
+def run(input_ooi: dict, raw: bytes) -> Iterable[OOI]:
+    """Normalize InternetDB output."""
+    result = json.loads(raw)
+    input_ooi_reference = Reference.from_str(input_ooi["primary_key"])
+    input_ooi_str = input_ooi["address"]
+
+    if not result:
+        logging.info("No InternetDB results available for normalization.")
+    elif "detail" in result:
+        if result["detail"] == "No information available":
+            logging.info("No information available for IP.")
+        else:
+            logging.warning("Unexpected detail: %s", result["detail"])
+    else:
+        for hostname in result["hostnames"]:
+            hostname_ooi = Hostname(name=hostname, network=Network(name=input_ooi["network"]["name"]).reference)
+            yield hostname_ooi
+            if hostname.endswith(DNS_PTR_STR):
+                yield DNSPTRRecord(hostname=hostname_ooi.reference, value=hostname, address=input_ooi_reference)
+
+        # ruff: noqa: ERA001
+        # for port in result["ports"]:
+        #     yield IPPort(address=input_ooi_reference, port=int(port), state=PortState("open"))
+
+        for cve in result["vulns"]:
+            finding_type = CVEFindingType(id=cve)
+            finding = Finding(
+                finding_type=finding_type.reference,
+                ooi=input_ooi_reference,
+                proof=f"https://internetdb.shodan.io/{input_ooi_str}",
+            )
+            yield finding_type
+            yield finding
+
+        for cpe in result["cpes"]:
+            name, version = cpe_to_name_version(cpe=cpe)
+            software = Software(name=name, version=version, cpe=cpe)
+            yield software
+            yield SoftwareInstance(software=software.reference, ooi=input_ooi_reference)

boefjes/boefjes/plugins/kat_shodan_internetdb/normalizer.json

@@ -0,0 +1,16 @@
+{
+  "id": "kat_shodan_internetdb_normalize",
+  "name": "Shodan InternetDB normalizer",
+  "consumes": [
+    "boefje/shodan_internetdb"
+  ],
+  "produces": [
+    "Finding",
+    "IPPort",
+    "Hostname",
+    "CVEFindingType",
+    "DNSPTRRecord",
+    "Software",
+    "SoftwareInstance"
+  ]
+}

boefjes/tests/integration/test_api.py

@@ -17,9 +17,9 @@ def test_get_local_plugin(test_client, organisation):
 
 def test_filter_plugins(test_client, organisation):
     response = test_client.get(f"/v1/organisations/{organisation.id}/plugins/")
-    assert len(response.json()) == 99
+    assert len(response.json()) == 101
     response = test_client.get(f"/v1/organisations/{organisation.id}/plugins?plugin_type=boefje")
-    assert len(response.json()) == 44
+    assert len(response.json()) == 45
 
     response = test_client.get(f"/v1/organisations/{organisation.id}/plugins?limit=10")
     assert len(response.json()) == 10
@@ -62,7 +62,7 @@ def test_add_boefje(test_client, organisation):
     assert response.status_code == 422
 
     response = test_client.get(f"/v1/organisations/{organisation.id}/plugins/?plugin_type=boefje")
-    assert len(response.json()) == 45
+    assert len(response.json()) == 46
 
     boefje_dict = boefje.model_dump()
     boefje_dict["consumes"] = list(boefje_dict["consumes"])
@@ -119,7 +119,7 @@ def test_add_normalizer(test_client, organisation):
     assert response.status_code == 201
 
     response = test_client.get(f"/v1/organisations/{organisation.id}/plugins/?plugin_type=normalizer")
-    assert len(response.json()) == 56
+    assert len(response.json()) == 57
 
     response = test_client.get(f"/v1/organisations/{organisation.id}/plugins/test_normalizer")
     assert response.json() == normalizer.model_dump()