"""Fail-closed validation primitives for Test1 runtime evidence."""
from __future__ import annotations

import json
import os
import re
import tempfile
from pathlib import Path
from typing import Any, Iterable, Mapping


SENSITIVE_NORMALIZED_KEYS = frozenset(
    {
        "authorization",
        "proxyauthorization",
        "cookie",
        "setcookie",
        "xrefreshedtoken",
        "xnexuscrmsessioncontrol",
        "xnexuscrmsessioncontrolintent",
        "xnexuscrmsessioncontrolproof",
        "authorizationcode",
        "codeverifier",
        "apikey",
        "privatekey",
        "publickey",
        "masterkey",
        "encryptionkey",
        "accesstoken",
        "refreshtoken",
        "idtoken",
        "bearertoken",
        "password",
        "passwd",
        "clientsecret",
        "secret",
        "secretvalue",
        "proof",
        "credential",
        "credentials",
        "tidbuser",
        "tidbcloudbranchid",
        "googleclientid",
        "mysqlpwd",
        "smtpuser",
        "awssecretaccesskey",
    }
)
SENSITIVE_NORMALIZED_SUFFIXES = (
    "password",
    "passwordconfirmation",
    "passwordbcrypt",
    "passwordbcryptliteral",
    "passwordhash",
    "token",
    "secret",
    "secretvalue",
    "proof",
    "apikey",
    "privatekey",
    "publickey",
    "masterkey",
    "encryptionkey",
    "secretaccesskey",
    "credential",
    "credentials",
    "tidbuser",
    "tidbcloudbranchid",
    "googleclientid",
    "mysqlpwd",
    "smtpuser",
)
SENSITIVE_REPRESENTATION_SUFFIXES = ("base64url", "base64", "b64")
FIELD_SCOPED_SENSITIVE_KEY_SUFFIXES = ("tidbuser",)
SALESFORCE_CUSTOM_FIELD_SUFFIX = re.compile(r"__(?:b|c|e|mdt|pc|x)$", re.IGNORECASE)
STRUCTURED_REDACTION_REPLACEMENT = ""
TEST1_CONNECTED_WORK_SCHEMA_VERSION = "test1-connected-work-evidence/v1"
TEST1_RUNTIME_GATE = "A1_RUNTIME_REF_SCHEMA"
TEST1_WORKITEM_GATE = "E0_WORKITEM_CLEAN_PILOT_20_20_16"
TEST1_WORKITEM_PROFILE = "noop-pilot"
TEST1_WORKITEM_RESULT = "noop_pilot_20_events_20_lanes"
TEST1_EXPECTED_EVENTS = 20
TEST1_EXPECTED_LANES = 20
TEST1_EXPECTED_WORKERS = 16
TEST1_EXPECTED_MAX_PROCESSING_ROWS = 32
TEST1_EXPECTED_QUEUE_PROFILE = "default-heavy"
GIT_COMMIT_RE = re.compile(r"[0-9a-f]{40}")
MODULE_VERSION_RE = re.compile(r"v[0-9]+\.[0-9]+\.[0-9]+(?:[-+][0-9A-Za-z.-]+)?")
SHA256_RE = re.compile(r"[0-9a-f]{64}")


class EvidenceBlocked(Exception):
    """A blocked evidence result whose fields are safe for callers to serialize."""

    def __init__(
        self,
        verdict: str,
        message: str,
        details: Mapping[str, Any] | None = None,
    ) -> None:
        super().__init__(message)
        self.verdict = verdict
        self.message = message
        self.details = dict(details or {})


def _normalized_sensitive_key(key: Any) -> str:
    return re.sub(r"[^a-z0-9]", "", str(key).lower())


