"""Fail-closed privacy boundary for model-bound fact packages."""
from __future__ import annotations

import re
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from typing import Any


class UnsafeFactPackage(ValueError):
    """Raised before a model call when a package may contain sensitive data."""


_DENIED_KEYS = {
    "name", "display_name", "email", "mail", "phone", "mobile", "contact",
    "ip", "ip_address", "public_ip", "private_ip", "hostname", "machine_name",
    "domain", "domain_name", "account", "account_name", "username", "login",
    "secret", "password", "token", "api_key", "access_key", "private_key",
    "policy", "raw_policy", "raw_data", "raw", "audit_text", "description",
}
_DENIED_KEY_PARTS = ("password", "secret", "token", "private_key", "access_key")
_EMAIL = re.compile(r"(?i)\b[a-z0-9.!#$%&'*+/=?^_`{|}~-]+@[a-z0-9-]+(?:\.[a-z0-9-]+)+\b")
_IPV4 = re.compile(r"(?<!\d)(?:\d{1,3}\.){3}\d{1,3}(?!\d)")
_DOMAIN = re.compile(r"(?i)\b(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+(?:com|net|org|cn|io|ai|cloud|internal)\b")
_SECRET = re.compile(
    r"(?i)(?:sk-[a-z0-9_-]{12,}|AKIA[0-9A-Z]{16}|-----BEGIN [A-Z ]*PRIVATE KEY-----|"
    r"(?:bearer|password|secret|token)\s*[:=]\s*[^\s,;]{8,})"
)
_ANON_REF = re.compile(r"^[a-z][a-z0-9]{1,15}_[0-9]{2,6}$")

# The separator the citation flattener joins addresses with. The owning
# definition is ``contracts.FLAT_KEY_SEPARATOR``, but ``contracts`` imports
# THIS module, so a second declaration here is the only way to name it
# without an import cycle; ``test_privacy`` pins the two to each other.
_FLAT_PATH_SEPARATOR = "."


@dataclass(frozen=True)
class EvidenceObject:
    kind: str
    object_id: str


def build_reference_map(objects: Iterable[EvidenceObject]) -> tuple[list[str], dict[str, Any]]:
    """Assign stable-in-package anonymous references to local objects."""
    counters: dict[str, int] = {}
    refs: list[str] = []
    mapping: dict[str, Any] = {}
    seen: set[tuple[str, str]] = set()
    # Sorting makes the same domain facts produce the same references/hash.
    for obj in sorted(objects, key=lambda item: (item.kind, item.object_id)):
        identity = (obj.kind, obj.object_id)
        if identity in seen:
            continue
        seen.add(identity)
        kind = re.sub(r"[^a-z0-9]", "", obj.kind.lower())[:12] or "obj"
        counters[kind] = counters.get(kind, 0) + 1
        ref = f"{kind}_{counters[kind]:02d}"
        refs.append(ref)
        mapping[ref] = {
            "kind": obj.kind,
            "object_id": obj.object_id,
        }
    return refs, mapping


