tegwick
ae7bc73b4f
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa
202 lines
7.9 KiB
Python
202 lines
7.9 KiB
Python
"""Tests for the `warden access` operator front door (WP-0014 T2)."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from warden.access import agent_read_boundary_identity, expand_handoff, policy_gate_status
|
|
from warden.cli import app
|
|
from warden.routing.models import RouteEntry
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def synthetic_openbao_advisory_identity(monkeypatch):
|
|
"""Advisory CLI tests must not inspect or use workstation credentials.
|
|
|
|
Identity-display tests override this stub explicitly. The real helper has
|
|
separate subprocess-mocked unit coverage in test_openbao_caller.py.
|
|
"""
|
|
monkeypatch.setattr(
|
|
"warden.openbao_caller.describe_ambient_openbao_caller",
|
|
lambda *args, **kwargs: {
|
|
"source": "none", "entity_id": None, "display_name": None,
|
|
"auth_path": None, "policies": [], "oidc_role": None, "ttl": None,
|
|
"error": "synthetic test: no ambient OpenBao token",
|
|
},
|
|
)
|
|
|
|
|
|
def test_advisory_does_not_invoke_ambient_lookup(monkeypatch):
|
|
monkeypatch.setenv("BAO_TOKEN", "fixture-only-never-a-real-token")
|
|
def forbidden(*args, **kwargs):
|
|
raise AssertionError("advisory unit tests must not launch a token lookup")
|
|
monkeypatch.setattr("warden.openbao_caller.subprocess.run", forbidden)
|
|
result = runner.invoke(app, ["access", "npm token", "--json"])
|
|
assert result.exit_code == 0
|
|
assert json.loads(result.stdout)["openbao_caller"]["source"] == "none"
|
|
|
|
|
|
|
|
def _repo_catalog() -> Path:
|
|
return Path(__file__).resolve().parents[1] / "registry" / "routing" / "catalog.yaml"
|
|
|
|
|
|
def _openbao_entry() -> RouteEntry:
|
|
return RouteEntry(
|
|
id="openbao-api-key",
|
|
title="API key, DB credential, or dynamic lease",
|
|
need_keywords=["api", "key", "npm", "token"],
|
|
owner_repo="railiance-platform",
|
|
subsystem="OpenBao",
|
|
warden_executes=False,
|
|
wiki_ref="wiki/CredentialRouting.md#routing-table",
|
|
canon_ref="net-kingdom/docs/x.md",
|
|
reviewed="2026-06-27",
|
|
status="active",
|
|
auth_method="key-cape OIDC → bao login -method=oidc role=<domain>",
|
|
path_template="platform/workloads/<domain>/<workload>/<bundle>",
|
|
fetch_command="bao kv get -field=<FIELD> <path_template>",
|
|
policy_ref="flex-auth check secret.read:<domain>",
|
|
exec_capable=True,
|
|
)
|
|
|
|
|
|
# --- pure expansion --------------------------------------------------------
|
|
|
|
def test_expand_inlines_path_template_token():
|
|
e = expand_handoff(_openbao_entry())
|
|
assert "<path_template>" not in e.fetch_command
|
|
assert e.fetch_command.startswith("bao kv get -field=<FIELD> platform/workloads/")
|
|
|
|
|
|
def test_expand_substitutes_domain():
|
|
e = expand_handoff(_openbao_entry(), domain="coulomb_social")
|
|
assert "coulomb_social" in e.path_template
|
|
assert "<domain>" not in e.path_template
|
|
assert "<domain>" not in e.auth_method
|
|
# owner-side names stay as placeholders — warden does not invent them
|
|
assert "<workload>" in e.path_template and "<bundle>" in e.path_template
|
|
|
|
|
|
def test_expand_without_domain_keeps_placeholder():
|
|
e = expand_handoff(_openbao_entry())
|
|
assert "<domain>" in e.path_template
|
|
|
|
|
|
def test_policy_gate_status_no_config(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("WARDEN_CONFIG", str(tmp_path / "nope.yaml"))
|
|
assert "advisory" in policy_gate_status()
|
|
|
|
|
|
def test_issued_coding_agent_subject_drives_read_boundary_without_legacy_marker():
|
|
env = {"WARDEN_POLICY_SUBJECT": "service:codex:railiance-platform"}
|
|
assert agent_read_boundary_identity(env) == "service:codex:railiance-platform"
|
|
|
|
|
|
def test_legacy_agent_marker_remains_fail_toward_safety_fallback():
|
|
env = {"WARDEN_AGENT_ID": "grok"}
|
|
assert agent_read_boundary_identity(env) == "grok"
|
|
|
|
|
|
def test_non_agent_policy_subject_does_not_manufacture_agent_identity():
|
|
env = {"WARDEN_POLICY_SUBJECT": "iam:bernd"}
|
|
assert agent_read_boundary_identity(env) is None
|
|
|
|
|
|
def test_explicit_empty_environment_does_not_inherit_process_identity(monkeypatch):
|
|
monkeypatch.setenv("WARDEN_AGENT_ID", "codex")
|
|
monkeypatch.setenv("WARDEN_POLICY_SUBJECT", "service:codex:railiance-platform")
|
|
assert agent_read_boundary_identity({}) is None
|
|
assert agent_read_boundary_identity() == "service:codex:railiance-platform"
|
|
|
|
|
|
def test_operator_subject_does_not_override_legacy_agent_marker():
|
|
env = {"WARDEN_POLICY_SUBJECT": "iam:bernd", "WARDEN_AGENT_ID": "codex"}
|
|
assert agent_read_boundary_identity(env) == "codex"
|
|
|
|
|
|
# --- CLI -------------------------------------------------------------------
|
|
|
|
def test_access_advisory_shows_ambient_openbao_caller(monkeypatch):
|
|
monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(_repo_catalog()))
|
|
|
|
def described(env=None, **kwargs):
|
|
return {
|
|
"source": "BAO_TOKEN",
|
|
"entity_id": "4a1f663c-0621-0483-edd0-f852274f4b6b",
|
|
"display_name": "platform-root",
|
|
"auth_path": "auth/netkingdom/oidc/callback",
|
|
"policies": ["agent-high-risk-boundary"],
|
|
"oidc_role": "coding-agent",
|
|
"ttl": 900,
|
|
"error": None,
|
|
}
|
|
|
|
monkeypatch.setattr("warden.openbao_caller.describe_ambient_openbao_caller", described)
|
|
r = runner.invoke(app, ["access", "reuse-surface hub write token", "--json"])
|
|
assert r.exit_code == 0
|
|
payload = json.loads(r.stdout)
|
|
caller = payload["openbao_caller"]
|
|
assert caller["entity_id"] == "4a1f663c-0621-0483-edd0-f852274f4b6b"
|
|
assert "agent-high-risk-boundary" in caller["boundary_note"]
|
|
assert "id" not in caller
|
|
|
|
|
|
def test_access_advisory_output(monkeypatch):
|
|
monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(_repo_catalog()))
|
|
r = runner.invoke(app, ["access", "npm token", "--domain", "coulomb_social"])
|
|
assert r.exit_code == 0
|
|
assert "railiance-platform" in r.stdout
|
|
assert "platform/workloads/coulomb_social/" in r.stdout
|
|
# npm is an exec_capable lane → the front door leads with the proxy, not "owner vends".
|
|
assert "can fetch this for you" in r.stdout
|
|
assert "never holds" in r.stdout
|
|
|
|
|
|
def test_access_native_exec_shows_primary_and_fallback(monkeypatch):
|
|
"""A secrets-engine-owned lane leads with the native exec; proxy is the fallback."""
|
|
monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(_repo_catalog()))
|
|
r = runner.invoke(app, ["access", "whynot-design-npm-publish"])
|
|
assert r.exit_code == 0
|
|
assert "secrets-engine exec --catalog whynot-design-npm-publish" in r.stdout
|
|
assert "Primary" in r.stdout and "Fallback" in r.stdout
|
|
|
|
|
|
def test_access_route_only_lane_says_owner_vends(monkeypatch):
|
|
"""A non-exec lane (host principal deploy) keeps the advise-only framing."""
|
|
monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(_repo_catalog()))
|
|
r = runner.invoke(app, ["access", "host principal deploy"])
|
|
assert r.exit_code == 0
|
|
assert "warden advises, the owner vends" in r.stdout
|
|
|
|
|
|
def test_access_json_shape_is_secret_free(monkeypatch):
|
|
monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(_repo_catalog()))
|
|
r = runner.invoke(app, ["access", "npm token", "--domain", "coulomb_social", "--json"])
|
|
assert r.exit_code == 0
|
|
payload = json.loads(r.stdout)
|
|
assert payload["id"] == "openbao-api-key"
|
|
assert payload["domain"] == "coulomb_social"
|
|
assert payload["handoff"]["exec_capable"] is True
|
|
# only placeholders/templates — never a concrete credential
|
|
assert "<FIELD>" in payload["handoff"]["fetch_command"]
|
|
|
|
|
|
def test_access_ssh_lane_points_to_sign(monkeypatch):
|
|
monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(_repo_catalog()))
|
|
r = runner.invoke(app, ["access", "ssh cert for host access"])
|
|
assert r.exit_code == 0
|
|
assert "issues this directly" in r.stdout
|
|
assert "warden sign" in r.stdout
|
|
|
|
|
|
def test_access_no_match_exits_nonzero(monkeypatch):
|
|
monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(_repo_catalog()))
|
|
r = runner.invoke(app, ["access", "zzzz qqqq xyzzy"])
|
|
assert r.exit_code == 1
|