Petru Paler
f2145e4827
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>
389 lines
12 KiB
Python
389 lines
12 KiB
Python
# ABOUTME: Tests for request ID generation and logging middleware.
|
|
# ABOUTME: Covers ULID generation, scope storage, and NDJSON log format.
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
from unittest.mock import MagicMock
|
|
|
|
|
|
def make_test_settings(
|
|
csrf_secret: str = "test-secret",
|
|
trusted_proxy_ips: str = "",
|
|
auth_header_name: str = "X-Oidc-Username",
|
|
):
|
|
"""Create Settings for testing by setting env vars temporarily."""
|
|
from animaltrack.config import Settings
|
|
|
|
old_env = os.environ.copy()
|
|
try:
|
|
os.environ["CSRF_SECRET"] = csrf_secret
|
|
os.environ["TRUSTED_PROXY_IPS"] = trusted_proxy_ips
|
|
os.environ["AUTH_HEADER_NAME"] = auth_header_name
|
|
return Settings()
|
|
finally:
|
|
os.environ.clear()
|
|
os.environ.update(old_env)
|
|
|
|
|
|
class TestRequestIdBefore:
|
|
"""Tests for request ID generation middleware."""
|
|
|
|
def test_generates_request_id(self):
|
|
"""Generates a request_id in the scope."""
|
|
from animaltrack.web.middleware import request_id_before
|
|
|
|
req = MagicMock()
|
|
req.scope = {}
|
|
|
|
request_id_before(req)
|
|
|
|
assert "request_id" in req.scope
|
|
assert req.scope["request_id"] is not None
|
|
|
|
def test_request_id_is_ulid(self):
|
|
"""Request ID is a valid 26-char ULID."""
|
|
from animaltrack.web.middleware import request_id_before
|
|
|
|
req = MagicMock()
|
|
req.scope = {}
|
|
|
|
request_id_before(req)
|
|
|
|
request_id = req.scope["request_id"]
|
|
assert len(request_id) == 26
|
|
# ULIDs are base32, should only contain valid chars
|
|
valid_chars = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
|
|
assert all(c in valid_chars for c in request_id.upper())
|
|
|
|
def test_generates_unique_ids(self):
|
|
"""Each request gets a unique request_id."""
|
|
from animaltrack.web.middleware import request_id_before
|
|
|
|
ids = set()
|
|
for _ in range(100):
|
|
req = MagicMock()
|
|
req.scope = {}
|
|
request_id_before(req)
|
|
ids.add(req.scope["request_id"])
|
|
|
|
assert len(ids) == 100 # All unique
|
|
|
|
def test_sets_start_time(self):
|
|
"""Sets request_start_time in scope for duration calculation."""
|
|
from animaltrack.web.middleware import request_id_before
|
|
|
|
req = MagicMock()
|
|
req.scope = {}
|
|
|
|
before = time.time()
|
|
request_id_before(req)
|
|
after = time.time()
|
|
|
|
assert "request_start_time" in req.scope
|
|
assert before <= req.scope["request_start_time"] <= after
|
|
|
|
|
|
class TestLoggingAfter:
|
|
"""Tests for request logging middleware."""
|
|
|
|
def test_logs_in_ndjson_format(self, capsys):
|
|
"""Log line is valid JSON."""
|
|
from animaltrack.web.middleware import logging_after
|
|
|
|
req = MagicMock()
|
|
req.scope = {"request_id": "TEST123", "request_start_time": time.time()}
|
|
req.url = MagicMock(path="/test")
|
|
req.method = "GET"
|
|
req.headers = {}
|
|
req.client = MagicMock(host="127.0.0.1")
|
|
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
|
|
settings = make_test_settings()
|
|
|
|
logging_after(req, resp, settings)
|
|
|
|
captured = capsys.readouterr()
|
|
log_line = captured.out.strip()
|
|
|
|
# Should be valid JSON
|
|
parsed = json.loads(log_line)
|
|
assert isinstance(parsed, dict)
|
|
|
|
def test_log_includes_required_fields(self, capsys):
|
|
"""Log includes: ts, level, route, actor, ip, method, status, duration_ms."""
|
|
from animaltrack.models.reference import User, UserRole
|
|
from animaltrack.web.middleware import logging_after
|
|
|
|
user = User(
|
|
username="testuser",
|
|
role=UserRole.RECORDER,
|
|
active=True,
|
|
created_at_utc=1000000,
|
|
updated_at_utc=1000000,
|
|
)
|
|
|
|
req = MagicMock()
|
|
req.scope = {
|
|
"request_id": "TEST123",
|
|
"request_start_time": time.time() - 0.1, # 100ms ago
|
|
"auth": user,
|
|
}
|
|
req.url = MagicMock(path="/test/route")
|
|
req.method = "POST"
|
|
req.headers = {}
|
|
req.client = MagicMock(host="127.0.0.1")
|
|
|
|
resp = MagicMock()
|
|
resp.status_code = 201
|
|
|
|
settings = make_test_settings()
|
|
|
|
logging_after(req, resp, settings)
|
|
|
|
captured = capsys.readouterr()
|
|
parsed = json.loads(captured.out.strip())
|
|
|
|
assert "ts" in parsed
|
|
assert parsed["level"] == "info"
|
|
assert parsed["route"] == "/test/route"
|
|
assert parsed["actor"] == "testuser"
|
|
assert parsed["method"] == "POST"
|
|
assert parsed["status"] == 201
|
|
assert "duration_ms" in parsed
|
|
assert parsed["duration_ms"] >= 0
|
|
|
|
def test_log_includes_request_id(self, capsys):
|
|
"""Log includes request_id field."""
|
|
from animaltrack.web.middleware import logging_after
|
|
|
|
req = MagicMock()
|
|
req.scope = {"request_id": "01ARZ3NDEKTSV4RRFFQ69G5FAV", "request_start_time": time.time()}
|
|
req.url = MagicMock(path="/test")
|
|
req.method = "GET"
|
|
req.headers = {}
|
|
req.client = MagicMock(host="127.0.0.1")
|
|
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
|
|
settings = make_test_settings()
|
|
|
|
logging_after(req, resp, settings)
|
|
|
|
captured = capsys.readouterr()
|
|
parsed = json.loads(captured.out.strip())
|
|
|
|
assert parsed["request_id"] == "01ARZ3NDEKTSV4RRFFQ69G5FAV"
|
|
|
|
def test_log_includes_event_id_when_provided(self, capsys):
|
|
"""Log includes event_id field when provided."""
|
|
from animaltrack.web.middleware import logging_after
|
|
|
|
req = MagicMock()
|
|
req.scope = {"request_id": "TEST123", "request_start_time": time.time()}
|
|
req.url = MagicMock(path="/actions/product-collected")
|
|
req.method = "POST"
|
|
req.headers = {}
|
|
req.client = MagicMock(host="127.0.0.1")
|
|
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
|
|
settings = make_test_settings()
|
|
|
|
logging_after(req, resp, settings, event_id="01ARZ3NDEKTSV4RRFFQ69G5EVT")
|
|
|
|
captured = capsys.readouterr()
|
|
parsed = json.loads(captured.out.strip())
|
|
|
|
assert parsed["event_id"] == "01ARZ3NDEKTSV4RRFFQ69G5EVT"
|
|
|
|
def test_duration_ms_accurate(self, capsys):
|
|
"""duration_ms reflects actual request time."""
|
|
from animaltrack.web.middleware import logging_after
|
|
|
|
req = MagicMock()
|
|
# Started 150ms ago
|
|
req.scope = {"request_id": "TEST123", "request_start_time": time.time() - 0.150}
|
|
req.url = MagicMock(path="/test")
|
|
req.method = "GET"
|
|
req.headers = {}
|
|
req.client = MagicMock(host="127.0.0.1")
|
|
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
|
|
settings = make_test_settings()
|
|
|
|
logging_after(req, resp, settings)
|
|
|
|
captured = capsys.readouterr()
|
|
parsed = json.loads(captured.out.strip())
|
|
|
|
# Should be approximately 150ms (with some tolerance)
|
|
assert 100 <= parsed["duration_ms"] <= 250
|
|
|
|
def test_logs_x_forwarded_for_ip(self, capsys):
|
|
"""Log ip field uses X-Forwarded-For when present."""
|
|
from animaltrack.web.middleware import logging_after
|
|
|
|
req = MagicMock()
|
|
req.scope = {"request_id": "TEST123", "request_start_time": time.time()}
|
|
req.url = MagicMock(path="/test")
|
|
req.method = "GET"
|
|
req.headers = {"x-forwarded-for": "203.0.113.50, 10.0.0.1"}
|
|
req.client = MagicMock(host="10.0.0.1") # Proxy IP
|
|
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
|
|
settings = make_test_settings()
|
|
|
|
logging_after(req, resp, settings)
|
|
|
|
captured = capsys.readouterr()
|
|
parsed = json.loads(captured.out.strip())
|
|
|
|
# Should use the client IP from X-Forwarded-For, not the proxy IP
|
|
assert parsed["ip"] == "203.0.113.50"
|
|
|
|
def test_actor_is_none_when_unauthenticated(self, capsys):
|
|
"""Log actor is None when no user in scope."""
|
|
from animaltrack.web.middleware import logging_after
|
|
|
|
req = MagicMock()
|
|
req.scope = {"request_id": "TEST123", "request_start_time": time.time()}
|
|
req.url = MagicMock(path="/healthz")
|
|
req.method = "GET"
|
|
req.headers = {}
|
|
req.client = MagicMock(host="127.0.0.1")
|
|
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
|
|
settings = make_test_settings()
|
|
|
|
logging_after(req, resp, settings)
|
|
|
|
captured = capsys.readouterr()
|
|
parsed = json.loads(captured.out.strip())
|
|
|
|
assert parsed["actor"] is None
|
|
|
|
def test_timestamp_is_milliseconds(self, capsys):
|
|
"""Log ts field is milliseconds since epoch."""
|
|
from animaltrack.web.middleware import logging_after
|
|
|
|
req = MagicMock()
|
|
req.scope = {"request_id": "TEST123", "request_start_time": time.time()}
|
|
req.url = MagicMock(path="/test")
|
|
req.method = "GET"
|
|
req.headers = {}
|
|
req.client = MagicMock(host="127.0.0.1")
|
|
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
|
|
settings = make_test_settings()
|
|
|
|
before_ms = int(time.time() * 1000)
|
|
logging_after(req, resp, settings)
|
|
after_ms = int(time.time() * 1000)
|
|
|
|
captured = capsys.readouterr()
|
|
parsed = json.loads(captured.out.strip())
|
|
|
|
# ts should be between before and after (in milliseconds)
|
|
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
|