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