"""Canonical security utility functions for Sublarr.

Centralizes path traversal prevention, ZIP extraction safety,
Git URL validation, URL SSRF validation, and environment variable sanitization.
All security-sensitive operations should use these helpers.
"""

import hashlib
import ipaddress
import os
import socket
import urllib.parse
import zipfile

# Domains allowed for git-based plugin installation
_ALLOWED_GIT_DOMAINS = {"github.com", "gitlab.com", "codeberg.org"}

# Schemes allowed for service/provider URLs
_ALLOWED_SERVICE_SCHEMES = {"http", "https"}

# Cloud metadata hostnames that must always be blocked
_BLOCKED_METADATA_HOSTS = {"metadata.google.internal", "metadata.goog"}

# Metadata IP ranges (link-local 169.254.x.x, Alibaba 100.100.100.200)
_METADATA_NETWORKS = [
    ipaddress.ip_network("169.254.0.0/16"),  # AWS / Azure / GCP link-local
    ipaddress.ip_network("100.100.100.200/32"),  # Alibaba Cloud metadata
]


# ---------------------------------------------------------------------------
# Per-provider domain allowlists for subtitle download URLs (P1)
# ---------------------------------------------------------------------------

# Providers that perform all downloads locally (no external URL needed).
_LOCAL_PROVIDERS = {"embedded", "whisper"}

# Providers that are self-hosted (operator chooses the URL).
# Scheme must be http/https, but any hostname is accepted.
# The generic Custom HTTP/JSON provider ("customapi") and its dynamically
# registered instances ("customapi-<name>") fall in this class too.
_SELF_HOSTED_PROVIDERS = {"subsdump", "customapi"}


def _is_self_hosted_provider(provider_name: str) -> bool:
    return provider_name in _SELF_HOSTED_PROVIDERS or provider_name.startswith("customapi-")


# Allowlists: a URL is accepted when netloc equals entry OR ends with ".<entry>".
_PROVIDER_DOWNLOAD_DOMAINS: dict[str, set[str]] = {
    "opensubtitles": {"opensubtitles.com", "opensubtitles.org"},
    "podnapisi": {"podnapisi.net"},
    "jimaku": {"jimaku.cc"},
    "addic7ed": {"addic7ed.com"},
    "betaseries": {"betaseries.com"},
    "gestdown": {"gestdown.info"},
    "kitsunekko": {"kitsunekko.net"},
    "legendasdivx": {"legendasdivx.pt"},
    "napisy24": {"napisy24.pl"},
    "subdl": {"subdl.com"},
    # animetosho.net serves the actual files: an index hit on .org/.xyz hands
    # out a download URL on storage.animetosho.net, and without it here every
    # such result is rejected at the last step. Prod 2026-09-27/28 lost 81
    # downloads in 24 h that way while .org-hosted ones kept working, so the
    # gap was invisible in the success count.
    "animetosho": {"animetosho.org", "animetosho.xyz", "animetosho.net"},
    # subf2m.co 302s the actual archive to its signed CDN.
    "subf2m": {"subf2m.co", "isubcdn.com"},
    "subsource": {"subsource.net"},
    "titlovi": {"titlovi.com"},
    "titrari": {"titrari.ro"},
    "tvsubtitles": {"tvsubtitles.net"},
    "turkcealtyazi": {"turkcealtyazi.org"},
    "yifysubtitles": {"yifysubtitles.ch"},
    "zimuku": {"zimuku.net"},
}


