tegwick
c26a6e59de
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02878-7c21-7692-bcd6-ce2838c4b448
266 lines
9.8 KiB
Python
266 lines
9.8 KiB
Python
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from ops_mason.kubernetes_plane import (
|
|
CommandResult,
|
|
PlaneBundle,
|
|
PlaneRefused,
|
|
_json_path,
|
|
apply,
|
|
preflight,
|
|
rollback_plan,
|
|
)
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_whitehat_bundle_is_exactly_four_allowlisted_objects() -> None:
|
|
bundle = PlaneBundle.load(ROOT / "bundles/whitehat-foundational-plane.yaml")
|
|
assert bundle.digest
|
|
assert [ref.display for ref in bundle.allowed_objects] == [
|
|
"Namespace/whitehat",
|
|
"NetworkPolicy/whitehat/default-deny",
|
|
"NetworkPolicy/whitehat/allow-audit-core-e2",
|
|
"ServiceAccount/whitehat/whitehat-runner",
|
|
]
|
|
plan = bundle.plan()
|
|
assert plan.status == "built"
|
|
assert plan.approved_by == "Bernd Worsch"
|
|
assert plan.approved_at == "2026-08-22"
|
|
assert {doc["kind"] for doc in bundle.documents} == {
|
|
"Namespace",
|
|
"NetworkPolicy",
|
|
"ServiceAccount",
|
|
}
|
|
|
|
|
|
def _fixture(tmp_path: Path, *, approved: bool = True, kind: str = "Namespace") -> Path:
|
|
(tmp_path / "bundles").mkdir()
|
|
(tmp_path / "manifests").mkdir()
|
|
(tmp_path / "plans").mkdir()
|
|
manifest = tmp_path / "manifests/plane.yaml"
|
|
if kind == "Namespace":
|
|
document = {
|
|
"apiVersion": "v1",
|
|
"kind": "Namespace",
|
|
"metadata": {
|
|
"name": "whitehat",
|
|
"labels": {"pod-security.kubernetes.io/enforce": "restricted"},
|
|
},
|
|
}
|
|
resource = "namespaces"
|
|
namespace = None
|
|
else:
|
|
document = {
|
|
"apiVersion": "v1",
|
|
"kind": kind,
|
|
"metadata": {"name": "forbidden", "namespace": "whitehat"},
|
|
}
|
|
resource = kind.lower() + "s"
|
|
namespace = "whitehat"
|
|
manifest.write_text(yaml.safe_dump(document, sort_keys=False))
|
|
digest = hashlib.sha256(manifest.read_bytes()).hexdigest()
|
|
plan = tmp_path / "plans/plane.md"
|
|
status = "approved" if approved else "reviewed"
|
|
approval = 'approved_by: "Bernd"\napproved_at: "2026-08-22"\n' if approved else ""
|
|
plan.write_text(f"---\nid: plane\nstatus: {status}\n{approval}---\n# Plan\n")
|
|
descriptor = {
|
|
"schema_version": "ops-mason.kubernetes-plane/v1",
|
|
"id": "plane",
|
|
"plan": "../plans/plane.md",
|
|
"expected_context": "default",
|
|
"expected_namespace": "whitehat",
|
|
"source_repo": "whitehat-security",
|
|
"source_revision": "abc",
|
|
"implementation_revision": "def",
|
|
"evidence_path": "../evidence/plane.json",
|
|
"forbidden_kinds": ["Pod", "Secret"],
|
|
"manifests": [
|
|
{
|
|
"path": "../manifests/plane.yaml",
|
|
"source_path": "plane.yaml",
|
|
"sha256": digest,
|
|
}
|
|
],
|
|
"allowed_objects": [
|
|
{
|
|
"api_version": "v1",
|
|
"kind": kind,
|
|
"resource": resource,
|
|
"namespace": namespace,
|
|
"name": document["metadata"]["name"],
|
|
}
|
|
],
|
|
}
|
|
bundle_path = tmp_path / "bundles/plane.yaml"
|
|
bundle_path.write_text(yaml.safe_dump(descriptor, sort_keys=False))
|
|
return bundle_path
|
|
|
|
|
|
def test_bundle_refuses_forbidden_pod_even_when_allowlisted(tmp_path: Path) -> None:
|
|
with pytest.raises(PlaneRefused, match="forbidden Kubernetes kind"):
|
|
PlaneBundle.load(_fixture(tmp_path, kind="Pod"))
|
|
|
|
|
|
def test_bundle_refuses_manifest_hash_drift(tmp_path: Path) -> None:
|
|
bundle_path = _fixture(tmp_path)
|
|
manifest = tmp_path / "manifests/plane.yaml"
|
|
manifest.write_text(manifest.read_text() + "# drift\n")
|
|
with pytest.raises(PlaneRefused, match="digest mismatch"):
|
|
PlaneBundle.load(bundle_path)
|
|
|
|
|
|
def test_bundle_refuses_secret_bearing_key_even_on_other_kind(tmp_path: Path) -> None:
|
|
bundle_path = _fixture(tmp_path)
|
|
manifest = tmp_path / "manifests/plane.yaml"
|
|
document = yaml.safe_load(manifest.read_text())
|
|
document["stringData"] = {"token": "must-never-enter-mason"}
|
|
manifest.write_text(yaml.safe_dump(document, sort_keys=False))
|
|
descriptor = yaml.safe_load(bundle_path.read_text())
|
|
descriptor["manifests"][0]["sha256"] = hashlib.sha256(manifest.read_bytes()).hexdigest()
|
|
bundle_path.write_text(yaml.safe_dump(descriptor, sort_keys=False))
|
|
with pytest.raises(PlaneRefused, match="secret-bearing"):
|
|
PlaneBundle.load(bundle_path)
|
|
|
|
|
|
def test_json_pointer_supports_label_keys_with_slashes_and_dots() -> None:
|
|
value = {"labels": {"kubernetes.io/metadata.name": "whitehat"}}
|
|
assert _json_path(value, "/labels/kubernetes.io~1metadata.name") == "whitehat"
|
|
|
|
|
|
class FakeCluster:
|
|
def __init__(
|
|
self, *, context: str = "default", drift: bool = False, dirty: bool = False
|
|
) -> None:
|
|
self.context = context
|
|
self.drift = drift
|
|
self.dirty = dirty
|
|
self.applied = False
|
|
self.calls: list[list[str]] = []
|
|
|
|
def __call__(self, args) -> CommandResult:
|
|
command = list(args)
|
|
self.calls.append(command)
|
|
if command[:4] == ["git", "-C", command[2], "status"]:
|
|
return CommandResult(0, " M src/ops_mason/kubernetes_plane.py\n" if self.dirty else "")
|
|
if command == ["kubectl", "config", "current-context"]:
|
|
return CommandResult(0, self.context + "\n")
|
|
if command[:4] == ["kubectl", "auth", "can-i", "create"]:
|
|
return CommandResult(0, "yes\n")
|
|
if "apply" in command:
|
|
if "--dry-run=client" not in command and "--dry-run=server" not in command:
|
|
self.applied = True
|
|
return CommandResult(0, "configured\n")
|
|
if command[:3] == ["kubectl", "get", "namespaces"]:
|
|
if not self.applied:
|
|
return CommandResult(1, "", 'Error from server (NotFound): namespaces "whitehat" not found')
|
|
labels = {"pod-security.kubernetes.io/enforce": "baseline" if self.drift else "restricted"}
|
|
return CommandResult(
|
|
0,
|
|
json.dumps(
|
|
{
|
|
"apiVersion": "v1",
|
|
"kind": "Namespace",
|
|
"metadata": {
|
|
"name": "whitehat",
|
|
"labels": labels,
|
|
"uid": "uid-1",
|
|
"resourceVersion": "10",
|
|
},
|
|
}
|
|
),
|
|
)
|
|
if command[:4] == ["kubectl", "-n", "whitehat", "get"]:
|
|
return CommandResult(0, json.dumps({"items": []}))
|
|
raise AssertionError(f"unexpected command: {command}")
|
|
|
|
|
|
def test_preflight_refuses_wrong_context_before_kubectl_apply(tmp_path: Path) -> None:
|
|
bundle = PlaneBundle.load(_fixture(tmp_path))
|
|
cluster = FakeCluster(context="wrong")
|
|
with pytest.raises(PlaneRefused, match="context mismatch"):
|
|
preflight(bundle, cluster)
|
|
assert not any("apply" in call for call in cluster.calls)
|
|
|
|
|
|
def test_apply_refuses_unapproved_plan_without_calling_runner(tmp_path: Path) -> None:
|
|
bundle = PlaneBundle.load(_fixture(tmp_path, approved=False))
|
|
calls = []
|
|
|
|
def runner(args):
|
|
calls.append(args)
|
|
raise AssertionError("runner must not be called")
|
|
|
|
with pytest.raises(PlaneRefused, match="not approved"):
|
|
apply(bundle, confirm_plan_id="plane", expected_digest=bundle.digest, runner=runner)
|
|
assert calls == []
|
|
|
|
|
|
def test_apply_refuses_digest_mismatch_without_calling_runner(tmp_path: Path) -> None:
|
|
bundle = PlaneBundle.load(_fixture(tmp_path))
|
|
calls = []
|
|
|
|
def runner(args):
|
|
calls.append(args)
|
|
raise AssertionError("runner must not be called")
|
|
|
|
with pytest.raises(PlaneRefused, match="digest confirmation mismatch"):
|
|
apply(bundle, confirm_plan_id="plane", expected_digest="wrong", runner=runner)
|
|
assert calls == []
|
|
|
|
|
|
def test_apply_refuses_dirty_repository_before_kubectl(tmp_path: Path) -> None:
|
|
bundle = PlaneBundle.load(_fixture(tmp_path))
|
|
cluster = FakeCluster(dirty=True)
|
|
with pytest.raises(PlaneRefused, match="committed and clean"):
|
|
apply(
|
|
bundle,
|
|
confirm_plan_id="plane",
|
|
expected_digest=bundle.digest,
|
|
runner=cluster,
|
|
)
|
|
assert not any(call and call[0] == "kubectl" for call in cluster.calls)
|
|
|
|
|
|
def test_apply_runs_guarded_path_and_writes_metadata_only_evidence(tmp_path: Path) -> None:
|
|
bundle = PlaneBundle.load(_fixture(tmp_path))
|
|
cluster = FakeCluster()
|
|
evidence = apply(
|
|
bundle,
|
|
confirm_plan_id="plane",
|
|
expected_digest=bundle.digest,
|
|
runner=cluster,
|
|
)
|
|
assert evidence["verification"]["negative_scope"] == {"pods": 0, "secrets": 0}
|
|
assert evidence["apply"] == {
|
|
"field_manager": "ops-mason",
|
|
"server_validated_manifests": ["manifests/plane.yaml"],
|
|
"persisted_manifests": ["manifests/plane.yaml"],
|
|
}
|
|
assert bundle.evidence_path.exists()
|
|
text = bundle.evidence_path.read_text()
|
|
assert "uid-1" in text
|
|
assert "data" not in evidence["verification"]
|
|
mutating = [call for call in cluster.calls if "apply" in call and "--dry-run=server" not in call and "--dry-run=client" not in call]
|
|
assert len(mutating) == 1
|
|
assert "--field-manager=ops-mason" in mutating[0]
|
|
|
|
|
|
def test_preflight_refuses_unmanaged_live_drift(tmp_path: Path) -> None:
|
|
bundle = PlaneBundle.load(_fixture(tmp_path))
|
|
cluster = FakeCluster(drift=True)
|
|
cluster.applied = True
|
|
with pytest.raises(PlaneRefused, match="unmanaged live drift"):
|
|
preflight(bundle, cluster)
|
|
|
|
|
|
def test_rollback_is_generated_but_never_executed(tmp_path: Path) -> None:
|
|
bundle = PlaneBundle.load(_fixture(tmp_path))
|
|
result = rollback_plan(bundle)
|
|
assert result["object_scoped_commands"] == []
|
|
assert result["conditional_namespace_commands"] == ["kubectl delete namespaces whitehat"]
|