Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import ipaddress
from typing import Any
from urllib.parse import ParseResult, urlparse
from urllib.parse import ParseResult, unquote, urlparse

from pydantic import Field

Expand Down Expand Up @@ -113,12 +113,53 @@ def _effective_port(url: ParseResult) -> int | None:
return None


def _normalize_path(path: str, decode_percent_encoded: bool = False) -> str:
"""Normalize a URL path the same way httpx does before routing.

Applies RFC 3986 §5.2.4 remove_dot_segments. Empty segments are preserved.
Paths that would climb above the root are clamped at ``/``.
"""
if decode_percent_encoded:
path = unquote(path)

# remove_dot_segments (RFC 3986 §5.2.4), preserving empty segments like httpx.
segments = path.split("/")
resolved: list[str] = []
for segment in segments:
if segment == ".":
continue
if segment == "..":
if resolved and resolved[-1] != "":
resolved.pop()
# else: clamp at root (leading empty from absolute path stays)
continue
resolved.append(segment)
if segments[-1] in (".", ".."):
resolved.append("")
# Absolute paths keep a leading empty segment from the initial "/".
if not resolved:
return "/"
normalized = "/".join(resolved)
if path.startswith("/") and not normalized.startswith("/"):
normalized = "/" + normalized
return normalized or "/"


def _matches_path_prefix(url_path: str, base_path: str) -> bool:
url_path = url_path or "/"
base_path = base_path or "/"

if url_path.lower() == base_path.lower():
return True
for decode_percent_encoded in (False, True):
normalized_url_path = _normalize_path(url_path, decode_percent_encoded)
normalized_base_path = _normalize_path(base_path, decode_percent_encoded)

if normalized_url_path.lower() == normalized_base_path.lower():
continue

base_path_with_slash = (
normalized_base_path if normalized_base_path.endswith("/") else f"{normalized_base_path}/"
)
if not normalized_url_path.lower().startswith(base_path_with_slash.lower()):
return False

base_path_with_slash = base_path if base_path.endswith("/") else f"{base_path}/"
return url_path.lower().startswith(base_path_with_slash.lower())
return True
Original file line number Diff line number Diff line change
Expand Up @@ -241,3 +241,76 @@ async def test_validate_server_url_returns_no_addresses_for_allowed_base_url():
async def test_validate_server_url_returns_no_addresses_when_private_access_is_allowed():
options = ServerUrlValidationOptions(allow_private_network_access=True)
assert await validate_server_url("https://internal.example/api", options) == []


@pytest.mark.parametrize(
"url",
[
"https://api.example.com/api/../admin",
"https://api.example.com/api/%2e%2e/admin",
"https://api.example.com/api/%2E%2E/admin",
"https://api.example.com/api/foo/../../admin",
],
)
async def test_validate_server_url_rejects_path_prefix_escape(url):
"""Dot-segment escapes must not satisfy an /api allowlist prefix."""
options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api"])

with pytest.raises(FunctionExecutionException, match="allowed base URLs"):
await validate_server_url(url, options)


@pytest.mark.parametrize(
"url",
[
"https://api.example.com/api/./users",
"https://api.example.com/api/users",
"https://api.example.com/api/",
"https://api.example.com/api",
],
)
async def test_validate_server_url_allows_normalized_paths_under_prefix(url):
options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api"])

assert await validate_server_url(url, options) == []


async def test_validate_server_url_still_rejects_path_that_only_shares_prefix_chars():
options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api"])

with pytest.raises(FunctionExecutionException, match="allowed base URLs"):
await validate_server_url("https://api.example.com/apiusers", options)


async def test_validate_server_url_rejects_path_that_only_matches_after_decoding():
options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api"])

with pytest.raises(FunctionExecutionException, match="allowed base URLs"):
await validate_server_url("https://api.example.com/admin/%2e%2e/api", options)


@pytest.mark.parametrize(
"url",
[
"https://api.example.com/api/.",
"https://api.example.com/api/users/..",
],
)
async def test_validate_server_url_preserves_trailing_slash_after_normalization(url):
options = ServerUrlValidationOptions(allowed_base_urls=["https://api.example.com/api/"])

assert await validate_server_url(url, options) == []


@pytest.mark.parametrize(
"allowed_base_url",
[
"https://api.example.com/api/.",
"https://api.example.com/api/",
],
)
async def test_validate_server_url_trailing_slash_allowlist_does_not_allow_bare_path(allowed_base_url):
options = ServerUrlValidationOptions(allowed_base_urls=[allowed_base_url])

with pytest.raises(FunctionExecutionException, match="allowed base URLs"):
await validate_server_url("https://api.example.com/api", options)
Loading