Scope gallery image endpoints by owner (#3001)

Three IMAP connection leaks were recently fixed via try/finally
(#1325, #1330, #1423). This commit applies the same pattern to the
remaining callsites that still used inline logout-only cleanup.

routes/email_helpers.py:
- _fetch_sender_thread_context: conn was uninitialized when the outer
  try/except returned early on connect failure, causing the finally
  block to crash on conn.close()/conn.logout(). Merged the two
  separate try blocks into one and added conn=None guard.
- _pre_retrieve_context: ctx_conn.logout() was inside the loop body
  with no finally, so any exception in the folder/search loop leaked
  the socket. Moved cleanup into a finally block with ctx_conn=None
  guard.

mcp_servers/email_server.py:
- _list_emails: multiple inline conn.logout() calls on early-return
  paths; exception between them leaked the socket. Wrapped in
  try/finally.
- _read_email: same pattern — four separate logout() calls replaced
  by a single finally block.
- _reply_to_email: logout() called before the error check, so an
  exception in conn.select() leaked the socket. Wrapped in
  try/finally.
- _download_attachment: same pattern as _reply_to_email.

Also adds tests/test_imap_leak_fixes.py with 9 regression tests (one
per function/failure-mode) that monkeypatch _imap_connect and assert
conn.logout() is called exactly once even when IMAP operations raise.

README.md

@@ -1,5 +1,7 @@
 # Odysseus
 
+> **Branch note:** `dev` is the default branch and contains the latest development changes, but it may be unstable. For the more stable curated branch, use [`main`](https://github.com/pewdiepie-archdaemon/odysseus/tree/main).
+
 ```
 ───────────────────────────────────────────────
  ⊹ ࣪ ˖ ૮( ˶ᵔ ᵕ ᵔ˶ )っ  Odysseus vers. 1.0

core/middleware.py

@@ -63,6 +63,14 @@ class SecurityHeadersMiddleware(BaseHTTPMiddleware):
 
         response.headers["X-Content-Type-Options"] = "nosniff"
         response.headers["Referrer-Policy"] = "no-referrer"
+        response.headers["Permissions-Policy"] = "camera=(), microphone=(self), geolocation=()"
+
+        is_https = (
+            request.url.scheme == "https"
+            or request.headers.get("X-Forwarded-Proto") == "https"
+        )
+        if is_https:
+            response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
 
         if is_report:
             response.headers["Content-Security-Policy"] = (

mcp_servers/email_server.py

@@ -423,68 +423,71 @@ def _list_emails(folder="INBOX", max_results=20, unresponded_only=False,
     Pass unread_only=True and/or unresponded_only=True for attention scans.
     account selects mailbox (None = default).
     """
-    conn = _imap_connect(account)
-    select_status, _ = conn.select(_q(folder), readonly=True)
-    if select_status != "OK":
-        conn.logout()
-        raise ValueError(f"IMAP folder not found: {folder}")
+    conn = None
+    try:
+        conn = _imap_connect(account)
+        select_status, _ = conn.select(_q(folder), readonly=True)
+        if select_status != "OK":
+            raise ValueError(f"IMAP folder not found: {folder}")
 
-    if unread_only and unresponded_only:
-        status, data = conn.uid("SEARCH", None, "(UNSEEN UNANSWERED)")
-    elif unread_only:
-        status, data = conn.uid("SEARCH", None, "(UNSEEN)")
-    elif unresponded_only:
-        # Was missing — unresponded_only=True (without unread_only) fell through
-        # to "ALL" and returned answered mail too, despite the documented
-        # "emails without replies" behaviour.
-        status, data = conn.uid("SEARCH", None, "(UNANSWERED)")
-    else:
-        # Include read too — IMAP search "ALL" returns the entire folder
-        status, data = conn.uid("SEARCH", None, "ALL")
+        if unread_only and unresponded_only:
+            status, data = conn.uid("SEARCH", None, "(UNSEEN UNANSWERED)")
+        elif unread_only:
+            status, data = conn.uid("SEARCH", None, "(UNSEEN)")
+        elif unresponded_only:
+            # Was missing — unresponded_only=True (without unread_only) fell through
+            # to "ALL" and returned answered mail too, despite the documented
+            # "emails without replies" behaviour.
+            status, data = conn.uid("SEARCH", None, "(UNANSWERED)")
+        else:
+            # Include read too — IMAP search "ALL" returns the entire folder
+            status, data = conn.uid("SEARCH", None, "ALL")
 
-    if status != "OK" or not data[0]:
-        conn.logout()
-        return []
+        if status != "OK" or not data[0]:
+            return []
 
-    uid_list = list(reversed(data[0].split()))[:max_results]
-    cache = _get_cached_summaries()
-    results = []
+        uid_list = list(reversed(data[0].split()))[:max_results]
+        cache = _get_cached_summaries()
+        results = []
 
-    for uid in uid_list:
-        try:
-            status, msg_data = conn.uid("FETCH", uid, "(RFC822.HEADER)")
-            if status != "OK":
+        for uid in uid_list:
+            try:
+                status, msg_data = conn.uid("FETCH", uid, "(RFC822.HEADER)")
+                if status != "OK":
+                    continue
+                raw_header = msg_data[0][1]
+                msg = email.message_from_bytes(raw_header)
+
+                subject = _decode_header(msg.get("Subject", "(no subject)"))
+                sender = _decode_header(msg.get("From", "unknown"))
+                date_str = msg.get("Date", "")
+                message_id = msg.get("Message-ID", "")
+
+                # Parse sender name
+                sender_name, sender_addr = email.utils.parseaddr(sender)
+                sender_display = sender_name or sender_addr
+
+                # Check cache for summary
+                cached = cache.get(subject, {})
+                summary = cached.get("summary", "")
+
+                results.append({
+                    "uid": uid.decode(),
+                    "message_id": message_id,
+                    "subject": subject,
+                    "from": sender_display,
+                    "from_address": sender_addr,
+                    "date": date_str,
+                    "summary": summary,
+                })
+            except Exception:
                 continue
-            raw_header = msg_data[0][1]
-            msg = email.message_from_bytes(raw_header)
 
-            subject = _decode_header(msg.get("Subject", "(no subject)"))
-            sender = _decode_header(msg.get("From", "unknown"))
-            date_str = msg.get("Date", "")
-            message_id = msg.get("Message-ID", "")
-
-            # Parse sender name
-            sender_name, sender_addr = email.utils.parseaddr(sender)
-            sender_display = sender_name or sender_addr
-
-            # Check cache for summary
-            cached = cache.get(subject, {})
-            summary = cached.get("summary", "")
-
-            results.append({
-                "uid": uid.decode(),
-                "message_id": message_id,
-                "subject": subject,
-                "from": sender_display,
-                "from_address": sender_addr,
-                "date": date_str,
-                "summary": summary,
-            })
-        except Exception:
-            continue
-
-    conn.logout()
-    return results
+        return results
+    finally:
+        if conn:
+            try: conn.logout()
+            except Exception: pass
 
 
 def _result_sort_time(result: dict) -> datetime:
@@ -657,54 +660,55 @@ def _extract_attachment_to_disk(msg, index, target_dir):
 def _read_email(uid=None, message_id=None, folder="INBOX", account=None):
     """Read full email content by UID or message-ID. account = mailbox selector."""
     cfg = _load_config(account)
-    conn = _imap_connect(account)
-    conn.select(_q(folder), readonly=True)
+    conn = None
+    try:
+        conn = _imap_connect(account)
+        conn.select(_q(folder), readonly=True)
 
-    if message_id and not uid:
-        status, data = conn.uid("SEARCH", None, f'(HEADER Message-ID "{message_id}")')
-        if status != "OK" or not data[0]:
-            conn.logout()
-            return {"error": f"Email not found with Message-ID: {message_id}"}
-        uid = data[0].split()[-1]
+        if message_id and not uid:
+            status, data = conn.uid("SEARCH", None, f'(HEADER Message-ID "{message_id}")')
+            if status != "OK" or not data[0]:
+                return {"error": f"Email not found with Message-ID: {message_id}"}
+            uid = data[0].split()[-1]
 
-    if not uid:
-        conn.logout()
-        return {"error": "No UID or Message-ID provided"}
+        if not uid:
+            return {"error": "No UID or Message-ID provided"}
 
-    status, msg_data = conn.uid("FETCH", _b(uid), "(BODY.PEEK[])")
-    if status != "OK":
-        conn.logout()
-        return {"error": f"Failed to fetch email UID {uid}"}
-    if not msg_data or not msg_data[0] or not isinstance(msg_data[0], tuple) or len(msg_data[0]) < 2:
-        conn.logout()
-        return {"error": f"Email not found with UID {uid}"}
+        status, msg_data = conn.uid("FETCH", _b(uid), "(BODY.PEEK[])")
+        if status != "OK":
+            return {"error": f"Failed to fetch email UID {uid}"}
+        if not msg_data or not msg_data[0] or not isinstance(msg_data[0], tuple) or len(msg_data[0]) < 2:
+            return {"error": f"Email not found with UID {uid}"}
 
-    raw = msg_data[0][1]
-    msg = email.message_from_bytes(raw)
+        raw = msg_data[0][1]
+        msg = email.message_from_bytes(raw)
 
-    subject = _decode_header(msg.get("Subject", "(no subject)"))
-    sender = _decode_header(msg.get("From", "unknown"))
-    date_str = msg.get("Date", "")
-    message_id_header = msg.get("Message-ID", "")
-    body = _extract_text(msg)
-    attachments = _list_attachments_from_msg(msg)
+        subject = _decode_header(msg.get("Subject", "(no subject)"))
+        sender = _decode_header(msg.get("From", "unknown"))
+        date_str = msg.get("Date", "")
+        message_id_header = msg.get("Message-ID", "")
+        body = _extract_text(msg)
+        attachments = _list_attachments_from_msg(msg)
 
-    sender_name, sender_addr = email.utils.parseaddr(sender)
+        sender_name, sender_addr = email.utils.parseaddr(sender)
 
-    conn.logout()
-    return {
-        "uid": uid.decode() if isinstance(uid, bytes) else str(uid),
-        "account": cfg.get("account_name") or cfg.get("imap_user") or "default",
-        "account_email": cfg.get("imap_user") or cfg.get("from_address") or "",
-        "account_id": cfg.get("account_id"),
-        "message_id": message_id_header,
-        "subject": subject,
-        "from": sender_name or sender_addr,
-        "from_address": sender_addr,
-        "date": date_str,
-        "body": body[:8000],
-        "attachments": attachments,
-    }
+        return {
+            "uid": uid.decode() if isinstance(uid, bytes) else str(uid),
+            "account": cfg.get("account_name") or cfg.get("imap_user") or "default",
+            "account_email": cfg.get("imap_user") or cfg.get("from_address") or "",
+            "account_id": cfg.get("account_id"),
+            "message_id": message_id_header,
+            "subject": subject,
+            "from": sender_name or sender_addr,
+            "from_address": sender_addr,
+            "date": date_str,
+            "body": body[:8000],
+            "attachments": attachments,
+        }
+    finally:
+        if conn:
+            try: conn.logout()
+            except Exception: pass
 
 
 def _read_email_across_accounts(uid=None, message_id=None, folder="INBOX"):
@@ -858,10 +862,15 @@ def _send_email(to, subject, body, in_reply_to=None, references=None, cc=None, b
 
 def _reply_to_email(uid, body, folder="INBOX", reply_all=False, account=None):
     """Reply to an existing email by UID. Threads via In-Reply-To/References."""
-    conn = _imap_connect(account)
-    conn.select(_q(folder), readonly=True)
-    status, msg_data = conn.uid("FETCH", _b(uid), "(BODY.PEEK[])")
-    conn.logout()
+    conn = None
+    try:
+        conn = _imap_connect(account)
+        conn.select(_q(folder), readonly=True)
+        status, msg_data = conn.uid("FETCH", _b(uid), "(BODY.PEEK[])")
+    finally:
+        if conn:
+            try: conn.logout()
+            except Exception: pass
     if status != "OK" or not msg_data or not msg_data[0]:
         return {"error": f"Failed to fetch email UID {uid}"}
     raw = msg_data[0][1]
@@ -1038,10 +1047,15 @@ def _archive_email(uid, folder="INBOX", account=None):
 
 def _download_attachment(uid, index, folder="INBOX", account=None):
     """Extract a specific attachment to disk and return its local path."""
-    conn = _imap_connect(account)
-    conn.select(_q(folder), readonly=True)
-    status, msg_data = conn.uid("FETCH", _b(uid), "(BODY.PEEK[])")
-    conn.logout()
+    conn = None
+    try:
+        conn = _imap_connect(account)
+        conn.select(_q(folder), readonly=True)
+        status, msg_data = conn.uid("FETCH", _b(uid), "(BODY.PEEK[])")
+    finally:
+        if conn:
+            try: conn.logout()
+            except Exception: pass
     if status != "OK":
         return {"error": f"Failed to fetch email UID {uid}"}
     raw = msg_data[0][1]

routes/calendar_routes.py

@@ -1368,7 +1368,7 @@ def setup_calendar_routes() -> APIRouter:
         "tomorrow", "next Tuesday", "in 30 minutes" resolve correctly.
         Uses the "utility" endpoint (small / fast model) to keep latency low.
         """
-        _require_user(request)
+        owner = _require_user(request)
         from src.endpoint_resolver import resolve_endpoint
         from src.llm_core import llm_call_async
         from src.text_helpers import strip_think
@@ -1394,9 +1394,9 @@ def setup_calendar_routes() -> APIRouter:
         if tz_hint:
             set_user_tz_name(tz_hint)
 
-        url, model, headers = resolve_endpoint("utility")
+        url, model, headers = resolve_endpoint("utility", owner=owner or None)
         if not url:
-            url, model, headers = resolve_endpoint("default")
+            url, model, headers = resolve_endpoint("default", owner=owner or None)
         if not url or not model:
             return {"ok": False, "error": "No LLM endpoint configured"}
 

routes/document_routes.py

@@ -7,7 +7,7 @@ from typing import Dict, Any, List, Optional
 
 from fastapi import APIRouter, HTTPException, Query, Request, UploadFile, File, Form
 
-from sqlalchemy import func
+from sqlalchemy import func, or_
 from core.database import SessionLocal, Document, DocumentVersion
 from core.database import Session as DbSession
 from src.auth_helpers import get_current_user
@@ -15,6 +15,15 @@ from src.auth_helpers import get_current_user
 logger = logging.getLogger(__name__)
 
 
+def _get_session_or_404(db, session_id: str, user: Optional[str]):
+    session = db.query(DbSession).filter(DbSession.id == session_id).first()
+    if not session:
+        raise HTTPException(404, "Session not found")
+    if user and session.owner != user:
+        raise HTTPException(404, "Session not found")
+    return session
+
+
 def _aggregate_language_facets(lang_rows):
     """Sum document counts per display language for the library facet.
 
@@ -69,17 +78,12 @@ def setup_document_routes(session_manager, upload_handler=None) -> APIRouter:
             # the doc is owner-stamped, so it lives in the library on its own.
             session = None
             if req.session_id:
-                session = db.query(DbSession).filter(DbSession.id == req.session_id).first()
-                if not session:
-                    raise HTTPException(404, "Session not found")
                 # Match the lenient ownership model the rest of the app uses
                 # (see _owner_filter): only block when an AUTHENTICATED user is
                 # writing into a DIFFERENT user's session. In single-user /
-                # unconfigured / localhost-bypass mode the middleware leaves
-                # current_user unset (None), and those sessions are already
-                # served freely everywhere else.
-                if user and session.owner and session.owner != user:
-                    raise HTTPException(403, "Cannot create document in another user's session")
+                # unconfigured / localhost-bypass mode, falsey users preserve
+                # the existing lenient path.
+                session = _get_session_or_404(db, req.session_id, user)
 
             doc_id = str(uuid.uuid4())
             ver_id = str(uuid.uuid4())
@@ -171,11 +175,7 @@ def setup_document_routes(session_manager, upload_handler=None) -> APIRouter:
         if session_id:
             db = SessionLocal()
             try:
-                sess = db.query(DbSession).filter(DbSession.id == session_id).first()
-                if not sess:
-                    raise HTTPException(404, "Session not found")
-                if user and sess.owner and sess.owner != user:
-                    raise HTTPException(403, "Cannot import into another user's session")
+                _get_session_or_404(db, session_id, user)
             finally:
                 db.close()
 
@@ -198,7 +198,7 @@ def setup_document_routes(session_manager, upload_handler=None) -> APIRouter:
 
         title = os.path.splitext(meta.get("original_name") or meta.get("name") or upload_id)[0]
         try:
-            body_text = strip_pdf_content_marker(_process_pdf(pdf_path))
+            body_text = strip_pdf_content_marker(_process_pdf(pdf_path, owner=user))
         except Exception:
             body_text = None
 
@@ -359,18 +359,17 @@ def setup_document_routes(session_manager, upload_handler=None) -> APIRouter:
         try:
             if not user:
                 raise HTTPException(403, "Authentication required")
-            session = db.query(DbSession).filter(DbSession.id == session_id).first()
             # v2 review HIGH-9: raise 403 explicitly when the caller
             # can't see this session, instead of returning [] which the
             # UI treats identically to "no docs" and silently masks
             # auth failures.
-            if not session:
-                raise HTTPException(404, "Session not found")
-            if user and session.owner and session.owner != user:
-                raise HTTPException(403, "Access denied")
-            docs = db.query(Document).filter(
+            _get_session_or_404(db, session_id, user)
+            q = db.query(Document).filter(
                 Document.session_id == session_id
-            ).order_by(Document.created_at.desc()).all()
+            )
+            if user:
+                q = q.filter(or_(Document.owner == user, Document.owner.is_(None)))
+            docs = q.order_by(Document.created_at.desc()).all()
             return [_doc_to_dict(d) for d in docs]
         finally:
             db.close()
@@ -437,7 +436,7 @@ def setup_document_routes(session_manager, upload_handler=None) -> APIRouter:
                 raise HTTPException(404, "Source PDF could not be located")
 
             try:
-                body_text = strip_pdf_content_marker(_process_pdf(pdf_path))
+                body_text = strip_pdf_content_marker(_process_pdf(pdf_path, owner=user))
             except Exception as e:
                 logger.error(f"extract_pdf_text failed for {pdf_path}: {e}")
                 raise HTTPException(500, f"Extraction failed: {e}")
@@ -606,6 +605,8 @@ def setup_document_routes(session_manager, upload_handler=None) -> APIRouter:
                 doc.language = req.language
             if req.session_id is not None:
                 # Empty string = unlink from session
+                if req.session_id:
+                    _get_session_or_404(db, req.session_id, user)
                 doc.session_id = req.session_id if req.session_id else None
                 if not req.session_id:
                     # Tab closed / doc detached from its session — drop the
@@ -853,10 +854,10 @@ def setup_document_routes(session_manager, upload_handler=None) -> APIRouter:
         from src.llm_core import llm_call_async
 
         user = get_current_user(request)
-        url, model, headers = resolve_task_endpoint()
+        url, model, headers = resolve_task_endpoint(owner=user or None)
         if not url or not model:
             # Fall back to default endpoint
-            url, model, headers = resolve_endpoint("default")
+            url, model, headers = resolve_endpoint("default", owner=user or None)
         if not url or not model:
             raise HTTPException(500, "No endpoint configured for AI tidy")
 
@@ -1156,7 +1157,7 @@ def setup_document_routes(session_manager, upload_handler=None) -> APIRouter:
         settings = _load_vl_settings()
         vl_model = settings.get("vision_model", "")
         try:
-            url, model_id, headers = _resolve_vl_model(vl_model)
+            url, model_id, headers = _resolve_vl_model(vl_model, owner=user)
         except Exception as e:
             raise HTTPException(503, f"No vision model available: {e}")
 

routes/email_helpers.py

@@ -1140,13 +1140,9 @@ def _fetch_sender_thread_context(sender_addr: str,
     if exclude_uid:
         seen_uids.add((exclude_folder or "INBOX", str(exclude_uid)))
 
+    conn = None
     try:
         conn = _imap_connect(account_id, owner=owner)
-    except Exception as e:
-        logger.warning(f"sender-thread-context: imap connect failed: {e}")
-        return ""
-
-    try:
         for folder in ["INBOX", "Sent", "Archive", "Drafts"]:
             if len(blocks) >= limit:
                 break
@@ -1213,11 +1209,14 @@ def _fetch_sender_thread_context(sender_addr: str,
                 if atts_text:
                     lines.append(atts_text)
                 blocks.append("\n".join(lines))
+    except Exception as e:
+        logger.warning(f"sender-thread-context: imap failed: {e}")
     finally:
-        try: conn.close()
-        except Exception: pass
-        try: conn.logout()
-        except Exception: pass
+        if conn:
+            try: conn.close()
+            except Exception: pass
+            try: conn.logout()
+            except Exception: pass
 
     if not blocks:
         return ""
@@ -1320,6 +1319,7 @@ def _pre_retrieve_context(
         if not terms_list:
             return context_snippets, terms_list
 
+        ctx_conn = None
         try:
             ctx_conn = _imap_connect(account_id, owner=owner)
             for folder in ["INBOX", "Sent", "Archive", "Drafts"]:
@@ -1356,12 +1356,12 @@ def _pre_retrieve_context(
                     except Exception as _e:
                         logger.warning(f"  search {folder} {term!r} failed: {_e}")
                         continue
-            try:
-                ctx_conn.logout()
-            except Exception:
-                pass
         except Exception as _e:
             logger.warning(f"IMAP context search failed: {_e}")
+        finally:
+            if ctx_conn:
+                try: ctx_conn.logout()
+                except Exception: pass
 
         try:
             from routes.contacts_routes import _fetch_contacts

routes/gallery_routes.py

@@ -53,6 +53,46 @@ def _gallery_image_path(filename: str) -> Path:
         raise HTTPException(400, "Unsafe gallery filename")
     return path
 
+
+def _normalize_image_endpoint_base(url: str) -> str:
+    base = (url or "").strip().rstrip("/")
+    if base.endswith("/v1"):
+        base = base[:-3].rstrip("/")
+    return base
+
+
+def _visible_image_endpoint_query(db, owner: str | None):
+    from src.auth_helpers import owner_filter
+    q = db.query(ModelEndpoint).filter(
+        ModelEndpoint.model_type == "image",
+        ModelEndpoint.is_enabled == True,  # noqa: E712
+    )
+    return owner_filter(q, ModelEndpoint, owner)
+
+
+def _first_visible_image_endpoint(db, owner: str | None):
+    endpoints = _visible_image_endpoint_query(db, owner).all()
+    if owner:
+        for ep in endpoints:
+            if getattr(ep, "owner", None) == owner:
+                return ep
+    return endpoints[0] if endpoints else None
+
+
+def _visible_image_endpoint_for_base(db, base: str, owner: str | None):
+    target = _normalize_image_endpoint_base(base)
+    if not target:
+        return None
+    fallback = None
+    for ep in _visible_image_endpoint_query(db, owner).all():
+        if _normalize_image_endpoint_base(getattr(ep, "base_url", "")) == target:
+            if owner and getattr(ep, "owner", None) == owner:
+                return ep
+            if fallback is None:
+                fallback = ep
+    return fallback
+
+
 def setup_gallery_routes() -> APIRouter:
     router = APIRouter(tags=["gallery"])
 
@@ -76,6 +116,9 @@ def setup_gallery_routes() -> APIRouter:
         file_hash = hashlib.sha256(content).hexdigest()
         db = SessionLocal()
         try:
+            if album_id and user is not None:
+                _get_or_404_album(db, album_id, user)
+
             # SECURITY: scope the dup-detect to THIS user — otherwise a
             # caller can probe whether someone else uploaded the same
             # file (the response leaks the existing row's id+filename).
@@ -269,7 +312,7 @@ def setup_gallery_routes() -> APIRouter:
         """AI upscale using img2img with the diffusion server."""
         import base64, httpx
 
-        require_privilege(request, "can_generate_images")
+        user = require_privilege(request, "can_generate_images")
         form = await request.form()
         file = form.get("image")
         if not file: raise HTTPException(400, "No image")
@@ -281,7 +324,7 @@ def setup_gallery_routes() -> APIRouter:
         # Find image endpoint
         db = SessionLocal()
         try:
-            ep = db.query(ModelEndpoint).filter(ModelEndpoint.model_type == "image", ModelEndpoint.is_enabled == True).first()
+            ep = _first_visible_image_endpoint(db, user)
         finally:
             db.close()
 
@@ -312,7 +355,7 @@ def setup_gallery_routes() -> APIRouter:
         """Style transfer using img2img with the diffusion server."""
         import base64, httpx
 
-        require_privilege(request, "can_generate_images")
+        user = require_privilege(request, "can_generate_images")
         form = await request.form()
         file = form.get("image")
         prompt = form.get("prompt", "")
@@ -324,7 +367,7 @@ def setup_gallery_routes() -> APIRouter:
 
         db = SessionLocal()
         try:
-            ep = db.query(ModelEndpoint).filter(ModelEndpoint.model_type == "image", ModelEndpoint.is_enabled == True).first()
+            ep = _first_visible_image_endpoint(db, user)
         finally:
             db.close()
 
@@ -957,7 +1000,7 @@ def setup_gallery_routes() -> APIRouter:
         the request for /v1/images/edits (multipart, inverted mask). Otherwise
         proxy through to a self-hosted diffusion server's /v1/images/inpaint."""
         import httpx
-        require_privilege(request, "can_generate_images")
+        user = require_privilege(request, "can_generate_images")
         body = await request.json()
         # Use endpoint from request body (editor dropdown) or fall back to DB lookup
         base = (body.pop("_endpoint", "") or "").rstrip("/")
@@ -976,14 +1019,11 @@ def setup_gallery_routes() -> APIRouter:
         if not base:
             db = SessionLocal()
             try:
-                eps = db.query(ModelEndpoint).filter(
-                    ModelEndpoint.is_enabled == True,
-                    ModelEndpoint.model_type == "image",
-                ).all()
-                if not eps:
+                ep = _first_visible_image_endpoint(db, user)
+                if not ep:
                     raise HTTPException(400, "No image generation endpoint configured. Serve a diffusion model via Cookbook first.")
-                base = eps[0].base_url.rstrip("/")
-                api_key = eps[0].api_key
+                base = ep.base_url.rstrip("/")
+                api_key = ep.api_key
             finally:
                 db.close()
         else:
@@ -1000,10 +1040,9 @@ def setup_gallery_routes() -> APIRouter:
             _target = _norm_url(base)
             db = SessionLocal()
             try:
-                for ep in db.query(ModelEndpoint).all():
-                    if _norm_url(ep.base_url) == _target:
-                        api_key = ep.api_key
-                        break
+                ep = _visible_image_endpoint_for_base(db, _target, user)
+                if ep:
+                    api_key = ep.api_key
             finally:
                 db.close()
 
@@ -1155,7 +1194,7 @@ def setup_gallery_routes() -> APIRouter:
         you get edge blending + lighting unification while keeping the
         composition recognisable."""
         import httpx, base64 as _b64
-        require_privilege(request, "can_generate_images")
+        user = require_privilege(request, "can_generate_images")
         body = await request.json()
 
         image_b64 = body.get("image")
@@ -1182,23 +1221,19 @@ def setup_gallery_routes() -> APIRouter:
         if not base:
             db = SessionLocal()
             try:
-                eps = db.query(ModelEndpoint).filter(
-                    ModelEndpoint.is_enabled == True,
-                    ModelEndpoint.model_type == "image",
-                ).all()
-                if not eps:
+                ep = _first_visible_image_endpoint(db, user)
+                if not ep:
                     raise HTTPException(400, "No image generation endpoint configured.")
-                base = eps[0].base_url.rstrip("/")
-                api_key = eps[0].api_key
+                base = ep.base_url.rstrip("/")
+                api_key = ep.api_key
             finally:
                 db.close()
         else:
             db = SessionLocal()
             try:
-                for ep in db.query(ModelEndpoint).all():
-                    if ep.base_url.rstrip("/").removesuffix("/v1").rstrip("/") == base.rstrip("/").removesuffix("/v1").rstrip("/"):
-                        api_key = ep.api_key
-                        break
+                ep = _visible_image_endpoint_for_base(db, base, user)
+                if ep:
+                    api_key = ep.api_key
             finally:
                 db.close()
 
@@ -1669,9 +1704,10 @@ def setup_gallery_routes() -> APIRouter:
         db = SessionLocal()
         try:
             album = _get_or_404_album(db, album_id, user)
-            db.query(GalleryImage).filter(GalleryImage.album_id == album_id).update(
-                {"album_id": None}, synchronize_session=False
-            )
+            q = db.query(GalleryImage).filter(GalleryImage.album_id == album_id)
+            if user is not None:
+                q = q.filter(GalleryImage.owner == user)
+            q.update({"album_id": None}, synchronize_session=False)
             db.delete(album)
             db.commit()
             return {"ok": True}
@@ -1760,7 +1796,7 @@ def setup_gallery_routes() -> APIRouter:
                 return {"error": "Vision is disabled — enable it in Settings → Vision"}
             configured = vl_settings.get("vision_model", "")
             try:
-                chat_url, model_name, headers = _resolve_vl_model(configured)
+                chat_url, model_name, headers = _resolve_vl_model(configured, owner=user)
             except ValueError:
                 return {"error": "No vision model configured — set one in Settings → Vision"}
             if not chat_url:

routes/history_routes.py

@@ -522,6 +522,8 @@ def setup_history_routes(session_manager) -> APIRouter:
     async def compact_session(request: Request, session_id: str):
         """Manually trigger context compaction for a session."""
         _verify_session_owner(request, session_id)
+        from src.auth_helpers import effective_user
+        owner = effective_user(request)
         try:
             session = session_manager.get_session(session_id)
         except KeyError:
@@ -555,7 +557,7 @@ def setup_history_routes(session_manager) -> APIRouter:
             )
 
             # Use utility model if available
-            util_url, util_model, util_headers = resolve_endpoint("utility")
+            util_url, util_model, util_headers = resolve_endpoint("utility", owner=owner or None)
             compact_url = util_url or session.endpoint_url
             compact_model = util_model or session.model
             compact_headers = util_headers if util_url else session.headers

routes/memory_routes.py

@@ -371,7 +371,7 @@ def setup_memory_routes(memory_manager: MemoryManager, session_manager: SessionM
                 tmp.write(content)
                 tmp_path = tmp.name
             try:
-                text = _process_pdf(tmp_path)
+                text = _process_pdf(tmp_path, owner=_owner(request))
             finally:
                 os.unlink(tmp_path)
         else:

routes/note_routes.py

@@ -181,9 +181,9 @@ async def dispatch_reminder(
         try:
             from src.endpoint_resolver import resolve_endpoint
             from src.llm_core import llm_call_async
-            url, model, headers = resolve_endpoint("utility")
+            url, model, headers = resolve_endpoint("utility", owner=owner or None)
             if not url:
-                url, model, headers = resolve_endpoint("default")
+                url, model, headers = resolve_endpoint("default", owner=owner or None)
             if url and model:
                 raw = await llm_call_async(
                     url=url, model=model,

routes/task_routes.py

@@ -429,6 +429,20 @@ def setup_task_routes(task_scheduler) -> APIRouter:
         except Exception:
             return False
 
+    def _validate_then_task_id(db, then_task_id: Optional[str], user: Optional[str], current_task_id: Optional[str] = None) -> Optional[str]:
+        target_id = (then_task_id or "").strip()
+        if not target_id:
+            return None
+        if current_task_id and target_id == current_task_id:
+            raise HTTPException(400, "Task cannot chain to itself")
+        q = db.query(ScheduledTask).filter(ScheduledTask.id == target_id)
+        if user:
+            q = q.filter(ScheduledTask.owner == user)
+        target = q.first()
+        if not target:
+            raise HTTPException(404, "Chained task not found")
+        return target.id
+
     @router.post("")
     async def create_task(request: Request, req: TaskCreate):
         user = _owner(request)
@@ -492,6 +506,7 @@ def setup_task_routes(task_scheduler) -> APIRouter:
         task_id = str(uuid.uuid4())
         db = SessionLocal()
         try:
+            then_task_id = _validate_then_task_id(db, req.then_task_id, user)
             notifications_enabled = (
                 False if req.task_type == "action" and req.notifications_enabled is None
                 else bool(req.notifications_enabled) if req.notifications_enabled is not None
@@ -518,7 +533,7 @@ def setup_task_routes(task_scheduler) -> APIRouter:
                 output_target=req.output_target,
                 model=req.model or None,
                 endpoint_url=req.endpoint_url or None,
-                then_task_id=req.then_task_id or None,
+                then_task_id=then_task_id,
                 webhook_token=webhook_token,
                 notifications_enabled=notifications_enabled,
             )
@@ -671,7 +686,7 @@ def setup_task_routes(task_scheduler) -> APIRouter:
             if req.trigger_count is not None:
                 task.trigger_count = req.trigger_count
             if req.then_task_id is not None:
-                task.then_task_id = req.then_task_id or None
+                task.then_task_id = _validate_then_task_id(db, req.then_task_id, user, current_task_id=task.id)
             if req.notifications_enabled is not None:
                 task.notifications_enabled = bool(req.notifications_enabled)
             if req.cron_expression is not None:
@@ -1047,6 +1062,7 @@ def setup_task_routes(task_scheduler) -> APIRouter:
         desc = (body.get("description") or "").strip()
         if not desc:
             return {"success": False, "message": "Nothing to parse"}
+        user = _owner(request)
 
         now = _dt.now()
         # Give the model the current date/time + weekday so relative phrasing
@@ -1073,9 +1089,9 @@ def setup_task_routes(task_scheduler) -> APIRouter:
             "use cron '0 H * * 1-5'. Keep the prompt actionable and self-contained."
         )
         try:
-            url, model, headers = resolve_endpoint("utility")
+            url, model, headers = resolve_endpoint("utility", owner=user or None)
             if not url:
-                url, model, headers = resolve_endpoint("default")
+                url, model, headers = resolve_endpoint("default", owner=user or None)
             if not (url and model):
                 return {"success": False, "message": "No model endpoint configured"}
             raw = await llm_call_async(

routes/upload_routes.py

@@ -225,7 +225,7 @@ def setup_upload_routes(upload_handler):
                 logger.warning(f"Vision cache read failed for {file_id}: {e}")
         from src.document_processor import analyze_image_with_vl
         try:
-            text = analyze_image_with_vl(path) or ""
+            text = analyze_image_with_vl(path, owner=current_user) or ""
         except Exception as e:
             logger.error(f"Vision analysis failed for {file_id}: {e}")
             raise HTTPException(500, f"Vision analysis failed: {e}")

src/caldav_sync.py

@@ -216,18 +216,43 @@ def _open_url_as_calendar(client, url: str):
     return client.calendar(url=target)
 
 
+def _build_dav_client(url: str, username: str, password: str):
+    """Construct a CalDAV client with automatic redirects disabled.
+
+    ``validate_caldav_url`` resolves and vets the *initial* host, but caldav's
+    underlying HTTP session follows 3xx redirects by default. So a URL that
+    passes validation can still be redirected — at request time — to
+    loopback / link-local / private space, re-opening the SSRF the host check
+    closes. Pin the session to zero redirects: any 3xx then raises instead of
+    silently following an attacker-chosen ``Location``. This mirrors the
+    test-connection path in ``routes/calendar_routes.py``, which already sets
+    ``follow_redirects=False``.
+
+    DAVClient exposes no per-request redirect flag, so we set it on the session
+    after construction (the session is created in ``__init__``).
+    """
+    import caldav
+
+    client = caldav.DAVClient(url=url, username=username, password=password)
+    # Unconditional: a redirect-disable that only sometimes applies is not a
+    # control. The session exists right after __init__ on every real client;
+    # test_build_dav_client_disables_redirects asserts it against installed
+    # caldav in CI.
+    client.session.max_redirects = 0
+    return client
+
+
 def _sync_blocking(owner: str, url: str, username: str, password: str, account_id: str = "") -> dict:
     """The actual sync — synchronous, intended to run in a threadpool.
     Returns counts: {calendars, events, deleted, errors}."""
     # Lazy imports so a missing `caldav` dep doesn't break app startup —
     # the integrations form still works, sync just no-ops with an error.
-    import caldav
     from caldav.lib.error import AuthorizationError, NotFoundError
     from core.database import CalendarCal, CalendarEvent, SessionLocal
 
     result = {"calendars": 0, "events": 0, "deleted": 0, "errors": []}
 
-    client = caldav.DAVClient(url=url, username=username, password=password)
+    client = _build_dav_client(url, username, password)
 
     # Discovery: try principal → calendars first; if the server doesn't
     # support discovery (or the URL points directly at a calendar), fall

src/caldav_writeback.py

@@ -143,8 +143,10 @@ def _discover_calendars(client):
 
 def _writeback_blocking(local_cal_id, ev, delete, url, username, password,
                         owner="", account_id="") -> dict:
-    import caldav
-    client = caldav.DAVClient(url=url, username=username, password=password)
+    from src.caldav_sync import _build_dav_client
+    # Redirects disabled here too: the write-back path opens its own DAVClient,
+    # so it needs the same SSRF-via-redirect protection as the pull path.
+    client = _build_dav_client(url, username, password)
     calendars = _discover_calendars(client)
     if not calendars:
         return {"ok": False, "error": "no remote calendars discovered"}

src/chat_handler.py

@@ -229,7 +229,7 @@ class ChatHandler:
                             except Exception:
                                 vl_desc = None
                         if not vl_desc:
-                            vl_result = analyze_image_with_vl_result(file_info["path"])
+                            vl_result = analyze_image_with_vl_result(file_info["path"], owner=owner)
                             vl_desc = vl_result.get("text", "")
                             vl_model = vl_result.get("model", "")
                             if vl_desc and not vl_desc.startswith("["):

src/document_processor.py

@@ -109,7 +109,7 @@ def _process_text_file(path: str) -> str:
         return result
 
 
-def _process_pdf(path: str) -> str:
+def _process_pdf(path: str, owner: str | None = None) -> str:
     """Process PDF file with text extraction (pypdf). Uses VL model for image-heavy pages."""
     try:
         from pypdf import PdfReader
@@ -133,7 +133,7 @@ def _process_pdf(path: str) -> str:
                             temp_img_path = tmp.name
                         try:
                             img.image.save(temp_img_path, "PNG")  # pypdf -> PIL image
-                            ocr_text = analyze_image_with_vl(temp_img_path)
+                            ocr_text = analyze_image_with_vl(temp_img_path, owner=owner)
                             if ocr_text and "unavailable" not in ocr_text.lower():
                                 pdf_text += f"\n\n[Page {page_num + 1} image {img_index + 1} text]: {ocr_text}"
                         finally:
@@ -254,7 +254,7 @@ def _load_vl_settings() -> dict:
         return {}
 
 
-def _resolve_vl_model(configured: str) -> tuple:
+def _resolve_vl_model(configured: str, owner: str | None = None) -> tuple:
     """Resolve the vision model to (url, model_id, headers).
 
     Uses admin-configured model if set, otherwise tries auto-detection
@@ -263,7 +263,7 @@ def _resolve_vl_model(configured: str) -> tuple:
     from src.ai_interaction import _resolve_model
 
     if configured:
-        return _resolve_model(configured)
+        return _resolve_model(configured, owner=owner)
 
     # Auto-detect: try known vision-capable models in priority order
     candidates = [
@@ -274,14 +274,14 @@ def _resolve_vl_model(configured: str) -> tuple:
     ]
     for candidate in candidates:
         try:
-            return _resolve_model(candidate)
+            return _resolve_model(candidate, owner=owner)
         except (ValueError, Exception):
             continue
 
     raise ValueError("No vision model available")
 
 
-def analyze_image_with_vl_result(image_path: str) -> dict:
+def analyze_image_with_vl_result(image_path: str, owner: str | None = None) -> dict:
     """Analyze an image and return both text and the model that produced it."""
     logger.info(f"Analyzing image with VL model: {image_path}")
     try:
@@ -291,7 +291,7 @@ def analyze_image_with_vl_result(image_path: str) -> dict:
         vl_model = settings.get("vision_model", "")
 
         try:
-            url, model_id, headers = _resolve_vl_model(vl_model)
+            url, model_id, headers = _resolve_vl_model(vl_model, owner=owner)
         except ValueError:
             return {"text": "[No vision model configured — set one in Settings → Vision]", "model": vl_model or ""}
 
@@ -316,7 +316,7 @@ def analyze_image_with_vl_result(image_path: str) -> dict:
         # — same shape as task/chat but its own list (`vision_model_fallbacks`).
         try:
             from src.endpoint_resolver import resolve_vision_fallback_candidates
-            _vl_candidates = [(url, model_id, headers)] + resolve_vision_fallback_candidates()
+            _vl_candidates = [(url, model_id, headers)] + resolve_vision_fallback_candidates(owner=owner)
         except Exception:
             _vl_candidates = [(url, model_id, headers)]
 
@@ -338,9 +338,9 @@ def analyze_image_with_vl_result(image_path: str) -> dict:
         return {"text": "[VL model unavailable - image not analyzed]", "model": ""}
 
 
-def analyze_image_with_vl(image_path: str) -> str:
+def analyze_image_with_vl(image_path: str, owner: str | None = None) -> str:
     """Analyze an image using the admin-configured Vision-Language model."""
-    return analyze_image_with_vl_result(image_path).get("text", "")
+    return analyze_image_with_vl_result(image_path, owner=owner).get("text", "")
 
 
 def build_user_content(
@@ -434,7 +434,7 @@ def build_user_content(
                         # Pull the PDF prose once — used as either intro_text
                         # (form path) or the doc body (plain path).
                         try:
-                            pdf_body_text = strip_pdf_content_marker(_process_pdf(path))
+                            pdf_body_text = strip_pdf_content_marker(_process_pdf(path, owner=owner))
                         except Exception:
                             pdf_body_text = None
 
@@ -517,7 +517,7 @@ def build_user_content(
                     except Exception as e:
                         logger.warning(f"PDF auto-doc creation failed for {path}: {e}")
                 if extracted_text is None:
-                    extracted_text = _process_pdf(path)
+                    extracted_text = _process_pdf(path, owner=owner)
             elif mime.startswith("text/") or _is_text_file(path):
                 extracted_text = _process_text_file(path)
             else:

src/task_scheduler.py

@@ -844,7 +844,13 @@ class TaskScheduler:
             # Task chaining — trigger the next task on success
             if run.status == "success" and task.then_task_id:
                 chain_id = task.then_task_id
-                if not self._has_chain_cycle(db, chain_id):
+                chain_task = db.query(ScheduledTask).filter(ScheduledTask.id == chain_id).first()
+                if not chain_task or chain_task.owner != task.owner:
+                    logger.warning(
+                        "Skipping chain from %r: target task %s is missing or not owned by %r",
+                        task.name, chain_id, task.owner,
+                    )
+                elif not self._has_chain_cycle(db, chain_id, owner=task.owner):
                     logger.info(f"Chaining: '{task.name}' → task {chain_id}")
                     asyncio.create_task(self._run_chained(chain_id))
                 else:
@@ -1791,7 +1797,7 @@ class TaskScheduler:
             self._executing.add(task_id)
         await self._execute_task(task_id)
 
-    def _has_chain_cycle(self, db, start_id: str, max_depth: int = 10) -> bool:
+    def _has_chain_cycle(self, db, start_id: str, max_depth: int = 10, owner: str | None = None) -> bool:
         """Detect cycles in task chains."""
         from core.database import ScheduledTask
         visited = set()
@@ -1801,6 +1807,8 @@ class TaskScheduler:
                 return True
             visited.add(current)
             task = db.query(ScheduledTask).filter(ScheduledTask.id == current).first()
+            if owner is not None and task and task.owner != owner:
+                return True
             if not task or not task.then_task_id:
                 return False
             current = task.then_task_id

src/tool_implementations.py

@@ -1828,6 +1828,22 @@ async def do_manage_notes(content: str, owner: Optional[str] = None) -> Dict:
         text = re.sub(r"^\s*reminder\s*:\s*", "", text)
         return re.sub(r"\s+", " ", text)
 
+    def _note_visible_to_owner(note, owner_value: Optional[str]) -> bool:
+        # Empty owner_value is single-user / auth-disabled mode. A real
+        # authenticated owner must match exactly; null/empty legacy rows are not
+        # shared between accounts.
+        if not owner_value:
+            return True
+        return getattr(note, "owner", None) == owner_value
+
+    def _note_by_prefix(note_id: str):
+        if not note_id:
+            return None
+        q = db.query(Note).filter(Note.id.startswith(note_id))
+        if owner:
+            q = q.filter(Note.owner == owner)
+        return q.first()
+
     try:
         if action == "list":
             q = db.query(Note)
@@ -1947,10 +1963,10 @@ async def do_manage_notes(content: str, owner: Optional[str] = None) -> Dict:
 
         elif action == "update":
             note_id = args.get("id", "")
-            note = db.query(Note).filter(Note.id.startswith(note_id)).first() if note_id else None
+            note = _note_by_prefix(note_id)
             if not note:
                 return {"error": f"Note '{note_id}' not found", "exit_code": 1}
-            if owner is not None and note.owner and note.owner != owner:
+            if not _note_visible_to_owner(note, owner):
                 return {"error": "Note not found", "exit_code": 1}
             for field in ("title", "content", "note_type", "color", "label"):
                 if field in args and args[field] is not None:
@@ -1983,10 +1999,10 @@ async def do_manage_notes(content: str, owner: Optional[str] = None) -> Dict:
 
         elif action == "delete":
             note_id = args.get("id", "")
-            note = db.query(Note).filter(Note.id.startswith(note_id)).first() if note_id else None
+            note = _note_by_prefix(note_id)
             if not note:
                 return {"error": f"Note '{note_id}' not found", "exit_code": 1}
-            if owner is not None and note.owner and note.owner != owner:
+            if not _note_visible_to_owner(note, owner):
                 return {"error": "Note not found", "exit_code": 1}
             title = note.title
             db.delete(note)
@@ -1996,10 +2012,10 @@ async def do_manage_notes(content: str, owner: Optional[str] = None) -> Dict:
         elif action == "toggle_item":
             note_id = args.get("id", "")
             index = args.get("index", 0)
-            note = db.query(Note).filter(Note.id.startswith(note_id)).first() if note_id else None
+            note = _note_by_prefix(note_id)
             if not note:
                 return {"error": f"Note '{note_id}' not found", "exit_code": 1}
-            if owner is not None and note.owner and note.owner != owner:
+            if not _note_visible_to_owner(note, owner):
                 return {"error": "Note not found", "exit_code": 1}
             if not note.items:
                 return {"error": "Note has no checklist items", "exit_code": 1}

src/webhook_manager.py

@@ -136,11 +136,62 @@ def validate_events(events_str: str) -> str:
     return ",".join(events)
 
 
+# Broad candidate matcher for the IP-redaction pass. Deliberately loose: a
+# bracketed host authority ([fe80::1%eth0]:8080 and friends) with an optional
+# :port, or a bare IPv6 run — hex groups joined by colons, an optional trailing
+# dotted-quad for IPv4-mapped forms (::ffff:192.168.0.1), and an optional %zone.
+# It does NOT encode the IPv6 grammar; ipaddress.ip_address() is the real
+# validator (see _redact_ip_candidate), so any colon-bearing string it rejects
+# (clock times, MACs, "std::vector") is left alone. Every branch is a single
+# greedy class or a repetition over a mandatory ':'/'.' delimiter, so there is no
+# nested-quantifier backtracking (ReDoS-safe).
+_IP_CANDIDATE = re.compile(
+    r'\[[^\[\]\s]*\](?::\d+)?'
+    r'|(?<![\w.:%])[0-9A-Fa-f]{0,4}(?::[0-9A-Fa-f]{0,4}){2,}'
+    r'(?:(?:\.[0-9]{1,3}){3})?(?:%[0-9A-Za-z._-]+)?'
+)
+
+
+def _redact_ip_candidate(match: re.Match) -> str:
+    """Redact a candidate token that the stdlib confirms is an IP address.
+
+    A bare token is redacted only when it parses as IPv6 — bare IPv4 is left to
+    the dedicated IPv4 pass. A bracketed token is a host authority, so a v4 or v6
+    literal inside [ ] is redacted as a whole. This keeps output consistent (one
+    [redacted], never nested or partial) for scoped/mapped/ported forms.
+    """
+    token = match.group(0)
+    bracketed = token.startswith('[')
+    candidate = token
+    if bracketed:
+        # Keep only what's inside [...]; the trailing :port is dropped.
+        candidate = candidate[1:candidate.index(']')]
+    # A zone id (fe80::1%eth0) is not part of the address ipaddress parses.
+    candidate = candidate.split('%', 1)[0]
+    # The loose bare pattern can trail one stray ':' (e.g. "::1:" in "host ::1:
+    # down"); drop it unless it's the "::" compression marker.
+    if candidate.endswith(':') and not candidate.endswith('::'):
+        candidate = candidate[:-1]
+    try:
+        addr = ipaddress.ip_address(candidate)
+    except ValueError:
+        return token
+    if bracketed or isinstance(addr, ipaddress.IPv6Address):
+        return '[redacted]'
+    return token
+
+
 def sanitize_error(error: str, max_len: int = 200) -> str:
     """Strip potentially sensitive details from error messages."""
-    # Remove IP addresses and ports
-    cleaned = re.sub(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}(:\d+)?', '[redacted]', error)
-    # Remove hostnames in URLs
+    # Redact IPv6 (and bracketed-authority) addresses first, so an IPv4-mapped
+    # form like ::ffff:192.168.0.1 is scrubbed as one unit instead of having its
+    # embedded IPv4 removed first and leaving a stray "::ffff:" behind. Broad
+    # candidates are validated by ipaddress.ip_address(), so the false-positive
+    # guards (clock times, MACs, C++ "::") come from the stdlib, not a regex.
+    cleaned = _IP_CANDIDATE.sub(_redact_ip_candidate, error)
+    # Remove remaining bare IPv4 addresses and ports.
+    cleaned = re.sub(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}(:\d+)?', '[redacted]', cleaned)
+    # Remove hostnames in URLs.
     cleaned = re.sub(r'https?://[^\s/]+', '[redacted-url]', cleaned)
     return cleaned[:max_len]
 

tests/test_build_user_content_pdf_marker.py

@@ -35,7 +35,7 @@ def test_pdf_body_marker_stripped_without_eating_text(monkeypatch, tmp_path):
 
     # Shape _process_pdf actually returns: marker, then a page-text marker, then body.
     raw = "\n\n[PDF content]:\n\n[Page 1 text]:\nto the board, the agenda is set"
-    monkeypatch.setattr(dp, "_process_pdf", lambda path: raw)
+    monkeypatch.setattr(dp, "_process_pdf", lambda path, owner=None: raw)
     monkeypatch.setattr(pdf_forms, "has_form_fields", lambda path: False)
     monkeypatch.setattr(pdf_form_doc, "create_plain_pdf_document", lambda **kw: "doc-123")
 

tests/test_caldav_google_principal_url.py

@@ -83,6 +83,9 @@ class _FakePrincipal:
 class _FakeClient:
     def __init__(self, url=None, username=None, password=None):
         self.url = url
+        # Mirror the real DAVClient: _build_dav_client sets
+        # session.max_redirects = 0 right after construction.
+        self.session = types.SimpleNamespace(max_redirects=30)
 
     def principal(self):
         return _FakePrincipal()

tests/test_caldav_redirect_hardening.py

@@ -0,0 +1,105 @@
+"""CalDAV SSRF-via-redirect hardening.
+
+``validate_caldav_url`` resolves and vets the initial host, but the CalDAV
+client's HTTP session follows 3xx redirects by default — so a validated public
+URL can be redirected, at request time, into loopback/private space (an SSRF
+that bypasses the host check). ``_build_dav_client`` pins the session to zero
+redirects. These tests exercise the real DAVClient request path (the sync /
+write-back surface), not just the settings/test-connection endpoint.
+"""
+
+import http.server
+import socketserver
+import threading
+
+import pytest
+
+from src import caldav_sync, caldav_writeback
+
+
+def test_build_dav_client_disables_redirects():
+    """The hardened client must carry a redirect-disabled session."""
+    pytest.importorskip("caldav")
+    client = caldav_sync._build_dav_client("https://calendar.example.com/dav", "u", "p")
+    assert client.session.max_redirects == 0
+
+
+def test_dav_client_does_not_follow_redirect_to_internal_host():
+    """End-to-end through the real DAVClient: a 302 toward an internal host
+    must NOT be followed. Without the fix the sink is contacted (SSRF); with it
+    the redirect is refused and the sink is never reached."""
+    pytest.importorskip("caldav")
+
+    sink_hits: list[str] = []
+    public_methods: list[str] = []
+
+    class _Internal(http.server.BaseHTTPRequestHandler):
+        # Stand-in for an internal service the attacker redirects toward.
+        def do_GET(self):  # noqa: N802
+            sink_hits.append(self.path)
+            self.send_response(207)
+            self.end_headers()
+
+        do_PROPFIND = do_GET
+
+        def log_message(self, *a):  # silence test server
+            pass
+
+    class _Public(http.server.BaseHTTPRequestHandler):
+        # The "validated" public CalDAV server that redirects everything inward.
+        def do_GET(self):  # noqa: N802
+            public_methods.append(self.command)
+            self.send_response(302)
+            self.send_header("Location", f"http://127.0.0.1:{internal_port}/leak")
+            self.end_headers()
+
+        do_PROPFIND = do_GET
+
+        def log_message(self, *a):
+            pass
+
+    internal = socketserver.TCPServer(("127.0.0.1", 0), _Internal)
+    internal_port = internal.server_address[1]
+    public = socketserver.TCPServer(("127.0.0.1", 0), _Public)
+    public_port = public.server_address[1]
+    threading.Thread(target=internal.serve_forever, daemon=True).start()
+    threading.Thread(target=public.serve_forever, daemon=True).start()
+    try:
+        public_url = f"http://127.0.0.1:{public_port}/dav"
+        client = caldav_sync._build_dav_client(public_url, "u", "p")
+        client.timeout = 5
+        try:
+            client.request(public_url, "PROPFIND", "")
+        except Exception:
+            # Refusing the redirect surfaces as an exception (TooManyRedirects);
+            # that is the intended fail-closed behavior. The security assertion
+            # is that the internal sink was never contacted.
+            pass
+        # The request must actually have left the building — otherwise an early
+        # error would make "sink not hit" pass vacuously.
+        assert public_methods == ["PROPFIND"], "the PROPFIND must reach the public server first"
+        assert sink_hits == [], "redirect toward an internal host must not be followed"
+    finally:
+        internal.shutdown()
+        public.shutdown()
+
+
+def test_sync_and_writeback_construct_clients_through_the_helper():
+    """Guard against a raw DAVClient (redirects enabled) creeping back in.
+    Every DAVClient on the sync/write-back paths must go through
+    ``_build_dav_client`` so the redirect protection can't be bypassed."""
+    sync_src = (caldav_sync.__file__)
+    wb_src = (caldav_writeback.__file__)
+    with open(sync_src, encoding="utf-8") as f:
+        sync_text = f.read()
+    with open(wb_src, encoding="utf-8") as f:
+        wb_text = f.read()
+
+    # In caldav_sync the only raw construction lives inside the helper itself.
+    assert sync_text.count("caldav.DAVClient(") == 1
+    assert "max_redirects = 0" in sync_text
+    assert "_build_dav_client(" in sync_text
+
+    # Write-back must not construct its own raw client; it reuses the helper.
+    assert "caldav.DAVClient(" not in wb_text
+    assert "_build_dav_client(" in wb_text

tests/test_caldav_url_hardening.py

@@ -82,6 +82,39 @@ def test_validate_caldav_url_fails_closed_when_hostname_does_not_resolve(monkeyp
         caldav_sync.validate_caldav_url("https://calendar.example.com/dav")
 
 
+def test_validate_caldav_url_fails_closed_when_host_resolves_to_no_usable_records(monkeypatch):
+    # Distinct from the OSError path above: here resolution *succeeds* but yields
+    # no usable A/AAAA records (the `if not addrs` branch). Fail closed there too
+    # rather than letting an un-vetted host through.
+    monkeypatch.setattr(caldav_sync, "_resolve_caldav_host_ips", lambda host: [])
+
+    with pytest.raises(ValueError, match="host does not resolve"):
+        caldav_sync.validate_caldav_url("https://calendar.example.com/dav")
+
+
+@pytest.mark.parametrize(
+    "addrs",
+    [
+        ["93.184.216.34", "127.0.0.1"],  # public first, internal second
+        ["127.0.0.1", "93.184.216.34"],  # internal first, public second
+    ],
+)
+def test_validate_caldav_url_blocks_mixed_dns_in_any_order(monkeypatch, addrs):
+    # A host that resolves to BOTH a public and an internal address must be
+    # rejected regardless of record order — every resolved address is checked,
+    # so one internal answer is enough to block. Defends DNS round-robin and a
+    # rebind that slips an internal A-record alongside a public one.
+    monkeypatch.delenv("ODYSSEUS_ALLOW_PRIVATE_CALDAV", raising=False)
+    monkeypatch.setattr(
+        caldav_sync,
+        "_resolve_caldav_host_ips",
+        lambda host: [ipaddress.ip_address(a) for a in addrs],
+    )
+
+    with pytest.raises(ValueError, match="host is not allowed"):
+        caldav_sync.validate_caldav_url("https://calendar.example.com/dav")
+
+
 def test_sync_caldav_decrypts_stored_password_and_validates_url(monkeypatch):
     monkeypatch.setattr(
         caldav_sync,

tests/test_document_session_owner_scope.py

@@ -0,0 +1,143 @@
+"""Document session owner-scope regressions.
+
+Route handlers are called directly, matching the pattern used by the existing
+document route tests. This keeps coverage on the real closures without spinning
+up middleware.
+"""
+
+import tempfile
+import uuid
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import NullPool
+
+from tests.helpers.import_state import clear_fake_database_modules
+
+clear_fake_database_modules()
+
+import core.database as cdb
+import routes.document_routes as droutes
+from core.database import Document
+from core.database import Session as DbSession
+from routes.document_helpers import DocumentPatch
+
+_TMPDB = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+_ENGINE = create_engine(
+    f"sqlite:///{_TMPDB.name}",
+    connect_args={"check_same_thread": False},
+    poolclass=NullPool,
+)
+cdb.Base.metadata.create_all(_ENGINE)
+_TS = sessionmaker(bind=_ENGINE, autoflush=False, autocommit=False)
+
+
+def _req(user="alice"):
+    return SimpleNamespace(state=SimpleNamespace(current_user=user))
+
+
+def _endpoint(method, path):
+    router = droutes.setup_document_routes(MagicMock(), None)
+    for route in router.routes:
+        if getattr(route, "path", None) == path and method in getattr(route, "methods", set()):
+            return route.endpoint
+    raise RuntimeError(f"{method} {path} not found")
+
+
+def _bind_test_db():
+    previous = droutes.SessionLocal
+    droutes.SessionLocal = _TS
+    return previous
+
+
+def _seed():
+    alice_session = "alice-" + uuid.uuid4().hex[:8]
+    bob_session = "bob-" + uuid.uuid4().hex[:8]
+    alice_doc = str(uuid.uuid4())
+    bob_doc = str(uuid.uuid4())
+    legacy_doc = str(uuid.uuid4())
+    db = _TS()
+    try:
+        db.add(DbSession(id=alice_session, owner="alice", name="alice", model="m", endpoint_url="http://x"))
+        db.add(DbSession(id=bob_session, owner="bob", name="bob", model="m", endpoint_url="http://x"))
+        db.add(Document(
+            id=alice_doc,
+            session_id=alice_session,
+            title="alice doc",
+            language="markdown",
+            current_content="alice body",
+            version_count=1,
+            is_active=True,
+            owner="alice",
+        ))
+        db.add(Document(
+            id=bob_doc,
+            session_id=bob_session,
+            title="bob doc",
+            language="markdown",
+            current_content="bob body",
+            version_count=1,
+            is_active=True,
+            owner="bob",
+        ))
+        db.add(Document(
+            id=legacy_doc,
+            session_id=alice_session,
+            title="legacy doc",
+            language="markdown",
+            current_content="legacy body",
+            version_count=1,
+            is_active=True,
+            owner=None,
+        ))
+        db.commit()
+        return alice_session, bob_session, alice_doc, bob_doc, legacy_doc
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_patch_document_rejects_cross_owner_session_link():
+    previous_session_local = _bind_test_db()
+    try:
+        patch_document = _endpoint("PATCH", "/api/document/{doc_id}")
+        alice_session, bob_session, _alice_doc, bob_doc, _legacy_doc = _seed()
+
+        with pytest.raises(HTTPException) as exc:
+            await patch_document(_req("bob"), bob_doc, DocumentPatch(session_id=alice_session))
+
+        assert exc.value.status_code == 404
+        db = _TS()
+        try:
+            assert db.query(Document).filter(Document.id == bob_doc).first().session_id == bob_session
+        finally:
+            db.close()
+    finally:
+        droutes.SessionLocal = previous_session_local
+
+
+@pytest.mark.asyncio
+async def test_list_documents_filters_foreign_docs_in_visible_session():
+    previous_session_local = _bind_test_db()
+    try:
+        list_documents = _endpoint("GET", "/api/documents/{session_id}")
+        alice_session, _bob_session, alice_doc, bob_doc, legacy_doc = _seed()
+        db = _TS()
+        try:
+            db.query(Document).filter(Document.id == bob_doc).update({"session_id": alice_session})
+            db.commit()
+        finally:
+            db.close()
+
+        rows = await list_documents(_req("alice"), alice_session)
+        ids = {row["id"] for row in rows}
+
+        assert alice_doc in ids
+        assert legacy_doc in ids
+        assert bob_doc not in ids
+    finally:
+        droutes.SessionLocal = previous_session_local

tests/test_gallery_album_owner_scope.py

@@ -30,6 +30,13 @@ def test_patch_validates_target_album_ownership():
     assert "_get_or_404_album(db, req.album_id, user)" in body
 
 
+def test_upload_validates_target_album_ownership():
+    fns = _function_sources()
+    body = fns["gallery_upload"]
+    assert "album_id" in body
+    assert "_get_or_404_album(db, album_id, user)" in body
+
+
 def test_list_albums_count_and_cover_are_owner_scoped():
     fns = _function_sources()
     body = fns["list_albums"]
@@ -38,6 +45,14 @@ def test_list_albums_count_and_cover_are_owner_scoped():
     assert body.count("GalleryImage.owner == user") >= 2
 
 
+def test_delete_album_cleanup_is_owner_scoped():
+    fns = _function_sources()
+    body = fns["delete_album"]
+    assert "GalleryImage.album_id == album_id" in body
+    assert "GalleryImage.owner == user" in body
+    assert 'q.update({"album_id": None}' in body
+
+
 def test_get_or_404_album_enforces_owner():
     # Guard the precedent we rely on: the helper rejects another user's album.
     fns = _function_sources()

tests/test_gallery_endpoint_matching.py

@@ -1,34 +1,11 @@
-import ast
-from pathlib import Path
-
 def test_gallery_url_normalization_bug():
-    # Read and parse the actual source file
-    source_path = Path("routes/gallery_routes.py")
-    assert source_path.exists(), "gallery_routes.py could not be found"
-    
-    source = source_path.read_text(encoding="utf-8")
-    tree = ast.parse(source)
-    
-    # Locate the comparison node within harmonize_image that references ep.base_url and base
-    compare_node = None
-    for node in ast.walk(tree):
-        if isinstance(node, ast.Compare):
-            segment = ast.get_source_segment(source, node) or ""
-            if "ep.base_url" in segment and "base" in segment and "_norm_url" not in segment:
-                compare_node = node
-                break
-                
-    assert compare_node is not None, "Could not find the ep.base_url vs base comparison inside gallery_routes.py"
-    
-    # Compile the compare node into an expression
-    expr = ast.Expression(body=compare_node)
-    compiled_code = compile(expr, "<string>", "eval")
-    
+    from routes.gallery_routes import _normalize_image_endpoint_base
+
     def check_match(ep_url: str, base_url: str) -> bool:
-        class MockEP:
-            def __init__(self, url):
-                self.base_url = url
-        return eval(compiled_code, {}, {"ep": MockEP(ep_url), "base": base_url})
+        return (
+            _normalize_image_endpoint_base(ep_url)
+            == _normalize_image_endpoint_base(base_url)
+        )
 
     # Test cases that SHOULD NOT match under a correct implementation
     # (Buggy rstrip('/v1') logic incorrectly treats these as equal)

tests/test_gallery_image_endpoint_owner_scope.py

@@ -0,0 +1,126 @@
+"""Owner-scope regression for gallery image endpoint selection.
+
+The image editor/upscale proxies select ``ModelEndpoint`` rows and may copy the
+row's stored ``api_key`` for OpenAI-compatible image endpoints. That lookup must
+only consider endpoints visible to the caller, otherwise users sharing the same
+base URL can borrow another account's private image API key.
+"""
+
+from types import SimpleNamespace
+
+import routes.gallery_routes as gallery_routes
+
+
+class _Predicate:
+    def __init__(self, check):
+        self._check = check
+
+    def __call__(self, row):
+        return self._check(row)
+
+    def __or__(self, other):
+        return _Predicate(lambda row: self(row) or other(row))
+
+
+class _Column:
+    def __init__(self, name):
+        self.name = name
+
+    def __eq__(self, value):
+        return _Predicate(lambda row: getattr(row, self.name) == value)
+
+
+class _ModelEndpoint:
+    base_url = _Column("base_url")
+    model_type = _Column("model_type")
+    is_enabled = _Column("is_enabled")
+    owner = _Column("owner")
+
+
+class _Query:
+    def __init__(self, rows):
+        self._rows = list(rows)
+
+    def filter(self, *predicates):
+        self._rows = [row for row in self._rows if all(pred(row) for pred in predicates)]
+        return self
+
+    def all(self):
+        return list(self._rows)
+
+
+class _DB:
+    def __init__(self, rows):
+        self._rows = rows
+
+    def query(self, model):
+        assert model is _ModelEndpoint
+        return _Query(self._rows)
+
+
+def _ep(base_url, owner, *, enabled=True, model_type="image", api_key="sk-secret"):
+    return SimpleNamespace(
+        base_url=base_url,
+        owner=owner,
+        is_enabled=enabled,
+        model_type=model_type,
+        api_key=api_key,
+    )
+
+
+def _patch_model(monkeypatch):
+    monkeypatch.setattr(gallery_routes, "ModelEndpoint", _ModelEndpoint)
+
+
+URL = "https://api.example.com/v1"
+
+
+def test_first_visible_image_endpoint_rejects_another_owner(monkeypatch):
+    _patch_model(monkeypatch)
+    rows = [_ep(URL, "bob")]
+
+    assert gallery_routes._first_visible_image_endpoint(_DB(rows), "alice") is None
+
+
+def test_first_visible_image_endpoint_prefers_callers_own_row(monkeypatch):
+    _patch_model(monkeypatch)
+    rows = [_ep(URL, None, api_key="shared"), _ep(URL, "alice", api_key="own")]
+
+    ep = gallery_routes._first_visible_image_endpoint(_DB(rows), "alice")
+
+    assert ep is not None
+    assert ep.owner == "alice"
+    assert ep.api_key == "own"
+
+
+def test_visible_image_endpoint_for_base_rejects_same_url_other_owner(monkeypatch):
+    _patch_model(monkeypatch)
+    rows = [_ep(URL, "bob")]
+
+    assert gallery_routes._visible_image_endpoint_for_base(_DB(rows), URL, "alice") is None
+
+
+def test_visible_image_endpoint_for_base_allows_shared_or_own(monkeypatch):
+    _patch_model(monkeypatch)
+    rows = [
+        _ep("https://other.example/v1", "alice"),
+        _ep(URL, None, api_key="shared"),
+        _ep(URL, "alice", api_key="own"),
+    ]
+
+    ep = gallery_routes._visible_image_endpoint_for_base(_DB(rows), "https://api.example.com", "alice")
+
+    assert ep is not None
+    assert ep.owner == "alice"
+    assert ep.api_key == "own"
+    assert ep.base_url == URL
+
+
+def test_image_endpoint_owner_filter_is_noop_in_single_user_mode(monkeypatch):
+    _patch_model(monkeypatch)
+    rows = [_ep(URL, "bob")]
+
+    ep = gallery_routes._visible_image_endpoint_for_base(_DB(rows), URL, None)
+
+    assert ep is not None
+    assert ep.owner == "bob"

tests/test_imap_leak_fixes.py

@@ -0,0 +1,230 @@
+"""Regression tests for IMAP connection leak fixes.
+
+Each test forces an exception after _imap_connect() succeeds and asserts
+that conn.logout() is still called exactly once (guaranteed by try/finally).
+
+Functions covered:
+  - routes/email_helpers.py: _fetch_sender_thread_context, _pre_retrieve_context
+  - mcp_servers/email_server.py: _list_emails, _read_email, _reply_to_email,
+    _download_attachment
+"""
+
+import os
+import sys
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+_TMP = Path(tempfile.mkdtemp(prefix="odysseus-imap-leak-fixes-"))
+os.environ.setdefault("DATA_DIR", str(_TMP))
+os.environ.setdefault("DATABASE_URL", f"sqlite:///{_TMP / 'app.db'}")
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+
+def _make_failing_conn(captured, *, raises_on="select"):
+    """Return a mock IMAP connection that raises on the first call to `raises_on`."""
+    conn = MagicMock()
+    conn.logout = MagicMock(side_effect=lambda: captured.__setitem__(
+        "logout_calls", captured.get("logout_calls", 0) + 1
+    ))
+
+    def _raise(*a, **kw):
+        raise RuntimeError("simulated IMAP failure")
+
+    getattr(conn, raises_on).side_effect = _raise
+    return conn
+
+
+# ── email_helpers ──────────────────────────────────────────────────────────────
+
+def test_fetch_sender_thread_context_logs_out_on_select_failure(monkeypatch):
+    import routes.email_helpers as helpers
+
+    captured = {}
+    conn = _make_failing_conn(captured, raises_on="select")
+    monkeypatch.setattr(helpers, "_imap_connect", lambda *a, **kw: conn)
+
+    result = helpers._fetch_sender_thread_context("user@example.com")
+
+    assert captured.get("logout_calls", 0) == 1, (
+        f"conn.logout() must be called on select failure. "
+        f"Got logout_calls={captured.get('logout_calls')}"
+    )
+    assert result == "", "Should return empty string on failure"
+
+
+def test_fetch_sender_thread_context_logs_out_on_connect_failure(monkeypatch):
+    """If _imap_connect itself raises, conn is None — no logout, no crash."""
+    import routes.email_helpers as helpers
+
+    def _fail(*a, **kw):
+        raise ConnectionRefusedError("cannot connect")
+
+    monkeypatch.setattr(helpers, "_imap_connect", _fail)
+    result = helpers._fetch_sender_thread_context("user@example.com")
+    assert result == "", "Should return empty string when connect fails"
+
+
+def test_pre_retrieve_context_logs_out_on_search_failure(monkeypatch):
+    import routes.email_helpers as helpers
+
+    captured = {}
+    conn = MagicMock()
+    conn.select.return_value = ("OK", [])
+    conn.logout = MagicMock(side_effect=lambda: captured.__setitem__(
+        "logout_calls", captured.get("logout_calls", 0) + 1
+    ))
+    conn.search.side_effect = RuntimeError("simulated search failure")
+
+    monkeypatch.setattr(helpers, "_imap_connect", lambda *a, **kw: conn)
+
+    # Bypass the known-sender check and term extraction so we reach the IMAP block
+    monkeypatch.setattr(helpers, "_imap", MagicMock(
+        return_value=MagicMock(
+            __enter__=MagicMock(return_value=MagicMock(
+                select=MagicMock(return_value=("OK", [])),
+                search=MagicMock(return_value=("OK", [b"1"])),
+            )),
+            __exit__=MagicMock(return_value=False),
+        )
+    ))
+
+    # Provide a body with a capitalised term so terms_list is non-empty
+    snippets, terms = helpers._pre_retrieve_context(
+        body="Project Alpha update",
+        sender="Known Sender <known@example.com>",
+    )
+
+    # The function is best-effort and never raises; logout must have been called
+    assert captured.get("logout_calls", 0) == 1, (
+        f"ctx_conn.logout() must be called even when search raises. "
+        f"Got logout_calls={captured.get('logout_calls')}"
+    )
+
+
+# ── email_server ───────────────────────────────────────────────────────────────
+
+def test_mcp_list_emails_logs_out_on_select_failure(monkeypatch):
+    import mcp_servers.email_server as srv
+
+    captured = {}
+    conn = _make_failing_conn(captured, raises_on="select")
+    monkeypatch.setattr(srv, "_imap_connect", lambda *a, **kw: conn)
+
+    try:
+        srv._list_emails()
+    except Exception:
+        pass
+
+    assert captured.get("logout_calls", 0) == 1, (
+        f"conn.logout() must be called after select raises. "
+        f"Got logout_calls={captured.get('logout_calls')}"
+    )
+
+
+def test_mcp_list_emails_logs_out_on_search_failure(monkeypatch):
+    import mcp_servers.email_server as srv
+
+    captured = {}
+    conn = MagicMock()
+    conn.select.return_value = ("OK", [])
+    conn.uid.side_effect = RuntimeError("simulated search failure")
+    conn.logout = MagicMock(side_effect=lambda: captured.__setitem__(
+        "logout_calls", captured.get("logout_calls", 0) + 1
+    ))
+    monkeypatch.setattr(srv, "_imap_connect", lambda *a, **kw: conn)
+
+    try:
+        srv._list_emails()
+    except Exception:
+        pass
+
+    assert captured.get("logout_calls", 0) == 1, (
+        f"conn.logout() must be called after uid search raises. "
+        f"Got logout_calls={captured.get('logout_calls')}"
+    )
+
+
+def test_mcp_read_email_logs_out_on_select_failure(monkeypatch):
+    import mcp_servers.email_server as srv
+
+    captured = {}
+    conn = _make_failing_conn(captured, raises_on="select")
+    monkeypatch.setattr(srv, "_imap_connect", lambda *a, **kw: conn)
+    monkeypatch.setattr(srv, "_load_config", lambda *a, **kw: {})
+
+    # The exception propagates out of _read_email (no outer catch in this fn);
+    # what matters is that logout was still called via finally before it did.
+    try:
+        srv._read_email(uid="1")
+    except RuntimeError:
+        pass
+
+    assert captured.get("logout_calls", 0) == 1, (
+        f"conn.logout() must be called after select raises. "
+        f"Got logout_calls={captured.get('logout_calls')}"
+    )
+
+
+def test_mcp_read_email_logs_out_on_fetch_failure(monkeypatch):
+    import mcp_servers.email_server as srv
+
+    captured = {}
+    conn = MagicMock()
+    conn.select.return_value = ("OK", [])
+    conn.uid.side_effect = RuntimeError("simulated fetch failure")
+    conn.logout = MagicMock(side_effect=lambda: captured.__setitem__(
+        "logout_calls", captured.get("logout_calls", 0) + 1
+    ))
+    monkeypatch.setattr(srv, "_imap_connect", lambda *a, **kw: conn)
+    monkeypatch.setattr(srv, "_load_config", lambda *a, **kw: {})
+
+    try:
+        srv._read_email(uid="1")
+    except RuntimeError:
+        pass
+
+    assert captured.get("logout_calls", 0) == 1, (
+        f"conn.logout() must be called after uid fetch raises. "
+        f"Got logout_calls={captured.get('logout_calls')}"
+    )
+
+
+def test_mcp_reply_to_email_logs_out_on_select_failure(monkeypatch):
+    import mcp_servers.email_server as srv
+
+    captured = {}
+    conn = _make_failing_conn(captured, raises_on="select")
+    monkeypatch.setattr(srv, "_imap_connect", lambda *a, **kw: conn)
+
+    # Exception propagates; the finally still runs before it does.
+    try:
+        srv._reply_to_email(uid="1", body="hi")
+    except RuntimeError:
+        pass
+
+    assert captured.get("logout_calls", 0) == 1, (
+        f"conn.logout() must be called after select raises in _reply_to_email. "
+        f"Got logout_calls={captured.get('logout_calls')}"
+    )
+
+
+def test_mcp_download_attachment_logs_out_on_select_failure(monkeypatch):
+    import mcp_servers.email_server as srv
+
+    captured = {}
+    conn = _make_failing_conn(captured, raises_on="select")
+    monkeypatch.setattr(srv, "_imap_connect", lambda *a, **kw: conn)
+
+    try:
+        srv._download_attachment(uid="1", index=0)
+    except RuntimeError:
+        pass
+
+    assert captured.get("logout_calls", 0) == 1, (
+        f"conn.logout() must be called after select raises in _download_attachment. "
+        f"Got logout_calls={captured.get('logout_calls')}"
+    )

tests/test_manage_notes_owner_gate.py

@@ -0,0 +1,120 @@
+import asyncio
+import json
+import sys
+import types
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from src import tool_implementations
+
+
+class _Query:
+    def __init__(self, note):
+        self.note = note
+
+    def filter(self, *args, **kwargs):
+        return self
+
+    def first(self):
+        return self.note
+
+
+class _Db:
+    def __init__(self, note):
+        self.note = note
+        self.deleted = []
+        self.commits = 0
+
+    def query(self, *args, **kwargs):
+        return _Query(self.note)
+
+    def delete(self, note):
+        self.deleted.append(note)
+
+    def commit(self):
+        self.commits += 1
+
+    def rollback(self):
+        pass
+
+    def close(self):
+        pass
+
+
+def _install_fakes(monkeypatch, note):
+    fake_sa_attrs = types.ModuleType("sqlalchemy.orm.attributes")
+    fake_sa_attrs.flag_modified = lambda *args, **kwargs: None
+    monkeypatch.setitem(sys.modules, "sqlalchemy.orm.attributes", fake_sa_attrs)
+
+    db = _Db(note)
+    fake_core_db = types.ModuleType("core.database")
+    fake_core_db.SessionLocal = lambda: db
+    fake_core_db.Note = MagicMock()
+    monkeypatch.setitem(sys.modules, "core.database", fake_core_db)
+    return db
+
+
+def _run(args, owner="alice"):
+    return asyncio.run(tool_implementations.do_manage_notes(json.dumps(args), owner=owner))
+
+
+def _note(owner=None, **overrides):
+    data = {
+        "id": "abc12345-existing",
+        "owner": owner,
+        "title": "Original",
+        "content": "",
+        "note_type": "note",
+        "color": None,
+        "label": None,
+        "items": '[{"text":"item","done":false}]',
+        "pinned": False,
+        "archived": False,
+        "due_date": None,
+    }
+    data.update(overrides)
+    return SimpleNamespace(**data)
+
+
+def test_update_rejects_legacy_null_owner_for_authenticated_owner(monkeypatch):
+    note = _note(owner=None)
+    db = _install_fakes(monkeypatch, note)
+
+    result = _run({"action": "update", "id": "abc12345", "title": "Changed"})
+
+    assert result == {"error": "Note not found", "exit_code": 1}
+    assert note.title == "Original"
+    assert db.commits == 0
+
+
+def test_delete_rejects_legacy_empty_owner_for_authenticated_owner(monkeypatch):
+    note = _note(owner="")
+    db = _install_fakes(monkeypatch, note)
+
+    result = _run({"action": "delete", "id": "abc12345"})
+
+    assert result == {"error": "Note not found", "exit_code": 1}
+    assert db.deleted == []
+    assert db.commits == 0
+
+
+def test_toggle_rejects_other_owner(monkeypatch):
+    note = _note(owner="bob")
+    db = _install_fakes(monkeypatch, note)
+
+    result = _run({"action": "toggle_item", "id": "abc12345", "index": 0})
+
+    assert result == {"error": "Note not found", "exit_code": 1}
+    assert json.loads(note.items)[0]["done"] is False
+    assert db.commits == 0
+
+
+def test_update_allows_matching_owner(monkeypatch):
+    note = _note(owner="alice")
+    db = _install_fakes(monkeypatch, note)
+
+    result = _run({"action": "update", "id": "abc12345", "title": "Changed"})
+
+    assert result["exit_code"] == 0
+    assert note.title == "Changed"
+    assert db.commits == 1

tests/test_model_helper_owner_scope.py

@@ -0,0 +1,45 @@
+"""Model-assisted route helpers must resolve endpoints with owner scope."""
+
+import ast
+from pathlib import Path
+
+
+def _function_source(path: str, name: str) -> str:
+    source = Path(path).read_text(encoding="utf-8")
+    tree = ast.parse(source)
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == name:
+            return ast.get_source_segment(source, node) or ""
+    raise AssertionError(f"{name} not found in {path}")
+
+
+def test_document_ai_tidy_resolves_with_owner_scope():
+    body = _function_source("routes/document_routes.py", "ai_tidy_documents")
+    assert "resolve_task_endpoint(owner=user or None)" in body
+    assert 'resolve_endpoint("default", owner=user or None)' in body
+
+
+def test_calendar_quick_parse_resolves_with_owner_scope():
+    body = _function_source("routes/calendar_routes.py", "quick_parse")
+    assert "owner = _require_user(request)" in body
+    assert 'resolve_endpoint("utility", owner=owner or None)' in body
+    assert 'resolve_endpoint("default", owner=owner or None)' in body
+
+
+def test_task_parse_resolves_with_owner_scope():
+    body = _function_source("routes/task_routes.py", "parse_task")
+    assert "user = _owner(request)" in body
+    assert 'resolve_endpoint("utility", owner=user or None)' in body
+    assert 'resolve_endpoint("default", owner=user or None)' in body
+
+
+def test_history_compact_resolves_with_owner_scope():
+    body = _function_source("routes/history_routes.py", "compact_session")
+    assert "owner = effective_user(request)" in body
+    assert 'resolve_endpoint("utility", owner=owner or None)' in body
+
+
+def test_note_reminder_synthesis_resolves_with_owner_scope():
+    body = _function_source("routes/note_routes.py", "dispatch_reminder")
+    assert 'resolve_endpoint("utility", owner=owner or None)' in body
+    assert 'resolve_endpoint("default", owner=owner or None)' in body

tests/test_security_headers_middleware.py

@@ -0,0 +1,67 @@
+# tests/test_security_headers_middleware.py
+"""
+Focused regression coverage for `SecurityHeadersMiddleware`
+(core/middleware.py), added alongside the HSTS + Permissions-Policy
+hardening:
+
+  1. HSTS is emitted only for HTTPS requests, including those reaching
+     the app over a reverse proxy (`X-Forwarded-Proto: https`).
+  2. HSTS is absent on plain HTTP so local/dev deployments are unaffected.
+  3. `Permissions-Policy` locks down camera/geolocation but preserves
+     same-origin microphone access (`microphone=(self)`), so the app's
+     own voice/STT flow (`getUserMedia({ audio: true })`) keeps working.
+"""
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from core.middleware import SecurityHeadersMiddleware
+
+
+def _build_app():
+    app = FastAPI()
+    app.add_middleware(SecurityHeadersMiddleware)
+
+    @app.get("/")
+    def root():
+        return {"ok": True}
+
+    return app
+
+
+def _client(base_url="http://testserver"):
+    return TestClient(_build_app(), base_url=base_url)
+
+
+def test_hsts_absent_on_plain_http():
+    response = _client().get("/")
+
+    assert "strict-transport-security" not in response.headers
+
+
+def test_hsts_present_for_direct_https_requests():
+    response = _client(base_url="https://testserver").get("/")
+
+    assert response.headers["strict-transport-security"] == (
+        "max-age=31536000; includeSubDomains"
+    )
+
+
+def test_hsts_present_via_x_forwarded_proto_https():
+    response = _client().get("/", headers={"X-Forwarded-Proto": "https"})
+
+    assert response.headers["strict-transport-security"] == (
+        "max-age=31536000; includeSubDomains"
+    )
+
+
+def test_permissions_policy_locks_camera_and_geolocation_but_allows_self_microphone():
+    response = _client().get("/")
+
+    policy = response.headers["permissions-policy"]
+    assert policy == "camera=(), microphone=(self), geolocation=()"
+
+    # Explicitly pin the contract the reviewer flagged: an empty allowlist
+    # would also block the app's own same-origin voice/STT button.
+    assert "microphone=()" not in policy
+    assert "microphone=(self)" in policy

tests/test_task_chain_owner_scope.py

@@ -0,0 +1,127 @@
+"""Task chaining must not cross owner boundaries."""
+
+import tempfile
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import NullPool
+
+from tests.helpers.import_state import clear_fake_database_modules
+
+clear_fake_database_modules()
+
+import core.database as cdb
+import routes.task_routes as task_routes
+from core.database import ScheduledTask
+
+_TMPDB = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+_ENGINE = create_engine(
+    f"sqlite:///{_TMPDB.name}",
+    connect_args={"check_same_thread": False},
+    poolclass=NullPool,
+)
+cdb.Base.metadata.create_all(_ENGINE)
+_TS = sessionmaker(bind=_ENGINE, autoflush=False, autocommit=False)
+task_routes.SessionLocal = _TS
+
+
+def _req(user="alice"):
+    return SimpleNamespace(state=SimpleNamespace(current_user=user))
+
+
+def _endpoint(method, path):
+    task_routes.SessionLocal = _TS
+    router = task_routes.setup_task_routes(MagicMock())
+    for route in router.routes:
+        if getattr(route, "path", None) == path and method in getattr(route, "methods", set()):
+            return route.endpoint
+    raise RuntimeError(f"{method} {path} not found")
+
+
+def _seed_task(task_id, owner, *, then_task_id=None):
+    db = _TS()
+    try:
+        task = ScheduledTask(
+            id=task_id,
+            owner=owner,
+            name=task_id,
+            prompt="do work",
+            task_type="llm",
+            trigger_type="webhook",
+            status="active",
+            output_target="session",
+            then_task_id=then_task_id,
+        )
+        db.add(task)
+        db.commit()
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_create_task_rejects_cross_owner_chain_target():
+    _seed_task("bob-target-create", "bob")
+    create_task = _endpoint("POST", "/api/tasks")
+
+    req = task_routes.TaskCreate(
+        prompt="alice source",
+        trigger_type="webhook",
+        then_task_id="bob-target-create",
+    )
+    with pytest.raises(HTTPException) as exc:
+        await create_task(_req("alice"), req)
+
+    assert exc.value.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_update_task_rejects_cross_owner_chain_target():
+    _seed_task("alice-source-update", "alice")
+    _seed_task("bob-target-update", "bob")
+    update_task = _endpoint("PUT", "/api/tasks/{task_id}")
+
+    with pytest.raises(HTTPException) as exc:
+        await update_task(
+            _req("alice"),
+            "alice-source-update",
+            task_routes.TaskUpdate(then_task_id="bob-target-update"),
+        )
+
+    assert exc.value.status_code == 404
+    db = _TS()
+    try:
+        source = db.query(ScheduledTask).filter(ScheduledTask.id == "alice-source-update").first()
+        assert source.then_task_id is None
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+async def test_update_task_allows_same_owner_chain_target():
+    _seed_task("alice-source-allow", "alice")
+    _seed_task("alice-target-allow", "alice")
+    update_task = _endpoint("PUT", "/api/tasks/{task_id}")
+
+    out = await update_task(
+        _req("alice"),
+        "alice-source-allow",
+        task_routes.TaskUpdate(then_task_id="alice-target-allow"),
+    )
+
+    assert out["then_task_id"] == "alice-target-allow"
+
+
+def test_scheduler_cycle_guard_treats_cross_owner_chain_as_unsafe():
+    _seed_task("bob-target-cycle", "bob")
+    from src.task_scheduler import TaskScheduler
+
+    scheduler = TaskScheduler.__new__(TaskScheduler)
+    db = _TS()
+    try:
+        assert scheduler._has_chain_cycle(db, "bob-target-cycle", owner="alice") is True
+    finally:
+        db.close()

tests/test_vision_owner_scope.py

@@ -0,0 +1,101 @@
+from pathlib import Path
+
+from src import ai_interaction
+from src import document_processor as dp
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_configured_vision_model_resolution_passes_owner(monkeypatch):
+    seen = []
+
+    def fake_resolve_model(spec, owner=None):
+        seen.append((spec, owner))
+        return ("http://example.test/chat/completions", spec, {"Authorization": "Bearer token"})
+
+    monkeypatch.setattr(ai_interaction, "_resolve_model", fake_resolve_model)
+
+    assert dp._resolve_vl_model("gpt-4o", owner="alice") == (
+        "http://example.test/chat/completions",
+        "gpt-4o",
+        {"Authorization": "Bearer token"},
+    )
+    assert seen == [("gpt-4o", "alice")]
+
+
+def test_auto_detected_vision_model_resolution_passes_owner(monkeypatch):
+    seen = []
+
+    def fake_resolve_model(spec, owner=None):
+        seen.append((spec, owner))
+        if spec == "llava":
+            return ("http://example.test/chat/completions", spec, {})
+        raise ValueError("not available")
+
+    monkeypatch.setattr(ai_interaction, "_resolve_model", fake_resolve_model)
+
+    assert dp._resolve_vl_model("", owner="alice") == (
+        "http://example.test/chat/completions",
+        "llava",
+        {},
+    )
+    assert seen
+    assert all(owner == "alice" for _spec, owner in seen)
+
+
+def test_vision_analysis_uses_owner_scoped_primary_and_fallback(monkeypatch, tmp_path):
+    seen = {}
+
+    def fake_resolve_vl_model(configured, owner=None):
+        seen["primary"] = (configured, owner)
+        return ("http://primary.test/chat/completions", "vision-primary", {"X-Test": "1"})
+
+    def fake_fallbacks(owner=None):
+        seen["fallback_owner"] = owner
+        return []
+
+    def fake_llm_call(url, model, messages, headers=None, timeout=None):
+        seen["llm"] = (url, model, headers, timeout, messages)
+        return "description"
+
+    monkeypatch.setattr(dp, "_load_vl_settings", lambda: {"vision_enabled": True, "vision_model": "gpt-4o"})
+    monkeypatch.setattr(dp, "_resolve_vl_model", fake_resolve_vl_model)
+    monkeypatch.setattr(dp, "llm_call", fake_llm_call)
+
+    from src import endpoint_resolver
+
+    monkeypatch.setattr(endpoint_resolver, "resolve_vision_fallback_candidates", fake_fallbacks)
+
+    image = tmp_path / "image.png"
+    image.write_bytes(b"not-a-real-png-but-base64-is-enough")
+
+    assert dp.analyze_image_with_vl_result(str(image), owner="alice") == {
+        "text": "description",
+        "model": "vision-primary",
+    }
+    assert seen["primary"] == ("gpt-4o", "alice")
+    assert seen["fallback_owner"] == "alice"
+    assert seen["llm"][:4] == (
+        "http://primary.test/chat/completions",
+        "vision-primary",
+        {"X-Test": "1"},
+        120,
+    )
+
+
+def test_request_vision_call_sites_pass_owner():
+    chat_source = (ROOT / "src" / "chat_handler.py").read_text()
+    processor_source = (ROOT / "src" / "document_processor.py").read_text()
+    upload_source = (ROOT / "routes" / "upload_routes.py").read_text()
+    document_source = (ROOT / "routes" / "document_routes.py").read_text()
+    gallery_source = (ROOT / "routes" / "gallery_routes.py").read_text()
+    memory_source = (ROOT / "routes" / "memory_routes.py").read_text()
+
+    assert 'analyze_image_with_vl_result(file_info["path"], owner=owner)' in chat_source
+    assert "analyze_image_with_vl(path, owner=current_user)" in upload_source
+    assert "_process_pdf(path, owner=owner)" in processor_source
+    assert "_process_pdf(pdf_path, owner=user)" in document_source
+    assert "_resolve_vl_model(vl_model, owner=user)" in document_source
+    assert "_resolve_vl_model(configured, owner=user)" in gallery_source
+    assert "_process_pdf(tmp_path, owner=_owner(request))" in memory_source

tests/test_webhook_sanitize_error_ipv6.py

@@ -0,0 +1,98 @@
+"""sanitize_error must scrub IPv6 addresses, not just IPv4.
+
+Webhook delivery errors are stored in Webhook.last_error and surfaced in the
+UI. The scrubber removed IPv4 literals but let IPv6 addresses through, so a
+failed delivery to an internal v6 host (::1, fe80::/fc00:: ...) leaked the
+address. This pins the v6 redaction while keeping the false-positive guards
+(clock times, MACs, C++ "::") that make the pattern safe on arbitrary text.
+"""
+
+import os
+import sys
+from unittest.mock import patch
+
+from tests.helpers.import_state import clear_module, preserve_import_state
+
+# Same import dance as test_webhook_ssrf_resilience.py: webhook_manager pulls in
+# core.database (init_db -> create_all), which needs a DB path at import time.
+# Pin DATABASE_URL to in-memory SQLite and restore module state afterwards.
+# sanitize_error itself is pure (stdlib re only).
+with patch.dict(os.environ, {"DATABASE_URL": "sqlite:///:memory:"}), \
+        preserve_import_state("src.database", "core.database"):
+    clear_module("src.database")
+    _core_database = sys.modules.get("core.database")
+    if _core_database is not None and not getattr(_core_database, "__file__", None):
+        del sys.modules["core.database"]
+    from src.webhook_manager import sanitize_error
+
+
+def test_ipv6_addresses_are_redacted():
+    leaky = [
+        "connect to [fd00::1234:5678]:8080 failed",   # bracketed + port
+        "ConnectError to fe80::1 refused",            # link-local
+        "no route to ::1",                            # loopback
+        "host fc00::abcd unreachable",                # unique-local
+        "connect to [::1]:443 refused",               # bracketed + port
+        "POST https://[2001:db8::1]:443/hook failed",  # inside a URL
+        "addr 2001:0db8:0000:0000:0000:ff00:0042:8329",  # full 8-group
+    ]
+    for msg in leaky:
+        out = sanitize_error(msg)
+        # Scrubbed via the v6 rule ([redacted]) or, inside a URL, the URL rule
+        # ([redacted-url]) — either way the address must not survive.
+        assert "[redacted" in out, out
+        assert "::" not in out and "[fd00" not in out, out
+
+
+def test_non_addresses_are_preserved():
+    # Colon-bearing strings that are NOT IPv6 must pass through untouched, so
+    # error messages stay readable.
+    safe = [
+        "failed at 12:34:56 today",                 # clock time
+        "2026-06-05T22:36:55 connection reset",     # ISO timestamp
+        "std::vector<int> overflow",                # C++ scope resolution
+        "device ab:cd:ef:01:23:45 offline",         # MAC address
+        "unsupported ratio 16:9",
+        "HTTP 500 from upstream",
+        "request [deadbeef] failed",                # bracketed hex id, no colon
+    ]
+    for msg in safe:
+        assert sanitize_error(msg) == msg, msg
+
+
+def test_ipv4_still_redacted_and_length_capped():
+    assert sanitize_error("dial 192.168.1.5:9000 refused") == "dial [redacted] refused"
+    assert len(sanitize_error("x" * 500)) == 200
+
+
+def test_ipv6_zone_id_is_redacted():
+    # Link-local addresses often carry a %zone (fe80::1%eth0). The whole token,
+    # zone included, must go — ipaddress validates the address part.
+    out = sanitize_error("bind fe80::1%eth0 unreachable")
+    assert "[redacted]" in out
+    assert "::" not in out and "%eth0" not in out and "fe80" not in out
+
+
+def test_ipv4_mapped_ipv6_is_scrubbed():
+    # ::ffff:192.168.0.1 must be redacted as a single unit (one [redacted]), not
+    # split into "[redacted][redacted]" by the v6 and v4 passes.
+    assert sanitize_error("to ::ffff:192.168.0.1 closed") == "to [redacted] closed"
+
+
+def test_bracketed_scoped_ipv6_with_port_is_one_redaction():
+    # [fe80::1%eth0]:8080 — the whole bracketed authority (zone + port) goes,
+    # with no leftover brackets/port and no nested [redacted].
+    assert sanitize_error("dial [fe80::1%eth0]:8080 timeout") == "dial [redacted] timeout"
+
+
+def test_bracketed_ipv4_mapped_with_port_is_one_redaction():
+    # [::ffff:192.168.0.1]:8080 — same, for an IPv4-mapped literal in brackets.
+    assert sanitize_error("dial [::ffff:192.168.0.1]:8080 timeout") == "dial [redacted] timeout"
+
+
+def test_invalid_ipv6_is_not_partially_mangled():
+    # Nine groups is not a valid address. Backing the scrub with ipaddress means
+    # the whole token is preserved, instead of a hand-rolled 8-group regex
+    # chewing off "1:2:3:4:5:6:7:8" and leaving a dangling ":9".
+    msg = "weird id 1:2:3:4:5:6:7:8:9 here"
+    assert sanitize_error(msg) == msg