def validate_download_url(url: str, provider_name: str) -> tuple[bool, str | None]:
    """Validate that a provider download URL points to an allowed domain (P1 SSRF guard).

    Returns:
        (True, None) if the URL is safe to fetch.
        (False, reason) if the URL should be rejected.
    """
    if provider_name in _LOCAL_PROVIDERS:
        return True, None

    if _is_self_hosted_provider(provider_name):
        if not url:
            return False, "Self-hosted provider download URL must not be empty"
        try:
            parsed = urllib.parse.urlparse(url)
        except Exception:
            return False, "URL could not be parsed"
        if parsed.scheme not in _ALLOWED_SERVICE_SCHEMES:
            return False, f"Invalid scheme {parsed.scheme!r} — only http/https are allowed"
        host = parsed.hostname
        if not host:
            return False, "URL has no hostname"
        if host.lower() in _BLOCKED_METADATA_HOSTS:
            return False, f"Blocked metadata host: {host!r}"
        try:
            addr = ipaddress.ip_address(host)
            if addr.is_link_local or addr.is_loopback:
                return False, f"Link-local/loopback addresses are not allowed: {host!r}"
            for network in _METADATA_NETWORKS:
                if addr in network:
                    return False, f"Blocked metadata IP range: {host!r}"
        except ValueError:
            pass  # hostname, not an IP — OK
        return True, None

    allowed_domains = _PROVIDER_DOWNLOAD_DOMAINS.get(provider_name)
    if allowed_domains is None:
        return False, f"Unknown provider {provider_name!r} — no download domain allowlist defined"

    if not url:
        return False, "Download URL must not be empty"

    try:
        parsed = urllib.parse.urlparse(url)
    except Exception:
        return False, "URL could not be parsed"

    if parsed.scheme not in _ALLOWED_SERVICE_SCHEMES:
        return False, f"Invalid scheme {parsed.scheme!r} — only http/https are allowed"

    host = parsed.hostname
    if not host:
        return False, "URL has no hostname"

    if host.lower() in _BLOCKED_METADATA_HOSTS:
        return False, f"Blocked metadata host: {host!r}"
    try:
        addr = ipaddress.ip_address(host)
        if addr.is_link_local:
            return False, f"Link-local addresses are not allowed: {host!r}"
        for network in _METADATA_NETWORKS:
            if addr in network:
                return False, f"Blocked metadata IP range: {host!r}"
    except ValueError:
        pass

    netloc = host.lower()
    for allowed in allowed_domains:
        if netloc == allowed or netloc.endswith("." + allowed):
            return True, None

    return False, (
        f"Download URL domain {netloc!r} is not in the allowlist for provider {provider_name!r}"
    )


def safe_subprocess_arg(path: str) -> str:
    """Return ``path`` made safe to pass as a positional argv to an external
    binary, by prefixing ``./`` when a *relative* path's basename starts with
    ``-``.

    subprocess with a list avoids shell parsing, but the called binary still
    does its own getopt-style flag parsing, so a file literally named
    ``-i.mkv`` would be mistaken for a flag. Prefixing ``./`` keeps the path
    semantically identical while removing the ambiguity. Absolute paths (which
    start with the path separator, not ``-``) are returned unchanged.
    """
    if not path:
        return path
    if os.path.isabs(path):
        return path
    base = os.path.basename(path)
    if base.startswith("-"):
        return os.path.join(".", path)
    return path


def is_safe_path(file_path: str, base_dir: str) -> bool:
    """Return True iff file_path resolves inside base_dir (symlinks resolved).

    Args:
        file_path: Path to validate.
        base_dir: Allowed base directory.

    A path the operating system refuses to look at — an embedded NUL is the
    one that reaches us from a query string — is not safe, so it is answered
    with False rather than an exception. Letting ValueError out made every
    caller responsible for catching it, and the one that did not turned a
    rejected path into HTTP 500 (found 2026-09-03 on the deployed rc.5,
    `file_path=/media%00/../../etc/passwd`). POSIX-only: Windows'
    ``realpath`` does not raise on a NUL, so this can only be observed on a
    Linux host — which is every install.

    A relative file_path under an absolute base_dir is not safe either: it
    resolves against the process's working directory, not against base_dir,
    so the answer depended on where the process was started (Bite run
    2026-11-06). Both relative still works: both resolve against the same
    working directory, which a relative config_dir relies on.

    Returns:
        True if file_path is inside base_dir after resolving symlinks.
    """
    if os.path.isabs(base_dir) and not os.path.isabs(file_path):
        return False
    try:
        real_path = os.path.realpath(file_path)
        real_base = os.path.realpath(base_dir)
    except (ValueError, OSError):
        return False
    # Normalize real_base to always end with sep so startswith works correctly
    # for root paths like "E:\" that already end with the separator.
    if not real_base.endswith(os.sep):
        real_base = real_base + os.sep
    return real_path.startswith(real_base) or real_path + os.sep == real_base


def safe_zip_extract(zip_file: zipfile.ZipFile, dest_dir: str) -> None:
    """Extract zip_file into dest_dir, rejecting any ZIP Slip entries.

    Args:
        zip_file: Open ZipFile object.
        dest_dir: Target extraction directory.

    Raises:
        ValueError: If any entry resolves outside dest_dir (ZIP Slip).
    """
    dest_real = os.path.realpath(dest_dir)
    for member in zip_file.namelist():
        target = os.path.realpath(os.path.join(dest_real, member))
        if not (target.startswith(dest_real + os.sep) or target == dest_real):
            raise ValueError(f"ZIP Slip detected: {member!r}")
    zip_file.extractall(dest_dir)


