From 12f205627233fdcdc16eb32ccd35091af3b15d99 Mon Sep 17 00:00:00 2001 From: codex Date: Thu, 29 Oct 2026 21:52:55 +0100 Subject: [PATCH] Prepare scoped recovery resources and bounded construction renewal Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b --- scripts/application_recovery_identities.py | 17 + scripts/application_recovery_resources.py | 419 ++++++++++++++++++ tests/test_application_recovery_identities.py | 18 +- tests/test_application_recovery_resources.py | 66 +++ 4 files changed, 519 insertions(+), 1 deletion(-) create mode 100644 scripts/application_recovery_resources.py create mode 100644 tests/test_application_recovery_resources.py diff --git a/scripts/application_recovery_identities.py b/scripts/application_recovery_identities.py index 02eca25..5c7507c 100644 --- a/scripts/application_recovery_identities.py +++ b/scripts/application_recovery_identities.py @@ -222,6 +222,23 @@ class Bootstrap: def key(self, role, app, project): path = self.directory / (role + "-pair.json") + # A paused attended window can outlast the bounded construction key. + # Renew only this role, after exact own-key revocation/readback. Never + # replace reader/custodian/publisher credentials as a bootstrap retry. + prior = self.state["keys"].get(role, {}) + if role == "bootstrap" and prior: + expired = datetime.fromisoformat(prior["expires_at"].replace("Z", "+00:00")) <= self.now + if expired: + if path.exists(): + old = private_json(path) + actual = self.api("GET", "/iam/v1alpha1/api-keys/" + old["AWS_ACCESS_KEY_ID"]) + native_expiry = datetime.fromisoformat(actual["expires_at"].replace("Z", "+00:00")) + if native_expiry > self.now: + raise Refused("native_bootstrap_expiry_not_reached") + self.retire_bootstrap() + self.state.setdefault("retired_bootstrap_keys", []).append(dict(prior)) + self.state["keys"].pop(role) + self.save() if role in self.state["keys"]: pair = private_json(path) if set(pair) != {"AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"}: diff --git a/scripts/application_recovery_resources.py b/scripts/application_recovery_resources.py new file mode 100644 index 0000000..a28ac31 --- /dev/null +++ b/scripts/application_recovery_resources.py @@ -0,0 +1,419 @@ +"""Coulomb-only protected resources, exercised through scoped API pairs. + +Preview is offline. Execute never prints secret values or reads sibling custody. +Current Scaleway bucket policy version is an allowlist intersected with IAM. +""" +from __future__ import annotations + +import argparse +import base64 +import hashlib +import hmac +import json +import os +import subprocess +import sys +from datetime import UTC, datetime +from pathlib import Path +from urllib.error import HTTPError +from urllib.parse import urlencode +from urllib.request import HTTPRedirectHandler, Request, build_opener + +from application_recovery_identities import ( + DECISION, + PUBLISHER, + Refused, + checkpoint, + private_directory, + private_json, +) + +REGION = "nl-ams" +BUCKETS = {"kits": "railiance-coulomb-social-recovery", + "archives": "railiance-coulomb-social-archives"} +SECRET_BASE = "/secret-manager/v1beta1/regions/nl-ams/secrets" + + +class NativeError(Refused): + def __init__(self, status): + self.status = status + super().__init__("native_request_refused") + + +class API: + def __init__(self, pair): + self.key = pair["AWS_SECRET_ACCESS_KEY"] + + class NoRedirect(HTTPRedirectHandler): + def redirect_request(self, *args, **kwargs): + raise Refused("provider_redirect_refused") + + self.opener = build_opener(NoRedirect()) + + def __call__(self, method, path, value=None): + if not path.startswith((SECRET_BASE, "/iam/v1alpha1/")) or ".." in path: + raise Refused("exact_api_scope_required") + req = Request("https://api.scaleway.com" + path, method=method, + headers={"X-Auth-Token": self.key, "Content-Type": "application/json"}, + data=None if value is None else json.dumps(value).encode()) + try: + with self.opener.open(req, timeout=25) as response: + data = response.read(1048577) + if len(data) > 1048576: + raise Refused("bounded_response_required") + return json.loads(data) if data else {} + except HTTPError as error: + raise NativeError(error.code) from None + + +def s3(pair, project): + import boto3 + from botocore.config import Config + return boto3.client("s3", endpoint_url="https://s3.nl-ams.scw.cloud", region_name=REGION, + aws_access_key_id=pair["AWS_ACCESS_KEY_ID"].split("@")[0] + "@" + project, + aws_secret_access_key=pair["AWS_SECRET_ACCESS_KEY"], + config=Config(signature_version="s3v4", connect_timeout=15, read_timeout=30, + retries={"max_attempts": 0}, + request_checksum_calculation="when_required", + response_checksum_validation="when_required", + s3={"addressing_style": "path"})) + + +def status(error): + if isinstance(error, NativeError): + return error.status + return getattr(error, "response", {}).get("ResponseMetadata", {}).get("HTTPStatusCode") + + +def denied(call): + try: + result = call() + if isinstance(result, dict) and "Body" in result: + result["Body"].close() + except Exception as error: # noqa: BLE001 -- normalize SDK and HTTP denial evidence + if status(error) == 403: + return {"http_status": 403, "denied": True} + raise Refused("native_denial_must_be_403") from None + raise Refused("native_boundary_unexpectedly_allowed") + + +def policy(bucket, prefix, apps, owner): + def statement(sid, principal, actions, resources, condition=None): + value = {"Sid": sid, "Effect": "Allow", "Principal": {"SCW": principal}, + "Action": actions, "Resource": resources} + if condition: + value["Condition"] = condition + return value + # Preserve an explicit platform operator recovery route, even after the + # construction key is revoked. Owner is recorded from existing owned policy. + statements = [statement("PlatformOwner", "user_id:" + owner, "s3:*", [bucket, bucket + "/*"]), + statement("TemporaryConstruction", "application_id:" + apps["bootstrap"], + "s3:*", [bucket, bucket + "/*"])] + for role in ("reader", "custodian", *( ["publisher"] if prefix == "archives" else [])): + principal = "application_id:" + apps[role] + actions = ["s3:GetObject", "s3:GetObjectVersion"] + if role in {"custodian", "publisher"}: + actions += ["s3:PutObject"] + if role == "publisher": + actions += ["s3:AbortMultipartUpload", "s3:ListMultipartUploadParts"] + statements.append(statement(role + "Objects", principal, actions, bucket + "/" + prefix + "/*")) + statements.append(statement(role + "List", principal, ["s3:ListBucket", "s3:ListBucketVersions"], + bucket, {"StringLike": {"s3:prefix": [prefix + "/", prefix + "/*"]}})) + statements.append(statement(role + "Metadata", principal, + ["s3:GetBucketLocation", "s3:GetBucketVersioning", "s3:GetLifecycleConfiguration"], bucket)) + return {"Version": "2023-04-17", "Id": "coulomb-independent-recovery", "Statement": statements} + + +def public_recipient(identity): + proc = subprocess.run(["age-keygen", "-y", "/dev/stdin"], input=identity, + capture_output=True, timeout=15, check=False) + value = proc.stdout.decode().strip() + if proc.returncode or not value.startswith("age1"): + raise Refused("dedicated_age_identity_required") + return value + + +class Resources: + def __init__(self, directory, owner): + private_directory(directory) + self.directory, self.owner = directory, owner + self.ids = private_json(directory / "identity-state.json") + if self.ids.get("decision") != DECISION or self.ids.get("status") != ( + "identity_bootstrap_complete_native_boundaries_pending" + ): + raise Refused("verified_identity_stage_required") + self.project = self.ids["project"]["id"] + self.apps = {r: v["id"] for r, v in self.ids["apps"].items()} | {"publisher": PUBLISHER} + self.pairs = {r: private_json(directory / (r + "-pair.json")) + for r in ("bootstrap", "reader", "custodian", "publisher")} + self.api = {r: API(p) for r, p in self.pairs.items()} + self.clients = {r: s3(p, self.project) for r, p in self.pairs.items()} + self.path = directory / "resource-state.json" + self.state = private_json(self.path) if self.path.exists() else { + "schema": "platform.coulomb-recovery-resources.v1", "decision": DECISION, + "project_id": self.project, "owner_user_id": owner, "buckets": {}, "secrets": {}} + if (self.state.get("project_id") != self.project + or self.state.get("owner_user_id") != owner or self.state.get("decision") != DECISION): + raise Refused("resource_journal_binding_refused") + if self.state.get("pending"): + raise Refused("uncertain_resource_mutation_requires_reconciliation") + + def save(self): + checkpoint(self.path, self.state) + + def mutate(self, label, call): + self.state["pending"] = label + self.save() + try: + result = call() + except Exception as error: + if status(error) in {400, 401, 403, 404, 409, 412, 422, 429}: + self.state.pop("pending") + self.save() + raise + self.state.pop("pending") + return result + + def bucket(self, prefix, bucket): + client = self.clients["bootstrap"] + if bucket not in self.state["buckets"]: + inventory = client.list_buckets()["Buckets"] + if any(v["Name"] == bucket for v in inventory): + raise Refused("untracked_bucket_requires_review") + self.mutate("create_bucket:" + bucket, lambda: client.create_bucket( + Bucket=bucket, ACL="private", CreateBucketConfiguration={"LocationConstraint": REGION})) + self.state["buckets"][bucket] = {"created": True} + self.save() + # No payload is ever placed before allowlist installation/readback. + intended = policy(bucket, prefix, self.apps, self.owner) + try: + live = json.loads(client.get_bucket_policy(Bucket=bucket)["Policy"]) + except Exception as error: + if status(error) != 404: + raise + live = None + if live is not None and live != intended: + raise Refused("existing_policy_drift_requires_review") + if live is None: + self.mutate("put_policy:" + bucket, lambda: client.put_bucket_policy( + Bucket=bucket, Policy=json.dumps(intended))) + self.save() + if json.loads(client.get_bucket_policy(Bucket=bucket)["Policy"]) != intended: + raise Refused("policy_readback_differs") + if client.get_bucket_versioning(Bucket=bucket).get("Status") != "Enabled": + self.mutate("versioning:" + bucket, lambda: client.put_bucket_versioning( + Bucket=bucket, VersioningConfiguration={"Status": "Enabled"})) + self.save() + acl = client.get_bucket_acl(Bucket=bucket) + if any(g.get("Grantee", {}).get("URI") for g in acl.get("Grants", [])): + raise Refused("public_acl_refused") + try: + lifecycle = client.get_bucket_lifecycle_configuration(Bucket=bucket)["Rules"] + except Exception as error: + if status(error) != 404: + raise + lifecycle = [] + # Retention compatibility has not yet been proved. Do not activate + # expiry even on rolling bucket until protected recovery exists. + if lifecycle: + raise Refused("unexpected_expiry_requires_review") + if client.get_bucket_versioning(Bucket=bucket).get("Status") != "Enabled": + raise Refused("versioning_readback_required") + self.state["buckets"][bucket] = {"created": True, "versioning": "Enabled", + "public_acl_absent": True, "lifecycle_rules": [], "policy": intended} + self.save() + + def secret(self, name): + api = self.api["bootstrap"] + if name not in self.state["secrets"]: + found = api("GET", SECRET_BASE + "?" + urlencode({"project_id": self.project, "page_size": 100})) + if int(found["total_count"]) != len(found["secrets"]): + raise Refused("complete_secret_inventory_required") + if any(s["name"] == name for s in found["secrets"]): + raise Refused("untracked_secret_requires_review") + created = self.mutate("create_secret:" + name, lambda: api("POST", SECRET_BASE, + {"name": name, "project_id": self.project, "type": "opaque", "protected": True, + "description": "Coulomb-only independent recovery; " + DECISION})) + self.state["secrets"][name] = {"id": created["id"]} + self.save() + ref = self.state["secrets"][name] + metadata = api("GET", SECRET_BASE + "/" + ref["id"]) + if (metadata.get("name") != name or metadata.get("project_id") != self.project + or metadata.get("type") != "opaque" or metadata.get("protected") is not True + or metadata.get("ephemeral_policy") is not None + or metadata.get("deletion_requested_at") is not None): + raise Refused("protected_nonexpiring_secret_required") + return ref + + def version(self, name, value): + ref = self.secret(name) + path = SECRET_BASE + "/" + ref["id"] + if not ref.get("revision"): + versions = self.api["bootstrap"]("GET", path + "/versions?page_size=100") + if int(versions["total_count"]) != 0: + raise Refused("untracked_secret_revision_requires_review") + created = self.mutate("create_version:" + name, lambda: self.api["bootstrap"]( + "POST", path + "/versions", {"data": base64.b64encode(value).decode(), + "description": "Protected Coulomb custody", "disable_previous": False})) + ref["revision"] = created["revision"] + self.save() + revision = str(ref["revision"]) + current = self.api["custodian"]("GET", path + "/versions/" + revision) + props = current.get("ephemeral_properties") or {} + if current.get("status") != "enabled" or props.get("expires_at") or props.get("expires_once_accessed"): + raise Refused("enabled_nonexpiring_revision_required") + actual = self.api["custodian"]("GET", path + "/versions/" + revision + "/access") + if not hmac.compare_digest(base64.b64decode(actual["data"], validate=True), value): + raise Refused("independent_secret_readback_differs") + return ref + + def run(self): + for prefix, bucket in BUCKETS.items(): + self.bucket(prefix, bucket) + # Canary is deliberately synthetic: denial checks cannot expose an age + # identity or a real API pair, even if IAM were accidentally too broad. + canary_name = "coulomb-social-recovery-access-canary" + canary = self.version(canary_name, b"synthetic-coulomb-recovery-canary-v1") + probe_path = SECRET_BASE + "/" + canary["id"] + "/versions/" + str(canary["revision"]) + "/access" + boundaries = {r + "_escrow": denied(lambda r=r: self.api[r]("GET", probe_path)) + for r in ("reader", "publisher")} + identity_path = self.directory / "coulomb.agekey" + if not identity_path.exists(): + proc = subprocess.run(["age-keygen"], capture_output=True, timeout=15, check=False) + if proc.returncode: + raise Refused("age_generation_failed") + fd = os.open(identity_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "wb") as stream: + stream.write(proc.stdout) + stream.flush() + os.fsync(stream.fileno()) + # Validate mode/ownership without JSON parsing. + info = identity_path.lstat() + if not identity_path.is_file() or identity_path.is_symlink() or info.st_uid != os.geteuid() or info.st_mode & 0o077: + raise Refused("private_age_identity_required") + identity = identity_path.read_bytes() + recipient = public_recipient(identity) + self.version("coulomb-social-age-recovery", identity) + for role in ("reader", "custodian"): + self.version("coulomb-social-recovery-" + role + "-api", json.dumps( + self.pairs[role], sort_keys=True).encode()) + boundaries.update(self.object_boundaries()) + self.state.update(status="scoped_api_custody_and_native_boundaries_verified", + public_recipient=recipient, secret_boundaries=boundaries, + completed_at=datetime.now(UTC).isoformat(), + native_archive_recovery_verified=False, timers_activated=False) + self.save() + return {"status": self.state["status"], "buckets": len(self.state["buckets"]), + "protected_secrets": len(self.state["secrets"]), "public_recipient": recipient, + "routine_pairs_escrow_readback_equal": True, "secret_boundaries": boundaries, + "private_values_printed": False, "timers_activated": False} + + def fixture(self, prefix): + bucket = BUCKETS[prefix] + key = prefix + "/bootstrap-boundary-v1.txt" + content = b"synthetic Coulomb recovery boundary fixture v1\n" + ref = self.state.setdefault("fixtures", {}).get(prefix) + if not ref: + response = self.mutate("put_fixture:" + prefix, lambda: self.clients["custodian"].put_object( + Bucket=bucket, Key=key, Body=content, IfNoneMatch="*")) + ref = {"bucket": bucket, "key": key, "version_id": response.get("VersionId")} + self.state["fixtures"][prefix] = ref + self.save() + if not ref["version_id"] or ref["version_id"] == "null": + raise Refused("versioned_fixture_required") + response = self.clients["reader"].get_object(Bucket=bucket, Key=key, VersionId=ref["version_id"]) + with response["Body"] as stream: + actual = stream.read(len(content) + 1) + if (actual != content or response.get("VersionId") != ref["version_id"]): + raise Refused("independent_version_readback_differs") + ref["sha256"] = hashlib.sha256(content).hexdigest() + self.save() + return ref + + def object_boundaries(self): + proofs = {} + for prefix, bucket in BUCKETS.items(): + ref = self.fixture(prefix) + key, version = ref["key"], ref["version_id"] + reader = self.clients["reader"] + listing = reader.list_object_versions(Bucket=bucket, Prefix=prefix + "/") + if not any(v["Key"] == key and v["VersionId"] == version for v in listing.get("Versions", [])): + raise Refused("independent_version_listing_required") + proofs[prefix + "_reader_version_get_hash"] = {"verified": True, "version_id": version} + for role in ("reader", "custodian", "publisher"): + client = self.clients[role] + # All deletion targets are newly made synthetic canaries. No + # actual retained backup is ever a destructive negative probe. + proofs[prefix + "_" + role + "_delete"] = denied(lambda client=client, bucket=bucket, key=key, version=version: client.delete_object( + Bucket=bucket, Key=key, VersionId=version)) + intended = self.state["buckets"][bucket]["policy"] + proofs[prefix + "_" + role + "_policy_write"] = denied(lambda client=client, bucket=bucket, intended=intended: client.put_bucket_policy( + Bucket=bucket, Policy=json.dumps(intended))) + proofs[prefix + "_" + role + "_outside_prefix"] = denied(lambda client=client, bucket=bucket: client.get_object( + Bucket=bucket, Key="outside-approved-prefix/synthetic-absent")) + proofs[prefix + "_reader_put"] = denied(lambda reader=reader, bucket=bucket, prefix=prefix: reader.put_object( + Bucket=bucket, Key=prefix + "/reader-write-denial.txt", Body=b"synthetic-only")) + kit = self.state["fixtures"]["kits"] + proofs["publisher_protected_kit_get"] = denied(lambda: self.clients["publisher"].get_object( + Bucket=kit["bucket"], Key=kit["key"], VersionId=kit["version_id"])) + publisher = self.clients["publisher"] + # Verify ordinary worker upload/readback in addition to its multipart + # abort. No content is captured from the application by this helper. + key = "archives/publisher-boundary-v1.txt" + payload = b"synthetic publisher fixture v1\n" + published = self.state.get("publisher_fixture") + if not published: + response = self.mutate("publisher_fixture", lambda: publisher.put_object( + Bucket=BUCKETS["archives"], Key=key, Body=payload, IfNoneMatch="*")) + published = {"version_id": response.get("VersionId")} + self.state["publisher_fixture"] = published + self.save() + if not published["version_id"] or published["version_id"] == "null": + raise Refused("versioned_publisher_fixture_required") + response = publisher.get_object(Bucket=BUCKETS["archives"], Key=key, VersionId=published["version_id"]) + with response["Body"] as stream: + if stream.read(len(payload) + 1) != payload: + raise Refused("publisher_version_readback_differs") + proofs["publisher_create_readback"] = {"verified": True} + upload = self.state.get("multipart_fixture") + if not upload: + upload = self.mutate("multipart_fixture", lambda: publisher.create_multipart_upload( + Bucket=BUCKETS["archives"], Key="archives/multipart-abort-boundary-v1")) + self.state["multipart_fixture"] = {"upload_id": upload["UploadId"], "aborted": False} + self.save() + multipart = self.state["multipart_fixture"] + if not multipart["aborted"]: + self.mutate("multipart_abort", lambda: publisher.abort_multipart_upload( + Bucket=BUCKETS["archives"], Key="archives/multipart-abort-boundary-v1", UploadId=multipart["upload_id"])) + multipart["aborted"] = True + self.save() + proofs["publisher_own_multipart_abort"] = {"verified": True} + for role in ("publisher", "reader", "custodian"): + proofs[role + "_iam_admin"] = denied(lambda role=role: self.api[role]( + "GET", "/iam/v1alpha1/applications?organization_id=" + self.ids["project"]["organization_id"])) + return proofs + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--execute", action="store_true") + parser.add_argument("--directory", type=Path) + parser.add_argument("--owner-user-id") + args = parser.parse_args() + if not args.execute: + print(json.dumps({"status": "preview", "provider_contacted": False, + "buckets": BUCKETS, "automatic_expiry": False})) + return + import uuid + uuid.UUID(args.owner_user_id) + print(json.dumps(Resources(args.directory, args.owner_user_id).run())) + + +if __name__ == "__main__": + try: + main() + except Exception as error: # noqa: BLE001 -- never expose SDK credential-bearing diagnostics + print(json.dumps({"status": "refused", "failure_type": type(error).__name__, + "native_status": status(error), + "reason": str(error) if isinstance(error, Refused) else "private_review_required"})) + sys.exit(1) diff --git a/tests/test_application_recovery_identities.py b/tests/test_application_recovery_identities.py index 8f5067d..f764b53 100644 --- a/tests/test_application_recovery_identities.py +++ b/tests/test_application_recovery_identities.py @@ -6,7 +6,7 @@ import json import os import tempfile import unittest -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from pathlib import Path from unittest.mock import patch from urllib.parse import parse_qs, urlsplit @@ -30,6 +30,7 @@ class Provider: 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))) @@ -56,6 +57,9 @@ class Provider: 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: @@ -129,6 +133,18 @@ class IdentityBootstrap(unittest.TestCase): 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"): diff --git a/tests/test_application_recovery_resources.py b/tests/test_application_recovery_resources.py new file mode 100644 index 0000000..a864a96 --- /dev/null +++ b/tests/test_application_recovery_resources.py @@ -0,0 +1,66 @@ +import importlib.util +import sys +import unittest +from pathlib import Path + +SCRIPTS = Path(__file__).resolve().parents[1] / "scripts" +sys.path.insert(0, str(SCRIPTS)) +spec = importlib.util.spec_from_file_location("resources", SCRIPTS / "application_recovery_resources.py") +m = importlib.util.module_from_spec(spec) +spec.loader.exec_module(m) + + +class ResourcesTests(unittest.TestCase): + def setUp(self): + self.apps = {r: "synthetic-" + r for r in ("bootstrap", "reader", "custodian", "publisher")} + + def grants(self, prefix, role): + policy = m.policy("synthetic-bucket", prefix, self.apps, "synthetic-owner") + return [s for s in policy["Statement"] + if s["Principal"]["SCW"] == "application_id:" + self.apps[role]] + + def test_reader_only_prefixed_read_and_list(self): + rows = self.grants("kits", "reader") + actions = {a for r in rows for a in r["Action"]} + self.assertTrue({"s3:GetObject", "s3:GetObjectVersion", "s3:ListBucketVersions"} <= actions) + self.assertFalse(any("Put" in a or "Delete" in a or a == "s3:*" for a in actions)) + objects = [r for r in rows if "s3:GetObject" in r["Action"]] + self.assertEqual(objects[0]["Resource"], "synthetic-bucket/kits/*") + listing = next(r for r in rows if "s3:ListBucket" in r["Action"]) + self.assertEqual(listing["Condition"]["StringLike"]["s3:prefix"], ["kits/", "kits/*"]) + + def test_publisher_has_no_kit_grant(self): + self.assertEqual(self.grants("kits", "publisher"), []) + + def test_publisher_can_abort_without_delete(self): + rows = self.grants("archives", "publisher") + actions = {a for r in rows for a in r["Action"]} + self.assertIn("s3:AbortMultipartUpload", actions) + self.assertIn("s3:PutObject", actions) + self.assertFalse(any("Delete" in a or "Policy" in a for a in actions)) + + def test_custodian_no_retained_delete_or_administration(self): + rows = self.grants("kits", "custodian") + actions = {a for r in rows for a in r["Action"]} + self.assertIn("s3:PutObject", actions) + self.assertFalse(any("Delete" in a or "Policy" in a or a == "s3:*" for a in actions)) + + def test_denial_requires_403_not_404_or_transport_error(self): + def failing(code): + raise m.NativeError(code) + self.assertTrue(m.denied(lambda: failing(403))["denied"]) + for code in (400, 401, 404, 500): + with self.assertRaises(m.Refused): + m.denied(lambda code=code: failing(code)) + with self.assertRaises(m.Refused): + m.denied(dict) + + def test_bucket_policy_preserves_owner_and_temporary_constructor(self): + p = m.policy("b", "kits", self.apps, "o") + self.assertEqual(p["Version"], "2023-04-17") + self.assertEqual(p["Statement"][0]["Principal"], {"SCW": "user_id:o"}) + self.assertTrue(all(s["Effect"] == "Allow" for s in p["Statement"])) + + +if __name__ == "__main__": + unittest.main()