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()