def _is_sensitive_normalized_key(normalized: str) -> bool:
    if normalized in SENSITIVE_NORMALIZED_KEYS:
        return True
    if (
        "fingerprint" in normalized
        or "digest" in normalized
        or normalized.endswith("hashprefix")
    ):
        return False
    if any(normalized.endswith(suffix) for suffix in SENSITIVE_NORMALIZED_SUFFIXES):
        return True
    return any(
        normalized.endswith(suffix)
        and _is_sensitive_normalized_key(normalized[: -len(suffix)])
        for suffix in SENSITIVE_REPRESENTATION_SUFFIXES
    )


def _sensitive_key_candidates(key: Any) -> set[str]:
    raw = str(key)
    candidates = {_normalized_sensitive_key(raw)}
    salesforce_base = SALESFORCE_CUSTOM_FIELD_SUFFIX.sub("", raw)
    if salesforce_base != raw:
        candidates.add(_normalized_sensitive_key(salesforce_base))
    return candidates


def _is_sensitive_key(key: Any) -> bool:
    return any(
        _is_sensitive_normalized_key(candidate)
        for candidate in _sensitive_key_candidates(key)
    )


def _looks_high_entropy_identifier(value: str) -> bool:
    return (
        len(value) >= 12
        and any(char.isalpha() for char in value)
        and any(char.isdigit() for char in value)
    )


def _should_collect_sensitive_value(key: Any, value: str) -> bool:
    candidates = _sensitive_key_candidates(key)
    if not any(_is_sensitive_normalized_key(item) for item in candidates):
        return False
    if any(
        candidate.endswith(suffix)
        for candidate in candidates
        for suffix in FIELD_SCOPED_SENSITIVE_KEY_SUFFIXES
    ):
        return _looks_high_entropy_identifier(value)
    return True


def sensitive_values_for_mapping(runtime_values: Mapping[str, Any]) -> list[str]:
    """Collect secrets from values already parsed by an owning runtime producer."""
    return _sensitive_values(runtime_values.items())


def _sensitive_values(assignments: Iterable[tuple[Any, Any]]) -> list[str]:
    sensitive = {
        value
        for key, value in os.environ.items()
        if value and _should_collect_sensitive_value(key, value)
    }
    sensitive.update(
        str(value)
        for key, value in assignments
        if value and _should_collect_sensitive_value(key, str(value))
    )
    return sorted(sensitive, key=len, reverse=True)


def _is_field_scoped_sensitive_key(key: Any) -> bool:
    return any(
        candidate.endswith(suffix)
        for candidate in _sensitive_key_candidates(key)
        for suffix in FIELD_SCOPED_SENSITIVE_KEY_SUFFIXES
    )


def _high_entropy_descendant_values(value: Any) -> set[str]:
    if isinstance(value, Mapping):
        return {
            item
            for child in value.values()
            for item in _high_entropy_descendant_values(child)
        }
    if isinstance(value, (list, tuple)):
        return {
            item
            for child in value
            for item in _high_entropy_descendant_values(child)
        }
    if value is None or isinstance(value, bool):
        return set()
    text = str(value)
    return {text} if _looks_high_entropy_identifier(text) else set()


def _known_structured_sensitive_values(
    value: Any,
    *,
    inherited: bool = False,
) -> set[str]:
    sensitive: set[str] = set()
    if isinstance(value, Mapping):
        for key, item in value.items():
            key_sensitive = _is_sensitive_key(key)
            field_scoped = key_sensitive and _is_field_scoped_sensitive_key(key)
            if field_scoped and isinstance(item, (Mapping, list, tuple)):
                sensitive.update(_high_entropy_descendant_values(item))
            if (
                key_sensitive
                and not isinstance(item, (Mapping, list, tuple))
                and item is not None
                and not isinstance(item, bool)
                and str(item)
                and _should_collect_sensitive_value(key, str(item))
            ):
                sensitive.add(str(item))
            sensitive.update(
                _known_structured_sensitive_values(
                    item,
                    inherited=inherited or (key_sensitive and not field_scoped),
                )
            )
    elif isinstance(value, (list, tuple)):
        for item in value:
            sensitive.update(
                _known_structured_sensitive_values(item, inherited=inherited)
            )
    elif inherited and value is not None and not isinstance(value, bool) and str(value):
        sensitive.add(str(value))
    return sensitive


