from __future__ import annotations

import re
from dataclasses import dataclass
from ipaddress import IPv4Address, IPv6Address
from urllib.parse import unquote, urlsplit, urlunsplit

import idna


@dataclass(frozen=True)
class NormalizedOrigin:
    scheme: str
    hostname: str
    port: int

    @property
    def authority(self) -> str:
        host = f"[{self.hostname}]" if ":" in self.hostname else self.hostname
        default_port = 443 if self.scheme == "https" else 80
        return host if self.port == default_port else f"{host}:{self.port}"

    @property
    def url(self) -> str:
        return f"{self.scheme}://{self.authority}"


def _contains_disallowed_ascii_control(value: str, *, allow_space: bool = False) -> bool:
    minimum = 32 if allow_space else 33
    return any(ord(character) < minimum or ord(character) == 127 for character in value)


def _is_clean_url_value(value: object) -> bool:
    return (
        isinstance(value, str)
        and bool(value)
        and value == value.strip()
        and "%" not in value
        and "\\" not in value
        and not _contains_disallowed_ascii_control(value)
    )


def _normalize_hostname(hostname: str) -> str | None:
    if ":" in hostname:
        try:
            return IPv6Address(hostname).compressed
        except ValueError:
            return None

    # Python's built-in IDNA codec uses transitional mappings that differ from
    # browser URL parsing (for example, "faß.de" aliases "fass.de"). Require
    # administrators to provide the browser-serialized ASCII/punycode hostname.
    if not hostname.isascii():
        return None
    ascii_hostname = hostname.lower()
    if ascii_hostname.endswith("."):
        return None

    if re.fullmatch(r"\d+(?:\.\d+){3}", ascii_hostname):
        try:
            return str(IPv4Address(ascii_hostname))
        except ValueError:
            return None

    if len(ascii_hostname) > 253:
        return None
    labels = ascii_hostname.split(".")
    # Numeric final labels are browser IPv4 aliases. Only the canonical
    # dotted-quad form handled above is accepted.
    final_label = labels[-1] if labels else ""
    if re.fullmatch(r"\d+", final_label) or re.fullmatch(r"0x[0-9a-f]*", final_label):
        return None
    if any(
        not label or len(label) > 63 or re.fullmatch(r"[a-z0-9](?:[a-z0-9-]*[a-z0-9])?", label) is None
        for label in labels
    ):
        return None
    try:
        if any(label.startswith("xn--") and idna.alabel(label).decode("ascii") != label for label in labels):
            return None
    except idna.IDNAError:
        return None
    return ascii_hostname


def normalize_origin(origin: object) -> NormalizedOrigin | None:
    if not _is_clean_url_value(origin):
        return None

    try:
        parsed = urlsplit(origin)
    except (TypeError, ValueError):
        return None

    try:
        hostname = parsed.hostname
        explicit_port = parsed.port
    except ValueError:
        return None

    scheme = parsed.scheme.lower()
    if (
        not parsed.netloc
        or not hostname
        or parsed.username is not None
        or parsed.password is not None
        or parsed.path not in {"", "/"}
        or parsed.query
        or parsed.fragment
    ):
        return None

    normalized_hostname = _normalize_hostname(hostname)
    if normalized_hostname is None or (explicit_port is not None and explicit_port < 1):
        return None
    if scheme not in {"http", "https"}:
        return None
    port = explicit_port or (443 if scheme == "https" else 80)

    return NormalizedOrigin(scheme=scheme, hostname=normalized_hostname, port=port)


def _decode_path(path: str) -> str | None:
    decoded = path
    for _ in range(8):
        next_decoded = unquote(decoded)
        if next_decoded == decoded:
            return decoded
        decoded = next_decoded
    return None


def normalize_path(path: str | None) -> str | None:
    decoded = _decode_path(path or "/")
    if decoded is None:
        return None

    if _contains_disallowed_ascii_control(decoded, allow_space=True):
        return None

    # Do not repair ambiguous separators. Browsers and reverse proxies may
    # normalize them differently after authorization.
    if "\\" in decoded or "//" in decoded:
        return None
    if not decoded.startswith("/"):
        decoded = f"/{decoded}"
    if any(segment in {".", ".."} for segment in decoded.split("/")):
        return None
    return decoded or "/"


def normalize_path_prefix(prefix: object) -> str | None:
    if not isinstance(prefix, str):
        return None

    trimmed = prefix.strip()
    if not trimmed:
        return ""
    if trimmed == "/":
        return "/"
    if trimmed != prefix or not trimmed.startswith("/") or trimmed.startswith("//"):
        return None

    if trimmed.endswith("/*"):
        base_prefix = trimmed[:-2]
    elif "*" in trimmed:
        return None
    else:
        base_prefix = trimmed

    if not base_prefix or base_prefix.endswith("/") or "%" in base_prefix or "\\" in base_prefix:
        return None

    normalized = normalize_path(base_prefix)
    if normalized is None or normalized == "/" or normalized != base_prefix:
        return None
    return normalized


def normalize_embed_url(url: object) -> tuple[str, NormalizedOrigin, str] | None:
    if not isinstance(url, str) or not url or url != url.strip():
        return None

    try:
        parsed = urlsplit(url)
    except ValueError:
        return None

    origin = normalize_origin(f"{parsed.scheme}://{parsed.netloc}")
    path = normalize_path(parsed.path)
    if origin is None or path is None:
        return None

    # Use the canonical path only for authorization. Preserve the submitted
    # encoded path in the destination so validation never redirects to a
    # different resource.
    destination_path = parsed.path or "/"
    normalized_url = urlunsplit((origin.scheme, origin.authority, destination_path, parsed.query, ""))
    return normalized_url, origin, path


