mirror of
https://github.com/lfnovo/open-notebook.git
synced 2026-08-14 11:33:51 +00:00
7cac3da240
The DNS-pinning guard added in #1063 is the highest-risk change in the v1.14.0 release for self-hosters: it sits on the outbound path of every credential save, connection test and model discovery, so a regression that over-blocks silently costs users their provider. TestPinnedHttpTarget already covers what the guard must reject. This adds the inverse assertions - the deployment shapes that must keep working - which prepare_pinned_http_target had no coverage for at all (the existing legitimate-use tests only exercised validate_url, which is not what gates outbound requests): - Ollama on localhost, including IPv4 preference when both families resolve - host.docker.internal (containerized app -> host service) - private LAN by IP literal (LM Studio) and by hostname - IPv6 loopback literal - Tailscale CGNAT space (100.64.0.0/10) - shared, not link-local - AAAA-only endpoints producing a bracketed, parseable URL - query strings surviving the rewrite (PPQ's ?type=all discovery URL) All pass against the current implementation - this pins the behavior rather than fixing a defect.
373 lines
16 KiB
Python
373 lines
16 KiB
Python
"""
|
|
Test URL validation for SSRF protection in API key configuration.
|
|
|
|
Note: The validation is intentionally permissive for self-hosted scenarios.
|
|
It only blocks:
|
|
- Invalid schemes (must be http or https)
|
|
- Malformed URLs
|
|
- Link-local addresses (169.254.x.x) - used for cloud metadata endpoints
|
|
|
|
Localhost and private IPs are ALLOWED because this is a self-hosted application
|
|
where users commonly run local services (Ollama, LM Studio, etc.).
|
|
|
|
validate_url() is async (the hostname-resolution branch runs
|
|
socket.getaddrinfo() via asyncio.to_thread so it doesn't block the event
|
|
loop - see open_notebook/utils/url_validation.py), so every test here is
|
|
async too.
|
|
"""
|
|
|
|
import socket
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from api.credentials_service import validate_url
|
|
from open_notebook.utils.url_validation import prepare_pinned_http_target
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
class TestUrlValidation:
|
|
"""Test suite for URL validation to prevent SSRF attacks."""
|
|
|
|
async def test_valid_https_url(self):
|
|
"""Valid HTTPS URLs should pass."""
|
|
await validate_url("https://api.openai.com", "openai")
|
|
await validate_url("https://example.com/api", "anthropic")
|
|
# Should not raise
|
|
|
|
async def test_valid_http_url(self):
|
|
"""Valid HTTP URLs should pass."""
|
|
await validate_url("http://example.com", "openai")
|
|
# Should not raise
|
|
|
|
async def test_invalid_scheme(self):
|
|
"""URLs with invalid schemes should be rejected."""
|
|
with pytest.raises(ValueError, match="Invalid URL scheme"):
|
|
await validate_url("ftp://example.com", "openai")
|
|
|
|
with pytest.raises(ValueError, match="Invalid URL scheme"):
|
|
await validate_url("file:///etc/passwd", "openai")
|
|
|
|
async def test_localhost_allowed_for_self_hosted(self):
|
|
"""Localhost should be allowed for self-hosted services."""
|
|
# This is a self-hosted app, localhost is valid for local services
|
|
await validate_url("http://localhost:8000", "openai")
|
|
await validate_url("http://127.0.0.1:8000", "azure")
|
|
# Should not raise
|
|
|
|
async def test_localhost_allowed_for_ollama(self):
|
|
"""Localhost should be allowed for Ollama provider."""
|
|
await validate_url("http://localhost:11434", "ollama")
|
|
await validate_url("http://127.0.0.1:11434", "ollama")
|
|
# Should not raise
|
|
|
|
async def test_private_ip_allowed_for_self_hosted(self):
|
|
"""Private IP addresses should be allowed for self-hosted scenarios."""
|
|
# This is a self-hosted app, private IPs are valid for internal services
|
|
await validate_url("http://10.0.0.1", "openai")
|
|
await validate_url("http://172.16.0.1:8080", "anthropic")
|
|
await validate_url("http://192.168.1.1", "azure")
|
|
# Should not raise
|
|
|
|
async def test_private_ip_allowed_for_ollama(self):
|
|
"""Private IP addresses should be allowed for Ollama provider."""
|
|
await validate_url("http://192.168.1.100:11434", "ollama")
|
|
await validate_url("http://10.0.0.50:11434", "ollama")
|
|
# Should not raise
|
|
|
|
async def test_loopback_allowed_for_self_hosted(self):
|
|
"""Loopback addresses should be allowed for self-hosted scenarios."""
|
|
await validate_url("http://127.0.0.2", "openai")
|
|
# Should not raise
|
|
|
|
async def test_link_local_rejection(self):
|
|
"""Link-local addresses should be rejected (cloud metadata protection)."""
|
|
with pytest.raises(ValueError, match="Link-local addresses"):
|
|
await validate_url("http://169.254.169.254", "openai")
|
|
|
|
# Also reject for ollama - link-local is never valid
|
|
with pytest.raises(ValueError, match="Link-local addresses"):
|
|
await validate_url("http://169.254.169.254", "ollama")
|
|
|
|
async def test_ipv6_localhost_allowed(self):
|
|
"""IPv6 localhost should be allowed for self-hosted scenarios."""
|
|
await validate_url("http://[::1]:8000", "openai")
|
|
# Should not raise
|
|
|
|
async def test_empty_url(self):
|
|
"""Empty URLs should not raise (handled elsewhere)."""
|
|
await validate_url("", "openai")
|
|
# None is handled by the function's early return check
|
|
# Should not raise
|
|
|
|
async def test_invalid_url_format(self):
|
|
"""Malformed URLs should be rejected."""
|
|
with pytest.raises(ValueError):
|
|
await validate_url("not-a-url", "openai")
|
|
|
|
async def test_public_hostnames_allowed(self):
|
|
"""Public hostnames should be allowed."""
|
|
await validate_url("https://api.openai.com/v1", "openai")
|
|
await validate_url("https://api.anthropic.com", "anthropic")
|
|
await validate_url("https://generativelanguage.googleapis.com", "google")
|
|
await validate_url("https://api.groq.com", "groq")
|
|
# Should not raise
|
|
|
|
async def test_azure_specific_urls(self):
|
|
"""Azure OpenAI endpoints should be validated."""
|
|
await validate_url(
|
|
"https://my-resource.openai.azure.com", "azure"
|
|
)
|
|
# Localhost is allowed for self-hosted
|
|
await validate_url("http://localhost:8000", "azure")
|
|
# Should not raise
|
|
|
|
async def test_openai_compatible_urls(self):
|
|
"""OpenAI-compatible provider URLs should be validated."""
|
|
await validate_url("https://api.together.xyz", "openai_compatible")
|
|
# Private IPs are allowed for self-hosted
|
|
await validate_url("http://192.168.1.1:8080", "openai_compatible")
|
|
# Should not raise
|
|
|
|
async def test_ipv4_mapped_ipv6_link_local_rejected(self):
|
|
"""IPv4-mapped IPv6 addresses pointing to link-local should be rejected."""
|
|
with pytest.raises(ValueError, match="Link-local addresses"):
|
|
await validate_url("http://[::ffff:169.254.169.254]", "openai")
|
|
|
|
async def test_ipv4_mapped_ipv6_private_allowed(self):
|
|
"""IPv4-mapped IPv6 addresses pointing to private IPs should be allowed."""
|
|
await validate_url("http://[::ffff:192.168.1.1]", "openai")
|
|
# Should not raise - private IPs allowed for self-hosted
|
|
|
|
async def test_aws_imds_v6_rejected(self):
|
|
"""AWS IMDSv6 metadata address must be rejected."""
|
|
with pytest.raises(ValueError, match="IMDSv6|metadata"):
|
|
await validate_url("http://[fd00:ec2::254]/", "openai")
|
|
|
|
async def test_scoped_aws_imds_v6_rejected(self):
|
|
"""Scoped IMDSv6 (fd00:ec2::254%eth0) must not bypass the sentinel check."""
|
|
with pytest.raises(ValueError, match="IMDSv6|metadata"):
|
|
await validate_url("http://[fd00:ec2::254%eth0]/", "openai")
|
|
|
|
with pytest.raises(ValueError, match="IMDSv6|metadata"):
|
|
await validate_url("http://[fd00:ec2::254%25eth0]/", "openai")
|
|
|
|
|
|
class TestPinnedHttpTarget:
|
|
"""DNS pinning closes the validate-then-httpx rebinding window."""
|
|
|
|
async def test_ip_literal_unchanged(self):
|
|
target = await prepare_pinned_http_target(
|
|
"http://127.0.0.1:11434/api/tags", "ollama"
|
|
)
|
|
assert target.url == "http://127.0.0.1:11434/api/tags"
|
|
assert target.headers == {}
|
|
assert target.extensions == {}
|
|
|
|
async def test_link_local_ip_rejected(self):
|
|
with pytest.raises(ValueError, match="Link-local"):
|
|
await prepare_pinned_http_target(
|
|
"http://169.254.169.254/latest/meta-data", "openai_compatible"
|
|
)
|
|
|
|
async def test_scoped_aws_imds_v6_rejected(self):
|
|
"""Scoped IMDSv6 literals must be rejected before pinning returns a target."""
|
|
with pytest.raises(ValueError, match="IMDSv6|metadata"):
|
|
await prepare_pinned_http_target(
|
|
"http://[fd00:ec2::254%eth0]/", "openai_compatible"
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="IMDSv6|metadata"):
|
|
await prepare_pinned_http_target(
|
|
"http://[fd00:ec2::254%25eth0]/", "openai_compatible"
|
|
)
|
|
|
|
async def test_hostname_pinned_to_resolved_ip(self):
|
|
fake_addrs = [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.168.1.50", 0)),
|
|
]
|
|
with patch(
|
|
"open_notebook.utils.url_validation.socket.getaddrinfo",
|
|
return_value=fake_addrs,
|
|
):
|
|
target = await prepare_pinned_http_target(
|
|
"http://ollama.local:11434/api/tags", "ollama"
|
|
)
|
|
|
|
assert target.url == "http://192.168.1.50:11434/api/tags"
|
|
assert target.headers == {"Host": "ollama.local:11434"}
|
|
assert target.extensions == {}
|
|
|
|
async def test_https_sets_sni_hostname(self):
|
|
fake_addrs = [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0)),
|
|
]
|
|
with patch(
|
|
"open_notebook.utils.url_validation.socket.getaddrinfo",
|
|
return_value=fake_addrs,
|
|
):
|
|
target = await prepare_pinned_http_target(
|
|
"https://api.example.com/v1/models", "openai_compatible"
|
|
)
|
|
|
|
assert target.url == "https://93.184.216.34/v1/models"
|
|
assert target.headers == {"Host": "api.example.com"}
|
|
assert target.extensions == {"sni_hostname": "api.example.com"}
|
|
|
|
async def test_unicode_hostname_idna_encoded_for_host_and_sni(self):
|
|
"""Internationalized hostnames must use ASCII IDNA for Host and SNI."""
|
|
fake_addrs = [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0)),
|
|
]
|
|
with patch(
|
|
"open_notebook.utils.url_validation.socket.getaddrinfo",
|
|
return_value=fake_addrs,
|
|
):
|
|
target = await prepare_pinned_http_target(
|
|
"https://bücher.example/v1/models", "openai_compatible"
|
|
)
|
|
|
|
assert target.url == "https://93.184.216.34/v1/models"
|
|
assert target.headers == {"Host": "xn--bcher-kva.example"}
|
|
assert target.extensions == {"sni_hostname": "xn--bcher-kva.example"}
|
|
|
|
async def test_hostname_resolving_to_link_local_rejected(self):
|
|
fake_addrs = [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("169.254.169.254", 0)),
|
|
]
|
|
with patch(
|
|
"open_notebook.utils.url_validation.socket.getaddrinfo",
|
|
return_value=fake_addrs,
|
|
):
|
|
with pytest.raises(ValueError, match="link-local"):
|
|
await prepare_pinned_http_target(
|
|
"http://evil.example/v1/models", "openai_compatible"
|
|
)
|
|
|
|
|
|
class TestPinnedHttpTargetSelfHostedUse:
|
|
"""The pinning guard must not break the self-hosted setups it protects.
|
|
|
|
TestPinnedHttpTarget above proves the guard blocks what it should. These
|
|
are the inverse assertions: every deployment shape a self-hoster actually
|
|
runs must survive pinning and reach its endpoint. A regression here means
|
|
users with a local Ollama, an LM Studio box on the LAN, a containerized
|
|
app talking to the host, or a Tailscale-hosted endpoint lose their
|
|
provider with no obvious cause.
|
|
"""
|
|
|
|
async def test_localhost_ollama_survives_pinning(self):
|
|
"""The single most common self-hosted setup: Ollama on localhost."""
|
|
fake_addrs = [
|
|
(socket.AF_INET6, socket.SOCK_STREAM, 6, "", ("::1", 0, 0, 0)),
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 0)),
|
|
]
|
|
with patch(
|
|
"open_notebook.utils.url_validation.socket.getaddrinfo",
|
|
return_value=fake_addrs,
|
|
):
|
|
target = await prepare_pinned_http_target(
|
|
"http://localhost:11434/api/tags", "ollama"
|
|
)
|
|
|
|
# IPv4 is preferred when both families resolve.
|
|
assert target.url == "http://127.0.0.1:11434/api/tags"
|
|
assert target.headers == {"Host": "localhost:11434"}
|
|
assert target.extensions == {}
|
|
|
|
async def test_host_docker_internal_survives_pinning(self):
|
|
"""Containerized app reaching a service on the host."""
|
|
fake_addrs = [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.168.65.2", 0)),
|
|
]
|
|
with patch(
|
|
"open_notebook.utils.url_validation.socket.getaddrinfo",
|
|
return_value=fake_addrs,
|
|
):
|
|
target = await prepare_pinned_http_target(
|
|
"http://host.docker.internal:11434/v1/models", "ollama"
|
|
)
|
|
|
|
assert target.url == "http://192.168.65.2:11434/v1/models"
|
|
assert target.headers == {"Host": "host.docker.internal:11434"}
|
|
|
|
async def test_private_lan_ip_literal_survives_pinning(self):
|
|
"""LM Studio (or any box) addressed by private IP literal on the LAN."""
|
|
target = await prepare_pinned_http_target(
|
|
"http://192.168.1.50:1234/v1/models", "openai_compatible"
|
|
)
|
|
assert target.url == "http://192.168.1.50:1234/v1/models"
|
|
assert target.headers == {}
|
|
|
|
async def test_private_lan_hostname_survives_pinning(self):
|
|
"""A LAN hostname resolving into RFC1918 space stays reachable."""
|
|
fake_addrs = [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.1.20", 0)),
|
|
]
|
|
with patch(
|
|
"open_notebook.utils.url_validation.socket.getaddrinfo",
|
|
return_value=fake_addrs,
|
|
):
|
|
target = await prepare_pinned_http_target(
|
|
"http://llm.lan/v1/models", "openai_compatible"
|
|
)
|
|
|
|
assert target.url == "http://10.0.1.20/v1/models"
|
|
# No port in the source URL means no port in the Host header.
|
|
assert target.headers == {"Host": "llm.lan"}
|
|
|
|
async def test_ipv6_loopback_literal_survives_pinning(self):
|
|
target = await prepare_pinned_http_target(
|
|
"http://[::1]:11434/api/tags", "ollama"
|
|
)
|
|
assert target.url == "http://[::1]:11434/api/tags"
|
|
assert target.headers == {}
|
|
|
|
async def test_tailscale_cgnat_address_survives_pinning(self):
|
|
"""Tailscale hands out 100.64.0.0/10 — shared space, not link-local."""
|
|
fake_addrs = [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("100.101.102.103", 0)),
|
|
]
|
|
with patch(
|
|
"open_notebook.utils.url_validation.socket.getaddrinfo",
|
|
return_value=fake_addrs,
|
|
):
|
|
target = await prepare_pinned_http_target(
|
|
"http://ollama.tail1234.ts.net:11434/api/tags", "ollama"
|
|
)
|
|
|
|
assert target.url == "http://100.101.102.103:11434/api/tags"
|
|
assert target.headers == {"Host": "ollama.tail1234.ts.net:11434"}
|
|
|
|
async def test_ipv6_only_host_pinned_with_brackets(self):
|
|
"""An AAAA-only endpoint must produce a bracketed, parseable URL."""
|
|
fake_addrs = [
|
|
(socket.AF_INET6, socket.SOCK_STREAM, 6, "", ("2606:4700::1111", 0, 0, 0)),
|
|
]
|
|
with patch(
|
|
"open_notebook.utils.url_validation.socket.getaddrinfo",
|
|
return_value=fake_addrs,
|
|
):
|
|
target = await prepare_pinned_http_target(
|
|
"https://v6.example.com/v1/models", "openai_compatible"
|
|
)
|
|
|
|
assert target.url == "https://[2606:4700::1111]/v1/models"
|
|
assert target.headers == {"Host": "v6.example.com"}
|
|
assert target.extensions == {"sni_hostname": "v6.example.com"}
|
|
|
|
async def test_query_string_and_path_preserved(self):
|
|
"""PPQ-style discovery URLs carry a query string that must survive."""
|
|
fake_addrs = [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0)),
|
|
]
|
|
with patch(
|
|
"open_notebook.utils.url_validation.socket.getaddrinfo",
|
|
return_value=fake_addrs,
|
|
):
|
|
target = await prepare_pinned_http_target(
|
|
"https://api.ppq.ai/v1/models?type=all", "openai_compatible"
|
|
)
|
|
|
|
assert target.url == "https://93.184.216.34/v1/models?type=all"
|