turnstone/tests/test_oauth_ssrf.py

"""Direct tests for the shared SSRF helpers in :mod:`turnstone.core.oauth_ssrf`.

The OIDC test suite already exercises these via the OIDC adapter
(``OIDCError`` re-raises). This file pins the canonical
:class:`OAuthSSRFError` exception so callers that don't go through OIDC
(notably ``mcp_oauth``) can rely on a stable contract.
"""

from __future__ import annotations

import urllib.parse
from unittest.mock import patch

import pytest

from turnstone.core.oauth_ssrf import (
    OAuthSSRFError,
    OAuthSSRFPrivateAddressError,
    effective_port,
    is_localhost,
    validate_discovered_endpoint,
    validate_url_no_ssrf,
)


class TestIsLocalhost:
    def test_loopback_names(self) -> None:
        assert is_localhost("localhost")
        assert is_localhost("127.0.0.1")
        assert is_localhost("::1")
        assert is_localhost("foo.localhost")

    def test_non_loopback(self) -> None:
        assert not is_localhost("example.com")
        assert not is_localhost("internal.corp")


class TestEffectivePort:
    def test_explicit_port(self) -> None:
        p = urllib.parse.urlparse("https://idp.example.com:9443/foo")
        assert effective_port(p) == 9443

    def test_default_https(self) -> None:
        p = urllib.parse.urlparse("https://idp.example.com/foo")
        assert effective_port(p) == 443

    def test_default_http(self) -> None:
        p = urllib.parse.urlparse("http://idp.example.com/foo")
        assert effective_port(p) == 80

    def test_unknown_scheme(self) -> None:
        p = urllib.parse.urlparse("ftp://idp.example.com/foo")
        assert effective_port(p) is None


class TestValidateUrlNoSSRF:
    _PUBLIC_ADDR = [(2, 1, 6, "", ("93.184.216.34", 0))]
    _PRIVATE_ADDR = [(2, 1, 6, "", ("10.0.0.1", 0))]
    _LOOPBACK_ADDR = [(2, 1, 6, "", ("127.0.0.1", 0))]

    def test_valid_https(self) -> None:
        with patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR):
            parsed = validate_url_no_ssrf("https://idp.example.com/foo", allow_http=False)
        assert parsed.scheme == "https"
        assert parsed.hostname == "idp.example.com"

    def test_rejects_http_when_not_allowed(self) -> None:
        with pytest.raises(OAuthSSRFError, match="must use HTTPS"):
            validate_url_no_ssrf("http://idp.example.com", allow_http=False)

    def test_allows_http_localhost_with_flag(self) -> None:
        with patch("socket.getaddrinfo", return_value=self._LOOPBACK_ADDR):
            validate_url_no_ssrf("http://localhost:8080", allow_http=True)

    def test_rejects_http_non_localhost_even_with_flag(self) -> None:
        with pytest.raises(OAuthSSRFError, match="must use HTTPS"):
            validate_url_no_ssrf("http://idp.example.com", allow_http=True)

    def test_rejects_userinfo(self) -> None:
        with pytest.raises(OAuthSSRFError, match="embedded credentials"):
            validate_url_no_ssrf("https://user:pass@idp.example.com", allow_http=False)

    def test_rejects_private_address(self) -> None:
        with (
            patch("socket.getaddrinfo", return_value=self._PRIVATE_ADDR),
            pytest.raises(OAuthSSRFError, match="non-public address"),
        ):
            validate_url_no_ssrf("https://corp.example.com", allow_http=False)

    def test_private_address_raises_distinct_subclass(self) -> None:
        # Callers with an operator opt-in (OIDC) catch the subclass to
        # append the remediation hint; plain OAuthSSRFError catches still work.
        with (
            patch("socket.getaddrinfo", return_value=self._PRIVATE_ADDR),
            pytest.raises(OAuthSSRFPrivateAddressError),
        ):
            validate_url_no_ssrf("https://corp.example.com", allow_http=False)

    def test_allow_private_accepts_rfc1918(self) -> None:
        with patch("socket.getaddrinfo", return_value=self._PRIVATE_ADDR):
            parsed = validate_url_no_ssrf(
                "https://auth.corp.example.com", allow_http=False, allow_private=True
            )
        assert parsed.hostname == "auth.corp.example.com"

    def test_allow_private_accepts_cgnat(self) -> None:
        # 100.64/10 (RFC 6598, shared address space) — e.g. an overlay-VPN IdP.
        with patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("100.64.0.7", 0))]):
            validate_url_no_ssrf("https://idp.tail.example", allow_http=False, allow_private=True)

    def test_allow_private_accepts_loopback_hostname(self) -> None:
        # A non-localhost hostname resolving to loopback (IdP behind a
        # local reverse proxy) is operator-trusted under the opt-in.
        with patch("socket.getaddrinfo", return_value=self._LOOPBACK_ADDR):
            validate_url_no_ssrf("https://auth.internal", allow_http=False, allow_private=True)

    def test_allow_private_still_rejects_link_local(self) -> None:
        # Cloud metadata services live on link-local; no legitimate IdP does.
        with (
            patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("169.254.169.254", 0))]),
            pytest.raises(OAuthSSRFError, match="refused even with private"),
        ):
            validate_url_no_ssrf("https://md.example.com", allow_http=False, allow_private=True)

    def test_allow_private_still_rejects_unspecified(self) -> None:
        # The message names the class so 0.0.0.0/:: rejections are unambiguous.
        with (
            patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("0.0.0.0", 0))]),
            pytest.raises(OAuthSSRFError, match="unspecified"),
        ):
            validate_url_no_ssrf("https://zero.example.com", allow_http=False, allow_private=True)

    def test_allow_private_does_not_relax_https(self) -> None:
        with pytest.raises(OAuthSSRFError, match="must use HTTPS"):
            validate_url_no_ssrf(
                "http://auth.corp.example.com", allow_http=False, allow_private=True
            )

    def test_rejects_unresolvable(self) -> None:
        import socket

        with (
            patch("socket.getaddrinfo", side_effect=socket.gaierror("fail")),
            pytest.raises(OAuthSSRFError, match="cannot be resolved"),
        ):
            validate_url_no_ssrf("https://no.such.host.invalid", allow_http=False)


