security(hermes): isolate HUX release evidence trust
This commit is contained in:
jenkins
parent
1eda927b71
commit
47165274f0
@ -11,6 +11,7 @@ reach at all.
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from collections.abc import Mapping
|
||||
|
||||
from hux.errors import FlagOff
|
||||
@ -18,6 +19,7 @@ from hux.identity import Identity
|
||||
from hux.rules import flag_enabled, flag_registry
|
||||
|
||||
CONTRACT_VERSION = "1.1.0"
|
||||
RELEASE_TAG = re.compile(r"^git-([0-9a-f]{40})-build-[1-9][0-9]*-release$")
|
||||
CARD_ROUTES: dict[str, list[str]] = {
|
||||
"HUX-11": ["/hux/v1/capabilities", "/hux/v1/manifest", "/hux/v1/context/bootstrap"],
|
||||
"HUX-01": ["/hux/v1/conversations/{id}/events", "/hux/v1/conversations/{id}/events/stream"],
|
||||
@ -66,10 +68,8 @@ WORKER_ROUTES: frozenset[tuple[str, str]] = frozenset({
|
||||
("GET", "/hux/v1/modes"),
|
||||
("GET", "/hux/v1/projects/{project_id}/conversations/{id}/mode"),
|
||||
("POST", "/hux/v1/projects/{project_id}/conversations/{id}/multimodal/items"),
|
||||
("POST", "/hux/v1/projects/{project_id}/conversations/{id}/releases"),
|
||||
("GET", "/hux/v1/projects/{project_id}/conversations/{id}/releases"),
|
||||
("GET", "/hux/v1/projects/{project_id}/conversations/{id}/releases/{release_id}"),
|
||||
("POST", "/hux/v1/projects/{project_id}/conversations/{id}/releases/{release_id}/transitions"),
|
||||
})
|
||||
|
||||
|
||||
@ -102,13 +102,23 @@ class Flags:
|
||||
def enabled(self, card: str) -> bool:
|
||||
"""True only for a route-backed card whose configured flag chain is on."""
|
||||
entry = self._registry.get(card)
|
||||
return bool(entry) and card in self._route_cards and flag_enabled(entry["flag"], self._environ)
|
||||
enabled = bool(entry) and card in self._route_cards and flag_enabled(entry["flag"], self._environ)
|
||||
if enabled and card == "HUX-12":
|
||||
from hux.release_security import configured
|
||||
|
||||
return configured(self._environ)
|
||||
return enabled
|
||||
|
||||
def require(self, card: str) -> None:
|
||||
"""Raise FlagOff unless the card is enabled."""
|
||||
if not self.enabled(card):
|
||||
raise FlagOff(f"{card} is not enabled")
|
||||
|
||||
@property
|
||||
def environ(self) -> Mapping[str, str]:
|
||||
"""Read-only process configuration for route-family policy checks."""
|
||||
return self._environ
|
||||
|
||||
def capabilities(self, identity: Identity, build: Mapping[str, str] | None = None) -> dict:
|
||||
"""Serialise ``hux.capabilities.v1`` for one caller."""
|
||||
cards = [
|
||||
@ -128,4 +138,13 @@ class Flags:
|
||||
def build_from_environ(environ: Mapping[str, str] | None = None) -> dict[str, str]:
|
||||
"""Commit and image digest the pod was started with, when the operator set them."""
|
||||
environ = os.environ if environ is None else environ
|
||||
return {"commit": environ.get("HUX_BUILD_COMMIT", ""), "image_digest": environ.get("HUX_IMAGE_DIGEST", "")}
|
||||
tag = environ.get("HUX_IMAGE_TAG", "")
|
||||
commit = environ.get("HUX_BUILD_COMMIT", "")
|
||||
if tag:
|
||||
match = RELEASE_TAG.fullmatch(tag)
|
||||
if not match:
|
||||
raise ValueError("HUX_IMAGE_TAG is not an immutable WebUI release tag")
|
||||
if commit and commit != match.group(1):
|
||||
raise ValueError("HUX build commit conflicts with the immutable image tag")
|
||||
commit = match.group(1)
|
||||
return {"commit": commit, "image_digest": environ.get("HUX_IMAGE_DIGEST", "")}
|
||||
|
||||
@ -207,6 +207,15 @@ class Router:
|
||||
# worker cannot even learn which cards are on.
|
||||
if identity.trust == "worker" and not worker_may_call(method, route.template):
|
||||
raise Forbidden("route is not available to worker trust")
|
||||
if identity.trust == "evidence":
|
||||
allowed = {
|
||||
("GET", "/hux/v1/capabilities"),
|
||||
("GET", "/hux/v1/projects/{project_id}/conversations/{id}/releases"),
|
||||
("GET", "/hux/v1/projects/{project_id}/conversations/{id}/releases/{release_id}"),
|
||||
("POST", "/hux/v1/projects/{project_id}/conversations/{id}/releases/{release_id}/transitions"),
|
||||
}
|
||||
if (method, route.template) not in allowed:
|
||||
raise Forbidden("route is not available to evidence trust")
|
||||
self.flags.require(route.card)
|
||||
try:
|
||||
parsed_query = parse_qs(parts.query, max_num_fields=MAX_QUERY_FIELDS)
|
||||
|
||||
@ -28,8 +28,11 @@ HEADER_KEY = "X-Hux-Relay-Key"
|
||||
SLOT_RE = re.compile(r"^slot-[0-9]{1,3}$")
|
||||
SUBJECT_RE = re.compile(r"^usr_[0-9a-f]{16,64}$")
|
||||
SURFACES = ("chat", "worker", "telegram", "voice", "api")
|
||||
TRUSTS = ("router", "relay", "worker")
|
||||
KEY_ENV = {"router": "HUX_ROUTER_KEY", "relay": "HUX_RELAY_KEY", "worker": "HUX_WORKER_KEY"}
|
||||
TRUSTS = ("router", "relay", "worker", "evidence")
|
||||
KEY_ENV = {
|
||||
"router": "HUX_ROUTER_KEY", "relay": "HUX_RELAY_KEY",
|
||||
"worker": "HUX_WORKER_KEY", "evidence": "HUX_RELEASE_EVIDENCE_KEY",
|
||||
}
|
||||
MAX_KEY_BYTES = 4096
|
||||
MAX_SUBJECT_BYTES = 128
|
||||
SUBJECT_BINDING_ENV = "HUX_SUBJECT_BINDING_FILE"
|
||||
@ -58,6 +61,13 @@ def _header(headers: Mapping[str, str], name: str) -> str:
|
||||
|
||||
def _expected_key(environ: Mapping[str, str], trust: str) -> str:
|
||||
"""Read a caller key, preferring a bounded Kubernetes secret file."""
|
||||
if trust == "evidence":
|
||||
from hux.release_security import evidence_key
|
||||
|
||||
try:
|
||||
return evidence_key(environ)
|
||||
except Exception: # noqa: BLE001 - authentication fails closed without revealing configuration detail
|
||||
return ""
|
||||
name = KEY_ENV[trust]
|
||||
key_file = environ.get(f"{name}_FILE", "")
|
||||
if key_file:
|
||||
@ -180,5 +190,7 @@ def resolve(headers: Mapping[str, str], environ: Mapping[str, str] | None = None
|
||||
raise Unauthorized(f"{trust} key missing or wrong")
|
||||
if trust == "router" and surface == "worker":
|
||||
raise Unauthorized("worker surface needs worker trust")
|
||||
if trust == "evidence" and surface != "api":
|
||||
raise Unauthorized("evidence trust needs api surface")
|
||||
_enforce_subject_binding(environ, trust, subject)
|
||||
return Identity(slot, subject, surface, trust)
|
||||
|
||||
134
dockerfiles/hermes-hux-foundation/hux/release_security.py
Normal file
134
dockerfiles/hermes-hux-foundation/hux/release_security.py
Normal file
@ -0,0 +1,134 @@
|
||||
"""Fail-closed authentication and policy checks for HUX-12 evidence.
|
||||
|
||||
The evidence producer is a loopback-only companion process. It receives a
|
||||
dedicated Vault key and a non-secret policy file; neither the browser relay nor
|
||||
the agent worker can use its trust class. This module deliberately performs no
|
||||
network discovery: the producer supplies observations and this boundary binds
|
||||
them to the configured Jenkins, Harbor, Flux, workload, pod and health facts.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import stat
|
||||
from collections.abc import Mapping
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
from hux.errors import Forbidden, Invalid
|
||||
|
||||
KEY_FILE_ENV = "HUX_RELEASE_EVIDENCE_KEY_FILE"
|
||||
POLICY_FILE_ENV = "HUX_RELEASE_EVIDENCE_POLICY_FILE"
|
||||
MAX_KEY_BYTES = 4096
|
||||
MAX_POLICY_BYTES = 64 * 1024
|
||||
POLICY_FIELDS = {"schema", "max_evidence_age_seconds", "workloads"}
|
||||
WORKLOAD_FIELDS = {
|
||||
"review_url_prefix", "jenkins_job_url", "image_repository",
|
||||
"flux_kustomization", "health_url",
|
||||
}
|
||||
|
||||
|
||||
def _regular_file(path: str, modes: set[int], limit: int) -> bytes:
|
||||
"""Read one bounded regular file without accepting a symlink or loose mode."""
|
||||
if not path:
|
||||
raise Invalid("HUX-12 evidence configuration is missing")
|
||||
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
|
||||
descriptor = -1
|
||||
try:
|
||||
descriptor = os.open(Path(path), flags)
|
||||
info = os.fstat(descriptor)
|
||||
if not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) not in modes:
|
||||
raise Invalid("HUX-12 evidence file permissions are invalid")
|
||||
data = os.read(descriptor, limit + 1)
|
||||
except OSError as error:
|
||||
raise Invalid("HUX-12 evidence configuration is unavailable") from error
|
||||
finally:
|
||||
if descriptor >= 0:
|
||||
os.close(descriptor)
|
||||
if not data or len(data) > limit:
|
||||
raise Invalid("HUX-12 evidence file is empty or too large")
|
||||
return data
|
||||
|
||||
|
||||
def evidence_key(environ: Mapping[str, str]) -> str:
|
||||
"""Return the file-only producer key; inline environment secrets are ignored."""
|
||||
data = _regular_file(environ.get(KEY_FILE_ENV, ""), {0o400}, MAX_KEY_BYTES)
|
||||
try:
|
||||
key = data.decode("utf-8", errors="strict").strip()
|
||||
except UnicodeDecodeError as error:
|
||||
raise Invalid("HUX-12 evidence key is invalid") from error
|
||||
if len(key) < 32:
|
||||
raise Invalid("HUX-12 evidence key is too short")
|
||||
return key
|
||||
|
||||
|
||||
def _https_url(value: Any, field: str, *, prefix: bool = False) -> str:
|
||||
if not isinstance(value, str) or not value or len(value) > 500:
|
||||
raise Invalid(f"{field} is missing or too long")
|
||||
parsed = urlsplit(value)
|
||||
if parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password or parsed.query or parsed.fragment:
|
||||
raise Invalid(f"{field} must be a credential-free HTTPS URL")
|
||||
if prefix and not value.endswith("/"):
|
||||
raise Invalid(f"{field} must end with /")
|
||||
return value
|
||||
|
||||
|
||||
def load_policy(environ: Mapping[str, str]) -> dict[str, Any]:
|
||||
"""Load and strictly validate the non-secret producer policy."""
|
||||
raw = _regular_file(environ.get(POLICY_FILE_ENV, ""), {0o400, 0o440, 0o444}, MAX_POLICY_BYTES)
|
||||
try:
|
||||
policy = json.loads(raw)
|
||||
except (UnicodeDecodeError, json.JSONDecodeError) as error:
|
||||
raise Invalid("HUX-12 evidence policy is not valid JSON") from error
|
||||
if not isinstance(policy, dict) or set(policy) != POLICY_FIELDS or policy.get("schema") != "hux.release_evidence_policy.v1":
|
||||
raise Invalid("HUX-12 evidence policy shape is invalid")
|
||||
age = policy.get("max_evidence_age_seconds")
|
||||
workloads = policy.get("workloads")
|
||||
if not isinstance(age, int) or isinstance(age, bool) or not 60 <= age <= 86400:
|
||||
raise Invalid("HUX-12 evidence age bound is invalid")
|
||||
if not isinstance(workloads, dict) or not workloads or len(workloads) > 16:
|
||||
raise Invalid("HUX-12 workload policy is invalid")
|
||||
for name, item in workloads.items():
|
||||
if not isinstance(name, str) or not isinstance(item, dict) or set(item) != WORKLOAD_FIELDS:
|
||||
raise Invalid("HUX-12 workload policy shape is invalid")
|
||||
_https_url(item["review_url_prefix"], "review_url_prefix", prefix=True)
|
||||
_https_url(item["jenkins_job_url"], "jenkins_job_url")
|
||||
_https_url(item["health_url"], "health_url")
|
||||
if not re.fullmatch(r"[a-z0-9./_-]+", str(item["image_repository"])):
|
||||
raise Invalid("HUX-12 image repository is invalid")
|
||||
if not re.fullmatch(r"[a-z0-9-]{1,63}", str(item["flux_kustomization"])):
|
||||
raise Invalid("HUX-12 Flux kustomization is invalid")
|
||||
return policy
|
||||
|
||||
|
||||
def configured(environ: Mapping[str, str]) -> bool:
|
||||
"""Whether producer authentication and policy are both healthy."""
|
||||
try:
|
||||
evidence_key(environ)
|
||||
load_policy(environ)
|
||||
except Invalid:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def workload_policy(environ: Mapping[str, str], workload: str) -> dict[str, Any]:
|
||||
"""Return the exact allowlist for a workload or fail closed."""
|
||||
item = load_policy(environ)["workloads"].get(workload)
|
||||
if item is None:
|
||||
raise Invalid("workload is not enabled for release evidence")
|
||||
return item
|
||||
|
||||
|
||||
def require_review_creator(trust: str) -> None:
|
||||
"""Only the authenticated browser path may record the human review decision."""
|
||||
if trust != "router":
|
||||
raise Forbidden("reviewed releases require router trust")
|
||||
|
||||
|
||||
def require_evidence_producer(trust: str) -> None:
|
||||
"""Only the dedicated producer may advance a reviewed release."""
|
||||
if trust != "evidence":
|
||||
raise Forbidden("release transitions require evidence trust")
|
||||
@ -14,7 +14,7 @@ import re
|
||||
from typing import Any
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
from hux import contracts, redaction
|
||||
from hux import contracts, redaction, release_security
|
||||
from hux.errors import Conflict, Invalid, NotFound, TooLarge
|
||||
from hux.http import Request, Response, Router, page
|
||||
from hux.store import check_id, new_id, now_iso
|
||||
@ -56,6 +56,8 @@ def _scope(request: Request) -> tuple[str, str]:
|
||||
|
||||
|
||||
def _actor(request: Request) -> dict[str, str]:
|
||||
if request.identity.trust == "evidence":
|
||||
return {"type": "system", "id": "release-evidence-producer"}
|
||||
if request.identity.trust == "worker":
|
||||
return {"type": "system", "id": "release-worker"}
|
||||
return {"type": "user", "id": request.identity.subject}
|
||||
@ -173,6 +175,7 @@ def _initial_evidence(body: dict[str, Any]) -> dict[str, Any]:
|
||||
|
||||
def create_release(request: Request) -> Response:
|
||||
"""Create the sole reviewed ledger for one scoped workload commit."""
|
||||
release_security.require_review_creator(request.identity.trust)
|
||||
body = _body(request, {"workload", "commit", "feature_flags", "evidence"})
|
||||
project_id, conversation_id = _scope(request)
|
||||
key, fingerprint = _key(request), _fingerprint(body, request.path)
|
||||
@ -262,6 +265,7 @@ def _evidence_refs(target: str, evidence: dict[str, Any]) -> list[dict[str, str]
|
||||
|
||||
def transition_release(request: Request) -> Response:
|
||||
"""Append exactly one legal, evidence-backed release transition."""
|
||||
release_security.require_evidence_producer(request.identity.trust)
|
||||
body = _body(request, {"to", "evidence"})
|
||||
project_id, conversation_id = _scope(request)
|
||||
key, fingerprint = _key(request), _fingerprint(body, request.path)
|
||||
|
||||
@ -250,11 +250,12 @@
|
||||
},
|
||||
"trust": {
|
||||
"type": "string",
|
||||
"description": "How the identity was asserted: router (browser via chat router), relay (Telegram relay key), worker (owner control plane).",
|
||||
"description": "How identity was asserted: router, relay, worker, or the loopback-only release evidence producer.",
|
||||
"enum": [
|
||||
"router",
|
||||
"relay",
|
||||
"worker"
|
||||
"worker",
|
||||
"evidence"
|
||||
]
|
||||
},
|
||||
"identity": {
|
||||
|
||||
@ -22,9 +22,9 @@ FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
|
||||
if str(FOUNDATION) not in sys.path:
|
||||
sys.path.insert(0, str(FOUNDATION))
|
||||
|
||||
from hux import audit, contracts, errors, flags, identity, store
|
||||
from hux.http import Response, Router, page, serve
|
||||
from hux.server import build_router
|
||||
from hux import audit, contracts, errors, flags, identity, store # noqa: E402
|
||||
from hux.http import Response, Router, page, serve # noqa: E402
|
||||
from hux.server import build_router # noqa: E402
|
||||
|
||||
SCHEMAS = contracts.load_all()
|
||||
HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "rk"}
|
||||
@ -127,11 +127,24 @@ def test_flags_fail_closed_and_capabilities_validate():
|
||||
record = on.capabilities(ident(), {"commit": "d3cbeb06" * 5, "image_digest": "sha256:" + "4a" * 32, "junk": "x"})
|
||||
assert contracts.validate_record(record, SCHEMAS) == []
|
||||
enabled = {card["card"]: card["enabled"] for card in record["cards"]}
|
||||
assert all(enabled[card] for card, routes in flags.CARD_ROUTES.items() if routes)
|
||||
assert all(enabled[card] for card, routes in flags.CARD_ROUTES.items() if routes and card != "HUX-12")
|
||||
assert not enabled["HUX-12"]
|
||||
assert all(not enabled[card] for card, routes in flags.CARD_ROUTES.items() if not routes)
|
||||
assert {card["card"] for card in record["cards"]} == {f"HUX-{n:02d}" for n in range(1, 13)}
|
||||
assert not on.enabled("HUX-99")
|
||||
assert flags.build_from_environ({"HUX_BUILD_COMMIT": "abc"}) == {"commit": "abc", "image_digest": ""}
|
||||
release_commit = "d3cbeb06" * 5
|
||||
release_tag = f"git-{release_commit}-build-23-release"
|
||||
assert flags.build_from_environ({"HUX_IMAGE_TAG": release_tag}) == {
|
||||
"commit": release_commit,
|
||||
"image_digest": "",
|
||||
}
|
||||
with pytest.raises(ValueError):
|
||||
flags.build_from_environ({"HUX_IMAGE_TAG": "latest"})
|
||||
with pytest.raises(ValueError):
|
||||
flags.build_from_environ(
|
||||
{"HUX_IMAGE_TAG": release_tag, "HUX_BUILD_COMMIT": "a" * 40}
|
||||
)
|
||||
assert "commit" in flags.build_from_environ()
|
||||
|
||||
|
||||
@ -442,7 +455,8 @@ def test_new_backend_cards_declare_routes_and_enable_with_dependencies():
|
||||
shipped = {"HUX-06", "HUX-07", "HUX-09", "HUX-12"}
|
||||
configured = flags.Flags({"HUX_FLAGS": ALL_ON})
|
||||
assert all(flags.CARD_ROUTES[card] for card in shipped)
|
||||
assert all(configured.enabled(card) for card in shipped)
|
||||
assert all(configured.enabled(card) for card in shipped - {"HUX-12"})
|
||||
assert not configured.enabled("HUX-12")
|
||||
|
||||
|
||||
def test_unexpected_handler_exceptions_become_a_500_error_record(tmp_path):
|
||||
|
||||
@ -10,7 +10,7 @@ FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
|
||||
if str(FOUNDATION) not in sys.path:
|
||||
sys.path.insert(0, str(FOUNDATION))
|
||||
|
||||
from hux import contracts, flags, server
|
||||
from hux import contracts, flags, server # noqa: E402
|
||||
|
||||
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
|
||||
EXPECTED_FAMILIES = ("modes", "multimodal", "suggestions", "releases")
|
||||
@ -47,7 +47,7 @@ def test_server_registers_each_family_and_exact_card_routes(tmp_path):
|
||||
for card, expected in EXPECTED_ROUTES.items():
|
||||
assert set(flags.CARD_ROUTES[card]) == expected
|
||||
assert expected <= registered
|
||||
assert router.flags.enabled(card)
|
||||
assert router.flags.enabled(card) is (card != "HUX-12")
|
||||
|
||||
|
||||
def test_release_followthrough_requires_foundation_but_not_activity():
|
||||
@ -56,8 +56,8 @@ def test_release_followthrough_requires_foundation_but_not_activity():
|
||||
assert release["depends_on"] == ["HUX-11"]
|
||||
assert release["contracts"] == ["release.schema.json", "release-ledger.schema.json"]
|
||||
assert not flags.Flags({"HUX_FLAGS": "hux.release_followthrough"}).enabled("HUX-12")
|
||||
enabled = flags.Flags({"HUX_FLAGS": "hux.foundation,hux.release_followthrough"})
|
||||
assert enabled.enabled("HUX-12")
|
||||
configured_without_evidence_key = flags.Flags({"HUX_FLAGS": "hux.foundation,hux.release_followthrough"})
|
||||
assert not configured_without_evidence_key.enabled("HUX-12")
|
||||
|
||||
|
||||
def test_contract_loader_and_worker_allowlist_are_complete():
|
||||
@ -69,10 +69,8 @@ def test_contract_loader_and_worker_allowlist_are_complete():
|
||||
("GET", "/hux/v1/modes"),
|
||||
("GET", "/hux/v1/projects/{project_id}/conversations/{id}/mode"),
|
||||
("POST", "/hux/v1/projects/{project_id}/conversations/{id}/multimodal/items"),
|
||||
("POST", "/hux/v1/projects/{project_id}/conversations/{id}/releases"),
|
||||
("GET", "/hux/v1/projects/{project_id}/conversations/{id}/releases"),
|
||||
("GET", "/hux/v1/projects/{project_id}/conversations/{id}/releases/{release_id}"),
|
||||
("POST", "/hux/v1/projects/{project_id}/conversations/{id}/releases/{release_id}/transitions"),
|
||||
}
|
||||
assert worker_expected <= flags.WORKER_ROUTES
|
||||
assert ("PUT", "/hux/v1/projects/{project_id}/conversations/{id}/mode") not in flags.WORKER_ROUTES
|
||||
|
||||
177
testing/tests/test_hermes_hux_release_security.py
Normal file
177
testing/tests/test_hermes_hux_release_security.py
Normal file
@ -0,0 +1,177 @@
|
||||
"""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
|
||||
@ -14,11 +14,12 @@ FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
|
||||
if str(FOUNDATION) not in sys.path:
|
||||
sys.path.insert(0, str(FOUNDATION))
|
||||
|
||||
from hux import contracts, identity, releases, store
|
||||
from hux.errors import Invalid
|
||||
from hux.server import build_router
|
||||
from hux import contracts, identity, releases, store # noqa: E402
|
||||
from hux.errors import Invalid # noqa: E402
|
||||
from hux.server import build_router # noqa: E402
|
||||
|
||||
HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "rk"}
|
||||
EVIDENCE = {**HEADERS, "X-Hux-Surface": "api", "X-Hux-Trust": "evidence", "X-Hux-Relay-Key": "evidence-key-value-that-is-at-least-32-bytes"}
|
||||
OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"}
|
||||
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
|
||||
DIGEST = "sha256:" + "a" * 64
|
||||
@ -39,7 +40,25 @@ def tenant(router):
|
||||
|
||||
@pytest.fixture
|
||||
def setup(tmp_path):
|
||||
router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk"})
|
||||
key_file = tmp_path / "evidence.key"
|
||||
key_file.write_text("evidence-key-value-that-is-at-least-32-bytes\n")
|
||||
key_file.chmod(0o400)
|
||||
policy_file = tmp_path / "evidence-policy.json"
|
||||
policy_file.write_text(json.dumps({
|
||||
"schema": "hux.release_evidence_policy.v1", "max_evidence_age_seconds": 900,
|
||||
"workloads": {"hermes-webui": {
|
||||
"review_url_prefix": "https://git.bstein.dev/atlas/titan-iac/pulls/",
|
||||
"jenkins_job_url": "https://jenkins.bstein.dev/job/hermes",
|
||||
"image_repository": "registry.bstein.dev/bstein/hermes",
|
||||
"flux_kustomization": "hermes", "health_url": "https://chat.bstein.dev/healthz",
|
||||
}},
|
||||
}))
|
||||
policy_file.chmod(0o444)
|
||||
router = build_router(tmp_path, {
|
||||
"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk",
|
||||
"HUX_RELEASE_EVIDENCE_KEY_FILE": str(key_file),
|
||||
"HUX_RELEASE_EVIDENCE_POLICY_FILE": str(policy_file),
|
||||
})
|
||||
_, project, _ = call(router, "POST", "/hux/v1/projects", {"name": "P"})
|
||||
_, conversation, _ = call(router, "POST", "/hux/v1/conversations", {"title": "C", "project_id": project["id"]})
|
||||
base = f"/hux/v1/projects/{project['id']}/conversations/{conversation['id']}/releases"
|
||||
@ -56,7 +75,9 @@ def create(setup, key="release-key-001", commit="1" * 40, **changes):
|
||||
|
||||
|
||||
def move(router, path, revision, key, target, evidence):
|
||||
return call(router, "POST", path + "/transitions", {"to": target, "evidence": evidence}, {"If-Match": str(revision), "Idempotency-Key": key})
|
||||
return call(router, "POST", path + "/transitions", {"to": target, "evidence": evidence}, {
|
||||
**EVIDENCE, "If-Match": str(revision), "Idempotency-Key": key,
|
||||
})
|
||||
|
||||
|
||||
def build_evidence(digest=DIGEST):
|
||||
@ -162,7 +183,24 @@ def test_transition_requires_exact_order_revision_and_evidence(setup):
|
||||
status, replay, headers = move(router, path, 1, "good-merge-key", "merged", {"merge_commit": "2" * 40})
|
||||
assert status == 200 and replay == merged and headers["HUX-Replayed"] == "true"
|
||||
assert move(router, path, 2, "good-merge-key", "built", build_evidence())[0] == 409
|
||||
assert call(router, "POST", path + "/transitions", {"to": "built", "evidence": build_evidence()}, {"Idempotency-Key": "missing-transition-if"})[0] == 400
|
||||
assert call(router, "POST", path + "/transitions", {"to": "built", "evidence": build_evidence()}, {
|
||||
**EVIDENCE, "Idempotency-Key": "missing-transition-if",
|
||||
})[0] == 400
|
||||
|
||||
|
||||
def test_only_router_reviews_and_only_evidence_trust_transitions(setup):
|
||||
router, _, _, base = setup
|
||||
created = create(setup, "trust-bound-create")[1]
|
||||
path = base + "/" + created["release"]["id"]
|
||||
transition = {"to": "merged", "evidence": {"merge_commit": "2" * 40}}
|
||||
headers = {"If-Match": "1", "Idempotency-Key": "trust-bound-transition"}
|
||||
assert call(router, "POST", path + "/transitions", transition, headers)[0] == 403
|
||||
worker = {**HEADERS, "X-Hux-Surface": "worker", "X-Hux-Trust": "worker", "X-Hux-Relay-Key": "wk"}
|
||||
assert call(router, "POST", path + "/transitions", transition, {**worker, **headers})[0] == 401
|
||||
assert call(router, "POST", base, {
|
||||
"workload": "hermes-webui", "commit": "8" * 40,
|
||||
"evidence": {"review_url": "https://git.bstein.dev/atlas/titan-iac/pulls/8"},
|
||||
}, {**EVIDENCE, "If-Match": "0", "Idempotency-Key": "evidence-cannot-review"})[0] == 403
|
||||
|
||||
|
||||
@pytest.mark.parametrize("evidence", [
|
||||
@ -258,6 +296,8 @@ def test_inconsistent_idempotency_and_cross_conversation_views_fail_closed(setup
|
||||
def test_internal_helpers_are_bounded_and_worker_actor_has_no_subject():
|
||||
worker = SimpleNamespace(identity=identity.Identity("slot-3", "usr_0123456789abcdef", "worker", "worker"))
|
||||
assert releases._actor(worker) == {"type": "system", "id": "release-worker"}
|
||||
producer = SimpleNamespace(identity=identity.Identity("slot-3", "usr_0123456789abcdef", "api", "evidence"))
|
||||
assert releases._actor(producer) == {"type": "system", "id": "release-evidence-producer"}
|
||||
with pytest.raises(Invalid, match="missing or too long"):
|
||||
releases._safe_url(None)
|
||||
with pytest.raises(Invalid, match="secret"):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user