feat: replace inline IP resolution with centralized client_ip utility across routers and audit

Consolidate scattered `X-Real-IP` / `request.client.host` fallback logic into the single `client_ip()` helper from `utils`, reducing duplication in the rate limiter middleware, project file zip routes, tools shared module, and audit record builder. Also refactor three admin endpoints (`/ai-usage/data`, `/analytics`, `/users/{uid}/ai-usage`) to use the existing `require_admin()` guard instead of repeating manual auth checks, and remove now-unused `get_current_user`/`is_admin` imports from those modules.
2026-07-04 18:07:33 +00:00
parent 244a524b91
commit 3fbac87723
14 changed files with 266 additions and 36 deletions
@@ -0,0 +1,99 @@
+# retoor <retoor@molodetz.nl>
+
+import time
+import pytest
+import requests
+from tests.conftest import BASE_URL
+from devplacepy.database import get_table, refresh_snapshot
+
+JSON_analytics = {"Accept": "application/json"}
+_counter_analytics = [0]
+
+
+def _db_user(name):
+    refresh_snapshot()
+    return get_table("users").find_one(username=name)
+
+
+def _unique(prefix="an"):
+    _counter_analytics[0] += 1
+    return f"{prefix}{int(time.time() * 1000)}{_counter_analytics[0]}"
+
+
+def _member_key():
+    name = _unique("anmem")
+    requests.post(
+        f"{BASE_URL}/auth/signup",
+        data={
+            "username": name,
+            "email": f"{name}@t.dev",
+            "password": "secret123",
+            "confirm_password": "secret123",
+        },
+        allow_redirects=True,
+    )
+    return _db_user(name)["api_key"]
+
+
+def _admin_key(seeded_db):
+    return _db_user("alice_test")["api_key"]
+
+
+def test_analytics_guest_denied(seeded_db):
+    assert (
+        requests.get(
+            f"{BASE_URL}/admin/analytics", headers=JSON_analytics, allow_redirects=False
+        ).status_code
+        == 401
+    )
+    assert (
+        requests.get(f"{BASE_URL}/admin/analytics", allow_redirects=False).status_code
+        == 303
+    )
+
+
+def test_analytics_member_denied(seeded_db):
+    key = _member_key()
+    assert (
+        requests.get(
+            f"{BASE_URL}/admin/analytics",
+            headers={**JSON_analytics, "X-API-KEY": key},
+            allow_redirects=False,
+        ).status_code
+        == 403
+    )
+
+
+def test_analytics_admin_ok(seeded_db):
+    key = _admin_key(seeded_db)
+    r = requests.get(
+        f"{BASE_URL}/admin/analytics", headers={**JSON_analytics, "X-API-KEY": key}
+    )
+    assert r.status_code == 200
+    assert isinstance(r.json(), dict)
+
+
+def test_analytics_denial_is_audited(seeded_db):
+    key = _member_key()
+    requests.get(
+        f"{BASE_URL}/admin/analytics",
+        headers={**JSON_analytics, "X-API-KEY": key},
+        allow_redirects=False,
+    )
+    admin_key = _admin_key(seeded_db)
+    hit = False
+    for _ in range(20):
+        data = requests.get(
+            f"{BASE_URL}/admin/audit-log",
+            headers={**JSON_analytics, "X-API-KEY": admin_key},
+            params={"event_key": "security.authz.denied"},
+        ).json()
+        hit = any(
+            "/admin/analytics" in (entry.get("summary") or "")
+            or "/admin/analytics" in (entry.get("request_path") or "")
+            for entry in data["entries"]
+        )
+        if hit:
+            break
+        time.sleep(0.25)
+    assert hit, "expected a security.authz.denied audit row for /admin/analytics"