def merge_anonymous_fact_packages(
    packages: Mapping[str, tuple[Any, Mapping[str, Any]]],
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Merge independently numbered packages into one collision-free namespace."""
    objects: list[EvidenceObject] = []
    for payload, reference_map in packages.values():
        validate_anonymous_refs(payload, set(reference_map))
        for ref, value in reference_map.items():
            if (
                not isinstance(ref, str)
                or not isinstance(value, Mapping)
                or set(value) != {"kind", "object_id"}
                or not all(
                    isinstance(value.get(key), str) and value[key]
                    for key in ("kind", "object_id")
                )
            ):
                raise UnsafeFactPackage("Fact package contains an invalid reference map")
            objects.append(EvidenceObject(value["kind"], value["object_id"]))

    _, merged_map = build_reference_map(objects)
    reverse = {
        (value["kind"], value["object_id"]): ref
        for ref, value in merged_map.items()
    }

    def replace(value: Any, replacements: Mapping[str, str]) -> Any:
        if isinstance(value, dict):
            return {key: replace(child, replacements) for key, child in value.items()}
        if isinstance(value, list):
            return [replace(child, replacements) for child in value]
        if isinstance(value, tuple):
            return tuple(replace(child, replacements) for child in value)
        if isinstance(value, str):
            return replacements.get(value, value)
        return value

    merged_payloads: dict[str, Any] = {}
    for name, (payload, reference_map) in packages.items():
        replacements = {
            local_ref: reverse[(value["kind"], value["object_id"])]
            for local_ref, value in reference_map.items()
        }
        merged_payloads[name] = replace(payload, replacements)
    assert_llm_safe(merged_payloads)
    validate_anonymous_refs(merged_payloads, set(merged_map))
    return merged_payloads, merged_map


def _sensitive_shape(candidate: str) -> str | None:
    """Name the recognizable sensitive shape in a string, if any."""
    if _EMAIL.search(candidate):
        return "email-like value"
    if _IPV4.search(candidate):
        return "IP-like value"
    if _DOMAIN.search(candidate):
        return "domain-like value"
    if _SECRET.search(candidate):
        return "credential-like value"
    return None


def _key_shape(key: str, *, flat_path: bool) -> str | None:
    """Name the sensitive shape in a KEY, reading flat keys as addresses.

    A flattened key is an address our own flattener joined — ``metrics`` +
    ``cloud.access.governed_coverage`` + ``value``. Handing that joined
    string to the value regexes reads our vocabulary as a hostname:
    ``metrics.cloud`` is ``<label>.<TLD>`` to ``_DOMAIN`` because ``cloud``
    is one of its TLDs, and every one of the 19 cloud-subscription metric
    keys begins with ``cloud.``. The whole subject's package was rejected
    fail-closed, which is a total outage of that analysis, not a leak.

    Scanning each SEGMENT instead keeps the net where content can actually
    enter. The caller guarantees the nested payload already passed the
    strict scan, and every segment of a flat key was, in that nested shape,
    either a structural name we wrote or a string VALUE that the value
    regexes have already seen — so a hostname promoted into key position was
    rejected one pass earlier, as a value. Dot-bearing shapes (domains,
    IPv4) cannot survive inside a single segment; ``@``-bearing and
    credential shapes still can, and are still caught here.
    """
    if not flat_path:
        return _sensitive_shape(key)
    for segment in key.split(_FLAT_PATH_SEPARATOR):
        if (shape := _sensitive_shape(segment)) is not None:
            return shape
    return None


def assert_llm_safe(payload: Any, *, keys_are_flat_paths: bool = False) -> None:
    """Recursively reject prohibited keys and recognizable sensitive values.

    Set ``keys_are_flat_paths`` only for a payload whose keys are joined
    citation ADDRESSES (``<member>.<...>.<leaf>``) and whose nested form has
    already passed this same guard — see the call in
    ``generation.build_model_fact_package``.
    """
    errors: list[str] = []

    def visit(value: Any, path: str) -> None:
        if isinstance(value, dict):
            for raw_key, child in value.items():
                key = str(raw_key).strip().lower()
                if key in _DENIED_KEYS or any(part in key for part in _DENIED_KEY_PARTS):
                    errors.append(f"{path}.{key}: prohibited field")
                    continue
                # KEYS are scanned with the value regexes too. A key used to be
                # a structural name written by us, so only the deny-list
                # applied — but the citation layer now addresses a list item by
                # its own ``key`` FIELD, which promotes arbitrary fact content
                # into the key position. Without this, a value that would be
                # rejected one line lower travels freely by being made a key.
                if (shape := _key_shape(key, flat_path=keys_are_flat_paths)) is not None:
                    errors.append(f"{path}.{key}: {shape} in field name")
                    continue
                visit(child, f"{path}.{key}")
        elif isinstance(value, (list, tuple)):
            for index, child in enumerate(value):
                visit(child, f"{path}[{index}]")
        elif isinstance(value, str):
            if (shape := _sensitive_shape(value.strip())) is not None:
                errors.append(f"{path}: {shape}")

    visit(payload, "$.")
    if errors:
        preview = "; ".join(errors[:8])
        raise UnsafeFactPackage(f"Model fact package rejected: {preview}")


def validate_anonymous_refs(payload: Any, allowed_refs: set[str]) -> None:
    """Reject anonymous-looking model references absent from the fact pack."""
    unknown: set[str] = set()

    def visit(value: Any) -> None:
        if isinstance(value, dict):
            for child in value.values():
                visit(child)
        elif isinstance(value, list):
            for child in value:
                visit(child)
        elif isinstance(value, str) and _ANON_REF.fullmatch(value) and value not in allowed_refs:
            unknown.add(value)

    visit(payload)
    if unknown:
        raise UnsafeFactPackage(f"Model returned unknown evidence refs: {sorted(unknown)}")