MASON-WP-0015: accept a can-i denial as negative-scope proof

kubectl auth can-i exits 1 when the answer is no. verify() was sending that
query through _run, which raised PlaneError and stopped apply under the
scoped builder. A denial with stdout no is now the negative proof. Any other
non-zero exit still fails.

Assistant: grok
Assistant-Session: 01a16071-5425-78d3-80b4-20fe2ee30f0a
This commit is contained in:
tegwick 2026-10-22 09:41:16 +02:00
parent c8b3e062cc
commit 827781f318
2 changed files with 34 additions and 2 deletions

View file

@ -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

View file

@ -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)