Feature/disallowed domains in csp

boefjes/boefjes/plugins/kat_answer_parser/normalize.py

@@ -18,3 +18,13 @@ def run(normalizer_meta: NormalizerMeta, raw: bytes | str) -> Iterable[OOI]:
             bit_id="port-classification-ip",
             config=json.loads(raw),
         )
+
+    if "/bit/disallowed-csp-hostnames" in mime_types:
+        if isinstance(raw, bytes):
+            raw = raw.decode()
+
+        yield Config(
+            ooi=normalizer_meta.raw_data.boefje_meta.input_ooi,
+            bit_id="disallowed-csp-hostnames",
+            config=json.loads(raw),
+        )

boefjes/boefjes/plugins/kat_kat_finding_types/kat_finding_types.json

@@ -467,5 +467,11 @@
     "risk": "low",
     "impact": "All Certificate Authorities may issue certificates for you domain.",
     "recommendation": "Set a CAA record to limit which CA's are allowed to issue certs."
+  },
+  "KAT-DISALLOWED-DOMAIN-IN-CSP": {
+    "description": "This CSP header contains domains that are not allowed, If the website contains a cross-site scripting vulnerability, then JavaScript code can be injected into the web page hosted on these domains which can host files for anyone.",
+    "risk": "medium",
+    "impact": "Disallowed domains are domains that are for example 'world writable', this opens up the possibility for an atacker to host malicious files on a csp whitelisted domain.",
+    "recommendation": "Remove the offending hostname from the CSP header."
   }
 }

octopoes/.ci/mock_bits/url_classification_mock/url_classification_mock.py

@@ -7,7 +7,7 @@
 from octopoes.models.ooi.web import URL, HostnameHTTPURL, IPAddressHTTPURL, WebScheme
 
 
-def run(url: URL, additional_oois: list, config: dict[str, str]) -> Iterator[OOI]:
+def run(url: URL, additional_oois: list, config: dict) -> Iterator[OOI]:
     if url.raw.scheme == "http" or url.raw.scheme == "https":
         port = url.raw.port
         if port is None:

octopoes/bits/ask_disallowed_domains/ask_disallowed_domains.py

@@ -0,0 +1,21 @@
+import json
+from collections.abc import Iterator
+from pathlib import Path
+from typing import Any
+
+from octopoes.models import OOI
+from octopoes.models.ooi.network import Network
+from octopoes.models.ooi.question import Question
+
+
+def run(
+    input_ooi: Network,
+    additional_oois: list,
+    config: dict[str, Any],
+) -> Iterator[OOI]:
+    network = input_ooi
+
+    with (Path(__file__).parent / "question_schema.json").open() as f:
+        schema = json.load(f)
+
+    yield Question(ooi=network.reference, schema_id=schema["$id"], json_schema=json.dumps(schema))

octopoes/bits/ask_disallowed_domains/bit.py

@@ -0,0 +1,10 @@
+from bits.definitions import BitDefinition
+from octopoes.models.ooi.network import Network
+
+BIT = BitDefinition(
+    id="ask-disallowed-domains",
+    consumes=Network,
+    parameters=[],
+    min_scan_level=0,
+    module="bits.ask_disallowed_domains.ask_disallowed_domains",
+)

octopoes/bits/ask_disallowed_domains/question_schema.json

@@ -0,0 +1,21 @@
+{
+  "$schema": "https://json-schema.org/draft/2019-09/schema",
+  "$id": "/bit/disallowed-csp-hostnames",
+  "type": "object",
+  "default": {},
+  "Port Configuration": "Root Schema",
+  "required": [],
+  "properties": {
+    "disallowed_hostnames": {
+      "description": "Comma separated list of disallowed hostnames in CSP headers",
+      "type": "string",
+      "pattern": "^(\\s*(,*)[^,]+,?\\s*)*$",
+      "default": "github.com,google.com"
+    },
+    "disallow_url_shorteners": {
+      "description": "Do you want to disallow url shorteners in csp headers?",
+      "type": "boolean",
+      "default": true
+    }
+  }
+}

octopoes/bits/ask_port_specification/ask_port_specification.py

@@ -1,6 +1,7 @@
 import json
 from collections.abc import Iterator
 from pathlib import Path
+from typing import Any
 
 from octopoes.models import OOI
 from octopoes.models.ooi.network import Network
@@ -10,7 +11,7 @@
 def run(
     input_ooi: Network,
     additional_oois: list,
-    config: dict[str, str],
+    config: dict[str, Any],
 ) -> Iterator[OOI]:
     network = input_ooi
 

octopoes/bits/check_csp_header/check_csp_header.py

@@ -1,6 +1,7 @@
 import ipaddress
 import re
 from collections.abc import Iterator
+from typing import Any
 
 from octopoes.models import OOI, Reference
 from octopoes.models.ooi.findings import Finding, KATFindingType
