Align the MAS IAM rule and read num_uses from the raw writer response.

Adds the one-session Scaleway IAM grant script, merges the publisher
bucket probes into one statement to fit the 10-statement limit, keeps
the Playwright cache under the account home during attended login, and
records the publisher-gate runs through the num_uses rejection.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 49552@bnt-lap001
Assistant-Session: 575802ee-c0c0-42a9-b2a1-9204c9f5b3f4

scripts/mas_evidence_iam_grant.py

@@ -0,0 +1,322 @@
+#!/usr/bin/env python3
+"""Compare and correct the MAS publisher IAM rule.
+
+A temporary Scaleway Console session is accepted on stdin only with --execute,
+in the same shape the Coulomb identity bootstrap uses. Preview contacts nothing.
+Receipts record policy names, project names, and permission sets. Session
+tokens, API access keys, and secret keys are never written.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import pwd
+import sys
+from pathlib import Path
+from urllib.parse import urlencode
+
+ROOT = Path("/home/worsch/railiance-platform")
+COULOMB_APPLICATION = "681d00bf-b965-4323-b657-97bfde423511"
+MAS_APPLICATION = "25a4cbca-220e-40c0-9af5-6c7713000174"
+COULOMB_POLICY_NAME = "coulomb-social-backup-primary"
+POLICY_NAME = "mas-evidence-backup-primary"
+PERMISSIONS = (
+    "ObjectStorageBucketsRead",
+    "ObjectStorageObjectsRead",
+    "ObjectStorageObjectsWrite",
+    "ObjectStorageObjectsDelete",
+)
+_FORBIDDEN_RECEIPT_FIELDS = frozenset({
+    "access_key",
+    "secret_key",
+    "session_token",
+    "token",
+    "secret",
+})
+
+
+class Refused(RuntimeError):
+    pass
+
+
+def _recovery():
+    spec = importlib.util.spec_from_file_location(
+        "application_recovery_identities",
+        ROOT / "scripts/application_recovery_identities.py",
+    )
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def receipt_path():
+    home = Path(pwd.getpwuid(os.getuid()).pw_dir)
+    return home / ".local/state/railiance/receipts/mas-evidence-iam-grant.json"
+
+
+def _write_receipt(value):
+    path = receipt_path()
+    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 assert_public(value):
+    if isinstance(value, dict):
+        if _FORBIDDEN_RECEIPT_FIELDS & set(value):
+            raise Refused("receipt_field_refused")
+        for item in value.values():
+            assert_public(item)
+    elif isinstance(value, list):
+        for item in value:
+            assert_public(item)
+
+
+def inventory(api, path, key, **filters):
+    rows = []
+    for page in range(1, 101):
+        value = api("GET", path + "?" + urlencode({**filters, "page_size": 100, "page": page}))
+        batch = value[key]
+        if not isinstance(batch, list):
+            raise Refused("invalid_provider_inventory")
+        rows.extend(batch)
+        count = int(value["total_count"])
+        if len(rows) == count:
+            return rows
+        if not batch or len(rows) > count:
+            break
+    raise Refused("complete_inventory_required")
+
+
+def _rule_view(rule, names):
+    project_ids = [item for item in (rule.get("project_ids") or []) if isinstance(item, str)]
+    return {
+        "permission_set_names": sorted(
+            item for item in (rule.get("permission_set_names") or []) if isinstance(item, str)
+        ),
+        "project_ids": project_ids,
+        "project_names": [names.get(item, "") for item in project_ids],
+        "organization_scope": bool(rule.get("organization_id")),
+        "deny": bool(rule.get("deny")),
+        "condition_present": bool(rule.get("condition")),
+    }
+
+
+def _publisher_shape(rule):
+    return (
+        set(rule.get("permission_set_names") or []) == set(PERMISSIONS)
+        and not rule.get("organization_id")
+        and not rule.get("condition")
+        and not rule.get("deny")
+    )
+
+
+def _matches(rule, project_id):
+    return _publisher_shape(rule) and list(rule.get("project_ids") or []) == [project_id]
+
+
+def _bucket_project(policies, names):
+    """The live bucket grant is the Coulomb policy of that name, not the recovery project."""
+    chosen = [item for item in policies if item.get("name") == COULOMB_POLICY_NAME]
+    if len(chosen) != 1 or len(chosen[0]["_rules"]) != 1:
+        raise Refused("coulomb_backup_project_unresolved")
+    rule = chosen[0]["_rules"][0]
+    project_ids = [item for item in (rule.get("project_ids") or []) if isinstance(item, str)]
+    if not _publisher_shape(rule) or len(project_ids) != 1 or project_ids[0] not in names:
+        raise Refused("coulomb_backup_project_unresolved")
+    return project_ids[0], names[project_ids[0]]
+
+
+def _policies_for(api, application_id, organization_id, names):
+    found = []
+    for policy in inventory(
+        api, "/iam/v1alpha1/policies", "policies", organization_id=organization_id,
+        application_id=application_id,
+    ):
+        if policy.get("application_id") != application_id:
+            continue
+        rules = inventory(api, "/iam/v1alpha1/rules", "rules", policy_id=policy["id"])
+        found.append({
+            "id": policy.get("id"),
+            "name": policy.get("name"),
+            "editable": bool(policy.get("editable", True)),
+            "rules": [_rule_view(rule, names) for rule in rules],
+            "_rules": rules,
+        })
+    return found
+
+
+def _key_view(keys, project_id):
+    projects = sorted({
+        item.get("default_project_id") for item in keys
+        if isinstance(item.get("default_project_id"), str)
+    })
+    return {
+        "count": len(keys),
+        "default_project_ids": projects,
+        "preferred_project_matches": len(keys) == 1 and keys[0].get("default_project_id") == project_id,
+    }
+
+
+def _desired_rule(project_id):
+    return {"permission_set_names": list(PERMISSIONS), "project_ids": [project_id]}
+
+
+def align(api):
+    """Read both applications. Correct one direct MAS rule when that is unambiguous."""
+    coulomb = api("GET", "/iam/v1alpha1/applications/" + COULOMB_APPLICATION)
+    mas = api("GET", "/iam/v1alpha1/applications/" + MAS_APPLICATION)
+    if coulomb.get("name") != "coulomb-social-backup-primary":
+        raise Refused("coulomb_application_binding_refused")
+    if mas.get("name") != POLICY_NAME:
+        raise Refused("mas_application_binding_refused")
+    organization = coulomb.get("organization_id")
+    if not organization or organization != mas.get("organization_id"):
+        raise Refused("organization_binding_refused")
+    available = {
+        item.get("name") for item in inventory(
+            api, "/iam/v1alpha1/permission-sets", "permission_sets", organization_id=organization,
+        )
+    }
+    if any(name not in available for name in PERMISSIONS):
+        raise Refused("required_permission_sets_unavailable")
+    projects = inventory(api, "/account/v3/projects", "projects", organization_id=organization)
+    names = {item.get("id"): item.get("name") for item in projects if isinstance(item.get("id"), str)}
+    coulomb_policies = _policies_for(api, COULOMB_APPLICATION, organization, names)
+    project_id, project_name = _bucket_project(coulomb_policies, names)
+    mas_policies = _policies_for(api, MAS_APPLICATION, organization, names)
+    groups = [
+        item.get("name") for item in inventory(
+            api, "/iam/v1alpha1/groups", "groups", organization_id=organization,
+        )
+        if MAS_APPLICATION in (item.get("application_ids") or [])
+    ]
+    keys = [
+        item for item in inventory(
+            api, "/iam/v1alpha1/api-keys", "api_keys", organization_id=organization,
+            application_id=MAS_APPLICATION,
+        )
+        if item.get("application_id") == MAS_APPLICATION
+    ]
+    changed = False
+    preferred_updated = False
+    if len(mas_policies) > 1:
+        outcome = "multiple_policies_require_review"
+    elif len(mas_policies) == 1 and len(mas_policies[0]["_rules"]) > 1:
+        outcome = "multiple_rules_require_review"
+    elif len(mas_policies) == 1 and not mas_policies[0]["editable"]:
+        outcome = "policy_not_editable"
+    elif len(mas_policies) == 1 and all(_matches(rule, project_id) for rule in mas_policies[0]["_rules"]):
+        outcome = "already_matches"
+    elif len(mas_policies) == 1:
+        policy_id = mas_policies[0]["id"]
+        api("PUT", "/iam/v1alpha1/rules", {"policy_id": policy_id, "rules": [_desired_rule(project_id)]})
+        changed = True
+        outcome = "rule_replaced"
+    else:
+        api("POST", "/iam/v1alpha1/policies", {
+            "name": POLICY_NAME,
+            "description": "MAS evidence publisher; one project; bucket policy withholds object delete",
+            "organization_id": organization,
+            "application_id": MAS_APPLICATION,
+            "rules": [_desired_rule(project_id)],
+        })
+        changed = True
+        outcome = "policy_created"
+    if outcome in {"already_matches", "rule_replaced", "policy_created"} and len(keys) == 1:
+        if keys[0].get("default_project_id") != project_id:
+            access_key = keys[0].get("access_key")
+            if not isinstance(access_key, str) or not access_key:
+                raise Refused("api_key_binding_refused")
+            api("PATCH", "/iam/v1alpha1/api-keys/" + access_key, {"default_project_id": project_id})
+            preferred_updated = True
+    mas_policies = _policies_for(api, MAS_APPLICATION, organization, names)
+    keys = [
+        item for item in inventory(
+            api, "/iam/v1alpha1/api-keys", "api_keys", organization_id=organization,
+            application_id=MAS_APPLICATION,
+        )
+        if item.get("application_id") == MAS_APPLICATION
+    ]
+    if outcome in {"rule_replaced", "policy_created"}:
+        if len(mas_policies) != 1 or not all(_matches(rule, project_id) for rule in mas_policies[0]["_rules"]):
+            raise Refused("effective_rule_refused")
+    receipt = {
+        "status": "aligned" if outcome in {"already_matches", "rule_replaced", "policy_created"} else "review",
+        "outcome": outcome,
+        "changed": changed,
+        "project_name": project_name,
+        "project_source": COULOMB_POLICY_NAME,
+        "project_id": project_id,
+        "permission_sets": list(PERMISSIONS),
+        "coulomb_policies": [_public_policy(item) for item in coulomb_policies],
+        "mas_policies": [_public_policy(item) for item in mas_policies],
+        "mas_application_groups": groups,
+        "mas_keys": _key_view(keys, project_id),
+        "preferred_project_updated": preferred_updated,
+        "credential_values_printed": False,
+    }
+    assert_public(receipt)
+    return receipt
+
+
+def _public_policy(policy):
+    return {"id": policy["id"], "name": policy["name"], "editable": policy["editable"], "rules": policy["rules"]}
+
+
+def preview():
+    return {
+        "status": "preview",
+        "coulomb_application": COULOMB_APPLICATION,
+        "mas_application": MAS_APPLICATION,
+        "project_source": COULOMB_POLICY_NAME,
+        "permission_sets": list(PERMISSIONS),
+        "provider_contacted": False,
+        "credential_values_printed": False,
+    }
+
+
+def main(argv=None):
+    args = list(sys.argv[1:] if argv is None else argv)
+    if args not in ([], ["--execute"]):
+        raise Refused("usage_refused")
+    if not args:
+        print(json.dumps(preview(), sort_keys=True))
+        return 0
+    raw = sys.stdin.read(65537)
+    if len(raw) > 65536:
+        raise Refused("bounded_temporary_auth_required")
+    recovery = _recovery()
+    try:
+        auth = json.loads(raw)
+    except json.JSONDecodeError:
+        raise Refused("temporary_stdin_auth_required") from None
+    try:
+        receipt = align(recovery.Transport(auth))
+    except recovery.Refused as error:
+        raise Refused(str(error)) from None
+    except recovery.ProviderError as error:
+        print(json.dumps({
+            "status": "failed", "http_status": error.status, "credential_values_printed": False,
+        }, sort_keys=True))
+        return 1
+    _write_receipt(receipt)
+    print(json.dumps(receipt, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except Refused as error:
+        print(json.dumps({
+            "status": "failed", "refusal": str(error), "credential_values_printed": False,
+        }, sort_keys=True))
+        raise SystemExit(1) from None
+    except (OSError, KeyError, ValueError, TypeError):
+        print(json.dumps({
+            "status": "failed", "refusal": "contained_grant_failure", "credential_values_printed": False,
+        }, sort_keys=True))
+        raise SystemExit(1) from None

scripts/mas_evidence_iam_session.mjs

@@ -0,0 +1,80 @@
+#!/usr/bin/env node
+/** One attended Console login. The helper compares and corrects the MAS IAM rule. */
+import {spawn} from 'node:child_process';
+import {readFileSync} from 'node:fs';
+import {dirname, join, resolve} from 'node:path';
+import {fileURLToPath, pathToFileURL} from 'node:url';
+import {parseArgs} from 'node:util';
+import {createInterface} from 'node:readline/promises';
+
+const {values} = parseArgs({options: {'playwright-module': {type: 'string'}}});
+const root = resolve(dirname(fileURLToPath(import.meta.url)), '..');
+const grant = join(root, 'scripts', 'mas_evidence_iam_grant.py');
+let browser;
+let child;
+
+async function sessionAuth(page) {
+  if (new URL(page.url()).origin !== 'https://console.scaleway.com') {
+    throw new Error('console_page_required');
+  }
+  return page.evaluate(() => {
+    const names = Object.keys(localStorage).filter((name) => name.startsWith('_scw_session_console_scaleway_com_'));
+    if (names.length !== 1) throw new Error('unambiguous_session_required');
+    const session = JSON.parse(localStorage.getItem(names[0]));
+    if (new Date(session.jwt.expiresAt).getTime() < Date.now() + 30000) {
+      throw new Error('fresh_console_session_required');
+    }
+    return {session_token: session.token, user_agent: session.jwt.userAgent};
+  });
+}
+
+function runGrant(auth) {
+  return new Promise((resolveResult, reject) => {
+    child = spawn('python3', [grant, '--execute'], {stdio: ['pipe', 'pipe', 'pipe']});
+    let output = '';
+    child.stdout.on('data', (data) => {
+      if (output.length < 65536) output += data.toString();
+    });
+    child.stderr.on('data', () => {});
+    child.on('error', () => reject(new Error('grant_child_failed')));
+    child.on('close', (code) => {
+      child = null;
+      resolveResult({exit_code: code, output});
+    });
+    child.stdin.end(JSON.stringify(auth));
+  });
+}
+
+try {
+  const modulePath = resolve(values['playwright-module'] || '');
+  const pkg = JSON.parse(readFileSync(join(dirname(modulePath), 'package.json'), 'utf8'));
+  if (pkg.name !== 'playwright' || pkg.version !== '1.62.1') throw new Error('locked_playwright_required');
+  const imported = await import(pathToFileURL(modulePath).href);
+  const playwright = imported.chromium ? imported : imported.default;
+  if (!playwright?.chromium?.launch) throw new Error('playwright_export_unusable');
+  const {chromium} = playwright;
+  browser = await chromium.launch({headless: false});
+  const context = await browser.newContext();
+  const page = await context.newPage();
+  await page.goto('https://console.scaleway.com/', {waitUntil: 'domcontentloaded', timeout: 60000});
+  const reader = createInterface({input: process.stdin, output: process.stdout});
+  console.log('Console opened. Sign in, then press Enter. The session token stays in this process.');
+  await reader.question('');
+  reader.close();
+  const outcome = await runGrant(await sessionAuth(page));
+  process.stdout.write(outcome.output.endsWith('\n') ? outcome.output : `${outcome.output}\n`);
+  process.exitCode = outcome.exit_code === 0 ? 0 : 1;
+  try {
+    await page.goto('https://account.scaleway.com/logout', {waitUntil: 'domcontentloaded', timeout: 30000});
+  } catch {}
+  await context.clearCookies();
+} catch (error) {
+  const code = typeof error?.message === 'string' && /^[a-z0-9_]+$/.test(error.message)
+    ? error.message
+    : (error?.name || 'Error');
+  console.error(`IAM grant session failed: ${code}; no credential values printed.`);
+  process.exitCode = 1;
+} finally {
+  if (child) child.kill('SIGTERM');
+  if (browser) await browser.close();
+}

scripts/mas_evidence_publisher_gate.py

@@ -9,9 +9,12 @@ from __future__ import annotations
 
 import json
 import os
+import pwd
+import re
 import subprocess
 import sys
 import tempfile
+import urllib.request
 from pathlib import Path
 
 ROOT = Path("/home/worsch/railiance-platform")
@@ -24,44 +27,111 @@ 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",
 )
+PROBE_SID = "PublisherBucketProbe"
+PROBE_ACTIONS = (
+    "s3:GetBucketLocation",
+    "s3:GetBucketVersioning",
+    "s3:GetLifecycleConfiguration",
+)
+REPLACED_PROBES = (
+    "CoulombSocialBackupProbe",
+    "EmailConnectBackupProbe",
+    "MasEvidenceBackupProbe",
+)
+STATEMENT_LIMIT = 10
 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"),
+PREFERRED_PROJECT = "reef-storage"
+BAO_ADDR = "http://127.0.0.1:18200"
+_PROJECT_ID = re.compile(
+    r"\A[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\Z"
 )
+_SECRET_AUTH_FIELDS = frozenset({"client_token", "accessor"})
 
 
 class GateError(RuntimeError):
-    def __init__(self, code):
+    def __init__(self, code, detail=None):
         super().__init__(code)
         self.code = code
+        self.detail = detail
 
 
-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:
+_ROLE = {
+    "awsaccesskeyid": "access",
+    "accesskey": "access",
+    "accesskeyid": "access",
+    "scwaccesskey": "access",
+    "awssecretaccesskey": "secret",
+    "secretkey": "secret",
+    "secretaccesskey": "secret",
+    "scwsecretkey": "secret",
+}
+
+
+def _role(name):
+    return _ROLE.get("".join(char.lower() for char in name if char.isalnum()))
+
+
+def _mapping_text(value):
+    """Read one credentials blob. Ignore anything that is not a mapping."""
+    text = value.strip()
+    if text.startswith("{") and text.endswith("}"):
+        try:
+            parsed = json.loads(text)
+        except json.JSONDecodeError:
+            return None
+        return parsed if isinstance(parsed, dict) else None
+    rows = {}
+    for line in text.splitlines():
+        line = line.split("#", 1)[0].strip()
+        if not line or line.startswith("[") or "=" not in line:
             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")
+        key, raw = line.split("=", 1)
+        rows[key.strip()] = raw.strip().strip('"').strip("'")
+    return rows or None
+
+
+def _shape_detail(path, data, reason):
+    detail = {"path": path, "reason": reason}
+    if isinstance(data, dict):
+        detail["keys"] = sorted(str(key) for key in data)
+    else:
+        detail["shape"] = type(data).__name__
+    return detail
+
+
+def supplier_pair(data, path="", unwrap=True):
+    """Return the two workload fields. Refuse a partial or identical pair."""
+    if isinstance(data, str):
+        parsed = _mapping_text(data)
+        if isinstance(parsed, dict):
+            return supplier_pair(parsed, path, unwrap=False)
+        raise GateError("supplier_shape_rejected", _shape_detail(path, data, "not_a_mapping"))
+    if not isinstance(data, dict):
+        raise GateError("supplier_shape_rejected", _shape_detail(path, data, "not_a_mapping"))
+    nested = [value for value in data.values() if isinstance(value, dict)]
+    strings = {key: value for key, value in data.items() if isinstance(value, str)}
+    access = []
+    secret = []
+    for name, value in strings.items():
+        role = _role(str(name))
+        if role == "access":
+            access.append(value.strip())
+        elif role == "secret":
+            secret.append(value.strip())
+    if not access and not secret and len(nested) == 1 and not strings:
+        return supplier_pair(nested[0], path, unwrap=False)
+    if unwrap and not access and not secret and len(strings) == 1 and not nested:
+        parsed = _mapping_text(next(iter(strings.values())))
+        if isinstance(parsed, dict):
+            return supplier_pair(parsed, path, unwrap=False)
+    if len(access) != 1 or len(secret) != 1 or not access or not secret or not access[0] or not secret[0] or access[0] == secret[0]:
+        reason = "empty_or_equal" if access and secret else "pair_not_found"
+        raise GateError("supplier_shape_rejected", _shape_detail(path, data, reason))
+    return {"AWS_ACCESS_KEY_ID": access[0], "AWS_SECRET_ACCESS_KEY": secret[0]}
 
 
 def mas_statements(document):
@@ -72,15 +142,36 @@ def mas_statements(document):
     return found
 
 
+def _actions(statement):
+    value = statement.get("Action") or []
+    if isinstance(value, str):
+        return [value]
+    return list(value)
+
+
+def _principals(statement):
+    principal = statement.get("Principal")
+    if not isinstance(principal, dict):
+        return []
+    value = principal.get("SCW")
+    if isinstance(value, str):
+        return [value]
+    if isinstance(value, list):
+        return [item for item in value if isinstance(item, str)]
+    return []
+
+
 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)
+        actions.extend(_actions(item))
+    probe = next(
+        (item for item in document.get("Statement", []) if item.get("Sid") == PROBE_SID),
+        None,
+    )
+    probe_actions = _actions(probe) if probe else []
     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}],
@@ -88,19 +179,58 @@ def policy_observation(document):
             action in {"s3:DeleteObject", "s3:DeleteObjectVersion"} for action in actions
         ),
         "principal": APPLICATION_ID,
+        "statement_count": len(document.get("Statement", [])),
+        "probe_includes_publisher": bool(
+            probe
+            and f"application_id:{APPLICATION_ID}" in _principals(probe)
+            and all(action in probe_actions for action in PROBE_ACTIONS)
+        ),
+    }
+
+
+def merged_policy(live, desired):
+    """Keep live statements and fit the publisher probe plus the MAS prefix into 10."""
+    desired_items = list(desired.get("Statement", []))
+    desired_by_sid = {item.get("Sid"): item for item in desired_items}
+    statements = []
+    seen = set()
+    for item in live.get("Statement", []):
+        sid = item.get("Sid")
+        if sid in REPLACED_PROBES:
+            continue
+        statements.append(desired_by_sid.get(sid, item))
+        seen.add(sid)
+    for item in desired_items:
+        sid = item.get("Sid")
+        if sid not in seen:
+            statements.append(item)
+            seen.add(sid)
+    sids = [item.get("Sid") for item in statements]
+    if len(statements) > STATEMENT_LIMIT:
+        raise GateError(
+            "policy_statement_limit",
+            {"count": len(statements), "limit": STATEMENT_LIMIT, "sids": sids},
+        )
+    return {
+        "Version": live.get("Version") or desired.get("Version"),
+        "Id": live.get("Id") or desired.get("Id"),
+        "Statement": statements,
     }
 
 
 def denial_observation(results):
     required = ("sibling_prefix", "retained_delete", "administration")
+    names = (*required, "own_prefix")
     denied = {name: bool(results.get(name, {}).get("denied")) for name in required}
+    own = results.get("own_prefix") or {}
+    own_allowed = own.get("http_status") == 200 and own.get("denied") is False
     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()),
+        "http_status": {name: results.get(name, {}).get("http_status") for name in names},
+        "error_code": {name: results.get(name, {}).get("error_code") for name in names},
+        "own_prefix_allowed": own_allowed,
+        "passed": all(denied.values()) and own_allowed,
     }
 
 
@@ -148,7 +278,24 @@ def _kv_data(path):
     return data
 
 
+def scoped_supplier_pair(pair, project_id):
+    """Scope only the proof client. The stored workload pair stays the raw key."""
+    key = pair["AWS_ACCESS_KEY_ID"]
+    project = project_id.strip() if isinstance(project_id, str) else ""
+    applied = bool(project and "@" not in key and _PROJECT_ID.fullmatch(project))
+    if not applied:
+        return dict(pair), False
+    scoped = dict(pair)
+    scoped["AWS_ACCESS_KEY_ID"] = f"{key}@{project}"
+    return scoped, True
+
+
 def _client(pair):
+    """Avoid the boto3 1.36+ default checksum on PutBucketPolicy.
+
+    Scaleway accepts the body from older clients. The default checksum makes
+    the same document come back as MalformedPolicy.
+    """
     import boto3
     from botocore.config import Config
 
@@ -160,16 +307,39 @@ def _client(pair):
         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 public_error(message):
+    import re
+
+    text = str(message)
+    text = re.sub(
+        r"(?i)(secret|token|password|signature)([^\s,;]{0,12}\s*[:=]\s*)\S+",
+        r"\1\2<redacted>",
+        text,
+    )
+    text = re.sub(r"SCW[A-Z0-9]{8,}", "SCW<redacted>", text)
+    text = re.sub(r"\b[A-Za-z0-9+/]{32,}={0,2}\b", "<redacted>", text)
+    return text[:300]
+
+
+def s3_failure(exc):
+    response = exc.response if isinstance(getattr(exc, "response", None), dict) else {}
+    error = response.get("Error") if isinstance(response.get("Error"), dict) else {}
+    meta = response.get("ResponseMetadata") if isinstance(response.get("ResponseMetadata"), dict) else {}
+    detail = {
+        "error_code": error.get("Code"),
+        "http_status": meta.get("HTTPStatusCode"),
+    }
+    if isinstance(error.get("Message"), str) and error.get("Message"):
+        detail["message"] = public_error(error["Message"])
+    return detail
+
+
 def _status(call):
     from botocore.exceptions import ClientError
 
@@ -179,54 +349,98 @@ def _status(call):
         if body is not None:
             body.close()
         code = result.get("ResponseMetadata", {}).get("HTTPStatusCode", 200)
-        return {"http_status": code, "denied": False}
+        return {"http_status": code, "denied": False, "error_code": None}
     except ClientError as exc:
-        code = exc.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
-        return {"http_status": code, "denied": code == 403}
+        failure = s3_failure(exc)
+        code = failure["http_status"]
+        return {"http_status": code, "denied": code == 403, "error_code": failure["error_code"]}
+
+
+def backup_python(account_home):
+    """Pinned backup runtime on the account home.
+
+    Attended login replaces the process home with an empty session. A venv
+    under that session, or under /tmp, is not there on the next login.
+    """
+    return (
+        Path(account_home)
+        / ".local/share/railiance-platform/backup-venv/bin/python"
+    )
+
+
+def selected_backup_python():
+    return backup_python(pwd.getpwuid(os.getuid()).pw_dir)
+
+
+def same_interpreter(current, candidate):
+    """Compare the invoked path. The venv binary is a symlink to system Python."""
+    return Path(current) == Path(candidate)
 
 
 def _ensure_boto():
     try:
         import boto3  # noqa: F401
+        return
     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
+        pass
+    candidate = selected_backup_python()
+    if (
+        candidate.is_file()
+        and os.access(candidate, os.X_OK)
+        and not same_interpreter(sys.executable, candidate)
+    ):
+        os.execv(candidate, [str(candidate), *sys.argv])
+    raise GateError("boto3_required")
 
 
 def apply_live_policy(reader_pair, writer_pair):
+    from botocore.exceptions import ClientError
+
     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,
-    }
+    try:
+        live_policy = reader.get_bucket_policy(Bucket=BUCKET)["Policy"]
+    except ClientError as exc:
+        raise GateError("policy_read_failed", s3_failure(exc)) from None
+    try:
+        live = json.loads(live_policy)
+    except (TypeError, json.JSONDecodeError):
+        raise GateError("policy_read_failed", {"reason": "response_shape"}) from None
+    policy = merged_policy(live, desired)
+    statements = policy["Statement"]
     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))
+        try:
+            _client(writer_pair).put_bucket_policy(Bucket=BUCKET, Policy=json.dumps(policy))
+        except ClientError as exc:
+            raise GateError("policy_write_failed", s3_failure(exc)) from None
         receipt["put"] = "ok"
-    after = json.loads(reader.get_bucket_policy(Bucket=BUCKET)["Policy"])
+    try:
+        after_policy = reader.get_bucket_policy(Bucket=BUCKET)["Policy"]
+    except ClientError as exc:
+        raise GateError("policy_read_failed", s3_failure(exc)) from None
+    try:
+        after = json.loads(after_policy)
+    except (TypeError, json.JSONDecodeError):
+        raise GateError("policy_read_failed", {"reason": "response_shape"}) from None
     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):
+    if (
+        not receipt["mas"]["object_delete_absent"]
+        or receipt["mas"]["mas_sids_present"] != list(MAS_SIDS)
+        or not receipt["mas"]["probe_includes_publisher"]
+        or receipt["mas"]["statement_count"] > STATEMENT_LIMIT
+    ):
         raise GateError("mas_policy_not_installed")
     return receipt
 
 
-def prove_denials(pair):
-    client = _client(pair)
+def prove_denials(pair, project_id=""):
+    scoped, applied = scoped_supplier_pair(pair, project_id)
+    client = _client(scoped)
     results = {
         "sibling_prefix": _status(
             lambda: client.list_objects_v2(Bucket=BUCKET, Prefix=SIBLING, MaxKeys=1)
@@ -241,24 +455,75 @@ def prove_denials(pair):
     }
     observation = denial_observation(results)
     observation["preferred_project"] = PREFERRED_PROJECT
+    observation["project_id_present"] = bool(isinstance(project_id, str) and project_id.strip())
+    observation["project_scope_applied"] = applied
     _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")
+def _string_list(value):
+    if isinstance(value, list) and all(isinstance(item, str) for item in value):
+        return value
+    return None
+
+
+def _bounded_int(value):
+    if type(value) is int:
+        return value
+    if isinstance(value, str) and value.isdigit():
+        return int(value)
+    return None
+
+
+def _identity_detail(auth, policies, ttl, uses):
+    identity = _string_list(auth.get("identity_policies"))
+    legacy = _string_list(auth.get("policies"))
     return {
         "policies": policies,
-        "identity_policies": list(auth.get("identity_policies") or []),
+        "legacy_policies": legacy or [],
+        "token_policies_present": _string_list(auth.get("token_policies")) is not None,
+        "policies_present": legacy is not None,
+        "identity_policies": identity or [],
+        "ttl": ttl,
+        "ttl_type": type(auth.get("lease_duration")).__name__,
+        "num_uses": uses,
+        "num_uses_type": type(auth.get("num_uses")).__name__,
+        "auth_fields": sorted(key for key in auth if key not in _SECRET_AUTH_FIELDS),
+    }
+
+
+def _writer_identity(auth):
+    """Use the create response so the two token uses stay available for put and revoke.
+
+    OpenBao 2.5 returns token_policies. A non-empty legacy policies list must
+    name the same single policy. default and any parent policy stay rejected.
+    """
+    token_policies = _string_list(auth.get("token_policies"))
+    legacy = _string_list(auth.get("policies"))
+    policies = token_policies if token_policies else (legacy or [])
+    identity = _string_list(auth.get("identity_policies"))
+    if identity is None and "identity_policies" not in auth:
+        identity = []
+    ttl = _bounded_int(auth.get("lease_duration"))
+    uses = _bounded_int(auth.get("num_uses"))
+    legacy_conflict = bool(legacy) and legacy != ["backup-pair-writer"]
+    accepted = (
+        not legacy_conflict
+        and policies == ["backup-pair-writer"]
+        and "default" not in policies
+        and identity == []
+        and type(ttl) is int
+        and 0 < ttl <= 300
+        and type(uses) is int
+        and 0 < uses <= 2
+    )
+    if not accepted:
+        raise GateError("writer_identity_rejected", _identity_detail(auth, policies, ttl, uses))
+    return {
+        "policies": policies,
+        "identity_policies": [],
         "ttl": ttl,
         "num_uses": uses,
     }
@@ -309,6 +574,53 @@ def _custody():
     return module
 
 
+class _NoRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, *args, **kwargs):
+        raise GateError("writer_token_failed")
+
+
+def writer_auth(payload):
+    """Keep the server's num_uses. The OpenBao 2.5 CLI drops that field."""
+    auth = payload.get("auth") if isinstance(payload, dict) else None
+    token = auth.get("client_token") if isinstance(auth, dict) else None
+    if not isinstance(token, str) or not token:
+        raise GateError("writer_token_failed")
+    return auth
+
+
+def _request_writer_auth():
+    addr = os.environ.get("BAO_ADDR") or os.environ.get("VAULT_ADDR") or BAO_ADDR
+    parent = os.environ.get("BAO_TOKEN") or os.environ.get("VAULT_TOKEN")
+    if addr.rstrip("/") != BAO_ADDR or not isinstance(parent, str) or not parent:
+        raise GateError("writer_token_failed")
+    body = json.dumps({
+        "policies": ["backup-pair-writer"],
+        "ttl": "5m",
+        "num_uses": 2,
+        "no_default_policy": True,
+    }).encode()
+    request = urllib.request.Request(
+        BAO_ADDR + "/v1/auth/token/create-orphan",
+        data=body,
+        method="POST",
+        headers={"X-Vault-Token": parent, "Content-Type": "application/json"},
+    )
+    opener = urllib.request.build_opener(_NoRedirect)
+    try:
+        with opener.open(request, timeout=20) as response:
+            raw = response.read(65537)
+    except GateError:
+        raise
+    except Exception:
+        raise GateError("writer_token_failed") from None
+    if len(raw) > 65536:
+        raise GateError("writer_token_failed")
+    try:
+        return writer_auth(json.loads(raw))
+    except json.JSONDecodeError:
+        raise GateError("writer_token_failed") from None
+
+
 def write_pair(pair):
     custody = _custody()
 
