"""Validation policy for Hermes' least-authority Atlas Forgejo client.

The screening catches structured credential assignments, known token formats,
and long high-entropy values. It is a fail-closed accident barrier, not proof
that text is secret-free: a short or multiword secret under an innocuous key is
not reliably distinguishable from prose and must never be supplied by callers.
JSON depth/nodes/documents and multiline gaps are deliberately bounded; large
or unusual configuration blobs are not a supported pull-request field format.
"""

# ruff: noqa: SIM905

from __future__ import annotations

import json
import re
import subprocess
import urllib.parse
from collections import Counter
from math import log2

REPO_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,99}\Z")
SHA_RE = re.compile(r"[0-9a-fA-F]{40}\Z")
DRAFT_TITLE_PREFIX = "WIP: "
GIT_BIN = "/usr/bin/git"
MAX_QUERY_LENGTH = 128
MAX_QUERY_ITEMS = 3
MAX_QUERY_KEY_LENGTH = 16
MAX_QUERY_VALUE_LENGTH = 16
MAX_PAGE = 10_000
MAX_LIMIT = 50
MAX_PR_NUMBER = 2_147_483_647
MAX_PR_NUMBER_DIGITS = 10
MAX_REF_CHARACTERS = 200
MAX_REF_UTF8_BYTES = 255
MAX_TITLE_UTF8_BYTES = 512
MAX_BODY_UTF8_BYTES = 32_768
MAX_JSON_DOCUMENTS = 32
MAX_JSON_NODES = 2_048
MAX_JSON_DEPTH = 32
CANONICAL_QUERY_RE = re.compile(r"[A-Za-z0-9_=&-]*\Z")
ASSIGNMENT_RE = re.compile(
    r"""(?mx)
    (?<![A-Za-z0-9_.-])
    (?P<key_quote>[\"']?)
    (?P<key>\.?[A-Za-z][A-Za-z0-9_.-]{0,127})
    (?P=key_quote)
    [ \t]*(?P<separator>[:=])[ \t]*
    (?P<value>
      \"(?:\\.|[^\"\\\r\n])*\" |
      '(?:\\.|[^'\\\r\n])*' |
      [^\r\n,;}&]{0,2048}
    )
    """
)
JSON_KEY_RE = re.compile(r'"(?P<key>[^"]{1,512})"(?P<spacing>[\x00-\x20\x7f]{0,256}):')
MULTILINE_ASSIGNMENT_RE = re.compile(
    r"""(?mx)
    (?<![A-Za-z0-9_.-])
    (?P<key_quote>["']?)
    (?P<key>\.?[A-Za-z][A-Za-z0-9_.-]{0,127})
    (?P=key_quote)
    [ \t]{0,64}(?:\r?\n[ \t]{0,64}){0,4}
    (?P<separator>[:=])
    [ \t]{0,64}(?:\r?\n[ \t]{0,64}){1,4}
    (?P<value>[^\r\n,;}&]{1,2048})
    """
)
CAMEL_BOUNDARY_RE = re.compile(r"(?<=[a-z0-9])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])")
CREDENTIAL_SCHEME_RE = re.compile(
    r"(?i)\b(?:bearer|basic|token)\s+"
    r"(?=[A-Za-z0-9+/_=.:-]{8,}\b)(?=[A-Za-z0-9+/_=.:-]*[0-9+/_=.-])"
    r"[A-Za-z0-9+/_=.:-]{8,}"
)
STANDALONE_SECRET_PATTERNS = tuple(
    re.compile(pattern, re.IGNORECASE)
    for pattern in (
        r"(?<![A-Za-z0-9])gh[pousr]_[A-Za-z0-9]{20,}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])github_pat_[A-Za-z0-9_]{20,}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])glpat-[A-Za-z0-9_-]{20,}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])xox[baprs]-[A-Za-z0-9-]{20,}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])sk-(?:ant-|proj-)?[A-Za-z0-9_-]{20,}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])[rs]k_(?:test|live)_[A-Za-z0-9]{20,}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])whsec_[A-Za-z0-9]{20,}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])npm_[A-Za-z0-9]{20,}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])pypi-[A-Za-z0-9_-]{30,}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])hf_[A-Za-z0-9]{20,}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])(?:gta|gto|gitea|forgejo)_[A-Za-z0-9_-]{20,}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])ya29\.[A-Za-z0-9_-]{20,}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])oy2[A-Za-z0-9]{40,}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])SG\.[A-Za-z0-9_-]{16,}\.[A-Za-z0-9_-]{20,}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])SK[0-9a-f]{32}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])(?:AKIA|ASIA)[0-9A-Z]{16}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])AIza[0-9A-Za-z_-]{35}(?![A-Za-z0-9])",
        r"(?<![A-Za-z0-9])eyJ[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{8,}(?![A-Za-z0-9])",
        r"-----BEGIN\s+[A-Z0-9 ][A-Z0-9 -]{1,62}-----",
        r"\bssh-(?:rsa|ed25519|dss|ecdsa-[A-Za-z0-9-]+)\s+[A-Za-z0-9+/]{16,}={0,3}",
        r"https://hooks\.slack\.com/services/[A-Za-z0-9/_-]{20,}",
        r"https://(?:discord(?:app)?\.com)/api/webhooks/[0-9]+/[A-Za-z0-9._-]{20,}",
        r"https://[^\s/]*webhook\.office\.com/[^\s]{20,}",
    )
)
HIGH_ENTROPY_TOKEN_RE = re.compile(
    r"(?<![A-Za-z0-9+/_=-])[A-Za-z0-9+/_=-]{48,16384}(?![A-Za-z0-9+/_=-])"
)
PROSE_LEAD_WORDS = frozenset(
    "add allow change check describe document ensure explain fix handle keep preserve "
    "prevent reject remove review support test validate".split()
)
SENSITIVE_KEY_WORDS = frozenset(
    "auth authentication authorization auths basic bearer credential credentials "
    "passwd password pat private sas secret secrets sig signature signing token tokens "
    "webhook".split()
)
KEY_MODIFIER_WORDS = frozenset(
    "access account api auth client encryption identity private registry secret service "
    "session signing ssh".split()
)
SENSITIVE_COMPACT_KEYS = frozenset(
    "accessid accesskeyid accountkey clientemail clientid connectionstring "
    "dockerconfigjson privatekeyid serviceaccountkey sharedaccesssignature".split()
)
SENSITIVE_COMPACT_SUFFIXES = tuple(
    "accesskey accesskeyid accountkey apikey authkey clientemail clientid clientkey "
    "connectionstring credential credentials encryptionkey identitykey password passwd "
    "privatekeyid privatekey secret secretkey servicekey sessionkey signature signingkey "
    "sshkey token webhook".split()
)


