mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
Patrick Buckley
4350248d8f
The discovered-endpoint wrapper converted every OAuthSSRFError to a bare OIDCError, so a private-resolving endpoint or trusted host got the non-public message without the allow_private_network remediation even though the same knob fixes it. Hoist the hint into a module constant and append it in both wrappers. Also name "unspecified" in the refused-even-with-opt-in message so 0.0.0.0/:: rejections read unambiguously.
236 lines
9.7 KiB
Python
236 lines
9.7 KiB
Python
"""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. a tailnet-hosted 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(),
|
|
)
|