fix: [scanmalware] list only browser-contacted hosts, report a degraded CT answer, trailing dot

misp_modules/modules/expansion/scanmalware.py

@@ -98,7 +98,7 @@ def _is_scan_of_host(scan, hostname):
     """
     if "url" not in (scan.get("matched_on") or []):
         return False
-    scanned = (urlparse(scan.get("url") or "").hostname or "").lower()
+    scanned = (urlparse(scan.get("url") or "").hostname or "").lower().rstrip(".")
     host = (hostname or "").lower().rstrip(".")
     return bool(host) and scanned in (host, "www." + host)
 
@@ -170,6 +170,19 @@ def parse_domain(self, domain):
             {"subdomain_limit": SUBDOMAIN_LIMIT},
         )
         if ct:
+            # `degraded` means the service could not look: an empty list is then an
+            # absence of data, not a finding that the domain has no subdomains. Added
+            # first so the result cap can never drop it.
+            if ct.get("degraded"):
+                self._add(
+                    type="text",
+                    value=(
+                        f"ScanMalware could not read its Certificate Transparency data for {domain} just now;"
+                        " a missing or short subdomain list is not a finding. Retry later."
+                    ),
+                    comment="ScanMalware: partial result",
+                    disable_correlation=True,
+                )
             self._add_hosts(ct.get("subdomains") or [], "ScanMalware: seen in Certificate Transparency DNS records")
             if ct.get("subdomains_truncated"):
                 self._add(
@@ -180,11 +193,13 @@ def parse_domain(self, domain):
                 )
 
         # A different population: hosts a browser resolved and requested. Dev and
-        # staging hosts holding no certificate appear only here.
+        # staging hosts holding no certificate appear only here. The endpoint's default
+        # also returns `stream` hosts, which were never fetched (existence only), so the
+        # two runtime-observed sources are asked for explicitly.
         hosts = _get(
             self.api_url,
             f"/api/v1/hosts/{quote(domain, safe='')}",
-            {"subdomains_only": "true"},
+            {"subdomains_only": "true", "sources": "observed,scripts"},
         )
         if hosts:
             self._add_hosts(hosts.get("subdomains") or [], "ScanMalware: contacted by a browser during a scan")

tests/test_scanmalware.py

@@ -52,6 +52,24 @@ def test_domain_returns_subdomains_from_both_sources():
     assert "observed.example.com" in values
 
 
+def test_browser_contacted_hosts_are_only_the_runtime_observed_sources():
+    """
+    The hosts endpoint's default also returns `stream` hosts, which were never fetched.
+    Labelling those "contacted by a browser during a scan" would be a false statement.
+    """
+    calls = []
+
+    def _get(url, params=None, headers=None, timeout=None):
+        calls.append((url, params or {}))
+        return MockResponse(HOSTS_PAYLOAD if "/hosts/" in url else {})
+
+    with patch.object(scanmalware.requests, "get", _get):
+        scanmalware.handler(_query("example.com"))
+
+    hosts_params = [params for url, params in calls if "/hosts/" in url]
+    assert hosts_params and hosts_params[0].get("sources") == "observed,scripts"
+
+
 def test_wildcard_certificate_entries_are_dropped():
     """`*.example.com` appears literally in certificate data and is not a resolvable host."""
     with patch.object(scanmalware.requests, "get", _by_path({"ct/dns": CT_PAYLOAD})):
@@ -72,6 +90,19 @@ def test_truncated_subdomain_list_is_reported():
     assert any("truncated" in a["value"] for a in results["results"]["Attribute"])
 
 
+def test_a_degraded_ct_answer_is_not_reported_as_no_subdomains():
+    """
+    `degraded: true` means the service could not look. Answering "No ScanMalware
+    results" would tell the analyst we looked and found nothing.
+    """
+    payload = {"subdomains": [], "subdomains_truncated": False, "degraded": True}
+    with patch.object(scanmalware.requests, "get", _by_path({"ct/dns": payload})):
+        results = scanmalware.handler(_query("example.com"))
+
+    texts = [a["value"] for a in (results.get("results") or {}).get("Attribute") or [] if a["type"] == "text"]
+    assert any("could not read its Certificate Transparency data" in t for t in texts)
+
+
 def test_results_are_capped_and_the_cap_is_reported():
     """
     A busy domain can return over a thousand subdomains. Adding them all to one event is
@@ -296,6 +327,18 @@ def test_a_subdomain_scan_is_not_a_scan_of_the_platform():
     assert not any("wallet-support" in str(a.get("value")) for a in attributes)
 
 
+def test_a_trailing_dot_on_the_scanned_host_is_the_same_host():
+    """`example.com.` is `example.com`, on the scan's side as on the attribute's."""
+    own = {"scan_id": "own", "url": "https://example.com./", "matched_on": ["url"]}
+    fake = _verdicts_by_scan([], [own], {"own": "Low Risk"})
+
+    with patch.object(scanmalware.requests, "get", fake):
+        results = scanmalware.handler(_query("https://example.com./login", "url"))
+
+    texts = [a["value"] for a in (results.get("results") or {}).get("Attribute") or [] if a["type"] == "text"]
+    assert texts and "nearest scan on the same host: https://example.com./" in texts[0]
+
+
 def test_the_exact_url_matches_without_its_trailing_slash():
     """The archive stores `https://example.com` for a scan of `https://example.com/`."""
     row = {"scan_id": "root", "url": "https://example.com"}