class PolicyError(ValueError):
    """Raised when a requested Forgejo operation exceeds the safe boundary."""


def _validate_repo(repo: str) -> str:
    if not REPO_RE.fullmatch(repo) or repo in {".", ".."}:
        raise PolicyError("repository name is outside the Atlas allowlist")
    return repo


def _reject_forbidden(value: object, name: str, forbidden: tuple[str, ...]) -> None:
    """Reject exact runtime credentials before invoking helpers or the network."""
    if not isinstance(value, str) or any(secret and secret in value for secret in forbidden):
        raise PolicyError(f"{name} contains runtime credential material")


def _validate_ref_bounds(
    value: object, name: str, *, forbidden: tuple[str, ...] = ()
) -> str:
    """Reject oversized or non-encodable refs before invoking Git."""
    _reject_forbidden(value, name, forbidden)
    if not value or len(value) > MAX_REF_CHARACTERS:
        raise PolicyError(f"{name} exceeds the safe branch-name limit")
    try:
        encoded = value.encode("utf-8", errors="strict")
    except UnicodeEncodeError as exc:
        raise PolicyError(f"{name} is not valid UTF-8 text") from exc
    if len(encoded) > MAX_REF_UTF8_BYTES:
        raise PolicyError(f"{name} exceeds the safe UTF-8 branch-name limit")
    return value


