fix(security): deny generated-image access when ownership verification fails (#6268)

* 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>

app.py

@@ -516,28 +516,69 @@ app.mount("/static", _RevalidatingStatic(directory=STATIC_DIR), name="static")
 @app.get("/api/generated-image/{filename}")
 async def serve_generated_image(filename: str, request: Request):
     """Serve generated images from the data directory."""
-    img_path = resolve_generated_image_path(filename)
-    # SECURITY: filename is the only key, so anyone who knows / guesses a
-    # 12-hex content hash could pull another user's image bytes. Require
-    # auth and verify ownership via the gallery row (when one exists).
-    try:
-        from src.auth_helpers import get_current_user
-        from core.database import SessionLocal as _SL, GalleryImage as _GI
-        _user = get_current_user(request)
-        if _user:
-            _db = _SL()
+    # 1. Validate filename syntax upfront, rejecting malformed requests (400)
+    # without probing the filesystem or database.
+    from src.generated_images import GENERATED_IMAGE_RE
+    if not isinstance(filename, str) or not GENERATED_IMAGE_RE.fullmatch(filename):
+        raise HTTPException(status_code=400, detail="Invalid filename")
+
+    # 2. Ownership verification:
+    # If auth is enabled, verify caller identity and ownership BEFORE probing disk,
+    # preventing existence oracles during outages or unauthorized requests.
+    from src.auth_helpers import auth_disabled, effective_user
+    if not auth_disabled():
+        _user = effective_user(request)
+        if not _user:
+            raise HTTPException(status_code=401, detail="Not authenticated")
+
+        from src.generated_images import get_registered_image_owner
+        _reg_owner = get_registered_image_owner(filename)
+        if _reg_owner is not None:
+            if _reg_owner != _user:
+                raise HTTPException(status_code=404, detail="Image not found")
+        else:
             try:
-                _row = _db.query(_GI).filter(_GI.filename == filename).first()
-                # Generated-but-not-yet-imported images have no row → allow.
-                # Row exists with a different owner → 404 (don't confirm existence).
-                if _row is not None and _row.owner and _row.owner != _user:
-                    raise HTTPException(status_code=404, detail="Image not found")
-            finally:
-                _db.close()
-    except HTTPException:
-        raise
-    except Exception as _e:
-        logger.warning("Image ownership verification failed for %r", filename, exc_info=_e)
+                from core.database import (
+                    SessionLocal as _SL,
+                    GalleryImage as _GI,
+                    Session as _DbSession,
+                    ChatMessage as _CM,
+                )
+                _db = _SL()
+                try:
+                    _row = _db.query(_GI).filter(_GI.filename == filename).first()
+                    if _row is not None:
+                        if _row.owner:
+                            if _row.owner != _user:
+                                raise HTTPException(status_code=404, detail="Image not found")
+                        elif _row.session_id:
+                            _sess = _db.query(_DbSession).filter(_DbSession.id == _row.session_id).first()
+                            if not _sess or not _sess.owner or _sess.owner != _user:
+                                raise HTTPException(status_code=404, detail="Image not found")
+                        else:
+                            # Gallery record has no owner or session link -> fail closed
+                            raise HTTPException(status_code=404, detail="Image not found")
+                    else:
+                        # Rowless image: check if referenced in a chat session owned by this user
+                        _chat_ref = (
+                            _db.query(_CM.id)
+                            .join(_DbSession, _CM.session_id == _DbSession.id)
+                            .filter(_DbSession.owner == _user, _CM.meta_data.like(f"%{filename}%"))
+                            .first()
+                        )
+                        if not _chat_ref:
+                            raise HTTPException(status_code=404, detail="Image not found")
+                finally:
+                    _db.close()
+            except HTTPException:
+                raise
+            except Exception as _e:
+                logger.warning("Image ownership verification failed for %r", filename, exc_info=_e)
+                raise HTTPException(status_code=503, detail="Ownership verification unavailable")
+
+    # 3. Resolve confined path and verify existence on disk
+    img_path = resolve_generated_image_path(filename)
+
     ext = filename.rsplit('.', 1)[-1].lower()
     mime = {
         "png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg",
@@ -545,10 +586,6 @@ async def serve_generated_image(filename: str, request: Request):
         "mp4": "video/mp4", "mov": "video/quicktime", "webm": "video/webm",
         "mkv": "video/x-matroska", "m4v": "video/mp4",
     }.get(ext, "application/octet-stream")
-    # Generated-image filenames are content hashes → the bytes for a given
-    # filename never change. Cache them hard so the gallery doesn't
-    # re-download every full-size image each time it's opened. `immutable`
-    # tells the browser it never needs to revalidate within the max-age.
     return FileResponse(
         str(img_path),
         media_type=mime,