Latte
cf3d2e17d3
The audit logger is opened at startup (exit 2 on failure) and obtained before every call, which is refused with audit_unavailable otherwise. Audit records only take validated path and sudo values and use the request id as correlation id. The rate limiter evicts idle keys, caps tracked keys, audits the first excess per key and window and uses the settings passed to the app. FORWARDED_ALLOW_IPS is now a setting, and startup warns about a missing host allow-list and unavailable OAuth.
121 lines
4.1 KiB
Python
121 lines
4.1 KiB
Python
"""Tests for the request rate limiter."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from aegis_gitea_mcp.audit import get_audit_logger, validate_audit_log_integrity
|
|
from aegis_gitea_mcp.config import get_settings
|
|
from aegis_gitea_mcp.rate_limit import RequestRateLimiter, SlidingWindowLimiter
|
|
from tests.test_server_helpers import server_settings
|
|
|
|
|
|
@pytest.fixture
|
|
def clock(monkeypatch: pytest.MonkeyPatch) -> list[float]:
|
|
now = [1000.0]
|
|
monkeypatch.setattr("aegis_gitea_mcp.rate_limit.time.time", lambda: now[0])
|
|
return now
|
|
|
|
|
|
def security_events() -> list[dict[str, object]]:
|
|
path = get_audit_logger().log_path
|
|
entries = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
|
|
return [e["payload"] for e in entries if e["event_type"] == "security_event"]
|
|
|
|
|
|
def test_sliding_window_blocks_after_limit(clock: list[float]) -> None:
|
|
limiter = SlidingWindowLimiter(max_requests=2, window_seconds=60)
|
|
|
|
assert limiter.allow("k")
|
|
assert limiter.allow("k")
|
|
assert not limiter.allow("k")
|
|
assert limiter.allow("other")
|
|
|
|
clock[0] += 61
|
|
assert limiter.allow("k")
|
|
|
|
|
|
def test_idle_keys_are_evicted(clock: list[float]) -> None:
|
|
limiter = SlidingWindowLimiter(max_requests=5, window_seconds=60)
|
|
for index in range(100):
|
|
limiter.allow(f"client-{index}")
|
|
assert len(limiter) == 100
|
|
|
|
clock[0] += 61
|
|
limiter.allow("fresh")
|
|
|
|
assert len(limiter) == 1
|
|
|
|
|
|
def test_key_count_is_capped_and_least_recently_used_keys_go_first(clock: list[float]) -> None:
|
|
limiter = SlidingWindowLimiter(max_requests=1, window_seconds=60, max_keys=3)
|
|
for key in ("a", "b", "c"):
|
|
limiter.allow(key)
|
|
limiter.allow("a") # touch: "b" is now the oldest
|
|
limiter.allow("d")
|
|
|
|
assert len(limiter) == 3
|
|
assert limiter.allow("b") # evicted earlier, so it starts with a clean window
|
|
assert not limiter.allow("d")
|
|
|
|
|
|
def test_first_excess_reports_once_per_window(clock: list[float]) -> None:
|
|
limiter = SlidingWindowLimiter(max_requests=1, window_seconds=60)
|
|
limiter.allow("k")
|
|
|
|
assert not limiter.allow("k") and limiter.first_excess("k")
|
|
assert not limiter.allow("k") and not limiter.first_excess("k")
|
|
clock[0] += 30
|
|
assert not limiter.allow("k") and not limiter.first_excess("k")
|
|
clock[0] += 31
|
|
assert limiter.allow("k") # the old event left the window
|
|
assert not limiter.allow("k") and limiter.first_excess("k")
|
|
|
|
|
|
def test_request_limiter_per_ip_and_token(mock_env: None) -> None:
|
|
limiter = RequestRateLimiter(server_settings(rate_limit_per_minute=2))
|
|
|
|
assert limiter.check("1.1.1.1", "tok").allowed
|
|
assert limiter.check("1.1.1.1", None).allowed
|
|
assert not limiter.check("1.1.1.1", "tok").allowed
|
|
|
|
# The same token from other IPs trips the per-token limiter.
|
|
assert limiter.check("2.2.2.2", "tok").allowed
|
|
third = limiter.check("3.3.3.3", "tok")
|
|
assert not third.allowed
|
|
assert third.reason == "Per-token rate limit exceeded"
|
|
|
|
valid, errors = validate_audit_log_integrity(get_settings().resolved_audit_log_path)
|
|
assert valid, errors
|
|
|
|
|
|
def test_only_the_first_excess_per_key_is_audited(mock_env: None, clock: list[float]) -> None:
|
|
limiter = RequestRateLimiter(server_settings(rate_limit_per_minute=1))
|
|
limiter.check("1.1.1.1", None)
|
|
for _ in range(20):
|
|
assert not limiter.check("1.1.1.1", None).allowed
|
|
|
|
events = security_events()
|
|
assert [e["event_type"] for e in events] == ["rate_limit_ip_exceeded"]
|
|
|
|
clock[0] += 61
|
|
limiter.check("1.1.1.1", None)
|
|
assert not limiter.check("1.1.1.1", None).allowed
|
|
assert len(security_events()) == 2
|
|
|
|
|
|
def test_unwritable_audit_log_does_not_break_rate_limiting(
|
|
mock_env: None, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
def broken() -> None:
|
|
raise OSError("read-only file system")
|
|
|
|
monkeypatch.setattr("aegis_gitea_mcp.rate_limit.get_audit_logger", broken)
|
|
limiter = RequestRateLimiter(server_settings(rate_limit_per_minute=1))
|
|
limiter.check("1.1.1.1", None)
|
|
|
|
assert not limiter.check("1.1.1.1", None).allowed
|
|
assert "OSError" in caplog.text
|