849f3947dc
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
452 lines
23 KiB
Python
452 lines
23 KiB
Python
"""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 fcntl
|
|
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 canonical_policy(value):
|
|
# Policy array ordering is not significant. Scaleway sorts actions and
|
|
# principal arrays on write. Preserve every value (including duplicates)
|
|
# and compare all fields; never discard unknown permission/condition data.
|
|
if isinstance(value, dict):
|
|
return {k: canonical_policy(v) for k, v in value.items()}
|
|
if isinstance(value, list):
|
|
return sorted((canonical_policy(v) for v in value), key=lambda v: json.dumps(v, sort_keys=True))
|
|
return value
|
|
|
|
|
|
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 + "/*"]}}))
|
|
if prefix != "archives":
|
|
statements.append(statement(role + "Metadata", principal,
|
|
["s3:GetBucketLocation", "s3:GetBucketVersioning", "s3:GetLifecycleConfiguration"], bucket))
|
|
if prefix == "archives":
|
|
# Scaleway accepts at most ten statements per policy. These metadata
|
|
# rights are identical for all three callers; sharing their statement
|
|
# changes no effective permission or prefix restriction.
|
|
statements.append(statement("ArchiveMetadata", ["application_id:" + apps[r]
|
|
for r in ("reader", "custodian", "publisher")],
|
|
["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 deny_s3(self, role, operation, **arguments):
|
|
# Some denied PUTs leave a provider keep-alive connection unusable for
|
|
# the next request (observed HTTP400 with an empty error). Isolate each
|
|
# negative probe so it cannot corrupt subsequent positive read proof.
|
|
client = s3(self.pairs[role], self.project)
|
|
try:
|
|
return denied(lambda: getattr(client, operation)(**arguments))
|
|
finally:
|
|
client.close()
|
|
|
|
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 canonical_policy(live) != canonical_policy(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 canonical_policy(json.loads(client.get_bucket_policy(Bucket=bucket)["Policy"])) != canonical_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"):
|
|
# All deletion targets are newly made synthetic canaries. No
|
|
# actual retained backup is ever a destructive negative probe.
|
|
proofs[prefix + "_" + role + "_delete"] = self.deny_s3(role, "delete_object",
|
|
Bucket=bucket, Key=key, VersionId=version)
|
|
intended = self.state["buckets"][bucket]["policy"]
|
|
proofs[prefix + "_" + role + "_policy_write"] = self.deny_s3(role, "put_bucket_policy",
|
|
Bucket=bucket, Policy=json.dumps(intended))
|
|
proofs[prefix + "_" + role + "_outside_prefix"] = self.deny_s3(role, "get_object",
|
|
Bucket=bucket, Key="outside-approved-prefix/synthetic-absent")
|
|
proofs[prefix + "_reader_put"] = self.deny_s3("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"] = self.deny_s3("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)
|
|
private_directory(args.directory)
|
|
fd = os.open(args.directory / "resource-bootstrap.lock", os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW, 0o600)
|
|
with os.fdopen(fd, "w") as lock:
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
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)
|