@@ -318,21 +630,7 @@ def write_pair(pair):
     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"]
+    auth = _request_writer_auth()
     token = auth["client_token"]
     try:
         identity = _writer_identity(auth)
@@ -365,21 +663,31 @@ def main():
     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)
+        supplier = supplier_pair(_kv_data(SUPPLIER), SUPPLIER)
+        backup_path = "platform/workloads/railiance/backup/object-storage"
+        bootstrap_path = "platform/workloads/railiance/scaleway/bootstrap"
+        backup = _kv_data(backup_path)
+        bootstrap = _kv_data(bootstrap_path)
+        reader = supplier_pair(backup, backup_path)
+        writer = supplier_pair(bootstrap, bootstrap_path)
         apply_live_policy(reader, writer)
-        prove_denials(supplier)
+        prove_denials(supplier, bootstrap.get("DEFAULT_PROJECT_ID", ""))
         write_pair(supplier)
     except GateError as exc:
-        _write_json(RECEIPT_DIR / "mas-evidence-publisher-gate.json", {"status": "failed", "code": exc.code})
+        receipt = {"status": "failed", "code": exc.code, "values_recorded": False}
+        if exc.detail:
+            receipt["detail"] = exc.detail
+        _write_json(RECEIPT_DIR / "mas-evidence-publisher-gate.json", receipt)
         return 71
