feat: add CIDR/netmask support for trusted proxy IPs
TRUSTED_PROXY_IPS now accepts CIDR notation (e.g., 192.168.1.0/24) in addition to exact IP addresses. Supports both IPv4 and IPv6. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
# ABOUTME: Application configuration loaded from environment variables.
|
||||
# ABOUTME: Uses Pydantic Settings for validation and type coercion.
|
||||
|
||||
import ipaddress
|
||||
import logging
|
||||
from functools import cached_property
|
||||
|
||||
@@ -44,6 +45,43 @@ class Settings(BaseSettings):
|
||||
"""Parse trusted proxy IPs from comma-separated raw string."""
|
||||
return _parse_comma_separated(self.trusted_proxy_ips_raw)
|
||||
|
||||
@cached_property
|
||||
def trusted_proxy_networks(
|
||||
self,
|
||||
) -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]:
|
||||
"""Parse trusted proxy IPs/CIDRs into network objects.
|
||||
|
||||
Plain IPs become /32 (IPv4) or /128 (IPv6) networks.
|
||||
CIDR notation is parsed directly.
|
||||
Entries that cannot be parsed as IP/CIDR are skipped (handled by
|
||||
trusted_proxy_literals for backwards compatibility).
|
||||
"""
|
||||
networks = []
|
||||
for entry in self.trusted_proxy_ips:
|
||||
try:
|
||||
# ip_network with strict=False allows "192.168.1.1/24" to work
|
||||
# (normalizes to "192.168.1.0/24")
|
||||
network = ipaddress.ip_network(entry, strict=False)
|
||||
networks.append(network)
|
||||
except ValueError:
|
||||
# Not a valid IP/network - will be handled by trusted_proxy_literals
|
||||
pass
|
||||
return networks
|
||||
|
||||
@cached_property
|
||||
def trusted_proxy_literals(self) -> frozenset[str]:
|
||||
"""Get non-IP entries for exact string matching.
|
||||
|
||||
For backwards compatibility with entries like "testclient" in tests.
|
||||
"""
|
||||
literals = set()
|
||||
for entry in self.trusted_proxy_ips:
|
||||
try:
|
||||
ipaddress.ip_network(entry, strict=False)
|
||||
except ValueError:
|
||||
literals.add(entry)
|
||||
return frozenset(literals)
|
||||
|
||||
@field_validator("log_level", mode="before")
|
||||
@classmethod
|
||||
def normalize_and_validate_log_level(cls, v: str) -> str:
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
# ABOUTME: Middleware functions for authentication, CSRF, and request logging.
|
||||
# ABOUTME: Implements Beforeware pattern for FastHTML request processing.
|
||||
|
||||
import ipaddress
|
||||
import json
|
||||
import time
|
||||
from urllib.parse import urlparse
|
||||
@@ -112,7 +113,7 @@ def get_client_ip(req: Request) -> str:
|
||||
|
||||
|
||||
def is_trusted_proxy(req: Request, settings: Settings) -> bool:
|
||||
"""Check if request comes from a trusted proxy IP.
|
||||
"""Check if request comes from a trusted proxy IP or CIDR range.
|
||||
|
||||
Args:
|
||||
req: The Starlette request object.
|
||||
@@ -121,18 +122,35 @@ def is_trusted_proxy(req: Request, settings: Settings) -> bool:
|
||||
Returns:
|
||||
True if request is from trusted proxy, False otherwise.
|
||||
"""
|
||||
trusted_ips = settings.trusted_proxy_ips
|
||||
if not trusted_ips:
|
||||
trusted_networks = settings.trusted_proxy_networks
|
||||
trusted_literals = settings.trusted_proxy_literals
|
||||
|
||||
if not trusted_networks and not trusted_literals:
|
||||
# If no trusted IPs configured, reject all (fail-secure)
|
||||
return False
|
||||
|
||||
# Get the immediate connection IP (not X-Forwarded-For)
|
||||
if req.client:
|
||||
client_ip = req.client.host
|
||||
client_ip_str = req.client.host
|
||||
else:
|
||||
return False
|
||||
|
||||
return client_ip in trusted_ips
|
||||
# Check literal matches first (for backwards compatibility with "testclient" etc)
|
||||
if client_ip_str in trusted_literals:
|
||||
return True
|
||||
|
||||
# Try to parse as IP address for network matching
|
||||
try:
|
||||
client_ip = ipaddress.ip_address(client_ip_str)
|
||||
except ValueError:
|
||||
# Not a valid IP and not in literals
|
||||
return False
|
||||
|
||||
for network in trusted_networks:
|
||||
if client_ip in network:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def get_expected_host(req: Request, settings: Settings) -> str:
|
||||
|
||||
@@ -227,6 +227,109 @@ class TestTrustedProxyIPs:
|
||||
settings = Settings()
|
||||
assert settings.trusted_proxy_ips == []
|
||||
|
||||
def test_cidr_notation_parsed(self):
|
||||
"""CIDR notation should be parsed into networks."""
|
||||
import ipaddress
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"TRUSTED_PROXY_IPS": "192.168.1.0/24", "CSRF_SECRET": "test-secret"},
|
||||
clear=True,
|
||||
):
|
||||
from animaltrack.config import Settings
|
||||
|
||||
settings = Settings()
|
||||
networks = settings.trusted_proxy_networks
|
||||
assert len(networks) == 1
|
||||
assert isinstance(networks[0], ipaddress.IPv4Network)
|
||||
assert str(networks[0]) == "192.168.1.0/24"
|
||||
|
||||
def test_plain_ip_parsed_as_single_host_network(self):
|
||||
"""Plain IP should be parsed as /32 network."""
|
||||
import ipaddress
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"TRUSTED_PROXY_IPS": "10.0.0.1", "CSRF_SECRET": "test-secret"},
|
||||
clear=True,
|
||||
):
|
||||
from animaltrack.config import Settings
|
||||
|
||||
settings = Settings()
|
||||
networks = settings.trusted_proxy_networks
|
||||
assert len(networks) == 1
|
||||
assert isinstance(networks[0], ipaddress.IPv4Network)
|
||||
# Plain IP becomes /32 network
|
||||
assert networks[0].num_addresses == 1
|
||||
|
||||
def test_mixed_ips_and_cidrs(self):
|
||||
"""Mix of plain IPs and CIDR notation should all be parsed."""
|
||||
import ipaddress
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"TRUSTED_PROXY_IPS": "10.0.0.1,192.168.0.0/16,172.16.0.0/12",
|
||||
"CSRF_SECRET": "test-secret",
|
||||
},
|
||||
clear=True,
|
||||
):
|
||||
from animaltrack.config import Settings
|
||||
|
||||
settings = Settings()
|
||||
networks = settings.trusted_proxy_networks
|
||||
assert len(networks) == 3
|
||||
# All should be network objects
|
||||
assert all(
|
||||
isinstance(n, (ipaddress.IPv4Network, ipaddress.IPv6Network)) for n in networks
|
||||
)
|
||||
|
||||
def test_ipv6_cidr_supported(self):
|
||||
"""IPv6 CIDR notation should be supported."""
|
||||
import ipaddress
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"TRUSTED_PROXY_IPS": "::1,fd00::/8", "CSRF_SECRET": "test-secret"},
|
||||
clear=True,
|
||||
):
|
||||
from animaltrack.config import Settings
|
||||
|
||||
settings = Settings()
|
||||
networks = settings.trusted_proxy_networks
|
||||
assert len(networks) == 2
|
||||
assert any(isinstance(n, ipaddress.IPv6Network) for n in networks)
|
||||
|
||||
def test_invalid_cidr_becomes_literal(self):
|
||||
"""Invalid CIDR notation should become a literal for string matching."""
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"TRUSTED_PROXY_IPS": "192.168.1.0/33", "CSRF_SECRET": "test-secret"},
|
||||
clear=True,
|
||||
):
|
||||
from animaltrack.config import Settings
|
||||
|
||||
settings = Settings()
|
||||
# Invalid CIDR should not appear in networks
|
||||
assert len(settings.trusted_proxy_networks) == 0
|
||||
# But should appear in literals
|
||||
assert "192.168.1.0/33" in settings.trusted_proxy_literals
|
||||
|
||||
def test_invalid_ip_becomes_literal(self):
|
||||
"""Invalid IP address should become a literal for string matching."""
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"TRUSTED_PROXY_IPS": "not-an-ip", "CSRF_SECRET": "test-secret"},
|
||||
clear=True,
|
||||
):
|
||||
from animaltrack.config import Settings
|
||||
|
||||
settings = Settings()
|
||||
# Invalid IP should not appear in networks
|
||||
assert len(settings.trusted_proxy_networks) == 0
|
||||
# But should appear in literals
|
||||
assert "not-an-ip" in settings.trusted_proxy_literals
|
||||
|
||||
|
||||
class TestCsrfSecretRequired:
|
||||
"""Test that CSRF_SECRET is required."""
|
||||
|
||||
@@ -300,3 +300,89 @@ class TestLoggingAfter:
|
||||
assert before_ms <= parsed["ts"] <= after_ms
|
||||
# Should be a reasonable timestamp (year 2020+)
|
||||
assert parsed["ts"] > 1577836800000 # 2020-01-01
|
||||
|
||||
|
||||
class TestIsTrustedProxyCIDR:
|
||||
"""Tests for CIDR support in is_trusted_proxy."""
|
||||
|
||||
def test_ip_within_cidr_is_trusted(self):
|
||||
"""IP within CIDR range should be trusted."""
|
||||
from animaltrack.web.middleware import is_trusted_proxy
|
||||
|
||||
req = MagicMock()
|
||||
req.client = MagicMock(host="192.168.1.50")
|
||||
|
||||
settings = make_test_settings(trusted_proxy_ips="192.168.1.0/24")
|
||||
|
||||
assert is_trusted_proxy(req, settings) is True
|
||||
|
||||
def test_ip_outside_cidr_not_trusted(self):
|
||||
"""IP outside CIDR range should not be trusted."""
|
||||
from animaltrack.web.middleware import is_trusted_proxy
|
||||
|
||||
req = MagicMock()
|
||||
req.client = MagicMock(host="192.168.2.50")
|
||||
|
||||
settings = make_test_settings(trusted_proxy_ips="192.168.1.0/24")
|
||||
|
||||
assert is_trusted_proxy(req, settings) is False
|
||||
|
||||
def test_exact_ip_still_works(self):
|
||||
"""Exact IP matching should still work."""
|
||||
from animaltrack.web.middleware import is_trusted_proxy
|
||||
|
||||
req = MagicMock()
|
||||
req.client = MagicMock(host="10.0.0.1")
|
||||
|
||||
settings = make_test_settings(trusted_proxy_ips="10.0.0.1")
|
||||
|
||||
assert is_trusted_proxy(req, settings) is True
|
||||
|
||||
def test_mixed_exact_and_cidr(self):
|
||||
"""Mix of exact IPs and CIDR should work."""
|
||||
from animaltrack.web.middleware import is_trusted_proxy
|
||||
|
||||
settings = make_test_settings(trusted_proxy_ips="10.0.0.1,192.168.0.0/16")
|
||||
|
||||
# Exact IP match
|
||||
req1 = MagicMock()
|
||||
req1.client = MagicMock(host="10.0.0.1")
|
||||
assert is_trusted_proxy(req1, settings) is True
|
||||
|
||||
# CIDR match
|
||||
req2 = MagicMock()
|
||||
req2.client = MagicMock(host="192.168.100.200")
|
||||
assert is_trusted_proxy(req2, settings) is True
|
||||
|
||||
# No match
|
||||
req3 = MagicMock()
|
||||
req3.client = MagicMock(host="172.16.0.1")
|
||||
assert is_trusted_proxy(req3, settings) is False
|
||||
|
||||
def test_ipv6_cidr_matching(self):
|
||||
"""IPv6 CIDR matching should work."""
|
||||
from animaltrack.web.middleware import is_trusted_proxy
|
||||
|
||||
settings = make_test_settings(trusted_proxy_ips="fd00::/8")
|
||||
|
||||
req = MagicMock()
|
||||
req.client = MagicMock(host="fd12:3456:789a::1")
|
||||
assert is_trusted_proxy(req, settings) is True
|
||||
|
||||
req2 = MagicMock()
|
||||
req2.client = MagicMock(host="fe80::1")
|
||||
assert is_trusted_proxy(req2, settings) is False
|
||||
|
||||
def test_localhost_cidr(self):
|
||||
"""Localhost CIDR should work."""
|
||||
from animaltrack.web.middleware import is_trusted_proxy
|
||||
|
||||
settings = make_test_settings(trusted_proxy_ips="127.0.0.0/8")
|
||||
|
||||
req = MagicMock()
|
||||
req.client = MagicMock(host="127.0.0.1")
|
||||
assert is_trusted_proxy(req, settings) is True
|
||||
|
||||
req2 = MagicMock()
|
||||
req2.client = MagicMock(host="127.255.255.255")
|
||||
assert is_trusted_proxy(req2, settings) is True
|
||||
|
||||
Reference in New Issue
Block a user