spb/internetpressure

1"""Cross-checks the Python HMAC verifier against the Go agent's test vector (services/probe-agent README) and the2self-exclusion rules."""34from internetpressure.engine.health import local_failures5from internetpressure.ingest import auth67KEY = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f"8TS = "1789189804"9BODY = b'{"probe_id":"ca-qc-01","agent_version":"0.1.0","measurements":[]}'101112def test_go_vector_post():13    assert auth.body_sha256(BODY) == "9fd6962cedcf4a7aedb13c38f7f63137fdbbbf3703b447ebd761234981847c65"14    assert auth.sign(KEY, "POST", "/ingest/v1/batch", TS, BODY) == "111075ee8d4c20723598c8f75d5ad89433f0385ff965f123302d2b9411280723"151617def test_go_vector_get():18    assert auth.sign(KEY, "GET", "/ingest/v1/config", TS, b"") == "3798599e7f1bed2dab170d2aacf5f5288ec7cc591a64bd37285d3ea3c3450a8b"192021def test_verify_skew_and_bad_signature():22    sig = auth.sign(KEY, "POST", "/ingest/v1/batch", TS, BODY)23    assert auth.verify(KEY, "POST", "/ingest/v1/batch", TS, BODY, sig, now=int(TS) + 10) == (True, "ok")24    assert auth.verify(KEY, "POST", "/ingest/v1/batch", TS, BODY, sig, now=int(TS) + 1000)[1] == "skew"25    assert auth.verify(KEY, "POST", "/ingest/v1/batch", TS, BODY + b" ", sig, now=int(TS))[1] == "bad_signature"26    assert auth.verify(KEY, "POST", "/ingest/v1/batch", "abc", BODY, sig, now=int(TS))[1] == "bad_timestamp"27    assert auth.verify(KEY, "POST", "/ingest/v1/other", TS, BODY, sig, now=int(TS))[1] == "bad_signature"282930def _rows(spec):31    rows = []32    for probe, fail in spec.items():33        for i in range(10):34            rows.append({"probe_id": probe, "target_id": f"t{i}", "kind": "http", "n": 4, "ok_n": 0 if i < fail else 4})35    return rows363738def test_local_failure_excludes_only_the_broken_probe():39    rows = _rows({"a": 9, "b": 0, "c": 1})40    excluded, reasons, everyone = local_failures(rows, ["a", "b", "c"], 0.8)41    assert excluded == {"a"} and not everyone and "a" in reasons424344def test_everyone_failing_is_internal_not_internet():45    rows = _rows({"a": 10, "b": 9, "c": 10})46    excluded, _, everyone = local_failures(rows, ["a", "b", "c"], 0.8)47    assert excluded == set() and everyone  # cannot exclude anyone: freeze the instrument instead484950def test_stale_probes_ignored():51    rows = _rows({"a": 10, "b": 0})52    excluded, _, _ = local_failures(rows, ["b"], 0.8)  # "a" is not fresh → not considered53    assert excluded == set()54