diff --git a/src/ops_mason/kubernetes_plane.py b/src/ops_mason/kubernetes_plane.py index 7e8a2f3..e9a202d 100644 --- a/src/ops_mason/kubernetes_plane.py +++ b/src/ops_mason/kubernetes_plane.py @@ -311,10 +311,20 @@ def _find_keys(value: Any, forbidden: set[str], prefix: str = "") -> list[str]: return found -def _run(runner: Runner, args: Sequence[str], *, allow_not_found: bool = False) -> CommandResult: +def _run( + runner: Runner, + args: Sequence[str], + *, + allow_not_found: bool = False, + allow_capability_denial: bool = False, +) -> CommandResult: result = runner(args) if result.returncode == 0: return result + # kubectl auth can-i exits 1 and prints "no" for a denial. That is the + # answer, not a broken command. Any other non-zero exit still fails. + if allow_capability_denial and result.stdout.strip().lower() == "no": + return result if allow_not_found and "notfound" in result.stderr.lower().replace(" ", ""): return result command = " ".join(args) @@ -510,6 +520,7 @@ def verify(bundle: PlaneBundle, runner: Runner = subprocess_runner) -> dict[str, "kubectl", "auth", "can-i", "get", resource, "-n", bundle.expected_namespace, ], + allow_capability_denial=True, ).stdout.strip().lower() if allowed != "yes": negative_scope[resource] = 0 diff --git a/tests/test_kubernetes_plane.py b/tests/test_kubernetes_plane.py index b49e3a2..cd90447 100644 --- a/tests/test_kubernetes_plane.py +++ b/tests/test_kubernetes_plane.py @@ -11,6 +11,7 @@ from ops_mason.readiness import inspect_readiness, resolve_tier from ops_mason.kubernetes_plane import ( CommandResult, PlaneBundle, + PlaneError, PlaneRefused, _json_path, apply, @@ -173,7 +174,7 @@ class FakeCluster: return CommandResult(0, self.context + "\n") if command[:3] == ["kubectl", "auth", "can-i"]: if command[3] == "get" and not self.can_get_negative: - return CommandResult(0, "no\n") + return CommandResult(1, "no\n") return CommandResult(0, "yes\n") if "apply" in command: if "--dry-run=client" not in command and "--dry-run=server" not in command: @@ -294,6 +295,26 @@ def test_apply_records_negative_scope_when_get_is_denied(tmp_path: Path) -> None assert [call[4] for call in denied] == ["pods", "secrets"] +def test_apply_rejects_a_broken_can_i_instead_of_a_negative_scope(tmp_path: Path) -> None: + bundle = PlaneBundle.load(_fixture(tmp_path)) + cluster = FakeCluster() + + def runner(args): + command = list(args) + if command[:4] == ["kubectl", "auth", "can-i", "get"]: + cluster.calls.append(command) + return CommandResult(1, "", "the server rejected the capability query") + return cluster(args) + + with pytest.raises(PlaneError, match="failed"): + apply( + bundle, + confirm_plan_id="plane", + expected_digest=bundle.digest, + runner=runner, + ) + + def test_preflight_refuses_unmanaged_live_drift(tmp_path: Path) -> None: bundle = PlaneBundle.load(_fixture(tmp_path)) cluster = FakeCluster(drift=True)