d41c5ace0a
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 4541@bnt-lap001 Assistant-Session: 2099b3aa-d7d9-49a9-ab4e-fa8b7d0e39e8
95 lines
3.9 KiB
Python
95 lines
3.9 KiB
Python
"""Reader boundary probe: passes only on an exact reader and never records values."""
|
|
|
|
import base64
|
|
import importlib.util
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location("boundary", ROOT / "scripts/backup_reader_boundary_probe.py")
|
|
probe = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(probe)
|
|
|
|
SECRET = "supersecretvalue12345"
|
|
ACCESS = "SCWACCESSKEY12345678"
|
|
|
|
|
|
def _jwt(app, aud, uid="sa-uid"):
|
|
claims = {"sub": f"system:serviceaccount:{app}:{probe.SA}", "aud": [aud],
|
|
"exp": int(time.time()) + 600, "kubernetes.io": {"serviceaccount": {"uid": uid}}}
|
|
body = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip("=")
|
|
return f"h.{body}.s"
|
|
|
|
|
|
class FakeBao:
|
|
def __init__(self, app="mas", version=1, allow_write=False, extra_policy=False, allow_foreign_role=False):
|
|
self.app, self.version = app, version
|
|
self.allow_write, self.extra_policy, self.allow_foreign_role = allow_write, extra_policy, allow_foreign_role
|
|
self.tokens, self.writes = {}, []
|
|
|
|
def jwt(self, application, audience):
|
|
return _jwt(application, audience)
|
|
|
|
def http(self, method, path, payload=None, token=None):
|
|
if path == "auth/kubernetes/login":
|
|
claims = probe.jwt_claims(payload["jwt"])
|
|
ok_role = payload["role"] == probe.role(self.app) or self.allow_foreign_role
|
|
if claims["aud"] != ["openbao"] or not ok_role:
|
|
return 403, {}
|
|
tok = f"t{len(self.tokens)}"
|
|
policies = [probe.policy(self.app)] + (["default"] if self.extra_policy else [])
|
|
self.tokens[tok] = {"uses": 4, "policies": policies}
|
|
return 200, {"auth": {"client_token": tok, "policies": policies, "lease_duration": 300}}
|
|
state = self.tokens.get(token)
|
|
if not state or state["uses"] <= 0:
|
|
return 403, {}
|
|
state["uses"] -= 1
|
|
if path == "auth/token/revoke-self":
|
|
del self.tokens[token]
|
|
return 204, {}
|
|
if path == "auth/token/lookup-self":
|
|
return 200, {"data": {"policies": state["policies"], "identity_policies": None, "ttl": 299,
|
|
"explicit_max_ttl": 300, "period": 0, "num_uses": state["uses"]}}
|
|
if method == "GET" and path == probe.data_path(self.app):
|
|
return 200, {"data": {"data": {"AWS_ACCESS_KEY_ID": ACCESS, "AWS_SECRET_ACCESS_KEY": SECRET},
|
|
"metadata": {"version": self.version, "destroyed": False, "deletion_time": ""}}}
|
|
if method == "POST" and path == probe.data_path(self.app) and self.allow_write:
|
|
self.writes.append(payload)
|
|
return 400, {} # CAS mismatch: the path already exists
|
|
return 403, {}
|
|
|
|
|
|
def run(fake, version=1):
|
|
return probe.probe(fake, "mas", version, sa_uid="sa-uid")
|
|
|
|
|
|
def test_exact_reader_passes_and_records_no_values():
|
|
fake = FakeBao()
|
|
receipt = run(fake)
|
|
assert receipt["passed"] is True, receipt
|
|
assert receipt["positive"]["kv_version"] == 1
|
|
assert receipt["positive"]["field_names"] == probe.FIELDS
|
|
text = json.dumps(receipt)
|
|
assert SECRET not in text and ACCESS not in text and "t0" not in text
|
|
assert fake.tokens == {} # every issued token revoked
|
|
|
|
|
|
def test_version_mismatch_fails():
|
|
assert run(FakeBao(version=2))["passed"] is False
|
|
|
|
|
|
def test_extra_policy_fails():
|
|
assert run(FakeBao(extra_policy=True))["positive"]["passed"] is False
|
|
|
|
|
|
def test_write_not_denied_fails_and_cas0_payload_cannot_replace():
|
|
fake = FakeBao(allow_write=True)
|
|
receipt = run(fake)
|
|
assert receipt["negative"]["kv_write"] == 400 and receipt["passed"] is False
|
|
assert fake.writes == [{"options": {"cas": 0}, "data": {"probe": "x"}}]
|
|
|
|
|
|
def test_foreign_role_login_accepted_fails():
|
|
receipt = run(FakeBao(allow_foreign_role=True))
|
|
assert receipt["negative"]["foreign_role_login"] == 200 and receipt["passed"] is False
|