@@ -9,7 +10,7 @@
 NON_DECIMAL_FILTER = re.compile(r"[^\d.]+")
 
 
-def run(input_ooi: HTTPHeader, additional_oois: list, config: dict[str, str]) -> Iterator[OOI]:
+def run(input_ooi: HTTPHeader, additional_oois: list, config: dict[str, Any]) -> Iterator[OOI]:
     header = input_ooi
     if header.key.lower() != "content-security-policy":
         return

octopoes/bits/check_cve_2021_41773/check_cve_2021_41773.py

@@ -1,11 +1,12 @@
 from collections.abc import Iterator
+from typing import Any
 
 from octopoes.models import OOI
 from octopoes.models.ooi.findings import CVEFindingType, Finding
 from octopoes.models.types import HTTPHeader
 
 
-def run(input_ooi: HTTPHeader, additional_oois: list, config: dict[str, str]) -> Iterator[OOI]:
+def run(input_ooi: HTTPHeader, additional_oois: list, config: dict[str, Any]) -> Iterator[OOI]:
     header = input_ooi
     if header.key.lower() != "server":
         return

octopoes/bits/check_hsts_header/check_hsts_header.py

@@ -1,12 +1,13 @@
 import datetime
 from collections.abc import Iterator
+from typing import Any
 
 from octopoes.models import OOI, Reference
 from octopoes.models.ooi.findings import Finding, KATFindingType
 from octopoes.models.types import HTTPHeader
 
 
-def run(input_ooi: HTTPHeader, additional_oois: list, config: dict[str, str]) -> Iterator[OOI]:
+def run(input_ooi: HTTPHeader, additional_oois: list, config: dict[str, Any]) -> Iterator[OOI]:
     header = input_ooi
     if header.key.lower() != "strict-transport-security":
         return

octopoes/bits/default_findingtype_risk/default_findingtype_risk.py

@@ -1,10 +1,11 @@
 from collections.abc import Iterator
+from typing import Any
 
 from octopoes.models import OOI
 from octopoes.models.ooi.findings import FindingType, RiskLevelSeverity
 
 
-def run(input_ooi: FindingType, additional_oois: list, config: dict[str, str]) -> Iterator[OOI]:
+def run(input_ooi: FindingType, additional_oois: list, config: dict[str, Any]) -> Iterator[OOI]:
     value_set = False
     if not input_ooi.risk_severity:
         input_ooi.risk_severity = RiskLevelSeverity.PENDING

octopoes/bits/disallowed_csp_hostnames/bit.py

@@ -0,0 +1,10 @@
+from bits.definitions import BitDefinition
+from octopoes.models.ooi.web import HTTPHeaderHostname
+
+BIT = BitDefinition(
+    id="disallowed-csp-hostnames",
+    consumes=HTTPHeaderHostname,
+    parameters=[],
+    module="bits.disallowed_csp_hostnames.disallowed_csp_hostnames",
+    config_ooi_relation_path="HTTPHeaderHostname.hostname.network",
+)

octopoes/bits/disallowed_csp_hostnames/disallowed_csp_hostnames.py

@@ -0,0 +1,45 @@
+import logging
+from collections.abc import Iterator
+from typing import Any
+
+from link_shorteners import link_shorteners_list
+
+from octopoes.models import OOI
+from octopoes.models.ooi.findings import Finding, KATFindingType
+from octopoes.models.ooi.web import HTTPHeaderHostname
+
+LINK_SHORTENERS = link_shorteners_list()
+
+logger = logging.getLogger(__name__)
+
+
+def get_disallowed_hostnames_from_config(config, config_key, default):
+    disallowed_hostnames = config.get(config_key, None)
+    if disallowed_hostnames is None:
+        return default
+    return list(disallowed_hostnames.strip().split(",")) if disallowed_hostnames else []
+
+
+def run(input_ooi: HTTPHeaderHostname, additional_oois: list, config: dict[str, Any]) -> Iterator[OOI]:
+    header_hostname = input_ooi
+    header = header_hostname.header
+
+    if header.tokenized.key.lower() != "content-security-policy":
+        return
+
+    disallow_url_shorteners = config.get("disallow_url_shorteners", True) if config else True
+
+    hostname = header_hostname.hostname.tokenized.name
+    disallowed_domains = link_shorteners_list() if disallow_url_shorteners else []
+    disallowed_hostnames_from_config = get_disallowed_hostnames_from_config(config, "disallowed_hostnames", [])
+
+    disallowed_domains.extend(disallowed_hostnames_from_config)
+
+    if hostname.lower() in disallowed_domains:
+        ft = KATFindingType(id="KAT-DISALLOWED-DOMAIN-IN-CSP")
+        f = Finding(
+            ooi=input_ooi.reference,
+            finding_type=ft.reference,
+        )
+        yield ft
+        yield f

