Protect advanced request diagnostics for non-admins
Magent CI/CD / verify (push) Successful in 11m8s
Magent CI/CD / deploy-prod (push) Skipped
Magent CI/CD / deploy-beta (push) Successful in 19s

This commit is contained in:
2026-08-31 16:09:17 +12:00
parent ecf9b230c1
commit 3402e53c31
3 changed files with 129 additions and 21 deletions
+80 -1
View File
@@ -12,7 +12,7 @@ from backend.app import db
from backend.app.auth import require_admin
from backend.app.config import settings
from backend.app.network_security import request_trusts_forwarded_headers, validate_notification_target_url
from backend.app.models import NormalizedState, RequestType, Snapshot, TimelineHop
from backend.app.models import ActionOption, NormalizedState, RequestType, Snapshot, TimelineHop
from backend.app.routers import auth as auth_router
from backend.app.routers import portal as portal_router
from backend.app.routers import requests as requests_router
@@ -223,6 +223,85 @@ class RequestCacheTests(unittest.TestCase):
self.assertEqual(requests_router._cache_get(key), {"id": 123})
class RequestVisibilityTests(unittest.TestCase):
def test_non_admin_snapshot_excludes_advanced_identifying_data(self) -> None:
snapshot = Snapshot(
request_id="3925",
title="Example",
timeline=[
TimelineHop(
service="Seerr",
status="approved",
details={"requestedBy": "viewer@example.com"},
)
],
raw={
"jellyseerr": {"requestedBy": {"email": "viewer@example.com"}},
"qbittorrent": {"downloadIds": ["secret-hash"]},
},
actions=[
ActionOption(
id="search_releases",
label="Search and choose a download",
risk="safe",
)
],
presentation={"status": {"label": "Needs attention"}},
)
filtered = requests_router._filter_snapshot_for_user(
snapshot, {"username": "helper", "role": "user"}
)
self.assertEqual(filtered.timeline, [])
self.assertEqual(filtered.raw, {})
self.assertEqual([action.id for action in filtered.actions], ["search_releases"])
self.assertEqual(filtered.presentation["status"]["label"], "Needs attention")
def test_admin_snapshot_retains_advanced_diagnostics(self) -> None:
snapshot = Snapshot(
request_id="3925",
title="Example",
timeline=[TimelineHop(service="Seerr", status="approved")],
raw={"jellyseerr": {"id": 3925}},
)
filtered = requests_router._filter_snapshot_for_user(
snapshot, {"username": "admin", "role": "admin"}
)
self.assertEqual(len(filtered.timeline), 1)
self.assertEqual(filtered.raw["jellyseerr"]["id"], 3925)
def test_non_admin_cannot_request_advanced_history(self) -> None:
with self.assertRaises(HTTPException) as context:
requests_router._require_advanced_request_access(
{"username": "helper", "role": "user"}
)
self.assertEqual(context.exception.status_code, 403)
def test_my_requests_cache_only_returns_signed_in_users_rows(self) -> None:
previous = dict(requests_router._recent_cache)
requests_router._recent_cache["items"] = [
{"request_id": 100, "requested_by_id": 7, "requested_by_norm": "zak"},
{"request_id": 101, "requested_by_id": 8, "requested_by_norm": "someone-else"},
]
try:
rows = requests_router._get_recent_from_cache(
requested_by_norm="zak",
requested_by_id=7,
limit=10,
offset=0,
since_iso=None,
)
finally:
requests_router._recent_cache.clear()
requests_router._recent_cache.update(previous)
self.assertEqual([row["request_id"] for row in rows], [100])
class RequestPresentationTests(unittest.TestCase):
def test_torrent_progress_keeps_tenths_for_live_updates(self) -> None:
self.assertEqual(_torrent_progress({"progress": 0.1344}), 13.4)