From 52e7c315a71ab63c85f8e368850608bb305a0e04 Mon Sep 17 00:00:00 2001 From: blocksifrdev Date: Tue, 15 Sep 2026 09:38:37 -0400 Subject: [PATCH] Add the Python AGT bridge and fix the trust scale in the guide MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AGT is Python-first (microsoft/agent-governance-toolkit ships Python, TypeScript, Go, Rust and .NET), so a JavaScript-only bridge would miss where most integrations actually live. sdk/python/agt.py mirrors packages/pctr/src/agt.mjs: consequence classification, AGT TrustScore and tiers, ExecutionRing mapping, SPIFFE ids, input.ttp claims for Rego, behavioural evidence, AgentMesh attestations, and event normalization over AGT's real shapes. Two implementations agree only as long as something checks, so scripts/check-agt-parity.mjs runs both over one corpus — 52 checks across classification, tiers, scores, rings, domains and event normalization — and fails the build on any divergence. Verified it catches drift: moving the Python "trusted" threshold from 0.6 to 0.65 fails with exit 1 and names the check. Also corrects docs/integration-guide.md 6.4 and docs/ecosystem-integrations.md, which specified agt_trust_score = round(ttp_score * 1000). Upstream AGT scores 0-1, banded untrusted 0.0 / provisional 0.30 / trusted 0.60 / verified 0.85 (agent-governance-typescript/src/trust.ts), so sending 918 where AGT expects 0.918 would read as Verified for every agent. The guide now states the real mapping, keeps the 0-1000 scale only for downstream consumers that ask for it, and adds 6.4.1 on mapping consequence severity onto ExecutionRing rather than duplicating it. CI now compiles agt.py, runs the 19 Python tests, and enforces parity. Co-Authored-By: Claude Opus 5 --- .github/workflows/ci.yml | 8 +- docs/ecosystem-integrations.md | 3 +- docs/integration-guide.md | 39 ++- packages/pctr/README.md | 5 + scripts/check-agt-parity.mjs | 115 +++++++++ sdk/python/README.md | 30 +++ sdk/python/__init__.py | 26 ++ sdk/python/agt.py | 446 +++++++++++++++++++++++++++++++++ sdk/python/test_agt.py | 198 +++++++++++++++ 9 files changed, 863 insertions(+), 7 deletions(-) create mode 100644 scripts/check-agt-parity.mjs create mode 100644 sdk/python/agt.py create mode 100644 sdk/python/test_agt.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b88ba0e..bb3deef 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -28,7 +28,13 @@ jobs: test -f docs/ops/persistent-storage.md - name: Compile Python SDK - run: python -m py_compile sdk/python/client.py + run: python -m py_compile sdk/python/client.py sdk/python/agt.py + + - name: Test Python SDK + run: python -m unittest discover -s sdk/python -p 'test_*.py' -v + + - name: Check PCTR/AGT cross-language parity + run: node scripts/check-agt-parity.mjs - name: Build TypeScript SDK run: npm run build:sdk diff --git a/docs/ecosystem-integrations.md b/docs/ecosystem-integrations.md index e73b9e7..5710fb9 100644 --- a/docs/ecosystem-integrations.md +++ b/docs/ecosystem-integrations.md @@ -14,7 +14,8 @@ Use TTP as the behavioral evidence layer in a closed loop: Key patterns: - OPA/Rego bridge (`input.ttp` claims) - SPIFFE/SVID identity compatibility -- Canonical score adapter: `agt_trust_score = round(ttp_score * 1000)` +- Trust score mapping: AGT's `TrustScore` is **0-1 with tiers** (untrusted 0.0 / provisional 0.30 / trusted 0.60 / verified 0.85), so a TTP score maps across unscaled. The 0-1000 integer scale is only for downstream components that ask for it. +- Execution rings: map consequence severity onto AGT's `ExecutionRing` rather than building a parallel privilege model - AgentMesh trust attestation bridge See full details in [integration-guide.md#part-6-agt-native-integration-recommended-priority](integration-guide.md#part-6-agt-native-integration-recommended-priority). diff --git a/docs/integration-guide.md b/docs/integration-guide.md index 45468ce..66a51fc 100644 --- a/docs/integration-guide.md +++ b/docs/integration-guide.md @@ -404,18 +404,47 @@ Benefits: - Immediate compatibility with SPIFFE-based workload identity. - TTP augments SPIFFE identity with behavioral trust. -### 6.4 Canonical Score Adapter: TTP -> AGT +### 6.4 Trust Score Mapping: TTP -> AGT -When downstream AGT components expect a 0-1000 trust scale, use the canonical mapping: +**AGT scores trust on 0-1, not 0-1000.** Upstream +[`microsoft/agent-governance-toolkit`](https://github.com/microsoft/agent-governance-toolkit) +defines `TrustScore` as `{ overall, dimensions, tier }` with `overall` in `[0.0, 1.0]`, +banded into tiers (`agent-governance-typescript/src/trust.ts`): + +| Tier | Threshold | +| --- | --- | +| `Untrusted` | 0.0 | +| `Provisional` | 0.30 | +| `Trusted` | 0.60 | +| `Verified` | 0.85 | + +A TTP score is already `[0.0, 1.0]`, so it maps across **unscaled**: ```text -agt_trust_score = round(ttp_score * 1000) +trust_score = { overall: ttp_score, dimensions: {...}, tier: tier_for(ttp_score) } ``` Reference adapter behavior: - Input: `ttp_score` in `[0.0, 1.0]`. -- Output: integer `agt_trust_score` in `[0, 1000]`. -- Preserve original `ttp_score` in logs/telemetry for auditability. +- Output: an AGT `TrustScore` whose `overall` is the same number and whose `tier` comes + from the thresholds above. +- Preserve the original `ttp_score` and the issuing evidence in logs/telemetry for + auditability. + +> **Earlier versions of this guide specified `agt_trust_score = round(ttp_score * 1000)`. +> That is wrong against upstream AGT** — sending `918` where AGT expects `0.918` puts +> every agent off the top of the scale and reads as `Verified`. The 0-1000 integer scale +> applies only to downstream components that explicitly ask for it; it is not the +> AGT-native path. Implementations: `toAgtTrustScore()` / `toAgtScore()` in +> `packages/pctr/src/agt.mjs`, `to_agt_trust_score()` / `to_agt_score()` in +> `sdk/python/agt.py`. + +### 6.4.1 Execution Rings + +AGT's `ExecutionRing` (Ring0 most privileged through Ring3) is the privilege construct to +map into rather than duplicate. PCTR proposes a required ring from the consequence an +action can cause — `CRITICAL -> Ring0`, `HIGH -> Ring1`, `MEDIUM -> Ring2`, `LOW -> Ring3` +— and AGT's own `actionRings` configuration stays authoritative. ### 6.5 AgentMesh / Peer Trust Attestation Bridge diff --git a/packages/pctr/README.md b/packages/pctr/README.md index 8051c9f..ab9671c 100644 --- a/packages/pctr/README.md +++ b/packages/pctr/README.md @@ -342,6 +342,11 @@ ran. Worked end to end: `npm run demo:pctr-agt`. +**Python.** AGT is Python-first, so the bridge exists there too — `sdk/python/agt.py`, +same functions, same behaviour. `scripts/check-agt-parity.mjs` runs both implementations +over one corpus in CI and fails the build on any divergence, so the bindings cannot drift +apart. + ## Relationship to TTP PCTR answers *which path through multiple agents is trustworthy enough to reach this diff --git a/scripts/check-agt-parity.mjs b/scripts/check-agt-parity.mjs new file mode 100644 index 0000000..4810b73 --- /dev/null +++ b/scripts/check-agt-parity.mjs @@ -0,0 +1,115 @@ +#!/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 +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 { trustTier, toAgtScore, ringForSeverity, domainFor, normalizeAgtEvent } from '../packages/pctr/src/agt.mjs'; + +const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..'); + +// One shared corpus, exercised by both implementations. +const ACTIONS = [ + ['payments.status', {}], ['payments.transfer', {}], ['payments.transfer', { amount: 18000 }], + ['customers.delete', {}], ['records.update', { recordsAffected: 5000 }], ['send_invoice_email', {}], + ['reports.read', {}], ['cluster.deploy', {}], ['vault.read', {}], ['nothing.special', {}] +]; +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']; +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 }], + ['policy-deny', { allowed: false, action: 'deny', agentId: 'f', operation: 'payments.transfer', approvers: [], rateLimited: false }], + ['policy-approval', { allowed: true, action: 'require_approval', agentId: 'f', operation: 'payments.transfer', approvers: ['ops'], rateLimited: false }], + ['audit-review', { timestamp: 't', agentId: 'f', action: 'customers.delete', decision: 'review', hash: 'h1', previousHash: 'h0' }], + ['audit-allow', { timestamp: 't', agentId: 'f', action: 'reports.read', decision: 'allow', hash: 'h1', previousHash: 'h0' }], + ['cascade-quarantine', { eventId: 'e1', timestamp: 't', sourceAgentId: 'f', affectedAgentIds: ['a'], action: 'agent_quarantined', reason: 'b', blastRadius: 3 }], + ['cascade-health', { eventId: 'e2', timestamp: 't', sourceAgentId: 'f', affectedAgentIds: [], action: 'health_propagated' }], + ['ring-violation', { agentId: 'f', action: 'prod.deploy', agentRing: 2, requiredRing: 0, message: 'm' }], + ['trust-verify', { verified: false, agentId: 'f', trustScore: { overall: 0.42, dimensions: {}, tier: 'Provisional' } }], + ['invocation', { type: 'action.invocation', agentId: 'f', action: 'payments.transfer', parameters: { amount: 18000 } }], + ['registration', { type: 'agent.registered', agentId: 'f', trustScore: { overall: 0.97, dimensions: {}, tier: 'Verified' } }], + ['delegation', { type: 'task.delegated', agentId: 'f', to: 'g' }], + ['junk', { type: 'telemetry.heartbeat', agentId: 'f' }], + ['malformed', { nothing: 'useful' }] +]; + +const key = (score) => (Number.isInteger(score) ? String(score) : String(score)); + +function javascript() { + const out = { classify: {}, tiers: {}, scores: {}, rings: {}, domains: {}, events: {} }; + for (const [action, params] of ACTIONS) { + const c = classifyAction(action, params); + out.classify[`${action}|${JSON.stringify(params)}`] = [c.consequence, c.severity, c.reversible]; + } + for (const score of SCORES) { out.tiers[key(score)] = trustTier(score); out.scores[key(score)] = toAgtScore(score); } + for (const severity of SEVERITIES) out.rings[severity] = ringForSeverity(severity); + for (const consequence of CONSEQUENCES) out.domains[consequence] = domainFor(consequence); + for (const [name, event] of EVENTS) { + const r = normalizeAgtEvent(event); + out.events[name] = r ? [r.event, r.subject ?? null] : null; + } + return out; +} + +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 + +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))}) + +out = {'classify': {}, 'tiers': {}, 'scores': {}, 'rings': {}, 'domains': {}, 'events': {}} +for action, params in actions: + c = classify_action(action, params) + out['classify'][action + '|' + json.dumps(params, separators=(',', ':'))] = [c['consequence'], c['severity'], c['reversible']] +for score in scores: + k = str(int(score)) if float(score).is_integer() else str(score) + out['tiers'][k] = trust_tier(score) + out['scores'][k] = to_agt_score(score) +for severity in severities: + out['rings'][severity] = ring_for_severity(severity) +for consequence in consequences: + out['domains'][consequence] = domain_for(consequence) +for name, event in events: + r = normalize_agt_event(event) + out['events'][name] = [r['event'], r.get('subject')] if r else None +print(json.dumps(out)) +`; + +function python() { + const stdout = execFileSync('python3', ['-c', PY], { encoding: 'utf8', cwd: root }); + return JSON.parse(stdout); +} + +const js = javascript(); +const py = python(); + +const differences = []; +let checks = 0; +for (const section of Object.keys(js)) { + for (const [name, value] of Object.entries(js[section])) { + checks++; + const other = py[section]?.[name]; + if (JSON.stringify(value) !== JSON.stringify(other)) { + differences.push(` ${section}.${name}\n js: ${JSON.stringify(value)}\n python: ${JSON.stringify(other)}`); + } + } +} + +if (differences.length) { + console.error(`PCTR/AGT bridge 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.'); + process.exit(1); +} +console.log(`PCTR/AGT bridge parity OK — JavaScript and Python agree across ${checks} checks.`); diff --git a/sdk/python/README.md b/sdk/python/README.md index 12f0644..b3306d9 100644 --- a/sdk/python/README.md +++ b/sdk/python/README.md @@ -37,3 +37,33 @@ print(resp.decision.value, resp.mode.value, resp.receipt.receiptId) - Enforce both `decision` and `mode`. - Treat missing receipt as deny. - Persist receipt integrity values for audit traceability. + +## AGT bridge (`agt.py`) + +PCTR <-> [Microsoft AGT](https://github.com/microsoft/agent-governance-toolkit) for +Python, mirroring `packages/pctr/src/agt.mjs`. AGT is Python-first, so this is where +most real integrations sit. + +```python +from agt import agt_claims, normalize_agt_event, to_trust_evidence, to_agt_trust_score + +# 1. AGT's runtime events -> canonical PCTR events (None when unrecognised) +event = normalize_agt_event(agt_policy_decision) + +# 2. What PCTR hands AGT's policy engine, under input.ttp +claims = agt_claims(ttp_score=0.9178, action="payments.transfer", issuer_count=2, + agents=["spiffe://blocksifr.com/ns/prod/sa/finance"]) + +# 3. Evidence back, so AGT decides better next time +evidence = to_trust_evidence(receipt) +``` + +**AGT trust is 0-1**, banded untrusted 0.0 / provisional 0.30 / trusted 0.60 / +verified 0.85 — `to_agt_trust_score()` returns AGT's `TrustScore {overall, dimensions, +tier}`. The 0-1000 integer scale (`to_agt_score()`) is only for downstream consumers that +ask for it. + +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. + +Tests: `python3 -m unittest discover -s sdk/python -p 'test_*.py'` diff --git a/sdk/python/__init__.py b/sdk/python/__init__.py index ac2f1f8..099dbbc 100644 --- a/sdk/python/__init__.py +++ b/sdk/python/__init__.py @@ -1,3 +1,17 @@ +from .agt import ( + AGT_TIER_THRESHOLDS, + TrustScore, + agt_claims, + classify_action, + domain_for, + normalize_agt_event, + parse_spiffe_id, + ring_for_severity, + to_agt_trust_score, + to_mesh_attestation, + to_trust_evidence, + trust_tier, +) from .client import ( AuthorityGrant, AuthorizeRequest, @@ -11,6 +25,7 @@ ) __all__ = [ + "AGT_TIER_THRESHOLDS", "AuthorityGrant", "AuthorizeRequest", "AuthorizeResponse", @@ -19,5 +34,16 @@ "Principal", "Receipt", "Resource", + "TrustScore", + "agt_claims", "authorize", + "classify_action", + "domain_for", + "normalize_agt_event", + "parse_spiffe_id", + "ring_for_severity", + "to_agt_trust_score", + "to_mesh_attestation", + "to_trust_evidence", + "trust_tier", ] diff --git a/sdk/python/agt.py b/sdk/python/agt.py new file mode 100644 index 0000000..051a374 --- /dev/null +++ b/sdk/python/agt.py @@ -0,0 +1,446 @@ +"""PCTR <-> Microsoft AGT bridge for Python. + +AGT (https://github.com/microsoft/agent-governance-toolkit) is Python-first, so this +mirrors ``packages/pctr/src/agt.mjs`` for integrations that live on that side. + +Types follow the toolkit's own definitions in +``agent-governance-typescript/src/types.ts`` — PolicyAction, PolicyDecisionResult, +AuditEntry, TrustScore, TrustTier, ExecutionRing, CascadeEvent, RingViolation. + +The loop (docs/integration-guide.md Part 6): + +1. AGT enforces pre-execution policy +2. PCTR observes the consequence, the route and the signed receipt +3. Trust is recomputed from that evidence +4. AGT consumes it and adjusts the next decision + +AGT decides allow/deny. PCTR reports evidence and never a competing verdict. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Dict, List, Mapping, Optional + +__all__ = [ + "AGT_TIER_THRESHOLDS", "RING_BY_SEVERITY", "TrustScore", "trust_tier", + "to_agt_trust_score", "from_agt_trust_score", "to_agt_score", "from_agt_score", + "ring_for_severity", "is_spiffe_id", "parse_spiffe_id", "classify_action", + "domain_for", "agt_claims", "to_trust_evidence", "to_mesh_attestation", + "normalize_agt_event", +] + +# ── Trust ──────────────────────────────────────────────────────────────────── +# AGT's TrustScore is 0-1, banded into tiers. Thresholds are the toolkit's own +# defaults (agent-governance-typescript/src/trust.ts). A TTP score is already 0-1, +# so it maps across unscaled — that is the point of sharing a scale. +AGT_TIER_THRESHOLDS: Dict[str, float] = { + "untrusted": 0.0, "provisional": 0.3, "trusted": 0.6, "verified": 0.85, +} + + +def _clamp01(value: Any) -> float: + try: + return max(0.0, min(1.0, float(value))) + except (TypeError, ValueError): + return 0.0 + + +def trust_tier(score: Any, thresholds: Optional[Mapping[str, float]] = None) -> str: + """Return AGT's TrustTier for a 0-1 score.""" + t = dict(AGT_TIER_THRESHOLDS if thresholds is None else thresholds) + value = _clamp01(score) + if value >= t["verified"]: + return "Verified" + if value >= t["trusted"]: + return "Trusted" + if value >= t["provisional"]: + return "Provisional" + return "Untrusted" + + +@dataclass +class TrustScore: + """AGT's TrustScore: ``{ overall, dimensions, tier }``.""" + + overall: float + dimensions: Dict[str, float] = field(default_factory=dict) + tier: str = "" + + def __post_init__(self) -> None: + self.overall = _clamp01(self.overall) + if not self.tier: + self.tier = trust_tier(self.overall) + + def as_dict(self) -> Dict[str, Any]: + return {"overall": self.overall, "dimensions": self.dimensions, "tier": self.tier} + + +def to_agt_trust_score(ttp_score: Any, dimensions: Optional[Mapping[str, float]] = None, + thresholds: Optional[Mapping[str, float]] = None) -> TrustScore: + overall = _clamp01(ttp_score) + return TrustScore(overall=overall, dimensions=dict(dimensions or {}), + tier=trust_tier(overall, thresholds)) + + +def from_agt_trust_score(trust_score: Any) -> float: + """Accept a TrustScore, a mapping, or a bare number.""" + if isinstance(trust_score, TrustScore): + return trust_score.overall + if isinstance(trust_score, Mapping): + return _clamp01(trust_score.get("overall", 0)) + return _clamp01(trust_score) + + +# The 0-1000 integer scale from integration-guide.md 6.4. Upstream AGT does NOT use it; +# this is only for downstream components that explicitly expect that scale. +def to_agt_score(ttp_score: Any) -> int: + return round(_clamp01(ttp_score) * 1000) + + +def from_agt_score(agt_score: Any) -> float: + try: + return _clamp01(float(agt_score) / 1000) + except (TypeError, ValueError): + return 0.0 + + +# AGT's ExecutionRing: Ring0 is the most privileged. PCTR proposes a ring from what the +# action can cause; AGT's own actionRings configuration stays authoritative. +RING_BY_SEVERITY: Dict[str, int] = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3} + + +def ring_for_severity(severity: Optional[str]) -> int: + return RING_BY_SEVERITY.get(severity or "", 3) + + +# ── SPIFFE ─────────────────────────────────────────────────────────────────── +_SPIFFE = re.compile(r"^spiffe://([^/]+)(/.*)?$") + + +def is_spiffe_id(agent_id: Any) -> bool: + return bool(_SPIFFE.match(str(agent_id or ""))) + + +def parse_spiffe_id(agent_id: Any) -> Optional[Dict[str, str]]: + match = _SPIFFE.match(str(agent_id or "")) + if not match: + return None + return {"trust_domain": match.group(1), "path": match.group(2) or "", "id": str(agent_id)} + + +# ── Consequences ───────────────────────────────────────────────────────────── +# Mirrors packages/pctr/src/consequences.mjs. The verb decides over the namespace: +# payments.status reads, payments.transfer moves money. +_READ_VERB = re.compile( + r"(?:^|[./_:-])(get|list|read|status|search|query|fetch|describe|show|view|count|preview)$", + re.IGNORECASE) + +_RULES = [ + (re.compile(r"payment|transfer|payout|refund|charge|wire|treasury|stripe|ledger\.move|\bpay\b", re.I), "MONEY_MOVED"), + (re.compile(r"delete|destroy|drop|purge|truncate|wipe|erase|remove_all|terminate", re.I), "DATA_DELETED"), + (re.compile(r"secret|credential|token|apikey|api_key|private_key|vault\.read|password", re.I), "SECRET_EXPOSED"), + (re.compile(r"contract|sign|execute_agreement|onchain|smart_contract", re.I), "CONTRACT_EXECUTED"), + (re.compile(r"device|actuator|valve|door|robot|plc|scada", re.I), "PHYSICAL_CHANGED"), + (re.compile(r"disable_user|deactivate|suspend_user|user\.update|identity|scim|role\.assign", re.I), "IDENTITY_CHANGED"), + (re.compile(r"grant|permission|policy\.attach|share|acl|invite", re.I), "ACCESS_GRANTED"), + (re.compile(r"deploy|release|rollback|migrate|prod|k8s\.apply|helm", re.I), "PRODUCTION_CHANGED"), + (re.compile(r"terraform|infra|instance|bucket|dns|firewall|network|cluster", re.I), "INFRA_MODIFIED"), + (re.compile(r"send|email|slack|sms|post_message|notify|publish|tweet", re.I), "MESSAGE_SENT"), + (re.compile(r"write|create|update|insert|upsert|patch|put", re.I), "DATA_WRITTEN"), + (re.compile(r"read|get|list|search|query|select|fetch", re.I), "DATA_READ"), +] + +_CONSEQUENCES = { + "MONEY_MOVED": ("money moved", "HIGH", False), + "DATA_DELETED": ("data deleted", "CRITICAL", False), + "IDENTITY_CHANGED": ("identity disabled or changed", "HIGH", True), + "SECRET_EXPOSED": ("secret exposed", "CRITICAL", False), + "PRODUCTION_CHANGED": ("production changed", "HIGH", True), + "INFRA_MODIFIED": ("infrastructure modified", "HIGH", True), + "ACCESS_GRANTED": ("access granted", "HIGH", True), + "MESSAGE_SENT": ("external message sent", "MEDIUM", False), + "CONTRACT_EXECUTED": ("contract executed", "CRITICAL", False), + "PHYSICAL_CHANGED": ("physical system changed", "CRITICAL", False), + "DATA_WRITTEN": ("data written", "MEDIUM", True), + "DATA_READ": ("data read", "LOW", True), + "NONE": ("no protected consequence", "LOW", True), +} + +_SEVERITY_RANK = {"LOW": 0, "MEDIUM": 1, "HIGH": 2, "CRITICAL": 3} + + +def severity_rank(severity: Optional[str]) -> int: + return _SEVERITY_RANK.get(severity or "", 0) + + +def classify_action(action: str, params: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: + """What can this action cause? Scale and money raise the stakes.""" + hints = dict(params or {}) + consequence = hints.get("consequence") + if consequence not in _CONSEQUENCES: + consequence = None + if consequence is None: + consequence = "DATA_READ" if _READ_VERB.search(action or "") else None + if consequence is None: + for pattern, name in _RULES: + if pattern.search(action or ""): + consequence = name + break + consequence = consequence or "NONE" + + label, severity, reversible = _CONSEQUENCES[consequence] + severity = hints.get("severity", severity) + reversible = hints.get("reversible", reversible) + + records = hints.get("recordsAffected", hints.get("records_affected", 0)) or 0 + amount = hints.get("amount", 0) or 0 + if records >= 1000 or amount >= 5000: + severity = severity if severity_rank(severity) >= 3 else "CRITICAL" + elif records >= 100 or amount >= 500: + severity = severity if severity_rank(severity) >= 2 else "HIGH" + + return { + "action": action, "consequence": consequence, "label": label, + "severity": severity, "reversible": reversible, + "protected": consequence != "NONE" and severity_rank(severity) >= 1, + "required_ring": ring_for_severity(severity), + } + + +_DOMAIN_BY_CONSEQUENCE = { + "MONEY_MOVED": "payments", "DATA_DELETED": "data-destruction", "SECRET_EXPOSED": "secrets", + "PRODUCTION_CHANGED": "prod-change", "INFRA_MODIFIED": "infrastructure", + "IDENTITY_CHANGED": "identity", "ACCESS_GRANTED": "access", "MESSAGE_SENT": "communications", + "CONTRACT_EXECUTED": "contracts", "PHYSICAL_CHANGED": "physical", + "DATA_WRITTEN": "data-write", "DATA_READ": "data-read", +} + + +def domain_for(consequence: Optional[str]) -> str: + return _DOMAIN_BY_CONSEQUENCE.get(consequence or "", "general") + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +# ── OPA/Rego bridge (integration-guide.md 6.2) ─────────────────────────────── +def agt_claims(*, ttp_score: float, action: Optional[str] = None, + consequence: Optional[str] = None, severity: Optional[str] = None, + reversible: Optional[bool] = None, domain: Optional[str] = None, + issuer_count: int = 0, agents: Optional[List[str]] = None, + route_id: Optional[str] = None, receipt: Optional[Mapping[str, Any]] = None, + at: Optional[str] = None) -> Dict[str, Any]: + """Build ``input.ttp`` for AGT's policy engine. Rego stays authoritative.""" + if action and not consequence: + classified = classify_action(action) + consequence = classified["consequence"] + severity = severity or classified["severity"] + reversible = classified["reversible"] if reversible is None else reversible + + agents = list(agents or []) + score = _clamp01(ttp_score) + return { + "ttp": { + "ttp_domain": domain or domain_for(consequence), + "ttp_score": round(score, 4), + "issuer_count": issuer_count, + "ttp_agt_score": to_agt_score(score), # legacy 0-1000 consumers + "trust_score": to_agt_trust_score(score).as_dict(), # AGT-native TrustScore + "trust_tier": trust_tier(score), + "required_ring": ring_for_severity(severity), + "consequence": consequence, + "severity": severity, + "reversible": reversible, + "route_id": route_id, + "agents": agents, + "spiffe_ids": [a for a in agents if is_spiffe_id(a)], + "receipt_id": (receipt or {}).get("receiptId"), + "receipt_hash": (receipt or {}).get("receiptHash"), + "evaluated_at": at or _now(), + } + } + + +# ── Evidence back to AGT (step 2 of the loop) ──────────────────────────────── +def to_trust_evidence(receipt: Optional[Mapping[str, Any]], + domain: Optional[str] = None) -> Optional[Dict[str, Any]]: + if not receipt: + return None + verifier = receipt.get("verifier") or {} + consequence = receipt.get("consequence") or {} + allowed = verifier.get("decision") == "EXECUTION_ALLOWED" + severity = consequence.get("severity", "LOW") + executed = receipt.get("executed") or {} + return { + "type": "TTPBehavioralEvidence", + "version": 1, + "subject": receipt.get("requestedBy") or receipt.get("principal"), + "domain": domain or domain_for(consequence.get("class")), + "action": (receipt.get("requested") or {}).get("action"), + "outcome": "allowed" if allowed else "denied", + # A denial at a high-consequence action is the strongest behavioural signal. + "weight": round((severity_rank(severity) + 1) / 4, 2), + "failures": [f.get("code") for f in verifier.get("failures", [])], + "receipt_id": receipt.get("receiptId"), + "receipt_hash": receipt.get("receiptHash"), + "observed_at": executed.get("executedAt") or receipt.get("issuedAt"), + } + + +def to_mesh_attestation(receipt: Optional[Mapping[str, Any]], peer: Optional[str] = None, + trust_domain: Optional[str] = None) -> Optional[Dict[str, Any]]: + """AgentMesh peer trust attestation (integration-guide.md 6.5).""" + if not receipt: + return None + route = receipt.get("routeSelected") or {} + principal = receipt.get("principal") + parsed = parse_spiffe_id(principal) if principal else None + score = _clamp01(route.get("effectiveTrust", 0)) + return { + "type": "AgentMeshTrustAttestation", + "version": 1, + "subject": receipt.get("requestedBy") or principal, + "peer": peer, + "trust_domain": trust_domain or (parsed or {}).get("trust_domain"), + "action": (receipt.get("requested") or {}).get("action"), + "consequence": (receipt.get("consequence") or {}).get("class"), + "outcome": (receipt.get("verifier") or {}).get("decision", "EXECUTION_DENIED"), + "trust_score": to_agt_trust_score(score).as_dict(), + "agt_trust_score": to_agt_score(score), + "ttp_score": score, + "evidence": { + "receipt_id": receipt.get("receiptId"), + "receipt_hash": receipt.get("receiptHash"), + "signed_by": receipt.get("keyId"), + }, + "issued_at": receipt.get("issuedAt"), + } + + +# ── AGT events -> canonical PCTR events ────────────────────────────────────── +_POLICY_ACTION = { + "allow": "EXECUTION_ALLOWED", "log": "EXECUTION_ALLOWED", "warn": "EXECUTION_ALLOWED", + "deny": "EXECUTION_DENIED", + "require_approval": "EXECUTION_DENIED", # not denied forever, but not executable yet +} +_LEGACY_DECISION = { + "allow": "EXECUTION_ALLOWED", "deny": "EXECUTION_DENIED", "review": "EXECUTION_DENIED", +} +# CascadeEvent actions all describe trust collapsing around an agent. +_CASCADE_TRUST = { + "circuit_opened": 0.2, "agent_degraded": 0.3, "agent_quarantined": 0.0, + "agent_killed": 0.0, "rollback_triggered": 0.4, "health_propagated": None, +} + + +def normalize_agt_event(event: Any) -> Optional[Dict[str, Any]]: + """Map one AGT event onto a canonical PCTR event, or None if unrecognised. + + Returning None is a valid, common answer: PCTR never invents a security event + from a shape it cannot read. + """ + if not isinstance(event, Mapping): + return None + + agent = (event.get("agentId") or event.get("agent_id") or event.get("sourceAgentId") + or event.get("agent") or event.get("subject")) + action = (event.get("action") or event.get("operation") or event.get("tool") + or event.get("capability")) + + # CascadeEvent + if event.get("eventId") and isinstance(event.get("affectedAgentIds"), list) and action in _CASCADE_TRUST: + to = _CASCADE_TRUST[action] + if to is None: + return None # health_propagated is telemetry, not a trust change + return {"event": "TRUST_CHANGED", "subject": agent, "detail": { + "to": to, "reason": event.get("reason") or f"AGT cascade containment: {action}", + "cascade_action": action, "affected": event.get("affectedAgentIds"), + "blast_radius": event.get("blastRadius")}} + + # RingViolation + if event.get("requiredRing") is not None and event.get("agentRing") is not None: + return {"event": "EXECUTION_DENIED", "subject": agent, "detail": { + "action": event.get("action"), + "reason": event.get("message") or + f"agent is Ring{event['agentRing']}; {event.get('action')} requires Ring{event['requiredRing']}", + "agent_ring": event.get("agentRing"), "required_ring": event.get("requiredRing")}} + + # KillSwitchResult + if event.get("killed") is True or event.get("terminated") is True: + return {"event": "TRUST_CHANGED", "subject": agent, "detail": { + "to": 0.0, "reason": event.get("reason") or "AGT kill switch engaged"}} + + # PolicyDecisionResult + if isinstance(event.get("allowed"), bool) and event.get("action") in _POLICY_ACTION: + policy_action = event["action"] + pending = policy_action == "require_approval" + approvers = event.get("approvers") or [] + return { + "event": "EXECUTION_ALLOWED" if (event["allowed"] and not pending) else _POLICY_ACTION[policy_action], + "subject": agent, + "detail": { + "action": event.get("operation") or event.get("tool"), + "reason": event.get("reason") or ( + f"AGT requires approval from: {', '.join(approvers) or 'an approver'}" if pending + else f"AGT policy {policy_action}"), + "policy": event.get("policyName") or event.get("matchedRule"), + "policy_action": policy_action, + "approvers": approvers, + "rate_limited": event.get("rateLimited", False), + }, + } + + # AuditEntry — AGT hash-chains its audit log as PCTR chains receipts. + if event.get("hash") and "previousHash" in event and isinstance(event.get("decision"), str): + mapped = _LEGACY_DECISION.get(event["decision"]) + if not mapped: + return None + return {"event": mapped, "subject": agent, "detail": { + "action": event.get("action"), "decision": event["decision"], + "reason": f'AGT audit entry records decision "{event["decision"]}"', + "audit_hash": event.get("hash"), "previous_hash": event.get("previousHash"), + "skill": (event.get("skillAuditMetadata") or {}).get("skillName")}} + + # TrustVerificationResult + if isinstance(event.get("verified"), bool) and event.get("trustScore") is not None: + score = from_agt_trust_score(event["trustScore"]) + tier = event["trustScore"].get("tier") if isinstance(event["trustScore"], Mapping) else None + return {"event": "TRUST_CHANGED", "subject": agent, "detail": { + "to": score, "tier": tier or trust_tier(score), + "reason": event.get("reason") or + f"AGT trust verification {'passed' if event['verified'] else 'failed'}"}} + + kind = str(event.get("type") or event.get("event") or event.get("eventType") or event.get("kind") or "").lower() + + if re.search(r"register|discover|onboard", kind) and agent: + raw = event.get("trustScore") + return {"event": "AGENT_DISCOVERED", "subject": agent, "detail": { + "framework": "microsoft-agt", + "trust": from_agt_trust_score(raw) if raw is not None else None, + "status": event.get("status")}} + + if re.search(r"delegat|handoff|assign", kind): + to = event.get("to") or event.get("target") or event.get("targetAgent") or event.get("delegateTo") + return {"event": "AGENT_DELEGATED", "subject": agent, "detail": {"to": to}} if to else None + + if re.search(r"(action|tool|capability).*(request|invoke|propos|call)|invocation", kind): + if not action: + return None + params = event.get("parameters") or event.get("params") or event.get("arguments") or event.get("input") or {} + classified = classify_action(action, params) + detail: Dict[str, Any] = {"action": action} + detail.update(params if isinstance(params, Mapping) else {}) + detail.update({"consequence": classified["consequence"], "severity": classified["severity"], + "required_ring": classified["required_ring"]}) + return {"event": "ACTION_PROPOSED", "subject": agent, "detail": detail} + + if re.search(r"complete|finish|result", kind) and action: + failed = bool(event.get("error") or event.get("failed")) + return {"event": "EXECUTION_COMPLETED", "subject": agent, + "detail": {"action": action, "status": "FAILED" if failed else "SUCCEEDED"}} + + return None diff --git a/sdk/python/test_agt.py b/sdk/python/test_agt.py new file mode 100644 index 0000000..cc4ef17 --- /dev/null +++ b/sdk/python/test_agt.py @@ -0,0 +1,198 @@ +"""Parity tests for the Python AGT bridge. + +These assert the same behaviour as packages/pctr/tests/agt.test.mjs, so the two +language bindings cannot drift apart on the parts that matter to AGT. + +Run: python3 -m unittest discover -s sdk/python -p 'test_*.py' +""" +import unittest + +from agt import ( + AGT_TIER_THRESHOLDS, agt_claims, classify_action, domain_for, from_agt_score, + from_agt_trust_score, is_spiffe_id, normalize_agt_event, parse_spiffe_id, + ring_for_severity, to_agt_score, to_agt_trust_score, to_mesh_attestation, + to_trust_evidence, trust_tier, +) + +RECEIPT = { + "receiptId": "rcpt-1", "receiptHash": "sha256:abc", "keyId": "ed25519:k1", + "principal": "spiffe://blocksifr.com/ns/prod/sa/planner", + "requestedBy": "spiffe://blocksifr.com/ns/prod/sa/finance", + "requested": {"action": "payments.transfer", "target": "acct:1", "params": {"amount": 1800}}, + "consequence": {"class": "MONEY_MOVED", "severity": "HIGH", "reversible": False}, + "routeSelected": {"routeId": "r1", "effectiveTrust": 0.9178, "agents": ["spiffe://blocksifr.com/ns/prod/sa/finance"]}, + "verifier": {"decision": "EXECUTION_ALLOWED", "failures": []}, + "executed": {"status": "SUCCEEDED", "executedAt": "2026-09-15T00:00:00Z"}, + "issuedAt": "2026-09-15T00:00:00Z", +} + + +class TrustScaleTests(unittest.TestCase): + def test_agt_scores_zero_to_one_with_tiers(self): + # Upstream AGT (agent-governance-typescript/src/trust.ts), not the 0-1000 scale + # our integration guide used to specify. + self.assertEqual(AGT_TIER_THRESHOLDS, + {"untrusted": 0.0, "provisional": 0.3, "trusted": 0.6, "verified": 0.85}) + self.assertEqual(to_agt_trust_score(0.9178).as_dict(), + {"overall": 0.9178, "dimensions": {}, "tier": "Verified"}) + self.assertEqual(trust_tier(0.1), "Untrusted") + self.assertEqual(trust_tier(0.3), "Provisional") + self.assertEqual(trust_tier(0.6), "Trusted") + self.assertEqual(trust_tier(0.85), "Verified") + + def test_scores_are_clamped_not_propagated(self): + self.assertEqual(to_agt_trust_score(1.5).overall, 1.0) + self.assertEqual(to_agt_trust_score(-2).overall, 0.0) + self.assertEqual(to_agt_trust_score("not a number").overall, 0.0) + + def test_trust_score_accepts_every_inbound_shape(self): + self.assertEqual(from_agt_trust_score({"overall": 0.42, "tier": "Provisional"}), 0.42) + self.assertEqual(from_agt_trust_score(0.42), 0.42) + self.assertEqual(from_agt_trust_score(to_agt_trust_score(0.42)), 0.42) + + def test_legacy_thousand_scale_still_available(self): + self.assertEqual(to_agt_score(0.9178), 918) + self.assertEqual(to_agt_score(1.5), 1000) + self.assertEqual(from_agt_score(918), 0.918) + + def test_rings_come_from_the_consequence(self): + self.assertEqual(ring_for_severity("CRITICAL"), 0) + self.assertEqual(ring_for_severity("HIGH"), 1) + self.assertEqual(ring_for_severity("MEDIUM"), 2) + self.assertEqual(ring_for_severity(None), 3) + + +class SpiffeTests(unittest.TestCase): + def test_svid_round_trips(self): + svid = "spiffe://blocksifr.com/ns/prod/sa/finance" + self.assertTrue(is_spiffe_id(svid)) + self.assertFalse(is_spiffe_id("finance")) + self.assertEqual(parse_spiffe_id(svid), + {"trust_domain": "blocksifr.com", "path": "/ns/prod/sa/finance", "id": svid}) + self.assertIsNone(parse_spiffe_id("not-a-svid")) + + +class ConsequenceTests(unittest.TestCase): + def test_verb_decides_over_namespace(self): + self.assertEqual(classify_action("payments.status")["consequence"], "DATA_READ") + self.assertEqual(classify_action("payments.transfer")["consequence"], "MONEY_MOVED") + self.assertEqual(classify_action("customers.delete")["consequence"], "DATA_DELETED") + self.assertEqual(classify_action("send_invoice_email")["consequence"], "MESSAGE_SENT") + + def test_scale_and_money_escalate_severity(self): + self.assertEqual(classify_action("payments.transfer")["severity"], "HIGH") + self.assertEqual(classify_action("payments.transfer", {"amount": 18000})["severity"], "CRITICAL") + self.assertEqual(classify_action("records.update", {"recordsAffected": 5000})["severity"], "CRITICAL") + + def test_domains_match_the_javascript_bridge(self): + self.assertEqual(domain_for("MONEY_MOVED"), "payments") + self.assertEqual(domain_for("DATA_DELETED"), "data-destruction") + self.assertEqual(domain_for("SOMETHING_NEW"), "general") + + +class ClaimsTests(unittest.TestCase): + def test_rego_input_carries_every_evaluated_claim(self): + claims = agt_claims(ttp_score=0.9178, action="payments.transfer", issuer_count=2, + agents=["spiffe://blocksifr.com/ns/prod/sa/finance", "planner"], + route_id="r1", receipt=RECEIPT)["ttp"] + self.assertEqual(claims["ttp_domain"], "payments") + self.assertEqual(claims["ttp_score"], 0.9178) + self.assertEqual(claims["issuer_count"], 2) + self.assertEqual(claims["trust_score"]["tier"], "Verified") + self.assertEqual(claims["required_ring"], ring_for_severity(claims["severity"])) + self.assertEqual(claims["spiffe_ids"], ["spiffe://blocksifr.com/ns/prod/sa/finance"]) + self.assertEqual(claims["receipt_hash"], "sha256:abc") + + +class EvidenceTests(unittest.TestCase): + def test_receipt_becomes_behavioural_evidence(self): + evidence = to_trust_evidence(RECEIPT) + self.assertEqual(evidence["type"], "TTPBehavioralEvidence") + self.assertEqual(evidence["outcome"], "allowed") + self.assertEqual(evidence["domain"], "payments") + self.assertEqual(evidence["receipt_hash"], "sha256:abc") + self.assertEqual(evidence["weight"], 0.75) + self.assertIsNone(to_trust_evidence(None)) + + def test_denial_at_high_consequence_weighs_more(self): + denied = dict(RECEIPT) + denied["consequence"] = {"class": "MONEY_MOVED", "severity": "CRITICAL"} + denied["verifier"] = {"decision": "EXECUTION_DENIED", "failures": [{"code": "APPROVAL_REQUIRED"}]} + evidence = to_trust_evidence(denied) + self.assertEqual(evidence["outcome"], "denied") + self.assertEqual(evidence["failures"], ["APPROVAL_REQUIRED"]) + self.assertGreater(evidence["weight"], to_trust_evidence(RECEIPT)["weight"]) + + def test_mesh_attestation_traces_back_to_the_receipt(self): + attestation = to_mesh_attestation(RECEIPT, peer="spiffe://partner.example/ns/prod/sa/mesh") + self.assertEqual(attestation["type"], "AgentMeshTrustAttestation") + self.assertEqual(attestation["trust_domain"], "blocksifr.com") + self.assertEqual(attestation["trust_score"]["tier"], "Verified") + self.assertEqual(attestation["evidence"]["receipt_hash"], "sha256:abc") + self.assertEqual(attestation["evidence"]["signed_by"], "ed25519:k1") + self.assertIsNone(to_mesh_attestation(None)) + + +class EventTests(unittest.TestCase): + def test_policy_decision_result_covers_every_policy_action(self): + def decision(action, allowed, **extra): + return normalize_agt_event({"allowed": allowed, "action": action, "agentId": "finance", + "operation": "payments.transfer", "policyName": "agt.authz", + "approvers": [], "rateLimited": False, **extra}) + + self.assertEqual(decision("allow", True)["event"], "EXECUTION_ALLOWED") + self.assertEqual(decision("log", True)["event"], "EXECUTION_ALLOWED") + self.assertEqual(decision("warn", True)["event"], "EXECUTION_ALLOWED") + self.assertEqual(decision("deny", False)["event"], "EXECUTION_DENIED") + + pending = decision("require_approval", True, approvers=["ops-oncall"]) + self.assertEqual(pending["event"], "EXECUTION_DENIED") + self.assertEqual(pending["detail"]["approvers"], ["ops-oncall"]) + self.assertEqual(pending["detail"]["policy_action"], "require_approval") + + def test_audit_entry_keeps_the_hash_chain(self): + entry = normalize_agt_event({"timestamp": "t", "agentId": "finance", "action": "customers.delete", + "decision": "review", "hash": "h1", "previousHash": "h0"}) + self.assertEqual(entry["event"], "EXECUTION_DENIED") + self.assertEqual(entry["detail"]["audit_hash"], "h1") + self.assertEqual(entry["detail"]["previous_hash"], "h0") + + def test_cascade_containment_and_ring_violations(self): + quarantined = normalize_agt_event({"eventId": "e1", "timestamp": "t", "sourceAgentId": "finance", + "affectedAgentIds": ["a", "b"], "action": "agent_quarantined", + "reason": "breach", "blastRadius": 3}) + self.assertEqual(quarantined["event"], "TRUST_CHANGED") + self.assertEqual(quarantined["detail"]["to"], 0.0) + self.assertEqual(quarantined["detail"]["blast_radius"], 3) + + self.assertIsNone(normalize_agt_event({"eventId": "e2", "timestamp": "t", "sourceAgentId": "f", + "affectedAgentIds": [], "action": "health_propagated"})) + + violation = normalize_agt_event({"agentId": "finance", "action": "prod.deploy", + "agentRing": 2, "requiredRing": 0, "message": "Ring2 cannot reach Ring0"}) + self.assertEqual(violation["event"], "EXECUTION_DENIED") + self.assertEqual(violation["detail"]["required_ring"], 0) + + def test_trust_verification_result_keeps_its_tier(self): + verification = normalize_agt_event({"verified": False, "agentId": "finance", "reason": "stale attestation", + "trustScore": {"overall": 0.42, "dimensions": {}, "tier": "Provisional"}}) + self.assertEqual(verification["detail"]["to"], 0.42) + self.assertEqual(verification["detail"]["tier"], "Provisional") + + def test_invocations_get_the_consequence_agt_omits(self): + proposed = normalize_agt_event({"type": "action.invocation", "agentId": "finance", + "action": "payments.transfer", "parameters": {"amount": 18000}}) + self.assertEqual(proposed["event"], "ACTION_PROPOSED") + self.assertEqual(proposed["detail"]["consequence"], "MONEY_MOVED") + self.assertEqual(proposed["detail"]["severity"], "CRITICAL") + self.assertEqual(proposed["detail"]["required_ring"], 0) + + def test_unrecognised_events_are_dropped_not_invented(self): + self.assertIsNone(normalize_agt_event({"type": "telemetry.heartbeat", "agentId": "finance"})) + self.assertIsNone(normalize_agt_event({"type": "action.invocation", "agentId": "finance"})) + self.assertIsNone(normalize_agt_event(None)) + self.assertIsNone(normalize_agt_event("not-a-mapping")) + + +if __name__ == "__main__": + unittest.main()