mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-10-30 09:42:21 +01:00
* fix(security): deny generated-image access when ownership verification fails The except Exception block in serve_generated_image catches all exceptions during ownership verification and falls through to serve the file anyway. If the database is down or corrupted, the ownership check is bypassed entirely — any authenticated user can download any generated image by guessing the content-hash filename. Raise HTTP 503 (Service Unavailable) instead of falling through, so the endpoint fails closed when verification is impossible. This matches the defense-in-depth principle: deny access on verification errors rather than granting it. * fix(security): enforce fail-closed ownership and secure caching for generated images - Fail closed on missing gallery records, null/empty owners, and DB errors - Support both cookie session and bearer API token owners via effective_user - Add in-memory registry for freshly generated images to preserve access - Eliminate file existence oracle by verifying ownership before disk probing - Update Cache-Control to private, no-cache, must-revalidate - Pass owner attribution through tool execution and image MCP server - Add comprehensive hermetic regression tests for ownership gates and caching * fix(security): sanitize tool owner arguments and drop unneeded MCP env fallback - Remove redundant and unsafe ODYSSEUS_MCP_IMAGE_OWNER fallback in image_gen_server.py to fix 4 tests in test_env_reference.py. - Sanitize caller-supplied tool arguments across all MCP and email tool dispatch paths in src/tool_execution.py: strip untrusted _odysseus_* keys from arguments. - Only inject _odysseus_owner when verified trusted server context is present, preventing untrusted/anonymous users from spoofing ownership claims. - Sanitize empty/whitespace owner strings in image_gen_server.py to None. - Add comprehensive regression tests in test_generated_image_ownership.py for argument sanitization, owner injection, fallback propagation, and server sanitization. --------- Co-authored-by: isharak7m <192635824+isharak7m@users.noreply.github.com> Co-authored-by: PR #6166 Reviewer <reviewer@odysseus-review.local>
71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
import re
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from src.constants import GENERATED_IMAGES_DIR
|
|
from src.path_confinement import confine
|
|
|
|
|
|
GENERATED_IMAGE_DIR = Path(GENERATED_IMAGES_DIR)
|
|
GENERATED_IMAGE_RE = re.compile(
|
|
r"^[a-f0-9]{8,64}\.(png|jpg|jpeg|webp|gif|mp4|mov|webm|mkv|m4v)$"
|
|
)
|
|
GENERATED_IMAGE_HEADERS = {
|
|
"Cache-Control": "private, no-cache, must-revalidate",
|
|
"X-Content-Type-Options": "nosniff",
|
|
}
|
|
|
|
_RECENT_GENERATED_IMAGES: dict[str, tuple[str, float]] = {}
|
|
_RECENT_IMAGES_LOCK = threading.Lock()
|
|
_RECENT_IMAGES_TTL = 3600 # 1 hour TTL
|
|
|
|
|
|
def register_generated_image(filename: str, owner: Optional[str]) -> None:
|
|
"""Record ownership of a freshly generated image.
|
|
|
|
Ensures legitimate access to freshly generated images even before
|
|
or in case Gallery database records are created/committed.
|
|
"""
|
|
if not filename or not owner:
|
|
return
|
|
now = time.time()
|
|
with _RECENT_IMAGES_LOCK:
|
|
if len(_RECENT_GENERATED_IMAGES) > 200:
|
|
expired = [k for k, (_, exp) in _RECENT_GENERATED_IMAGES.items() if exp < now]
|
|
for k in expired:
|
|
_RECENT_GENERATED_IMAGES.pop(k, None)
|
|
_RECENT_GENERATED_IMAGES[filename] = (owner, now + _RECENT_IMAGES_TTL)
|
|
|
|
|
|
def get_registered_image_owner(filename: str) -> Optional[str]:
|
|
"""Return the owner of a freshly generated image if still valid in the registry."""
|
|
if not filename:
|
|
return None
|
|
now = time.time()
|
|
with _RECENT_IMAGES_LOCK:
|
|
entry = _RECENT_GENERATED_IMAGES.get(filename)
|
|
if not entry:
|
|
return None
|
|
owner, expiry = entry
|
|
if now > expiry:
|
|
_RECENT_GENERATED_IMAGES.pop(filename, None)
|
|
return None
|
|
return owner
|
|
|
|
|
|
def resolve_generated_image_path(filename: str) -> Path:
|
|
if not isinstance(filename, str) or not GENERATED_IMAGE_RE.fullmatch(filename):
|
|
raise HTTPException(status_code=400, detail="Invalid filename")
|
|
try:
|
|
path = Path(confine(GENERATED_IMAGE_DIR, filename, allow_root=False))
|
|
except (ValueError, OSError):
|
|
raise HTTPException(status_code=400, detail="Invalid filename")
|
|
if not path.exists():
|
|
raise HTTPException(status_code=404, detail="Image not found")
|
|
return path
|
|
|