tegwick
9a2cf213d1
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a11fd9-caf6-7a63-b578-9daaaa134ff2
120 lines
5.1 KiB
Python
120 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Attended age escrow to a protected Scaleway Secret Manager secret, no key output."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import hmac
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import uuid
|
|
from pathlib import Path
|
|
from urllib.request import Request, urlopen
|
|
|
|
from ledger_archive import Refused, checkpoint, private_directory, private_file
|
|
|
|
NAME = "email-connect-ledger-age-recovery"
|
|
REGION = "nl-ams"
|
|
RECIPIENT = "age1zvryunvjhvpkmasskauga2heeg0ztnte9ymgppvjge36ekumk50syr3tsz"
|
|
BASE = "https://api.scaleway.com/secret-manager/v1beta1/regions/nl-ams/secrets/"
|
|
|
|
|
|
def request(secret_id, suffix="", payload=None):
|
|
key = os.environ.get("SCW_SECRET_KEY", "")
|
|
if not key:
|
|
raise Refused("attended_scaleway_identity_required")
|
|
req = Request(BASE + secret_id + suffix,
|
|
data=None if payload is None else json.dumps(payload).encode(),
|
|
headers={"X-Auth-Token": key, "Content-Type": "application/json"},
|
|
method="GET" if payload is None else "POST")
|
|
with urlopen(req, timeout=30) as response:
|
|
return json.loads(response.read(262144))
|
|
|
|
|
|
def recipient(identity):
|
|
result = subprocess.run(["age-keygen", "-y", "/dev/stdin"], input=identity,
|
|
capture_output=True, timeout=30)
|
|
if result.returncode:
|
|
raise Refused("age_identity_invalid")
|
|
return result.stdout.decode().strip()
|
|
|
|
|
|
def escrow(secret_id, project_id, identity, transport=request):
|
|
uuid.UUID(secret_id)
|
|
uuid.UUID(project_id)
|
|
if recipient(identity) != RECIPIENT or not 0 < len(identity) <= 65536:
|
|
raise Refused("configured_age_recipient_required")
|
|
metadata = transport(secret_id)
|
|
if (metadata.get("name") != NAME or metadata.get("project_id") != project_id
|
|
or metadata.get("region") != REGION 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_exact_secret_required")
|
|
versions = transport(secret_id, "/versions?page_size=100")
|
|
count = versions.get("total_count")
|
|
if type(count) is not int or count != len(versions.get("versions", [])) or count > 1:
|
|
raise Refused("existing_versions_require_custodian_review")
|
|
if count == 0:
|
|
written = transport(secret_id, "/versions", {
|
|
"data": base64.b64encode(identity).decode(),
|
|
"description": "Existing email-connect age identity; attended recovery only.",
|
|
"disable_previous": False,
|
|
})
|
|
revision = written["revision"]
|
|
if revision != 1:
|
|
raise Refused("initial_secret_version_required")
|
|
else:
|
|
version = versions["versions"][0]
|
|
properties = version.get("ephemeral_properties") or {}
|
|
if (version.get("status") != "enabled" or properties.get("expires_at") is not None
|
|
or properties.get("expires_once_accessed") is True
|
|
or version.get("revision") != 1):
|
|
raise Refused("nonexpiring_enabled_version_required")
|
|
revision = 1
|
|
returned = transport(secret_id, "/versions/" + str(revision) + "/access")
|
|
restored = base64.b64decode(returned["data"], validate=True)
|
|
if (returned.get("secret_id") != secret_id or returned.get("revision") != revision
|
|
or not hmac.compare_digest(restored, identity) or recipient(restored) != RECIPIENT):
|
|
raise Refused("independent_escrow_readback_differs")
|
|
return {"status": "scaleway_age_escrow_verified", "secret_id": secret_id,
|
|
"revision": revision, "region": REGION, "public_recipient_matches": True,
|
|
"readback_equal": True, "private_identity_printed": False,
|
|
"independent_account_login_verified": False,
|
|
"worker_access_denial_verified": False}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--execute", action="store_true")
|
|
parser.add_argument("--secret-id")
|
|
parser.add_argument("--project-id")
|
|
parser.add_argument("--identity", type=Path)
|
|
parser.add_argument("--receipt", type=Path)
|
|
args = parser.parse_args()
|
|
if not args.execute:
|
|
print(json.dumps({"status": "prepared", "name": NAME, "region": REGION,
|
|
"protected": True, "ephemeral_policy": None,
|
|
"provider_contacted": False, "private_identity_read": False}))
|
|
return
|
|
if not args.identity or not args.receipt or not args.receipt.is_absolute():
|
|
raise Refused("private_identity_and_receipt_paths_required")
|
|
private_directory(args.receipt.parent)
|
|
private_file(args.identity)
|
|
if args.receipt.exists() or args.receipt.is_symlink():
|
|
raise Refused("fresh_receipt_required")
|
|
try:
|
|
result = escrow(args.secret_id, args.project_id, args.identity.read_bytes())
|
|
except Exception as error:
|
|
checkpoint(args.receipt, {"status": "failed", "failure_type": type(error).__name__})
|
|
raise
|
|
checkpoint(args.receipt, result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception:
|
|
raise SystemExit(1) from None
|