"""SSRF, allow-list, DNS and redirect-target validation tests."""

from __future__ import annotations

import socket

import pytest

from app.errors import (
    DomainNotAllowedError,
    InvalidUrlError,
    NetworkError,
    SsrfBlockedError,
)
from app.services.url_validator import NetworkGuard, ValidatingResolver

PUBLIC_V4 = "93.184.216.34"
PUBLIC_V4_OTHER = "1.1.1.1"
PUBLIC_V6 = "2606:4700:4700::1111"


def _resolver_for(*addresses: str, family: int = socket.AF_INET):
    async def resolver(host: str, port: int):
        return [
            (family, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", (address, port, 0, 0))
            for address in addresses
        ]

    return resolver


@pytest.fixture
def guard() -> NetworkGuard:
    return NetworkGuard(["example.com"], resolver=_resolver_for(PUBLIC_V4))


@pytest.mark.asyncio
async def test_allowed_domain_is_normalized(guard: NetworkGuard) -> None:
    result = await guard.validate(
        "HTTPS://Media.Example.com.:443/videos/a.mp4?x=1#fragment"
    )
    assert result.host == "media.example.com"
    assert result.url == "https://media.example.com/videos/a.mp4?x=1"
    assert result.port == 443
    assert result.addresses == (PUBLIC_V4,)
    assert result.original.startswith("HTTPS://")


@pytest.mark.asyncio
async def test_default_http_port_is_omitted_from_normalized_url(guard: NetworkGuard) -> None:
    result = await guard.validate("http://example.com:80", resolve_dns=False)
    assert result.url == "http://example.com/"
    result = await guard.validate("http://example.com:8080/a", resolve_dns=False)
    assert result.url == "http://example.com:8080/a"


@pytest.mark.asyncio
async def test_domain_boundary_is_not_suffix_matching(guard: NetworkGuard) -> None:
    assert guard.is_domain_allowed("example.com")
    assert guard.is_domain_allowed("deep.media.example.com")
    assert not guard.is_domain_allowed("notexample.com")
    assert not guard.is_domain_allowed("example.com.evil.test")
    assert not guard.is_domain_allowed("evil.test")


@pytest.mark.asyncio
async def test_disallowed_domain_never_resolves() -> None:
    calls: list[str] = []

    async def resolver(host: str, port: int):
        calls.append(host)
        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (PUBLIC_V4, port))]

    guard = NetworkGuard(["example.com"], resolver=resolver)
    with pytest.raises(DomainNotAllowedError):
        await guard.validate("https://evil.test/video.mp4")
    assert calls == []


@pytest.mark.parametrize(
    "url",
    [
        "",
        "   ",
        "ftp://example.com/a",
        "file:///etc/passwd",
        "javascript:alert(1)",
        "//example.com/a",
        "http:///path",
        "http://user:pass@example.com/a",
        "http://user@example.com/a",
        "http://example.com\\@evil.test/a",
        "http://example.com/a\nb",
        "http://example.com/a\x00b",
        "http://example.com:0/a",
        "http://example.com:65536/a",
        "http://example.com:notaport/a",
        "https://example.com:https/a",
        "http://" + "a" * 5000 + ".com/",
    ],
)
@pytest.mark.asyncio
async def test_structurally_invalid_urls_are_rejected(guard: NetworkGuard, url: str) -> None:
    with pytest.raises(InvalidUrlError):
        await guard.validate(url)


@pytest.mark.parametrize(
    "url",
    [
        "http://localhost/a",
        "http://LOCALHOST./a",
        "http://foo.local/a",
        "http://printer.local./a",
        "http://metadata.google.internal/computeMetadata/v1/",
        "http://127.0.0.1/a",
        "http://127.1/a",
        "http://0.0.0.0/a",
        "http://10.1.2.3/a",
        "http://192.168.1.5/a",
        "http://169.254.169.254/latest/meta-data/",
        "http://[::1]/a",
        "http://[fe80::1]/a",
        "http://[::ffff:127.0.0.1]/a",
        "http://100.64.0.1/a",
        "http://224.0.0.1/a",
    ],
)
@pytest.mark.asyncio
async def test_private_and_internal_hosts_are_blocked(guard: NetworkGuard, url: str) -> None:
    with pytest.raises(SsrfBlockedError):
        await guard.validate(url)


@pytest.mark.asyncio
async def test_private_dns_answer_blocks_even_allowed_name() -> None:
    guard = NetworkGuard(["example.com"], resolver=_resolver_for("10.0.0.7"))
    with pytest.raises(SsrfBlockedError):
        await guard.validate("https://example.com/a")


@pytest.mark.asyncio
async def test_one_private_answer_among_public_answers_blocks() -> None:
    guard = NetworkGuard(
        ["example.com"],
        resolver=_resolver_for(PUBLIC_V4, "192.168.0.1"),
    )
    with pytest.raises(SsrfBlockedError):
        await guard.validate("https://example.com/a")


