fix(skills): require manage_skills action (#5856)

* fix(integrations): pin api_call to the SSRF-validated IP

execute_api_call runs check_outbound_url on the target, but that guard only
resolves the host to answer (ok, reason) and hands back no address. The request
right after it opened a plain httpx.AsyncClient, which resolves the host again at
connect time. A base_url host on a low TTL can pass the guard as a public IP and
then flip to 169.254.169.254 for the connect, so the call lands on cloud metadata
with the integration's stored auth headers attached.

Resolve once, remember the IPs the guard actually validated, and pin the client's
socket to that set through a small AnyIO-backed transport. SNI and the Host header
still come from the URL, so TLS and vhost routing are unchanged; connect-time
fallback stays inside the approved address set over one shared deadline. This is
the same pinning the webhook sender and web-fetch paths already do -- api_call was
the last outbound path that skipped it.

Fixes #5513

* fix(integrations): de-duplicate the pinned IP list

_default_resolver calls getaddrinfo(host, None) with no socktype filter, so
glibc returns one record per socktype and a single-homed host comes back three
times over. _validated_ips kept every entry, so the transport pinned the same
address repeatedly and the connect fallback could spend its shared deadline
retrying one dead address instead of moving on to a genuinely different one.

Windows getaddrinfo collapses those duplicate records, which is why the
ip-literal pin test only failed on CI and not locally.

app.py

@@ -739,7 +739,7 @@ app.include_router(setup_stt_routes(stt_service))
 logger.info("STT service initialized (provider managed via settings)")
 
 # Documents (artifacts/canvas)
-from routes.document_routes import setup_document_routes
+from routes.document.document_routes import setup_document_routes
 document_router = setup_document_routes(session_manager, upload_handler)
 app.include_router(document_router)
 

routes/document/__init__.py

@@ -0,0 +1,6 @@
+"""Document route domain package (slice 2m, #4082/#4071).
+
+Contains document_routes.py and document_helpers.py, migrated from the flat
+routes/ directory. Backward-compat shims at routes/document_routes.py and
+routes/document_helpers.py re-export from here.
+"""

routes/document/document_helpers.py

@@ -0,0 +1,243 @@
+"""document_helpers.py — Pydantic models, doc serializers, owner gating, file-locator helpers shared with document_routes.py."""
+
+"""Document routes — CRUD for living documents with version history."""
+
+import logging
+import os
+import re
+from typing import Any, Dict, Optional
+
+from fastapi import HTTPException, Request
+from pydantic import BaseModel
+
+from core.database import Document, DocumentVersion
+from core.database import Session as DbSession
+from src.auth_helpers import _auth_disabled
+from src.upload_handler import UploadHandler
+
+logger = logging.getLogger(__name__)
+
+
+# ---- Request schemas ----
+
+class DocumentCreate(BaseModel):
+    session_id: Optional[str] = None
+    title: str = "Untitled"
+    language: Optional[str] = None
+    content: str = ""
+
+class DocumentUpdate(BaseModel):
+    content: str
+    summary: Optional[str] = None
+    force_version: bool = False
+
+class DocumentPatch(BaseModel):
+    title: Optional[str] = None
+    language: Optional[str] = None
+    session_id: Optional[str] = None  # link/unlink document to a session
+
+
+# ---- Helpers ----
+
+def _doc_to_dict(doc: Document) -> Dict[str, Any]:
+    return {
+        "id": doc.id,
+        "session_id": doc.session_id,
+        "title": doc.title,
+        "language": doc.language,
+        "current_content": doc.current_content,
+        "version_count": doc.version_count,
+        "is_active": doc.is_active,
+        "archived": bool(getattr(doc, "archived", False)),
+        "created_at": (doc.created_at.isoformat() + "Z") if doc.created_at else None,
+        "updated_at": (doc.updated_at.isoformat() + "Z") if doc.updated_at else None,
+        # Source-email provenance (set when doc was created from an email
+        # attachment) — drives the "Send signed reply" menu item.
+        "source_email_uid":        getattr(doc, "source_email_uid", None),
+        "source_email_folder":     getattr(doc, "source_email_folder", None),
+        "source_email_account_id": getattr(doc, "source_email_account_id", None),
+        "source_email_message_id": getattr(doc, "source_email_message_id", None),
+    }
+
+def _version_to_dict(v: DocumentVersion) -> Dict[str, Any]:
+    return {
+        "id": v.id,
+        "document_id": v.document_id,
+        "version_number": v.version_number,
+        "content": v.content,
+        "summary": v.summary,
+        "source": v.source,
+        "created_at": v.created_at.isoformat() if v.created_at else None,
+    }
+
+
+def _verify_doc_owner(db, doc: Document, user: str):
+    """Verify `user` owns this document. Raise 404 if not.
+
+    Documents now carry their own `owner` column, so a doc whose session
+    was deleted (session_id → NULL) can still prove ownership and stay
+    openable / cloneable. We trust that column first and only fall back to
+    the session join for any not-yet-backfilled legacy row.
+    """
+    if user is None:
+        if _auth_disabled():
+            return  # Single-user / no-auth mode: allow access
+        raise HTTPException(403, "Authentication required")
+    if doc.owner is not None:
+        if doc.owner != user:
+            raise HTTPException(404, "Document not found")
+        return
+    # Legacy fallback: derive ownership from the linked session.
+    if not doc.session_id:
+        raise HTTPException(404, "Document not found")
+    session = db.query(DbSession).filter(DbSession.id == doc.session_id).first()
+    if not session or session.owner != user:
+        raise HTTPException(404, "Document not found")
+
+
+def _owner_session_filter(q, user):
+    """Restrict a documents query to those owned by `user`.
+
+    Documents now carry their own `owner` column (backfilled at boot from
+    the linked session, or assigned to the admin user for legacy/orphaned
+    docs). We filter on that directly rather than on a session join, so a
+    document whose session was deleted (session_id → NULL) still shows up
+    for its owner instead of silently vanishing from the Library + search.
+
+    The owner backfill runs in init_db before the app serves requests, so
+    by the time this filter is live there are no NULL-owner rows to leak;
+    we therefore match the owner strictly for authenticated callers."""
+    if not user:
+        if user == "" or _auth_disabled():
+            return q
+        return q.filter(False)
+    return q.filter(Document.owner == user)
+
+
+
+def _slug(name: str) -> str:
+    """Filesystem-friendly version of a document title.
+
+    Whitespace becomes underscores; other unsafe punctuation is dropped.
+    Preserves letters, digits, dot, hyphen, underscore. Idempotent.
+    """
+    import re as _re
+    s = (name or "").strip()
+    # Drop the trailing extension if the title happens to include one
+    s = _re.sub(r'\.pdf$', '', s, flags=_re.IGNORECASE)
+    s = _re.sub(r'\s+', '_', s)
+    s = _re.sub(r'[^A-Za-z0-9._-]', '', s)
+    s = _re.sub(r'_+', '_', s).strip('_')
+    return s or "form"
+
+
+# DPI scale for the interactive PDF view. ~150 DPI (2x of 72 PDF user-units).
+_PDF_RENDER_SCALE = 2.0
+
+
+def _upload_path_inside(upload_dir: str, path: str) -> bool:
+    base = os.path.realpath(upload_dir)
+    p = os.path.realpath(path)
+    try:
+        return os.path.commonpath([base, p]) == base
+    except Exception:
+        return False
+
+
+def _resolve_user_upload_path(
+    upload_handler: Any,
+    upload_id: str,
+    owner: Optional[str],
+    auth_manager=None,
+) -> Optional[str]:
+    """Resolve an upload id to a filesystem path the caller may read."""
+    if upload_handler is None:
+        return None
+    resolved = upload_handler.resolve_upload(
+        upload_id,
+        owner=owner,
+        auth_manager=auth_manager,
+    )
+    if not isinstance(resolved, dict) or not resolved:
+        return None
+    path = resolved.get("path")
+    upload_dir = getattr(upload_handler, "upload_dir", None)
+    if path and upload_dir and not _upload_path_inside(upload_dir, path):
+        logger.warning("Upload path outside upload directory: %s", path)
+        return None
+    return path
+
+
+def _locate_upload(
+    upload_dir: str,
+    file_id: str,
+    owner: Optional[str] = None,
+    auth_manager=None,
+    upload_handler: Any = None,
+):
+    """Find an upload by its filename ID via UploadHandler.resolve_upload."""
+    if upload_handler is None:
+        from src.upload_handler import UploadHandler
+
+        base_dir = os.path.dirname(os.path.abspath(upload_dir))
+        upload_handler = UploadHandler(base_dir, upload_dir)
+    return _resolve_user_upload_path(upload_handler, file_id, owner, auth_manager)
+
+
+def _assert_pdf_marker_upload_owned(
+    request: Request,
+    content: str,
+    user: Optional[str],
+    upload_handler: Any,
+) -> None:
+    """Reject document content whose pdf_source marker points at another user's upload."""
+    if upload_handler is None:
+        return
+    from src.pdf_form_doc import find_source_upload_id
+
+    upload_id = find_source_upload_id(content or "")
+    if not upload_id:
+        return
+    auth_manager = getattr(getattr(request.app, "state", None), "auth_manager", None)
+    if not _resolve_user_upload_path(upload_handler, upload_id, user, auth_manager):
+        raise HTTPException(
+            400,
+            "Document PDF marker references an upload you do not own",
+        )
+
+
+def _derive_title(content: str) -> str:
+    """Derive a title from document content."""
+    import re
+    if not isinstance(content, str):
+        return "Untitled"
+    text = content.strip()
+    if not text:
+        return "Untitled"
+
+    # Markdown header
+    md = re.match(r'^#{1,3}\s+(.+)', text, re.MULTILINE)
+    if md:
+        title = md.group(1).strip()
+        if len(title) > 50:
+            title = title[:48] + "…"
+        return title
+
+    # HTML heading
+    html = re.search(r'<h[1-3][^>]*>([^<]+)</h[1-3]>', text, re.IGNORECASE)
+    if html:
+        title = html.group(1).strip()
+        if len(title) > 50:
+            title = title[:48] + "…"
+        return title
+
+    # First non-empty line (if short enough)
+    for line in text.split('\n'):
+        line = line.strip()
+        if line and 2 <= len(line) <= 60:
+            title = re.sub(r'[:#*`]+$', '', line).strip()
+            if title and len(title) > 50:
+                title = title[:48] + "…"
+            return title or "Untitled"
+
+    return "Untitled"

routes/document_helpers.py

@@ -1,243 +1,14 @@
-"""document_helpers.py — Pydantic models, doc serializers, owner gating, file-locator helpers shared with document_routes.py."""
+"""Backward-compat shim — canonical location is routes/document/document_helpers.py.
 
-"""Document routes — CRUD for living documents with version history."""
+This module is replaced in ``sys.modules`` by the canonical module object so
+that ``import routes.document_helpers``, ``from routes.document_helpers import
+X``, and the ``sys.modules.pop("routes.document_helpers")`` + re-import
+pattern used by test_security_regressions.py all operate on the *same* object.
+Keeps existing import paths working after slice 2m (#4082/#4071).
+"""
 
-import logging
+import sys as _sys
-import os
-import re
-from typing import Any, Dict, Optional
 
-from fastapi import HTTPException, Request
+from routes.document import document_helpers as _canonical  # noqa: F401
-from pydantic import BaseModel
 
-from core.database import Document, DocumentVersion
+_sys.modules[__name__] = _canonical
-from core.database import Session as DbSession
-from src.auth_helpers import _auth_disabled
-from src.upload_handler import UploadHandler
-
-logger = logging.getLogger(__name__)
-
-
-# ---- Request schemas ----
-
-class DocumentCreate(BaseModel):
-    session_id: Optional[str] = None
-    title: str = "Untitled"
-    language: Optional[str] = None
-    content: str = ""
-
-class DocumentUpdate(BaseModel):
-    content: str
-    summary: Optional[str] = None
-    force_version: bool = False
-
-class DocumentPatch(BaseModel):
-    title: Optional[str] = None
-    language: Optional[str] = None
-    session_id: Optional[str] = None  # link/unlink document to a session
-
-
-# ---- Helpers ----
-
-def _doc_to_dict(doc: Document) -> Dict[str, Any]:
-    return {
-        "id": doc.id,
-        "session_id": doc.session_id,
-        "title": doc.title,
-        "language": doc.language,
-        "current_content": doc.current_content,
-        "version_count": doc.version_count,
-        "is_active": doc.is_active,
-        "archived": bool(getattr(doc, "archived", False)),
-        "created_at": (doc.created_at.isoformat() + "Z") if doc.created_at else None,
-        "updated_at": (doc.updated_at.isoformat() + "Z") if doc.updated_at else None,
-        # Source-email provenance (set when doc was created from an email
-        # attachment) — drives the "Send signed reply" menu item.
-        "source_email_uid":        getattr(doc, "source_email_uid", None),
-        "source_email_folder":     getattr(doc, "source_email_folder", None),
-        "source_email_account_id": getattr(doc, "source_email_account_id", None),
-        "source_email_message_id": getattr(doc, "source_email_message_id", None),
-    }
-
-def _version_to_dict(v: DocumentVersion) -> Dict[str, Any]:
-    return {
-        "id": v.id,
-        "document_id": v.document_id,
-        "version_number": v.version_number,
-        "content": v.content,
-        "summary": v.summary,
-        "source": v.source,
-        "created_at": v.created_at.isoformat() if v.created_at else None,
-    }
-
-
-def _verify_doc_owner(db, doc: Document, user: str):
-    """Verify `user` owns this document. Raise 404 if not.
-
-    Documents now carry their own `owner` column, so a doc whose session
-    was deleted (session_id → NULL) can still prove ownership and stay
-    openable / cloneable. We trust that column first and only fall back to
-    the session join for any not-yet-backfilled legacy row.
-    """
-    if user is None:
-        if _auth_disabled():
-            return  # Single-user / no-auth mode: allow access
-        raise HTTPException(403, "Authentication required")
-    if doc.owner is not None:
-        if doc.owner != user:
-            raise HTTPException(404, "Document not found")
-        return
-    # Legacy fallback: derive ownership from the linked session.
-    if not doc.session_id:
-        raise HTTPException(404, "Document not found")
-    session = db.query(DbSession).filter(DbSession.id == doc.session_id).first()
-    if not session or session.owner != user:
-        raise HTTPException(404, "Document not found")
-
-
-def _owner_session_filter(q, user):
-    """Restrict a documents query to those owned by `user`.
-
-    Documents now carry their own `owner` column (backfilled at boot from
-    the linked session, or assigned to the admin user for legacy/orphaned
-    docs). We filter on that directly rather than on a session join, so a
-    document whose session was deleted (session_id → NULL) still shows up
-    for its owner instead of silently vanishing from the Library + search.
-
-    The owner backfill runs in init_db before the app serves requests, so
-    by the time this filter is live there are no NULL-owner rows to leak;
-    we therefore match the owner strictly for authenticated callers."""
-    if not user:
-        if user == "" or _auth_disabled():
-            return q
-        return q.filter(False)
-    return q.filter(Document.owner == user)
-
-
-
-def _slug(name: str) -> str:
-    """Filesystem-friendly version of a document title.
-
-    Whitespace becomes underscores; other unsafe punctuation is dropped.
-    Preserves letters, digits, dot, hyphen, underscore. Idempotent.
-    """
-    import re as _re
-    s = (name or "").strip()
-    # Drop the trailing extension if the title happens to include one
-    s = _re.sub(r'\.pdf$', '', s, flags=_re.IGNORECASE)
-    s = _re.sub(r'\s+', '_', s)
-    s = _re.sub(r'[^A-Za-z0-9._-]', '', s)
-    s = _re.sub(r'_+', '_', s).strip('_')
-    return s or "form"
-
-
-# DPI scale for the interactive PDF view. ~150 DPI (2x of 72 PDF user-units).
-_PDF_RENDER_SCALE = 2.0
-
-
-def _upload_path_inside(upload_dir: str, path: str) -> bool:
-    base = os.path.realpath(upload_dir)
-    p = os.path.realpath(path)
-    try:
-        return os.path.commonpath([base, p]) == base
-    except Exception:
-        return False
-
-
-def _resolve_user_upload_path(
-    upload_handler: Any,
-    upload_id: str,
-    owner: Optional[str],
-    auth_manager=None,
-) -> Optional[str]:
-    """Resolve an upload id to a filesystem path the caller may read."""
-    if upload_handler is None:
-        return None
-    resolved = upload_handler.resolve_upload(
-        upload_id,
-        owner=owner,
-        auth_manager=auth_manager,
-    )
-    if not isinstance(resolved, dict) or not resolved:
-        return None
-    path = resolved.get("path")
-    upload_dir = getattr(upload_handler, "upload_dir", None)
-    if path and upload_dir and not _upload_path_inside(upload_dir, path):
-        logger.warning("Upload path outside upload directory: %s", path)
-        return None
-    return path
-
-
-def _locate_upload(
-    upload_dir: str,
-    file_id: str,
-    owner: Optional[str] = None,
-    auth_manager=None,
-    upload_handler: Any = None,
-):
-    """Find an upload by its filename ID via UploadHandler.resolve_upload."""
-    if upload_handler is None:
-        from src.upload_handler import UploadHandler
-
-        base_dir = os.path.dirname(os.path.abspath(upload_dir))
-        upload_handler = UploadHandler(base_dir, upload_dir)
-    return _resolve_user_upload_path(upload_handler, file_id, owner, auth_manager)
-
-
-def _assert_pdf_marker_upload_owned(
-    request: Request,
-    content: str,
-    user: Optional[str],
-    upload_handler: Any,
-) -> None:
-    """Reject document content whose pdf_source marker points at another user's upload."""
-    if upload_handler is None:
-        return
-    from src.pdf_form_doc import find_source_upload_id
-
-    upload_id = find_source_upload_id(content or "")
-    if not upload_id:
-        return
-    auth_manager = getattr(getattr(request.app, "state", None), "auth_manager", None)
-    if not _resolve_user_upload_path(upload_handler, upload_id, user, auth_manager):
-        raise HTTPException(
-            400,
-            "Document PDF marker references an upload you do not own",
-        )
-
-
-def _derive_title(content: str) -> str:
-    """Derive a title from document content."""
-    import re
-    if not isinstance(content, str):
-        return "Untitled"
-    text = content.strip()
-    if not text:
-        return "Untitled"
-
-    # Markdown header
-    md = re.match(r'^#{1,3}\s+(.+)', text, re.MULTILINE)
-    if md:
-        title = md.group(1).strip()
-        if len(title) > 50:
-            title = title[:48] + "…"
-        return title
-
-    # HTML heading
-    html = re.search(r'<h[1-3][^>]*>([^<]+)</h[1-3]>', text, re.IGNORECASE)
-    if html:
-        title = html.group(1).strip()
-        if len(title) > 50:
-            title = title[:48] + "…"
-        return title
-
-    # First non-empty line (if short enough)
-    for line in text.split('\n'):
-        line = line.strip()
-        if line and 2 <= len(line) <= 60:
-            title = re.sub(r'[:#*`]+$', '', line).strip()
-            if title and len(title) > 50:
-                title = title[:48] + "…"
-            return title or "Untitled"
-
-    return "Untitled"

src/integrations.py

@@ -1,11 +1,14 @@
+import ipaddress
 import json
 import os
+import time
 import uuid
 import logging
 import re
 from typing import Dict, List, Optional, Any
 from urllib.parse import urljoin, urlparse, urlunparse
 
+import httpcore
 import httpx
 from fastapi import HTTPException
 
@@ -354,6 +357,152 @@ def _find_integration(identifier: str) -> Optional[Dict[str, Any]]:
     return None
 
 
+# httpcore raises its own exception hierarchy; map the ones a simple request can
+# surface back to their httpx equivalents so the caller's `except httpx.*` blocks
+# below behave exactly as they did with the default transport.
+_HTTPCORE_TO_HTTPX_EXC = {
+    httpcore.ConnectError: httpx.ConnectError,
+    httpcore.ConnectTimeout: httpx.ConnectTimeout,
+    httpcore.NetworkError: httpx.NetworkError,
+    httpcore.PoolTimeout: httpx.PoolTimeout,
+    httpcore.ProtocolError: httpx.ProtocolError,
+    httpcore.ReadError: httpx.ReadError,
+    httpcore.ReadTimeout: httpx.ReadTimeout,
+    httpcore.RemoteProtocolError: httpx.RemoteProtocolError,
+    httpcore.TimeoutException: httpx.TimeoutException,
+    httpcore.WriteError: httpx.WriteError,
+    httpcore.WriteTimeout: httpx.WriteTimeout,
+}
+
+
+class _PinnedAsyncBackend(httpcore.AsyncNetworkBackend):
+    """Network backend that connects only to the pre-validated IPs, in order.
+
+    Every address here came out of the single SSRF resolution, so moving to the
+    next one after a connect failure is not re-resolution — it's ordinary
+    multi-address fallback restricted to the set the guard already approved.
+    httpcore takes TLS SNI and the ``Host`` header from the request URL rather
+    than the connect host, so pinning the socket destination leaves certificate
+    validation and vhost routing pointed at the original hostname.
+    """
+
+    def __init__(self, ips: List[ipaddress._BaseAddress]):
+        self._ips = [str(ip) for ip in ips]
+        self._real = httpcore.AnyIOBackend()
+
+    async def connect_tcp(self, host, port, timeout=None, local_address=None,
+                          socket_options=None):
+        # One shared connect budget: each attempt gets the time left until the
+        # original deadline, so N dead addresses can't stretch the connect
+        # phase to N * timeout.
+        deadline = None if timeout is None else time.monotonic() + timeout
+        last_exc: Optional[Exception] = None
+        for ip in self._ips:
+            remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
+            try:
+                return await self._real.connect_tcp(
+                    ip, port, remaining, local_address, socket_options
+                )
+            except (httpcore.ConnectError, httpcore.ConnectTimeout) as exc:
+                last_exc = exc
+                if deadline is not None and time.monotonic() >= deadline:
+                    break
+        raise last_exc
+
+    async def connect_unix_socket(self, path, timeout=None, socket_options=None):
+        return await self._real.connect_unix_socket(path, timeout, socket_options)
+
+    async def sleep(self, seconds: float) -> None:
+        return await self._real.sleep(seconds)
+
+
+class _PinnedAsyncTransport(httpx.AsyncBaseTransport):
+    """httpx transport that pins the TCP connect to the pre-resolved IP(s).
+
+    Kept local, mirroring the per-module pinned transports web fetch and
+    webhook delivery already carry, rather than coupling api_call to the
+    webhook subsystem. The request URL passes through unchanged, so SNI and the
+    ``Host`` header stay the original hostname; only the socket destination is
+    pinned, which is what closes the rebinding window.
+    """
+
+    def __init__(self, ips: List[ipaddress._BaseAddress]):
+        self._pinned_ips = list(ips)
+        self._pool = httpcore.AsyncConnectionPool(
+            # Reuse the CA trust the default httpx client would build (certifi
+            # plus SSL_CERT_FILE / SSL_CERT_DIR when trust_env is set) so
+            # swapping in this transport doesn't quietly change which chains
+            # verify. ssl.create_default_context() would use system roots.
+            ssl_context=httpx.create_ssl_context(),
+            http1=True,
+            http2=False,
+            network_backend=_PinnedAsyncBackend(ips),
+        )
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        core_req = httpcore.Request(
+            method=request.method,
+            url=httpcore.URL(
+                scheme=request.url.raw_scheme,
+                host=request.url.raw_host,
+                port=request.url.port,
+                target=request.url.raw_path,
+            ),
+            headers=request.headers.raw,
+            content=request.stream,
+            extensions=request.extensions,
+        )
+        try:
+            core_resp = await self._pool.handle_async_request(core_req)
+            content = b"".join([chunk async for chunk in core_resp.aiter_stream()])
+            await core_resp.aclose()
+        except Exception as exc:
+            mapped = _HTTPCORE_TO_HTTPX_EXC.get(type(exc))
+            if mapped is not None:
+                raise mapped(str(exc)) from exc
+            raise
+        return httpx.Response(
+            status_code=core_resp.status,
+            headers=core_resp.headers,
+            content=content,
+            extensions=core_resp.extensions,
+        )
+
+    async def aclose(self) -> None:
+        await self._pool.aclose()
+
+
+def _validated_ips(raw_ips: List[str]) -> List[ipaddress._BaseAddress]:
+    """Return every entry that parses as an IP address, de-duplicated, order
+    preserved.
+
+    check_outbound_url only reports ok when *all* of these classify as safe, so
+    the whole list is guard-approved and any of them is a legitimate connect
+    target. Skipping unparseable entries mirrors how the guard walks the same
+    resolver output.
+
+    De-duplication matters because the resolver is getaddrinfo(host, None) with
+    no socktype filter, so glibc reports the same address once per socktype
+    (SOCK_STREAM/SOCK_DGRAM/SOCK_RAW) — a single-homed host comes back three
+    times. Without this, the connect fallback would spend the shared deadline
+    retrying one dead address instead of moving on to a genuinely different one.
+    """
+    ips: List[ipaddress._BaseAddress] = []
+    seen = set()
+    for raw in raw_ips:
+        if not isinstance(raw, str):
+            continue
+        try:
+            ip = ipaddress.ip_address(raw.split("%")[0])  # strip IPv6 zone id
+        except ValueError:
+            continue
+        if ip in seen:
+            continue
+        seen.add(ip)
+        ips.append(ip)
+    return ips
+
+
 async def execute_api_call(
     integration_id: str,
     method: str,
@@ -409,13 +558,31 @@ async def execute_api_call(
     # loopback for locked-down deployments. Private stays allowed by default
     # because LAN integrations (Home Assistant, Miniflux, ntfy) are the
     # primary use case.
-    from src.url_safety import check_outbound_url
+    from src.url_safety import check_outbound_url, _default_resolver
     block_private = os.getenv(
         "INTEGRATION_API_BLOCK_PRIVATE_IPS", "false"
     ).lower() == "true"
-    ok, reason = check_outbound_url(url, block_private=block_private)
+    # Resolve the host exactly once and remember the IPs the guard validated so
+    # the request below can be pinned to them. check_outbound_url only reports
+    # (ok, reason); a plain httpx client re-resolves the host at connect time,
+    # which reopens a DNS-rebinding TOCTOU — a base_url host that answers with a
+    # public IP for the guard and then flips to 169.254.169.254 for the connect
+    # would reach cloud metadata with the integration's auth headers attached.
+    resolved_ips: List[str] = []
+
+    def _recording_resolver(host: str) -> List[str]:
+        ips = _default_resolver(host)
+        resolved_ips[:] = ips
+        return ips
+
+    ok, reason = check_outbound_url(
+        url, block_private=block_private, resolver=_recording_resolver
+    )
     if not ok:
         return {"error": f"URL rejected: {reason}", "exit_code": 1}
+    pinned_ips = _validated_ips(resolved_ips)
+    if not pinned_ips:
+        return {"error": "URL rejected: host did not resolve to a usable address", "exit_code": 1}
 
     method = method.upper()
 
@@ -455,7 +622,9 @@ async def execute_api_call(
             auth = httpx.BasicAuth(parts[0], parts[1])
 
     try:
-        async with httpx.AsyncClient(timeout=30.0) as client:
+        async with httpx.AsyncClient(
+            timeout=30.0, transport=_PinnedAsyncTransport(pinned_ips)
+        ) as client:
             response = await client.request(
                 method,
                 url,

src/tools/system.py

@@ -46,7 +46,9 @@ async def do_manage_skills(content: str, owner: Optional[str] = None) -> Dict:
     except ValueError:
         return {"error": "Invalid JSON arguments", "exit_code": 1}
 
-    action = (args.get("action") or "").lower()
+    action = (args.get("action") or "").strip().lower()
+    if not action:
+        return {"error": "action is required (list|view|view_ref|add|edit|patch|publish|delete|search)", "exit_code": 1}
     from services.memory.skills import SkillsManager
     from services.memory.skill_format import Skill, slugify
     from src.constants import DATA_DIR
@@ -55,7 +57,7 @@ async def do_manage_skills(content: str, owner: Optional[str] = None) -> Dict:
     # Accept legacy `skill_id` as an alias for `name`.
     name = (args.get("name") or args.get("skill_id") or "").strip()
 
-    if action in ("list", "index", ""):
+    if action in ("list", "index"):
         all_skills = sm.load(owner=owner)
         if not all_skills:
             return {"results": "No skills yet. Create one with action='add'."}

tests/test_document_routes_shim.py

@@ -0,0 +1,29 @@
+"""Regression test for the document route shim (slice 2m, #4082/#4071).
+
+The backward-compat shims at ``routes/document_routes.py`` and
+``routes/document_helpers.py`` use ``sys.modules`` replacement so the legacy
+import paths and the canonical ``routes.document.*`` paths resolve to the
+*same* module objects. This is required because multiple tests do
+``import routes.document_routes as droutes`` followed by
+``droutes.SessionLocal = ...`` / ``monkeypatch.setattr(droutes, ...)`` and
+``sys.modules.pop("routes.document_helpers")`` + re-import — for those to
+take effect at runtime, the legacy and canonical module objects must be
+identical.
+"""
+
+import importlib
+
+import routes.document_routes as _shim_routes  # noqa: F401
+import routes.document_helpers as _shim_helpers  # noqa: F401
+
+
+def test_legacy_and_canonical_routes_are_same_object():
+    legacy = importlib.import_module("routes.document_routes")
+    canonical = importlib.import_module("routes.document.document_routes")
+    assert legacy is canonical
+
+
+def test_legacy_and_canonical_helpers_are_same_object():
+    legacy = importlib.import_module("routes.document_helpers")
+    canonical = importlib.import_module("routes.document.document_helpers")
+    assert legacy is canonical

tests/test_imap_mailbox_quoting.py

@@ -87,7 +87,7 @@ def test_known_imap_mailbox_call_sites_are_quoted():
     assert "conn.select(sent_name" not in pollers
     assert "imap.append(sent_folder" not in pollers
 
-    document_routes = Path("routes/document_routes.py").read_text()
+    document_routes = Path("routes/document/document_routes.py").read_text()
     assert "conn.select(doc.source_email_folder" not in document_routes
 
 

tests/test_integration_api_call_ssrf.py

@@ -9,8 +9,13 @@ link-local/metadata is always rejected; RFC-1918/loopback only when
 INTEGRATION_API_BLOCK_PRIVATE_IPS=true (LAN integrations are the primary
 use case, so private stays allowed by default).
 """
+import asyncio
+import ipaddress
+import ssl
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import httpcore
+import httpx
 import pytest
 
 from src import integrations
@@ -97,3 +102,238 @@ async def test_private_base_url_allowed_by_default_blocked_with_knob(monkeypatch
     assert result["exit_code"] == 1
     assert "rejected" in result["error"].lower()
     client.request.assert_not_called()
+
+
+async def _call_capturing_transport(base_url, path="/items"):
+    """Drive execute_api_call and return (result, transport) where transport is
+    the object passed to httpx.AsyncClient(transport=...)."""
+    resp = MagicMock()
+    resp.status_code = 200
+    resp.headers = {"content-type": "application/json"}
+    resp.json.return_value = {"ok": True}
+    resp.text = '{"ok": true}'
+
+    client = AsyncMock()
+    client.__aenter__ = AsyncMock(return_value=client)
+    client.__aexit__ = AsyncMock(return_value=None)
+    client.request = AsyncMock(return_value=resp)
+
+    captured = {}
+
+    def _fake_async_client(*args, **kwargs):
+        captured.update(kwargs)
+        return client
+
+    with (
+        patch.object(integrations, "_find_integration",
+                     return_value=_integration(base_url)),
+        patch("httpx.AsyncClient", side_effect=_fake_async_client),
+    ):
+        result = await integrations.execute_api_call("test_integ", "GET", path)
+    return result, captured.get("transport"), client
+
+
+@pytest.mark.asyncio
+async def test_connection_is_pinned_to_the_validated_ip(monkeypatch):
+    """DNS-rebinding defense: the guard resolves the host once to a benign
+    public IP, and the request must be pinned to *that* IP so a host that
+    rebinds to the metadata range at connect time can't be reached with the
+    integration's auth headers. Static resolution passing the guard is not
+    enough — a plain client would re-resolve at connect."""
+    monkeypatch.setattr("src.url_safety._default_resolver",
+                        lambda host: ["93.184.216.34"])
+    result, transport, client = await _call_capturing_transport(
+        "http://rebinding.attacker.example")
+
+    assert result.get("exit_code") == 0
+    client.request.assert_called_once()
+    assert isinstance(transport, integrations._PinnedAsyncTransport)
+    assert [str(ip) for ip in transport._pinned_ips] == ["93.184.216.34"]
+
+
+@pytest.mark.asyncio
+async def test_pin_carries_the_whole_validated_ip_set(monkeypatch):
+    """When a host resolves to several records the transport keeps all of them
+    (check_outbound_url validated every one), in resolver order, so it can fall
+    back past a dead first address instead of failing the whole call."""
+    monkeypatch.setattr("src.url_safety._default_resolver",
+                        lambda host: ["93.184.216.34", "198.51.100.7"])
+    result, transport, _ = await _call_capturing_transport("http://multi.example")
+
+    assert result.get("exit_code") == 0
+    assert [str(ip) for ip in transport._pinned_ips] == ["93.184.216.34", "198.51.100.7"]
+
+
+class _FakeStream:
+    """Stand-in for the connected socket the real backend returns."""
+
+
+class _RecordingBackend:
+    """Fake httpcore backend: connect_tcp fails for the addresses in `dead`
+    and succeeds for the rest, recording the order it was asked to connect."""
+
+    def __init__(self, dead):
+        self.dead = set(dead)
+        self.attempts = []
+
+    async def connect_tcp(self, host, port, timeout=None, local_address=None,
+                          socket_options=None):
+        self.attempts.append((host, timeout))
+        if host in self.dead:
+            raise httpcore.ConnectError(f"connection refused: {host}")
+        return _FakeStream()
+
+
+def _pinned_backend(ips, dead):
+    """A _PinnedAsyncBackend whose underlying connect is the recording fake."""
+    backend = integrations._PinnedAsyncBackend(ips)
+    backend._real = _RecordingBackend(dead)
+    return backend
+
+
+@pytest.mark.asyncio
+async def test_connect_falls_back_from_dead_first_to_live_second():
+    """first-dead / second-live: the pinned backend must try the next validated
+    address when the first refuses, rather than surfacing the failure. It also
+    ignores the `host` httpcore passes (the original hostname) and connects to
+    the pinned IPs, which is what keeps TLS SNI / Host on the real hostname."""
+    ips = [ipaddress.ip_address("203.0.113.10"), ipaddress.ip_address("198.51.100.7")]
+    backend = _pinned_backend(ips, dead={"203.0.113.10"})
+
+    stream = await backend.connect_tcp("original.hostname.example", 443, timeout=5.0)
+
+    assert isinstance(stream, _FakeStream)
+    # Tried the dead address first, then the live one — never the hostname.
+    assert [host for host, _ in backend._real.attempts] == ["203.0.113.10", "198.51.100.7"]
+    # Fallback shared one budget: the second attempt got the time left, not a fresh 5s.
+    assert backend._real.attempts[1][1] <= 5.0
+
+
+@pytest.mark.asyncio
+async def test_connect_raises_when_every_validated_address_is_dead():
+    ips = [ipaddress.ip_address("203.0.113.10"), ipaddress.ip_address("198.51.100.7")]
+    backend = _pinned_backend(ips, dead={"203.0.113.10", "198.51.100.7"})
+
+    with pytest.raises(httpcore.ConnectError):
+        await backend.connect_tcp("original.hostname.example", 443, timeout=5.0)
+    assert [host for host, _ in backend._real.attempts] == ["203.0.113.10", "198.51.100.7"]
+
+
+@pytest.mark.asyncio
+async def test_pinned_transport_reuses_httpx_ca_trust(monkeypatch):
+    """TLS trust must come from the same builder the default httpx client uses
+    (certifi + SSL_CERT_FILE / SSL_CERT_DIR via trust_env), not from
+    ssl.create_default_context()'s system roots — otherwise chains that verified
+    under the old default client can silently stop verifying."""
+    sentinel = ssl.create_default_context()
+    calls = []
+
+    def _fake_create(*args, **kwargs):
+        calls.append(kwargs)
+        return sentinel
+
+    monkeypatch.setattr(httpx, "create_ssl_context", _fake_create)
+    transport = integrations._PinnedAsyncTransport([ipaddress.ip_address("93.184.216.34")])
+    try:
+        assert calls, "transport did not build its context via httpx.create_ssl_context"
+        assert transport._pool._ssl_context is sentinel
+    finally:
+        await transport.aclose()
+
+
+@pytest.mark.asyncio
+async def test_real_socket_falls_back_from_dead_first_to_live_second():
+    """End-to-end over real loopback sockets: pin [127.0.0.2 (nothing
+    listening), 127.0.0.1 (live)], and the request must succeed by falling back
+    to the second address while the Host header stays the original hostname —
+    i.e. only the socket destination moved, vhost/SNI routing did not."""
+    captured = {}
+
+    async def handle(reader, writer):
+        request = await reader.read(4096)
+        for line in request.split(b"\r\n"):
+            if line.lower().startswith(b"host:"):
+                captured["host"] = line.split(b":", 1)[1].strip().decode()
+        writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nhi")
+        await writer.drain()
+        writer.close()
+
+    server = await asyncio.start_server(handle, "127.0.0.1", 0)
+    port = server.sockets[0].getsockname()[1]
+    async with server:
+        await server.start_serving()
+        transport = integrations._PinnedAsyncTransport(
+            [ipaddress.ip_address("127.0.0.2"), ipaddress.ip_address("127.0.0.1")]
+        )
+        try:
+            async with httpx.AsyncClient(transport=transport) as client:
+                resp = await client.get(f"http://pinned.example:{port}/health")
+        finally:
+            await transport.aclose()
+
+    assert resp.status_code == 200
+    assert resp.text == "hi"
+    assert captured.get("host") == f"pinned.example:{port}"
+
+
+@pytest.mark.asyncio
+async def test_ip_literal_base_url_still_pins_and_is_not_rejected():
+    """A base_url that is already an IP has nothing to rebind, but it must not
+    trip the "did not resolve" guard either.
+
+    check_outbound_url resolves even a literal (getaddrinfo returns the address
+    itself), so the captured list is populated and the pin is a no-op rather
+    than a rejection. Uses the real resolver on purpose — no monkeypatch — so
+    this would catch the fail-closed branch firing on a literal.
+    """
+    result, transport, client = await _call_capturing_transport(
+        "http://93.184.216.34")
+
+    assert result.get("exit_code") == 0
+    assert isinstance(transport, integrations._PinnedAsyncTransport)
+    assert [str(ip) for ip in transport._pinned_ips] == ["93.184.216.34"]
+
+
+@pytest.mark.asyncio
+async def test_ipv6_base_url_pins_every_validated_address(monkeypatch):
+    """IPv6 goes down the same path as v4.
+
+    Resolution is stubbed rather than using a literal so this doesn't depend on
+    the runner having IPv6 configured.
+    """
+    v6 = "2606:2800:220:1:248:1893:25c8:1946"
+    monkeypatch.setattr("src.url_safety._default_resolver", lambda host: [v6])
+    result, transport, client = await _call_capturing_transport("http://v6.example")
+
+    assert result.get("exit_code") == 0
+    assert isinstance(transport, integrations._PinnedAsyncTransport)
+    assert [str(ip) for ip in transport._pinned_ips] == [v6]
+
+
+def test_validated_ips_strips_zone_id_and_drops_junk():
+    """getaddrinfo can hand back a scoped v6 address like 'fe80::1%eth0'."""
+    got = integrations._validated_ips(
+        ["93.184.216.34", "fe80::1%eth0", "not-an-ip", None, "2001:db8::5"]
+    )
+    assert [str(ip) for ip in got] == ["93.184.216.34", "fe80::1", "2001:db8::5"]
+
+
+def test_validated_ips_deduplicates_repeated_addresses():
+    """The resolver is getaddrinfo(host, None) with no socktype filter, so glibc
+    returns one record per socktype and a single-homed host arrives three times
+    over. Duplicates must collapse (first-seen order kept) or the connect
+    fallback wastes its shared deadline retrying one dead address."""
+    got = integrations._validated_ips(
+        ["93.184.216.34", "93.184.216.34", "93.184.216.34"]
+    )
+    assert [str(ip) for ip in got] == ["93.184.216.34"]
+
+    # Order is first-seen, and distinct addresses all survive.
+    got = integrations._validated_ips(
+        ["198.51.100.7", "93.184.216.34", "198.51.100.7", "2001:db8::5"]
+    )
+    assert [str(ip) for ip in got] == ["198.51.100.7", "93.184.216.34", "2001:db8::5"]
+
+    # A zone-id variant is the same address once stripped, so it collapses too.
+    got = integrations._validated_ips(["fe80::1%eth0", "fe80::1%eth1", "fe80::1"])
+    assert [str(ip) for ip in got] == ["fe80::1"]

tests/test_integrations_api_call_truncation.py

@@ -83,9 +83,10 @@ async def _call(json_data, status=200):
     with (
         patch.object(integrations, "_find_integration", return_value=DUMMY_INTEGRATION),
         patch("httpx.AsyncClient", return_value=mock_client),
-        # api.example.com doesn't resolve; the SSRF guard would fail closed.
+        # api.example.com doesn't resolve. Point the resolver at a public
-        # These tests are about truncation, so stub the guard open.
+        # address instead of stubbing the guard open, so the real check (and
-        patch("src.url_safety.check_outbound_url", return_value=(True, "ok")),
+        # the connect-IP pinning that reads its result) still runs.
+        patch("src.url_safety._default_resolver", lambda host: ["93.184.216.34"]),
     ):
         return await integrations.execute_api_call("test_integ", "GET", "/items")
 
@@ -101,9 +102,10 @@ async def _call_with_integration(integration, path="/items"):
     with (
         patch.object(integrations, "_find_integration", return_value=integration),
         patch("httpx.AsyncClient", return_value=mock_client),
-        # api.example.com doesn't resolve; the SSRF guard would fail closed.
+        # api.example.com doesn't resolve. Point the resolver at a public
-        # These tests are about URL joining, so stub the guard open.
+        # address instead of stubbing the guard open, so the real check (and
-        patch("src.url_safety.check_outbound_url", return_value=(True, "ok")),
+        # the connect-IP pinning that reads its result) still runs.
+        patch("src.url_safety._default_resolver", lambda host: ["93.184.216.34"]),
     ):
         result = await integrations.execute_api_call("test_integ", "GET", path)
     return result, mock_client

tests/test_manage_skills_action_required.py

@@ -0,0 +1,24 @@
+import json
+
+import pytest
+
+from src.tools.system import do_manage_skills
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "payload",
+    [
+        {},
+        {"action": ""},
+        {"action": "   "},
+        {"name": "demo", "description": "x", "procedure": ["step"]},
+    ],
+)
+async def test_manage_skills_requires_action(payload):
+    result = await do_manage_skills(json.dumps(payload), owner="test")
+
+    assert result == {
+        "error": "action is required (list|view|view_ref|add|edit|patch|publish|delete|search)",
+        "exit_code": 1,
+    }

tests/test_model_helper_owner_scope.py

@@ -14,7 +14,7 @@ def _function_source(path: str, name: str) -> str:
 
 
 def test_document_ai_tidy_resolves_with_owner_scope():
-    body = _function_source("routes/document_routes.py", "ai_tidy_documents")
+    body = _function_source("routes/document/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
 

tests/test_vision_owner_scope.py

@@ -88,7 +88,7 @@ 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()
+    document_source = (ROOT / "routes" / "document" / "document_routes.py").read_text()
     gallery_source = (ROOT / "routes" / "gallery" / "gallery_routes.py").read_text()
     memory_source = (ROOT / "routes" / "memory" / "memory_routes.py").read_text()