From 037d28da7bee69dcf9fcd6c8c1ff888da583d33c Mon Sep 17 00:00:00 2001 From: Joshua Davis Date: Sat, 31 Oct 2026 18:04:50 -0400 Subject: [PATCH] Score saved scans as of when they were taken (roadmap 3.6) Rescores aged saved scans against today, so results drifted by the day and a past figure couldn't be reproduced. New scans record their time, and older ones are listed in the report and can be fixed to a date with --as-of. Story: PDX-28 Co-Authored-By: Claude Opus 5.5 --- README.md | 4 ++- ROADMAP.md | 8 +++--- paradox/calibrate.py | 53 +++++++++++++++++++++++++++-------- paradox/sources.py | 6 ++-- tests/test_scoring.py | 64 +++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 115 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index cbe7e7b..8c85941 100644 --- a/README.md +++ b/README.md @@ -216,7 +216,9 @@ GreyNoise), both `explained`. The other 28 malicious IPs: AUC 1.000, detection 1 With it on, both scanners read as `veneer` instead (see the note below). Raw OTX responses are saved, so OTX settings (`OTX_*` in `paradox/sources.py`) and the scoring weights -can be re-tested offline with `--rescore`, with no API calls. `--refresh ` re-runs +can be re-tested offline with `--rescore`, with no API calls. Each scan is aged as of when it was taken, so a +rescore gives the same result on any day. Scans saved before that was recorded are listed in the report and age +against today unless `--as-of YYYY-MM-DD` says otherwise. `--refresh ` re-runs just those beacons on the saved scans, keeping the old reading wherever a call fails. If OTX starts rate-limiting, the run stops early and fills in the gaps next time. diff --git a/ROADMAP.md b/ROADMAP.md index a699881..17501fa 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -24,7 +24,7 @@ Effort: **S** is about a day, **M** a few days, **L** a week or more. Each item has a permanent story ID (`PDX-n`) and a "done when" line for the [Muster board](https://github.com/Castor-o7/muster/blob/main/docs/board.md). A checked box means shipped, not verified. -The watchman keeps each story's state. **Progress: 21 of 40 stories shipped (52%),** counting each story the same +The watchman keeps each story's state. **Progress: 22 of 40 stories shipped (55%),** counting each story the same whatever its size, and the total grows as work finds new stories. --- @@ -336,9 +336,9 @@ with no network calls. "Guarantee" tests must keep holding, and "known weakness" *Limits:* these pin today's rules against attacks we thought of, on fixtures where the other beacons are set by hand. They don't show what a real attacker can do, whether OTX would catch sock puppets, or how often this happens. -### 3.6 Rescores that reproduce · S +### 3.6 Rescores that reproduce · S · ✅ shipped -- [ ] **PDX-28: Score saved scans as of when they were taken.** Done when: `paradox calibrate --rescore` ages each saved scan against its recorded scan time, scans without one say so in the report, and a test shows the same saved scan gets the same score on different days. +- [x] **PDX-28: Score saved scans as of when they were taken.** Done when: `paradox calibrate --rescore` ages each saved scan against its recorded scan time, scans without one say so in the report, and a test shows the same saved scan gets the same score on different days. Saved scans are aged against today's date, so recency rules such as 1.6's window and the veneer's freshness check give different results depending on the day of the rescore, and a past result can't be reproduced. @@ -589,7 +589,7 @@ and fix the top issues. Repeat after 5.6. It also doubles as class material. | When | Items | Why first | |---|---|---| | Now | ~~1.1, 1.2, 1.5, 3.2, 5.1~~ (done) | Small, keyless, and fix the biggest measured weakness (the scanner rule's dependence on GreyNoise); 5.1 makes that dependence visible | -| Next | 1.3, 3.1, 3.6, 3.8, 3.9, 3.10, 4.1, ~~3.5, 4.6, 4.7, 4.8, 4.9, 4.10, 4.5, 1.7, 1.8, 3.7, 1.4, 1.6, 5.2, 5.4~~ (done) | Close the known false positives, build adversarial and hiding-place test sets, test the AI summarizer under attack, and show watch state and claim checks in the UI | +| Next | 1.3, 3.1, 3.8, 3.9, 3.10, 4.1, ~~3.6, 3.5, 4.6, 4.7, 4.8, 4.9, 4.10, 4.5, 1.7, 1.8, 3.7, 1.4, 1.6, 5.2, 5.4~~ (done) | Close the known false positives, build adversarial and hiding-place test sets, test the AI summarizer under attack, and show watch state and claim checks in the UI | | Then | 2.1, 2.3, 3.3, 3.4, 4.2, ~~4.3, 4.4~~ (done), 5.3, 5.5–5.7 | New unmasking signals plus self-measurement once history has accumulated, with the UI restructured to hold them | | Later | 2.2, 2.4, 2.5 | Bigger extensions: domains, campaigns, more beacons | diff --git a/paradox/calibrate.py b/paradox/calibrate.py index d1c5879..eb52824 100644 --- a/paradox/calibrate.py +++ b/paradox/calibrate.py @@ -36,9 +36,10 @@ import time from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone from . import config, scoring -from .sources import (BEACONS, NEIGHBOR_LISTS, Signal, _drop_networks, _feodo_index, _ip_list, +from .sources import (BEACONS, NEIGHBOR_LISTS, Signal, _drop_networks, _feodo_index, _ip_list, parse_dt, apply_otx, fetch_otx, run_beacon, virustotal_confidence) DIR = config.DATA_DIR / "calibration" @@ -103,16 +104,17 @@ def scan_raw(ip: str) -> dict: others = [b for b in BEACONS if b.id != "otx"] with ThreadPoolExecutor(max_workers=len(others)) as pool: signals = list(pool.map(lambda b: run_beacon(b, ip), others)) - return {"signals": [s.to_dict() for s in signals], "otx_raw": otx_raw, "otx_error": otx_error} + return {"scanned_at": datetime.now(timezone.utc).isoformat(), + "signals": [s.to_dict() for s in signals], "otx_raw": otx_raw, "otx_error": otx_error} -def build_signals(ip: str, saved: dict) -> list[Signal]: +def build_signals(ip: str, saved: dict, as_of: datetime | None = None) -> list[Signal]: """Rebuild a scan's signals, re-scoring OTX from its raw response with the - current settings in sources.py.""" + current settings in sources.py, aged to as_of.""" otx = next(b for b in BEACONS if b.id == "otx") sig = Signal(source=otx.id, name=otx.name, tier=otx.tier) if saved.get("otx_raw") is not None: - apply_otx(ip, sig, saved["otx_raw"]) + apply_otx(ip, sig, saved["otx_raw"], as_of) else: sig.status, sig.summary = "error", saved.get("otx_error") or "no OTX data" others = [Signal(**s) for s in saved["signals"]] @@ -126,9 +128,14 @@ def build_signals(ip: str, saved: dict) -> list[Signal]: return [sig] + others -def collect(ips: list[str], pause: float, offline: bool) -> dict[str, list[Signal]]: +def collect(ips: list[str], pause: float, offline: bool, + undated_as_of: datetime | None = None) -> tuple[dict[str, list[Signal]], dict[str, datetime], list[str]]: + """The signals for each IP, when each scan counts as taken, and which scans have no + recorded time. Those use undated_as_of, or are left to age against today.""" SCANS.mkdir(parents=True, exist_ok=True) out = {} + times: dict[str, datetime] = {} + undated = [] limited = 0 for i, ip in enumerate(ips, 1): path = SCANS / f"{ip.replace(':', '_')}.json" @@ -153,8 +160,14 @@ def collect(ips: list[str], pause: float, offline: bool) -> dict[str, list[Signa limited = 0 path.write_text(json.dumps(saved)) time.sleep(pause) - out[ip] = build_signals(ip, saved) - return out + as_of = parse_dt(saved.get("scanned_at")) + if as_of is None: + undated.append(ip) + as_of = undated_as_of + if as_of is not None: + times[ip] = as_of + out[ip] = build_signals(ip, saved, as_of) + return out, times, undated def refresh(ips: list[str], beacon_ids: list[str], pause: float) -> None: @@ -170,6 +183,7 @@ def refresh(ips: list[str], beacon_ids: list[str], pause: float) -> None: new = run_beacon(b, ip) if new.status == "ok": # replace the saved reading, or add a beacon newer than the scan saved["signals"] = [s for s in saved["signals"] if s["source"] != b.id] + [new.to_dict()] + saved["scanned_at"] = datetime.now(timezone.utc).isoformat() # its newest reading mustn't be future-dated else: print(f" {ip}: {b.id} {new.status}, keeping saved reading", file=sys.stderr) path.write_text(json.dumps(saved)) @@ -185,12 +199,12 @@ def auc(pos: list[float], neg: list[float]) -> float | None: return wins / (len(pos) * len(neg)) -def evaluate(labels: dict, scans: dict[str, list[Signal]]) -> dict: +def evaluate(labels: dict, scans: dict[str, list[Signal]], times: dict[str, datetime] | None = None) -> dict: rows = [] for cls in ("malicious", "benign"): for ip in labels[cls]: if ip in scans: - sc = scoring.compute(scans[ip]) + sc = scoring.compute(scans[ip], now=(times or {}).get(ip)) scanner_by = sorted({s.name for s in scans[ip] if s.status == "ok" and s.facts.get("label_kind") == "scanner"}) rows.append({"ip": ip, "label": cls, "threat": sc["threat"], "paradox": sc["paradox"], @@ -284,6 +298,13 @@ def render(ev: dict) -> str: return "\n".join(lines) + "\n" +def as_of_date(text: str) -> datetime: + try: + return datetime.strptime(text, "%Y-%m-%d").replace(tzinfo=timezone.utc) + except ValueError: + raise argparse.ArgumentTypeError(f"'{text}' isn't a YYYY-MM-DD date") from None + + def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(prog="paradox calibrate", description=__doc__.split("\n\n")[0]) ap.add_argument("--rescore", action="store_true", help="score saved scans only; no API calls") @@ -291,6 +312,8 @@ def main(argv: list[str] | None = None) -> int: ap.add_argument("--size", type=int, default=30, help="malicious sample size (default 30)") ap.add_argument("--seed", type=int, default=7) ap.add_argument("--pause", type=float, default=1.5, help="seconds between live scans") + ap.add_argument("--as-of", type=as_of_date, metavar="YYYY-MM-DD", + help="when to treat scans with no recorded time as taken (default: today, so results drift)") ap.add_argument("--refresh", nargs="+", metavar="BEACON", help="re-run only these beacon ids on saved scans") args = ap.parse_args(argv) @@ -303,9 +326,15 @@ def main(argv: list[str] | None = None) -> int: if unknown: ap.error(f"unknown beacon id(s): {', '.join(sorted(unknown))} (OTX is re-scored with --rescore)") refresh(labels["malicious"] + labels["benign"], args.refresh, args.pause) - scans = collect(labels["malicious"] + labels["benign"], args.pause, offline=args.rescore or bool(args.refresh)) - ev = evaluate(labels, scans) + scans, times, undated = collect(labels["malicious"] + labels["benign"], args.pause, + offline=args.rescore or bool(args.refresh), undated_as_of=args.as_of) + ev = evaluate(labels, scans, times) + ev["undated_scans"] = {"ips": undated, "as_of": args.as_of.date().isoformat() if args.as_of else "today"} print(render(ev)) + if undated: + print(f" {len(undated)} of {len(scans)} scans have no recorded time, so they were aged as of " + f"{ev['undated_scans']['as_of']}." + + ("" if args.as_of else " These results change with the day; use --as-of to fix them.") + "\n") (DIR / "report.json").write_text(json.dumps(ev, indent=1, default=str)) return 0 diff --git a/paradox/sources.py b/paradox/sources.py index be43093..57a5577 100644 --- a/paradox/sources.py +++ b/paradox/sources.py @@ -306,10 +306,10 @@ def otx_scan_share(data: dict) -> float | None: return round(scan / (scan + attack), 3) if scan + attack else None -def apply_otx(ip: str, sig: Signal, data: dict) -> None: +def apply_otx(ip: str, sig: Signal, data: dict, now: datetime | None = None) -> None: """Fill an OTX signal from a raw response (separate so calibration can - re-score saved responses offline).""" - s = score_otx(data) + re-score saved responses offline, as of when they were taken).""" + s = score_otx(data, now) sig.threat, sig.confidence = s["threat"], s["confidence"] sig.last_seen = iso_date(s["newest"]) sig.link = f"https://otx.alienvault.com/indicator/ip/{ip}" diff --git a/tests/test_scoring.py b/tests/test_scoring.py index 8c28f30..5c6227d 100644 --- a/tests/test_scoring.py +++ b/tests/test_scoring.py @@ -1,3 +1,4 @@ +import json import tempfile import unittest from datetime import datetime, timedelta, timezone @@ -463,6 +464,69 @@ def test_saved_scans_recover_the_reporter_count(self): self.assertEqual(abuse.facts["reporters"], 7) +class RescoreAsOfTests(unittest.TestCase): + """PDX-28: a saved scan gets the same score whatever day it's rescored.""" + + PULSES = {"pulse_info": {"count": 3, "pulses": [ + {"author": {"username": f"a{i}"}, "name": "attack campaign", "modified": "2026-10-20T00:00:00+00:00", + "indicator_type_counts": {"IPv4": 20}} for i in range(3)]}, "validation": []} + + def rescore(self, saved, today, **kwargs): + """Run the real collect and evaluate with the clock set to `today`.""" + from paradox import calibrate, sources as src + import paradox.scoring as sc + + class Clock(datetime): + @classmethod + def now(cls, tz=None): + return today + + with tempfile.TemporaryDirectory() as d: + scans = Path(d) + (scans / "45.1.1.1.json").write_text(json.dumps(saved)) + with mock.patch.object(calibrate, "SCANS", scans), mock.patch.object(src, "datetime", Clock), \ + mock.patch.object(sc, "datetime", Clock): + signals, times, undated = calibrate.collect(["45.1.1.1"], 0, offline=True, **kwargs) + row = calibrate.evaluate({"malicious": ["45.1.1.1"], "benign": []}, signals, times) + return row["threat_mean"]["malicious"], row["paradox_mean"]["malicious"], undated + + def saved(self, **fields): + abuse = {"source": "abuseipdb", "name": "a", "tier": "fast", "threat": 0.9, "confidence": 0.8, + "last_seen": "2026-10-20", "facts": {"reporters": 5}, "summary": "x"} + return {"otx_raw": self.PULSES, "signals": [abuse], "otx_error": None, **fields} + + DAY1, DAY2 = datetime(2026, 10, 21, tzinfo=timezone.utc), datetime(2027, 6, 1, tzinfo=timezone.utc) + + def test_dated_scan_scores_the_same_on_any_day(self): + saved = self.saved(scanned_at="2026-10-21T00:00:00+00:00") + first, later = self.rescore(saved, self.DAY1), self.rescore(saved, self.DAY2) + self.assertEqual(first, later) + self.assertEqual(first[2], []) + + def test_undated_scan_is_reported_and_drifts(self): + first, later = self.rescore(self.saved(), self.DAY1), self.rescore(self.saved(), self.DAY2) + self.assertEqual(first[2], ["45.1.1.1"]) + self.assertNotEqual(first[:2], later[:2]) + + def test_as_of_fixes_an_undated_scan_without_hiding_it(self): + fixed = self.DAY1 + first = self.rescore(self.saved(), self.DAY1, undated_as_of=fixed) + later = self.rescore(self.saved(), self.DAY2, undated_as_of=fixed) + self.assertEqual(first, later) + self.assertEqual(later[2], ["45.1.1.1"]) + + def test_recorded_time_wins_over_as_of(self): + saved = self.saved(scanned_at="2026-10-21T00:00:00+00:00") + self.assertEqual(self.rescore(saved, self.DAY2)[:2], + self.rescore(saved, self.DAY2, undated_as_of=self.DAY2)[:2]) + + def test_new_scans_record_their_time(self): + from paradox import calibrate + with mock.patch.object(calibrate, "fetch_otx", return_value={}), mock.patch.object( + calibrate, "run_beacon", return_value=Signal("x", "x", "slow")): + self.assertIsNotNone(_sources.parse_dt(calibrate.scan_raw("45.1.1.1")["scanned_at"])) + + class ClaimTests(unittest.TestCase): def rdns(self, status="ok", **facts): s = sig("rdns", "context", None, 0.0, status=status)