Prepare bounded Coulomb API identity bootstrap with private restartable custody

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b

scripts/application_recovery_identities.py

@@ -0,0 +1,383 @@
+"""Coulomb-only identity bootstrap; stdin auth is temporary, API pairs stay private.
+
+No personal Console session is a routine credential. Default preview has no
+effects. Native S3/escrow boundary checks remain mandatory after this IAM stage.
+"""
+
+from __future__ import annotations
+
+import argparse
+import fcntl
+import json
+import os
+import stat
+import sys
+import tempfile
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from urllib.error import HTTPError
+from urllib.parse import urlencode
+from urllib.request import HTTPRedirectHandler, Request, build_opener
+
+PUBLISHER = "681d00bf-b965-4323-b657-97bfde423511"
+PROJECT = "coulomb-social-recovery"
+DECISION = "27bbc1ac-e275-42be-ba34-6cf16eae4e78"
+PROGRAMMATIC_DIRECTION = "6139e02b-3fb7-4eff-be8a-efe127c45ce5"
+ROLES = {
+    "bootstrap": {"name": "coulomb-social-recovery-bootstrap", "permissions": [
+        "ObjectStorageFullAccess", "SecretManagerFullAccess"]},
+    "reader": {"name": "coulomb-social-recovery-reader", "permissions": [
+        "ObjectStorageReadOnly"]},
+    "custodian": {"name": "coulomb-social-recovery-custodian", "permissions": [
+        "ObjectStorageReadOnly", "ObjectStorageObjectsWrite", "SecretManagerReadOnly",
+        "SecretManagerSecretAccess"]},
+    "publisher": {"name": "coulomb-social-backup-primary", "permissions": [
+        "ObjectStorageObjectsRead", "ObjectStorageObjectsWrite", "ObjectStorageObjectsDelete",
+        "ObjectStorageBucketsRead"]},
+}
+
+
+class Refused(RuntimeError):
+    pass
+
+
+class ProviderError(Refused):
+    def __init__(self, status):
+        super().__init__("provider_request_refused")
+        self.status = status
+
+
+def inside_git(path):
+    return any((parent / ".git").exists() for parent in [path.resolve(), *path.resolve().parents])
+
+
+def private_directory(path):
+    info = path.lstat()
+    if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077:
+        raise Refused("private_owner_directory_required")
+    if inside_git(path):
+        raise Refused("credential_directory_must_be_outside_git")
+
+
+def fsync_directory(path):
+    fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY)
+    try:
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+
+
+def private_json(path):
+    info = path.lstat()
+    if not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077:
+        raise Refused("private_owner_file_required")
+    return json.loads(path.read_text())
+
+
+def checkpoint(path, value):
+    fd, name = tempfile.mkstemp(prefix=".checkpoint-", dir=path.parent)
+    try:
+        with os.fdopen(fd, "w") as stream:
+            json.dump(value, stream, sort_keys=True, indent=2)
+            stream.write("\n")
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(name, path)
+        fsync_directory(path.parent)
+    finally:
+        if os.path.exists(name):
+            os.unlink(name)
+
+
+class Transport:
+    def __init__(self, auth):
+        if set(auth) != {"session_token", "user_agent"} or any(
+            not isinstance(v, str) or not v or "\n" in v or "\r" in v for v in auth.values()
+        ):
+            raise Refused("temporary_stdin_auth_required")
+        self.headers = {"X-Session-Token": auth["session_token"],
+                        "User-Agent": auth["user_agent"], "Content-Type": "application/json"}
+
+        class NoRedirect(HTTPRedirectHandler):
+            def redirect_request(self, *args, **kwargs):
+                raise Refused("provider_redirect_refused")
+
+        self.opener = build_opener(NoRedirect())
+
+    def __call__(self, method, path, data=None):
+        if not path.startswith(("/iam/v1alpha1/", "/account/v3/")) or ".." in path:
+            raise Refused("provider_api_scope_refused")
+        request = Request("https://api.scaleway.com" + path, method=method,
+                          headers=self.headers,
+                          data=None if data is None else json.dumps(data).encode())
+        try:
+            with self.opener.open(request, timeout=20) as response:
+                raw = response.read(8 * 1024**2)
+                return json.loads(raw) if raw else {}
+        except HTTPError as error:
+            raise ProviderError(error.code) from None
+
+
+class Bootstrap:
+    def __init__(self, api, directory, now=None):
+        self.api, self.directory = api, directory
+        private_directory(directory)
+        self.now = now or datetime.now(UTC)
+        self.path = directory / "identity-state.json"
+        self.state = private_json(self.path) if self.path.exists() else {
+            "schema": "platform.coulomb-recovery-identities.v1", "decision": DECISION,
+            "access_decision": PROGRAMMATIC_DIRECTION, "apps": {}, "policies": {}, "keys": {},
+        }
+        if self.state.get("decision") != DECISION or self.state.get("access_decision") != (
+            PROGRAMMATIC_DIRECTION
+        ) or self.state.get("schema") != "platform.coulomb-recovery-identities.v1":
+            raise Refused("journal_scope_refused")
+        if self.state.get("pending_key_issue"):
+            raise Refused("uncertain_key_issue_needs_reconciliation")
+
+    def save(self):
+        checkpoint(self.path, self.state)
+
+    def inventory(self, path, key, **filters):
+        rows = []
+        for page in range(1, 101):
+            value = self.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 project(self, organization):
+        if self.state.get("project"):
+            project = self.api("GET", "/account/v3/projects/" + self.state["project"]["id"])
+        else:
+            found = [v for v in self.inventory("/account/v3/projects", "projects",
+                                              organization_id=organization) if v["name"] == PROJECT]
+            if found:
+                raise Refused("untracked_existing_project_requires_review")
+            project = self.api("POST", "/account/v3/projects", {
+                "name": PROJECT, "organization_id": organization,
+                "description": "Coulomb-only recovery custody; decision " + DECISION,
+            })
+            self.state["project"] = project
+            self.save()
+        if project["name"] != PROJECT or project["organization_id"] != organization:
+            raise Refused("project_binding_refused")
+        return project
+
+    def application(self, role, organization, applications):
+        spec = ROLES[role]
+        if role == "publisher":
+            app = self.api("GET", "/iam/v1alpha1/applications/" + PUBLISHER)
+        elif role in self.state["apps"]:
+            app = self.api("GET", "/iam/v1alpha1/applications/" + self.state["apps"][role]["id"])
+        else:
+            if any(v["name"] == spec["name"] for v in applications):
+                raise Refused("untracked_existing_application_requires_review")
+            app = self.api("POST", "/iam/v1alpha1/applications", {
+                "name": spec["name"], "organization_id": organization,
+                "description": "Coulomb-only " + role + "; " + DECISION,
+            })
+            self.state["apps"][role] = app
+            self.save()
+        if app["name"] != spec["name"] or app["organization_id"] != organization:
+            raise Refused("application_binding_refused")
+        return app
+
+    def policy(self, role, app, project, organization):
+        name = "coulomb-recovery-project-" + role
+        if role in self.state["policies"]:
+            policy = self.api("GET", "/iam/v1alpha1/policies/" + self.state["policies"][role]["id"])
+        else:
+            policies = self.inventory("/iam/v1alpha1/policies", "policies",
+                                      organization_id=organization)
+            if any(p["name"] == name for p in policies):
+                raise Refused("untracked_existing_policy_requires_review")
+            policy = self.api("POST", "/iam/v1alpha1/policies", {
+                "name": name, "description": "Coulomb recovery project only; " + DECISION,
+                "organization_id": organization, "application_id": app["id"],
+                "rules": [{"project_ids": [project["id"]],
+                           "permission_set_names": ROLES[role]["permissions"]}],
+            })
+            self.state["policies"][role] = policy
+            self.save()
+        if (policy["application_id"] != app["id"] or policy["name"] != name
+                or policy["organization_id"] != organization):
+            raise Refused("policy_binding_refused")
+        rules = self.api("GET", "/iam/v1alpha1/rules?" + urlencode({"policy_id": policy["id"]}))
+        if len(rules["rules"]) != 1:
+            raise Refused("single_project_rule_required")
+        rule = rules["rules"][0]
+        if (rule.get("organization_id") or rule.get("project_ids") != [project["id"]]
+                or set(rule.get("permission_set_names", [])) != set(ROLES[role]["permissions"])
+                or rule.get("condition") or rule.get("deny")):
+            raise Refused("effective_project_rule_refused")
+
+    def key(self, role, app, project):
+        path = self.directory / (role + "-pair.json")
+        if role in self.state["keys"]:
+            pair = private_json(path)
+            if set(pair) != {"AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"}:
+                raise Refused("exact_pair_fields_required")
+            current = self.api("GET", "/iam/v1alpha1/api-keys/" + pair["AWS_ACCESS_KEY_ID"])
+            if (current["application_id"] != app["id"]
+                    or current["default_project_id"] != project["id"]):
+                raise Refused("existing_key_binding_refused")
+            return
+        if path.exists() or path.is_symlink():
+            raise Refused("untracked_pair_requires_review")
+        self.state["pending_key_issue"] = {"role": role, "application_id": app["id"]}
+        self.save()
+        payload = {"application_id": app["id"], "default_project_id": project["id"],
+                   "description": "Coulomb recovery " + role + "; " + DECISION}
+        if role == "bootstrap":
+            payload["expires_at"] = (self.now + timedelta(hours=2)).isoformat()
+        try:
+            issued = self.api("POST", "/iam/v1alpha1/api-keys", payload)
+        except ProviderError as error:
+            if error.status in {400, 401, 403, 404, 422, 429}:
+                self.state.pop("pending_key_issue")
+                self.save()
+            raise
+        try:
+            if issued.get("application_id") != app["id"]:
+                raise Refused("issued_key_binding_refused")
+            if role == "bootstrap":
+                expires = datetime.fromisoformat(
+                    (issued.get("expires_at") or "").replace("Z", "+00:00"))
+                if not self.now < expires <= self.now + timedelta(hours=2):
+                    raise Refused("bounded_bootstrap_expiry_required")
+            pair = {"AWS_ACCESS_KEY_ID": issued["access_key"],
+                    "AWS_SECRET_ACCESS_KEY": issued["secret_key"]}
+            if any(not isinstance(v, str) or not v for v in pair.values()):
+                raise Refused("complete_pair_required")
+            fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+            with os.fdopen(fd, "w") as stream:
+                json.dump(pair, stream)
+                stream.write("\n")
+                stream.flush()
+                os.fsync(stream.fileno())
+            fsync_directory(path.parent)
+        except (Refused, OSError, KeyError, ValueError, TypeError):
+            if issued.get("access_key") and issued.get("application_id") == app["id"]:
+                try:
+                    self.api("DELETE", "/iam/v1alpha1/api-keys/" + issued["access_key"])
+                    self.state.pop("pending_key_issue")
+                    self.save()
+                except (Refused, OSError, KeyError, ValueError, TypeError):
+                    raise Refused("pair_persistence_and_revocation_failed_review_required") from None
+            raise Refused("pair_persistence_failed_review_required") from None
+        self.state["keys"][role] = {"application_id": app["id"], "file": path.name,
+                                    "expires_at": issued.get("expires_at")}
+        self.state.pop("pending_key_issue")
+        self.save()
+
+    def run(self):
+        publisher = self.api("GET", "/iam/v1alpha1/applications/" + PUBLISHER)
+        if publisher["id"] != PUBLISHER or publisher["name"] != ROLES["publisher"]["name"]:
+            raise Refused("publisher_binding_refused")
+        organization = publisher["organization_id"]
+        permissions = self.inventory("/iam/v1alpha1/permission-sets", "permission_sets",
+                                     organization_id=organization)
+        names = {v["name"] for v in permissions}
+        if any(set(role["permissions"]) - names for role in ROLES.values()):
+            raise Refused("required_permission_sets_unavailable")
+        applications = self.inventory("/iam/v1alpha1/applications", "applications",
+                                       organization_id=organization)
+        project = self.project(organization)
+        for role in ROLES:
+            app = self.application(role, organization, applications)
+            self.policy(role, app, project, organization)
+            if role != "publisher":
+                self.key(role, app, project)
+        self.state["status"] = "identity_bootstrap_complete_native_boundaries_pending"
+        self.save()
+        return {"status": self.state["status"], "new_project": PROJECT,
+                "managed_recovery_applications": 3, "project_policies_verified": 4,
+                "private_pairs": 3,
+                "publisher_pair_replaced": False, "organization_wide_grants": False,
+                "bootstrap_key_max_hours": 2, "native_s3_escrow_boundaries_verified": False,
+                "credential_values_printed": False}
+
+    def retire_bootstrap(self):
+        if self.state.get("status") != "identity_bootstrap_complete_native_boundaries_pending":
+            raise Refused("completed_identity_stage_required")
+        app = self.state["apps"]["bootstrap"]
+        actual = self.api("GET", "/iam/v1alpha1/applications/" + app["id"])
+        if actual["name"] != ROLES["bootstrap"]["name"]:
+            raise Refused("bootstrap_application_binding_refused")
+        path = self.directory / "bootstrap-pair.json"
+        if self.state["keys"]["bootstrap"].get("revoked_verified") and not path.exists():
+            return {"status": "bootstrap_api_key_retired", "credential_values_printed": False}
+        pair = private_json(path)
+        key_path = "/iam/v1alpha1/api-keys/" + pair["AWS_ACCESS_KEY_ID"]
+        try:
+            current = self.api("GET", key_path)
+        except ProviderError as error:
+            if error.status != 404:
+                raise
+        else:
+            if (current["application_id"] != app["id"] or current["default_project_id"] != (
+                self.state["project"]["id"]
+            )):
+                raise Refused("retirement_key_binding_refused")
+            self.api("DELETE", key_path)
+            try:
+                self.api("GET", key_path)
+            except ProviderError as error:
+                if error.status != 404:
+                    raise
+            else:
+                raise Refused("bootstrap_revocation_unconfirmed")
+        self.state["keys"]["bootstrap"]["revoked_verified"] = True
+        self.save()
+        path.unlink()  # Only this just-revoked temporary pair, never retained archives/pairs.
+        fsync_directory(path.parent)
+        return {"status": "bootstrap_api_key_retired", "own_temporary_pair_removed": True,
+                "credential_values_printed": False}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--execute", action="store_true")
+    parser.add_argument("--directory", type=Path)
+    parser.add_argument("--retire-bootstrap", action="store_true")
+    args = parser.parse_args()
+    if not args.execute:
+        print(json.dumps({"status": "preview", "project": PROJECT,
+                          "applications": [v["name"] for k, v in ROLES.items() if k != "publisher"],
+                          "provider_contacted": False, "credential_values_printed": False}))
+        return 0
+    if not args.directory or not args.directory.is_absolute():
+        raise Refused("private_absolute_directory_required")
+    private_directory(args.directory)
+    fd = os.open(args.directory / "identity-bootstrap.lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW,
+                 0o600)
+    with os.fdopen(fd, "w") as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        raw = sys.stdin.read(65537)
+        if len(raw) > 65536:
+            raise Refused("bounded_temporary_auth_required")
+        auth = json.loads(raw)
+        bootstrap = Bootstrap(Transport(auth), args.directory)
+        print(json.dumps(bootstrap.retire_bootstrap() if args.retire_bootstrap else bootstrap.run()))
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except (Refused, OSError, KeyError, ValueError, TypeError) as error:
+        print(json.dumps({"status": "failed", "failure_type": type(error).__name__,
+                          "http_status": getattr(error, "status", None),
+                          "refusal": str(error) if isinstance(error, Refused) and all(
+                              c.islower() or c == "_" for c in str(error)) else None,
+                          "credential_values_printed": False}))
+        raise SystemExit(1) from None

