generated from coulomb/repo-seed
tegwick
41a55c95b0
Finish the Workload Security Posture workplan (all five tasks done). T3 — scripts/check_secret_posture_conformance.py: read-only checker that asserts env-posture conformance (backend/unseal/real_values per tier) and evaluates the secret-flow lattice via posture.can_deliver. Metadata-only manifest, no secret values, exit 0/1/2. examples/posture-conformance.example.yaml as the reference. T4 — src/warden/doubles.py: generalizes "fake bao" into materialize_doubles() — hermetic, synthetic-only (synthetic- prefix) stand-ins for bao/key-cape honoring each argv/stdout/exit contract, for fully offline dev/test access flows. Documented as the sanctioned dev backend in WorkloadSecurityPosture.md R1. T5 — INTENT/SCOPE/wiki aligned; canon landing in net-kingdom/info-tech-canon left owner-driven (tracked via coordination messages). 16 new tests, 200 passing, ruff clean. Archived WP-0012/0014/0015 to workplans/archived/ with 260627- prefix. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
115 lines
3.3 KiB
Python
115 lines
3.3 KiB
Python
"""Tests for the dev-tier contract-double fixture library (WP-0015 T4)."""
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from warden.doubles import (
|
|
SYNTHETIC_PREFIX,
|
|
available_doubles,
|
|
doubles_path_prepended,
|
|
materialize_doubles,
|
|
)
|
|
|
|
|
|
def test_available_doubles_includes_routed_subsystems():
|
|
names = available_doubles()
|
|
assert "bao" in names
|
|
assert "key-cape" in names
|
|
|
|
|
|
def test_materialize_writes_executables(tmp_path):
|
|
paths = materialize_doubles(tmp_path)
|
|
assert set(paths) == set(available_doubles())
|
|
for p in paths.values():
|
|
assert p.exists()
|
|
import os
|
|
|
|
assert os.access(p, os.X_OK)
|
|
|
|
|
|
def test_bao_kv_get_emits_synthetic_value(tmp_path):
|
|
materialize_doubles(tmp_path, ["bao"])
|
|
out = subprocess.run(
|
|
[str(tmp_path / "bao"), "kv", "get", "-field=NPM_AUTH_TOKEN", "platform/x/y"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
value = out.stdout.strip()
|
|
assert value.startswith(SYNTHETIC_PREFIX)
|
|
assert "NPM_AUTH_TOKEN" in value
|
|
|
|
|
|
def test_bao_login_emits_synthetic_token(tmp_path):
|
|
materialize_doubles(tmp_path, ["bao"])
|
|
out = subprocess.run(
|
|
[str(tmp_path / "bao"), "login", "-method=oidc"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
assert out.stdout.strip().startswith(SYNTHETIC_PREFIX)
|
|
|
|
|
|
def test_keycape_login_emits_synthetic_session(tmp_path):
|
|
materialize_doubles(tmp_path, ["key-cape"])
|
|
out = subprocess.run(
|
|
[str(tmp_path / "key-cape"), "login"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
assert out.stdout.strip().startswith(SYNTHETIC_PREFIX)
|
|
|
|
|
|
def test_double_rejects_unknown_contract(tmp_path):
|
|
materialize_doubles(tmp_path, ["bao"])
|
|
out = subprocess.run(
|
|
[str(tmp_path / "bao"), "write", "secret/x"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert out.returncode == 2
|
|
|
|
|
|
def test_unknown_double_raises(tmp_path):
|
|
with pytest.raises(KeyError):
|
|
materialize_doubles(tmp_path, ["nonesuch"])
|
|
|
|
|
|
def test_path_prepended_puts_doubles_first(tmp_path):
|
|
path = doubles_path_prepended(tmp_path, base_path="/usr/bin")
|
|
assert path.split(":")[0] == str(tmp_path)
|
|
|
|
|
|
def test_proxy_fetch_runs_fully_offline_against_double(tmp_path):
|
|
"""End-to-end: the proxy fetch lane resolves `bao` from the doubles dir."""
|
|
import os
|
|
|
|
materialize_doubles(tmp_path, ["bao"])
|
|
from warden.proxy import resolve_fetch_command
|
|
from warden.routing.models import RouteEntry
|
|
|
|
entry = RouteEntry(
|
|
id="openbao-api-key",
|
|
title="API key",
|
|
need_keywords=["npm"],
|
|
owner_repo="railiance-platform",
|
|
subsystem="OpenBao",
|
|
warden_executes=False,
|
|
wiki_ref="w",
|
|
canon_ref="c",
|
|
reviewed="2026-06-27",
|
|
status="active",
|
|
path_template="platform/x/y/z",
|
|
fetch_command="bao kv get -field=<FIELD> <path_template>",
|
|
exec_capable=True,
|
|
)
|
|
argv = resolve_fetch_command(entry, field="API_KEY", path="platform/x/y/z")
|
|
env = dict(os.environ, PATH=doubles_path_prepended(tmp_path))
|
|
# proxy_fetch inherits stdout; run it in a child so we can capture the stream.
|
|
result = subprocess.run(argv, capture_output=True, text=True, env=env, check=True)
|
|
assert result.stdout.strip().startswith(SYNTHETIC_PREFIX)
|