Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 15 additions & 4 deletions obs/api/v1/history.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,20 @@ def _parse_ts(s: str | None, default: datetime) -> datetime:
)


async def _resolve_page_access(db: Database, node_id: str) -> str:
"""Traversiert die parent_id-Kette und gibt das effektive Access-Level zurück."""
current_id: str | None = node_id
while current_id:
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Guard against cycles in access-chain traversal

Track visited node IDs (or enforce a max depth) in _resolve_page_access, because the current while current_id loop can become non-terminating when visu_nodes.parent_id contains a cycle. Cycles are possible from existing write paths (for example, node moves do not enforce acyclic parent relationships), and with such data a history request using that X-Page-Id will repeatedly query forever, tying up a worker and creating a denial-of-service condition for affected endpoints.

Useful? React with 👍 / 👎.

async with db.conn.execute("SELECT access, parent_id FROM visu_nodes WHERE id = ?", (current_id,)) as cur:
row = await cur.fetchone()
if not row:
return "private" # Unbekannter Knoten → sicher ablehnen
if row["access"] is not None:
return row["access"]
current_id = row["parent_id"]
return "public"


async def _check_history_access(
request: Request,
user: str | None,
Expand All @@ -70,10 +84,7 @@ async def _check_history_access(
if not page_id:
raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Authentication required")

async with db.conn.execute("SELECT access, parent_id FROM visu_nodes WHERE id = ?", (page_id,)) as cur:
row = await cur.fetchone()

access = row["access"] if row and row["access"] else "public"
access = await _resolve_page_access(db, page_id)

if access in ("public", "readonly"):
return # Öffentliche Seite → History-Lesen erlaubt
Expand Down