From d94e956ae0b9d010feed074b2370849937c52fdf Mon Sep 17 00:00:00 2001 From: blocksifrdev Date: Tue, 15 Sep 2026 12:16:47 -0400 Subject: [PATCH] Follow the v1.1 correction through the docs and the Python binding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three things a normative change is not finished without. 1. docs/security.md stated, as a mitigation against a compromised high-volume issuer: "Even if one issuer submits 1000 receipts, it cannot exceed 40% of the aggregate weight." That was not true of v1.0 — 50 receipts reached roughly 87% — so anyone who read that section received an assurance the algorithm did not provide. The claims now name the version they hold from, the residual-risk note is corrected, and a dated correction explains what was wrong and points at the vector that pins the fixed behaviour. A security guarantee that silently starts being true is worse than one that says when it started. 2. sdk/python/aggregate.py implements the same algorithm, so both bindings carry the normative one rather than only the JavaScript side. All eleven vectors run against it. 3. The parity gate now covers aggregation as well as the AGT bridge: 59 checks across both languages, including the capped-issuer case. Verified it catches drift — moving the Python cap to 0.45 fails with the divergence named. CI compiles and tests the new module. Co-Authored-By: Claude Opus 5 --- .github/workflows/ci.yml | 2 +- docs/security.md | 23 +++++-- scripts/check-agt-parity.mjs | 53 +++++++++++++--- sdk/python/README.md | 20 +++++- sdk/python/aggregate.py | 119 +++++++++++++++++++++++++++++++++++ sdk/python/test_aggregate.py | 66 +++++++++++++++++++ 6 files changed, 268 insertions(+), 15 deletions(-) create mode 100644 sdk/python/aggregate.py create mode 100644 sdk/python/test_aggregate.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bb3deef..65223b8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -28,7 +28,7 @@ jobs: test -f docs/ops/persistent-storage.md - name: Compile Python SDK - run: python -m py_compile sdk/python/client.py sdk/python/agt.py + run: python -m py_compile sdk/python/client.py sdk/python/agt.py sdk/python/aggregate.py - name: Test Python SDK run: python -m unittest discover -s sdk/python -p 'test_*.py' -v diff --git a/docs/security.md b/docs/security.md index bd02175..4bf8b43 100644 --- a/docs/security.md +++ b/docs/security.md @@ -33,7 +33,7 @@ When correctly deployed with conformant implementations: 8. **Stateless verifiability** — Verifiers can validate tokens without contacting the Trust Authority, eliminating dependency on TA availability at verification time. -9. **Single-issuer resistance** — The aggregation algorithm's issuer weight cap prevents any single issuer from dominating a trust score. +9. **Single-issuer resistance** — The aggregation algorithm's issuer weight cap prevents any single issuer from dominating a trust score. *(Holds as of aggregation algorithm **v1.1**. See the correction note in §3.)* ### 1.2 What TTP Does NOT Guarantee @@ -116,6 +116,11 @@ Compromise at a higher level in this hierarchy has broader impact. **Residual risk:** If an issuer's private key is compromised, the attacker can generate receipts up to the issuer weight cap (40%). Multi-issuer requirements for sensitive domains mitigate this. +> Under aggregation algorithm **v1.0 this bound did not hold**: capped weight was +> re-normalized back across all issuers, so a compromised high-volume issuer could reach +> roughly 87% of the aggregate weight while the others stayed light. Fixed in v1.1 — see +> the correction note below. + --- ### T4: Receipt Replay @@ -135,11 +140,21 @@ Compromise at a higher level in this hierarchy has broader impact. **Description:** An attacker controls or compromises a single high-volume issuer and uses it to inflate an agent's trust score. **Mitigations:** -- The aggregation algorithm caps any single issuer's contribution at `max_issuer_weight` (default: 40%). -- Even if one issuer submits 1000 receipts, it cannot exceed 40% of the aggregate weight. +- The aggregation algorithm caps any single issuer's contribution at `effective_cap = max(max_issuer_weight, 1 / issuer_count)` (default `max_issuer_weight`: 40%), and redistributes the excess to the **uncapped** issuers. +- Even if one issuer submits 1000 receipts, it cannot exceed that cap of the aggregate weight. - Multi-issuer requirements (`min_issuer_count ≥ 2`) prevent single-issuer tokens entirely for sensitive domains. -**Residual risk:** If only one issuer is deployed, that issuer's compromise fully controls agent trust. Deploy at least 2 independent issuers per domain. +**Residual risk:** If only one issuer is deployed, that issuer's compromise fully controls agent trust — the cap cannot bind when there is nobody to redistribute to (`effective_cap = 1/1`). Deploy at least 2 independent issuers per domain. + +> **Correction — aggregation v1.1 (2026-09-15).** The second mitigation above was not true +> of v1.0. That version capped a dominant issuer's *fraction* and then re-normalized +> across all issuers, which returned most of the capped weight to it whenever the others +> were light: 50 receipts from one issuer against one each from two others left the +> "capped" issuer holding **87%** of the weight, and the aggregate at 0.90 rather than +> 0.52. The guarantee stated in §1.1 item 9 and in this section holds from v1.1 onward. +> Anyone who deployed against v1.0 and relied on the 40% bound should re-evaluate; +> `protocol/test-vectors/aggregation-vectors.json` case `agg-009` pins the corrected +> behaviour. --- diff --git a/scripts/check-agt-parity.mjs b/scripts/check-agt-parity.mjs index 4810b73..05a8104 100644 --- a/scripts/check-agt-parity.mjs +++ b/scripts/check-agt-parity.mjs @@ -1,13 +1,15 @@ #!/usr/bin/env node -// The PCTR <-> AGT bridge exists twice: packages/pctr/src/agt.mjs for JavaScript and -// sdk/python/agt.py for Python, because AGT is Python-first. Two implementations agree -// only as long as something checks, so this runs both over the same inputs and fails on -// any divergence. Run: node scripts/check-agt-parity.mjs +// PCTR exists twice: JavaScript in packages/pctr/src, Python in sdk/python, because AGT +// is Python-first and the aggregation algorithm is normative for both. Two +// implementations agree only as long as something checks, so this runs both over the same +// inputs — AGT bridge and trust aggregation — and fails on any divergence. +// Run: node scripts/check-agt-parity.mjs import { execFileSync } from 'node:child_process'; import path from 'node:path'; import { fileURLToPath } from 'node:url'; import { classifyAction } from '../packages/pctr/src/consequences.mjs'; +import { aggregateTrust } from '../packages/pctr/src/aggregate.mjs'; import { trustTier, toAgtScore, ringForSeverity, domainFor, normalizeAgtEvent } from '../packages/pctr/src/agt.mjs'; const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..'); @@ -21,6 +23,29 @@ const ACTIONS = [ const SCORES = [0, 0.1, 0.29, 0.3, 0.6, 0.849, 0.85, 0.9178, 1]; const SEVERITIES = ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW', '?']; const CONSEQUENCES = ['MONEY_MOVED', 'DATA_DELETED', 'SECRET_EXPOSED', 'SOMETHING_NEW']; +// Aggregation is implemented in both languages against the same normative spec, so the +// two must agree on scores, not merely each pass the vectors alone. +const NOW = 1_700_000_000_000; +const AGGREGATION = { + 'single-perfect': [{ receipt_id: 'a', issuer_id: 'A', score: 1.0, timestamp: NOW }], + 'single-zero': [{ receipt_id: 'a', issuer_id: 'A', score: 0.0, timestamp: NOW }], + 'two-issuers-divergent': [ + { receipt_id: 'a', issuer_id: 'A', score: 1.0, timestamp: NOW }, + { receipt_id: 'b', issuer_id: 'B', score: 0.0, timestamp: NOW }], + 'dominant-issuer-capped': [ + ...Array.from({ length: 50 }, (_, i) => ({ receipt_id: `l${i}`, issuer_id: 'LOUD', score: 1.0, timestamp: NOW })), + { receipt_id: 'b', issuer_id: 'B', score: 0.2, timestamp: NOW }, + { receipt_id: 'c', issuer_id: 'C', score: 0.2, timestamp: NOW }], + 'decayed': [ + { receipt_id: 'fresh', issuer_id: 'A', score: 1.0, timestamp: NOW }, + { receipt_id: 'old', issuer_id: 'A', score: 0.0, timestamp: NOW - 120_000 }], + 'outside-window': [{ receipt_id: 'a', issuer_id: 'A', score: 1.0, timestamp: NOW - 400_000 }], + 'three-issuers': [ + { receipt_id: 'a', issuer_id: 'A', score: 0.95, timestamp: NOW }, + { receipt_id: 'b', issuer_id: 'B', score: 0.92, timestamp: NOW - 10_000 }, + { receipt_id: 'c', issuer_id: 'C', score: 0.88, timestamp: NOW - 20_000 }] +}; + const EVENTS = [ ['policy-allow', { allowed: true, action: 'allow', agentId: 'f', operation: 'reports.read', approvers: [], rateLimited: false }], ['policy-warn', { allowed: true, action: 'warn', agentId: 'f', operation: 'reports.read', approvers: [], rateLimited: false }], @@ -42,7 +67,7 @@ const EVENTS = [ const key = (score) => (Number.isInteger(score) ? String(score) : String(score)); function javascript() { - const out = { classify: {}, tiers: {}, scores: {}, rings: {}, domains: {}, events: {} }; + const out = { classify: {}, tiers: {}, scores: {}, rings: {}, domains: {}, events: {}, aggregation: {} }; for (const [action, params] of ACTIONS) { const c = classifyAction(action, params); out.classify[`${action}|${JSON.stringify(params)}`] = [c.consequence, c.severity, c.reversible]; @@ -54,6 +79,10 @@ function javascript() { const r = normalizeAgtEvent(event); out.events[name] = r ? [r.event, r.subject ?? null] : null; } + for (const [name, receipts] of Object.entries(AGGREGATION)) { + const r = aggregateTrust(receipts, NOW); + out.aggregation[name] = r.error ? [r.error] : [Number(r.score.toFixed(6)), r.contributing_issuers]; + } return out; } @@ -61,14 +90,17 @@ const PY = ` import json, sys sys.path.insert(0, ${JSON.stringify(path.join(root, 'sdk', 'python'))}) from agt import classify_action, trust_tier, to_agt_score, ring_for_severity, domain_for, normalize_agt_event +from aggregate import aggregate_trust actions = json.loads(${JSON.stringify(JSON.stringify(ACTIONS))}) scores = json.loads(${JSON.stringify(JSON.stringify(SCORES))}) severities = json.loads(${JSON.stringify(JSON.stringify(SEVERITIES))}) consequences = json.loads(${JSON.stringify(JSON.stringify(CONSEQUENCES))}) events = json.loads(${JSON.stringify(JSON.stringify(EVENTS))}) +aggregation = json.loads(${JSON.stringify(JSON.stringify(AGGREGATION))}) +now = ${NOW} -out = {'classify': {}, 'tiers': {}, 'scores': {}, 'rings': {}, 'domains': {}, 'events': {}} +out = {'classify': {}, 'tiers': {}, 'scores': {}, 'rings': {}, 'domains': {}, 'events': {}, 'aggregation': {}} for action, params in actions: c = classify_action(action, params) out['classify'][action + '|' + json.dumps(params, separators=(',', ':'))] = [c['consequence'], c['severity'], c['reversible']] @@ -83,6 +115,9 @@ for consequence in consequences: for name, event in events: r = normalize_agt_event(event) out['events'][name] = [r['event'], r.get('subject')] if r else None +for name, receipts in aggregation.items(): + r = aggregate_trust(receipts, now) + out['aggregation'][name] = [r['error']] if r.get('error') else [round(r['score'], 6), r['contributing_issuers']] print(json.dumps(out)) `; @@ -107,9 +142,9 @@ for (const section of Object.keys(js)) { } if (differences.length) { - console.error(`PCTR/AGT bridge parity FAILED — ${differences.length} of ${checks} checks differ:\n`); + console.error(`PCTR parity FAILED — ${differences.length} of ${checks} checks differ:\n`); console.error(differences.join('\n')); - console.error('\npackages/pctr/src/agt.mjs and sdk/python/agt.py must agree. Fix both.'); + console.error('\nThe JavaScript and Python implementations must agree. Fix both.'); process.exit(1); } -console.log(`PCTR/AGT bridge parity OK — JavaScript and Python agree across ${checks} checks.`); +console.log(`PCTR parity OK — JavaScript and Python agree across ${checks} checks (AGT bridge + trust aggregation).`); diff --git a/sdk/python/README.md b/sdk/python/README.md index b3306d9..5a4e0c9 100644 --- a/sdk/python/README.md +++ b/sdk/python/README.md @@ -63,7 +63,25 @@ verified 0.85 — `to_agt_trust_score()` returns AGT's `TrustScore {overall, dim tier}`. The 0-1000 integer scale (`to_agt_score()`) is only for downstream consumers that ask for it. +## Trust aggregation (`aggregate.py`) + +The normative algorithm from +[`protocol/aggregation-spec.md`](../../protocol/aggregation-spec.md) **v1.1** — time +decay, negative-signal amplification, and issuer weight capping that redistributes to the +uncapped issuers. The same eleven test vectors run against both bindings. + +```python +from aggregate import aggregate_trust, score_label + +result = aggregate_trust(receipts, current_time_ms) +print(result["score"], score_label(result["score"])) +``` + +An empty receipt window returns `INSUFFICIENT_TRUST_DATA` with `score: None` — absent +evidence is never a score. + The two implementations are held in step by `scripts/check-agt-parity.mjs`, which runs -both over the same corpus in CI and fails on any divergence. +both over the same corpus in CI — AGT bridge and trust aggregation, 59 checks — and fails +on any divergence. Tests: `python3 -m unittest discover -s sdk/python -p 'test_*.py'` diff --git a/sdk/python/aggregate.py b/sdk/python/aggregate.py new file mode 100644 index 0000000..8a0a3c9 --- /dev/null +++ b/sdk/python/aggregate.py @@ -0,0 +1,119 @@ +"""TTP trust score aggregation — the normative algorithm. + +Implements protocol/aggregation-spec.md v1.1 step for step, mirroring +packages/pctr/src/aggregate.mjs. The same eleven test vectors run against both +bindings (scripts/check-agt-parity.mjs), so they cannot drift apart. + +Two properties, both deliberate: + - Negative signals weigh more (default 1.5x), so an agent behaving well most of the + time cannot average away a few dangerous actions. + - No single issuer may contribute more than effective_cap of the score, and the excess + goes to the *uncapped* issuers. v1.0 re-normalized across everyone, which handed the + excess straight back whenever the other issuers were light. +""" + +from __future__ import annotations + +import math +from typing import Any, Dict, List, Mapping, Optional + +__all__ = ["DEFAULT_PARAMS", "INSUFFICIENT_TRUST_DATA", "aggregate_trust", "score_label"] + +DEFAULT_PARAMS: Dict[str, float] = { + "receipt_window_s": 300, + "max_issuer_weight": 0.40, + "negative_weight_multiplier": 1.5, + "decay_half_life_s": 120, +} + +INSUFFICIENT_TRUST_DATA = "INSUFFICIENT_TRUST_DATA" + + +def aggregate_trust(receipts: Optional[List[Mapping[str, Any]]] = None, + current_time_ms: Optional[int] = None, + params: Optional[Mapping[str, float]] = None) -> Dict[str, Any]: + """Aggregate behavioural receipts into a trust score in [0.0, 1.0].""" + p = {**DEFAULT_PARAMS, **(params or {})} + receipts = list(receipts or []) + + # Step 1 — filter to the receipt window. + window = [r for r in receipts + if current_time_ms - r["timestamp"] <= p["receipt_window_s"] * 1000] + if not window: + return {"error": INSUFFICIENT_TRUST_DATA, "score": None, + "contributing_receipts": 0, "contributing_issuers": 0} + + # Steps 2 and 3 — time decay, then negative signal amplification. + weighted = [] + for r in window: + age_s = (current_time_ms - r["timestamp"]) / 1000 + decay = math.exp(-math.log(2) * age_s / p["decay_half_life_s"]) + negative = r["score"] < 0.5 + weighted.append({**r, "age_s": age_s, + "signal_weight": decay * p["negative_weight_multiplier"] if negative else decay}) + + # Step 4 — per-issuer weighted score. + by_issuer: Dict[str, Dict[str, float]] = {} + for r in weighted: + e = by_issuer.setdefault(r["issuer_id"], {"weighted_sum": 0.0, "total_weight": 0.0}) + e["weighted_sum"] += r["score"] * r["signal_weight"] + e["total_weight"] += r["signal_weight"] + + issuers = [{"issuer_id": k, "issuer_score": e["weighted_sum"] / e["total_weight"], + "issuer_raw_weight": e["total_weight"]} + for k, e in by_issuer.items()] + + # Step 5 — cap, then water-fill the excess onto the uncapped issuers. A cap below + # 1/n is infeasible, so that is the floor on the cap actually applied. + total_raw = sum(i["issuer_raw_weight"] for i in issuers) + effective_cap = max(p["max_issuer_weight"], 1 / len(issuers)) + for i in issuers: + i["weight"] = i["issuer_raw_weight"] / total_raw + i["capped"] = False + + for _ in range(len(issuers) + 1): + over = [i for i in issuers if not i["capped"] and i["weight"] > effective_cap + 1e-12] + if not over: + break + excess = 0.0 + for i in over: + excess += i["weight"] - effective_cap + i["weight"] = effective_cap + i["capped"] = True + free = [i for i in issuers if not i["capped"]] + free_total = sum(i["weight"] for i in free) + if not free or free_total == 0: + break + for i in free: + i["weight"] += excess * (i["weight"] / free_total) + + # Step 6 — combine, and clamp for floating point. + raw_score = sum(i["issuer_score"] * i["weight"] for i in issuers) + + return { + "score": max(0.0, min(1.0, raw_score)), + "contributing_receipts": len(window), + "contributing_issuers": len(issuers), + "oldest_receipt_age_s": round(max(r["age_s"] for r in weighted)), + "issuers": [{"issuer_id": i["issuer_id"], + "issuer_score": round(i["issuer_score"], 6), + "weight": round(i["weight"], 6), + "capped": i["capped"]} for i in issuers], + } + + +# The scale from protocol/scoring-semantics.md, so a score can be read in words. +def score_label(score: Optional[float]) -> str: + if score is None: + return "Unknown" + if score >= 0.90: + return "Excellent" + if score >= 0.70: + return "Good" + if score >= 0.50: + return "Marginal" + if score >= 0.30: + return "Poor" + if score >= 0.10: + return "Bad" + return "Critical" diff --git a/sdk/python/test_aggregate.py b/sdk/python/test_aggregate.py new file mode 100644 index 0000000..71a6917 --- /dev/null +++ b/sdk/python/test_aggregate.py @@ -0,0 +1,66 @@ +"""The normative aggregation vectors, run against the Python binding. + +The same vectors run against the JavaScript binding in +packages/pctr/tests/aggregation.test.mjs. Both must agree with the spec and each other. +""" +import json +import os +import unittest + +from aggregate import DEFAULT_PARAMS, INSUFFICIENT_TRUST_DATA, aggregate_trust, score_label + +VECTORS = json.load(open(os.path.join(os.path.dirname(__file__), "..", "..", + "protocol", "test-vectors", "aggregation-vectors.json"))) + + +class ConformanceTests(unittest.TestCase): + def test_defaults_match_the_spec(self): + self.assertEqual(DEFAULT_PARAMS, VECTORS["params"]) + + def test_every_vector(self): + for case in VECTORS["cases"]: + with self.subTest(case=case["id"]): + result = aggregate_trust(case["receipts"], case["current_time_ms"], VECTORS["params"]) + if case["expected"].get("error"): + self.assertEqual(result["error"], case["expected"]["error"]) + continue + self.assertAlmostEqual(result["score"], case["expected"]["score"], delta=0.001, + msg=f'{case["id"]}: {case["description"]}') + if "contributing_receipts" in case["expected"]: + self.assertEqual(result["contributing_receipts"], case["expected"]["contributing_receipts"]) + if "contributing_issuers" in case["expected"]: + self.assertEqual(result["contributing_issuers"], case["expected"]["contributing_issuers"]) + + +class PropertyTests(unittest.TestCase): + def test_empty_window_is_insufficient_data_not_zero(self): + result = aggregate_trust([{"receipt_id": "r", "issuer_id": "A", "score": 1.0, "timestamp": 0}], 10_000_000) + self.assertEqual(result["error"], INSUFFICIENT_TRUST_DATA) + self.assertIsNone(result["score"]) + + def test_a_capped_issuer_holds_its_cap_and_no_more(self): + now = 1_700_000_000_000 + receipts = [{"receipt_id": f"loud{i}", "issuer_id": "LOUD", "score": 1.0, "timestamp": now} for i in range(50)] + receipts += [{"receipt_id": "b", "issuer_id": "B", "score": 0.2, "timestamp": now}, + {"receipt_id": "c", "issuer_id": "C", "score": 0.2, "timestamp": now}] + result = aggregate_trust(receipts, now) + loud = next(i for i in result["issuers"] if i["issuer_id"] == "LOUD") + self.assertTrue(loud["capped"]) + self.assertAlmostEqual(loud["weight"], 0.4, delta=0.001) + self.assertLess(result["score"], 0.55) + self.assertAlmostEqual(sum(i["weight"] for i in result["issuers"]), 1.0, delta=0.001) + + def test_danger_cannot_be_averaged_away(self): + now = 1_700_000_000_000 + receipts = [{"receipt_id": f"g{i}", "issuer_id": "A", "score": 1.0, "timestamp": now} for i in range(9)] + receipts.append({"receipt_id": "bad", "issuer_id": "A", "score": 0.0, "timestamp": now}) + self.assertLess(aggregate_trust(receipts, now)["score"], 0.87) + + def test_labels(self): + self.assertEqual(score_label(0.95), "Excellent") + self.assertEqual(score_label(0.05), "Critical") + self.assertEqual(score_label(None), "Unknown") + + +if __name__ == "__main__": + unittest.main()