def _path_pattern_key(pattern: object) -> tuple[str, bool] | None:
    if not isinstance(pattern, str):
        return None
    normalized_prefix = normalize_path_prefix(pattern)
    if not normalized_prefix:
        return None
    return normalized_prefix, pattern.endswith("/*")


def validate_embed_origins(entries: object, *, field_name: str) -> list[dict]:
    errors: list[dict] = []
    if not isinstance(entries, list):
        return errors

    seen_origins: dict[NormalizedOrigin, int] = {}
    for index, entry in enumerate(entries):
        if not isinstance(entry, dict):
            continue
        normalized_origin = normalize_origin(entry.get("origin"))
        if normalized_origin is None:
            errors.append(
                {
                    "type": "value_error",
                    "loc": [field_name, index, "origin"],
                    "msg": ("Origin must be an HTTP or HTTPS origin without credentials, path, query, or fragment"),
                }
            )
            continue

        first_index = seen_origins.get(normalized_origin)
        if first_index is not None:
            errors.append(
                {
                    "type": "value_error",
                    "loc": [field_name, index, "origin"],
                    "msg": f"Origin duplicates the normalized origin at index {first_index}",
                }
            )
            continue
        seen_origins[normalized_origin] = index
    return errors


def validate_embed_path_prefixes(allowed_entries: object) -> list[dict]:
    errors: list[dict] = []

    if isinstance(allowed_entries, list):
        for origin_index, entry in enumerate(allowed_entries):
            if not isinstance(entry, dict):
                continue

            rules = entry.get("rules")
            seen_rule_patterns: dict[tuple[str, bool], int] = {}
            for rule_index, rule in enumerate(rules if isinstance(rules, list) else []):
                if not isinstance(rule, dict):
                    continue

                pattern_key = _path_pattern_key(rule.get("pathPrefix"))
                if isinstance(rule.get("pathPrefix"), str) and rule["pathPrefix"].strip() and pattern_key is None:
                    errors.append(
                        {
                            "type": "value_error",
                            "loc": [
                                "allowedEmbedOrigins",
                                origin_index,
                                "rules",
                                rule_index,
                                "pathPrefix",
                            ],
                            "msg": (
                                "Path prefix must start with '/'. Without '*', only the exact path matches. "
                                "Use '/' for the whole origin or a terminal '/*' for a subtree."
                            ),
                        }
                    )
                    continue
                if pattern_key is None:
                    continue

                first_index = seen_rule_patterns.get(pattern_key)
                if first_index is not None:
                    errors.append(
                        {
                            "type": "value_error",
                            "loc": [
                                "allowedEmbedOrigins",
                                origin_index,
                                "rules",
                                rule_index,
                                "pathPrefix",
                            ],
                            "msg": f"Path rule duplicates the rule at index {first_index}",
                        }
                    )
                    continue

                seen_rule_patterns[pattern_key] = rule_index

            excluded_prefixes = entry.get("excludedPathPrefixes")
            seen_excluded_patterns: dict[tuple[str, bool], int] = {}
            for prefix_index, prefix in enumerate(excluded_prefixes if isinstance(excluded_prefixes, list) else []):
                pattern_key = _path_pattern_key(prefix)
                if isinstance(prefix, str) and prefix.strip() and pattern_key is None:
                    errors.append(
                        {
                            "type": "value_error",
                            "loc": [
                                "allowedEmbedOrigins",
                                origin_index,
                                "excludedPathPrefixes",
                                prefix_index,
                            ],
                            "msg": (
                                "Excluded path must start with '/'. Without '*', only the exact path matches. "
                                "Use '/' for the whole origin or a terminal '/*' for a subtree."
                            ),
                        }
                    )
                    continue
                if pattern_key is None:
                    continue

                first_index = seen_excluded_patterns.get(pattern_key)
                if first_index is not None:
                    errors.append(
                        {
                            "type": "value_error",
                            "loc": [
                                "allowedEmbedOrigins",
                                origin_index,
                                "excludedPathPrefixes",
                                prefix_index,
                            ],
                            "msg": f"Excluded path duplicates the entry at index {first_index}",
                        }
                    )
                    continue

                seen_excluded_patterns[pattern_key] = prefix_index

    return errors


def validate_embed_query_parameters(allowed_entries: object) -> list[dict]:
    errors: list[dict] = []
    if not isinstance(allowed_entries, list):
        return errors

    for origin_index, entry in enumerate(allowed_entries):
        if not isinstance(entry, dict) or not isinstance(entry.get("rules"), list):
            continue
        for rule_index, rule in enumerate(entry["rules"]):
            if not isinstance(rule, dict):
                continue
            parameters = rule.get("allowedQueryParameters")
            if not isinstance(parameters, list):
                continue
            seen: set[str] = set()
            for parameter_index, parameter in enumerate(parameters):
                if (
                    not isinstance(parameter, str)
                    or not parameter
                    or parameter != parameter.strip()
                    or any(character in parameter for character in "?#&=%+; \t\r\n")
                    or parameter in seen
                ):
                    errors.append(
                        {
                            "type": "value_error",
                            "loc": [
                                "allowedEmbedOrigins",
                                origin_index,
                                "rules",
                                rule_index,
                                "allowedQueryParameters",
                                parameter_index,
                            ],
                            "msg": "Query parameter names must be unique names without URL separators or whitespace",
                        }
                    )
                elif parameter:
                    seen.add(parameter)
    return errors