class TestValidateDiscoveredEndpoint:
    _PUBLIC_ADDR = [(2, 1, 6, "", ("93.184.216.34", 0))]

    def test_same_origin_passes(self) -> None:
        issuer = urllib.parse.urlparse("https://idp.example.com")
        with patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR):
            validate_discovered_endpoint(
                "https://idp.example.com/token",
                issuer,
                allow_http=False,
                trusted_endpoint_hosts=frozenset(),
            )

    def test_third_party_host_rejected(self) -> None:
        issuer = urllib.parse.urlparse("https://idp.example.com")
        with (
            patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR),
            pytest.raises(OAuthSSRFError, match="not trusted"),
        ):
            validate_discovered_endpoint(
                "https://attacker.example.com/token",
                issuer,
                allow_http=False,
                trusted_endpoint_hosts=frozenset(),
            )

    def test_allow_private_passes_through(self) -> None:
        # Same-origin endpoint on a private-resolving issuer host is accepted
        # when the operator opted in.
        issuer = urllib.parse.urlparse("https://auth.corp.example.com")
        with patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("10.0.0.5", 0))]):
            validate_discovered_endpoint(
                "https://auth.corp.example.com/token",
                issuer,
                allow_http=False,
                trusted_endpoint_hosts=frozenset(),
                allow_private=True,
            )

    def test_trusted_endpoint_host_passes(self) -> None:
        issuer = urllib.parse.urlparse("https://idp.example.com")
        with patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR):
            validate_discovered_endpoint(
                "https://shard.example.com/token",
                issuer,
                allow_http=False,
                trusted_endpoint_hosts=frozenset({"shard.example.com"}),
            )

    def test_known_google_alias_passes(self) -> None:
        """The hard-coded Google alias map covers oauth2.googleapis.com."""
        issuer = urllib.parse.urlparse("https://accounts.google.com")
        with patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR):
            validate_discovered_endpoint(
                "https://oauth2.googleapis.com/token",
                issuer,
                allow_http=False,
                trusted_endpoint_hosts=frozenset(),
            )

    def test_scheme_mismatch_rejected(self) -> None:
        # When the issuer is http://localhost (allow_http=True), an
        # https:// endpoint must still be rejected as a scheme mismatch.
        issuer = urllib.parse.urlparse("http://localhost:8080")
        with (
            patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("127.0.0.1", 0))]),
            pytest.raises(OAuthSSRFError, match="scheme"),
        ):
            validate_discovered_endpoint(
                "https://localhost:8080/token",
                issuer,
                allow_http=True,
                trusted_endpoint_hosts=frozenset(),
            )

    def test_port_mismatch_rejected(self) -> None:
        issuer = urllib.parse.urlparse("https://idp.example.com")
        with (
            patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR),
            pytest.raises(OAuthSSRFError, match="port"),
        ):
            validate_discovered_endpoint(
                "https://idp.example.com:9443/token",
                issuer,
                allow_http=False,
                trusted_endpoint_hosts=frozenset(),
            )