"""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()