geointel/backend/tests/test_outbound_request_guard.py

"""Bounded acquisition must stay bounded to the official host.

Every acquisition service builds its URL from configured settings, so the
request payload cannot point the runtime anywhere. The redirect chain can:
``urlopen`` follows redirects by default, so a misconfigured or compromised
upstream can send the runtime to ``127.0.0.1``, to the container network, or to
a cloud metadata endpoint — and the response is then persisted as if it were
official source data.

The product's stated rule is that acquisition fails closed and never
substitutes fabricated data for official data. A redirect off the configured
host is exactly that substitution.
"""

from __future__ import annotations

import pytest

from app.core.errors import AppError
from app.services.outbound_request_guard import (
    assert_public_http_url,
    assert_same_origin_redirect,
)


class TestUrlShape:
    def test_an_official_https_endpoint_is_accepted(self) -> None:
        assert_public_http_url("https://geo.api.vlaanderen.be/dhmv/wcs?SERVICE=WCS")

    def test_a_non_http_scheme_is_refused(self) -> None:
        with pytest.raises(AppError) as exc_info:
            assert_public_http_url("file:///etc/passwd")

        assert exc_info.value.code == "OUTBOUND_URL_NOT_ALLOWED"

    @pytest.mark.parametrize(
        "url",
        [
            "http://127.0.0.1:8000/internal",
            "http://localhost/internal",
            "http://10.1.2.3/internal",
            "http://192.168.10.150/internal",
            "http://172.16.0.9/internal",
            "http://169.254.169.254/latest/meta-data/",
            "http://[::1]/internal",
        ],
    )
    def test_private_and_loopback_destinations_are_refused(self, url: str) -> None:
        with pytest.raises(AppError) as exc_info:
            assert_public_http_url(url)

        assert exc_info.value.code == "OUTBOUND_URL_NOT_ALLOWED"

    def test_a_url_without_a_host_is_refused(self) -> None:
        with pytest.raises(AppError):
            assert_public_http_url("https:///no-host")


class TestRedirects:
    def test_a_redirect_within_the_same_origin_is_allowed(self) -> None:
        assert_same_origin_redirect(
            "https://geo.api.vlaanderen.be/dhmv/wcs",
            "https://geo.api.vlaanderen.be/dhmv/wcs/v2?x=1",
        )

    def test_a_redirect_to_another_host_is_refused(self) -> None:
        with pytest.raises(AppError) as exc_info:
            assert_same_origin_redirect(
                "https://geo.api.vlaanderen.be/dhmv/wcs",
                "https://cdn.example.net/payload.tif",
            )

        assert exc_info.value.code == "OUTBOUND_REDIRECT_NOT_ALLOWED"
        assert "cdn.example.net" in str(exc_info.value.details)

    def test_a_downgrade_to_plain_http_is_refused(self) -> None:
        with pytest.raises(AppError) as exc_info:
            assert_same_origin_redirect(
                "https://geo.api.vlaanderen.be/wcs",
                "http://geo.api.vlaanderen.be/wcs",
            )

        assert exc_info.value.code == "OUTBOUND_REDIRECT_NOT_ALLOWED"

    def test_a_redirect_to_the_loopback_is_refused_even_on_the_same_scheme(self) -> None:
        with pytest.raises(AppError):
            assert_same_origin_redirect("https://geo.api.vlaanderen.be/wcs", "https://127.0.0.1/wcs")

    def test_an_upgrade_to_https_stays_allowed(self) -> None:
        assert_same_origin_redirect("http://geo.example.be/wcs", "https://geo.example.be/wcs")


def test_the_guard_opener_refuses_a_cross_host_redirect() -> None:
    """The opener is what the acquisition services actually call."""

    from app.services.outbound_request_guard import guarded_opener

    opener = guarded_opener("https://geo.api.vlaanderen.be/wcs")

    class _Redirecting:
        def __init__(self, location: str) -> None:
            self.url = location

        def __enter__(self):
            return self

        def __exit__(self, *_args):
            return False

    with pytest.raises(AppError) as exc_info:
        with opener(
            type("Req", (), {"full_url": "https://geo.api.vlaanderen.be/wcs"})(),
            timeout=1,
            _transport=lambda *_a, **_k: _Redirecting("https://evil.example.net/x"),
        ):
            pass

    assert exc_info.value.code == "OUTBOUND_REDIRECT_NOT_ALLOWED"


