reputation: internetdb, captcha risk and reputation assembly

src/netcheck/reputation.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import ipaddress
 import os
 import time
@@ -7,10 +8,12 @@ from bisect import bisect_right
 from dataclasses import dataclass, field
 from pathlib import Path
 
+import dns.asyncresolver
+import dns.exception
 import httpx
 
 from netcheck.config import Providers
-from netcheck.models import DnsblHit
+from netcheck.models import DnsblHit, InternetDbResult, Reputation
 
 
 def parse_netset(text: str) -> list[str]:
@@ -145,3 +148,104 @@ def summarize_dnsbl(outcomes: list[DnsblOutcome]) -> tuple[list[DnsblHit], bool]
     ]
     blocked = any(o.unavailable_reason is not None for o in outcomes)
     return hits, blocked
+
+
+def normalize_internetdb(payload: dict) -> InternetDbResult:
+    if "ip" not in payload:
+        return InternetDbResult()
+    return InternetDbResult(
+        ip=payload.get("ip"),
+        ports=list(payload.get("ports") or []),
+        hostnames=list(payload.get("hostnames") or []),
+        tags=list(payload.get("tags") or []),
+        cpes=list(payload.get("cpes") or []),
+        vulns=list(payload.get("vulns") or []),
+    )
+
+
+def captcha_risk(
+    firehol_hits: list[str],
+    dnsbl_hits: list[DnsblHit],
+    ip_type: str,
+    abuseipdb_score: int | None,
+) -> tuple[str, str]:
+    reasons: list[str] = []
+    risk = "low"
+    if firehol_hits:
+        risk = "high"
+        reasons.append(f"listed on {', '.join(firehol_hits)}")
+    if dnsbl_hits:
+        risk = "high"
+        reasons.append(f"listed on {', '.join(h.zone for h in dnsbl_hits)}")
+    if abuseipdb_score is not None and abuseipdb_score >= 50:
+        risk = "high"
+        reasons.append(f"AbuseIPDB confidence {abuseipdb_score}")
+    elif abuseipdb_score is not None and abuseipdb_score >= 25 and risk == "low":
+        risk = "medium"
+        reasons.append(f"AbuseIPDB confidence {abuseipdb_score}")
+    if risk == "low" and ip_type == "hosting":
+        risk = "medium"
+        reasons.append("egress looks like hosting/proxy space, which many sites challenge by default")
+    if not reasons:
+        reasons.append("no blocklist match and the address looks like ordinary end-user space")
+    return risk, "; ".join(reasons)
+
+
+def build_reputation(
+    internetdb: InternetDbResult | None,
+    firehol_hits: list[str],
+    dnsbl_outcomes: list[DnsblOutcome] | None,
+    ip_type: str,
+    abuseipdb_score: int | None,
+    abuseipdb_reports: int | None,
+) -> Reputation:
+    hits: list[DnsblHit] | None = None
+    blocked = False
+    if dnsbl_outcomes is not None:
+        hits, blocked = summarize_dnsbl(dnsbl_outcomes)
+    risk, rationale = captcha_risk(firehol_hits, hits or [], ip_type, abuseipdb_score)
+    return Reputation(
+        internetdb=internetdb,
+        firehol_hits=firehol_hits,
+        dnsbl_hits=hits,
+        dnsbl_query_blocked=blocked,
+        abuseipdb_score=abuseipdb_score,
+        abuseipdb_reports=abuseipdb_reports,
+        captcha_risk=risk,
+        rationale=rationale,
+    )
+
+
+async def query_dnsbl(ip: str, zones: list[str], timeout: float) -> list[DnsblOutcome]:
+    label = reverse_ip(ip)
+    resolver = dns.asyncresolver.Resolver()
+    resolver.lifetime = timeout
+
+    async def one(zone: str) -> DnsblOutcome:
+        try:
+            answer = await resolver.resolve(f"{label}.{zone}", "A")
+        except dns.exception.DNSException:
+            return DnsblOutcome(zone=zone, listed=False, codes=[])
+        return decode_dnsbl(zone, sorted(record.address for record in answer))
+
+    return list(await asyncio.gather(*(one(zone) for zone in zones)))
+
+
+async def fetch_internetdb(client: httpx.AsyncClient, providers: Providers, ip: str) -> dict:
+    response = await client.get(f"{providers.internetdb_url}{ip}")
+    if response.status_code == 404:
+        return {"detail": "No information available"}
+    response.raise_for_status()
+    return response.json()
+
+
+async def fetch_abuseipdb(
+    client: httpx.AsyncClient, providers: Providers, ip: str, key: str
+) -> dict:
+    response = await client.get(
+        providers.abuseipdb_url,
+        params={"ipAddress": ip, "maxAgeInDays": "90"},
+        headers={"Key": key, "Accept": "application/json"},
+    )
+    response.raise_for_status()
+    return response.json()

