diff --git a/src/netsleuth/cli.py b/src/netsleuth/cli.py index 272ca2f..9b438da 100644 --- a/src/netsleuth/cli.py +++ b/src/netsleuth/cli.py @@ -208,6 +208,7 @@ async def _bgp_section(client, settings: Settings, asn: str | None, raw: dict): async def _reputation_section(client, settings: Settings, options: Options, geo: IpGeo, raw: dict): from netsleuth.reputation import ( build_reputation, + ReputationContext, fetch_abuseipdb, fetch_internetdb, normalize_internetdb, @@ -235,12 +236,14 @@ async def _reputation_section(client, settings: Settings, options: Options, geo: else: warnings.append("ABUSEIPDB_API_KEY not set — abuse score skipped") reputation = build_reputation( - internetdb=normalize_internetdb(payload), - firehol_hits=index.hits(geo.ip), - dnsbl_outcomes=outcomes, - ip_type=geo.ip_type, - abuseipdb_score=score, - abuseipdb_reports=reports, + ReputationContext( + internetdb=normalize_internetdb(payload), + firehol_hits=index.hits(geo.ip), + dnsbl_outcomes=outcomes, + ip_type=geo.ip_type, + abuseipdb_score=score, + abuseipdb_reports=reports, + ) ) return ModuleResult( name="reputation", status="partial" if warnings else "ok", data=reputation, warnings=warnings diff --git a/src/netsleuth/reputation.py b/src/netsleuth/reputation.py index d838566..f489da1 100644 --- a/src/netsleuth/reputation.py +++ b/src/netsleuth/reputation.py @@ -208,26 +208,29 @@ def _captcha_risk_full( return risk, "; ".join(reasons), "; ".join(reasons_ru) -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: +@dataclass +class ReputationContext: + ip_type: str + internetdb: InternetDbResult | None + firehol_hits: list[str] + dnsbl_outcomes: list[DnsblOutcome] | None + abuseipdb_score: int | None + abuseipdb_reports: int | None + + +def build_reputation(ctx: ReputationContext) -> Reputation: hits: list[DnsblHit] | None = None blocked = False - if dnsbl_outcomes is not None: - hits, blocked = summarize_dnsbl(dnsbl_outcomes) - risk, rationale, rationale_ru = _captcha_risk_full(firehol_hits, hits or [], ip_type, abuseipdb_score) + if ctx.dnsbl_outcomes is not None: + hits, blocked = summarize_dnsbl(ctx.dnsbl_outcomes) + risk, rationale, rationale_ru = _captcha_risk_full(ctx.firehol_hits, hits or [], ctx.ip_type, ctx.abuseipdb_score) return Reputation( - internetdb=internetdb, - firehol_hits=firehol_hits, + internetdb=ctx.internetdb, + firehol_hits=ctx.firehol_hits, dnsbl_hits=hits, dnsbl_query_blocked=blocked, - abuseipdb_score=abuseipdb_score, - abuseipdb_reports=abuseipdb_reports, + abuseipdb_score=ctx.abuseipdb_score, + abuseipdb_reports=ctx.abuseipdb_reports, captcha_risk=risk, rationale=rationale, rationale_ru=rationale_ru, diff --git a/tests/test_reputation.py b/tests/test_reputation.py index 8d9592e..325f119 100644 --- a/tests/test_reputation.py +++ b/tests/test_reputation.py @@ -3,6 +3,7 @@ from __future__ import annotations from netsleuth.reputation import ( NetsetIndex, build_reputation, + ReputationContext, captcha_risk, decode_dnsbl, normalize_internetdb, @@ -132,12 +133,14 @@ def test_captcha_risk_escalates_on_a_high_abuseipdb_score(): 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, + ReputationContext( + 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 @@ -146,12 +149,14 @@ def test_build_reputation_marks_the_query_as_blocked_without_inventing_hits(api_ 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, + ReputationContext( + 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 @@ -181,12 +186,14 @@ def test_build_reputation_rationale_ru_is_populated_and_cyrillic_for_every_reaso ] for case in cases: rep = build_reputation( - internetdb=normalize_internetdb(api_fixture("internetdb.json")), - firehol_hits=case["firehol_hits"], - dnsbl_outcomes=case["dnsbl_outcomes"], - ip_type=case["ip_type"], - abuseipdb_score=case["abuseipdb_score"], - abuseipdb_reports=None, + ReputationContext( + internetdb=normalize_internetdb(api_fixture("internetdb.json")), + firehol_hits=case["firehol_hits"], + dnsbl_outcomes=case["dnsbl_outcomes"], + ip_type=case["ip_type"], + abuseipdb_score=case["abuseipdb_score"], + abuseipdb_reports=None, + ) ) assert rep.rationale assert rep.rationale_ru