build: add guarded Kubernetes plane executor
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02878-7c21-7692-bcd6-ce2838c4b448
This commit is contained in:
parent
1dd98b4f27
commit
2b318634b6
13 changed files with 1240 additions and 3 deletions
258
tests/test_kubernetes_plane.py
Normal file
258
tests/test_kubernetes_plane.py
Normal file
|
|
@ -0,0 +1,258 @@
|
|||
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",
|
||||
]
|
||||
assert bundle.plan().is_approved()
|
||||
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 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"]
|
||||
Loading…
Add table
Add a link
Reference in a new issue