def _redact_known_text(
    value: str,
    sensitive_pattern: re.Pattern[str] | None,
) -> str:
    if sensitive_pattern is None:
        return value
    while sensitive_pattern.search(value):
        value = sensitive_pattern.sub(STRUCTURED_REDACTION_REPLACEMENT, value)
    return value


def _redact_known_structured_values(
    value: Any,
    sensitive_pattern: re.Pattern[str] | None,
) -> Any:
    if isinstance(value, Mapping):
        return {
            _redact_known_text(str(key), sensitive_pattern): (
                STRUCTURED_REDACTION_REPLACEMENT
                if _is_sensitive_key(key)
                else _redact_known_structured_values(item, sensitive_pattern)
            )
            for key, item in value.items()
        }
    if isinstance(value, (list, tuple)):
        return [
            _redact_known_structured_values(item, sensitive_pattern) for item in value
        ]
    if isinstance(value, str):
        return _redact_known_text(value, sensitive_pattern)
    if (
        sensitive_pattern is not None
        and value is not None
        and not isinstance(value, bool)
        and sensitive_pattern.search(str(value))
    ):
        return STRUCTURED_REDACTION_REPLACEMENT
    return value


def redact_known_structured_values(
    value: Any,
    sensitive_values: Iterable[str],
) -> Any:
    """Remove classifier-owned and explicit exact values from a detached copy."""
    literal_values = {
        *{str(item) for item in sensitive_values if str(item)},
        *_known_structured_sensitive_values(value),
    }
    ordered = sorted(literal_values, key=len, reverse=True)
    pattern = re.compile("|".join(re.escape(item) for item in ordered)) if ordered else None
    return _redact_known_structured_values(value, pattern)


def _blocked_report(message: str) -> EvidenceBlocked:
    return EvidenceBlocked("BLOCK_TEST1_EVIDENCE_SCHEMA", message)


def _object(value: Any, path: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping):
        raise _blocked_report(f"{path} must be an object")
    return value


def _array(value: Any, path: str) -> list[Any]:
    if not isinstance(value, list):
        raise _blocked_report(f"{path} must be an array")
    return value


def _string(value: Any, path: str) -> str:
    if not isinstance(value, str) or not value:
        raise _blocked_report(f"{path} must be a non-empty string")
    return value


def _boolean(value: Any, path: str) -> bool:
    if type(value) is not bool:
        raise _blocked_report(f"{path} must be a boolean")
    return value


def _integer(value: Any, path: str) -> int:
    if type(value) is not int or value < 0:
        raise _blocked_report(f"{path} must be a non-negative integer")
    return value


def _one_gate(summary: Mapping[str, Any], gate_name: str) -> Mapping[str, Any]:
    gates = _array(summary.get("gates"), "matrix.gates")
    matches = [
        _object(item, "matrix.gates[]")
        for item in gates
        if isinstance(item, Mapping) and item.get("gate") == gate_name
    ]
    if len(matches) != 1:
        raise _blocked_report(f"matrix must contain exactly one {gate_name} gate")
    return matches[0]


def _require_equal(value: Any, expected: Any, path: str) -> None:
    if value != expected or type(value) is not type(expected):
        raise _blocked_report(f"{path} does not match the Test1 evidence contract")