def _validate_ref(
    value: object, name: str, *, forbidden: tuple[str, ...] = ()
) -> str:
    """Validate the complete Git ref grammar using Git itself."""
    value = _validate_ref_bounds(value, name, forbidden=forbidden)
    if value.startswith("-"):
        raise PolicyError(f"{name} must be a same-repository branch name")
    if any(ord(character) < 32 or ord(character) == 127 for character in value):
        raise PolicyError(f"{name} must be a safe same-repository branch name")
    result = subprocess.run(
        [GIT_BIN, "check-ref-format", f"refs/heads/{value}"],
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
    )
    if result.returncode != 0:
        raise PolicyError(f"{name} must be a safe same-repository branch name")
    return value


def _validate_sha(value: object, name: str = "head SHA") -> str:
    if not isinstance(value, str) or not SHA_RE.fullmatch(value):
        raise PolicyError(f"{name} must be a full Git SHA-1")
    return value.lower()


def _validate_pr_number_segment(value: object) -> int:
    """Validate a canonical bounded pull-request number from an API path."""
    if (
        not isinstance(value, str)
        or len(value) > MAX_PR_NUMBER_DIGITS
        or not re.fullmatch(r"[1-9][0-9]*", value)
    ):
        raise PolicyError("pull-request number must use bounded canonical ASCII digits")
    number = int(value)
    if number > MAX_PR_NUMBER:
        raise PolicyError("pull-request number is outside its allowed range")
    return number


def _validate_pr_number(value: object) -> int:
    """Validate a bounded pull-request number returned by Forgejo."""
    if (
        not isinstance(value, int)
        or isinstance(value, bool)
        or not 1 <= value <= MAX_PR_NUMBER
    ):
        raise PolicyError("Forgejo omitted a valid pull-request number")
    return value


def _validate_text(
    value: object,
    name: str,
    maximum: int,
    maximum_bytes: int,
    *,
    required: bool,
) -> str:
    if not isinstance(value, str):
        raise PolicyError(f"{name} must be text")
    if required and not value.strip():
        raise PolicyError(f"{name} must not be empty")
    try:
        encoded = value.encode("utf-8", errors="strict")
    except UnicodeEncodeError as exc:
        raise PolicyError(f"{name} is not valid UTF-8 text") from exc
    if len(value) > maximum or len(encoded) > maximum_bytes or "\x00" in value:
        raise PolicyError(f"{name} exceeds the safe request limit")
    return value


def _normalized_key(value: str) -> tuple[tuple[str, ...], str]:
    """Split camelCase and separator-based assignment keys into semantics."""
    separated = CAMEL_BOUNDARY_RE.sub(" ", value.strip(".\"'"))
    words = tuple(re.findall(r"[A-Za-z0-9]+", separated.lower()))
    return words, "".join(words)


def _is_sensitive_key(value: str) -> bool:
    words, compact = _normalized_key(value)
    word_set = set(words)
    if word_set & SENSITIVE_KEY_WORDS:
        return True
    if compact in SENSITIVE_COMPACT_KEYS:
        return True
    if compact.endswith(SENSITIVE_COMPACT_SUFFIXES):
        return True
    if "key" in word_set and word_set & KEY_MODIFIER_WORDS:
        return True
    if {"client", "email"} <= word_set or {"client", "id"} <= word_set:
        return True
    if {"access", "id"} <= word_set or {"connection", "string"} <= word_set:
        return True
    return compact.endswith("configjson")


def _strip_assignment_value(value: str) -> tuple[str, bool]:
    stripped = value.strip()
    quoted = (
        len(stripped) >= 2 and stripped[0] in {'"', "'"} and stripped[-1] == stripped[0]
    )
    if quoted:
        stripped = stripped[1:-1].strip()
    return stripped, quoted


def _looks_like_prose(value: str) -> bool:
    words = re.findall(r"[A-Za-z]+(?:[-'][A-Za-z]+)?", value)
    if len(words) < 2:
        return False
    if re.search(r"[$`{}\[\]\\@/:=+_]", value):
        return False
    return words[0].lower() in PROSE_LEAD_WORDS