class TestTheGuardIsWiredIntoAcquisition:
    """Behavioural, not a grep: each service is called on its real fetch path.

    Every existing acquisition test injects an ``opener``, which bypasses the
    guard by design — that is how those tests stub the network. These call the
    production default instead.
    """

    def _settings(self):
        from app.core.config import Settings

        return Settings(_env_file=None)

    def test_dhmv_refuses_a_loopback_endpoint(self) -> None:
        from app.services.dhmv_acquisition_service import DhmvAcquisitionService

        with pytest.raises(AppError) as exc_info:
            DhmvAcquisitionService._fetch("http://127.0.0.1:9/wcs", self._settings())

        assert exc_info.value.code == "OUTBOUND_URL_NOT_ALLOWED"

    def test_flood_hazard_refuses_a_link_local_endpoint(self) -> None:
        from app.services.flood_hazard_acquisition_service import FloodHazardAcquisitionService

        with pytest.raises(AppError) as exc_info:
            FloodHazardAcquisitionService._fetch("http://169.254.169.254/latest/", self._settings())

        assert exc_info.value.code == "OUTBOUND_URL_NOT_ALLOWED"

    def test_thematic_raster_refuses_a_private_endpoint(self) -> None:
        from app.services.thematic_raster_acquisition_service import ThematicRasterAcquisitionService

        with pytest.raises(AppError) as exc_info:
            ThematicRasterAcquisitionService._fetch("http://10.0.0.5/product.tif", self._settings())

        assert exc_info.value.code == "OUTBOUND_URL_NOT_ALLOWED"

    def test_orthophoto_refuses_a_private_endpoint(self) -> None:
        from app.services.orthophoto_acquisition_service import OrthophotoAcquisitionService

        with pytest.raises(AppError) as exc_info:
            OrthophotoAcquisitionService._fetch("http://192.168.10.150/wms", self._settings())

        assert exc_info.value.code == "OUTBOUND_URL_NOT_ALLOWED"


class TestOneRedirectPolicy:
    """Two acquisition services rejected every redirect through their own
    opener while eight allowed a same-origin one through this guard. Two
    policies with no stated reason, and only one of them checked where the
    response actually came from."""

    def test_the_strict_policy_refuses_any_redirect(self) -> None:
        from app.services.outbound_request_guard import guarded_opener

        opener = guarded_opener("https://geo.api.vlaanderen.be/GRB/wfs", allow_redirect=False)

        class _Redirected:
            url = "https://geo.api.vlaanderen.be/GRB/wfs/v2"

            def __enter__(self):
                return self

            def __exit__(self, *_args):
                return False

        with pytest.raises(AppError) as exc_info:
            with opener(object(), timeout=1, _transport=lambda *_a, **_k: _Redirected()):
                pass

        assert exc_info.value.code == "OUTBOUND_REDIRECT_NOT_ALLOWED"

    def test_the_strict_policy_still_allows_the_response_it_asked_for(self) -> None:
        from app.services.outbound_request_guard import guarded_opener

        url = "https://geo.api.vlaanderen.be/GRB/wfs"
        opener = guarded_opener(url, allow_redirect=False)

        class _Direct:
            def __init__(self) -> None:
                self.url = url

            def __enter__(self):
                return self

            def __exit__(self, *_args):
                return False

        with opener(object(), timeout=1, _transport=lambda *_a, **_k: _Direct()) as response:
            assert response.url == url

    def test_both_policies_refuse_a_private_destination(self) -> None:
        from app.services.outbound_request_guard import guarded_opener

        for allow_redirect in (True, False):
            with pytest.raises(AppError) as exc_info:
                guarded_opener("http://10.0.0.5/wfs", allow_redirect=allow_redirect)
            assert exc_info.value.code == "OUTBOUND_URL_NOT_ALLOWED"

    def test_the_strict_services_use_the_shared_guard(self) -> None:
        """Behavioural: their own fetch paths refuse a private endpoint, which
        the hand-rolled opener never checked."""

        from app.core.config import Settings
        from app.services.grb_acquisition_service import GrbAcquisitionService
        from app.services.official_vector_acquisition_service import OfficialVectorAcquisitionService

        settings = Settings(_env_file=None)
        for service in (GrbAcquisitionService, OfficialVectorAcquisitionService):
            with pytest.raises(AppError) as exc_info:
                service._read_page("http://127.0.0.1:9/wfs", settings, None)
            assert exc_info.value.code == "OUTBOUND_URL_NOT_ALLOWED"


def test_a_refused_redirect_is_never_requested() -> None:
    """Rejecting after the fact still sends the request.

    Checking ``response.url`` means urllib has already followed the chain: the
    connection to the redirect target was opened and the response read. For a
    destination like a metadata endpoint that is the whole attack. The strict
    policy must refuse to follow, not refuse afterwards.
    """

    from app.services.outbound_request_guard import no_redirect_opener

    opener = no_redirect_opener()
    handlers = [type(handler).__name__ for handler in opener.handlers]

    assert "_RejectRedirects" in handlers


def test_the_rejecting_handler_returns_no_new_request() -> None:
    from app.services.outbound_request_guard import _RejectRedirects

    handler = _RejectRedirects()

    assert handler.redirect_request(None, None, 302, "Found", {}, "http://169.254.169.254/") is None