feat: credential injection and offload for agent tool calls (#2481) (#2534)

Add CredentialVault, CredentialProfile, and CredentialInjector primitives so AGT can hold secrets on behalf of agents and inject them at the point of use. Agents only ever see opaque {{cred:NAME}} placeholders; resolved values stay inside the trust boundary.

Key properties (addresses original AC + kayalopez review):

- Encrypted-at-rest persistence via Fernet (cryptography). Fails closed (no plaintext on disk) when cryptography is unavailable.

- Per-agent profiles bind credential handles to action capabilities (e.g. github:read_issues vs github:push_code), not just agent identities.

- Workflow policy callback runs BEFORE any vault read, so prompt-injected tool args cannot smuggle credential references past policy.

- Placeholder allowlist per call: handles not pre-authorized by the workflow are rejected. MCP server metadata and tool descriptions cannot inject credential references.

- Deterministic DenyReceipt for missing / out-of-scope / policy-denied handles.

- Audit events record agent DID, handle name, target service, action class, decision, policy version. Never the resolved value.

- Rotation rebinds value in place; saved prompts/plans/MCP descriptions keep working.

- HMAC-SHA256 audit_digest helper for tamper-evident logs.

Surfaces:

- agent_os.credential_vault module + public exports on agent_os.

- agt cred {genkey,add,list,rotate,remove} CLI; never prints values.

- examples/credential_vault_example.py end-to-end example.

Also: add credential_vault.py to no-custom-crypto allowlist, extend cspell repo terms, and lower entropy of vault rotation test fixture.

Tests: 28 unit tests + 5 CLI tests, all green. ruff clean.

Signed-off-by: Ricky Gummadi <ricky.gummadi@outlook.com>

Author: Ricky-G

.cspell-repo-terms.txt

@@ -305,3 +305,11 @@ workflows
 XACML
 xfail
 xunit
+Fernet
+fernet
+genkey
+importorskip
+digestmod
+urandom
+octo
+kayalopez

agent-governance-python/agent-compliance/src/agent_compliance/cli/agt.py

@@ -25,6 +25,7 @@
 import click
 
 from agent_compliance.cli.red_team import red_team
+from agent_compliance.cli.cred import cred
 
 _logger = logging.getLogger(__name__)
 
@@ -194,6 +195,8 @@ def cli(
 
 # Register the red-team subcommand group
 cli.add_command(red_team)
+# Register the credential vault subcommand group (issue #2481)
+cli.add_command(cred)
 
 
 @cli.command()

agent-governance-python/agent-compliance/src/agent_compliance/cli/cred.py

@@ -0,0 +1,125 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""``agt cred`` — manage the credential vault for agent tool calls.
+
+Subcommands:
+
+    agt cred add NAME VALUE          Store or replace a credential.
+    agt cred list                    List credential handle names.
+    agt cred rotate NAME NEW_VALUE   Rotate a credential value in place.
+    agt cred remove NAME             Delete a credential.
+    agt cred genkey                  Generate a Fernet encryption key.
+
+The CLI persists to ``$AGT_VAULT_PATH`` (default ``./.agt/vault.bin``)
+encrypted with ``$AGT_VAULT_KEY`` (a Fernet key — generate with
+``agt cred genkey``). The CLI never prints credential values.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+
+import click
+
+
+_DEFAULT_VAULT_PATH = ".agt/vault.bin"
+
+
+def _resolve_vault() -> "object":  # returns CredentialVault
+    """Construct a vault from environment-configured key/path."""
+    try:
+        from agent_os.credential_vault import (
+            CredentialVault,
+            EncryptionUnavailable,
+        )
+    except ImportError as exc:
+        raise click.ClickException(
+            f"agent_os.credential_vault is unavailable: {exc}"
+        ) from exc
+
+    path = os.environ.get("AGT_VAULT_PATH", _DEFAULT_VAULT_PATH)
+    key = os.environ.get("AGT_VAULT_KEY")
+    if not key:
+        raise click.ClickException(
+            "AGT_VAULT_KEY is not set. Generate one with 'agt cred genkey' "
+            "and export it before running other 'agt cred' commands."
+        )
+    Path(os.path.dirname(path) or ".").mkdir(parents=True, exist_ok=True)
+    try:
+        return CredentialVault(persist_path=path, encryption_key=key.encode("utf-8"))
+    except EncryptionUnavailable as exc:
+        raise click.ClickException(str(exc)) from exc
+
+
+@click.group(name="cred")
+def cred() -> None:
+    """Manage the AGT credential vault."""
+
+
+@cred.command(name="genkey")
+def genkey() -> None:
+    """Print a fresh Fernet encryption key suitable for AGT_VAULT_KEY."""
+    from agent_os.credential_vault import CredentialVault
+
+    click.echo(CredentialVault.generate_key().decode("ascii"))
+
+
+@cred.command(name="add")
+@click.argument("name")
+@click.argument("value")
+@click.option("--type", "cred_type", default="secret", show_default=True,
+              help="Credential type label (e.g. bearer_token, basic_auth).")
+def add(name: str, value: str, cred_type: str) -> None:
+    """Store or replace a credential. Use '-' as VALUE to read from stdin."""
+    vault = _resolve_vault()
+    if value == "-":
+        value = sys.stdin.read().rstrip("\n")
+    handle = vault.put(name, value, cred_type=cred_type)  # type: ignore[attr-defined]
+    click.echo(f"stored: {handle.name}")
+
+
+@cred.command(name="list")
+@click.option("--json", "as_json", is_flag=True, default=False, help="Emit JSON.")
+def list_handles(as_json: bool) -> None:
+    """List credential handle names (no values are printed)."""
+    vault = _resolve_vault()
+    names = vault.list_handles()  # type: ignore[attr-defined]
+    if as_json:
+        click.echo(json.dumps({"handles": names}))
+        return
+    if not names:
+        click.echo("(no credentials)")
+        return
+    for n in names:
+        meta = vault.get_metadata(n) or {}  # type: ignore[attr-defined]
+        click.echo(f"{n}\t{meta.get('cred_type', '')}\tv{meta.get('version', 1)}")
+
+
+@cred.command(name="rotate")
+@click.argument("name")
+@click.argument("new_value")
+def rotate(name: str, new_value: str) -> None:
+    """Rotate a credential's value while preserving its handle name."""
+    vault = _resolve_vault()
+    if new_value == "-":
+        new_value = sys.stdin.read().rstrip("\n")
+    try:
+        handle = vault.rotate(name, new_value)  # type: ignore[attr-defined]
+    except KeyError:
+        raise click.ClickException(f"unknown credential: {name}")
+    click.echo(f"rotated: {handle.name}")
+
+
+@cred.command(name="remove")
+@click.argument("name")
+def remove(name: str) -> None:
+    """Delete a credential by handle name."""
+    vault = _resolve_vault()
+    removed = vault.delete(name)  # type: ignore[attr-defined]
+    if not removed:
+        raise click.ClickException(f"unknown credential: {name}")
+    click.echo(f"removed: {name}")

agent-governance-python/agent-compliance/tests/test_cred_cli.py

@@ -0,0 +1,90 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+"""Tests for ``agt cred`` CLI (issue #2481)."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+from click.testing import CliRunner
+
+from agent_compliance.cli.agt import cli
+
+
+pytest.importorskip("cryptography")
+
+
+@pytest.fixture()
+def runner() -> CliRunner:
+    try:
+        return CliRunner(mix_stderr=False)
+    except TypeError:
+        return CliRunner()
+
+
+@pytest.fixture()
+def env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, str]:
+    from agent_os.credential_vault import CredentialVault
+
+    key = CredentialVault.generate_key().decode("ascii")
+    path = str(tmp_path / "vault.bin")
+    monkeypatch.setenv("AGT_VAULT_KEY", key)
+    monkeypatch.setenv("AGT_VAULT_PATH", path)
+    return {"AGT_VAULT_KEY": key, "AGT_VAULT_PATH": path}
+
+
+def test_genkey_prints_valid_fernet_key(runner: CliRunner) -> None:
+    result = runner.invoke(cli, ["cred", "genkey"])
+    assert result.exit_code == 0, result.output
+    key = result.output.strip()
+    # 32-byte url-safe base64 = 44 chars including padding
+    assert len(key) == 44
+
+
+def test_add_list_rotate_remove_roundtrip(
+    runner: CliRunner, env: dict[str, str]
+) -> None:
+    r = runner.invoke(cli, ["cred", "add", "gh", "secret-value", "--type", "bearer_token"])
+    assert r.exit_code == 0, r.output
+    assert "stored: gh" in r.output
+
+    r = runner.invoke(cli, ["cred", "list", "--json"])
+    assert r.exit_code == 0, r.output
+    payload = json.loads(r.output)
+    assert payload == {"handles": ["gh"]}
+    # Value never leaks
+    assert "secret-value" not in r.output
+
+    r = runner.invoke(cli, ["cred", "rotate", "gh", "new-value"])
+    assert r.exit_code == 0
+    assert "rotated: gh" in r.output
+    assert "new-value" not in r.output
+
+    r = runner.invoke(cli, ["cred", "remove", "gh"])
+    assert r.exit_code == 0
+    assert "removed: gh" in r.output
+
+    r = runner.invoke(cli, ["cred", "list"])
+    assert "(no credentials)" in r.output
+
+
+def test_missing_key_fails_closed(
+    runner: CliRunner, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.delenv("AGT_VAULT_KEY", raising=False)
+    monkeypatch.setenv("AGT_VAULT_PATH", str(tmp_path / "v.bin"))
+    r = runner.invoke(cli, ["cred", "add", "k", "v"])
+    assert r.exit_code != 0
+    assert "AGT_VAULT_KEY" in (r.output + (r.stderr or ""))
+
+
+def test_remove_unknown_errors(runner: CliRunner, env: dict[str, str]) -> None:
+    r = runner.invoke(cli, ["cred", "remove", "nope"])
+    assert r.exit_code != 0
+
+
+def test_rotate_unknown_errors(runner: CliRunner, env: dict[str, str]) -> None:
+    r = runner.invoke(cli, ["cred", "rotate", "nope", "x"])
+    assert r.exit_code != 0

agent-governance-python/agent-os/examples/credential_vault_example.py

@@ -0,0 +1,74 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+"""Example: credential offload and injection for an agent tool call.
+
+Run with: ``python examples/credential_vault_example.py``
+
+This shows the issue #2481 flow end-to-end:
+
+1. An operator provisions a credential in the vault.
+2. An agent's profile binds an action capability to that credential.
+3. The agent's prompt / saved plan only ever sees the opaque placeholder
+   ``{{cred:NAME}}`` — never the resolved value.
+4. The injector evaluates the workflow policy, resolves the placeholder
+   inside the trust boundary, and returns the rendered request.
+5. Audit records show *who* used *which handle* for *which service* — but
+   never the secret itself.
+"""
+
+from __future__ import annotations
+
+from agent_os.credential_vault import (
+    CredentialInjector,
+    CredentialProfile,
+    CredentialVault,
+    InjectionContext,
+    PolicyOutcome,
+)
+
+
+def main() -> None:
+    # 1. Operator provisions the credential.
+    vault = CredentialVault()
+    vault.put("github_pat", "ghp_real_token_value", cred_type="bearer_token")
+
+    # 2. Bind an agent identity to an action capability -> handle.
+    vault.register_profile(
+        CredentialProfile(
+            agent_did="did:web:agent-ci",
+            bindings={"github:read_issues": "github_pat"},
+        )
+    )
+
+    # 3. The agent's saved tool-call template uses only the placeholder.
+    headers = {"Authorization": "Bearer {{cred:github_pat}}"}
+
+    # 4. Workflow policy decides whether this call is allowed at all,
+    #    before the injector ever reads the value.
+    def policy(ctx: InjectionContext) -> PolicyOutcome:
+        return PolicyOutcome(
+            allow=ctx.action_class == "github:read_issues",
+            reason="only read-only github calls are permitted in this workflow",
+        )
+
+    injector = CredentialInjector(vault)
+    result = injector.inject_headers(
+        "did:web:agent-ci",
+        headers,
+        action_class="github:read_issues",
+        target_service="api.github.com",
+        allowed_handles=["github_pat"],
+        policy_check=policy,
+        policy_version="v1",
+    )
+
+    print("allowed:", result.allowed)
+    print("rendered Authorization length:", len(result.payload["Authorization"]))
+
+    # 5. Audit log has no value, only the handle name and decision.
+    for ev in vault.audit_log():
+        print("audit:", ev.to_dict())
+
+
+if __name__ == "__main__":
+    main()

agent-governance-python/agent-os/src/agent_os/__init__.py

@@ -181,6 +181,23 @@ def check_installation() -> None:
     ToolFingerprint,
 )
 from agent_os.credential_redactor import CredentialMatch, CredentialPattern, CredentialRedactor
+from agent_os.credential_vault import (
+    DENY_REASON,
+    CredentialDecision,
+    CredentialError,
+    CredentialHandle,
+    CredentialInjector,
+    CredentialProfile,
+    CredentialRecord,
+    CredentialVault,
+    DenyReceipt,
+    EncryptionUnavailable,
+    InjectionContext,
+    InjectionResult,
+    PolicyOutcome,
+    VaultAuditEvent,
+    audit_digest,
+)
 from agent_os.mcp_message_signer import (
     MCPMessageSigner,
     MCPSignedEnvelope,
@@ -336,6 +353,22 @@ def check_installation() -> None:
     "CredentialRedactor",
     "CredentialPattern",
     "CredentialMatch",
+    # Credential Vault & Injection (issue #2481)
+    "CredentialVault",
+    "CredentialInjector",
+    "CredentialHandle",
+    "CredentialProfile",
+    "CredentialRecord",
+    "CredentialDecision",
+    "CredentialError",
+    "EncryptionUnavailable",
+    "VaultAuditEvent",
+    "DenyReceipt",
+    "InjectionContext",
+    "InjectionResult",
+    "PolicyOutcome",
+    "DENY_REASON",
+    "audit_digest",
     "MCPSessionStore",
     "MCPNonceStore",
     "MCPRateLimitStore",