def _looks_sensitive_assignment_value(value: str) -> bool:
    stripped, quoted = _strip_assignment_value(value)
    if not stripped:
        return False
    if quoted or stripped.startswith(("{", "[", "|", ">", "!!", "&", "*")):
        return True
    if CREDENTIAL_SCHEME_RE.search(stripped):
        return True
    if any(pattern.search(stripped) for pattern in STANDALONE_SECRET_PATTERNS):
        return True
    if re.search(r"(?:https?://|[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+)", stripped):
        return True
    if re.search(r"(?:\$\{|\$\(|\\[nrt]|[A-Za-z0-9+/]{16,}={0,3}\Z)", stripped):
        return True
    return not _looks_like_prose(stripped)


def _is_structural_credential_assignment(key: str, value: str) -> bool:
    words, compact_key = _normalized_key(key)
    stripped, _quoted = _strip_assignment_value(value)
    _value_words, compact_value = _normalized_key(stripped)
    return (
        compact_key == "type" and compact_value in {"serviceaccount", "credential"}
    ) or ("service" in words and "account" in words and bool(stripped))


def _decode_json_key(raw: str) -> tuple[str, bool]:
    """Decode one bounded JSON key without silently accepting bad escapes."""
    try:
        decoded = json.loads(f'"{raw}"')
    except (json.JSONDecodeError, UnicodeError):
        approximate = re.sub(r"[^A-Za-z0-9]+", "_", raw.replace("\\", ""))
        return approximate, False
    return decoded if isinstance(decoded, str) else "", True


def _json_value_has_sensitive_assignment(
    value: object, *, depth: int = 0, nodes: list[int] | None = None
) -> bool:
    """Walk one decoded JSON value with explicit depth and node ceilings."""
    if nodes is None:
        nodes = [0]
    nodes[0] += 1
    if nodes[0] > MAX_JSON_NODES or depth > MAX_JSON_DEPTH:
        raise PolicyError("structured pull-request text exceeds the scan limit")
    if isinstance(value, dict):
        for key, assigned in value.items():
            if not isinstance(key, str):
                continue
            if _is_sensitive_key(key):
                return True
            assigned_text = assigned if isinstance(assigned, str) else ""
            if _is_structural_credential_assignment(key, assigned_text):
                return True
            if _json_value_has_sensitive_assignment(
                assigned, depth=depth + 1, nodes=nodes
            ):
                return True
    elif isinstance(value, list):
        return any(
            _json_value_has_sensitive_assignment(item, depth=depth + 1, nodes=nodes)
            for item in value
        )
    return False


def _decoded_json_documents(value: str):
    """Yield bounded embedded JSON object or array fragments."""
    decoder = json.JSONDecoder()
    index = 0
    attempts = 0
    while attempts < MAX_JSON_DOCUMENTS:
        positions = [
            position for token in "{[" if (position := value.find(token, index)) >= 0
        ]
        if not positions:
            return
        start = min(positions)
        attempts += 1
        try:
            document, end = decoder.raw_decode(value, start)
        except (json.JSONDecodeError, RecursionError, ValueError):
            index = start + 1
            continue
        yield document
        index = max(end, start + 1)
    if "{" in value[index:] or "[" in value[index:]:
        raise PolicyError("structured pull-request text exceeds the scan limit")


def _has_structured_sensitive_assignment(value: str) -> bool:
    """Detect bounded JSON and multiline YAML/env credential assignments."""
    for candidate in JSON_KEY_RE.finditer(value):
        decoded, valid = _decode_json_key(candidate.group("key"))
        if _is_sensitive_key(decoded):
            return True
        if not valid and _is_sensitive_key(candidate.group("key").replace("\\", "")):
            return True
    for candidate in MULTILINE_ASSIGNMENT_RE.finditer(value):
        key = candidate.group("key")
        assigned = candidate.group("value")
        if _is_structural_credential_assignment(key, assigned) or (
            _is_sensitive_key(key) and _looks_sensitive_assignment_value(assigned)
        ):
            return True
    return any(
        _json_value_has_sensitive_assignment(item)
        for item in _decoded_json_documents(value)
    )


