|
# 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"
|