4ff113f2cf
Some checks failed
secrets-guard / encrypted (push) Has been cancelled
shell-lint / shellcheck (push) Has been cancelled
Deploy backend to LAN dev server / build (push) Successful in 1m22s
Build + push backend image (Forgejo registry) / build-push (push) Successful in 1m24s
Deploy backend to LAN dev server / deploy (push) Successful in 19s
87 lines
3.7 KiB
Python
87 lines
3.7 KiB
Python
"""The request-logging middleware (web/app.py's revalidate_static): what gets
|
|
counted, what gets written to the access log, and what gets truncated before
|
|
it's persisted. Companion to core/test_metrics.py's classify_inbound tests --
|
|
these exercise the actual ASGI request path, not just the classifier function.
|
|
"""
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from web import app as appmod
|
|
|
|
metrics = appmod.metrics # the exact module object app.py's own calls use --
|
|
audit = appmod.audit # see the note on module-reload isolation below.
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
return TestClient(appmod.app)
|
|
|
|
|
|
def _patch_log_access(monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(audit, "log_access", lambda record: calls.append(record))
|
|
return calls
|
|
|
|
|
|
def test_healthz_is_never_audited(monkeypatch, client):
|
|
"""/healthz is ~47% of prod's daily access log; it must never even reach
|
|
audit.log_access (not just be cheap once there -- see classify_inbound's
|
|
"health" category and the exclusion in revalidate_static)."""
|
|
calls = _patch_log_access(monkeypatch)
|
|
r = client.get("/healthz")
|
|
assert r.status_code == 200
|
|
assert calls == []
|
|
|
|
|
|
def test_internal_category_is_never_audited(monkeypatch, client):
|
|
"""The SSR content API's own category ("internal") is excluded from the
|
|
access log the same way -- it's a server-to-server hop (frontend_ssr's
|
|
api_client.py calling backend), not a page view."""
|
|
calls = _patch_log_access(monkeypatch)
|
|
monkeypatch.setattr(metrics, "classify_inbound", lambda path, base: "internal")
|
|
r = client.get("/thermograph/api/version")
|
|
assert r.status_code == 200
|
|
assert calls == []
|
|
|
|
|
|
def test_access_log_ip_is_truncated_not_raw(monkeypatch, client):
|
|
calls = _patch_log_access(monkeypatch)
|
|
r = client.get("/thermograph/api/version",
|
|
headers={"X-Forwarded-For": "203.0.113.77, 10.0.0.1"})
|
|
assert r.status_code == 200
|
|
assert len(calls) == 1
|
|
assert calls[0]["ip"] == "203.0.113.0" # /24, not the exact address
|
|
assert calls[0]["ip"] != "203.0.113.77"
|
|
|
|
|
|
def test_loggable_ip_truncation():
|
|
assert appmod._loggable_ip("203.0.113.77") == "203.0.113.0"
|
|
assert appmod._loggable_ip("2001:db8:1234:5678::1") == "2001:db8:1234::"
|
|
assert appmod._loggable_ip(None) is None
|
|
assert appmod._loggable_ip("") == ""
|
|
assert appmod._loggable_ip("not-an-ip") == "not-an-ip" # best-effort, never raises
|
|
|
|
|
|
def test_rate_limiter_still_sees_the_full_precision_ip(monkeypatch, client):
|
|
"""_client_ip (which feeds both the access log and the event rate limiter)
|
|
is never itself truncated -- only what audit.log_access persists is. A
|
|
/24-truncated rate-limit key would let one abuser exhaust a whole NAT'd
|
|
office's quota, which is exactly what must NOT happen here."""
|
|
seen = {}
|
|
monkeypatch.setattr(metrics, "record_event", lambda name, **kw: seen.update(kw))
|
|
r = client.post("/thermograph/api/v2/event", json={"event": "home.locate"},
|
|
headers={"X-Forwarded-For": "203.0.113.77"})
|
|
assert r.status_code == 204
|
|
assert seen["ip"] == "203.0.113.77"
|
|
|
|
|
|
def test_event_route_records_end_to_end(client):
|
|
"""Full HTTP round trip through the real ASGI route -- confirms the write
|
|
path backend/api/event -> metrics.record_event -> the counters store works
|
|
end to end (the existing suite only ever called record_event directly)."""
|
|
before = metrics.snapshot()["events_total"]
|
|
r = client.post("/thermograph/api/v2/event", json={"event": "home.locate"})
|
|
assert r.status_code == 204
|
|
after = metrics.snapshot()
|
|
assert after["events_total"] == before + 1
|
|
assert after["events"]["home.locate"]["direct"] # no Referer sent -> "direct"
|