atlas-iac/testing/tests/test_hermes_hux_release_security.py

178 lines
6.8 KiB
Python

"""HUX-12 evidence identity and configuration fail-closed tests."""
from __future__ import annotations
import json
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
if str(FOUNDATION) not in sys.path:
sys.path.insert(0, str(FOUNDATION))
from hux import identity, release_security # noqa: E402
from hux.errors import Forbidden, Invalid, Unauthorized # noqa: E402
KEY = "evidence-key-value-that-is-at-least-32-bytes"
def policy() -> dict:
"""Return one complete non-secret release allowlist."""
return {
"schema": "hux.release_evidence_policy.v1",
"max_evidence_age_seconds": 900,
"workloads": {"hermes-webui": {
"review_url_prefix": "https://scm.bstein.dev/atlas/titan-iac/pulls/",
"jenkins_job_url": "https://jenkins.bstein.dev/job/hermes-webui-image",
"image_repository": "registry.bstein.dev/bstein/hermes-webui",
"flux_kustomization": "hermes",
"health_url": "https://chat.bstein.dev/healthz",
}},
}
def files(tmp_path: Path, content: bytes | None = None) -> dict[str, str]:
"""Create strict key/policy projections and return their environment."""
key = tmp_path / "evidence.key"
key.write_bytes((KEY + "\n").encode() if content is None else content)
key.chmod(0o400)
config = tmp_path / "policy.json"
config.write_text(json.dumps(policy()))
config.chmod(0o444)
return {
release_security.KEY_FILE_ENV: str(key),
release_security.POLICY_FILE_ENV: str(config),
}
def test_good_configuration_and_evidence_identity(tmp_path):
env = files(tmp_path)
env["HUX_TENANT_SLOT"] = "slot-3"
assert release_security.configured(env)
assert release_security.evidence_key(env) == KEY
assert release_security.workload_policy(env, "hermes-webui")["flux_kustomization"] == "hermes"
headers = {
"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef",
"X-Hux-Surface": "api", "X-Hux-Trust": "evidence", "X-Hux-Relay-Key": KEY,
}
assert identity.resolve(headers, env).trust == "evidence"
with pytest.raises(Unauthorized, match="api surface"):
identity.resolve({**headers, "X-Hux-Surface": "chat"}, env)
with pytest.raises(Unauthorized, match="key missing or wrong"):
identity.resolve(headers, {**env, release_security.KEY_FILE_ENV: "", "HUX_RELEASE_EVIDENCE_KEY": KEY})
@pytest.mark.parametrize("content", [b"short", b"\xff" * 40, b"", b"x" * 4097])
def test_key_rejects_invalid_contents(tmp_path, content):
env = files(tmp_path, content)
with pytest.raises(Invalid):
release_security.evidence_key(env)
assert not release_security.configured(env)
def test_key_rejects_missing_loose_and_symlink_files(tmp_path):
env = files(tmp_path)
Path(env[release_security.KEY_FILE_ENV]).chmod(0o440)
with pytest.raises(Invalid, match="permissions"):
release_security.evidence_key(env)
env[release_security.KEY_FILE_ENV] = str(tmp_path / "missing")
with pytest.raises(Invalid, match="unavailable"):
release_security.evidence_key(env)
target = tmp_path / "target"
target.write_text(KEY)
link = tmp_path / "link"
link.symlink_to(target)
env[release_security.KEY_FILE_ENV] = str(link)
with pytest.raises(Invalid, match="unavailable"):
release_security.evidence_key(env)
with pytest.raises(Invalid, match="missing"):
release_security._regular_file("", {0o400}, 10)
def test_policy_rejects_encoding_json_shape_and_bounds(tmp_path):
env = files(tmp_path)
path = Path(env[release_security.POLICY_FILE_ENV])
for value in (b"\xff", b"{", json.dumps([]).encode(), json.dumps({"schema": "bad"}).encode()):
path.chmod(0o600)
path.write_bytes(value)
path.chmod(0o444)
with pytest.raises(Invalid):
release_security.load_policy(env)
for age in (True, 59, 86401):
item = policy()
item["max_evidence_age_seconds"] = age
path.chmod(0o600)
path.write_text(json.dumps(item))
path.chmod(0o444)
with pytest.raises(Invalid, match="age bound"):
release_security.load_policy(env)
@pytest.mark.parametrize("workloads", [None, {}, {str(index): {} for index in range(17)}])
def test_policy_rejects_workload_collection(tmp_path, workloads):
env = files(tmp_path)
item = policy()
item["workloads"] = workloads
path = Path(env[release_security.POLICY_FILE_ENV])
path.chmod(0o600)
path.write_text(json.dumps(item))
path.chmod(0o444)
with pytest.raises(Invalid, match="workload policy"):
release_security.load_policy(env)
@pytest.mark.parametrize("field,value,message", [
("review_url_prefix", "https://scm.bstein.dev/pulls", "end with"),
("review_url_prefix", "http://scm.bstein.dev/pulls/", "HTTPS"),
("jenkins_job_url", "https://user:pass@jenkins.bstein.dev/job/x", "HTTPS"),
("health_url", "https://chat.bstein.dev/healthz?token=x", "HTTPS"),
("image_repository", "registry:5000/Bad", "image repository"),
("flux_kustomization", "BAD_name", "Flux kustomization"),
])
def test_policy_rejects_unsafe_workload_fields(tmp_path, field, value, message):
env = files(tmp_path)
item = policy()
item["workloads"]["hermes-webui"][field] = value
path = Path(env[release_security.POLICY_FILE_ENV])
path.chmod(0o600)
path.write_text(json.dumps(item))
path.chmod(0o444)
with pytest.raises(Invalid, match=message):
release_security.load_policy(env)
def test_policy_rejects_bad_workload_shape_and_unknown_name(tmp_path):
env = files(tmp_path)
item = policy()
item["workloads"]["hermes-webui"]["extra"] = True
path = Path(env[release_security.POLICY_FILE_ENV])
path.chmod(0o600)
path.write_text(json.dumps(item))
path.chmod(0o444)
with pytest.raises(Invalid, match="shape"):
release_security.load_policy(env)
path.chmod(0o600)
path.write_text(json.dumps(policy()))
path.chmod(0o444)
with pytest.raises(Invalid, match="not enabled"):
release_security.workload_policy(env, "hermes-agent")
def test_trust_gates_are_exact():
release_security.require_review_creator("router")
release_security.require_evidence_producer("evidence")
for trust in ("relay", "worker", "evidence"):
if trust != "router":
with pytest.raises(Forbidden, match="router trust"):
release_security.require_review_creator(trust)
for trust in ("router", "relay", "worker"):
with pytest.raises(Forbidden, match="evidence trust"):
release_security.require_evidence_producer(trust)
def test_source_is_bounded():
assert len((FOUNDATION / "hux" / "release_security.py").read_text().splitlines()) <= 500