m4
c46ae17084
EndpointPolicy validates provider base URLs (section 4.3): public https endpoints with hostname and optional port pass; loopback, private, link-local, multicast, unspecified, and cloud-metadata addresses are denied unless the normalized URL exactly matches a registered development_endpoints entry (no prefix or wildcard matching). URLs with user info, fragments, or non-http(s) schemes are rejected with the new stable 422 code ENDPOINT_NOT_ALLOWED. SafeHttpTransport is the single network egress for adapters: a custom httpcore NetworkBackend resolves DNS under control on every connect (retries included), filters denied ranges, and connects directly to the selected IP, while TLS SNI/certificate checks and the HTTP Host header keep the original hostname. Redirects and env proxies are disabled; every request origin re-passes URL-layer validation before any I/O.
297 lines
10 KiB
Python
297 lines
10 KiB
Python
"""Tests for the EndpointPolicy URL allow rules (design doc section 4.3).
|
|
|
|
Public endpoints require https plus a hostname with an optional port.
|
|
Loopback, private, link-local, multicast, unspecified, and cloud-metadata
|
|
addresses are denied unless the exact normalized URL is registered as a
|
|
platform ``development_endpoints`` entry. Matching is an exact full-string
|
|
match after normalization — no prefix or wildcard matching.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from EvoScientist.model_registry.endpoint_policy import EndpointPolicy
|
|
from EvoScientist.model_registry.errors import ENDPOINT_NOT_ALLOWED, ModelRegistryError
|
|
from EvoScientist.model_registry.schemas import (
|
|
DevelopmentEndpoint,
|
|
EndpointPolicyPublic,
|
|
)
|
|
|
|
|
|
def _ollama_policy() -> EndpointPolicy:
|
|
return EndpointPolicy(
|
|
development_endpoints=[
|
|
DevelopmentEndpoint(
|
|
id="ollama", url="http://localhost:11434", label="Local Ollama"
|
|
)
|
|
]
|
|
)
|
|
|
|
|
|
class TestPublicEndpoints:
|
|
@pytest.mark.parametrize(
|
|
("url", "normalized"),
|
|
[
|
|
("https://api.example.com", "https://api.example.com"),
|
|
("HTTPS://API.Example.COM/", "https://api.example.com"),
|
|
("https://api.example.com:443", "https://api.example.com"),
|
|
("https://api.example.com:8443/v1/", "https://api.example.com:8443/v1"),
|
|
("https://8.8.8.8", "https://8.8.8.8"),
|
|
],
|
|
)
|
|
def test_public_https_passes(self, url, normalized):
|
|
policy = EndpointPolicy()
|
|
assert policy.validate_base_url(url) == normalized
|
|
|
|
def test_http_public_rejected(self):
|
|
policy = EndpointPolicy()
|
|
with pytest.raises(ModelRegistryError) as excinfo:
|
|
policy.validate_base_url("http://api.example.com")
|
|
assert excinfo.value.code == ENDPOINT_NOT_ALLOWED
|
|
assert excinfo.value.http_status == 422
|
|
|
|
@pytest.mark.parametrize(
|
|
"url",
|
|
[
|
|
# loopback
|
|
"https://127.0.0.1",
|
|
"https://127.0.0.1:8443",
|
|
"http://127.0.0.1:8080",
|
|
"https://[::1]",
|
|
# private
|
|
"https://10.0.0.5",
|
|
"https://172.16.0.1",
|
|
"https://192.168.1.1",
|
|
"https://[fd00::1]",
|
|
# link-local
|
|
"https://169.254.1.1",
|
|
"https://[fe80::1]",
|
|
# multicast
|
|
"https://224.0.0.1",
|
|
# unspecified
|
|
"https://0.0.0.0",
|
|
"https://[::]",
|
|
# cloud metadata
|
|
"https://169.254.169.254",
|
|
"http://169.254.169.254",
|
|
],
|
|
)
|
|
def test_denied_ip_literals_rejected(self, url):
|
|
policy = EndpointPolicy()
|
|
with pytest.raises(ModelRegistryError) as excinfo:
|
|
policy.validate_base_url(url)
|
|
assert excinfo.value.code == ENDPOINT_NOT_ALLOWED
|
|
|
|
@pytest.mark.parametrize(
|
|
"url",
|
|
[
|
|
"https://user:pass@api.example.com",
|
|
"https://user@api.example.com",
|
|
"http://user@localhost:11434",
|
|
],
|
|
)
|
|
def test_userinfo_rejected(self, url):
|
|
policy = _ollama_policy()
|
|
with pytest.raises(ModelRegistryError) as excinfo:
|
|
policy.validate_base_url(url)
|
|
assert excinfo.value.code == ENDPOINT_NOT_ALLOWED
|
|
|
|
def test_fragment_rejected(self):
|
|
policy = EndpointPolicy()
|
|
with pytest.raises(ModelRegistryError) as excinfo:
|
|
policy.validate_base_url("https://api.example.com/#section")
|
|
assert excinfo.value.code == ENDPOINT_NOT_ALLOWED
|
|
|
|
@pytest.mark.parametrize(
|
|
"url",
|
|
[
|
|
"ftp://example.com",
|
|
"file:///etc/passwd",
|
|
"wss://example.com",
|
|
"example.com",
|
|
],
|
|
)
|
|
def test_disallowed_scheme_rejected(self, url):
|
|
policy = EndpointPolicy()
|
|
with pytest.raises(ModelRegistryError) as excinfo:
|
|
policy.validate_base_url(url)
|
|
assert excinfo.value.code == ENDPOINT_NOT_ALLOWED
|
|
|
|
@pytest.mark.parametrize("url", ["https://", "https://:8443", ""])
|
|
def test_missing_host_rejected(self, url):
|
|
policy = EndpointPolicy()
|
|
with pytest.raises(ModelRegistryError) as excinfo:
|
|
policy.validate_base_url(url)
|
|
assert excinfo.value.code == ENDPOINT_NOT_ALLOWED
|
|
|
|
def test_error_message_is_safe(self):
|
|
policy = EndpointPolicy()
|
|
with pytest.raises(ModelRegistryError) as excinfo:
|
|
policy.validate_base_url("https://user:secret@127.0.0.1/x")
|
|
assert "secret" not in excinfo.value.message
|
|
|
|
|
|
class TestDevelopmentEndpoints:
|
|
def test_registered_local_endpoint_passes(self):
|
|
policy = _ollama_policy()
|
|
assert (
|
|
policy.validate_base_url("http://localhost:11434")
|
|
== "http://localhost:11434"
|
|
)
|
|
|
|
def test_unregistered_local_endpoint_rejected(self):
|
|
policy = EndpointPolicy()
|
|
with pytest.raises(ModelRegistryError) as excinfo:
|
|
policy.validate_base_url("http://localhost:11434")
|
|
assert excinfo.value.code == ENDPOINT_NOT_ALLOWED
|
|
|
|
@pytest.mark.parametrize(
|
|
"url",
|
|
[
|
|
# different port — exact match, not prefix or wildcard
|
|
"http://localhost:11435",
|
|
# extra path segment — no prefix matching
|
|
"http://localhost:11434/api",
|
|
# a different host that merely shares the prefix
|
|
"http://localhost:11434.evil.com",
|
|
],
|
|
)
|
|
def test_exact_match_only(self, url):
|
|
policy = _ollama_policy()
|
|
with pytest.raises(ModelRegistryError) as excinfo:
|
|
policy.validate_base_url(url)
|
|
assert excinfo.value.code == ENDPOINT_NOT_ALLOWED
|
|
|
|
@pytest.mark.parametrize(
|
|
"url",
|
|
[
|
|
"HTTP://LOCALHOST:11434/",
|
|
"http://Localhost:11434",
|
|
],
|
|
)
|
|
def test_normalization_equivalence_hits_same_entry(self, url):
|
|
policy = _ollama_policy()
|
|
assert policy.validate_base_url(url) == "http://localhost:11434"
|
|
|
|
def test_default_port_normalization_hits_same_entry(self):
|
|
policy = EndpointPolicy(
|
|
development_endpoints=[
|
|
DevelopmentEndpoint(
|
|
id="local-http", url="http://127.0.0.1", label="Local HTTP"
|
|
)
|
|
]
|
|
)
|
|
assert policy.validate_base_url("http://127.0.0.1:80/") == "http://127.0.0.1"
|
|
|
|
def test_registered_private_ip_passes(self):
|
|
policy = EndpointPolicy(
|
|
development_endpoints=[
|
|
DevelopmentEndpoint(
|
|
id="lan", url="http://192.168.1.10:8080", label="LAN box"
|
|
)
|
|
]
|
|
)
|
|
assert (
|
|
policy.validate_base_url("http://192.168.1.10:8080/")
|
|
== "http://192.168.1.10:8080"
|
|
)
|
|
|
|
def test_registration_normalization_applies_to_entries(self):
|
|
policy = EndpointPolicy(
|
|
development_endpoints=[
|
|
DevelopmentEndpoint(
|
|
id="ollama", url="HTTP://LOCALHOST:11434/", label="Local Ollama"
|
|
)
|
|
]
|
|
)
|
|
assert (
|
|
policy.validate_base_url("http://localhost:11434")
|
|
== "http://localhost:11434"
|
|
)
|
|
|
|
def test_entry_with_path_matches_exactly(self):
|
|
policy = EndpointPolicy(
|
|
development_endpoints=[
|
|
DevelopmentEndpoint(
|
|
id="lan", url="http://localhost:11434/api", label="LAN API"
|
|
)
|
|
]
|
|
)
|
|
assert (
|
|
policy.validate_base_url("http://localhost:11434/api/")
|
|
== "http://localhost:11434/api"
|
|
)
|
|
with pytest.raises(ModelRegistryError):
|
|
policy.validate_base_url("http://localhost:11434/other")
|
|
|
|
def test_entry_with_path_allows_whole_origin_for_requests(self):
|
|
policy = EndpointPolicy(
|
|
development_endpoints=[
|
|
DevelopmentEndpoint(
|
|
id="lan", url="http://localhost:11434/api", label="LAN API"
|
|
)
|
|
]
|
|
)
|
|
assert (
|
|
policy.validate_request_origin("http://localhost:11434")
|
|
== "http://localhost:11434"
|
|
)
|
|
|
|
def test_request_origin_uses_public_rules_without_registration(self):
|
|
policy = _ollama_policy()
|
|
assert (
|
|
policy.validate_request_origin("https://api.example.com")
|
|
== "https://api.example.com"
|
|
)
|
|
with pytest.raises(ModelRegistryError) as excinfo:
|
|
policy.validate_request_origin("http://localhost:11435")
|
|
assert excinfo.value.code == ENDPOINT_NOT_ALLOWED
|
|
|
|
def test_duplicate_registration_rejected(self):
|
|
with pytest.raises(ValueError, match=r"[Dd]uplicate"):
|
|
EndpointPolicy(
|
|
development_endpoints=[
|
|
DevelopmentEndpoint(
|
|
id="a", url="http://localhost:11434", label="A"
|
|
),
|
|
DevelopmentEndpoint(
|
|
id="b", url="HTTP://LOCALHOST:11434/", label="B"
|
|
),
|
|
]
|
|
)
|
|
|
|
@pytest.mark.parametrize(
|
|
"url",
|
|
["ftp://localhost:11434", "https://user@localhost:11434", "not-a-url"],
|
|
)
|
|
def test_invalid_registration_rejected(self, url):
|
|
with pytest.raises(ValueError, match="development endpoint"):
|
|
EndpointPolicy(
|
|
development_endpoints=[
|
|
DevelopmentEndpoint(id="bad", url=url, label="Bad")
|
|
]
|
|
)
|
|
|
|
|
|
class TestPublicView:
|
|
def test_public_view_shape(self):
|
|
policy = _ollama_policy()
|
|
view = policy.public_view()
|
|
assert isinstance(view, EndpointPolicyPublic)
|
|
assert view.model_dump() == {
|
|
"public_https_allowed": True,
|
|
"development_endpoints": [
|
|
{
|
|
"id": "ollama",
|
|
"url": "http://localhost:11434",
|
|
"label": "Local Ollama",
|
|
}
|
|
],
|
|
}
|
|
|
|
def test_public_view_empty(self):
|
|
view = EndpointPolicy().public_view()
|
|
assert view.public_https_allowed is True
|
|
assert view.development_endpoints == []
|