dad91e8c44
Round-10 review (railiance-platform) found that ContainedOperator inherited an ssh-only kubectl, so an E2 would write OpenBao and then fail its sync, restart and rollback. Also: a bad AUDIT_KUBECONFIG now leaves window-error evidence, and the receipt script validates bodies and lets a later request-changes withdraw an approval. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 236506@bnt-lap001 Assistant-Session: 09087018-91e1-48c5-ab2f-acfbb62c92e4
109 lines
4.7 KiB
Python
109 lines
4.7 KiB
Python
"""AUDIT_KUBECONFIG routes cluster reads and the port-forward locally.
|
|
|
|
Since RCLUSTER-WP-0019 the node account cannot read the k3s admin kubeconfig.
|
|
Attended runs name the operator's own kubeconfig for one command. Unset keeps
|
|
the earlier ssh path unchanged.
|
|
"""
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def load(name, rel):
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
spec = importlib.util.spec_from_file_location(name, ROOT / rel)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
PRE = load("pre_kc", "scripts/audit-core-recovery-preflight.py")
|
|
WIN = load("win_kc", "scripts/t02-attended-window.py")
|
|
|
|
|
|
class Done:
|
|
def __init__(self, rc=0, out="{}"):
|
|
self.returncode, self.stdout, self.stderr = rc, out, "secret-looking stderr"
|
|
|
|
|
|
def test_unset_keeps_ssh_path(monkeypatch):
|
|
monkeypatch.delenv("AUDIT_KUBECONFIG", raising=False)
|
|
remote = PRE.Remote("railiance01")
|
|
with patch.object(PRE.subprocess, "run", return_value=Done()) as run:
|
|
remote.kubectl(["get", "nodes"], label="read node state")
|
|
assert run.call_args.args[0][:3] == ["ssh", "-o", "BatchMode=yes"]
|
|
assert WIN.PortForward("railiance01", 18080).command()[0] == "ssh"
|
|
|
|
|
|
def test_set_runs_kubectl_locally_with_that_file(monkeypatch, tmp_path):
|
|
kc = tmp_path / "config"; kc.write_text("not read")
|
|
monkeypatch.setenv("AUDIT_KUBECONFIG", str(kc))
|
|
remote = PRE.Remote("railiance01")
|
|
with patch.object(PRE.subprocess, "run", return_value=Done()) as run:
|
|
remote.kubectl(["get", "nodes", "-o", "json"], label="read node state")
|
|
assert run.call_args.args[0] == ["kubectl", "--kubeconfig", str(kc), "get", "nodes", "-o", "json"]
|
|
# Host-level reads still use ssh to the node.
|
|
with patch.object(PRE.subprocess, "run", return_value=Done(out="12.0 1.0")) as run:
|
|
remote.run(["cat", "/proc/uptime"], label="read host uptime")
|
|
assert run.call_args.args[0][0] == "ssh"
|
|
command = WIN.PortForward("railiance01", 18080).command()
|
|
assert command[:5] == ["timeout", str(WIN.REMOTE_PF_TIMEOUT_SECONDS), "kubectl", "--kubeconfig", str(kc)]
|
|
assert "ssh" not in command and "127.0.0.1" in command
|
|
|
|
|
|
def test_failure_reports_only_the_step(monkeypatch, tmp_path):
|
|
kc = tmp_path / "config"; kc.write_text("")
|
|
monkeypatch.setenv("AUDIT_KUBECONFIG", str(kc))
|
|
with patch.object(PRE.subprocess, "run", return_value=Done(rc=1)):
|
|
with pytest.raises(PRE.PreflightError) as err:
|
|
PRE.Remote("railiance01").kubectl(["get", "nodes"], label="read node state")
|
|
assert str(err.value) == "read node state failed (exit 1)"
|
|
|
|
|
|
@pytest.mark.parametrize("value", ["relative/config", "/nonexistent/config"])
|
|
def test_bad_path_is_refused(monkeypatch, value):
|
|
monkeypatch.setenv("AUDIT_KUBECONFIG", value)
|
|
with pytest.raises(PRE.PreflightError):
|
|
PRE.Remote("railiance01")
|
|
|
|
|
|
LANE = load("lane_kc", "scripts/audit_core_synthetic_sender_lane.py")
|
|
|
|
|
|
def operator(monkeypatch):
|
|
monkeypatch.setattr(LANE.LEASE, "contained_bao_env", lambda: {"BAO_ADDR": "http://127.0.0.1:18200"})
|
|
return LANE.ContainedOperator("railiance01")
|
|
|
|
|
|
def test_lane_operator_writes_use_the_attended_kubeconfig(monkeypatch, tmp_path):
|
|
kc = tmp_path / "config"; kc.write_text("not read")
|
|
monkeypatch.setenv("AUDIT_KUBECONFIG", str(kc))
|
|
op = operator(monkeypatch)
|
|
with patch.object(LANE.CREDS.subprocess, "run", return_value=Done()) as run:
|
|
op.kubectl(["-n", "audit-core", "rollout", "restart", "deploy/audit-core"], label="restart audit-core")
|
|
assert run.call_args.args[0] == ["kubectl", "--kubeconfig", str(kc), "-n", "audit-core",
|
|
"rollout", "restart", "deploy/audit-core"]
|
|
|
|
|
|
def test_lane_operator_unset_keeps_ssh(monkeypatch):
|
|
monkeypatch.delenv("AUDIT_KUBECONFIG", raising=False)
|
|
op = operator(monkeypatch)
|
|
with patch.object(LANE.CREDS.subprocess, "run", return_value=Done()) as run:
|
|
op.kubectl(["get", "pods"], label="read pods")
|
|
assert run.call_args.args[0][:3] == ["ssh", "-o", "BatchMode=yes"]
|
|
|
|
|
|
def test_window_records_bad_kubeconfig_as_window_error(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("AUDIT_KUBECONFIG", "relative/config")
|
|
evidence = tmp_path / "evidence"; evidence.mkdir(mode=0o700)
|
|
monkeypatch.setattr(WIN, "window", lambda args: WIN.PortForward("railiance01", 18080))
|
|
code = WIN.main(["--approval", str(tmp_path / "a.json"), "--load-driver", str(tmp_path / "d"),
|
|
"--bearer-file", str(tmp_path / "b"), "--evidence-dir", str(evidence),
|
|
"--confirm", "x"])
|
|
assert code == 1
|
|
assert [p.name.startswith("window-error-") for p in evidence.iterdir()] == [True]
|