def build_test1_connected_work_evidence(
    matrix_summary: Mapping[str, Any],
) -> dict[str, Any]:
    """Build one allowlisted certificate from the live Connected Work producer."""
    summary = _object(matrix_summary, "matrix")
    _require_equal(summary.get("targetEnv"), "test1", "matrix.targetEnv")
    status = _string(summary.get("status"), "matrix.status")
    if not status.startswith("PASS_CONNECTED_WORK_MATRIX_"):
        raise _blocked_report("matrix.status is not a passing Connected Work result")

    runtime_gate = _one_gate(summary, TEST1_RUNTIME_GATE)
    _require_equal(
        runtime_gate.get("status"),
        "PASS_RUNTIME_REF_SCHEMA_GATE",
        "runtime.status",
    )
    runtime = _object(runtime_gate.get("runtime"), "runtime.runtime")
    observed = _object(runtime.get("observed"), "runtime.runtime.observed")
    requirements = _object(
        runtime.get("requirements"), "runtime.runtime.requirements"
    )
    readback = _object(runtime.get("readback"), "runtime.runtime.readback")
    refs = _object(readback.get("refs"), "runtime.runtime.readback.refs")
    checks = _object(runtime_gate.get("checks"), "runtime.checks")

    expected_commit = _string(runtime.get("expectedSha"), "runtime.expectedSha")
    readback_commit = _string(
        refs.get("CONNECTORS_MODULE_READBACK_REF"),
        "runtime.CONNECTORS_MODULE_READBACK_REF",
    )
    module_version = _string(
        refs.get("CONNECTORS_MODULE_VERSION"),
        "runtime.CONNECTORS_MODULE_VERSION",
    )
    if not GIT_COMMIT_RE.fullmatch(expected_commit):
        raise _blocked_report("runtime.expectedSha must be an immutable lowercase commit")
    if not GIT_COMMIT_RE.fullmatch(readback_commit):
        raise _blocked_report(
            "runtime.CONNECTORS_MODULE_READBACK_REF must be an immutable lowercase commit"
        )
    if not MODULE_VERSION_RE.fullmatch(module_version):
        raise _blocked_report("runtime.CONNECTORS_MODULE_VERSION is invalid")
    _require_equal(readback_commit, expected_commit, "runtime module readback")
    for key in (
        "health200",
        "moduleReadbackMatches",
        "schemaAuthorityPresent",
        "workItemClaimEnabled",
        "workerCountExpected",
        "maxProcessingRowsExpected",
        "queueProfileExpected",
    ):
        _require_equal(
            _boolean(checks.get(key), f"runtime.checks.{key}"),
            True,
            f"runtime.checks.{key}",
        )
    for key in (
        "schemaAuthorityPresent",
        "workItemClaimEnabled",
        "workerCountMatches",
        "maxProcessingRowsMatches",
        "queueProfileMatches",
    ):
        _require_equal(
            _boolean(observed.get(key), f"runtime.observed.{key}"),
            True,
            f"runtime.observed.{key}",
        )
    for key in (
        "requireSchemaAuthority",
        "requireWorkItemClaim",
        "requireWorkerCount",
        "requireMaxProcessingRows",
        "requireQueueProfile",
    ):
        _require_equal(
            _boolean(requirements.get(key), f"runtime.requirements.{key}"),
            True,
            f"runtime.requirements.{key}",
        )
    _require_equal(
        _integer(requirements.get("workers"), "runtime.requirements.workers"),
        TEST1_EXPECTED_WORKERS,
        "runtime.requirements.workers",
    )
    _require_equal(
        _integer(
            requirements.get("maxProcessingRows"),
            "runtime.requirements.maxProcessingRows",
        ),
        TEST1_EXPECTED_MAX_PROCESSING_ROWS,
        "runtime.requirements.maxProcessingRows",
    )
    _require_equal(
        _string(
            requirements.get("queueProfile"),
            "runtime.requirements.queueProfile",
        ),
        TEST1_EXPECTED_QUEUE_PROFILE,
        "runtime.requirements.queueProfile",
    )

    workitem_gate = _one_gate(summary, TEST1_WORKITEM_GATE)
    _require_equal(
        workitem_gate.get("status"),
        f"PASS_{TEST1_WORKITEM_GATE}",
        "workitem.status",
    )
    _require_equal(
        workitem_gate.get("profile"), TEST1_WORKITEM_PROFILE, "workitem.profile"
    )
    results = _array(workitem_gate.get("results"), "workitem.results")
    if len(results) != 1:
        raise _blocked_report("workitem.results must contain exactly one pilot result")
    result = _object(results[0], "workitem.results[0]")
    for key, expected in (
        ("name", TEST1_WORKITEM_RESULT),
        ("events", TEST1_EXPECTED_EVENTS),
        ("lanes", TEST1_EXPECTED_LANES),
        ("workers", TEST1_EXPECTED_WORKERS),
        ("accepted", TEST1_EXPECTED_EVENTS),
        ("materialized", TEST1_EXPECTED_EVENTS),
        ("work_materialized", TEST1_EXPECTED_EVENTS),
        ("claimed", TEST1_EXPECTED_EVENTS),
        ("processed", TEST1_EXPECTED_EVENTS),
        ("failed", 0),
        ("retry", 0),
        ("dead", 0),
        ("nonterminal", 0),
        ("claim_scope_workitem_count", TEST1_EXPECTED_EVENTS),
    ):
        actual = (
            _string(result.get(key), f"workitem.{key}")
            if isinstance(expected, str)
            else _integer(result.get(key), f"workitem.{key}")
        )
        _require_equal(actual, expected, f"workitem.{key}")
    for key, expected in (
        ("active_pack_match", True),
        ("explain_available", True),
        ("explain_has_broad_scan", False),
        ("scheduler_lease_read_path_used", False),
        ("lane_ordering_ok", True),
    ):
        _require_equal(
            _boolean(result.get(key), f"workitem.{key}"),
            expected,
            f"workitem.{key}",
        )
    _require_equal(
        result.get("candidate_claim_plan"),
        "workitem_row_cas",
        "workitem.candidate_claim_plan",
    )
    _require_equal(result.get("verdict"), "PASS", "workitem.verdict")
    installed_pack_fingerprint = _string(
        result.get("installed_pack_id_sha256"),
        "workitem.installed_pack_id_sha256",
    )
    if not SHA256_RE.fullmatch(installed_pack_fingerprint):
        raise _blocked_report("workitem installed-pack fingerprint is invalid")
    generation = _integer(result.get("generation"), "workitem.generation")

    report = {
        "schema_version": TEST1_CONNECTED_WORK_SCHEMA_VERSION,
        "verdict": "PASS_TEST1_CONNECTED_WORK_EVIDENCE",
        "target_env": "test1",
        "module_identity": {
            "module": "github.com/pingcap-inc/nexuscrm-connectors",
            "version": module_version,
            "expected_commit": expected_commit,
            "readback_commit": readback_commit,
            "readback_matches_expected": True,
        },
        "workitem": {
            "gate": TEST1_WORKITEM_GATE,
            "profile": TEST1_WORKITEM_PROFILE,
            "events": TEST1_EXPECTED_EVENTS,
            "lanes": TEST1_EXPECTED_LANES,
            "workers": TEST1_EXPECTED_WORKERS,
            "accepted": TEST1_EXPECTED_EVENTS,
            "materialized": TEST1_EXPECTED_EVENTS,
            "work_materialized": TEST1_EXPECTED_EVENTS,
            "claimed": TEST1_EXPECTED_EVENTS,
            "processed": TEST1_EXPECTED_EVENTS,
            "failed": 0,
            "retry": 0,
            "dead": 0,
            "nonterminal": 0,
            "installed_pack_id_sha256": installed_pack_fingerprint,
            "generation": generation,
            "active_pack_match": True,
            "claim_path": "workitem_row_cas",
            "explain_available": True,
            "explain_has_broad_scan": False,
            "scheduler_lease_read_path_used": False,
            "lane_ordering_ok": True,
        },
    }
    validate_test1_connected_work_evidence(report)
    return report


