from __future__ import annotations

import io
import unittest
from unittest.mock import patch

from govoplan_core.security.http_fetch import _PolicyRedirectHandler, is_http_url, validate_http_url
from govoplan_core.security.outbound_http import (
    DEFAULT_FILE_TRANSFER_BYTES,
    DEFAULT_STRUCTURED_RESPONSE_BYTES,
    OutboundHttpBlocked,
    OutboundResponseTooLarge,
    bounded_chunks_bytes,
    bounded_response_bytes,
    create_outbound_connection,
    outbound_http_policy,
    validate_outbound_http_url,
    validate_unpinned_sdk_host,
    validate_unpinned_sdk_http_url,
)


class HttpFetchTests(unittest.TestCase):
    def test_validate_http_url_accepts_absolute_http_urls_without_credentials(self) -> None:
        self.assertEqual("https://example.test/catalog.json", validate_http_url("https://example.test/catalog.json"))
        self.assertTrue(is_http_url("http://example.test/catalog.json"))

    def test_validate_http_url_rejects_non_http_urls_and_embedded_credentials(self) -> None:
        for value in (
            "file:///etc/passwd",
            "/relative/catalog.json",
            "https://user@example.test/catalog.json",
            "https://user:secret@example.test/catalog.json",
        ):
            with self.subTest(value=value):
                self.assertFalse(is_http_url(value))
                with self.assertRaises(ValueError):
                    validate_http_url(value)

    def test_outbound_policy_has_practical_bounded_defaults(self) -> None:
        policy = outbound_http_policy({"APP_ENV": "production"})
        self.assertFalse(policy.allow_private_networks)
        self.assertEqual(DEFAULT_STRUCTURED_RESPONSE_BYTES, policy.structured_response_bytes)
        self.assertEqual(DEFAULT_FILE_TRANSFER_BYTES, policy.file_transfer_bytes)

        configured = outbound_http_policy(
            {
                "APP_ENV": "production",
                "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true",
                "GOVOPLAN_CONNECTOR_MAX_STRUCTURED_RESPONSE_BYTES": "1024",
                "GOVOPLAN_CONNECTOR_MAX_FILE_TRANSFER_BYTES": "2048",
            }
        )
        self.assertTrue(configured.allow_private_networks)
        self.assertEqual(1024, configured.structured_response_bytes)
        self.assertEqual(2048, configured.file_transfer_bytes)

    def test_private_network_policy_is_deployment_wide(self) -> None:
        denied_policy = outbound_http_policy({"APP_ENV": "production"})
        with patch(
            "govoplan_core.security.outbound_http.socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("127.0.0.1", 443))],
        ), self.assertRaisesRegex(OutboundHttpBlocked, "non-public network"):
            validate_outbound_http_url("https://connector.example.test/path", policy=denied_policy)

        allowed_policy = outbound_http_policy(
            {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"}
        )
        with patch(
            "govoplan_core.security.outbound_http.socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("127.0.0.1", 443))],
        ):
            self.assertEqual(
                "https://127.0.0.1/path",
                validate_outbound_http_url("https://127.0.0.1/path", policy=allowed_policy),
            )

    def test_private_network_policy_still_blocks_non_peer_and_metadata_addresses(self) -> None:
        policy = outbound_http_policy(
            {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"}
        )
        for address in (
            "0.0.0.0",
            "169.254.169.254",
            "224.0.0.1",
            "255.255.255.255",
            "::",
            "::ffff:255.255.255.255",
            "64:ff9b::169.254.169.254",
            "64:ff9b::255.255.255.255",
            "fd00:ec2::254",
            "fe80::1",
            "ff02::1",
        ):
            with self.subTest(address=address), patch(
                "govoplan_core.security.outbound_http.socket.getaddrinfo",
                return_value=[(2, 1, 6, "", (address, 443))],
            ), self.assertRaisesRegex(OutboundHttpBlocked, "forbidden special-purpose network"):
                validate_outbound_http_url("https://connector.example.test/path", policy=policy)

    def test_public_policy_rejects_reserved_and_ipv4_embedded_ipv6_addresses(self) -> None:
        policy = outbound_http_policy({"APP_ENV": "production"})
        for address in (
            "fec0::1",
            "::127.0.0.1",
            "64:ff9b::8.8.8.8",
            "64:ff9b::127.0.0.1",
            "2002:7f00:1::",
        ):
            with self.subTest(address=address), patch(
                "govoplan_core.security.outbound_http.socket.getaddrinfo",
                return_value=[(10, 1, 6, "", (address, 443, 0, 0))],
            ), self.assertRaises(OutboundHttpBlocked):
                validate_outbound_http_url("https://connector.example.test/path", policy=policy)

    def test_outbound_policy_rejects_mixed_public_and_private_dns_answers(self) -> None:
        policy = outbound_http_policy({"APP_ENV": "production"})
        with patch(
            "govoplan_core.security.outbound_http.socket.getaddrinfo",
            return_value=[
                (2, 1, 6, "", ("93.184.216.34", 443)),
                (2, 1, 6, "", ("10.0.0.4", 443)),
            ],
        ), self.assertRaises(OutboundHttpBlocked):
            validate_outbound_http_url("https://connector.example.test/path", policy=policy)

    def test_bounded_response_rejects_declared_and_streamed_oversize_payloads(self) -> None:
        with self.assertRaises(OutboundResponseTooLarge):
            bounded_response_bytes(io.BytesIO(b"small"), headers={"Content-Length": "20"}, max_bytes=10)
        with self.assertRaises(OutboundResponseTooLarge):
            bounded_response_bytes(io.BytesIO(b"eleven-byte"), max_bytes=10)
        self.assertEqual(b"ten-bytes!", bounded_response_bytes(io.BytesIO(b"ten-bytes!"), max_bytes=10))
        with patch.dict("os.environ", {"GOVOPLAN_CONNECTOR_MAX_STRUCTURED_RESPONSE_BYTES": "5"}), self.assertRaises(
            OutboundResponseTooLarge
        ):
            bounded_response_bytes(io.BytesIO(b"123456"), max_bytes=10)

    def test_bounded_chunks_rejects_an_oversized_chunk_before_retaining_it(self) -> None:
        with self.assertRaises(OutboundResponseTooLarge):
            bounded_chunks_bytes((b"one-large-chunk",), max_bytes=8)

    def test_connection_time_resolution_blocks_dns_rebinding_before_socket_open(self) -> None:
        policy = outbound_http_policy({"APP_ENV": "production"})
        public = [(2, 1, 6, "", ("93.184.216.34", 443))]
        private = [(2, 1, 6, "", ("127.0.0.1", 443))]
        with patch(
            "govoplan_core.security.outbound_http.socket.getaddrinfo",
            side_effect=(public, private),
        ) as resolver, patch("govoplan_core.security.outbound_http.socket.socket") as socket_factory:
            validate_outbound_http_url("https://connector.example.test/path", policy=policy)
            with self.assertRaisesRegex(OutboundHttpBlocked, "non-public network"):
                create_outbound_connection("connector.example.test", 443, policy=policy)

        self.assertEqual(2, resolver.call_count)
        socket_factory.assert_not_called()

    def test_unpinned_sdk_endpoints_fail_closed_in_public_and_private_modes(self) -> None:
        for allow_private, address in ((False, "93.184.216.34"), (True, "10.0.0.5")):
            with self.subTest(allow_private=allow_private):
                policy = outbound_http_policy(
                    {
                        "APP_ENV": "production",
                        "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": str(allow_private).lower(),
                    }
                )
                with patch(
                    "govoplan_core.security.outbound_http.socket.getaddrinfo",
                    return_value=[(2, 1, 6, "", (address, 443))],
                ), self.assertRaisesRegex(OutboundHttpBlocked, "until that transport supports.*DNS/IP pinning"):
                    validate_unpinned_sdk_http_url(
                        "https://objects.example.test",
                        label="S3 endpoint",
                        policy=policy,
                    )

    def test_unpinned_host_sdk_fails_closed_for_dns_and_explicit_ip_targets(self) -> None:
        policy = outbound_http_policy(
            {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"}
        )
        for host in ("files.internal.example", "10.0.0.5"):
            with self.subTest(host=host), patch(
                "govoplan_core.security.outbound_http.socket.getaddrinfo",
                return_value=[(2, 1, 6, "", ("10.0.0.5", 445))],
            ), self.assertRaisesRegex(OutboundHttpBlocked, "redirects/referrals.*DNS/IP pinning"):
                validate_unpinned_sdk_host(host, port=445, label="SMB endpoint", policy=policy)

    def test_core_redirects_strip_credentials_cross_origin_and_reject_https_downgrades(self) -> None:
        import urllib.request

        request = urllib.request.Request(
            "https://catalog.example.test/releases",
            headers={"Authorization": "Bearer secret", "X-Request-ID": "request-1"},
        )
        handler = _PolicyRedirectHandler(label="Catalog URL")
        with patch.dict("os.environ", {"APP_ENV": "test"}), patch(
            "govoplan_core.security.outbound_http.socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("127.0.0.1", 443))],
        ):
            redirected = handler.redirect_request(
                request,
                None,
                302,
                "Found",
                {},
                "https://cdn.example.test/releases",
            )
            downgrade = handler.redirect_request(
                request,
                None,
                302,
                "Found",
                {},
                "http://catalog.example.test/releases",
            )

        self.assertIsNotNone(redirected)
        self.assertIsNone(redirected.get_header("Authorization"))
        self.assertEqual("request-1", redirected.get_header("X-request-id"))
        self.assertIsNone(downgrade)


if __name__ == "__main__":
    unittest.main()