scripts/bootstrap_coulomb_recovery_session.mjs

@@ -0,0 +1,100 @@
+#!/usr/bin/env node
+/** One attended bootstrap; committed helper creates scoped routine API identities. */
+import {createHash} from 'node:crypto';
+import {execFileSync,spawn} from 'node:child_process';
+import {readFileSync,mkdirSync,lstatSync} from 'node:fs';
+import {homedir} from 'node:os';
+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'},'preflight-only':{type:'boolean'}}});
+const root=resolve(dirname(fileURLToPath(import.meta.url)),'..');
+const factory=join(root,'scripts','application_recovery_identities.py');
+const directory=join(homedir(),'.config','railiance-platform','coulomb-social-recovery');
+const digest=data=>createHash('sha256').update(data).digest('hex');
+let browser,reader,child;
+
+function sourcePreflight(){
+ const committed=execFileSync('git',['show','HEAD:scripts/application_recovery_identities.py'],{cwd:root});
+ if(digest(committed)!==digest(readFileSync(factory)))throw new Error('committed_factory_required');
+ mkdirSync(directory,{recursive:true,mode:0o700});
+ const info=lstatSync(directory);
+ if(!info.isDirectory()||info.uid!==process.getuid()||(info.mode&0o077))throw new Error('private_directory_required');
+ const validation="import importlib.util,sys;from pathlib import Path;"+
+  "s=importlib.util.spec_from_file_location('recovery_bootstrap',sys.argv[1]);"+
+  "m=importlib.util.module_from_spec(s);s.loader.exec_module(m);m.private_directory(Path(sys.argv[2]))";
+ execFileSync('python3',['-c',validation,factory,directory],{stdio:['ignore','pipe','pipe']});
+ return digest(committed);
+}
+
+async function runFactory(page,retire=false){
+ const sourceHash=sourcePreflight();
+ if(new URL(page.url()).origin!=='https://console.scaleway.com')throw new Error('console_page_required');
+ const auth=await page.evaluate(()=>{
+  const names=Object.keys(localStorage).filter(n=>n.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_bootstrap_session_required');
+  return {session_token:session.token,user_agent:session.jwt.userAgent};
+ });
+ const result=await new Promise((resolveResult,reject)=>{
+  child=spawn('python3',[factory,'--execute','--directory',directory,
+   ...(retire?['--retire-bootstrap']:[])],{stdio:['pipe','pipe','pipe']});
+  let output='';
+  child.stdout.on('data',data=>{if(output.length<65536)output+=data.toString();});
+  child.stderr.on('data',()=>{}); // Provider/credential diagnostics never enter chat.
+  child.on('error',()=>reject(new Error('bootstrap_child_failed')));
+  child.on('close',code=>{
+   child=null;
+   try {resolveResult({exit_code:code,result:JSON.parse(output.trim()),factory_source_sha256:sourceHash});}
+   catch {reject(new Error('redacted_bootstrap_result_required'));}
+  });
+  child.stdin.end(JSON.stringify(auth));
+ });
+ return result;
+}
+
+try{
+ const sourceHash=sourcePreflight();
+ 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');
+ if(values['preflight-only']){
+  console.log(JSON.stringify({status:'preflight_passed',factory_source_sha256:sourceHash,
+   private_directory_validated:true,locked_playwright_verified:true,provider_contacted:false}));
+  process.exit(0);
+ }
+ const {chromium}=await import(pathToFileURL(modulePath).href);
+ 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});
+ reader=createInterface({input:process.stdin,output:process.stdout});
+ console.log('One-time bootstrap browser opened. Sign in privately; no personal state is saved.');
+ for(;;){
+  const answer=await reader.question('After Console login: Enter/run = bootstrap; finish = retire temporary key and close; stop = close. ');
+  if(answer.trim()==='stop')break;
+  try{
+   const retire=answer.trim()==='finish';
+   const outcome=await runFactory(page,retire);
+   console.log(JSON.stringify(outcome));
+   if(outcome.exit_code===0&&retire){
+    try{await page.goto('https://account.scaleway.com/logout',{waitUntil:'domcontentloaded',timeout:30000});}catch{}
+    await context.clearCookies();
+    break;
+   }
+   if(outcome.exit_code===0){
+    console.log('Identity stage complete. Use scoped APIs for resource/boundary proofs, then finish. Keep this window for bootstrap only.');
+    continue;
+   }
+  }catch{console.error('Bootstrap stopped safely. No credential values printed. Keep this window for review.');}
+  console.log('No automatic retry. Review the private journal; keep this browser open until the next reviewed attempt.');
+ }
+}catch{
+ console.error('Bootstrap preflight/session failed; no credential values printed.');process.exitCode=1;
+}finally{
+ if(child)child.kill('SIGTERM');
+ if(reader)reader.close();
+ if(browser)await browser.close();
+}

