715b0064ce
The helper checks both draft archive CCRs and refuses a CAS write until the writer identity is admitted. No provider key is created. Assistant: grok Assistant-Session: 01a11cdb-f9fe-7011-947c-936e72007e54
143 lines
4.6 KiB
Python
143 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Preview custody for the two backup publisher pairs.
|
|
|
|
A later attended act may store both AWS fields in one CAS=0 write. This
|
|
command does not generate provider keys, contact OpenBao, or perform that
|
|
write. The delivery reader in application_backup_credentials.py cannot
|
|
create or rotate these values.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
FIELDS = ("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY")
|
|
WRITER_POLICY = "backup-pair-writer"
|
|
WRITER_ADMITTED = False
|
|
PROFILES = {
|
|
"coulomb-social": "CCR-2026-0036",
|
|
"email-connect": "CCR-2026-0037",
|
|
}
|
|
VALUE = re.compile(r"[!-~]{16,128}\Z")
|
|
|
|
|
|
class CustodyError(Exception):
|
|
"""Fixed diagnostic. It never carries key material."""
|
|
|
|
|
|
def guard(condition, code):
|
|
if not condition:
|
|
raise CustodyError(code)
|
|
|
|
|
|
def kv_path(application):
|
|
guard(application in PROFILES, "unknown_application")
|
|
return f"platform/workloads/{application}/backup-primary"
|
|
|
|
|
|
def ccr_path(application):
|
|
prefix = PROFILES[application]
|
|
return ROOT / "credential-change-requests" / f"{prefix}-{application}-backup-primary.yaml"
|
|
|
|
|
|
def preview(application):
|
|
raw = ccr_path(application).read_text(encoding="utf-8")
|
|
document = json.loads(raw)
|
|
guard(document.get("id") == PROFILES[application], "ccr_mismatch")
|
|
guard(document.get("openbao", {}).get("fields") == list(FIELDS), "field_mismatch")
|
|
guard(document.get("openbao", {}).get("kv_path") == kv_path(application), "path_mismatch")
|
|
return {
|
|
"status": "preview",
|
|
"application": application,
|
|
"ccr": document["id"],
|
|
"ccr_status": document.get("status"),
|
|
"kv_path": kv_path(application),
|
|
"fields": list(FIELDS),
|
|
"cas": 0,
|
|
"writes": 1,
|
|
"writer_policy": WRITER_POLICY,
|
|
"writer_admitted": WRITER_ADMITTED,
|
|
"provider_keys": "not_created",
|
|
"openbao_write": False,
|
|
"effects": [],
|
|
}
|
|
|
|
|
|
def validate_pair(values):
|
|
guard(isinstance(values, dict) and set(values) == set(FIELDS), "exact_pair_required")
|
|
stored = []
|
|
for name in FIELDS:
|
|
value = values[name]
|
|
guard(isinstance(value, str) and VALUE.fullmatch(value), "invalid_pair_value")
|
|
stored.append(value)
|
|
guard(stored[0] != stored[1], "pair_values_must_differ")
|
|
return {name: values[name] for name in FIELDS}
|
|
|
|
|
|
def assert_writer(identity, application):
|
|
"""Refuse the delivery reader, a generic grant, and any unreviewed writer."""
|
|
guard(WRITER_ADMITTED, "writer_identity_unreviewed")
|
|
policies = identity.get("policies")
|
|
reader = f"workload-kv-read-{application}-backup-primary"
|
|
guard(
|
|
policies == [WRITER_POLICY]
|
|
and identity.get("identity_policies") == []
|
|
and reader not in policies
|
|
and "default" not in policies,
|
|
"writer_identity_unreviewed",
|
|
)
|
|
ttl = identity.get("ttl")
|
|
uses = identity.get("num_uses")
|
|
guard(type(ttl) is int and 0 < ttl <= 300, "writer_ttl_unbounded")
|
|
guard(type(uses) is int and 0 < uses <= 2, "writer_uses_unbounded")
|
|
|
|
|
|
def provision(application, values, identity, transport):
|
|
"""One metadata check and one CAS=0 write. Tests inject transport.
|
|
|
|
Live use stays disconnected while WRITER_ADMITTED is false.
|
|
"""
|
|
path = kv_path(application)
|
|
pair = validate_pair(values)
|
|
assert_writer(identity, application)
|
|
metadata = transport.metadata(f"platform/metadata/workloads/{application}/backup-primary")
|
|
guard(metadata is None, "existing_custody_requires_rotation_review")
|
|
written = transport.put_cas0(
|
|
f"platform/data/workloads/{application}/backup-primary",
|
|
{"options": {"cas": 0}, "data": pair},
|
|
)
|
|
try:
|
|
guard(written.get("version") == 1 and set(written) <= {"version", "request_id"}, "unexpected_version")
|
|
return {
|
|
"application": application,
|
|
"kv_path": path,
|
|
"cas": 0,
|
|
"kv_version": 1,
|
|
"field_names": list(FIELDS),
|
|
"request_id": written.get("request_id"),
|
|
"secret_material": False,
|
|
}
|
|
finally:
|
|
transport.revoke()
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--application", required=True, choices=sorted(PROFILES))
|
|
args = parser.parse_args(argv)
|
|
json.dump(preview(args.application), sys.stdout, indent=2)
|
|
sys.stdout.write("\n")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except CustodyError as exc:
|
|
print(str(exc), file=sys.stderr)
|
|
raise SystemExit(2) from None
|