def validate_test1_connected_work_evidence(report: Mapping[str, Any]) -> None:
    """Reject any field or value outside the versioned Test1 certificate."""
    root = _object(report, "report")
    module = _object(root.get("module_identity"), "report.module_identity")
    workitem = _object(root.get("workitem"), "report.workitem")
    version = _string(module.get("version"), "report.module_identity.version")
    expected_commit = _string(
        module.get("expected_commit"), "report.module_identity.expected_commit"
    )
    readback_commit = _string(
        module.get("readback_commit"), "report.module_identity.readback_commit"
    )
    if not MODULE_VERSION_RE.fullmatch(version):
        raise _blocked_report("report module version is invalid")
    if not GIT_COMMIT_RE.fullmatch(expected_commit) or not GIT_COMMIT_RE.fullmatch(readback_commit):
        raise _blocked_report("report module commits are invalid")
    _require_equal(readback_commit, expected_commit, "report module commits")
    fingerprint = _string(
        workitem.get("installed_pack_id_sha256"),
        "report.workitem.installed_pack_id_sha256",
    )
    if not SHA256_RE.fullmatch(fingerprint):
        raise _blocked_report("report installed-pack fingerprint is invalid")
    generation = _integer(workitem.get("generation"), "report.workitem.generation")
    expected = {
        "schema_version": TEST1_CONNECTED_WORK_SCHEMA_VERSION,
        "verdict": "PASS_TEST1_CONNECTED_WORK_EVIDENCE",
        "target_env": "test1",
        "module_identity": {
            "module": "github.com/pingcap-inc/nexuscrm-connectors",
            "version": version,
            "expected_commit": expected_commit,
            "readback_commit": readback_commit,
            "readback_matches_expected": True,
        },
        "workitem": {
            "gate": TEST1_WORKITEM_GATE,
            "profile": TEST1_WORKITEM_PROFILE,
            "events": TEST1_EXPECTED_EVENTS,
            "lanes": TEST1_EXPECTED_LANES,
            "workers": TEST1_EXPECTED_WORKERS,
            "accepted": TEST1_EXPECTED_EVENTS,
            "materialized": TEST1_EXPECTED_EVENTS,
            "work_materialized": TEST1_EXPECTED_EVENTS,
            "claimed": TEST1_EXPECTED_EVENTS,
            "processed": TEST1_EXPECTED_EVENTS,
            "failed": 0,
            "retry": 0,
            "dead": 0,
            "nonterminal": 0,
            "installed_pack_id_sha256": fingerprint,
            "generation": generation,
            "active_pack_match": True,
            "claim_path": "workitem_row_cas",
            "explain_available": True,
            "explain_has_broad_scan": False,
            "scheduler_lease_read_path_used": False,
            "lane_ordering_ok": True,
        },
    }
    try:
        actual_json = json.dumps(dict(root), sort_keys=True, separators=(",", ":"))
    except (TypeError, ValueError) as exc:
        raise _blocked_report("report is not canonical JSON") from exc
    expected_json = json.dumps(expected, sort_keys=True, separators=(",", ":"))
    if actual_json != expected_json:
        raise _blocked_report("report contains missing or unapproved fields")


def write_test1_connected_work_evidence(
    path: Path,
    report: Mapping[str, Any],
    *,
    sensitive_values: Iterable[str],
) -> None:
    """Atomically persist only an exact, already-safe Test1 certificate."""
    validate_test1_connected_work_evidence(report)
    detached = json.loads(json.dumps(dict(report), sort_keys=True))
    safe = redact_known_structured_values(detached, sensitive_values)
    if safe != detached:
        raise EvidenceBlocked(
            "BLOCK_TEST1_EVIDENCE_CONTAINS_SENSITIVE_DATA",
            "Test1 evidence contains a classified or known sensitive value",
        )
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = json.dumps(safe, indent=2, sort_keys=True) + "\n"
    temporary_name = ""
    try:
        with tempfile.NamedTemporaryFile(
            mode="w",
            encoding="utf-8",
            dir=path.parent,
            prefix=f".{path.name}.",
            suffix=".tmp",
            delete=False,
        ) as handle:
            temporary_name = handle.name
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        Path(temporary_name).replace(path)
    finally:
        if temporary_name:
            Path(temporary_name).unlink(missing_ok=True)