@pytest.mark.asyncio
async def test_public_ipv6_is_allowed_and_resolver_reports_family() -> None:
    guard = NetworkGuard(
        ["example.com"],
        resolver=_resolver_for(PUBLIC_V6, family=socket.AF_INET6),
    )
    validated = await guard.validate("https://example.com/a")
    assert validated.addresses == (PUBLIC_V6,)

    records = await ValidatingResolver(guard).resolve("example.com", 443)
    assert len(records) == 1
    assert records[0]["family"] == socket.AF_INET6
    assert records[0]["host"] == PUBLIC_V6
    assert records[0]["sockaddr"] == (PUBLIC_V6, 443, 0, 0)


@pytest.mark.asyncio
async def test_ipv4_mapped_ipv6_loopback_is_blocked() -> None:
    guard = NetworkGuard(["example.com"], resolver=_resolver_for("::ffff:127.0.0.1"))
    with pytest.raises(SsrfBlockedError):
        await guard.validate("https://example.com/a")


@pytest.mark.asyncio
async def test_public_ip_literal_works_only_when_allow_listed() -> None:
    guard = NetworkGuard(["8.8.8.8"], resolver=_resolver_for(PUBLIC_V4))
    validated = await guard.validate("http://8.8.8.8/dns-query")
    assert validated.host == "8.8.8.8"
    assert validated.addresses == ("8.8.8.8",)

    other = NetworkGuard(["example.com"], resolver=_resolver_for(PUBLIC_V4))
    with pytest.raises(DomainNotAllowedError):
        await other.validate("http://8.8.8.8/")


@pytest.mark.asyncio
async def test_dns_answers_are_cached_until_ttl() -> None:
    answers = [PUBLIC_V4, PUBLIC_V4_OTHER]
    calls = 0

    async def resolver(host: str, port: int):
        nonlocal calls
        address = answers[min(calls, len(answers) - 1)]
        calls += 1
        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (address, port))]

    guard = NetworkGuard(["example.com"], resolver=resolver, dns_cache_ttl=60)
    first = await guard.validate("https://example.com/a")
    second = await guard.validate("https://example.com/b")
    assert calls == 1
    assert first.addresses == second.addresses == (PUBLIC_V4,)
    assert await guard.resolve_addresses("example.com", 443) == (PUBLIC_V4,)

    uncached = NetworkGuard(["example.com"], resolver=resolver, dns_cache_ttl=0)
    assert (await uncached.resolve_addresses("example.com", 443)) == (PUBLIC_V4_OTHER,)
    assert (await uncached.resolve_addresses("example.com", 443)) == (PUBLIC_V4_OTHER,)


@pytest.mark.asyncio
async def test_resolver_errors_become_network_errors() -> None:
    async def broken(host: str, port: int):
        raise OSError("dns failure")

    guard = NetworkGuard(["example.com"], resolver=broken)
    with pytest.raises(NetworkError):
        await guard.validate("https://example.com/a")


@pytest.mark.asyncio
async def test_empty_dns_answer_is_a_network_error() -> None:
    async def empty(host: str, port: int):
        return []

    guard = NetworkGuard(["example.com"], resolver=empty)
    with pytest.raises(NetworkError):
        await guard.validate("https://example.com/a")


@pytest.mark.asyncio
async def test_dns_invalid_address_is_ssrf_blocked() -> None:
    async def garbage(host: str, port: int):
        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("not-an-ip", port))]

    guard = NetworkGuard(["example.com"], resolver=garbage)
    with pytest.raises(SsrfBlockedError):
        await guard.validate("https://example.com/a")


def test_host_only_validation_skips_dns() -> None:
    calls = 0

    async def resolver(host: str, port: int):
        nonlocal calls
        calls += 1
        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (PUBLIC_V4, port))]

    guard = NetworkGuard(["example.com"], resolver=resolver)
    validated = guard.validate_host_only("https://EXAMPLE.com./clip.mp4")
    assert validated.host == "example.com"
    assert validated.addresses == ()
    assert validated.url == "https://example.com/clip.mp4"
    assert calls == 0

    with pytest.raises(DomainNotAllowedError):
        guard.validate_host_only("https://evil.test/clip.mp4")
    with pytest.raises(InvalidUrlError):
        guard.validate_host_only("https://example.com:0/clip.mp4")


def test_guard_requires_an_allow_list() -> None:
    with pytest.raises(ValueError):
        NetworkGuard([])


def test_idna_hosts_are_normalized() -> None:
    guard = NetworkGuard(["münich.example"], resolver=_resolver_for(PUBLIC_V4))
    assert guard.is_domain_allowed("xn--mnich-kva.example")
    assert guard.validate_host_only("https://bücher.münich.example/a").host == (
        "xn--bcher-kva.xn--mnich-kva.example"
    )


@pytest.mark.asyncio
async def test_validating_resolver_ipv4_sockaddr() -> None:
    guard = NetworkGuard(["example.com"], resolver=_resolver_for(PUBLIC_V4))
    records = await ValidatingResolver(guard).resolve("example.com", 8443)
    assert records == [
        {
            "hostname": "example.com",
            "host": PUBLIC_V4,
            "port": 8443,
            "family": socket.AF_INET,
            "proto": socket.IPPROTO_TCP,
            "flags": 0,
            "sockaddr": (PUBLIC_V4, 8443),
        }
    ]
    assert await ValidatingResolver(guard).close() is None