Add the silent MAS publisher gate for the prefix policy and CAS=0 pair.

The attended platform-admin login timed out before the gate ran, so the live policy, workload pair, and delivery admission are unchanged.

Assistant: grok
Assistant-Session: 01a198fa-fd27-7e41-a0a2-57355c3e4efd

openbao/policies/backup-pair-writer.hcl

@@ -1,7 +1,8 @@
-# Attended writer for the two application-archive publisher pairs.
-# Founder admission 2026-10-08. CAS=0 and one-write behavior are enforced by
-# scripts/application_backup_pair_custody.py. This policy cannot delete,
-# list, or read secret values.
+# Attended writer for the application-archive publisher pairs.
+# Founder admission 2026-10-08 for coulomb-social and email-connect.
+# The MAS path is included for the one CAS=0 pair write. CAS=0 and one-write
+# behavior are enforced by scripts/application_backup_pair_custody.py.
+# This policy cannot delete, list, or read secret values.
 
 path "platform/data/workloads/coulomb-social/backup-primary" {
   capabilities = ["create", "update"]
@@ -11,6 +12,10 @@ path "platform/data/workloads/email-connect/backup-primary" {
   capabilities = ["create", "update"]
 }
 
+path "platform/data/workloads/mas/backup-primary" {
+  capabilities = ["create", "update"]
+}
+
 path "platform/metadata/workloads/coulomb-social/backup-primary" {
   capabilities = ["read"]
 }
@@ -19,6 +24,10 @@ path "platform/metadata/workloads/email-connect/backup-primary" {
   capabilities = ["read"]
 }
 
+path "platform/metadata/workloads/mas/backup-primary" {
+  capabilities = ["read"]
+}
+
 path "auth/token/lookup-self" {
   capabilities = ["read"]
 }

scripts/application_backup_pair_custody.py

@@ -24,6 +24,7 @@ WRITER_ADMITTED = True
 PROFILES = {
     "coulomb-social": "CCR-2026-0036",
     "email-connect": "CCR-2026-0037",
+    "mas": "CCR-2026-0062",
 }
 VALUE = re.compile(r"[!-~]{16,128}\Z")
 

scripts/mas_evidence_publisher_gate.py

@@ -0,0 +1,395 @@
+#!/usr/bin/env python3
+"""Apply the MAS publisher boundary and write its one CAS=0 pair.
+
+Stdout stays empty. Receipts record statement ids, HTTP statuses, versions,
+and field names. Values are never written to a receipt or to stderr.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+ROOT = Path("/home/worsch/railiance-platform")
+DESIRED = ROOT / "docs/object-storage/railiance-platform-pg-backup-policy.json"
+POLICY_FILE = ROOT / "openbao/policies/backup-pair-writer.hcl"
+RECEIPT_DIR = Path("/home/worsch/.local/state/railiance/receipts")
+BUCKET = "railiance-platform-pg-backup"
+ENDPOINT = "https://s3.nl-ams.scw.cloud"
+APPLICATION_ID = "25a4cbca-220e-40c0-9af5-6c7713000174"
+PREFIX = "platform-pg/application-archives/mas/"
+SIBLING = "platform-pg/application-archives/email-connect/"
+MAS_SIDS = (
+    "MasEvidenceBackupProbe",
+    "MasEvidenceBackupList",
+    "MasEvidenceBackupObjects",
+)
+SUPPLIER = "platform/suppliers/scaleway/mas-evidence-backup-primary"
+WORKLOAD = "platform/workloads/mas/backup-primary"
+PREFERRED_PROJECT = "mail-address-semantics"
+FIELD_ALIASES = (
+    ("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"),
+    ("ACCESS_KEY", "SECRET_KEY"),
+    ("access_key", "secret_key"),
+    ("SCW_ACCESS_KEY", "SCW_SECRET_KEY"),
+)
+
+
+class GateError(RuntimeError):
+    def __init__(self, code):
+        super().__init__(code)
+        self.code = code
+
+
+def supplier_pair(data):
+    """Return the two workload fields. Refuse a partial or identical pair."""
+    if not isinstance(data, dict):
+        raise GateError("supplier_shape_rejected")
+    for access_name, secret_name in FIELD_ALIASES:
+        access = data.get(access_name)
+        secret = data.get(secret_name)
+        if access is None and secret is None:
+            continue
+        if not isinstance(access, str) or not isinstance(secret, str):
+            raise GateError("supplier_shape_rejected")
+        access = access.strip()
+        secret = secret.strip()
+        if access == secret or not access or not secret:
+            raise GateError("supplier_shape_rejected")
+        return {"AWS_ACCESS_KEY_ID": access, "AWS_SECRET_ACCESS_KEY": secret}
+    raise GateError("supplier_shape_rejected")
+
+
+def mas_statements(document):
+    found = []
+    for item in document.get("Statement", []):
+        if item.get("Sid") in MAS_SIDS:
+            found.append(item)
+    return found
+
+
+def policy_observation(document):
+    """Non-secret view of the MAS statements and whether object delete is absent."""
+    rows = mas_statements(document)
+    actions = []
+    for item in rows:
+        value = item.get("Action") or []
+        if isinstance(value, str):
+            value = [value]
+        actions.extend(value)
+    return {
+        "sids": [item.get("Sid") for item in rows],
+        "mas_sids_present": [sid for sid in MAS_SIDS if sid in {item.get("Sid") for item in rows}],
+        "object_delete_absent": not any(
+            action in {"s3:DeleteObject", "s3:DeleteObjectVersion"} for action in actions
+        ),
+        "principal": APPLICATION_ID,
+    }
+
+
+def denial_observation(results):
+    required = ("sibling_prefix", "retained_delete", "administration")
+    denied = {name: bool(results.get(name, {}).get("denied")) for name in required}
+    return {
+        "application_id": APPLICATION_ID,
+        "denied": denied,
+        "http_status": {
+            name: results.get(name, {}).get("http_status") for name in (*required, "own_prefix")
+        },
+        "passed": all(denied.values()),
+    }
+
+
+def version_from_put(payload):
+    data = payload.get("data") or {}
+    version = data.get("version")
+    if version != 1:
+        raise GateError("unexpected_version")
+    return {"version": 1, "request_id": payload.get("request_id")}
+
+
+def logical_path(path):
+    for marker in ("/metadata/", "/data/"):
+        if path.startswith("platform" + marker):
+            return "platform/" + path.split(marker, 1)[1]
+    return path
+
+
+def _write_json(path, value):
+    path.parent.mkdir(mode=0o700, exist_ok=True)
+    path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    path.chmod(0o600)
+
+
+def _bao(args, token=None):
+    env = os.environ.copy()
+    env.setdefault("BAO_ADDR", "http://127.0.0.1:18200")
+    env.setdefault("VAULT_ADDR", env["BAO_ADDR"])
+    if token is None:
+        pass
+    else:
+        env["BAO_TOKEN"] = token
+        env["VAULT_TOKEN"] = token
+    return subprocess.run(["bao", *args], capture_output=True, env=env)
+
+
+def _kv_data(path):
+    proc = _bao(["kv", "get", "-format=json", path])
+    if proc.returncode:
+        raise GateError("kv_read_failed")
+    body = json.loads(proc.stdout)
+    data = ((body.get("data") or {}).get("data")) or {}
+    if not isinstance(data, dict):
+        raise GateError("kv_read_failed")
+    return data
+
+
+def _client(pair):
+    import boto3
+    from botocore.config import Config
+
+    return boto3.client(
+        "s3",
+        region_name="nl-ams",
+        endpoint_url=ENDPOINT,
+        aws_access_key_id=pair["AWS_ACCESS_KEY_ID"],
+        aws_secret_access_key=pair["AWS_SECRET_ACCESS_KEY"],
+        config=Config(
+            signature_version="s3v4",
+            connect_timeout=15,
+            read_timeout=30,
+            retries={"max_attempts": 2},
+            s3={"addressing_style": "path"},
+            request_checksum_calculation="when_required",
+            response_checksum_validation="when_required",
+        ),
+    )
+
+
+def _status(call):
+    from botocore.exceptions import ClientError
+
+    try:
+        result = call()
+        body = result.get("Body") if isinstance(result, dict) else None
+        if body is not None:
+            body.close()
+        code = result.get("ResponseMetadata", {}).get("HTTPStatusCode", 200)
+        return {"http_status": code, "denied": False}
+    except ClientError as exc:
+        code = exc.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
+        return {"http_status": code, "denied": code == 403}
+
+
+def _ensure_boto():
+    try:
+        import boto3  # noqa: F401
+    except ImportError:
+        candidate = Path("/tmp/mas-bucket-policy-venv/bin/python")
+        if candidate.exists() and Path(sys.executable) != candidate:
+            os.execv(candidate, [str(candidate), *sys.argv])
+        raise GateError("boto3_required") from None
+
+
+def apply_live_policy(reader_pair, writer_pair):
+    desired = json.loads(DESIRED.read_text(encoding="utf-8"))
+    reader = _client(reader_pair)
+    live = json.loads(reader.get_bucket_policy(Bucket=BUCKET)["Policy"])
+    present = {item.get("Sid") for item in live.get("Statement", [])}
+    statements = list(live.get("Statement", []))
+    for item in desired.get("Statement", []):
+        if item.get("Sid") not in present:
+            statements.append(item)
+            present.add(item.get("Sid"))
+    policy = {
+        "Version": live.get("Version") or desired.get("Version"),
+        "Id": live.get("Id") or desired.get("Id"),
+        "Statement": statements,
+    }
+    receipt = {"before": [item.get("Sid") for item in live.get("Statement", [])]}
+    if receipt["before"] == [item.get("Sid") for item in statements]:
+        receipt["put"] = "unchanged"
+    else:
+        _client(writer_pair).put_bucket_policy(Bucket=BUCKET, Policy=json.dumps(policy))
+        receipt["put"] = "ok"
+    after = json.loads(reader.get_bucket_policy(Bucket=BUCKET)["Policy"])
+    receipt["after"] = [item.get("Sid") for item in after.get("Statement", [])]
+    receipt["mas"] = policy_observation(after)
+    _write_json(RECEIPT_DIR / "backup-bucket-policy-apply.json", receipt)
+    if not receipt["mas"]["object_delete_absent"] or receipt["mas"]["mas_sids_present"] != list(MAS_SIDS):
+        raise GateError("mas_policy_not_installed")
+    return receipt
+
+
+def prove_denials(pair):
+    client = _client(pair)
+    results = {
+        "sibling_prefix": _status(
+            lambda: client.list_objects_v2(Bucket=BUCKET, Prefix=SIBLING, MaxKeys=1)
+        ),
+        "retained_delete": _status(
+            lambda: client.delete_object(Bucket=BUCKET, Key=PREFIX + "retained-delete-probe")
+        ),
+        "administration": _status(lambda: client.get_bucket_policy(Bucket=BUCKET)),
+        "own_prefix": _status(
+            lambda: client.list_objects_v2(Bucket=BUCKET, Prefix=PREFIX, MaxKeys=1)
+        ),
+    }
+    observation = denial_observation(results)
+    observation["preferred_project"] = PREFERRED_PROJECT
+    _write_json(RECEIPT_DIR / "mas-evidence-publisher-denials.json", observation)
+    if not observation["passed"]:
+        raise GateError("publisher_denials_failed")
+    return observation
+
+
+def _writer_identity(auth):
+    """Use the create response so the two token uses stay available for put and revoke."""
+    policies = list(auth.get("policies") or [])
+    ttl = auth.get("lease_duration")
+    uses = auth.get("num_uses")
+    if "default" in policies or policies != ["backup-pair-writer"]:
+        raise GateError("writer_identity_rejected")
+    if not isinstance(ttl, int) or not isinstance(uses, int):
+        raise GateError("writer_identity_rejected")
+    return {
+        "policies": policies,
+        "identity_policies": list(auth.get("identity_policies") or []),
+        "ttl": ttl,
+        "num_uses": uses,
+    }
+
+
+class _Transport:
+    def __init__(self, token, absent):
+        self.token = token
+        self.absent = absent
+
+    def metadata(self, path):
+        if logical_path(path) != WORKLOAD or not self.absent:
+            raise GateError("pair_metadata_rejected")
+        return None
+
+    def put_cas0(self, path, body):
+        handle = tempfile.NamedTemporaryFile(
+            "w", prefix="mas-pair-", dir=RECEIPT_DIR, delete=False
+        )
+        name = Path(handle.name)
+        try:
+            json.dump(body["data"], handle)
+            handle.flush()
+            os.fchmod(handle.fileno(), 0o600)
+            handle.close()
+            proc = _bao(
+                ["kv", "put", "-format=json", "-cas=0", logical_path(path), f"@{name}"],
+                token=self.token,
+            )
+        finally:
+            name.unlink(missing_ok=True)
+        if proc.returncode:
+            raise GateError("pair_write_failed")
+        return version_from_put(json.loads(proc.stdout))
+
+    def revoke(self):
+        _bao(["token", "revoke", "-self"], token=self.token)
+
+
+def _custody():
+    import importlib.util
+
+    spec = importlib.util.spec_from_file_location(
+        "application_backup_pair_custody", ROOT / "scripts/application_backup_pair_custody.py"
+    )
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def write_pair(pair):
+    custody = _custody()
+
+    existing = _bao(["kv", "metadata", "get", "-format=json", WORKLOAD])
+    if existing.returncode == 0:
+        raise GateError("existing_custody_requires_rotation_review")
+    written = _bao(["policy", "write", "backup-pair-writer", str(POLICY_FILE)])
+    if written.returncode:
+        raise GateError("writer_policy_install_failed")
+    created = _bao(
+        [
+            "token",
+            "create",
+            "-policy=backup-pair-writer",
+            "-ttl=5m",
+            "-use-limit=2",
+            "-no-default-policy",
+            "-orphan",
+            "-format=json",
+        ]
+    )
+    if created.returncode:
+        raise GateError("writer_token_failed")
+    auth = json.loads(created.stdout)["auth"]
+    token = auth["client_token"]
+    try:
+        identity = _writer_identity(auth)
+        receipt = custody.provision("mas", pair, identity, _Transport(token, absent=True))
+    except Exception:
+        _bao(["token", "revoke", "-self"], token=token)
+        raise
+    meta = _bao(["kv", "metadata", "get", "-format=json", WORKLOAD])
+    if meta.returncode:
+        raise GateError("pair_metadata_unreadable")
+    data = json.loads(meta.stdout).get("data") or {}
+    names = sorted(_kv_data(WORKLOAD))
+    summary = {
+        "path": WORKLOAD,
+        "current_version": data.get("current_version"),
+        "field_names": names,
+        "cas": 0,
+        "kv_version": receipt["kv_version"],
+        "values_recorded": False,
+    }
+    if summary["current_version"] != 1 or names != ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"]:
+        raise GateError("pair_metadata_rejected")
+    _write_json(RECEIPT_DIR / "mas-evidence-pair.json", summary)
+    return summary
+
+
+def main():
+    sys.stdout = open(os.devnull, "w")
+    sys.stderr = open(os.devnull, "w")
+    try:
+        _ensure_boto()
+        RECEIPT_DIR.mkdir(mode=0o700, exist_ok=True)
+        supplier = supplier_pair(_kv_data(SUPPLIER))
+        backup = _kv_data("platform/workloads/railiance/backup/object-storage")
+        bootstrap = _kv_data("platform/workloads/railiance/scaleway/bootstrap")
+        reader = supplier_pair(backup)
+        writer = supplier_pair(bootstrap)
+        apply_live_policy(reader, writer)
+        prove_denials(supplier)
+        write_pair(supplier)
+    except GateError as exc:
+        _write_json(RECEIPT_DIR / "mas-evidence-publisher-gate.json", {"status": "failed", "code": exc.code})
+        return 71
+    except Exception:
+        _write_json(
+            RECEIPT_DIR / "mas-evidence-publisher-gate.json",
+            {"status": "failed", "code": "contained_gate_failure"},
+        )
+        return 72
+    _write_json(RECEIPT_DIR / "mas-evidence-publisher-gate.json", {"status": "passed", "values_recorded": False})
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except SystemExit:
+        raise
+    except Exception:
+        raise SystemExit(72) from None

tests/test_mas_evidence_publisher_gate.py

@@ -0,0 +1,94 @@
+"""MAS publisher gate records boundaries and refuses to keep secret values."""
+
+import importlib.util
+import json
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+spec = importlib.util.spec_from_file_location(
+    "mas_gate", ROOT / "scripts/mas_evidence_publisher_gate.py"
+)
+gate = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(gate)
+
+
+def test_supplier_pair_maps_aliases_and_refuses_partial_or_equal_values():
+    mapped = gate.supplier_pair({"ACCESS_KEY": "access-one", "SECRET_KEY": "secret-two"})
+    assert list(mapped) == ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"]
+    assert mapped["AWS_ACCESS_KEY_ID"] != mapped["AWS_SECRET_ACCESS_KEY"]
+    with pytest.raises(gate.GateError):
+        gate.supplier_pair({"ACCESS_KEY": "access-one"})
+    with pytest.raises(gate.GateError):
+        gate.supplier_pair({"ACCESS_KEY": "same-value", "SECRET_KEY": "same-value"})
+
+
+def test_policy_observation_requires_the_three_prefix_statements_without_delete():
+    document = {
+        "Statement": [
+            {
+                "Sid": "MasEvidenceBackupObjects",
+                "Action": ["s3:GetObject", "s3:PutObject"],
+            },
+            {"Sid": "MasEvidenceBackupProbe", "Action": ["s3:GetBucketLocation"]},
+            {"Sid": "MasEvidenceBackupList", "Action": ["s3:ListBucket"]},
+            {"Sid": "Other", "Action": ["s3:DeleteObject"]},
+        ]
+    }
+    observed = gate.policy_observation(document)
+    assert observed["mas_sids_present"] == list(gate.MAS_SIDS)
+    assert observed["object_delete_absent"] is True
+    assert observed["principal"] == gate.APPLICATION_ID
+    document["Statement"][0]["Action"].append("s3:DeleteObject")
+    assert gate.policy_observation(document)["object_delete_absent"] is False
+
+
+def test_denial_observation_passes_only_when_all_three_boundaries_deny():
+    results = {
+        "sibling_prefix": {"http_status": 403, "denied": True},
+        "retained_delete": {"http_status": 403, "denied": True},
+        "administration": {"http_status": 403, "denied": True},
+        "own_prefix": {"http_status": 200, "denied": False},
+    }
+    passed = gate.denial_observation(results)
+    assert passed["passed"] is True
+    assert passed["application_id"] == "25a4cbca-220e-40c0-9af5-6c7713000174"
+    results["retained_delete"]["denied"] = False
+    assert gate.denial_observation(results)["passed"] is False
+    rendered = json.dumps(passed)
+    assert "access-one" not in rendered
+    assert "secret-two" not in rendered
+
+
+def test_put_parser_keeps_version_and_drops_any_nested_value():
+    parsed = gate.version_from_put(
+        {"request_id": "req-1", "data": {"version": 1, "data": {"AWS_SECRET_ACCESS_KEY": "not-a-live-value"}}}
+    )
+    assert parsed == {"version": 1, "request_id": "req-1"}
+    assert "not-a-live-value" not in json.dumps(parsed)
+    with pytest.raises(gate.GateError):
+        gate.version_from_put({"data": {"version": 2}})
+
+
+def test_logical_path_strips_only_the_openbao_storage_marker():
+    assert gate.logical_path("platform/metadata/workloads/mas/backup-primary") == (
+        "platform/workloads/mas/backup-primary"
+    )
+    assert gate.logical_path("platform/data/workloads/mas/backup-primary") == (
+        "platform/workloads/mas/backup-primary"
+    )
+
+
+def test_pair_custody_preview_includes_mas_and_the_writer_policy_names_it():
+    spec_custody = importlib.util.spec_from_file_location(
+        "pair_custody_for_gate", ROOT / "scripts/application_backup_pair_custody.py"
+    )
+    custody = importlib.util.module_from_spec(spec_custody)
+    spec_custody.loader.exec_module(custody)
+    preview = custody.preview("mas")
+    assert preview["kv_path"] == "platform/workloads/mas/backup-primary"
+    assert preview["openbao_write"] is False
+    text = (ROOT / "openbao/policies/backup-pair-writer.hcl").read_text(encoding="utf-8")
+    assert 'path "platform/data/workloads/mas/backup-primary"' in text
+    assert "read" not in text.split('path "platform/data/workloads/mas/backup-primary"')[1].split("path ")[0]

workplans/RPF-WP-0070-mas-evidence-recovery.md

@@ -77,3 +77,5 @@ workload KV pair and `/etc/mas-backup/primary.env` do not exist.
 Pair-custody lists still
 name only coulomb-social and email-connect.
 Design: `docs/credential-lane-designs/mas-evidence-backup-grant-2026-11-02.md`.
+
+2026-11-02: `scripts/mas_evidence_publisher_gate.py` is the silent gate for the live prefix policy, the three denials, and the one CAS=0 workload pair. The pair-writer source now names the MAS path. An attended platform-admin login exited 10 with category `callback_timeout` before that gate started. The live policy, workload pair, delivery admission, node file, and offsite object are unchanged.