tests/fixtures/api/internetdb.json

@@ -0,0 +1,8 @@
+{
+  "ip": "203.0.113.44",
+  "ports": [22, 80, 443, 7547],
+  "hostnames": ["host-203-0-113-44.example.net"],
+  "cpes": ["cpe:/a:nginx:nginx:1.24.0"],
+  "tags": ["cdn", "iot"],
+  "vulns": ["CVE-2024-1234"]
+}

tests/test_reputation.py

@@ -1,6 +1,14 @@
 from __future__ import annotations
 
-from netcheck.reputation import NetsetIndex, parse_netset
+from netcheck.reputation import (
+    NetsetIndex,
+    build_reputation,
+    captcha_risk,
+    decode_dnsbl,
+    normalize_internetdb,
+    parse_netset,
+    summarize_dnsbl,
+)
 
 
 def test_parse_netset_strips_comments_and_blank_lines(fixtures_dir):
@@ -75,3 +83,76 @@ def test_index_built_from_the_real_fixture_matches_the_expected_entries(fixtures
     assert index.hits("10.1.2.3") == ["firehol_level1"]
     assert index.hits("192.0.2.10") == ["firehol_level1"]
     assert index.hits("192.0.2.200") == []
+
+
+def test_internetdb_normalizes_into_the_typed_result(api_fixture):
+    result = normalize_internetdb(api_fixture("internetdb.json"))
+    assert result.ip == "203.0.113.44"
+    assert result.ports == [22, 80, 443, 7547]
+    assert result.tags == ["cdn", "iot"]
+    assert result.vulns == ["CVE-2024-1234"]
+
+
+def test_internetdb_404_shape_becomes_an_empty_result():
+    result = normalize_internetdb({"detail": "No information available"})
+    assert result.ip is None
+    assert result.ports == []
+
+
+def test_captcha_risk_is_low_for_a_clean_residential_address():
+    risk, rationale = captcha_risk([], [], "residential", None)
+    assert risk == "low"
+    assert rationale
+
+
+def test_captcha_risk_is_medium_for_a_hosting_address_with_no_listings():
+    risk, rationale = captcha_risk([], [], "hosting", None)
+    assert risk == "medium"
+    assert "hosting" in rationale.lower()
+
+
+def test_captcha_risk_is_high_when_a_blocklist_matches():
+    risk, rationale = captcha_risk(["firehol_level1"], [], "residential", None)
+    assert risk == "high"
+    assert "firehol_level1" in rationale
+
+
+def test_captcha_risk_is_high_on_a_real_dnsbl_listing():
+    hits, _ = summarize_dnsbl([decode_dnsbl("zen.spamhaus.org", ["127.0.0.2"])])
+    risk, rationale = captcha_risk([], hits, "residential", None)
+    assert risk == "high"
+    assert "zen.spamhaus.org" in rationale
+
+
+def test_captcha_risk_escalates_on_a_high_abuseipdb_score():
+    assert captcha_risk([], [], "residential", 90)[0] == "high"
+    assert captcha_risk([], [], "residential", 30)[0] == "medium"
+    assert captcha_risk([], [], "residential", 5)[0] == "low"
+
+
+def test_build_reputation_marks_the_query_as_blocked_without_inventing_hits(api_fixture):
+    rep = build_reputation(
+        internetdb=normalize_internetdb(api_fixture("internetdb.json")),
+        firehol_hits=[],
+        dnsbl_outcomes=[decode_dnsbl("zen.spamhaus.org", ["127.255.255.254"])],
+        ip_type="residential",
+        abuseipdb_score=None,
+        abuseipdb_reports=None,
+    )
+    assert rep.dnsbl_hits == []
+    assert rep.dnsbl_query_blocked is True
+    assert rep.captcha_risk == "low"
+
+
+def test_build_reputation_without_dnsbl_leaves_the_field_none(api_fixture):
+    rep = build_reputation(
+        internetdb=normalize_internetdb(api_fixture("internetdb.json")),
+        firehol_hits=["firehol_level1"],
+        dnsbl_outcomes=None,
+        ip_type="residential",
+        abuseipdb_score=None,
+        abuseipdb_reports=None,
+    )
+    assert rep.dnsbl_hits is None
+    assert rep.dnsbl_query_blocked is False
+    assert rep.captcha_risk == "high"