From b6377103f98bb9273234d83a4a6a7282ff03f4b0 Mon Sep 17 00:00:00 2001 From: Carlos Hernandez Date: Fri, 17 Jul 2026 13:29:01 +0200 Subject: [PATCH 1/2] feat(codex): add AgenTrust agent integrity plugin --- .agents/plugins/marketplace.json | 20 + .github/workflows/agentrust-codex-tests.yml | 70 + README.md | 1 + .../agentrust-codex/.codex-plugin/plugin.json | 38 + plugins/agentrust-codex/.gitignore | 6 + plugins/agentrust-codex/PRIVACY.md | 45 + plugins/agentrust-codex/README.md | 141 ++ plugins/agentrust-codex/engine/capture.py | 1149 +++++++++++++++++ plugins/agentrust-codex/hooks/hooks.json | 18 + plugins/agentrust-codex/integration.yaml | 15 + plugins/agentrust-codex/requirements.txt | 5 + .../agentrust-codex/scripts/session-start.cmd | 15 + .../agentrust-codex/scripts/session-start.sh | 12 + .../agentrust-codex/scripts/signing-python.py | 69 + .../skills/agent-integrity/SKILL.md | 88 ++ plugins/agentrust-codex/tests/test_capture.py | 378 ++++++ .../tests/test_signing_python.py | 22 + .../tests/validate_structure.py | 48 + 18 files changed, 2140 insertions(+) create mode 100644 .agents/plugins/marketplace.json create mode 100644 .github/workflows/agentrust-codex-tests.yml create mode 100644 plugins/agentrust-codex/.codex-plugin/plugin.json create mode 100644 plugins/agentrust-codex/.gitignore create mode 100644 plugins/agentrust-codex/PRIVACY.md create mode 100644 plugins/agentrust-codex/README.md create mode 100644 plugins/agentrust-codex/engine/capture.py create mode 100644 plugins/agentrust-codex/hooks/hooks.json create mode 100644 plugins/agentrust-codex/integration.yaml create mode 100644 plugins/agentrust-codex/requirements.txt create mode 100644 plugins/agentrust-codex/scripts/session-start.cmd create mode 100644 plugins/agentrust-codex/scripts/session-start.sh create mode 100644 plugins/agentrust-codex/scripts/signing-python.py create mode 100644 plugins/agentrust-codex/skills/agent-integrity/SKILL.md create mode 100644 plugins/agentrust-codex/tests/test_capture.py create mode 100644 plugins/agentrust-codex/tests/test_signing_python.py create mode 100644 plugins/agentrust-codex/tests/validate_structure.py diff --git a/.agents/plugins/marketplace.json b/.agents/plugins/marketplace.json new file mode 100644 index 0000000..8b5357f --- /dev/null +++ b/.agents/plugins/marketplace.json @@ -0,0 +1,20 @@ +{ + "name": "agentrust", + "interface": { + "displayName": "AgenTrust" + }, + "plugins": [ + { + "name": "agentrust-codex", + "source": { + "source": "local", + "path": "./plugins/agentrust-codex" + }, + "policy": { + "installation": "AVAILABLE", + "authentication": "ON_INSTALL" + }, + "category": "Security" + } + ] +} diff --git a/.github/workflows/agentrust-codex-tests.yml b/.github/workflows/agentrust-codex-tests.yml new file mode 100644 index 0000000..8acec8f --- /dev/null +++ b/.github/workflows/agentrust-codex-tests.yml @@ -0,0 +1,70 @@ +name: agentrust-codex tests + +on: + pull_request: + paths: + - ".agents/plugins/marketplace.json" + - "plugins/agentrust-codex/**" + - ".github/workflows/agentrust-codex-tests.yml" + - "README.md" + push: + branches: [main] + paths: + - ".agents/plugins/marketplace.json" + - "plugins/agentrust-codex/**" + - ".github/workflows/agentrust-codex-tests.yml" + - "README.md" + +permissions: + contents: read + +jobs: + stdlib: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.9", "3.11", "3.13"] + steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + persist-credentials: false + - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0 + with: + python-version: "${{ matrix.python-version }}" + - name: Run dependency-free capture tests + run: | + pip install pytest + python -m pytest plugins/agentrust-codex/tests -q + + signing: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.11", "3.12", "3.13"] + steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + persist-credentials: false + - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0 + with: + python-version: "${{ matrix.python-version }}" + - name: Run signing and conformance tests + run: | + pip install pytest -r plugins/agentrust-codex/requirements.txt + python -m pytest plugins/agentrust-codex/tests -q + + structure: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + persist-credentials: false + - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0 + with: + python-version: "3.12" + - name: Validate integration and plugin structure + run: | + pip install jsonschema pyyaml + python plugins/agentrust-codex/tests/validate_structure.py diff --git a/README.md b/README.md index aa8cf1b..520a01d 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,7 @@ TRACE only works as a standard if it is genuinely neutral. Integrations are list | Integration | Vendor | Integrates with | Tier | |---|---|---|---| | [claude-code](claude-code/) | agentrust-io | agent-manifest, trace | community | +| [agentrust-codex](plugins/agentrust-codex/) | agentrust-io | agent-manifest, trace | community | ## Community diff --git a/plugins/agentrust-codex/.codex-plugin/plugin.json b/plugins/agentrust-codex/.codex-plugin/plugin.json new file mode 100644 index 0000000..f5e63f8 --- /dev/null +++ b/plugins/agentrust-codex/.codex-plugin/plugin.json @@ -0,0 +1,38 @@ +{ + "name": "agentrust-codex", + "version": "0.1.0", + "description": "Detect Codex agent-composition drift and emit signed Agent Manifest and TRACE session records.", + "author": { + "name": "agentrust-io", + "url": "https://github.com/agentrust-io" + }, + "homepage": "https://github.com/agentrust-io/integrations/tree/main/plugins/agentrust-codex", + "repository": "https://github.com/agentrust-io/integrations", + "license": "Apache-2.0", + "keywords": [ + "codex", + "agent-integrity", + "agent-manifest", + "trace", + "governance" + ], + "skills": "./skills/", + "interface": { + "displayName": "AgenTrust for Codex", + "shortDescription": "Detect agent drift and create signed session records", + "longDescription": "Fingerprint Codex configuration at session start, compare it with a workspace baseline, and create signed Agent Manifest and TRACE Level 0 records on request.", + "developerName": "agentrust-io", + "category": "Developer Tools", + "capabilities": [ + "Local configuration integrity", + "Signed governance records" + ], + "websiteURL": "https://github.com/agentrust-io/integrations/tree/main/plugins/agentrust-codex", + "privacyPolicyURL": "https://github.com/agentrust-io/integrations/blob/main/plugins/agentrust-codex/PRIVACY.md", + "defaultPrompt": [ + "Verify my Codex agent against its AgenTrust baseline.", + "Approve the current Codex agent baseline.", + "Generate signed Agent Manifest and TRACE records." + ] + } +} diff --git a/plugins/agentrust-codex/.gitignore b/plugins/agentrust-codex/.gitignore new file mode 100644 index 0000000..d2f06af --- /dev/null +++ b/plugins/agentrust-codex/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +.pytest_cache/ +*.py[cod] +manifest.json +trace.json +verification_key.json diff --git a/plugins/agentrust-codex/PRIVACY.md b/plugins/agentrust-codex/PRIVACY.md new file mode 100644 index 0000000..5b38f02 --- /dev/null +++ b/plugins/agentrust-codex/PRIVACY.md @@ -0,0 +1,45 @@ +# Privacy + +The AgenTrust for Codex plugin processes agent configuration on your machine. +It sends no telemetry, analytics, snapshots, baselines, signing keys, or +records to agentrust-io. + +## Files it processes + +The plugin reads selected Codex configuration and instruction files to compute +SHA-256 hashes: + +- `AGENTS.md`, `AGENTS.override.md`, and the Codex memory summary +- skill definitions, plugin manifests, plugin hooks, and plugin scripts +- `config.toml`, `hooks.json`, `requirements.toml`, and rule files + +It parses config files only to extract plugin names, MCP server names, approval +policy, sandbox mode, and enabled state. It does not store raw file contents or +other config values. + +The SessionStart hook supplies the active model, permission mode, workspace, +and session ID. The plugin stores a hash of the workspace path and session ID. +It derives a pseudonymous local agent identifier from a hash of the username +and hostname. + +The plugin does not open Codex `auth.json`, transcript files, tool inputs, +tool outputs, or environment secrets. + +## Local data + +The plugin writes baselines, latest snapshots, and its signing key below +`$CODEX_HOME/agentrust`, which defaults to `~/.codex/agentrust`. It requests +owner-only permissions for private state. Signed reports go to the output +directory you choose. + +The plugin makes no network request during SessionStart, snapshot, verify, or +approve. A signed-report request can run the included bootstrap, which installs +the exact released packages in `requirements.txt` from PyPI into a dedicated +local virtual environment. + +Removing the plugin does not delete `$CODEX_HOME/agentrust`. This preserves +your baseline and signing identity if you reinstall it. You control removal of +that local state. + +Questions and corrections: +https://github.com/agentrust-io/integrations/issues diff --git a/plugins/agentrust-codex/README.md b/plugins/agentrust-codex/README.md new file mode 100644 index 0000000..7aef85c --- /dev/null +++ b/plugins/agentrust-codex/README.md @@ -0,0 +1,141 @@ +# AgenTrust for Codex + +A Codex plugin that fingerprints the agent configuration in each workspace, +warns when that composition changes, and creates signed Agent Manifest and +TRACE Level 0 records on request. + +## Install + +You need a Codex release with plugin and hook support. Drift detection supports +Python 3.9 or newer. Signed-record generation requires Python 3.11 or newer. + +```bash +codex plugin marketplace add agentrust-io/integrations +codex plugin add agentrust-codex@agentrust +``` + +Start a new Codex session after installation. Open `/hooks`, review the +AgenTrust SessionStart hook, and trust it. Codex skips plugin hooks until you +trust their current definition. + +The first trusted session records a baseline for the current Git workspace. +Later sessions stay quiet when the fingerprints match. Drift produces a +warning that names each changed category. + +## Use it + +The plugin packages the `agentrust-codex:agent-integrity` skill. Ask Codex: + +```text +Use $agentrust-codex:agent-integrity to verify this Codex agent. +``` + +```text +Use $agentrust-codex:agent-integrity to approve the current baseline. +``` + +```text +Use $agentrust-codex:agent-integrity to generate signed records in ./agentrust-records. +``` + +Approval replaces the workspace baseline. The skill requires an explicit +approval request and leaves unexpected drift unapproved. + +Signed-record generation, invoked with Python 3.11 or newer, creates a pinned +Python environment under +`$CODEX_HOME/agentrust/signing-venv` when needed. The plugin installs released +versions of `agent-manifest`, `agentrust-trace`, and +`agentrust-trace-tests` in that environment. + +## Fingerprinted composition + +The engine hashes these local inputs: + +- the active user and workspace `AGENTS.md` or `AGENTS.override.md` files, + plus the Codex memory summary when present +- user, workspace, and enabled-plugin skill definitions +- Codex config, hooks, requirements, and rule files +- installed enabled plugins and configured MCP server names +- the active model and permission mode supplied by the SessionStart hook + +Snapshots store logical names and SHA-256 hashes. They do not store raw prompt +text, config values, tool inputs, tool outputs, transcripts, auth data, or +environment secrets. + +Each Git root gets its own baseline: + +```text +$CODEX_HOME/agentrust/workspaces//baseline.json +``` + +The plugin hashes the absolute Git-root path to select that directory but does +not store the path. It hashes the Codex session ID before storing it. It also +uses a pseudonymous hash for the local agent identity. + +## Signed records + +The report command writes: + +- `manifest.json`, an Ed25519-signed Agent Manifest +- `trace.json`, a signed TRACE Level 0 session-context record +- `verification_key.json`, the public key for independent signature checks + +The plugin keeps one manifest signing key at +`$CODEX_HOME/agentrust/signing_key.json`. It writes the key with owner-only +permissions where the operating system supports them. A corrupt key stops +signing so the plugin cannot replace an existing identity without notice. + +Check TRACE conformance from a source checkout: + +```bash +python3 -m venv .venv-agentrust-codex +.venv-agentrust-codex/bin/pip install -r plugins/agentrust-codex/requirements.txt +.venv-agentrust-codex/bin/python plugins/agentrust-codex/engine/capture.py \ + report --model gpt-test --out ./agentrust-records +.venv-agentrust-codex/bin/trace-tests verify \ + --record ./agentrust-records/trace.json --level 0 +``` + +Use the real active model slug for a real record. The `gpt-test` value above +exists only to make a local smoke-test command reproducible. + +## Limits + +- TRACE Level 0 represents software integrity. It provides no TEE or + hardware-attestation claim. +- The instruction fingerprint covers visible `AGENTS.md` layers and the + local memory summary, not Codex's internal system prompt. +- A third party can verify the manifest signature with + `verification_key.json`. Verifying each artifact binding requires runtime + hashes measured outside the signed manifest. +- SessionStart exposes the model and permission mode but no complete built-in + tool roster. The tool catalog covers configured MCP servers, installed + enabled plugins, and any tools supplied through the engine's explicit CLI + option. +- The engine fingerprints workspace `.codex` files it can see on disk. It + cannot prove that Codex trusted and loaded each project layer. +- Codex may update the memory summary as you work. Such an update changes the + instruction fingerprint and requires review or approval. + +## Test + +```bash +python3 -m pytest plugins/agentrust-codex/tests -q +python3 /path/to/plugin-creator/scripts/validate_plugin.py \ + plugins/agentrust-codex +``` + +The full test suite verifies workspace isolation, drift detection, hook failure +safety, corrupt-state handling, persistent signing identity, independent +manifest signature verification, tamper detection, and TRACE Level 0 +conformance. + +## Privacy + +Read [PRIVACY.md](PRIVACY.md). The plugin processes configuration on the local +machine and sends no telemetry. The signing-environment bootstrap accesses PyPI +only after a signed-report request. + +## License + +Apache-2.0. diff --git a/plugins/agentrust-codex/engine/capture.py b/plugins/agentrust-codex/engine/capture.py new file mode 100644 index 0000000..3d68498 --- /dev/null +++ b/plugins/agentrust-codex/engine/capture.py @@ -0,0 +1,1149 @@ +"""AgenTrust capture engine for OpenAI Codex. + +The SessionStart path uses the Python standard library. It fingerprints Codex +configuration and stores names plus hashes, never file contents, tokens, +environment values, auth data, or transcripts. Signing imports the released +Agent Manifest and TRACE packages only when a user requests a report. +""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import json +import os +import platform +import re +import socket +import stat +import sys +import tempfile +import time +import uuid +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Set, Tuple + + +VERSION = "0.1.0" +HOME = Path.home() +CODEX_HOME = Path(os.environ.get("CODEX_HOME", str(HOME / ".codex"))).expanduser() +STATE_DIR = Path( + os.environ.get("AGENTRUST_STATE_DIR", str(CODEX_HOME / "agentrust")) +).expanduser() +SIGNING_KEY = STATE_DIR / "signing_key.json" + +_CONFIG_VALUE_KEYS = ("approval_policy", "sandbox_mode") +_LIVE_KEYS = { + "builtin_tools", + "capability_level", + "cwd", + "data_class", + "mcp_servers", + "model_id", + "model_provider", + "model_version", + "permission_mode", + "session_id_hash", +} + + +def _sha_bytes(value: bytes) -> str: + return "sha256:" + hashlib.sha256(value).hexdigest() + + +def _sha_file(path: Path) -> str: + return _sha_bytes(path.read_bytes()) + + +def _sha_mapping(value: Mapping[str, Any]) -> str: + encoded = json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8") + return _sha_bytes(encoded) + + +def _safe_hash(path: Path) -> Optional[str]: + try: + if path.is_file() and not path.is_symlink(): + return _sha_file(path) + except OSError: + return None + return None + + +def _now_iso() -> str: + return ( + datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") + ) + + +def _uuid7() -> str: + milliseconds = int(time.time() * 1000) + raw = bytearray(milliseconds.to_bytes(6, "big") + os.urandom(10)) + raw[6] = 0x70 | (raw[6] & 0x0F) + raw[8] = 0x80 | (raw[8] & 0x3F) + return str(uuid.UUID(bytes=bytes(raw))) + + +def _slug(value: str) -> str: + normalized = re.sub(r"[^a-z0-9._-]+", "-", value.lower()).strip("-") + return normalized or "unknown" + + +def _workspace(cwd: Optional[str] = None) -> Tuple[Path, List[Path], str]: + try: + current = Path(cwd or os.getcwd()).expanduser().resolve() + except (OSError, RuntimeError): + current = Path(os.getcwd()) + if not current.is_dir(): + current = current.parent + + upward = [current] + list(current.parents) + root = next((path for path in upward if (path / ".git").exists()), current) + chain: List[Path] = [] + cursor = current + while True: + chain.append(cursor) + if cursor == root: + break + if root not in cursor.parents: + chain = [current] + root = current + break + cursor = cursor.parent + chain.reverse() + workspace_id = hashlib.sha256(str(root).encode("utf-8")).hexdigest() + return root, chain, workspace_id + + +def _workspace_state(workspace_id: str) -> Tuple[Path, Path]: + root = STATE_DIR / "workspaces" / workspace_id + return root / "baseline.json", root / "session-latest.json" + + +def _logical_workspace_path(root: Path, path: Path) -> str: + try: + relative = path.relative_to(root).as_posix() + except ValueError: + relative = path.name + return "workspace:" + (relative or ".") + + +def _instruction_fingerprints(root: Path, chain: Sequence[Path]) -> Dict[str, str]: + found: Dict[str, str] = {} + + global_override = CODEX_HOME / "AGENTS.override.md" + global_default = CODEX_HOME / "AGENTS.md" + global_active = global_override if global_override.is_file() else global_default + digest = _safe_hash(global_active) + if digest: + found["user:" + global_active.name] = digest + + memory_summary = CODEX_HOME / "memories" / "memory_summary.md" + digest = _safe_hash(memory_summary) + if digest: + found["user:memories/memory_summary.md"] = digest + + for directory in chain: + override = directory / "AGENTS.override.md" + default = directory / "AGENTS.md" + active = override if override.is_file() else default + digest = _safe_hash(active) + if digest: + found[_logical_workspace_path(root, active)] = digest + return dict(sorted(found.items())) + + +def _policy_fingerprints(root: Path, chain: Sequence[Path]) -> Dict[str, str]: + found: Dict[str, str] = {} + user_files = ( + CODEX_HOME / "config.toml", + CODEX_HOME / "hooks.json", + CODEX_HOME / "requirements.toml", + ) + for path in user_files: + digest = _safe_hash(path) + if digest: + found["user:" + path.name] = digest + + user_rules = CODEX_HOME / "rules" + if user_rules.is_dir(): + for path in sorted(user_rules.rglob("*")): + digest = _safe_hash(path) + if digest: + found["user:rules/" + path.relative_to(user_rules).as_posix()] = digest + + for directory in chain: + layer = directory / ".codex" + for filename in ("config.toml", "hooks.json", "requirements.toml"): + path = layer / filename + digest = _safe_hash(path) + if digest: + found[_logical_workspace_path(root, path)] = digest + rules = layer / "rules" + if rules.is_dir(): + for path in sorted(rules.rglob("*")): + digest = _safe_hash(path) + if digest: + found[_logical_workspace_path(root, path)] = digest + return dict(sorted(found.items())) + + +def _skill_roots(chain: Sequence[Path]) -> List[Tuple[str, Path]]: + roots: List[Tuple[str, Path]] = [ + ("user:agents", HOME / ".agents" / "skills"), + ("user:codex", CODEX_HOME / "skills"), + ] + for index, directory in enumerate(chain): + roots.append(("workspace:%d" % index, directory / ".agents" / "skills")) + return roots + + +def _skill_fingerprints(chain: Sequence[Path]) -> Dict[str, str]: + found: Dict[str, str] = {} + for prefix, root in _skill_roots(chain): + if not root.is_dir(): + continue + try: + candidates = sorted(root.rglob("SKILL.md")) + except OSError: + continue + for path in candidates: + digest = _safe_hash(path) + if not digest: + continue + try: + relative = path.parent.relative_to(root).as_posix() + except ValueError: + relative = path.parent.name + found["%s:%s" % (prefix, relative)] = digest + return dict(sorted(found.items())) + + +def _read_toml(path: Path) -> Optional[Dict[str, Any]]: + try: + import tomllib # type: ignore[import-not-found] + except ImportError: + return None + try: + data = tomllib.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + return data if isinstance(data, dict) else None + + +def _fallback_toml_state(path: Path) -> Tuple[Set[str], Set[str], Dict[str, str]]: + mcp: Set[str] = set() + plugins: Set[str] = set() + scopes: Dict[str, str] = {} + try: + lines = path.read_text(encoding="utf-8").splitlines() + except OSError: + return mcp, plugins, scopes + + header = re.compile( + r'^\s*\[\s*(mcp_servers|plugins)\.(?:"([^"]+)"|([A-Za-z0-9_.@/-]+))\s*\]\s*$' + ) + value_line = re.compile( + r'^\s*(approval_policy|sandbox_mode)\s*=\s*["\']([^"\']+)["\']\s*$' + ) + enabled_line = re.compile(r"^\s*enabled\s*=\s*(true|false)\s*$", re.IGNORECASE) + current_plugin: Optional[str] = None + for line in lines: + match = header.match(line) + if match: + name = match.group(2) or match.group(3) + current_plugin = name if match.group(1) == "plugins" else None + (mcp if match.group(1) == "mcp_servers" else plugins).add(name) + continue + match = enabled_line.match(line) + if match and current_plugin and match.group(1).lower() == "false": + plugins.discard(current_plugin) + continue + match = value_line.match(line) + if match: + scopes[match.group(1)] = match.group(2) + return mcp, plugins, scopes + + +def _config_state(paths: Iterable[Path]) -> Tuple[List[str], List[str], Dict[str, str]]: + mcp: Set[str] = set() + plugins: Set[str] = set() + scopes: Dict[str, str] = {} + for path in paths: + if not path.is_file(): + continue + data = _read_toml(path) + if data is None: + fallback_mcp, fallback_plugins, fallback_scopes = _fallback_toml_state(path) + mcp.update(fallback_mcp) + plugins.update(fallback_plugins) + scopes.update(fallback_scopes) + continue + + server_map = data.get("mcp_servers") + if isinstance(server_map, dict): + mcp.update(str(name) for name in server_map) + + plugin_map = data.get("plugins") + if isinstance(plugin_map, dict): + for name, config in plugin_map.items(): + if isinstance(config, dict) and config.get("enabled") is False: + continue + plugins.add(str(name)) + + for key in _CONFIG_VALUE_KEYS: + value = data.get(key) + if isinstance(value, (str, int, float, bool)): + scopes[key] = str(value) + return sorted(mcp), sorted(plugins), dict(sorted(scopes.items())) + + +def _plugin_root(plugin_ref: str) -> Optional[Path]: + if "@" not in plugin_ref: + return None + name, marketplace = plugin_ref.rsplit("@", 1) + cache = CODEX_HOME / "plugins" / "cache" + bases = ( + cache / marketplace / name, + cache / (marketplace + "-remote") / name, + ) + candidates: List[Path] = [] + for base in bases: + if not base.is_dir(): + continue + try: + candidates.extend(path for path in base.iterdir() if path.is_dir()) + except OSError: + continue + if not candidates: + return None + try: + return max(candidates, key=lambda path: path.stat().st_mtime_ns) + except OSError: + return sorted(candidates, key=lambda path: path.name)[-1] + + +def _plugin_digest(root: Path) -> str: + selected: List[Path] = [] + direct = ( + root / ".codex-plugin" / "plugin.json", + root / ".mcp.json", + root / ".app.json", + ) + selected.extend(path for path in direct if path.is_file()) + for pattern in ( + "skills/**/SKILL.md", + "hooks/**/*.json", + "scripts/**/*", + "engine/**/*.py", + ): + try: + selected.extend(path for path in root.glob(pattern) if path.is_file()) + except OSError: + continue + + digest = hashlib.sha256() + for path in sorted(set(selected)): + if path.is_symlink(): + continue + try: + relative = path.relative_to(root).as_posix().encode("utf-8") + content = path.read_bytes() + except (OSError, ValueError): + continue + digest.update(relative) + digest.update(b"\0") + digest.update(content) + digest.update(b"\0") + return "sha256:" + digest.hexdigest() + + +def _plugin_fingerprints( + plugin_refs: Sequence[str], +) -> Tuple[Dict[str, str], Dict[str, str]]: + plugins: Dict[str, str] = {} + skills: Dict[str, str] = {} + for plugin_ref in plugin_refs: + root = _plugin_root(plugin_ref) + if root is None: + continue + plugins[plugin_ref] = _plugin_digest(root) + skill_root = root / "skills" + if not skill_root.is_dir(): + continue + try: + candidates = sorted(skill_root.rglob("SKILL.md")) + except OSError: + continue + for path in candidates: + digest = _safe_hash(path) + if digest: + try: + name = path.parent.relative_to(skill_root).as_posix() + except ValueError: + name = path.parent.name + skills["plugin:%s:%s" % (plugin_ref, name)] = digest + return dict(sorted(plugins.items())), dict(sorted(skills.items())) + + +def _identity() -> str: + host = ( + os.environ.get("COMPUTERNAME") + or os.environ.get("HOSTNAME") + or socket.gethostname() + ) + user = os.environ.get("USERNAME") or os.environ.get("USER") or "user" + pseudonym = hashlib.sha256((user + "\0" + host).encode("utf-8")).hexdigest()[:32] + return "spiffe://codex.local/host/%s" % pseudonym + + +def _runtime_hash() -> str: + engine_hash = _safe_hash(Path(__file__)) or _sha_bytes(b"unknown-engine") + facts = { + "engine": engine_hash, + "machine": platform.machine(), + "python": "%d.%d" % sys.version_info[:2], + "system": platform.system(), + "version": VERSION, + } + return _sha_mapping(facts) + + +def _sanitize_live(live: Optional[Mapping[str, Any]]) -> Dict[str, Any]: + if not isinstance(live, Mapping): + return {} + cleaned = {key: live[key] for key in _LIVE_KEYS if key in live} + for key in ("builtin_tools", "mcp_servers"): + value = cleaned.get(key) + if not isinstance(value, list): + cleaned.pop(key, None) + else: + cleaned[key] = sorted({str(item) for item in value if str(item).strip()}) + return cleaned + + +def snapshot(live: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: + live_state = _sanitize_live(live) + root, chain, workspace_id = _workspace( + str(live_state.get("cwd")) if live_state.get("cwd") else None + ) + instructions = _instruction_fingerprints(root, chain) + policy_files = _policy_fingerprints(root, chain) + config_paths = [CODEX_HOME / "config.toml"] + [ + directory / ".codex" / "config.toml" for directory in chain + ] + configured_mcp, enabled_plugin_refs, config_scope = _config_state(config_paths) + plugins, plugin_skills = _plugin_fingerprints(enabled_plugin_refs) + skills = _skill_fingerprints(chain) + skills.update(plugin_skills) + skills = dict(sorted(skills.items())) + + permission_mode = str(live_state.get("permission_mode", "unknown")) + model_id = str(live_state.get("model_id", "unknown")) + model_version = str(live_state.get("model_version", model_id)) + model_provider = str(live_state.get("model_provider", "openai")) + live_mcp = live_state.get("mcp_servers") + builtin_tools = list(live_state.get("builtin_tools") or []) + tools = sorted( + set(builtin_tools) + | {"mcp:" + name for name in configured_mcp} + | {"plugin:" + name for name in plugins} + ) + + observed = ["instructions", "mcp_config", "plugins", "policy", "skills"] + if model_id != "unknown": + observed.append("model") + if permission_mode != "unknown": + observed.append("permission") + if isinstance(live_mcp, list): + observed.append("mcp_live") + if builtin_tools: + observed.append("tools") + + policy_scope = dict(config_scope) + if permission_mode != "unknown": + policy_scope["permission_mode"] = permission_mode + policy_material = { + "files": policy_files, + "mcp": configured_mcp, + "plugins": plugins, + "scope": policy_scope, + } + runtime_context = { + "capability_level": live_state.get("capability_level"), + "data_class": str(live_state.get("data_class", "internal")), + "model_id": model_id, + "model_provider": model_provider, + "model_version": model_version, + "permission_mode": permission_mode, + } + if isinstance(live_mcp, list): + runtime_context["mcp_servers"] = sorted(live_mcp) + if builtin_tools: + runtime_context["builtin_tools"] = sorted(builtin_tools) + if live_state.get("session_id_hash"): + runtime_context["session_id_hash"] = str(live_state["session_id_hash"]) + + return { + "captured_at": _now_iso(), + "workspace_id": workspace_id, + "observed": sorted(observed), + "agent_id": _identity(), + "model": { + "provider": model_provider, + "model_id": model_id, + "version": model_version, + "capability_level": live_state.get("capability_level"), + }, + "permission_mode": permission_mode, + "instructions": instructions, + "skills": skills, + "plugins": plugins, + "policy_files": policy_files, + "policy_scope": policy_scope, + "configured_mcp_servers": configured_mcp, + "live_mcp_servers": sorted(live_mcp) if isinstance(live_mcp, list) else [], + "tools": tools, + "data_class": str(live_state.get("data_class", "internal")), + "runtime_context": runtime_context, + "hashes": { + "system_prompt": _sha_mapping(instructions), + "policy_bundle": _sha_mapping(policy_material), + "skills_set": _sha_mapping(skills), + "plugin_set": _sha_mapping(plugins), + "tool_catalog": _sha_mapping({name: True for name in tools}), + "runtime": _runtime_hash(), + }, + } + + +def _map_changes( + before: Mapping[str, str], + after: Mapping[str, str], + label: str, +) -> List[Dict[str, str]]: + changes: List[Dict[str, str]] = [] + before_keys, after_keys = set(before), set(after) + for name in sorted(after_keys - before_keys): + changes.append({"change": "added", "what": label, "detail": name}) + for name in sorted(before_keys - after_keys): + changes.append({"change": "removed", "what": label, "detail": name}) + for name in sorted(before_keys & after_keys): + if before[name] != after[name]: + changes.append({"change": "changed", "what": label, "detail": name}) + return changes + + +def _set_changes( + before: Iterable[str], + after: Iterable[str], + label: str, +) -> List[Dict[str, str]]: + changes: List[Dict[str, str]] = [] + before_set, after_set = set(before), set(after) + for name in sorted(after_set - before_set): + changes.append({"change": "added", "what": label, "detail": name}) + for name in sorted(before_set - after_set): + changes.append({"change": "removed", "what": label, "detail": name}) + return changes + + +def diff(base: Mapping[str, Any], current: Mapping[str, Any]) -> List[Dict[str, str]]: + common = set(base.get("observed", [])) & set(current.get("observed", [])) + changes: List[Dict[str, str]] = [] + if "instructions" in common: + changes += _map_changes( + base.get("instructions", {}), current.get("instructions", {}), "instruction" + ) + if "policy" in common: + changes += _map_changes( + base.get("policy_files", {}), current.get("policy_files", {}), "policy file" + ) + before_scope = base.get("policy_scope", {}) + after_scope = current.get("policy_scope", {}) + changes += _map_changes(before_scope, after_scope, "policy setting") + if "skills" in common: + changes += _map_changes( + base.get("skills", {}), current.get("skills", {}), "skill" + ) + if "plugins" in common: + changes += _map_changes( + base.get("plugins", {}), current.get("plugins", {}), "plugin" + ) + if "mcp_config" in common: + changes += _set_changes( + base.get("configured_mcp_servers", []), + current.get("configured_mcp_servers", []), + "configured MCP server", + ) + if "mcp_live" in common: + changes += _set_changes( + base.get("live_mcp_servers", []), + current.get("live_mcp_servers", []), + "live MCP server", + ) + if "tools" in common: + changes += _set_changes(base.get("tools", []), current.get("tools", []), "tool") + if "model" in common and base.get("model") != current.get("model"): + before = base.get("model", {}).get("model_id", "unknown") + after = current.get("model", {}).get("model_id", "unknown") + changes.append( + { + "change": "changed", + "what": "model", + "detail": "%s -> %s" % (before, after), + } + ) + if "permission" in common and base.get("permission_mode") != current.get( + "permission_mode" + ): + changes.append( + { + "change": "changed", + "what": "permission mode", + "detail": "%s -> %s" + % ( + base.get("permission_mode", "unknown"), + current.get("permission_mode", "unknown"), + ), + } + ) + return changes + + +def _enforcement_mode(permission_mode: str) -> str: + if permission_mode == "bypassPermissions": + return "audit-only" + if permission_mode == "unknown": + return "advisory" + return "enforce" + + +def _trace_enforcement_mode(permission_mode: str) -> str: + if permission_mode == "bypassPermissions": + return "silent" + if permission_mode == "unknown": + return "advisory" + return "enforce" + + +def build_manifest(current: Mapping[str, Any]) -> Dict[str, Any]: + now = _now_iso() + expires = ( + (datetime.now(timezone.utc) + timedelta(days=90)) + .isoformat(timespec="seconds") + .replace("+00:00", "Z") + ) + model = current["model"] + tools = [] + for name in current.get("tools", []): + kind = ( + "mcp" + if name.startswith("mcp:") + else "plugin" + if name.startswith("plugin:") + else "builtin" + ) + clean_name = name.split(":", 1)[1] if ":" in name else name + tools.append( + { + "tool_id": "codex.%s.%s" % (kind, _slug(clean_name)), + "tool_name": name, + "endpoint_id": "spiffe://codex.local/%s/%s" % (kind, _slug(clean_name)), + "schema_hash": _sha_bytes(name.encode("utf-8")), + "description_hash": _sha_bytes(("codex:" + name).encode("utf-8")), + "version": "1.0.0", + "egress_destinations": [], + } + ) + + scope = [ + "codex:%s=%s" % (key, value) + for key, value in sorted(current.get("policy_scope", {}).items()) + ] or ["codex:configuration"] + return { + "@context": "https://agentmanifest.agentrust.io/v0.1/context.json", + "@type": "AgentManifest", + "manifest_id": _uuid7(), + "agent_id": current["agent_id"], + "version": "0.1", + "issued_at": now, + "expires_at": expires, + "issuer": "spiffe://codex.local/self-signed", + "crypto_profile": "standard", + "artifacts": { + "system_prompt": { + "hash": current["hashes"]["system_prompt"], + "hash_algorithm": "SHA-256", + "version": "1.0.0", + "classification": "internal", + "bound_at": now, + }, + "policy_bundle": { + "hash": current["hashes"]["policy_bundle"], + "policy_language": "composite", + "version": "1.0.0", + "enforcement_mode": _enforcement_mode( + current.get("permission_mode", "unknown") + ), + "scope": scope, + "agt_version": "0.0.0-codex", + "bound_at": now, + }, + "tool_manifest": { + "catalog_hash": current["hashes"]["tool_catalog"], + "tools": tools, + "allow_dynamic_registration": True, + "rug_pull_policy": "deny-and-alert", + "bound_at": now, + }, + "model_identity": { + "provider": model["provider"], + "model_id": model["model_id"], + "version": model["version"], + "capability_level": model.get("capability_level"), + "deployment_type": "api", + "model_attestation_type": "provider-asserted", + "bound_at": now, + }, + }, + } + + +def build_trace(current: Mapping[str, Any]) -> Dict[str, Any]: + return { + "eat_profile": "tag:agentrust.io,2026:trace-v0.1", + "iat": int(time.time()), + "subject": current["agent_id"], + "model": { + key: current["model"][key] for key in ("provider", "model_id", "version") + }, + "runtime": { + "platform": "software-only", + "measurement": current["hashes"]["runtime"], + }, + "policy": { + "bundle_hash": current["hashes"]["policy_bundle"], + "enforcement_mode": _trace_enforcement_mode( + current.get("permission_mode", "unknown") + ), + }, + "data_class": current["data_class"], + "build_provenance": { + "slsa_level": 0, + "digest": current["hashes"]["plugin_set"], + }, + "appraisal": {"status": "none", "verifier": "https://codex.local"}, + "transparency": "https://registry.agentrust.io/claim/placeholder", + } + + +def _atomic_write(path: Path, content: str, mode: int = 0o600) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary: Optional[str] = None + try: + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=str(path.parent), + prefix=".%s." % path.name, + delete=False, + ) as handle: + handle.write(content) + handle.flush() + os.fsync(handle.fileno()) + temporary = handle.name + try: + os.chmod(temporary, mode) + except OSError: + pass + os.replace(temporary, path) + finally: + if temporary and os.path.exists(temporary): + try: + os.unlink(temporary) + except OSError: + pass + + +def _save(path: Path, value: Mapping[str, Any], mode: int = 0o600) -> None: + _atomic_write(path, json.dumps(value, indent=2, sort_keys=True) + "\n", mode) + + +def _load(path: Path) -> Tuple[Optional[Dict[str, Any]], str]: + if not path.is_file(): + return None, "missing" + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None, "corrupt" + return (data, "ok") if isinstance(data, dict) else (None, "corrupt") + + +def _load_or_create_manifest_keypair(): + from agent_manifest import Ed25519KeyPair, generate_ed25519 + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + existing, status = _load(SIGNING_KEY) + if status == "corrupt": + raise SystemExit( + "The AgenTrust signing key is unreadable. Restore it from backup or move it " + "aside before generating a new identity: %s" % SIGNING_KEY + ) + if existing and isinstance(existing.get("private_b64url"), str): + try: + padding = "=" * (-len(existing["private_b64url"]) % 4) + raw = base64.urlsafe_b64decode(existing["private_b64url"] + padding) + private = Ed25519PrivateKey.from_private_bytes(raw) + return Ed25519KeyPair(private, private.public_key()) + except (TypeError, ValueError): + raise SystemExit( + "The AgenTrust signing key is invalid. Restore it or move it aside: %s" + % SIGNING_KEY + ) + + keypair = generate_ed25519() + _save( + SIGNING_KEY, + { + "algorithm": "Ed25519", + "key_id": keypair.key_id, + "public_key_b64url": keypair.public_b64url(), + "private_b64url": keypair.private_b64url(), + "created_at": _now_iso(), + }, + stat.S_IRUSR | stat.S_IWUSR, + ) + return keypair + + +def sign_all( + current: Mapping[str, Any], outdir: Path +) -> Tuple[Dict[str, Any], Dict[str, Any]]: + if current["model"]["model_id"] == "unknown": + raise SystemExit( + "A signed record needs the active Codex model. Start a session with the " + "AgenTrust hook enabled, or pass --model." + ) + try: + from agent_manifest import Ed25519Signer, Ed25519Verifier, Manifest + from agentrust_trace import generate_key, sign_record + except ImportError as error: + raise SystemExit( + "Signing needs the released packages in requirements.txt " + "(missing module: %s)." % error.name + ) + + keypair = _load_or_create_manifest_keypair() + manifest = build_manifest(current) + Manifest.model_validate(manifest) + manifest["signature"] = Ed25519Signer(keypair).sign(manifest) + Ed25519Verifier(keypair.public_bytes).verify( + manifest, manifest["signature"]["signature_value"] + ) + trace = sign_record(build_trace(current), generate_key()) + + outdir.mkdir(parents=True, exist_ok=True) + _save(outdir / "manifest.json", manifest, 0o644) + _save(outdir / "trace.json", trace, 0o644) + verification_key = { + "algorithm": "Ed25519", + "key_id": keypair.key_id, + "public_key_b64url": keypair.public_b64url(), + "note": "Load this key into agent-manifest trusted_keys to verify the manifest signature.", + } + _save(outdir / "verification_key.json", verification_key, 0o644) + return manifest, trace + + +def render_report( + current: Mapping[str, Any], + changes: Optional[Sequence[Mapping[str, str]]], + signed: bool, +) -> str: + model = current["model"] + lines = [ + "=" * 68, + " AGENTRUST CODEX INTEGRITY REPORT", + "=" * 68, + "", + " Workspace : %s" % current["workspace_id"][:16], + " Agent identity : %s" % current["agent_id"], + " Model : %s/%s" % (model["provider"], model["model_id"]), + " Permission mode: %s" % current["permission_mode"], + " Captured : %s" % current["captured_at"], + "", + " Composition", + " " + "-" * 64, + " Instructions : %d fingerprint(s)" % len(current["instructions"]), + " Skills : %d fingerprint(s)" % len(current["skills"]), + " Plugins : %d enabled" % len(current["plugins"]), + " Configured MCP : %d server(s)" % len(current["configured_mcp_servers"]), + "", + " Fingerprints", + " instruction layer : %s..." % current["hashes"]["system_prompt"][:23], + " policy bundle : %s..." % current["hashes"]["policy_bundle"][:23], + " skills set : %s..." % current["hashes"]["skills_set"][:23], + " plugin set : %s..." % current["hashes"]["plugin_set"][:23], + " tool catalog : %s..." % current["hashes"]["tool_catalog"][:23], + "", + ] + if changes is not None: + lines += [" Baseline comparison", " " + "-" * 64] + if not changes: + lines.append(" Verified: no composition changes.") + else: + symbols = {"added": "+", "removed": "-", "changed": "~"} + for change in changes: + lines.append( + " %s %s %s: %s" + % ( + symbols[change["change"]], + change["change"].upper(), + change["what"], + change["detail"], + ) + ) + lines.append(" %d change(s) need review." % len(changes)) + lines.append("") + if signed: + lines += [ + " Wrote manifest.json, trace.json, and verification_key.json.", + " TRACE scope: Level 0 software integrity.", + "", + ] + lines.append("=" * 68) + return "\n".join(lines) + + +def _live_from_file(path: Optional[str]) -> Dict[str, Any]: + if not path: + return {} + try: + data = json.loads(Path(path).read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as error: + raise SystemExit("Could not read live context: %s" % error) + if not isinstance(data, dict): + raise SystemExit("Live context must contain a JSON object.") + return _sanitize_live(data) + + +def _live_for_args(args: argparse.Namespace) -> Dict[str, Any]: + cwd = args.cwd or os.getcwd() + _, _, workspace_id = _workspace(cwd) + _, latest_path = _workspace_state(workspace_id) + latest, status = _load(latest_path) + live: Dict[str, Any] = {} + if status == "ok" and latest: + live.update(_sanitize_live(latest.get("runtime_context", {}))) + live.update(_live_from_file(args.live_context)) + live["cwd"] = cwd + if args.model: + live["model_id"] = args.model + live["model_version"] = args.model + live["model_provider"] = "openai" + if args.permission_mode: + live["permission_mode"] = args.permission_mode + if args.tool: + live["builtin_tools"] = sorted(set(args.tool)) + if args.mcp_server: + live["mcp_servers"] = sorted(set(args.mcp_server)) + if args.data_class: + live["data_class"] = args.data_class + return live + + +def _snapshot_for_args(args: argparse.Namespace) -> Dict[str, Any]: + return snapshot(_live_for_args(args)) + + +def cmd_snapshot(args: argparse.Namespace) -> int: + current = _snapshot_for_args(args) + _, latest_path = _workspace_state(current["workspace_id"]) + _save(latest_path, current) + if args.json: + print(json.dumps(current, indent=2, sort_keys=True)) + else: + print(render_report(current, None, False)) + return 0 + + +def _emit_context(message: str, warning: bool = False) -> None: + output: Dict[str, Any] = { + "continue": True, + "hookSpecificOutput": { + "hookEventName": "SessionStart", + "additionalContext": message, + }, + } + if warning: + output["systemMessage"] = message + print(json.dumps(output)) + + +def _hook_payload() -> Dict[str, Any]: + try: + payload = json.load(sys.stdin) if not sys.stdin.isatty() else {} + except (json.JSONDecodeError, ValueError): + payload = {} + if not isinstance(payload, dict): + return {} + live: Dict[str, Any] = { + "cwd": payload.get("cwd") or os.getcwd(), + "model_id": payload.get("model") or "unknown", + "model_provider": "openai", + "model_version": payload.get("model") or "unknown", + "permission_mode": payload.get("permission_mode") or "unknown", + } + session_id = payload.get("session_id") + if isinstance(session_id, str) and session_id: + live["session_id_hash"] = _sha_bytes(session_id.encode("utf-8")) + return live + + +def _hook_body() -> None: + current = snapshot(_hook_payload()) + baseline_path, latest_path = _workspace_state(current["workspace_id"]) + _save(latest_path, current) + baseline, status = _load(baseline_path) + + if status == "missing": + _save(baseline_path, current) + _emit_context( + "AgenTrust established a baseline for this workspace " + "(%d skills, %d plugins, %d configured MCP servers)." + % ( + len(current["skills"]), + len(current["plugins"]), + len(current["configured_mcp_servers"]), + ) + ) + return + if status == "corrupt": + _emit_context( + "AgenTrust WARNING: the workspace baseline is unreadable and was not " + "replaced. Verify the current composition, then approve a new baseline.", + warning=True, + ) + return + + changes = diff(baseline or {}, current) + if not changes: + return + detail = "; ".join( + "%s %s %s" % (item["change"], item["what"], item["detail"]) + for item in changes[:8] + ) + _emit_context( + "AgenTrust WARNING: %d Codex composition change(s): %s. " + "Use the AgenTrust agent-integrity skill to verify or approve." + % (len(changes), detail), + warning=True, + ) + + +def cmd_hook(_args: argparse.Namespace) -> int: + try: + _hook_body() + except Exception: + _emit_context( + "AgenTrust skipped the integrity check because it could not read the " + "Codex configuration. Run a manual verification.", + warning=True, + ) + return 0 + + +def cmd_verify(args: argparse.Namespace) -> int: + current = _snapshot_for_args(args) + baseline_path, latest_path = _workspace_state(current["workspace_id"]) + _save(latest_path, current) + baseline, status = _load(baseline_path) + if status == "missing": + print( + "No baseline exists for this workspace. Approve the current composition first." + ) + return 2 + if status == "corrupt": + print( + "The workspace baseline is unreadable. Review the snapshot before approving a replacement." + ) + return 2 + print(render_report(current, diff(baseline or {}, current), False)) + return 0 + + +def cmd_approve(args: argparse.Namespace) -> int: + current = _snapshot_for_args(args) + baseline_path, latest_path = _workspace_state(current["workspace_id"]) + _save(latest_path, current) + _save(baseline_path, current) + print(render_report(current, [], False)) + print("\nApproved baseline: %s" % baseline_path) + if args.sign: + sign_all(current, Path(args.out)) + print("Signed records: %s" % Path(args.out).resolve()) + return 0 + + +def cmd_report(args: argparse.Namespace) -> int: + current = _snapshot_for_args(args) + baseline_path, latest_path = _workspace_state(current["workspace_id"]) + _save(latest_path, current) + baseline, status = _load(baseline_path) + changes = diff(baseline or {}, current) if status == "ok" else None + sign_all(current, Path(args.out)) + print(render_report(current, changes, True)) + return 0 + + +def _add_common_arguments(parser: argparse.ArgumentParser) -> None: + parser.add_argument( + "--cwd", help="workspace to inspect; defaults to the current directory" + ) + parser.add_argument( + "--live-context", help="JSON file with explicit live session facts" + ) + parser.add_argument("--model", help="active Codex model slug") + parser.add_argument("--permission-mode", help="active Codex permission mode") + parser.add_argument( + "--tool", action="append", help="observed built-in tool name; repeat as needed" + ) + parser.add_argument( + "--mcp-server", + action="append", + help="observed live MCP server; repeat as needed", + ) + parser.add_argument( + "--data-class", default=None, help="TRACE data class; defaults to internal" + ) + parser.add_argument("--out", default=".", help="directory for signed records") + parser.add_argument( + "--json", action="store_true", help="print the raw snapshot JSON" + ) + parser.add_argument( + "--sign", action="store_true", help="with approve, also write signed records" + ) + + +def main(argv: Optional[Sequence[str]] = None) -> int: + parser = argparse.ArgumentParser(prog="agentrust-codex") + parser.add_argument("--version", action="version", version=VERSION) + subparsers = parser.add_subparsers(dest="command", required=True) + commands = { + "snapshot": cmd_snapshot, + "hook": cmd_hook, + "verify": cmd_verify, + "approve": cmd_approve, + "report": cmd_report, + } + for name in commands: + command_parser = subparsers.add_parser(name) + _add_common_arguments(command_parser) + args = parser.parse_args(argv) + return commands[args.command](args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/plugins/agentrust-codex/hooks/hooks.json b/plugins/agentrust-codex/hooks/hooks.json new file mode 100644 index 0000000..4ce533f --- /dev/null +++ b/plugins/agentrust-codex/hooks/hooks.json @@ -0,0 +1,18 @@ +{ + "hooks": { + "SessionStart": [ + { + "matcher": "startup|resume|clear", + "hooks": [ + { + "type": "command", + "command": "sh \"$PLUGIN_ROOT/scripts/session-start.sh\"", + "commandWindows": "\"%PLUGIN_ROOT%\\scripts\\session-start.cmd\"", + "timeout": 20, + "statusMessage": "Checking AgenTrust agent baseline" + } + ] + } + ] + } +} diff --git a/plugins/agentrust-codex/integration.yaml b/plugins/agentrust-codex/integration.yaml new file mode 100644 index 0000000..830199a --- /dev/null +++ b/plugins/agentrust-codex/integration.yaml @@ -0,0 +1,15 @@ +name: agentrust-codex +vendor: agentrust-io +integrates_with: + - agent-manifest + - trace +description: Codex plugin that fingerprints agent configuration, detects workspace-scoped drift, and emits signed Agent Manifest and TRACE Level 0 records. +maintainer: + github: carloshvp +repository: https://github.com/agentrust-io/integrations +license: Apache-2.0 +tier: community +trace_conformance_level: 0 +tested_against: + agent-manifest: 0.3.0 + agentrust-trace: 0.3.0 diff --git a/plugins/agentrust-codex/requirements.txt b/plugins/agentrust-codex/requirements.txt new file mode 100644 index 0000000..781f873 --- /dev/null +++ b/plugins/agentrust-codex/requirements.txt @@ -0,0 +1,5 @@ +# Snapshot, drift detection, and SessionStart use the Python standard library. +# Install these released packages only when generating signed records. +agent-manifest==0.3.0 +agentrust-trace==0.3.0 +agentrust-trace-tests==0.2.0 diff --git a/plugins/agentrust-codex/scripts/session-start.cmd b/plugins/agentrust-codex/scripts/session-start.cmd new file mode 100644 index 0000000..165c434 --- /dev/null +++ b/plugins/agentrust-codex/scripts/session-start.cmd @@ -0,0 +1,15 @@ +@echo off +where py >nul 2>nul +if %errorlevel% equ 0 ( + py -3 "%PLUGIN_ROOT%\engine\capture.py" hook + exit /b 0 +) + +where python >nul 2>nul +if %errorlevel% equ 0 ( + python "%PLUGIN_ROOT%\engine\capture.py" hook + exit /b 0 +) + +echo {"continue":true,"systemMessage":"AgenTrust: skipped the integrity check because Python is unavailable."} +exit /b 0 diff --git a/plugins/agentrust-codex/scripts/session-start.sh b/plugins/agentrust-codex/scripts/session-start.sh new file mode 100644 index 0000000..7ce60df --- /dev/null +++ b/plugins/agentrust-codex/scripts/session-start.sh @@ -0,0 +1,12 @@ +#!/bin/sh + +if command -v python3 >/dev/null 2>&1; then + exec python3 "$PLUGIN_ROOT/engine/capture.py" hook +fi + +if command -v python >/dev/null 2>&1; then + exec python "$PLUGIN_ROOT/engine/capture.py" hook +fi + +printf '%s\n' '{"continue":true,"systemMessage":"AgenTrust: skipped the integrity check because Python is unavailable."}' +exit 0 diff --git a/plugins/agentrust-codex/scripts/signing-python.py b/plugins/agentrust-codex/scripts/signing-python.py new file mode 100644 index 0000000..2957cbb --- /dev/null +++ b/plugins/agentrust-codex/scripts/signing-python.py @@ -0,0 +1,69 @@ +"""Create or reuse the pinned AgenTrust signing environment.""" + +from __future__ import annotations + +import hashlib +import os +import subprocess +import sys +import venv +from pathlib import Path + + +PLUGIN_ROOT = Path(__file__).resolve().parent.parent +REQUIREMENTS = PLUGIN_ROOT / "requirements.txt" +CODEX_HOME = Path( + os.environ.get("CODEX_HOME", str(Path.home() / ".codex")) +).expanduser() +ENV_ROOT = CODEX_HOME / "agentrust" / "signing-venv" +MARKER = ENV_ROOT / ".requirements.sha256" + + +def _python_path() -> Path: + if os.name == "nt": + return ENV_ROOT / "Scripts" / "python.exe" + return ENV_ROOT / "bin" / "python" + + +def _require_supported_python(version_info=None) -> None: + version_info = sys.version_info if version_info is None else version_info + if version_info < (3, 11): + raise SystemExit( + "Signed AgenTrust records require Python 3.11 or newer. " + "Run this bootstrap with Python 3.11+." + ) + + +def main() -> int: + _require_supported_python() + digest = hashlib.sha256(REQUIREMENTS.read_bytes()).hexdigest() + python = _python_path() + try: + current = MARKER.read_text(encoding="utf-8").strip() + except OSError: + current = "" + + if not python.is_file() or current != digest: + ENV_ROOT.parent.mkdir(parents=True, exist_ok=True) + if not python.is_file(): + venv.EnvBuilder(with_pip=True).create(ENV_ROOT) + subprocess.run( + [ + str(python), + "-m", + "pip", + "install", + "--disable-pip-version-check", + "--requirement", + str(REQUIREMENTS), + ], + check=True, + ) + MARKER.write_text(digest + "\n", encoding="utf-8") + + print(python) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/plugins/agentrust-codex/skills/agent-integrity/SKILL.md b/plugins/agentrust-codex/skills/agent-integrity/SKILL.md new file mode 100644 index 0000000..8fb7da7 --- /dev/null +++ b/plugins/agentrust-codex/skills/agent-integrity/SKILL.md @@ -0,0 +1,88 @@ +--- +name: agent-integrity +description: Verify or approve a workspace-scoped Codex agent baseline, inspect configuration drift, or generate signed Agent Manifest and TRACE Level 0 session records. +--- + +# AgenTrust agent integrity + +Use the capture engine shipped with this plugin. Resolve +`../../engine/capture.py` from this SKILL.md to an absolute path before +running it. Resolve `../../scripts/signing-python.py` the same way. + +The engine stores only names and hashes. Do not inspect `auth.json`, transcript +files, tool inputs, tool outputs, environment secrets, or raw configuration +values for this workflow. + +## Select the operation + +- Verify is the default. Run it when the user asks whether the Codex agent + changed, requests an integrity check, or names no operation. +- Snapshot shows the current fingerprints without comparing or changing the + baseline. +- Approve replaces the baseline. Run it only when the user asked to establish, + accept, approve, or rebaseline the current composition. A verification result + alone does not authorize approval. +- Report creates signed Agent Manifest and TRACE records. The report does not + alter the approved baseline. + +Use the current working directory as the workspace unless the user names +another one. The engine scopes baselines by a hash of the Git root, so checks in +different repositories do not overwrite each other. + +## Run the engine + +Use `python3` on macOS or Linux and `py -3` on Windows. + +Verify: + +```text +python3 verify --cwd +``` + +Snapshot: + +```text +python3 snapshot --cwd +``` + +Approve after explicit user authorization: + +```text +python3 approve --cwd +``` + +For a signed report, run the signing environment bootstrap with Python 3.11 or +newer. It creates a venv under `$CODEX_HOME/agentrust/signing-venv` and installs +the exact released versions from this plugin's `requirements.txt`. This is the +only step in the workflow that accesses PyPI. + +```text +python3 + report --cwd --out +``` + +If the SessionStart hook did not capture the active model, add +`--model ` only when the model slug is present in the +current Codex session. Never guess it. + +## Interpret results + +Name each added, removed, or changed instruction, skill, plugin, policy file, +policy setting, MCP server, model, or permission mode. Tell the user whether +the change matches their stated intent. + +Never approve unexpected drift. Leave the existing baseline intact and point +to the changed category. + +For signed output, report these checks: + +- `manifest.json` carries an Ed25519 signature and + `verification_key.json` contains the public verification key. State that + this verifies signature integrity. Artifact-binding verification needs + separately measured runtime hashes. +- `trace.json` is a TRACE Level 0 software-integrity record. Level 0 does not + claim a TEE, hardware attestation, or a complete action log. + +The engine fingerprints the on-disk instruction and policy layers it can see. +It does not prove that Codex loaded each project file, expose Codex's internal +system prompt, or enumerate every built-in tool from a SessionStart hook. diff --git a/plugins/agentrust-codex/tests/test_capture.py b/plugins/agentrust-codex/tests/test_capture.py new file mode 100644 index 0000000..c55a7d9 --- /dev/null +++ b/plugins/agentrust-codex/tests/test_capture.py @@ -0,0 +1,378 @@ +"""Tests for the AgenTrust Codex capture engine.""" + +from __future__ import annotations + +import io +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "engine")) +import capture # noqa: E402 + + +def _isolated_layout(tmp_path, monkeypatch): + home = tmp_path / "home" + codex_home = home / ".codex" + state = codex_home / "agentrust" + workspace = tmp_path / "work" + (workspace / ".git").mkdir(parents=True) + codex_home.mkdir(parents=True) + + monkeypatch.setattr(capture, "HOME", home) + monkeypatch.setattr(capture, "CODEX_HOME", codex_home) + monkeypatch.setattr(capture, "STATE_DIR", state) + monkeypatch.setattr(capture, "SIGNING_KEY", state / "signing_key.json") + monkeypatch.setenv("USER", "tester") + monkeypatch.setenv("HOSTNAME", "test-host") + return home, codex_home, state, workspace + + +def _write_plugin(codex_home: Path, marketplace="local", name="demo", version="1.0.0"): + root = codex_home / "plugins" / "cache" / marketplace / name / version + (root / ".codex-plugin").mkdir(parents=True) + (root / ".codex-plugin" / "plugin.json").write_text( + json.dumps({"name": name, "version": version}), encoding="utf-8" + ) + (root / "skills" / "review").mkdir(parents=True) + (root / "skills" / "review" / "SKILL.md").write_text( + "---\nname: review\ndescription: Review code.\n---\n", encoding="utf-8" + ) + return root + + +def _write_config(codex_home: Path): + (codex_home / "config.toml").write_text( + """ +approval_policy = "never" +sandbox_mode = "workspace-write" + +[mcp_servers.github] +url = "https://example.invalid/mcp" +secret = "must-not-appear" + +[plugins."demo@local"] +enabled = true +""".strip() + + "\n", + encoding="utf-8", + ) + + +def _live(workspace: Path, **overrides): + value = { + "cwd": str(workspace), + "model_id": "gpt-test", + "model_provider": "openai", + "model_version": "gpt-test", + "permission_mode": "dontAsk", + } + value.update(overrides) + return value + + +def test_snapshot_captures_names_and_hashes_without_config_values( + tmp_path, monkeypatch +): + home, codex_home, _state, workspace = _isolated_layout(tmp_path, monkeypatch) + _write_config(codex_home) + _write_plugin(codex_home) + (workspace / "AGENTS.md").write_text("Follow repository rules.\n", encoding="utf-8") + skill = home / ".agents" / "skills" / "local-check" + skill.mkdir(parents=True) + (skill / "SKILL.md").write_text( + "---\nname: local-check\ndescription: Check locally.\n---\n", encoding="utf-8" + ) + + current = capture.snapshot(_live(workspace)) + encoded = json.dumps(current, sort_keys=True) + + assert current["model"]["model_id"] == "gpt-test" + assert current["permission_mode"] == "dontAsk" + assert current["configured_mcp_servers"] == ["github"] + assert list(current["plugins"]) == ["demo@local"] + assert any(name.endswith("local-check") for name in current["skills"]) + assert any(name.endswith("AGENTS.md") for name in current["instructions"]) + assert "must-not-appear" not in encoded + assert "https://example.invalid/mcp" not in encoded + assert all(value.startswith("sha256:") for value in current["hashes"].values()) + + +def test_disabled_plugin_is_not_fingerprinted(tmp_path, monkeypatch): + _home, codex_home, _state, workspace = _isolated_layout(tmp_path, monkeypatch) + _write_plugin(codex_home) + (codex_home / "config.toml").write_text( + '[plugins."demo@local"]\nenabled = false\n', encoding="utf-8" + ) + + current = capture.snapshot(_live(workspace)) + assert current["plugins"] == {} + assert "plugin:demo@local" not in current["tools"] + + +def test_fallback_parser_honors_disabled_plugins(tmp_path): + config = tmp_path / "config.toml" + config.write_text( + """ +approval_policy = "never" +[mcp_servers.github] +url = "stdio" +[plugins."off@local"] +enabled = false +[plugins."on@local"] +enabled = true +""".strip() + + "\n", + encoding="utf-8", + ) + mcp, plugins, scope = capture._fallback_toml_state(config) + assert mcp == {"github"} + assert plugins == {"on@local"} + assert scope == {"approval_policy": "never"} + + +def test_diff_reports_specific_composition_changes(): + base = { + "observed": [ + "instructions", + "mcp_config", + "model", + "permission", + "plugins", + "policy", + "skills", + ], + "instructions": {"workspace:AGENTS.md": "sha256:" + "1" * 64}, + "policy_files": {"user:config.toml": "sha256:" + "2" * 64}, + "policy_scope": {"approval_policy": "never"}, + "skills": {"user:review": "sha256:" + "3" * 64}, + "plugins": {"demo@local": "sha256:" + "4" * 64}, + "configured_mcp_servers": ["github"], + "model": {"model_id": "gpt-a"}, + "permission_mode": "default", + } + current = json.loads(json.dumps(base)) + current["skills"]["user:exfil"] = "sha256:" + "5" * 64 + current["plugins"]["demo@local"] = "sha256:" + "6" * 64 + current["configured_mcp_servers"].append("shadow") + current["model"] = {"model_id": "gpt-b"} + current["permission_mode"] = "bypassPermissions" + + changes = capture.diff(base, current) + assert {"change": "added", "what": "skill", "detail": "user:exfil"} in changes + assert {"change": "changed", "what": "plugin", "detail": "demo@local"} in changes + assert { + "change": "added", + "what": "configured MCP server", + "detail": "shadow", + } in changes + assert any(item["what"] == "model" for item in changes) + assert any(item["what"] == "permission mode" for item in changes) + + +def test_workspace_baselines_are_isolated(tmp_path, monkeypatch): + _home, _codex_home, state, first = _isolated_layout(tmp_path, monkeypatch) + second = tmp_path / "second" + (second / ".git").mkdir(parents=True) + + first_snapshot = capture.snapshot(_live(first)) + second_snapshot = capture.snapshot(_live(second)) + first_path, _ = capture._workspace_state(first_snapshot["workspace_id"]) + second_path, _ = capture._workspace_state(second_snapshot["workspace_id"]) + + assert first_snapshot["workspace_id"] != second_snapshot["workspace_id"] + assert first_path != second_path + assert state in first_path.parents + assert state in second_path.parents + + +def _run_hook(payload, monkeypatch): + monkeypatch.setattr(capture.sys, "stdin", io.StringIO(json.dumps(payload))) + args = type("Args", (), {})() + return capture.cmd_hook(args) + + +def test_hook_establishes_baseline_stays_quiet_then_warns( + tmp_path, monkeypatch, capsys +): + _home, codex_home, _state, workspace = _isolated_layout(tmp_path, monkeypatch) + _write_config(codex_home) + _write_plugin(codex_home) + agents = workspace / "AGENTS.md" + agents.write_text("Initial instructions.\n", encoding="utf-8") + payload = { + "cwd": str(workspace), + "model": "gpt-test", + "permission_mode": "default", + "session_id": "session-secret", + } + + assert _run_hook(payload, monkeypatch) == 0 + first = json.loads(capsys.readouterr().out) + assert "established a baseline" in first["hookSpecificOutput"]["additionalContext"] + baseline_id = capture.snapshot(_live(workspace))["workspace_id"] + baseline_path, _ = capture._workspace_state(baseline_id) + stored = json.loads(baseline_path.read_text(encoding="utf-8")) + assert stored["runtime_context"]["session_id_hash"].startswith("sha256:") + assert "session-secret" not in baseline_path.read_text(encoding="utf-8") + + assert _run_hook(payload, monkeypatch) == 0 + assert capsys.readouterr().out == "" + + agents.write_text("Changed instructions.\n", encoding="utf-8") + assert _run_hook(payload, monkeypatch) == 0 + warning = json.loads(capsys.readouterr().out) + assert "AgenTrust WARNING" in warning["systemMessage"] + assert "changed instruction" in warning["systemMessage"] + + +def test_corrupt_baseline_is_never_replaced(tmp_path, monkeypatch, capsys): + _home, _codex_home, _state, workspace = _isolated_layout(tmp_path, monkeypatch) + current = capture.snapshot(_live(workspace)) + baseline_path, _ = capture._workspace_state(current["workspace_id"]) + baseline_path.parent.mkdir(parents=True) + baseline_path.write_text("{broken", encoding="utf-8") + + payload = { + "cwd": str(workspace), + "model": "gpt-test", + "permission_mode": "default", + } + assert _run_hook(payload, monkeypatch) == 0 + warning = json.loads(capsys.readouterr().out) + assert "was not replaced" in warning["systemMessage"] + assert baseline_path.read_text(encoding="utf-8") == "{broken" + + +def test_hook_never_crashes(tmp_path, monkeypatch, capsys): + _isolated_layout(tmp_path, monkeypatch) + + def fail(_live=None): + raise RuntimeError("simulated failure") + + monkeypatch.setattr(capture, "snapshot", fail) + monkeypatch.setattr(capture.sys, "stdin", io.StringIO("not-json")) + args = type("Args", (), {})() + assert capture.cmd_hook(args) == 0 + output = json.loads(capsys.readouterr().out) + assert output["continue"] is True + assert "skipped the integrity check" in output["systemMessage"] + + +def test_verify_resnapshots_instead_of_trusting_latest(tmp_path, monkeypatch, capsys): + _home, _codex_home, _state, workspace = _isolated_layout(tmp_path, monkeypatch) + agents = workspace / "AGENTS.md" + agents.write_text("Approved.\n", encoding="utf-8") + approved = capture.snapshot(_live(workspace)) + baseline_path, latest_path = capture._workspace_state(approved["workspace_id"]) + capture._save(baseline_path, approved) + capture._save(latest_path, approved) + agents.write_text("Drifted.\n", encoding="utf-8") + + args = type( + "Args", + (), + { + "cwd": str(workspace), + "live_context": None, + "model": None, + "permission_mode": None, + "tool": None, + "mcp_server": None, + "data_class": None, + }, + )() + assert capture.cmd_verify(args) == 0 + output = capsys.readouterr().out + assert "CHANGED instruction" in output + assert "Verified: no composition changes." not in output + + +def _signable_snapshot(tmp_path, monkeypatch): + _home, codex_home, _state, workspace = _isolated_layout(tmp_path, monkeypatch) + _write_config(codex_home) + _write_plugin(codex_home) + return capture.snapshot( + _live( + workspace, + builtin_tools=["exec_command"], + permission_mode="bypassPermissions", + ) + ) + + +def test_signing_key_is_stable_and_owner_only(tmp_path, monkeypatch): + pytest.importorskip("agent_manifest") + pytest.importorskip("cryptography") + _isolated_layout(tmp_path, monkeypatch) + + first = capture._load_or_create_manifest_keypair() + second = capture._load_or_create_manifest_keypair() + assert first.key_id == second.key_id + if os.name == "posix": + assert stat_mode(capture.SIGNING_KEY) == 0o600 + + +def stat_mode(path: Path): + return path.stat().st_mode & 0o777 + + +def test_corrupt_signing_key_does_not_mint_a_new_identity(tmp_path, monkeypatch): + pytest.importorskip("agent_manifest") + pytest.importorskip("cryptography") + _isolated_layout(tmp_path, monkeypatch) + capture.SIGNING_KEY.parent.mkdir(parents=True) + capture.SIGNING_KEY.write_text("{broken", encoding="utf-8") + + with pytest.raises(SystemExit, match="signing key is unreadable"): + capture._load_or_create_manifest_keypair() + assert capture.SIGNING_KEY.read_text(encoding="utf-8") == "{broken" + + +def test_signed_outputs_verify_and_pass_trace_level_zero(tmp_path, monkeypatch): + pytest.importorskip("agent_manifest") + pytest.importorskip("agentrust_trace") + pytest.importorskip("trace_tests") + from agent_manifest import RevocationStore, VerificationContext, verify_manifest + + current = _signable_snapshot(tmp_path, monkeypatch) + out = tmp_path / "records" + manifest, _trace = capture.sign_all(current, out) + assert manifest["artifacts"]["policy_bundle"]["enforcement_mode"] == "audit-only" + assert _trace["policy"]["enforcement_mode"] == "silent" + verification_key = json.loads( + (out / "verification_key.json").read_text(encoding="utf-8") + ) + context = VerificationContext( + trusted_keys={verification_key["key_id"]: verification_key["public_key_b64url"]} + ) + good = verify_manifest(manifest, context, RevocationStore()) + assert good.signature_verified is True + + tampered = json.loads(json.dumps(manifest)) + tampered["artifacts"]["policy_bundle"]["hash"] = "sha256:" + "0" * 64 + bad = verify_manifest(tampered, context, RevocationStore()) + assert bad.signature_verified is False + + result = subprocess.run( + [ + sys.executable, + "-m", + "trace_tests.cli", + "verify", + "--record", + str(out / "trace.json"), + "--level", + "0", + ], + text=True, + capture_output=True, + check=False, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert "PASS" in result.stdout diff --git a/plugins/agentrust-codex/tests/test_signing_python.py b/plugins/agentrust-codex/tests/test_signing_python.py new file mode 100644 index 0000000..749d728 --- /dev/null +++ b/plugins/agentrust-codex/tests/test_signing_python.py @@ -0,0 +1,22 @@ +"""Tests for the pinned signing-environment bootstrap.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import pytest + + +SCRIPT = Path(__file__).resolve().parent.parent / "scripts" / "signing-python.py" +SPEC = importlib.util.spec_from_file_location("agentrust_signing_python", SCRIPT) +assert SPEC and SPEC.loader +signing_python = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(signing_python) + + +def test_signing_bootstrap_requires_python_311(): + with pytest.raises(SystemExit, match="Python 3.11 or newer"): + signing_python._require_supported_python((3, 10)) + + signing_python._require_supported_python((3, 11)) diff --git a/plugins/agentrust-codex/tests/validate_structure.py b/plugins/agentrust-codex/tests/validate_structure.py new file mode 100644 index 0000000..a0d6bc0 --- /dev/null +++ b/plugins/agentrust-codex/tests/validate_structure.py @@ -0,0 +1,48 @@ +"""Validate the integration manifest, plugin package, and marketplace entry.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import jsonschema +import yaml + + +REPO = Path(__file__).resolve().parents[3] +PLUGIN = REPO / "plugins" / "agentrust-codex" + + +def main() -> int: + schema = json.loads((REPO / "schema" / "integration.schema.json").read_text()) + integration = yaml.safe_load((PLUGIN / "integration.yaml").read_text()) + jsonschema.validate(integration, schema) + + manifest = json.loads( + (PLUGIN / ".codex-plugin" / "plugin.json").read_text(encoding="utf-8") + ) + assert manifest["name"] == PLUGIN.name + assert manifest["skills"] == "./skills/" + assert (PLUGIN / "hooks" / "hooks.json").is_file() + assert (PLUGIN / "skills" / "agent-integrity" / "SKILL.md").is_file() + + marketplace = json.loads( + (REPO / ".agents" / "plugins" / "marketplace.json").read_text(encoding="utf-8") + ) + entry = next( + item for item in marketplace["plugins"] if item["name"] == manifest["name"] + ) + assert marketplace["name"] == "agentrust" + assert entry["source"] == { + "source": "local", + "path": "./plugins/agentrust-codex", + } + assert entry["policy"]["installation"] == "AVAILABLE" + assert entry["policy"]["authentication"] == "ON_INSTALL" + assert entry["category"] == "Security" + print("AgenTrust Codex structure validation passed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) From 9685507db15764f05611e7bb7dc09d0b1055e2ba Mon Sep 17 00:00:00 2001 From: Carlos Hernandez Date: Fri, 17 Jul 2026 13:32:56 +0200 Subject: [PATCH 2/2] fix(codex): keep signed records owner-only --- plugins/agentrust-codex/PRIVACY.md | 4 ++-- plugins/agentrust-codex/engine/capture.py | 16 +++++++--------- plugins/agentrust-codex/tests/test_capture.py | 3 +++ 3 files changed, 12 insertions(+), 11 deletions(-) diff --git a/plugins/agentrust-codex/PRIVACY.md b/plugins/agentrust-codex/PRIVACY.md index 5b38f02..094c4ea 100644 --- a/plugins/agentrust-codex/PRIVACY.md +++ b/plugins/agentrust-codex/PRIVACY.md @@ -29,8 +29,8 @@ tool outputs, or environment secrets. The plugin writes baselines, latest snapshots, and its signing key below `$CODEX_HOME/agentrust`, which defaults to `~/.codex/agentrust`. It requests -owner-only permissions for private state. Signed reports go to the output -directory you choose. +owner-only permissions for private state and signed report files. Signed +reports go to the output directory you choose. The plugin makes no network request during SessionStart, snapshot, verify, or approve. A signed-report request can run the included bootstrap, which installs diff --git a/plugins/agentrust-codex/engine/capture.py b/plugins/agentrust-codex/engine/capture.py index 3d68498..2c7d5f0 100644 --- a/plugins/agentrust-codex/engine/capture.py +++ b/plugins/agentrust-codex/engine/capture.py @@ -16,7 +16,6 @@ import platform import re import socket -import stat import sys import tempfile import time @@ -739,7 +738,7 @@ def build_trace(current: Mapping[str, Any]) -> Dict[str, Any]: } -def _atomic_write(path: Path, content: str, mode: int = 0o600) -> None: +def _atomic_write(path: Path, content: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary: Optional[str] = None try: @@ -755,7 +754,7 @@ def _atomic_write(path: Path, content: str, mode: int = 0o600) -> None: os.fsync(handle.fileno()) temporary = handle.name try: - os.chmod(temporary, mode) + os.chmod(temporary, 0o600) except OSError: pass os.replace(temporary, path) @@ -767,8 +766,8 @@ def _atomic_write(path: Path, content: str, mode: int = 0o600) -> None: pass -def _save(path: Path, value: Mapping[str, Any], mode: int = 0o600) -> None: - _atomic_write(path, json.dumps(value, indent=2, sort_keys=True) + "\n", mode) +def _save(path: Path, value: Mapping[str, Any]) -> None: + _atomic_write(path, json.dumps(value, indent=2, sort_keys=True) + "\n") def _load(path: Path) -> Tuple[Optional[Dict[str, Any]], str]: @@ -813,7 +812,6 @@ def _load_or_create_manifest_keypair(): "private_b64url": keypair.private_b64url(), "created_at": _now_iso(), }, - stat.S_IRUSR | stat.S_IWUSR, ) return keypair @@ -845,15 +843,15 @@ def sign_all( trace = sign_record(build_trace(current), generate_key()) outdir.mkdir(parents=True, exist_ok=True) - _save(outdir / "manifest.json", manifest, 0o644) - _save(outdir / "trace.json", trace, 0o644) + _save(outdir / "manifest.json", manifest) + _save(outdir / "trace.json", trace) verification_key = { "algorithm": "Ed25519", "key_id": keypair.key_id, "public_key_b64url": keypair.public_b64url(), "note": "Load this key into agent-manifest trusted_keys to verify the manifest signature.", } - _save(outdir / "verification_key.json", verification_key, 0o644) + _save(outdir / "verification_key.json", verification_key) return manifest, trace diff --git a/plugins/agentrust-codex/tests/test_capture.py b/plugins/agentrust-codex/tests/test_capture.py index c55a7d9..d4915c2 100644 --- a/plugins/agentrust-codex/tests/test_capture.py +++ b/plugins/agentrust-codex/tests/test_capture.py @@ -353,6 +353,9 @@ def test_signed_outputs_verify_and_pass_trace_level_zero(tmp_path, monkeypatch): ) good = verify_manifest(manifest, context, RevocationStore()) assert good.signature_verified is True + if os.name == "posix": + for name in ("manifest.json", "trace.json", "verification_key.json"): + assert stat_mode(out / name) == 0o600 tampered = json.loads(json.dumps(manifest)) tampered["artifacts"]["policy_bundle"]["hash"] = "sha256:" + "0" * 64