#!/usr/bin/env python3
"""Shared fail-closed primitives for the SigNoz metadata backup lane.

The contract intentionally works only through authenticated HTTP APIs.  It
never opens the SigNoz SQLite database or a Docker volume and never accepts a
secret value on the command line.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import stat
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


REPO_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_POLICY = REPO_ROOT / "config" / "signoz" / "metadata-export-policy.json"
SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
SAFE_ASSET_ID = re.compile(r"^[a-z][a-z0-9_]{0,63}$")
SHA256 = re.compile(r"^[0-9a-f]{64}$")
ALLOWED_TOKEN_MODES = {0o400, 0o600}
LOOPBACK_HOSTS = {"127.0.0.1", "::1", "localhost"}


class ContractError(ValueError):
    """A bounded policy, transport, or artifact contract failure."""


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(
        self,
        req: urllib.request.Request,
        fp: Any,
        code: int,
        msg: str,
        headers: Any,
        newurl: str,
    ) -> None:
        return None


def utc_now() -> str:
    return (
        datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
    )


def canonical_json_bytes(value: Any) -> bytes:
    return (
        json.dumps(
            value,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
        )
        + "\n"
    ).encode("utf-8")


def sha256_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def read_json_object(path: Path, *, label: str) -> dict[str, Any]:
    require_no_symlink_components(path, label=label)
    try:
        metadata = path.stat()
        if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > 1024 * 1024:
            raise ContractError(f"{label}_size_or_type_invalid")
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
        raise ContractError(f"{label}_read_failed") from exc
    if not isinstance(value, dict):
        raise ContractError(f"{label}_not_object")
    return value


def _normalized_key(value: str) -> str:
    return re.sub(r"[^a-z0-9]", "", value.casefold())


def require_no_symlink_components(path: Path, *, label: str) -> None:
    if not path.is_absolute():
        path = Path.cwd() / path
    current = path
    while True:
        try:
            metadata = current.lstat()
        except OSError as exc:
            raise ContractError(f"{label}_path_component_unavailable") from exc
        if stat.S_ISLNK(metadata.st_mode):
            raise ContractError(f"{label}_symlink_component_forbidden")
        if current == current.parent:
            break
        current = current.parent


def load_policy(path: Path) -> dict[str, Any]:
    policy = read_json_object(path, label="policy")
    if policy.get("schema") != "awoooi_signoz_metadata_export_policy_v1":
        raise ContractError("unsupported_policy_schema")

    runtime = policy.get("source_runtime")
    if not isinstance(runtime, dict):
        raise ContractError("policy_source_runtime_missing")
    if runtime.get("raw_database_access_allowed") is not False:
        raise ContractError("policy_raw_database_must_be_forbidden")
    if runtime.get("raw_volume_access_allowed") is not False:
        raise ContractError("policy_raw_volume_must_be_forbidden")
    if runtime.get("github_supply_chain_allowed") is not False:
        raise ContractError("policy_github_supply_chain_must_be_forbidden")
    if runtime.get("minimum_python_version") != "3.10":
        raise ContractError("policy_minimum_python_version_invalid")

    limits = policy.get("limits")
    if not isinstance(limits, dict):
        raise ContractError("policy_limits_missing")
    for key in (
        "request_timeout_seconds",
        "max_response_bytes",
        "max_asset_count",
        "max_items_per_asset",
        "max_restore_actions",
        "stable_read_count",
    ):
        value = limits.get(key)
        if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
            raise ContractError(f"policy_limit_invalid_{key}")
    if limits["stable_read_count"] != 2:
        raise ContractError("policy_requires_exactly_two_stable_reads")

    authentication = policy.get("authentication")
    if not isinstance(authentication, dict):
        raise ContractError("policy_authentication_missing")
    if authentication.get("header_name") != "SIGNOZ-API-KEY":
        raise ContractError("policy_auth_header_not_fixed")
    if authentication.get("secret_value_in_arguments_allowed") is not False:
        raise ContractError("policy_secret_argument_must_be_forbidden")
    if authentication.get("accepted_file_modes") != ["0400", "0600"]:
        raise ContractError("policy_credential_modes_invalid")

    assets = policy.get("assets")
    if not isinstance(assets, list) or not assets:
        raise ContractError("policy_assets_missing")
    if len(assets) > limits["max_asset_count"]:
        raise ContractError("policy_asset_count_exceeds_limit")

    forbidden_paths = {
        item.get("path")
        for item in policy.get("forbidden_endpoints", [])
        if isinstance(item, dict)
    }
    asset_ids: set[str] = set()
    asset_paths: set[str] = set()
    for asset in assets:
        if not isinstance(asset, dict):
            raise ContractError("policy_asset_not_object")
        asset_id = asset.get("id")
        endpoint = asset.get("endpoint")
        if not isinstance(asset_id, str) or not SAFE_ASSET_ID.fullmatch(asset_id):
            raise ContractError("policy_asset_id_unsafe")
        if asset_id in asset_ids:
            raise ContractError(f"policy_asset_duplicate_{asset_id}")
        asset_ids.add(asset_id)
        if not isinstance(endpoint, str) or not endpoint.startswith("/api/v1/"):
            raise ContractError(f"policy_asset_endpoint_unsafe_{asset_id}")
        if "?" in endpoint or "#" in endpoint or ".." in endpoint:
            raise ContractError(f"policy_asset_endpoint_unsafe_{asset_id}")
        if endpoint in forbidden_paths:
            raise ContractError(f"policy_forbidden_endpoint_allowlisted_{asset_id}")
        if endpoint in asset_paths:
            raise ContractError(f"policy_asset_endpoint_duplicate_{asset_id}")
        asset_paths.add(endpoint)
        if asset.get("method") != "GET":
            raise ContractError(f"policy_asset_export_must_be_get_{asset_id}")
        if asset.get("response_kind") not in {"collection", "document"}:
            raise ContractError(f"policy_asset_response_kind_invalid_{asset_id}")
        if asset.get("classification") not in {"portable", "export_only"}:
            raise ContractError(f"policy_asset_classification_invalid_{asset_id}")
        if asset.get("classification") == "portable":
            restore = asset.get("restore")
            if not isinstance(restore, dict):
                raise ContractError(f"policy_restore_missing_{asset_id}")
            if restore.get("method") not in {"POST", "PUT", "PATCH"}:
                raise ContractError(f"policy_restore_method_invalid_{asset_id}")
            if restore.get("strategy") != "collection_items":
                raise ContractError(f"policy_restore_strategy_invalid_{asset_id}")
            if restore.get("endpoint") != endpoint:
                raise ContractError(f"policy_restore_endpoint_mismatch_{asset_id}")
            if asset.get("response_kind") != "collection":
                raise ContractError(f"policy_portable_asset_not_collection_{asset_id}")

    completion = policy.get("completion_contract")
    if not isinstance(completion, dict):
        raise ContractError("policy_completion_contract_missing")
    if completion.get("portable_export_alone_is_completion") is not False:
        raise ContractError("policy_export_must_not_claim_completion")
    if completion.get("full_sqlite_completion_claim_allowed") is not False:
        raise ContractError("policy_full_sqlite_claim_must_be_forbidden")
    if completion.get("durable_terminal_receipt_required_for_apply") is not True:
        raise ContractError("policy_durable_apply_receipt_must_be_required")
    if completion.get("failure_terminal_receipt_required") is not True:
        raise ContractError("policy_failure_receipt_must_be_required")
    restore_isolation = policy.get("restore_isolation")
    if not isinstance(restore_isolation, dict):
        raise ContractError("policy_restore_isolation_missing")
    if restore_isolation.get("image_id_sha256_required") is not True:
        raise ContractError("policy_restore_image_id_must_be_required")

    forbidden_names = policy.get("forbidden_key_names")
    if not isinstance(forbidden_names, list) or not forbidden_names:
        raise ContractError("policy_forbidden_keys_missing")
    normalized = [_normalized_key(str(item)) for item in forbidden_names]
    if not all(normalized) or len(set(normalized)) != len(normalized):
        raise ContractError("policy_forbidden_keys_invalid")
    for pattern in policy.get("forbidden_value_patterns", []):
        try:
            re.compile(str(pattern))
        except re.error as exc:
            raise ContractError("policy_forbidden_value_pattern_invalid") from exc
    return policy


def validate_identity(value: str, *, label: str) -> str:
    if not SAFE_ID.fullmatch(value):
        raise ContractError(f"{label}_unsafe")
    return value


def validate_base_url(value: str, *, loopback_only: bool = False) -> str:
    parsed = urllib.parse.urlsplit(value)
    if parsed.username or parsed.password or parsed.query or parsed.fragment:
        raise ContractError("base_url_contains_forbidden_components")
    if parsed.path not in {"", "/"}:
        raise ContractError("base_url_path_must_be_empty")
    host = (parsed.hostname or "").casefold()
    is_loopback = host in LOOPBACK_HOSTS
    if loopback_only and not is_loopback:
        raise ContractError("restore_target_must_be_loopback")
    if parsed.scheme == "http" and not is_loopback:
        raise ContractError("plain_http_allowed_only_for_loopback")
    if parsed.scheme not in {"http", "https"} or not host:
        raise ContractError("base_url_scheme_or_host_invalid")
    try:
        parsed.port
    except ValueError as exc:
        raise ContractError("base_url_port_invalid") from exc
    return value.rstrip("/")


def validate_token_reference(path: Path, *, read_value: bool) -> str | None:
    if not path.is_absolute():
        raise ContractError("credential_reference_must_be_absolute")
    require_no_symlink_components(path, label="credential_reference")
    try:
        metadata = path.lstat()
    except OSError as exc:
        raise ContractError("credential_reference_unavailable") from exc
    if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
        raise ContractError("credential_reference_not_regular_file")
    if metadata.st_uid != os.geteuid():
        raise ContractError("credential_reference_owner_mismatch")
    if stat.S_IMODE(metadata.st_mode) not in ALLOWED_TOKEN_MODES:
        raise ContractError("credential_reference_mode_must_be_0400_or_0600")
    if metadata.st_size <= 0 or metadata.st_size > 4096:
        raise ContractError("credential_reference_size_invalid")
    if not read_value:
        return None
    try:
        raw = path.read_bytes()
    except OSError as exc:
        raise ContractError("credential_reference_read_failed") from exc
    token = raw.rstrip(b"\r\n")
    if not token or len(token) > 4096 or any(byte < 33 or byte > 126 for byte in token):
        raise ContractError("credential_reference_value_shape_invalid")
    return token.decode("ascii")


def scan_forbidden(value: Any, policy: dict[str, Any]) -> None:
    forbidden_keys = {
        _normalized_key(str(item)) for item in policy["forbidden_key_names"]
    }
    value_patterns = [
        re.compile(str(item)) for item in policy.get("forbidden_value_patterns", [])
    ]

    def walk(item: Any, pointer: str) -> None:
        if isinstance(item, dict):
            for key, child in item.items():
                if not isinstance(key, str):
                    raise ContractError("json_object_key_not_string")
                if _normalized_key(key) in forbidden_keys:
                    raise ContractError(
                        f"forbidden_key_detected_at_{pointer or 'root'}"
                    )
                walk(child, f"{pointer}/{key}")
        elif isinstance(item, list):
            for index, child in enumerate(item):
                walk(child, f"{pointer}/{index}")
        elif isinstance(item, str):
            if any(pattern.search(item) for pattern in value_patterns):
                raise ContractError(
                    f"forbidden_value_pattern_detected_at_{pointer or 'root'}"
                )

    walk(value, "")


def resolve_pointer(value: Any, pointer: str) -> Any:
    if pointer == "":
        return value
    if not pointer.startswith("/"):
        raise ContractError("response_pointer_invalid")
    current = value
    for raw_part in pointer[1:].split("/"):
        part = raw_part.replace("~1", "/").replace("~0", "~")
        if isinstance(current, dict) and part in current:
            current = current[part]
        else:
            raise ContractError("response_pointer_not_found")
    return current


def validate_asset_document(
    value: Any,
    asset: dict[str, Any],
    policy: dict[str, Any],
) -> int:
    scan_forbidden(value, policy)
    selected = resolve_pointer(value, str(asset.get("response_pointer", "")))
    if asset["response_kind"] == "collection":
        if not isinstance(selected, list):
            raise ContractError(f"asset_collection_shape_invalid_{asset['id']}")
        if len(selected) > policy["limits"]["max_items_per_asset"]:
            raise ContractError(f"asset_item_limit_exceeded_{asset['id']}")
        if any(not isinstance(item, dict) for item in selected):
            raise ContractError(f"asset_collection_item_invalid_{asset['id']}")
        return len(selected)
    if not isinstance(selected, dict):
        raise ContractError(f"asset_document_shape_invalid_{asset['id']}")
    return 1


def semantic_items(value: Any, asset: dict[str, Any]) -> list[dict[str, Any]]:
    selected = resolve_pointer(value, str(asset.get("response_pointer", "")))
    if not isinstance(selected, list):
        raise ContractError(f"asset_collection_shape_invalid_{asset['id']}")
    restore = asset.get("restore")
    if not isinstance(restore, dict):
        raise ContractError(f"asset_not_portable_{asset['id']}")
    strip_fields = set(restore.get("strip_top_level_fields", []))
    normalized: list[dict[str, Any]] = []
    for item in selected:
        if not isinstance(item, dict):
            raise ContractError(f"asset_collection_item_invalid_{asset['id']}")
        normalized.append(
            {key: value for key, value in item.items() if key not in strip_fields}
        )
    return sorted(normalized, key=lambda item: canonical_json_bytes(item))


class ApiClient:
    def __init__(
        self,
        *,
        base_url: str,
        header_name: str,
        token: str,
        timeout_seconds: int,
        max_response_bytes: int,
    ) -> None:
        self.base_url = validate_base_url(base_url)
        self.header_name = header_name
        self.token = token
        self.timeout_seconds = timeout_seconds
        self.max_response_bytes = max_response_bytes
        self.opener = urllib.request.build_opener(_NoRedirect())

    def _url(self, endpoint: str) -> str:
        if not endpoint.startswith("/api/v1/") or any(
            item in endpoint for item in ("..", "?", "#")
        ):
            raise ContractError("api_endpoint_unsafe")
        return f"{self.base_url}{endpoint}"

    def request_json(self, endpoint: str) -> Any:
        request = urllib.request.Request(
            self._url(endpoint),
            method="GET",
            headers={
                self.header_name: self.token,
                "Accept": "application/json",
                "User-Agent": "AWOOOI-P0-OBS-002-metadata-export/1",
            },
        )
        try:
            with self.opener.open(request, timeout=self.timeout_seconds) as response:
                status_code = int(response.status)
                content_type = response.headers.get_content_type()
                payload = response.read(self.max_response_bytes + 1)
        except urllib.error.HTTPError as exc:
            raise ContractError(f"api_http_status_{exc.code}") from exc
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
            raise ContractError("api_transport_failed") from exc
        if status_code != 200:
            raise ContractError(f"api_http_status_{status_code}")
        if content_type != "application/json":
            raise ContractError("api_content_type_not_json")
        if len(payload) > self.max_response_bytes:
            raise ContractError("api_response_size_exceeded")
        try:
            return json.loads(payload)
        except (UnicodeError, json.JSONDecodeError) as exc:
            raise ContractError("api_response_json_invalid") from exc

    def request_write(self, method: str, endpoint: str, payload: Any) -> None:
        if method not in {"POST", "PUT", "PATCH"}:
            raise ContractError("api_write_method_forbidden")
        body = canonical_json_bytes(payload)
        if len(body) > self.max_response_bytes:
            raise ContractError("api_request_size_exceeded")
        request = urllib.request.Request(
            self._url(endpoint),
            method=method,
            data=body,
            headers={
                self.header_name: self.token,
                "Accept": "application/json",
                "Content-Type": "application/json",
                "User-Agent": "AWOOOI-P0-OBS-002-metadata-restore/1",
            },
        )
        try:
            with self.opener.open(request, timeout=self.timeout_seconds) as response:
                status_code = int(response.status)
                response_payload = response.read(self.max_response_bytes + 1)
        except urllib.error.HTTPError as exc:
            raise ContractError(f"api_write_http_status_{exc.code}") from exc
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
            raise ContractError("api_write_transport_failed") from exc
        if status_code not in {200, 201, 202, 204}:
            raise ContractError(f"api_write_http_status_{status_code}")
        if len(response_payload) > self.max_response_bytes:
            raise ContractError("api_write_response_size_exceeded")


def receipt(
    *,
    trace_id: str,
    run_id: str,
    work_item_id: str,
    phase: str,
    terminal: str,
    detail: str,
) -> dict[str, Any]:
    return {
        "schema": "awoooi_signoz_metadata_receipt_v1",
        "trace_id": trace_id,
        "run_id": run_id,
        "work_item_id": work_item_id,
        "observed_at": utc_now(),
        "phase": phase,
        "terminal": terminal,
        "detail": detail,
    }


def write_private(path: Path, payload: bytes) -> None:
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    descriptor = os.open(path, flags, 0o600)
    try:
        os.fchmod(descriptor, 0o600)
        with os.fdopen(descriptor, "wb", closefd=True) as stream:
            stream.write(payload)
            stream.flush()
            os.fsync(stream.fileno())
    except BaseException:
        try:
            os.close(descriptor)
        except OSError:
            pass
        raise


def validate_new_private_destination(path: Path, *, label: str) -> None:
    if not path.is_absolute():
        raise ContractError(f"{label}_path_must_be_absolute")
    require_no_symlink_components(path.parent, label=f"{label}_parent")
    try:
        parent_metadata = path.parent.lstat()
    except OSError as exc:
        raise ContractError(f"{label}_parent_unavailable") from exc
    if not stat.S_ISDIR(parent_metadata.st_mode):
        raise ContractError(f"{label}_parent_not_directory")
    if parent_metadata.st_uid != os.geteuid():
        raise ContractError(f"{label}_parent_owner_mismatch")
    if path.exists() or path.is_symlink():
        raise ContractError(f"{label}_path_exists")


def write_new_private_atomic(path: Path, payload: bytes, *, label: str) -> None:
    validate_new_private_destination(path, label=label)
    temporary = path.with_name(f".{path.name}.partial.{os.getpid()}")
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    try:
        descriptor = os.open(temporary, flags, 0o600)
        try:
            with os.fdopen(descriptor, "wb", closefd=True) as stream:
                stream.write(payload)
                stream.flush()
                os.fsync(stream.fileno())
        except BaseException:
            try:
                os.close(descriptor)
            except OSError:
                pass
            raise
        os.replace(temporary, path)
        try:
            parent_descriptor = os.open(path.parent, os.O_RDONLY)
            try:
                os.fsync(parent_descriptor)
            finally:
                os.close(parent_descriptor)
        except OSError:
            pass
    finally:
        try:
            temporary.unlink()
        except FileNotFoundError:
            pass