From c3382cdd19d07328b27b259a4b2848401c80d06b Mon Sep 17 00:00:00 2001 From: Tymofii Pidlisnyi <171286556+aeoess@users.noreply.github.com> Date: Mon, 3 Aug 2026 12:01:55 -0700 Subject: [PATCH 1/2] Add Agent Passport System TRACE exporter integration Maps exactly one signed APS policy decision, the dict returned by agent_passport.policy.evaluate_intent, onto exactly one TRACE Trust Record at EAT profile tag:agentrust-io.com,2026:trace-v0.2. The mapper verifies the APS evaluator signature and expiry through verify_policy_decision before it maps anything, and raises rather than emit a record for a decision that fails either check. The APS evaluator signature and the TRACE record signature are separate keys and the module never conflates them. Conformance level 0. The graded artifact is the unsigned record, so TR-SIG-005 reports UNVERIFIED; the signed form is written alongside it and verifies with agentrust_trace.verify_record. model, data_class and build_provenance are absent because an APS policy decision carries none of them, and a test pins that set so it cannot widen silently. --- .github/workflows/aeoess-aps-conformance.yml | 51 ++++ integrations/aeoess-aps/.gitignore | 2 + integrations/aeoess-aps/README.md | 111 +++++++ integrations/aeoess-aps/aps_trace.py | 236 ++++++++++++++ .../aeoess-aps/examples/emit_record.py | 114 +++++++ integrations/aeoess-aps/integration.yaml | 20 ++ integrations/aeoess-aps/pyproject.toml | 23 ++ integrations/aeoess-aps/tests/test_mapping.py | 288 ++++++++++++++++++ 8 files changed, 845 insertions(+) create mode 100644 .github/workflows/aeoess-aps-conformance.yml create mode 100644 integrations/aeoess-aps/.gitignore create mode 100644 integrations/aeoess-aps/README.md create mode 100644 integrations/aeoess-aps/aps_trace.py create mode 100644 integrations/aeoess-aps/examples/emit_record.py create mode 100644 integrations/aeoess-aps/integration.yaml create mode 100644 integrations/aeoess-aps/pyproject.toml create mode 100644 integrations/aeoess-aps/tests/test_mapping.py diff --git a/.github/workflows/aeoess-aps-conformance.yml b/.github/workflows/aeoess-aps-conformance.yml new file mode 100644 index 0000000..87e6659 --- /dev/null +++ b/.github/workflows/aeoess-aps-conformance.yml @@ -0,0 +1,51 @@ +# Agent Passport System TRACE exporter conformance workflow. +# Lives at repo root because GitHub Actions only discovers workflows there. +# Scoped to this integration via the paths filter. +name: aeoess-aps conformance +on: + push: + paths: + - "integrations/aeoess-aps/**" + - ".github/workflows/aeoess-aps-conformance.yml" + pull_request: + paths: + - "integrations/aeoess-aps/**" + - ".github/workflows/aeoess-aps-conformance.yml" + schedule: + - cron: "0 6 * * 1" # weekly: catch drift against the latest released packages + workflow_dispatch: + +permissions: + contents: read + +jobs: + conformance: + strategy: + fail-fast: false + matrix: + python: ["3.11", "3.12", "3.13", "3.14"] + os: [ubuntu-latest] + runs-on: ${{ matrix.os }} + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: ${{ matrix.python }} + - name: Install released agentrust-io packages + run: | + python -m pip install --upgrade pip + pip install agentrust-trace agentrust-trace-tests agent-passport-system + - name: Install this integration + run: pip install -e "integrations/aeoess-aps[test]" + - name: Integration tests + run: pytest integrations/aeoess-aps/tests -q + - name: Emit a sample TRACE record + run: python integrations/aeoess-aps/examples/emit_record.py --out trust-record.jwt + - name: TRACE conformance level 0 + run: trace-tests verify --record trust-record.jwt --level 0 + - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: conformance-${{ matrix.os }}-py${{ matrix.python }} + path: | + trust-record.jwt + trust-record.jwt.signed.json diff --git a/integrations/aeoess-aps/.gitignore b/integrations/aeoess-aps/.gitignore new file mode 100644 index 0000000..9c882b7 --- /dev/null +++ b/integrations/aeoess-aps/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.egg-info/ diff --git a/integrations/aeoess-aps/README.md b/integrations/aeoess-aps/README.md new file mode 100644 index 0000000..6a8780e --- /dev/null +++ b/integrations/aeoess-aps/README.md @@ -0,0 +1,111 @@ +# Agent Passport System integration with TRACE + +The Agent Passport System (APS) is an open protocol for agent identity and +scoped delegation, in which an evaluator checks an agent's declared intent +against a Values Floor and returns an Ed25519-signed policy decision. + +This integration maps exactly one signed APS policy decision, the dict returned +by `agent_passport.policy.evaluate_intent`, onto exactly one TRACE Trust Record +(EAT profile `tag:agentrust-io.com,2026:trace-v0.2`). Nothing else in APS is +mapped. Action receipts, revocation, identity binding, delegation chains and +attribution are out of scope here. + +## Two different signatures + +Conflating these two is the mistake this integration exists to avoid. + +1. **The APS evaluator signature**, carried in the decision's `signature` field + and verified against the `evaluatorPublicKey` embedded in the same decision. + `aps_trace` verifies it through `agent_passport.policy.verify_policy_decision` + before it maps anything. A decision whose signature fails, or whose + `expiresAt` has passed, raises `ValueError` and produces no record. +2. **The TRACE record signature**, applied afterwards by + `agentrust_trace.sign_record` with a separate key, whose public JWK is bound + into `cnf.jwk`. `aps_trace` never applies it. + +Verifying signature 1 says an APS evaluator authorized this action. Verifying +signature 2 says this exported record is the one the exporter produced. Neither +implies the other. + +## Run it + +Against released packages: + +```bash +pip install agentrust-trace agentrust-trace-tests agent-passport-system +pip install -e "integrations/aeoess-aps[test]" +pytest integrations/aeoess-aps/tests -q +python integrations/aeoess-aps/examples/emit_record.py --out trust-record.jwt +trace-tests verify --record trust-record.jwt --level 0 +``` + +The example mints a decision at run time with ephemeral keys instead of loading +a committed fixture. APS decisions expire five minutes after evaluation and the +mapper refuses expired decisions, so a committed decision fixture would be +permanently unmappable. No network access and no credentials are needed. + +## Field mapping + +| TRACE field | APS source | +|---|---| +| `eat_profile` | constant `tag:agentrust-io.com,2026:trace-v0.2` | +| `iat` | `evaluatedAt`, parsed to Unix seconds | +| `subject` | `spiffe://agent-passport.org/evaluator//decision/` | +| `cnf.jwk` | caller-supplied public JWK for the TRACE signing key | +| `policy.bundle_hash` | sha256 over the APS canonical bytes of `{floorVersion, principlesEvaluated}` | +| `policy.enforcement_mode` | `enforce` when any principle was evaluated `inline`, else `advisory` | +| `policy.version` | `floorVersion` | +| `runtime.platform` | `software-only` | +| `runtime.measurement` | sha256 over the APS canonical bytes of the full signed decision | +| `appraisal.status` | `permit` to `affirming`, `narrow` to `warning`, `deny` to `contraindicated` | +| `appraisal.verifier` | `urn:aps:evaluator:` | +| `appraisal.policy_ref` | `urn:aps:floor:` | +| `appraisal.timestamp` | `iat` | +| `transparency` | `urn:aps:transparency:none` | + +An APS verdict this mapper does not know is refused rather than appraised. + +## What is verified + +- `aps_trace.build_trace_record` refuses a decision with a tampered verdict, a + corrupt signature, a passed `expiresAt`, a missing field, or an unknown + verdict. `tests/` covers each refusal. +- `agentrust_trace.sign_record` signs the record with an ephemeral Ed25519 key + and `agentrust_trace.verify_record(..., allow_embedded_key=True)` verifies the + round-trip. `tests/` includes a tamper probe that must fail verification. +- `trace-tests verify --level 0` passes on the emitted record: 8 checks, of + which TR-SIG-005 is UNVERIFIED. See below. +- Every field the mapper emits validates against the TRACE v0.2 JSON Schema. + The fields it does not emit are pinned by a test. + +## What it does NOT claim + +See rules 2 and 4 in [CONTRIBUTING.md](../../CONTRIBUTING.md). + +- **The record is a partial TRACE record.** `model`, `data_class` and + `build_provenance` are required by the v0.2 JSON Schema and are absent. An APS + policy decision carries no model identity, no data classification and no build + provenance, so any value there would be invented. The record passes + `trace-tests verify --level 0`, which does not grade those fields, and it does + not satisfy the full v0.2 schema. `tests/test_mapping.py` pins the exact set of + absent required fields so the gap cannot widen silently. +- **Level 0 carries an explicit `TR-SIG-005 UNVERIFIED` finding.** The graded + artifact is the unsigned record, so trace-tests reports that it is not + cryptographically verified. The signed form is written next to it as + `.signed.json` and verifies with `agentrust_trace.verify_record`. +- **`runtime.platform` is `software-only` and there is no hardware attestation.** + `runtime.measurement` is a digest of the signed APS decision, not a TEE + measurement. Per the v0.2 schema, `software-only` records must never be treated + as attested evidence. +- **The ephemeral TRACE signing key proves the sign and verify path works.** It + does not chain to a trusted issuer. +- **`transparency` is `urn:aps:transparency:none`.** This integration publishes + nothing to a transparency log, so there is no SCITT receipt to resolve. +- **No conformance level above 0 is claimed or configured.** + +## Conformance CI + +`.github/workflows/aeoess-aps-conformance.yml` at the repository root, scoped to +`integrations/aeoess-aps/**`. It installs the released packages, runs the tests, +emits a record and runs `trace-tests verify --level 0` across Python 3.11 to +3.14. diff --git a/integrations/aeoess-aps/aps_trace.py b/integrations/aeoess-aps/aps_trace.py new file mode 100644 index 0000000..fe9df5a --- /dev/null +++ b/integrations/aeoess-aps/aps_trace.py @@ -0,0 +1,236 @@ +"""aps_trace: export one signed APS policy decision as a TRACE Trust Record. + +The Agent Passport System (APS) evaluates an ActionIntent against a Values +Floor and returns a PolicyDecision: a dict signed by the evaluator, carrying a +verdict of ``permit``, ``narrow`` or ``deny``. This module maps exactly one +such decision onto one TRACE Trust Record dict (EAT profile +``tag:agentrust-io.com,2026:trace-v0.2``). + +Two different signatures are involved and they are never the same key: + +1. The APS evaluator signature carried inside the decision. It is verified here, + before any mapping happens, against the ``evaluatorPublicKey`` embedded in + the decision, using ``agent_passport.policy.verify_policy_decision``. +2. The TRACE record signature, applied afterwards by + ``agentrust_trace.sign_record`` with a separate key whose public JWK is bound + into ``cnf.jwk``. This module never applies it. + +A record that fails step 1 is never produced. The mapper raises instead, so an +unverified or expired APS decision cannot become a TRACE record that looks +appraised. + +Field mapping +============= + +TRACE field APS source +-------------------------- --------------------------------------------------- +eat_profile constant EAT_PROFILE +iat ``evaluatedAt``, parsed to Unix seconds +subject spiffe:///evaluator/ + /decision/ +cnf.jwk caller-supplied ``trace_jwk`` +policy.bundle_hash sha256 over the canonical bytes of + {"floorVersion", "principlesEvaluated"} +policy.enforcement_mode "enforce" when any principle was evaluated with + enforcementMode "inline", else "advisory" +policy.version ``floorVersion`` (omitted when empty) +runtime.platform "software-only" +runtime.measurement sha256 over the canonical bytes of the full signed + decision +appraisal.status ``verdict`` through VERDICT_TO_APPRAISAL +appraisal.verifier urn:aps:evaluator: +appraisal.policy_ref urn:aps:floor: (omitted when empty) +appraisal.timestamp iat +transparency "urn:aps:transparency:none" + +Deliberately absent +=================== + +``model``, ``data_class`` and ``build_provenance`` are required by the TRACE +v0.2 JSON Schema and are absent here. An APS policy decision carries no model +identity, no data classification and no build provenance, so any value would be +invented. The record is therefore a partial TRACE record: it passes +``trace-tests verify --level 0``, and it does not satisfy the full v0.2 schema. +``tests/test_mapping.py`` pins the exact set of absent required fields so the +gap stays deliberate. See the README section "What it does NOT claim". + +``tool_transcript`` is absent because a policy decision is not a tool +transcript. ``delegation`` is absent because delegation chains are out of scope +for this integration. +""" +from __future__ import annotations + +import hashlib +from datetime import datetime, timezone +from typing import Any +from urllib.parse import quote + +from agent_passport.canonical import canonicalize +from agent_passport.policy import verify_policy_decision + +#: EAT profile URI, identical to the constant the other integrations use. +EAT_PROFILE = "tag:agentrust-io.com,2026:trace-v0.2" + +#: SPIFFE trust domain for APS workload identities. +TRUST_DOMAIN = "agent-passport.org" + +#: APS verdict to EAR appraisal status (draft-ietf-rats-ar4si). +#: permit authorizes the action, narrow authorizes it under added constraints, +#: deny withholds authorization. TRACE has no "narrow", and "warning" is the +#: EAR status for an appraisal that affirms with reservations. +VERDICT_TO_APPRAISAL = { + "permit": "affirming", + "narrow": "warning", + "deny": "contraindicated", +} + +#: Keys a signed APS policy decision must carry. Checked before signature +#: verification so a malformed input fails with a precise message. +REQUIRED_DECISION_KEYS = frozenset( + { + "decisionId", + "intentId", + "evaluatorId", + "evaluatorPublicKey", + "verdict", + "principlesEvaluated", + "reason", + "floorVersion", + "evaluatedAt", + "expiresAt", + "signature", + } +) + + +def build_trace_record(decision: dict[str, Any], *, trace_jwk: dict[str, str]) -> dict[str, Any]: + """Map one signed APS policy decision onto an unsigned TRACE Trust Record. + + Args: + decision: The dict returned by ``agent_passport.policy.evaluate_intent``, + including its evaluator ``signature``. + trace_jwk: Public JWK bound into ``cnf.jwk``. It belongs to the key used + later by ``agentrust_trace.sign_record``, and has nothing to do with + the APS evaluator key. + + Returns: + An unsigned TRACE Trust Record dict. + + Raises: + ValueError: if the decision is malformed, its evaluator signature does + not verify, it has expired, or its verdict is not one this mapper + knows how to appraise. No record is returned in any of those cases. + """ + _validate_decision(decision) + + evaluator_id = decision["evaluatorId"] + decision_id = decision["decisionId"] + floor_version = decision["floorVersion"] + iat = _unix_seconds(decision["evaluatedAt"]) + + appraisal: dict[str, Any] = { + "status": VERDICT_TO_APPRAISAL[decision["verdict"]], + "timestamp": iat, + "verifier": f"urn:aps:evaluator:{quote(str(evaluator_id), safe='')}", + } + if floor_version: + appraisal["policy_ref"] = f"urn:aps:floor:{quote(str(floor_version), safe='')}" + + policy: dict[str, Any] = { + "bundle_hash": _sha256_of( + { + "floorVersion": floor_version, + "principlesEvaluated": decision["principlesEvaluated"], + } + ), + "enforcement_mode": _enforcement_mode(decision["principlesEvaluated"]), + } + if floor_version: + policy["version"] = str(floor_version) + + return { + "eat_profile": EAT_PROFILE, + "iat": iat, + "subject": ( + f"spiffe://{TRUST_DOMAIN}" + f"/evaluator/{quote(str(evaluator_id), safe='')}" + f"/decision/{quote(str(decision_id), safe='')}" + ), + "cnf": {"jwk": trace_jwk}, + "policy": policy, + "runtime": { + "platform": "software-only", + "measurement": _sha256_of(decision), + }, + "appraisal": appraisal, + "transparency": "urn:aps:transparency:none", + } + + +def _validate_decision(decision: dict[str, Any]) -> None: + """Refuse anything that must not become a TRACE Trust Record. + + Checks shape, then the evaluator signature and expiry through + ``verify_policy_decision``, then the verdict. Every rejection raises + ``ValueError`` with the reason, so a caller can never mistake a refusal for + a mapped record. + """ + if not isinstance(decision, dict): + raise ValueError(f"decision must be a dict, got {type(decision).__name__}") + + missing = REQUIRED_DECISION_KEYS - decision.keys() + if missing: + raise ValueError(f"decision missing required fields: {sorted(missing)}") + + check = verify_policy_decision(decision) + if not check["valid"]: + raise ValueError( + "refusing to map an APS decision that failed verification: " + + "; ".join(check["errors"]) + ) + + verdict = decision["verdict"] + if verdict not in VERDICT_TO_APPRAISAL: + raise ValueError( + f"unknown APS verdict {verdict!r}; expected one of " + f"{sorted(VERDICT_TO_APPRAISAL)}" + ) + + +def _enforcement_mode(principles: Any) -> str: + """Return the TRACE enforcement mode implied by the evaluated principles. + + APS records a per-principle ``enforcementMode`` of ``inline``, ``audit`` or + ``warn``. Inline blocks the action when the principle fails, which is what + TRACE calls ``enforce``. Audit and warn log and let the action proceed, + which is ``advisory``. ``silent`` is never emitted: APS has no mode that + suppresses the operational record. + """ + if isinstance(principles, list): + for principle in principles: + if isinstance(principle, dict) and principle.get("enforcementMode") == "inline": + return "enforce" + return "advisory" + + +def _sha256_of(obj: Any) -> str: + """Return ``sha256:`` over the APS canonical bytes of *obj*. + + Uses the same canonicalization the APS evaluator signs over, so the digest + is reproducible by any APS implementation from the same input. + """ + digest = hashlib.sha256(canonicalize(obj).encode("utf-8")).hexdigest() + return f"sha256:{digest}" + + +def _unix_seconds(timestamp: str) -> int: + """Parse an APS ISO 8601 timestamp into integer Unix seconds (UTC).""" + if not isinstance(timestamp, str) or not timestamp: + raise ValueError(f"timestamp must be a non-empty ISO 8601 string, got {timestamp!r}") + try: + parsed = datetime.fromisoformat(timestamp.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError(f"timestamp {timestamp!r} is not ISO 8601: {exc}") from exc + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return int(parsed.timestamp()) diff --git a/integrations/aeoess-aps/examples/emit_record.py b/integrations/aeoess-aps/examples/emit_record.py new file mode 100644 index 0000000..fcf1223 --- /dev/null +++ b/integrations/aeoess-aps/examples/emit_record.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +"""Emit a TRACE Trust Record from a freshly evaluated APS policy decision. + +Runs the real APS path with ephemeral keys: an agent declares an ActionIntent, +an evaluator evaluates it against the Values Floor with FloorValidatorV1, and +the resulting signed PolicyDecision is mapped onto a TRACE Trust Record by +:func:`aps_trace.build_trace_record`. No network access and no credentials. + +The decision is minted at run time rather than loaded from a committed fixture +because an APS policy decision expires five minutes after evaluation, and +``build_trace_record`` refuses expired decisions. A committed fixture would be +permanently unmappable, which is the mapper behaving correctly. + +Two files are written: + + Unsigned record, the artifact ``trace-tests verify`` grades + .signed.json Signed record, verifiable with + ``agentrust_trace.verify_record(..., allow_embedded_key=True)`` + +Usage: + python examples/emit_record.py --out trust-record.jwt +""" +from __future__ import annotations + +import argparse +import json +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import agentrust_trace +from agent_passport.crypto import generate_key_pair +from agent_passport.policy import FloorValidatorV1, create_action_intent, evaluate_intent + +from aps_trace import build_trace_record + +FLOOR_VERSION = "floor-1.0" + + +def mint_decision() -> dict: + """Run one full APS intent-to-decision cycle and return the signed decision.""" + agent = generate_key_pair() + evaluator = generate_key_pair() + + intent = create_action_intent( + agent_id="agent_example", + agent_public_key=agent["publicKey"], + delegation_id="dlg_example", + action={"scopeRequired": "repo:read", "description": "read one repository file"}, + private_key=agent["privateKey"], + context="agentrust-io integration example", + ) + + expires_at = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat() + validation_context = { + "floorVersion": FLOOR_VERSION, + "agentRegistered": True, + "agentAttestationValid": True, + "delegation": { + "scope": ["repo:read"], + "revoked": False, + "expiresAt": expires_at, + "maxDepth": 3, + "currentDepth": 1, + }, + } + + return evaluate_intent( + intent=intent, + validator=FloorValidatorV1(), + validation_context=validation_context, + evaluator_id="eval_example", + evaluator_public_key=evaluator["publicKey"], + evaluator_private_key=evaluator["privateKey"], + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--out", required=True, help="Path for the trace-tests gradable record") + args = parser.parse_args() + + decision = mint_decision() + + key = agentrust_trace.generate_key() + jwk = agentrust_trace.key_to_jwk(key) + record = build_trace_record(decision, trace_jwk=jwk) + + signed = agentrust_trace.sign_record(dict(record), key) + agentrust_trace.verify_record(signed, allow_embedded_key=True, max_age_seconds=None) + + out = Path(args.out) + out.write_text( + json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n", + encoding="utf-8", + ) + signed_out = out.with_name(out.name + ".signed.json") + signed_out.write_text( + json.dumps(signed, sort_keys=True, separators=(",", ":")) + "\n", + encoding="utf-8", + ) + + print(f"APS verdict: {decision['verdict']}") + print(f"appraisal.status: {record['appraisal']['status']}") + print(f"subject: {record['subject']}") + print(f"unsigned (for trace-tests): {out}") + print(f"signed (verify_record OK): {signed_out}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/integrations/aeoess-aps/integration.yaml b/integrations/aeoess-aps/integration.yaml new file mode 100644 index 0000000..0d1fb21 --- /dev/null +++ b/integrations/aeoess-aps/integration.yaml @@ -0,0 +1,20 @@ +name: Agent Passport System +vendor: aeoess +integrates_with: + - trace +description: >- + Maps one Ed25519-signed Agent Passport System policy decision onto a TRACE + Trust Record, refusing decisions that fail signature or expiry verification. +maintainer: + github: aeoess +repository: https://github.com/aeoess/agent-passport-python +homepage: https://agent-passport.org +license: Apache-2.0 +tier: community +# Level 0 passes with an explicit TR-SIG-005 UNVERIFIED finding: the graded +# artifact is the unsigned record, so trace-tests reports it as not +# cryptographically verified. The signed form is written alongside it and +# verifies via agentrust_trace.verify_record (see README). +trace_conformance_level: 0 +tested_against: + agentrust-trace: "0.5.1" diff --git a/integrations/aeoess-aps/pyproject.toml b/integrations/aeoess-aps/pyproject.toml new file mode 100644 index 0000000..5996f6f --- /dev/null +++ b/integrations/aeoess-aps/pyproject.toml @@ -0,0 +1,23 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "aeoess-aps-integration" +version = "0.1.0" +description = "Maps signed Agent Passport System policy decisions onto TRACE Trust Records" +requires-python = ">=3.11" +license = "Apache-2.0" +dependencies = [ + "agentrust-trace>=0.5", + "agent-passport-system>=2.10", +] + +[project.optional-dependencies] +test = [ + "pytest", + "agentrust-trace-tests>=0.4", +] + +[tool.setuptools] +py-modules = ["aps_trace"] diff --git a/integrations/aeoess-aps/tests/test_mapping.py b/integrations/aeoess-aps/tests/test_mapping.py new file mode 100644 index 0000000..5c99526 --- /dev/null +++ b/integrations/aeoess-aps/tests/test_mapping.py @@ -0,0 +1,288 @@ +"""APS to TRACE mapping tests. + +Covers the mapping itself, the refusals that keep an unverified APS decision +from becoming a TRACE record, and a level 0 conformance run against the emitted +record. Decisions are minted in-process with ephemeral keys, so no network +access, no credentials and no committed fixtures are involved. + +A committed decision fixture is impossible here on purpose: APS decisions +expire five minutes after evaluation and the mapper refuses expired input. +""" +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import agentrust_trace +import pytest +from agent_passport.crypto import generate_key_pair +from agent_passport.policy import FloorValidatorV1, create_action_intent, evaluate_intent + +from aps_trace import EAT_PROFILE, VERDICT_TO_APPRAISAL, build_trace_record + +FLOOR_VERSION = "floor-1.0" + +#: TRACE v0.2 schema-required fields an APS policy decision cannot supply. +#: Pinned so the omission stays deliberate. See the README. +DOCUMENTED_ABSENT_REQUIRED = {"model", "data_class", "build_provenance"} + + +def _jwk() -> dict: + return agentrust_trace.key_to_jwk(agentrust_trace.generate_key()) + + +def _context(*, scope: list[str], spend_limit: int | None = None, spent: int = 0) -> dict: + delegation: dict = { + "scope": scope, + "revoked": False, + "expiresAt": (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat(), + "maxDepth": 3, + "currentDepth": 1, + } + if spend_limit is not None: + delegation["spendLimit"] = spend_limit + delegation["spentAmount"] = spent + return { + "floorVersion": FLOOR_VERSION, + "agentRegistered": True, + "agentAttestationValid": True, + "delegation": delegation, + } + + +def _decide(action: dict, context: dict, *, ttl_minutes: int = 5) -> dict: + agent = generate_key_pair() + evaluator = generate_key_pair() + intent = create_action_intent( + agent_id="agent_test", + agent_public_key=agent["publicKey"], + delegation_id="dlg_test", + action=action, + private_key=agent["privateKey"], + ) + return evaluate_intent( + intent=intent, + validator=FloorValidatorV1(), + validation_context=context, + evaluator_id="eval_test", + evaluator_public_key=evaluator["publicKey"], + evaluator_private_key=evaluator["privateKey"], + decision_ttl_minutes=ttl_minutes, + ) + + +def _permit() -> dict: + return _decide({"scopeRequired": "repo:read"}, _context(scope=["repo:read"])) + + +@pytest.fixture() +def permit_decision() -> dict: + return _permit() + + +@pytest.fixture() +def record(permit_decision: dict) -> dict: + return build_trace_record(permit_decision, trace_jwk=_jwk()) + + +# --- the decision the mapper consumes --------------------------------------- + +def test_permit_decision_has_the_expected_keys(permit_decision): + assert set(permit_decision) == { + "decisionId", "intentId", "evaluatorId", "evaluatorPublicKey", "verdict", + "principlesEvaluated", "constraints", "reason", "floorVersion", + "evaluatedAt", "expiresAt", "signature", + } + assert permit_decision["verdict"] == "permit" + + +# --- field mapping ----------------------------------------------------------- + +def test_eat_profile_is_the_v02_string(record): + assert record["eat_profile"] == "tag:agentrust-io.com,2026:trace-v0.2" + assert record["eat_profile"] == EAT_PROFILE + + +def test_subject_is_a_spiffe_uri_naming_evaluator_and_decision(record, permit_decision): + assert record["subject"].startswith("spiffe://agent-passport.org/") + assert permit_decision["evaluatorId"] in record["subject"] + assert permit_decision["decisionId"] in record["subject"] + + +def test_iat_comes_from_evaluated_at(record, permit_decision): + evaluated = datetime.fromisoformat(permit_decision["evaluatedAt"]) + assert record["iat"] == int(evaluated.timestamp()) + + +def test_policy_bundle_hash_is_a_sha256_digest(record): + assert record["policy"]["bundle_hash"].startswith("sha256:") + assert len(record["policy"]["bundle_hash"]) == len("sha256:") + 64 + + +def test_runtime_measurement_is_a_sha256_digest_and_platform_is_software_only(record): + assert record["runtime"]["platform"] == "software-only" + assert record["runtime"]["measurement"].startswith("sha256:") + assert len(record["runtime"]["measurement"]) == len("sha256:") + 64 + + +def test_bundle_hash_and_measurement_are_different_preimages(record): + assert record["policy"]["bundle_hash"] != record["runtime"]["measurement"] + + +def test_cnf_carries_the_supplied_trace_jwk(permit_decision): + jwk = _jwk() + rec = build_trace_record(permit_decision, trace_jwk=jwk) + assert rec["cnf"]["jwk"] == jwk + assert rec["cnf"]["jwk"]["kty"] == "OKP" + + +def test_appraisal_names_the_aps_evaluator(record, permit_decision): + assert record["appraisal"]["verifier"] == f"urn:aps:evaluator:{permit_decision['evaluatorId']}" + assert record["appraisal"]["policy_ref"] == f"urn:aps:floor:{FLOOR_VERSION}" + assert record["appraisal"]["timestamp"] == record["iat"] + + +def test_record_carries_no_aps_signature(record): + """The APS evaluator signature must not leak into the TRACE record.""" + assert "signature" not in record + assert "evaluatorPublicKey" not in record + + +# --- verdict to appraisal ---------------------------------------------------- + +def test_permit_maps_to_affirming(record): + assert record["appraisal"]["status"] == "affirming" + + +def test_narrow_maps_to_warning(): + decision = _decide( + {"scopeRequired": "repo:read", "spend": {"amount": 100, "currency": "USD"}}, + _context(scope=["repo:read"], spend_limit=50, spent=0), + ) + assert decision["verdict"] == "narrow" + assert build_trace_record(decision, trace_jwk=_jwk())["appraisal"]["status"] == "warning" + + +def test_deny_maps_to_contraindicated(): + decision = _decide({"scopeRequired": "repo:write"}, _context(scope=["repo:read"])) + assert decision["verdict"] == "deny" + assert build_trace_record(decision, trace_jwk=_jwk())["appraisal"]["status"] == "contraindicated" + + +def test_every_known_verdict_maps_to_a_valid_ear_status(): + valid = set(agentrust_trace.SCHEMA["properties"]["appraisal"]["properties"]["status"]["enum"]) + assert set(VERDICT_TO_APPRAISAL.values()) <= valid + + +def test_enforcement_mode_is_enforce_when_a_principle_is_inline(record, permit_decision): + modes = {p.get("enforcementMode") for p in permit_decision["principlesEvaluated"]} + assert "inline" in modes + assert record["policy"]["enforcement_mode"] == "enforce" + + +# --- refusals ---------------------------------------------------------------- + +def test_tampered_verdict_is_refused(permit_decision): + """Flipping the verdict breaks the evaluator signature, so no record is produced.""" + forged = dict(permit_decision, verdict="deny") + with pytest.raises(ValueError, match="failed verification"): + build_trace_record(forged, trace_jwk=_jwk()) + + +def test_corrupt_signature_is_refused(permit_decision): + bad = dict(permit_decision, signature="00" * 64) + with pytest.raises(ValueError, match="Invalid decision signature"): + build_trace_record(bad, trace_jwk=_jwk()) + + +def test_expired_decision_is_refused(): + expired = _decide({"scopeRequired": "repo:read"}, _context(scope=["repo:read"]), ttl_minutes=-1) + with pytest.raises(ValueError, match="expired"): + build_trace_record(expired, trace_jwk=_jwk()) + + +def test_missing_field_is_refused(permit_decision): + incomplete = {k: v for k, v in permit_decision.items() if k != "evaluatorPublicKey"} + with pytest.raises(ValueError, match="missing required fields"): + build_trace_record(incomplete, trace_jwk=_jwk()) + + +def test_unknown_verdict_is_refused(): + """A verdict this mapper cannot appraise is refused rather than guessed.""" + agent = generate_key_pair() + evaluator = generate_key_pair() + + class OddValidator(FloorValidatorV1): + def evaluate(self, intent, ctx): + result = super().evaluate(intent, ctx) + result["verdict"] = "escalate" + return result + + intent = create_action_intent( + agent_id="agent_test", + agent_public_key=agent["publicKey"], + delegation_id="dlg_test", + action={"scopeRequired": "repo:read"}, + private_key=agent["privateKey"], + ) + decision = evaluate_intent( + intent=intent, + validator=OddValidator(), + validation_context=_context(scope=["repo:read"]), + evaluator_id="eval_test", + evaluator_public_key=evaluator["publicKey"], + evaluator_private_key=evaluator["privateKey"], + ) + with pytest.raises(ValueError, match="unknown APS verdict"): + build_trace_record(decision, trace_jwk=_jwk()) + + +def test_non_dict_is_refused(): + with pytest.raises(ValueError, match="must be a dict"): + build_trace_record("not-a-decision", trace_jwk=_jwk()) # type: ignore[arg-type] + + +# --- TRACE sign and verify round-trip --------------------------------------- + +def test_sign_verify_roundtrip(record): + key = agentrust_trace.generate_key() + signed = agentrust_trace.sign_record(dict(record), key) + agentrust_trace.verify_record(signed, allow_embedded_key=True, max_age_seconds=None) + + +def test_tampered_signed_record_fails_verification(record): + key = agentrust_trace.generate_key() + signed = agentrust_trace.sign_record(dict(record), key) + signed["appraisal"]["status"] = "affirming" if signed["appraisal"]["status"] != "affirming" else "denying" + with pytest.raises(Exception): + agentrust_trace.verify_record(signed, allow_embedded_key=True, max_age_seconds=None) + + +# --- conformance ------------------------------------------------------------- + +def test_absent_schema_required_fields_are_exactly_the_documented_set(record): + """APS carries no model, data class or build provenance. Nothing is invented.""" + required = set(agentrust_trace.SCHEMA["required"]) + assert required - record.keys() == DOCUMENTED_ABSENT_REQUIRED + + +def test_present_fields_all_validate_against_the_v02_schema(record): + """Every field the mapper does emit must be schema-clean.""" + errors = agentrust_trace.iter_errors(record) + unexpected = [e for e in errors if e.message.split("'")[1::2][:1] != [] + and e.validator != "required"] + assert unexpected == [], [e.message for e in unexpected] + missing = {e.message.split("'")[1] for e in errors if e.validator == "required"} + assert missing == DOCUMENTED_ABSENT_REQUIRED + + +def test_record_passes_trace_tests_level_0(record): + """Level 0 must pass, with TR-SIG-005 UNVERIFIED on the unsigned record.""" + runner = pytest.importorskip("trace_tests.runner") + from trace_tests.result import Status + + results = runner.run(record, "trace", 0) + findings = [f for module in results.values() for f in module] + + assert [f.code for f in findings if f.status is Status.FAIL] == [] + unverified = [f for f in findings if f.status is Status.UNVERIFIED] + assert [f.code for f in unverified] == ["TR-SIG-005"] From f5295149be19092ea2ae8342d0821bfd70a3fa0a Mon Sep 17 00:00:00 2001 From: Tymofii Pidlisnyi <171286556+aeoess@users.noreply.github.com> Date: Mon, 3 Aug 2026 14:31:14 -0700 Subject: [PATCH 2/2] tests: assert the exact subject URI instead of a prefix and two substrings CodeQL flagged the startswith check as incomplete URL substring sanitization. The assertion now builds the expected spiffe URI from TRUST_DOMAIN and the quoted evaluator and decision ids and compares for equality, which also pins the path structure and the percent-encoding that the previous three assertions did not. --- integrations/aeoess-aps/tests/test_mapping.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/integrations/aeoess-aps/tests/test_mapping.py b/integrations/aeoess-aps/tests/test_mapping.py index 5c99526..d504a19 100644 --- a/integrations/aeoess-aps/tests/test_mapping.py +++ b/integrations/aeoess-aps/tests/test_mapping.py @@ -11,13 +11,14 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone +from urllib.parse import quote import agentrust_trace import pytest from agent_passport.crypto import generate_key_pair from agent_passport.policy import FloorValidatorV1, create_action_intent, evaluate_intent -from aps_trace import EAT_PROFILE, VERDICT_TO_APPRAISAL, build_trace_record +from aps_trace import EAT_PROFILE, TRUST_DOMAIN, VERDICT_TO_APPRAISAL, build_trace_record FLOOR_VERSION = "floor-1.0" @@ -103,9 +104,12 @@ def test_eat_profile_is_the_v02_string(record): def test_subject_is_a_spiffe_uri_naming_evaluator_and_decision(record, permit_decision): - assert record["subject"].startswith("spiffe://agent-passport.org/") - assert permit_decision["evaluatorId"] in record["subject"] - assert permit_decision["decisionId"] in record["subject"] + expected = ( + f"spiffe://{TRUST_DOMAIN}" + f"/evaluator/{quote(permit_decision['evaluatorId'], safe='')}" + f"/decision/{quote(permit_decision['decisionId'], safe='')}" + ) + assert record["subject"] == expected def test_iat_comes_from_evaluated_at(record, permit_decision):