tests/test_application_recovery_identities.py

@@ -0,0 +1,223 @@
+"""Synthetic provider responses only; no live identity or network access."""
+
+import copy
+import importlib.util
+import json
+import os
+import tempfile
+import unittest
+from datetime import UTC, datetime
+from pathlib import Path
+from unittest.mock import patch
+from urllib.parse import parse_qs, urlsplit
+from urllib.request import HTTPRedirectHandler, Request
+
+SOURCE = Path(__file__).resolve().parents[1] / "scripts/application_recovery_identities.py"
+spec = importlib.util.spec_from_file_location("recovery_identities", SOURCE)
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+NOW = datetime(2026, 10, 28, 10, tzinfo=UTC)
+
+
+class Provider:
+    def __init__(self):
+        self.calls, self.projects, self.policies, self.keys = [], {}, {}, {}
+        self.apps = {module.PUBLISHER: {"id": module.PUBLISHER, "name":
+                     "coulomb-social-backup-primary", "organization_id": "synthetic-org"}}
+        self.rules = {}
+        self.extra_org_rule = False
+        self.uncertain_key = False
+        self.key_http_error = None
+        self.foreign_key_response = False
+        self.unbounded_bootstrap = False
+
+    def __call__(self, method, url, data=None):
+        self.calls.append((method, url, copy.deepcopy(data)))
+        path = urlsplit(url).path
+        if path.endswith("/permission-sets"):
+            names = sorted({p for role in module.ROLES.values() for p in role["permissions"]})
+            return {"permission_sets": [{"name": n} for n in names], "total_count": str(len(names))}
+        if path.endswith("/rules"):
+            policy_id = parse_qs(urlsplit(url).query)["policy_id"][0]
+            rules = copy.deepcopy(self.rules[policy_id])
+            if self.extra_org_rule:
+                rules[0]["organization_id"] = "synthetic-org"
+            return {"rules": rules}
+        for suffix, store, key in [("projects", self.projects, "projects"),
+                                    ("applications", self.apps, "applications"),
+                                    ("policies", self.policies, "policies"),
+                                    ("api-keys", self.keys, "api_keys")]:
+            if path.endswith("/" + suffix):
+                if method == "GET":
+                    return {key: list(store.values()), "total_count": str(len(store))}
+                if method == "POST":
+                    identity = suffix + "-" + str(len(store) + 1)
+                    created = {"id": identity, **copy.deepcopy(data)}
+                    if suffix == "policies":
+                        self.rules[identity] = created.pop("rules")
+                    if suffix == "api-keys":
+                        created.update(access_key=identity,
+                                       secret_key="SYNTHETIC_SECRET_" + identity)
+                        if self.foreign_key_response:
+                            created["application_id"] = "foreign-application"
+                        if self.unbounded_bootstrap:
+                            created["expires_at"] = None
+                    store[identity] = created
+                    if suffix == "api-keys" and self.uncertain_key:
+                        raise OSError("synthetic lost acknowledgement")
+                    if suffix == "api-keys" and self.key_http_error:
+                        raise module.ProviderError(self.key_http_error)
+                    return copy.deepcopy(created)
+            marker = "/" + suffix + "/"
+            if marker in path:
+                identity = path.split(marker, 1)[1]
+                if method == "DELETE":
+                    del store[identity]
+                    return {}
+                if identity not in store:
+                    raise module.ProviderError(404)
+                return copy.deepcopy(store[identity])
+        raise AssertionError("Unexpected synthetic request")
+
+
+class IdentityBootstrap(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory(dir="/tmp")
+        self.addCleanup(self.temporary.cleanup)
+        self.directory = Path(self.temporary.name)
+        self.directory.chmod(0o700)
+        # Some workstations manage /tmp itself as a Git worktree. Isolate the
+        # synthetic provider cases; exercise the real Git guard separately below.
+        self.git_guard = patch.object(module, "inside_git", return_value=False)
+        self.git_guard.start()
+        self.addCleanup(self.git_guard.stop)
+        self.api = Provider()
+
+    def run_bootstrap(self):
+        return module.Bootstrap(self.api, self.directory, NOW).run()
+
+    def test_project_only_grants_private_pairs_no_publisher_rotation_and_idempotent_resume(self):
+        before = copy.deepcopy(self.api.apps[module.PUBLISHER])
+        result = self.run_bootstrap()
+        self.assertFalse(result["organization_wide_grants"])
+        self.assertFalse(result["native_s3_escrow_boundaries_verified"])
+        policies = [d for method, path, d in self.api.calls
+                    if method == "POST" and path.endswith("/policies")]
+        self.assertEqual(len(policies), 4)
+        for policy in policies:
+            self.assertEqual(policy["rules"][0]["project_ids"], ["projects-1"])
+            self.assertNotIn("organization_id", policy["rules"][0])
+            self.assertFalse(any(p.startswith(("IAM", "AllProducts"))
+                                 for p in policy["rules"][0]["permission_set_names"]))
+        self.assertEqual(self.api.apps[module.PUBLISHER], before)
+        self.assertFalse(any(k["application_id"] == module.PUBLISHER
+                             for k in self.api.keys.values()))
+        for role in ["bootstrap", "reader", "custodian"]:
+            path = self.directory / (role + "-pair.json")
+            self.assertEqual(path.stat().st_mode & 0o777, 0o600)
+            self.assertIn("AWS_SECRET_ACCESS_KEY", json.loads(path.read_text()))
+        self.assertNotIn("SYNTHETIC_SECRET", (self.directory / "identity-state.json").read_text())
+        self.assertNotIn("SYNTHETIC_SECRET", json.dumps(result))
+        posts = sum(method == "POST" for method, _, _ in self.api.calls)
+        self.run_bootstrap()
+        self.assertEqual(sum(method == "POST" for method, _, _ in self.api.calls), posts)
+
+    def test_untracked_existing_project_is_not_adopted_or_granted_access(self):
+        self.api.projects["foreign"] = {"id": "foreign", "name": module.PROJECT,
+                                        "organization_id": "synthetic-org"}
+        with self.assertRaisesRegex(module.Refused, "untracked_existing_project"):
+            self.run_bootstrap()
+        self.assertFalse(any(method == "POST" for method, _, _ in self.api.calls))
+
+    def test_effective_org_wide_rule_refuses_before_any_key_is_issued(self):
+        self.api.extra_org_rule = True
+        with self.assertRaisesRegex(module.Refused, "effective_project_rule"):
+            self.run_bootstrap()
+        self.assertEqual(self.api.keys, {})
+
+    def test_unknown_key_outcome_stops_restart_without_minting_a_duplicate(self):
+        self.api.uncertain_key = True
+        with self.assertRaises(OSError):
+            self.run_bootstrap()
+        self.assertEqual(len(self.api.keys), 1)
+        calls = len(self.api.calls)
+        with self.assertRaisesRegex(module.Refused, "uncertain_key_issue"):
+            self.run_bootstrap()
+        self.assertEqual(len(self.api.calls), calls)
+
+    def test_server_error_during_issuance_remains_uncertain_and_blocks_duplicate(self):
+        self.api.key_http_error = 503
+        with self.assertRaises(module.ProviderError):
+            self.run_bootstrap()
+        calls = len(self.api.calls)
+        with self.assertRaisesRegex(module.Refused, "uncertain_key_issue"):
+            self.run_bootstrap()
+        self.assertEqual(len(self.api.calls), calls)
+
+    def test_unbounded_bootstrap_key_is_revoked_and_never_delivered(self):
+        self.api.unbounded_bootstrap = True
+        with self.assertRaisesRegex(module.Refused, "pair_persistence_failed"):
+            self.run_bootstrap()
+        self.assertEqual(self.api.keys, {})
+        self.assertFalse((self.directory / "bootstrap-pair.json").exists())
+
+    def test_mismatched_returned_application_never_authorizes_deleting_a_foreign_key(self):
+        self.api.foreign_key_response = True
+        with self.assertRaisesRegex(module.Refused, "pair_persistence_failed"):
+            self.run_bootstrap()
+        self.assertFalse(any(method == "DELETE" for method, _, _ in self.api.calls))
+        self.assertIn("pending_key_issue", json.loads((
+            self.directory / "identity-state.json").read_text()))
+
+    def test_pair_write_failure_revokes_only_the_just_issued_owned_key(self):
+        real_open = os.open
+
+        def fail_reader(path, *args, **kwargs):
+            if str(path).endswith("reader-pair.json"):
+                raise OSError("synthetic disk failure")
+            return real_open(path, *args, **kwargs)
+
+        with (patch.object(module.os, "open", fail_reader),
+              self.assertRaisesRegex(module.Refused, "pair_persistence_failed")):
+            self.run_bootstrap()
+        deletions = [path for method, path, _ in self.api.calls if method == "DELETE"]
+        self.assertEqual(deletions, ["/iam/v1alpha1/api-keys/api-keys-2"])
+        self.assertEqual(len(self.api.keys), 1)  # Earlier persisted bootstrap pair is retained.
+
+    def test_private_credentials_cannot_be_written_inside_a_git_checkout(self):
+        self.git_guard.stop()
+        (self.directory / ".git").mkdir()
+        with self.assertRaisesRegex(module.Refused, "outside_git"):
+            module.Bootstrap(self.api, self.directory, NOW)
+        self.assertEqual(self.api.calls, [])
+
+    def test_bootstrap_retirement_revokes_only_temporary_key_and_retains_routine_pairs(self):
+        self.run_bootstrap()
+        before = {role: (self.directory / (role + "-pair.json")).read_bytes()
+                  for role in ["reader", "custodian"]}
+        result = module.Bootstrap(self.api, self.directory, NOW).retire_bootstrap()
+        self.assertEqual(result["status"], "bootstrap_api_key_retired")
+        self.assertEqual(len(self.api.keys), 2)
+        self.assertFalse((self.directory / "bootstrap-pair.json").exists())
+        for role, value in before.items():
+            self.assertEqual((self.directory / (role + "-pair.json")).read_bytes(), value)
+        self.assertTrue(module.Bootstrap(self.api, self.directory, NOW).retire_bootstrap())
+
+    def test_retirement_cannot_delete_a_key_after_foreign_application_binding_drift(self):
+        self.run_bootstrap()
+        bootstrap_key = json.loads((self.directory / "bootstrap-pair.json").read_text())
+        self.api.keys[bootstrap_key["AWS_ACCESS_KEY_ID"]]["application_id"] = "foreign"
+        with self.assertRaisesRegex(module.Refused, "retirement_key_binding_refused"):
+            module.Bootstrap(self.api, self.directory, NOW).retire_bootstrap()
+        self.assertFalse(any(method == "DELETE" for method, _, _ in self.api.calls))
+
+    def test_provider_redirect_cannot_forward_the_temporary_operator_token(self):
+        transport = module.Transport({"session_token": "synthetic-only", "user_agent": "test"})
+        handler = next(h for h in transport.opener.handlers if isinstance(h, HTTPRedirectHandler))
+        with self.assertRaisesRegex(module.Refused, "provider_redirect_refused"):
+            handler.redirect_request(Request("https://api.scaleway.com/"), None, 302, "",
+                                     {}, "https://example.invalid/")
+
+
+if __name__ == "__main__":
+    unittest.main()

workplans/RPF-WP-0053-coulomb-paired-backup-admission.md

@@ -319,3 +319,38 @@ regular recovery/drill credential. Retire the temporary session after native API
 verification. No standing organization-wide operator key or publisher IAM admin
 right follows from this instruction. First aborted attempt saved no state; fresh
 attended browser is waiting for user confirmation, with no provider mutation.
+
+## Prepared deterministic API identity bootstrap — 2026-10-28
+
+The first saved Console session could read the exact publisher metadata but
+expired before provisioning; API access now401 and renewal400. No project, policy
+or API key was created. The consumer records responsibility for the delay and
+will prepare the complete bootstrap before another login.
+
+New scripts/application_recovery_identities.py previews without effects and
+executes only through temporary auth on stdin. It binds the organization to the
+exact existing Coulomb publisher, creates an isolated recovery project, three
+separate applications (temporary construction, read-only archive reader,
+custodian), and four project-only policies including the publisher's mirror
+policy. No IAM/org-wide grants or replacement of the existing publisher pair/
+primary-bucket policy. The construction key has verified expiry at most two hours.
+The custodian gets explicit SecretManagerSecretAccess plus metadata read and
+limited object publication/read; archive reader has no secret access. Native
+bucket policy/escrow denials and protected custody remain mandatory after this
+identity stage; no archive, timer or operating acceptance follows automatically.
+
+Atomic owner-private output outside Git, fsync, effective-rule readback, explicit
+uncertain-issuance refusal, and own newly issued key revocation on persistence
+failure prevent silent duplicate/relabelled credentials. Retirement deletes only
+the exact verified construction key and its private temporary pair, retaining
+routine pairs and all backup material. Twelve synthetic cases pass; targeted
+platform Ruff and Node syntax checks pass. Native provisioning remains unperformed.
+
+The committed bootstrap_coulomb_recovery_session.mjs validates source/private
+output/locked Playwright before opening a browser, then executes the factory
+immediately after confirmation, reading current in-memory auth. It keeps the
+browser open for reviewed correction and API resource/boundary checks, then
+retires construction authority and logs out on finish. It saves no personal
+browser state. Run --preflight-only first; routine access uses the scoped API
+pairs, never that personal session. The expired earlier capture is not a
+credential lane. Existing independent recovery and service gates stay open.