def _normalise_to_ip(host: str) -> "ipaddress.IPv4Address | ipaddress.IPv6Address | None":
    """Coerce ``host`` into a canonical ``IPv4Address``/``IPv6Address`` if possible.

    Handles bypass-prone forms that ``ipaddress.ip_address`` rejects out of
    the box (pentest 2026-05-08 finding):
    - Decimal-encoded IPv4: ``"2130706433"`` (= 127.0.0.1) and ``"2852039166"``
      (= 169.254.169.254 / AWS metadata).
    - Octal/hex IPv4: ``"0177.0.0.1"``, ``"0x7f.0.0.1"``.
    - IPv4-mapped IPv6: ``"::ffff:127.0.0.1"`` is unwrapped to its v4 form so
      loopback/link-local checks fire.

    Returns ``None`` for true hostnames (DNS) — caller decides how to handle.
    """
    # First try canonical dotted-decimal / canonical IPv6.
    try:
        addr = ipaddress.ip_address(host)
    except ValueError:
        addr = None

    if addr is None:
        # Try non-canonical IPv4 forms (octal, hex, decimal int).
        # socket.inet_aton accepts decimal int strings, octal "0177.0.0.1",
        # and hex "0x7f.0.0.1". A pure host name still raises OSError.
        try:
            packed = socket.inet_aton(host)
            addr = ipaddress.IPv4Address(packed)
        except OSError:
            return None

    # Unwrap IPv4-mapped IPv6 ("::ffff:127.0.0.1" or "::ffff:7f00:1") so
    # the loopback / metadata checks see the underlying IPv4.
    if isinstance(addr, ipaddress.IPv6Address):
        v4 = addr.ipv4_mapped
        if v4 is not None:
            return v4

    return addr


def validate_service_url(url: str, *, allow_loopback: bool = False) -> tuple[bool, str | None]:
    """Validate that a service/provider URL is safe (SSRF prevention).

    Blocks dangerous schemes (file://, ftp://, gopher://, etc.), cloud
    metadata endpoints, loopback addresses, link-local addresses, and the
    unspecified address (0.0.0.0 / ::). Private LAN addresses (RFC 1918)
    are allowed because Sublarr's primary use case is reaching homelab
    Sonarr/Radarr/Ollama instances on 10/8, 172.16/12, 192.168/16.

    Hostname resolution: when ``host`` is a DNS name we resolve it to all
    A/AAAA records and re-run the IP checks against each — this catches
    "localhost", "metadata.google.internal" pointing at metadata, etc.
    DNS rebinding (validate-time vs fetch-time mismatch) is NOT prevented
    here; the caller is expected to pin the resolved IP if that matters.

    Args:
        url: URL to validate (e.g. Sonarr, Radarr, Ollama endpoint).
        allow_loopback: accept 127.0.0.0/8 and ::1 — for a service that is
            commonly run on the same host as Sublarr (Ollama on a bare-metal
            install). Metadata, link-local and unspecified stay blocked.

    Returns:
        (True, None) if the URL is acceptable.
        (False, reason) if the URL is rejected, where reason is a human-readable string.
    """
    if not url:
        return False, "URL must not be empty"

    try:
        parsed = urllib.parse.urlparse(url)
    except Exception:
        return False, "URL could not be parsed"

    if parsed.scheme not in _ALLOWED_SERVICE_SCHEMES:
        return False, f"Invalid scheme {parsed.scheme!r} — only http/https are allowed"

    host = parsed.hostname
    if not host:
        return False, "URL has no hostname"

    # Block cloud metadata hostnames before any IP coercion.
    if host.lower() in _BLOCKED_METADATA_HOSTS:
        return False, f"Blocked metadata host: {host!r}"

    # Try to coerce the host into an IP, including non-canonical forms
    # (decimal/octal/hex IPv4, IPv4-mapped IPv6) that the original
    # urlparse → ip_address pipeline silently allowed (pentest finding).
    addr = _normalise_to_ip(host)

    if addr is not None:
        ok, reason = _check_ip_address(addr, host, allow_loopback=allow_loopback)
        if not ok:
            return False, reason
        return True, None

    # Hostname (not an IP literal) — resolve and re-check every answer.
    # Fail-open on DNS resolution errors so legitimate homelab hostnames
    # like ``sonarr.local`` (mDNS only resolvable on the home LAN, not on
    # the dev/CI box) still pass validation. The trade-off: an attacker
    # who controls a hostname can theoretically register an A record only
    # at fetch-time (DNS rebinding) to bypass — caller pins the resolved
    # IP if that matters. Documented as deferred above.
    try:
        infos = socket.getaddrinfo(host, parsed.port or None, proto=socket.IPPROTO_TCP)
    except (socket.gaierror, OSError):
        return True, None

    seen: set[str] = set()
    for info in infos:
        sockaddr = info[4]
        ip_str = sockaddr[0]
        if ip_str in seen:
            continue
        seen.add(ip_str)
        try:
            resolved = ipaddress.ip_address(ip_str)
        except ValueError:
            continue
        if isinstance(resolved, ipaddress.IPv6Address) and resolved.ipv4_mapped is not None:
            resolved = resolved.ipv4_mapped
        ok, reason = _check_ip_address(
            resolved, f"{host} ({ip_str})", allow_loopback=allow_loopback
        )
        if not ok:
            return False, reason

    return True, None