-    except Exception:
+    except Exception as exc:
         _write_json(
             RECEIPT_DIR / "mas-evidence-publisher-gate.json",
-            {"status": "failed", "code": "contained_gate_failure"},
+            {
+                "status": "failed",
+                "code": "contained_gate_failure",
+                "exception": type(exc).__name__,
+                "values_recorded": False,
+            },
         )
         return 72
     _write_json(RECEIPT_DIR / "mas-evidence-publisher-gate.json", {"status": "passed", "values_recorded": False})

scripts/operator-browser/xdg-open

@@ -7,6 +7,7 @@ headed Chromium binary pinned by coulomb-social's Playwright install.
 """
 import json
 import os
+import pwd
 import subprocess
 import sys
 import tempfile
@@ -62,8 +63,22 @@ def chromium_executable(browsers_json, cache_root):
     return ""
 
 
+def playwright_cache(process_home, account_home):
+    """The Chromium pin lives in the account cache.
+
+    Attended platform-admin login replaces the process home with an empty
+    session before OpenBao starts. Looking under that session misses the
+    binary, the window never opens, and the OIDC callback times out.
+    """
+    del process_home
+    return Path(account_home) / ".cache" / "ms-playwright"
+
+
 def browser_path():
-    return chromium_executable(BROWSERS_JSON, Path.home() / ".cache/ms-playwright")
+    account_home = pwd.getpwuid(os.getuid()).pw_dir
+    return chromium_executable(
+        BROWSERS_JSON, playwright_cache(Path.home(), account_home)
+    )
 
 
 def launch_argv(browser, profile_dir, url):