def _has_high_entropy_token(value: str) -> bool:
    for match in HIGH_ENTROPY_TOKEN_RE.finditer(value):
        token = match.group(0)
        if len(token) > 256:
            return True
        groups = sum(
            bool(re.search(pattern, token))
            for pattern in (r"[a-z]", r"[A-Z]", r"[0-9]", r"[+/_=-]")
        )
        counts = Counter(token)
        entropy = -sum(
            (count / len(token)) * log2(count / len(token)) for count in counts.values()
        )
        if groups >= 3 and entropy >= 4.0:
            return True
    return False


def _reject_sensitive(value: str, name: str, forbidden: tuple[str, ...]) -> None:
    if any(secret and secret in value for secret in forbidden):
        raise PolicyError(f"pull-request {name} contains runtime credential material")
    if CREDENTIAL_SCHEME_RE.search(value):
        raise PolicyError(f"pull-request {name} resembles credential material")
    if any(pattern.search(value) for pattern in STANDALONE_SECRET_PATTERNS):
        raise PolicyError(f"pull-request {name} resembles credential material")
    if _has_structured_sensitive_assignment(value):
        raise PolicyError(f"pull-request {name} resembles credential material")
    for candidate in ASSIGNMENT_RE.finditer(value):
        key = candidate.group("key")
        assigned = candidate.group("value")
        if _is_structural_credential_assignment(key, assigned) or (
            _is_sensitive_key(key) and _looks_sensitive_assignment_value(assigned)
        ):
            raise PolicyError(f"pull-request {name} resembles credential material")
    if _has_high_entropy_token(value):
        raise PolicyError(f"pull-request {name} resembles credential material")


def _validate_body(value: object, *, forbidden: tuple[str, ...] = ()) -> str:
    body = _validate_text(value, "body", 16384, MAX_BODY_UTF8_BYTES, required=False)
    _reject_sensitive(body, "body", forbidden)
    return body


def _draft_title(value: object, *, forbidden: tuple[str, ...] = ()) -> str:
    """Return a bounded, secret-screened Gitea draft title."""
    title = _validate_text(
        value, "title", 251, MAX_TITLE_UTF8_BYTES, required=True
    ).strip()
    _reject_sensitive(title, "title", forbidden)
    for prefix in ("WIP:", "[WIP]"):
        if title.upper().startswith(prefix):
            title = title[len(prefix) :].lstrip()
            break
    if not title:
        raise PolicyError("title must contain text after the draft prefix")
    return DRAFT_TITLE_PREFIX + title


def _validate_query(target: urllib.parse.SplitResult, allowed: set[str]) -> None:
    """Accept only short canonical ASCII query strings with bounded pagination."""
    raw = target.query
    if (
        len(raw) > MAX_QUERY_LENGTH
        or not raw.isascii()
        or not CANONICAL_QUERY_RE.fullmatch(raw)
        or "%" in raw
    ):
        raise PolicyError("API query must use short canonical ASCII form")
    try:
        pairs = urllib.parse.parse_qsl(raw, keep_blank_values=True, strict_parsing=True)
    except ValueError as exc:
        raise PolicyError("invalid API query") from exc
    if len(pairs) > MAX_QUERY_ITEMS:
        raise PolicyError("too many API query parameters")
    if len({key for key, _ in pairs}) != len(pairs):
        raise PolicyError("duplicate API query parameters are not allowed")
    if any(
        not key
        or len(key) > MAX_QUERY_KEY_LENGTH
        or len(value) > MAX_QUERY_VALUE_LENGTH
        for key, value in pairs
    ):
        raise PolicyError("API query key or value exceeds its safe limit")
    if any(key not in allowed for key, _ in pairs):
        raise PolicyError("API query parameter is outside the read allowlist")
    values = dict(pairs)
    for name in ("page", "limit"):
        if name not in values:
            continue
        if not re.fullmatch(r"[0-9]+", values[name]):
            raise PolicyError(f"{name} must use ASCII decimal digits")
        number = int(values[name])
        if values[name] != str(number):
            raise PolicyError(f"{name} must use canonical ASCII decimal form")
        ceiling = MAX_PAGE if name == "page" else MAX_LIMIT
        if not 1 <= number <= ceiling:
            raise PolicyError(f"{name} is outside its allowed range")
    if "state" in values and values["state"] not in {"open", "closed", "all"}:
        raise PolicyError("pull-request state is invalid")