octopoes/bits/dns_alias_resolving/dns_alias_resolving.py

@@ -1,4 +1,5 @@
 from collections.abc import Iterator
+from typing import Any
 
 from octopoes.models import OOI
 from octopoes.models.ooi.dns.records import DNSCNAMERecord
@@ -7,7 +8,7 @@
 
 
 def run(
-    hostname: Hostname, additional_oois: list[DNSCNAMERecord | ResolvedHostname], config: dict[str, str]
+    hostname: Hostname, additional_oois: list[DNSCNAMERecord | ResolvedHostname], config: dict[str, Any]
 ) -> Iterator[OOI]:
     cname_records = [ooi for ooi in additional_oois if isinstance(ooi, DNSCNAMERecord)]
     resolved_hostnames = [ooi for ooi in additional_oois if isinstance(ooi, ResolvedHostname)]

octopoes/bits/dns_resolving/dns_resolving.py

@@ -1,11 +1,12 @@
 from collections.abc import Iterator
+from typing import Any
 
 from octopoes.models import OOI
 from octopoes.models.ooi.dns.records import DNSAAAARecord, DNSARecord
 from octopoes.models.ooi.dns.zone import Hostname, ResolvedHostname
 
 
-def run(hostname: Hostname, additional_oois: list[DNSARecord | DNSAAAARecord], config: dict[str, str]) -> Iterator[OOI]:
+def run(hostname: Hostname, additional_oois: list[DNSARecord | DNSAAAARecord], config: dict[str, Any]) -> Iterator[OOI]:
     for record in additional_oois:
         yield ResolvedHostname(
             hostname=hostname.reference,

octopoes/bits/expiring_certificate/expiring_certificate.py

@@ -1,5 +1,6 @@
 import datetime
 from collections.abc import Iterator
+from typing import Any
 
 from octopoes.models import OOI
 from octopoes.models.ooi.certificate import X509Certificate
@@ -9,7 +10,7 @@
 THRESHOLD = datetime.timedelta(weeks=2)
 
 
-def run(input_ooi: X509Certificate, additional_oois: list[Website], config: dict[str, str]) -> Iterator[OOI]:
+def run(input_ooi: X509Certificate, additional_oois: list[Website], config: dict[str, Any]) -> Iterator[OOI]:
     # only applies to OOIs referencing the certificate
     if input_ooi.expired:
         ft = KATFindingType(id="KAT-CERTIFICATE-EXPIRED")

octopoes/bits/https_availability/https_availability.py

@@ -1,12 +1,13 @@
 from collections.abc import Iterator
+from typing import Any
 
 from octopoes.models import OOI
 from octopoes.models.ooi.findings import Finding, KATFindingType
 from octopoes.models.ooi.network import IPAddress, IPPort
 from octopoes.models.ooi.web import Website
 
 
-def run(input_ooi: IPAddress, additional_oois: list[IPPort | Website], config: dict[str, str]) -> Iterator[OOI]:
+def run(input_ooi: IPAddress, additional_oois: list[IPPort | Website], config: dict[str, Any]) -> Iterator[OOI]:
     websites = [website for website in additional_oois if isinstance(website, Website)]
 
     open_ports = [port.port for port in additional_oois if isinstance(port, IPPort)]

octopoes/bits/https_redirect/https_redirect.py

@@ -1,11 +1,12 @@
 from collections.abc import Iterator
+from typing import Any
 
 from octopoes.models import OOI
 from octopoes.models.ooi.findings import Finding, KATFindingType
 from octopoes.models.ooi.web import HostnameHTTPURL, HTTPHeader
 
 
-def run(input_ooi: HostnameHTTPURL, additional_oois: list[HTTPHeader], config: dict[str, str]) -> Iterator[OOI]:
+def run(input_ooi: HostnameHTTPURL, additional_oois: list[HTTPHeader], config: dict[str, Any]) -> Iterator[OOI]:
     header_keys = [header.key.lower() for header in additional_oois if isinstance(header, HTTPHeader)]
 
     # only check for http urls

octopoes/bits/internetnl/internetnl.py

@@ -1,12 +1,13 @@
 from collections.abc import Iterator
+from typing import Any
 
 from octopoes.models import OOI
 from octopoes.models.ooi.dns.zone import Hostname
 from octopoes.models.ooi.findings import Finding, KATFindingType
 from octopoes.models.ooi.web import Website
 
 
-def run(input_ooi: Hostname, additional_oois: list[Finding | Website], config: dict[str, str]) -> Iterator[OOI]:
+def run(input_ooi: Hostname, additional_oois: list[Finding | Website], config: dict[str, Any]) -> Iterator[OOI]:
     # only websites have to comply with the internetnl rules
     websites = [websites for websites in additional_oois if isinstance(websites, Website)]
     if not websites:

octopoes/bits/ipv6_nameservers/ipv6_nameservers.py

@@ -1,12 +1,13 @@
 from collections.abc import Iterator
+from typing import Any
 
 from octopoes.models import OOI
 from octopoes.models.ooi.dns.records import DNSAAAARecord, DNSARecord, DNSNSRecord
 from octopoes.models.ooi.dns.zone import Hostname
 from octopoes.models.ooi.findings import Finding, KATFindingType
 
 
-def run(hostname: Hostname, additional_oois: list[DNSAAAARecord | DNSARecord], config: dict[str, str]) -> Iterator[OOI]:
+def run(hostname: Hostname, additional_oois: list[DNSAAAARecord | DNSARecord], config: dict[str, Any]) -> Iterator[OOI]:
     dns_ns_records = [dns_ns_record for dns_ns_record in additional_oois if isinstance(dns_ns_record, DNSNSRecord)]
     dns_aaaa_records = [
         dns_aaaa_record for dns_aaaa_record in additional_oois if isinstance(dns_aaaa_record, DNSAAAARecord)

octopoes/bits/ipv6_webservers/ipv6_webservers.py

@@ -7,7 +7,7 @@
 
 
 def run(
-    hostname: Hostname, additional_oois: list[DNSAAAARecord | DNSARecord | DNSNSRecord], config: dict[str, str]
+    hostname: Hostname, additional_oois: list[DNSAAAARecord | DNSARecord | DNSNSRecord], config: dict
 ) -> Iterator[OOI]:
     dns_a_records = [dns_a_record for dns_a_record in additional_oois if isinstance(dns_a_record, DNSARecord)]
     dns_aaaa_records = [

octopoes/bits/missing_caa/missing_caa.py

@@ -1,4 +1,5 @@
 from collections.abc import Iterator
+from typing import Any
 
 import tldextract
 
@@ -8,7 +9,7 @@
 from octopoes.models.ooi.findings import Finding, KATFindingType
 
 
-def run(input_ooi: Hostname, additional_oois: list[DNSCAARecord | NXDOMAIN], config: dict[str, str]) -> Iterator[OOI]:
+def run(input_ooi: Hostname, additional_oois: list[DNSCAARecord | NXDOMAIN], config: dict[str, Any]) -> Iterator[OOI]:
     caa_records = [ooi for ooi in additional_oois if isinstance(ooi, DNSCAARecord)]
     nxdomains = (ooi for ooi in additional_oois if isinstance(ooi, NXDOMAIN))
 

octopoes/bits/missing_certificate/missing_certificate.py

@@ -1,11 +1,12 @@
 from collections.abc import Iterator
+from typing import Any
 
 from octopoes.models import OOI
 from octopoes.models.ooi.findings import Finding, KATFindingType
 from octopoes.models.ooi.web import Website
 
 
-def run(input_ooi: Website, additional_oois, config: dict[str, str]) -> Iterator[OOI]:
+def run(input_ooi: Website, additional_oois, config: dict[str, Any]) -> Iterator[OOI]:
     if input_ooi.ip_service.tokenized.service.name.lower() != "https":
         return
 

octopoes/bits/missing_dkim/missing_dkim.py

@@ -1,4 +1,5 @@
 from collections.abc import Iterator
+from typing import Any
 
 import tldextract
 
@@ -9,7 +10,7 @@
 from octopoes.models.ooi.findings import Finding, KATFindingType
 
 
-def run(input_ooi: Hostname, additional_oois: list[DKIMExists | NXDOMAIN], config: dict[str, str]) -> Iterator[OOI]:
+def run(input_ooi: Hostname, additional_oois: list[DKIMExists | NXDOMAIN], config: dict[str, Any]) -> Iterator[OOI]:
     dkim_exists = [ooi for ooi in additional_oois if isinstance(ooi, DKIMExists)]
     nxdomains = (ooi for ooi in additional_oois if isinstance(ooi, NXDOMAIN))
 

octopoes/bits/missing_dmarc/missing_dmarc.py

@@ -1,4 +1,5 @@
 from collections.abc import Iterator
+from typing import Any
 
 import tldextract
 
@@ -9,7 +10,7 @@
 from octopoes.models.ooi.findings import Finding, KATFindingType
 
 
-def run(input_ooi: Hostname, additional_oois: list[DMARCTXTRecord | NXDOMAIN], config: dict[str, str]) -> Iterator[OOI]:
+def run(input_ooi: Hostname, additional_oois: list[DMARCTXTRecord | NXDOMAIN], config: dict[str, Any]) -> Iterator[OOI]:
     dmarc_records = [ooi for ooi in additional_oois if isinstance(ooi, DMARCTXTRecord)]
     nxdomains = (ooi for ooi in additional_oois if isinstance(ooi, NXDOMAIN))