_MAX_CHECKED_REDIRECTS = 5


def get_with_checked_redirects(url: str, *, https_only: bool = False, **kwargs):
    """``requests.get`` that validates every redirect hop like the first URL.

    ``validate_service_url`` only sees the URL it is given; ``requests``
    follows redirects on its own, so an allowed host could bounce the request
    to a blocked one (code check 2026-11-05, N-2). Redirects are followed
    here, each target checked before it is fetched. DNS rebinding between
    check and fetch is still not covered (see ``validate_service_url``).

    Raises ``ValueError`` for a blocked hop; HTTP errors are left to the caller.
    """
    import requests

    for _ in range(_MAX_CHECKED_REDIRECTS + 1):
        if https_only and urllib.parse.urlparse(url).scheme != "https":
            raise ValueError(f"Only HTTPS is allowed: {url}")
        ok, reason = validate_service_url(url)
        if not ok:
            raise ValueError(f"URL blocked for security reasons: {reason}")
        response = requests.get(url, allow_redirects=False, **kwargs)
        if not response.is_redirect:
            return response
        url = urllib.parse.urljoin(url, response.headers["Location"])
        response.close()
    raise ValueError(f"Too many redirects (more than {_MAX_CHECKED_REDIRECTS})")


def _check_ip_address(addr, label: str, *, allow_loopback: bool = False) -> tuple[bool, str | None]:
    """Return (False, reason) for any IP class we never want service URLs to point at.

    Used by ``validate_service_url`` for both literal IPs and DNS-resolved
    hostnames so the bypass surface is identical no matter how the host
    was expressed in the URL.
    """
    if addr.is_unspecified:
        return False, f"Unspecified address (0.0.0.0 / ::) is not a valid service host: {label!r}"
    if addr.is_loopback and not allow_loopback:
        return False, f"Loopback addresses are not allowed: {label!r}"
    if addr.is_link_local:
        return False, f"Link-local addresses are not allowed: {label!r}"
    for network in _METADATA_NETWORKS:
        if addr in network:
            return False, f"Blocked metadata IP range: {label!r}"
    return True, None


def validate_git_url(url: str) -> None:
    """Validate that a Git repository URL is safe to clone.

    Only HTTPS URLs on allow-listed domains are permitted.

    Args:
        url: Git repository URL to validate.

    Raises:
        ValueError: If the URL uses a non-HTTPS scheme or a non-allowed domain.
    """
    parsed = urllib.parse.urlparse(url)
    if parsed.scheme != "https":
        raise ValueError(f"Git URL must use HTTPS, got: {parsed.scheme!r}")
    domain = parsed.netloc.lower().split(":")[0]
    if not any(domain == d or domain.endswith("." + d) for d in _ALLOWED_GIT_DOMAINS):
        raise ValueError(f"Git URL domain not in allowlist: {domain!r}")


def sanitize_env_value(value: str) -> str:
    """Sanitize a string for safe use as an environment variable value.

    Strips newlines and null bytes that could enable env-var injection,
    and truncates to a safe maximum length.

    Args:
        value: Raw string to sanitize.

    Returns:
        Sanitized string safe for use as an env var value.
    """
    return value.replace("\n", " ").replace("\r", " ").replace("\x00", "")[:1024]


def api_key_actor(api_key: str | None) -> str:
    """Actor label for an audit log line: a short SHA-256 fingerprint of the key.

    The admin audit lines used to log the first six characters of the API key,
    which puts part of a live credential into every log and support bundle.
    A fingerprint still tells two callers apart and matches across lines, and
    gives away nothing about the key itself.
    """
    if not api_key:
        return "anon"
    return "key-" + hashlib.sha256(api_key.encode("utf-8")).hexdigest()[:8]