From d7b4c49786cd094cd2495e2bad7c2c1bd6e5e30c Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sun, 9 Aug 2026 02:21:57 +0300 Subject: [PATCH] reputation: internetdb, captcha risk and reputation assembly --- src/netcheck/reputation.py | 106 ++++++++++++++++++++++++++++- tests/fixtures/api/internetdb.json | 8 +++ tests/test_reputation.py | 83 +++++++++++++++++++++- 3 files changed, 195 insertions(+), 2 deletions(-) create mode 100644 tests/fixtures/api/internetdb.json diff --git a/src/netcheck/reputation.py b/src/netcheck/reputation.py index d80a7b2..d80824e 100644 --- a/src/netcheck/reputation.py +++ b/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() diff --git a/tests/fixtures/api/internetdb.json b/tests/fixtures/api/internetdb.json new file mode 100644 index 0000000..8f15f07 --- /dev/null +++ b/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"] +} diff --git a/tests/test_reputation.py b/tests/test_reputation.py index 8d00f62..347b25e 100644 --- a/tests/test_reputation.py +++ b/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"