diff --git a/scripts/application_recovery_identities.py b/scripts/application_recovery_identities.py new file mode 100644 index 0000000..02eca25 --- /dev/null +++ b/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 diff --git a/scripts/bootstrap_coulomb_recovery_session.mjs b/scripts/bootstrap_coulomb_recovery_session.mjs new file mode 100644 index 0000000..48eb6cc --- /dev/null +++ b/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(){ + 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(); +} diff --git a/tests/test_application_recovery_identities.py b/tests/test_application_recovery_identities.py new file mode 100644 index 0000000..8f5067d --- /dev/null +++ b/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() diff --git a/workplans/RPF-WP-0053-coulomb-paired-backup-admission.md b/workplans/RPF-WP-0053-coulomb-paired-backup-admission.md index da815bc..9285a8d 100644 --- a/workplans/RPF-WP-0053-coulomb-paired-backup-admission.md +++ b/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.