12f2056272
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
239 lines
12 KiB
Python
239 lines
12 KiB
Python
"""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, timedelta
|
|
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
|
|
self.issued_keys = 0
|
|
|
|
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":
|
|
self.issued_keys += 1
|
|
identity = suffix + "-" + str(self.issued_keys)
|
|
created["id"] = identity
|
|
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_expired_construction_key_renews_only_its_owned_role(self):
|
|
self.run_bootstrap()
|
|
routine = {r: (self.directory / (r + "-pair.json")).read_bytes()
|
|
for r in ("reader", "custodian")}
|
|
prior = json.loads((self.directory / "bootstrap-pair.json").read_text())["AWS_ACCESS_KEY_ID"]
|
|
result = module.Bootstrap(self.api, self.directory, NOW + timedelta(days=1)).run()
|
|
self.assertEqual(result["status"], "identity_bootstrap_complete_native_boundaries_pending")
|
|
self.assertNotIn(prior, self.api.keys)
|
|
self.assertEqual(len(self.api.keys), 3)
|
|
for role, value in routine.items():
|
|
self.assertEqual((self.directory / (role + "-pair.json")).read_bytes(), value)
|
|
|
|
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()
|