fix(dav): reject ambiguous nested authorization encodings

backend/api/data.py

@@ -2489,10 +2489,16 @@ async def _get_workspace_document(
     auth_context: AuthContext,
     document_id: str,
 ) -> Document:
+    organization_filter = (
+        Document.organization_id == auth_context.organization_id
+        if auth_context.organization_id is not None
+        else Document.organization_id.is_(None)
+    )
     result = await db.execute(
         select(Document).where(
             Document.document_id == document_id,
             Document.workspace_id == auth_context.workspace_id,
+            organization_filter,
         )
     )
     document = result.scalar_one_or_none()
@@ -3928,10 +3934,18 @@ async def get_data_quality_surface(
             ProjectFolder.folder_uid.asc(),
         ),
     )
+    document_organization_filter = (
+        Document.organization_id == auth_context.organization_id
+        if auth_context.organization_id is not None
+        else Document.organization_id.is_(None)
+    )
     documents = await _scoped_rows(
         db,
         select(Document)
-        .where(Document.workspace_id == auth_context.workspace_id)
+        .where(
+            Document.workspace_id == auth_context.workspace_id,
+            document_organization_filter,
+        )
         .order_by(Document.created_at.desc(), Document.document_id.asc())
         .limit(8),
     )

backend/api/dav.py

@@ -1,6 +1,5 @@
 import logging
 from html import escape as escape_xml_text
-from urllib.parse import unquote
 
 from fastapi import APIRouter, Depends, HTTPException, Request, Response
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -13,15 +12,67 @@
 
 router = APIRouter(prefix="/dav", tags=["dav"])
 
+_DAV_AUTHORIZATION_PATH_MAX_CHARACTERS = 8192
+_HEX_DIGITS = frozenset("0123456789abcdefABCDEF")
+_DAV_STRUCTURAL_OCTETS = frozenset(
+    {*range(0x20), 0x2E, 0x2F, 0x5C, 0x7F}
+)
+
+
+def _residual_percent_octet(path: str, percent_index: int) -> int | None:
+    """Return the octet exposed by another decode, following nested ``%25``."""
+
+    cursor = percent_index + 1
+    while cursor + 1 < len(path):
+        pair = path[cursor : cursor + 2]
+        if pair[0] not in _HEX_DIGITS or pair[1] not in _HEX_DIGITS:
+            return None
+        octet = int(pair, 16)
+        if octet != 0x25:
+            return octet
+        cursor += 2
+    return None
+
+
+def _has_ambiguous_percent_encoding(path: str) -> bool:
+    """Detect residual encodings that another decode would make structural."""
+
+    for index, character in enumerate(path):
+        if character != "%":
+            continue
+        octet = _residual_percent_octet(path, index)
+        if octet in _DAV_STRUCTURAL_OCTETS:
+            return True
+    return False
+
 
 def _normalize_dav_authorization_path(path: str) -> str:
+    """Validate the framework-decoded DAV path without decoding it again.
+
+    ASGI routing has already decoded the request-target path once. Authorization
+    therefore treats residual percent text as data unless another decode would
+    introduce a traversal dot, separator, backslash, or control octet. Literal
+    backslashes are normalized to separators for owner/traversal checks.
+    """
+
+    if len(path) > _DAV_AUTHORIZATION_PATH_MAX_CHARACTERS:
+        raise HTTPException(
+            status_code=414,
+            detail="DAV path exceeds authorization length limit",
+        )
+    if any(ord(character) < 0x20 or ord(character) == 0x7F for character in path):
+        raise HTTPException(
+            status_code=400,
+            detail="DAV path contains control characters",
+        )
+
     normalized_path = path.replace("\\", "/")
-    for _ in range(100):
-        decoded_path = unquote(normalized_path).replace("\\", "/")
-        if decoded_path == normalized_path:
-            return normalized_path
-        normalized_path = decoded_path
-    raise HTTPException(status_code=400, detail="DAV path decoding limit exceeded")
+    if _has_ambiguous_percent_encoding(normalized_path):
+        raise HTTPException(
+            status_code=400,
+            detail="DAV path contains ambiguous percent encoding",
+        )
+    return normalized_path
 
 
 def _dav_path_owner_user_id(path: str) -> str | None:
@@ -173,8 +224,9 @@ async def dav_handler(
     Provider-backed writeback stays fail-closed until source capability and
     ETag/If-Match enforcement are available through signed writeback intents.
     """
-    _ensure_dav_owner_scope(path, auth_context)
-    safe_path = repr(path)[1:-1]
+    normalized_path = _normalize_dav_authorization_path(path)
+    _ensure_dav_owner_scope(normalized_path, auth_context)
+    safe_path = repr(normalized_path)[1:-1]
     logger.info("DAV Request: %s /%s", request.method, safe_path)
 
     if request.method == "OPTIONS":
@@ -190,14 +242,13 @@ async def dav_handler(
     if request.method == "PROPFIND":
         return await _handle_project_propfind(
             request=request,
-            path=path,
+            path=normalized_path,
             auth_context=auth_context,
             db=db,
         )
 
     if request.method == "PUT":
         body = await request.body()
-        safe_path = repr(path)[1:-1]
         logger.info("DAV PUT received %s bytes at /%s", len(body), safe_path)
         logger.warning(
             "DAV PUT rejected at /%s: provider-backed DAV writeback is not "

backend/services/llm_provider_urls.py

@@ -16,6 +16,12 @@
 _DNS_RESOLUTION_TIMEOUT_SECONDS = 5.0
 _LOCAL_DEV_HOSTNAMES = {"localhost", "localhost.localdomain"}
 _LOCAL_DEV_IP_LITERALS = {"127.0.0.1", "::1"}
+_LOCAL_PROVIDER_NETWORKS = (
+    ipaddress.ip_network("10.0.0.0/8"),
+    ipaddress.ip_network("172.16.0.0/12"),
+    ipaddress.ip_network("192.168.0.0/16"),
+    ipaddress.ip_network("fc00::/7"),
+)
 
 
 def _has_url_control_character(value: str) -> bool:
@@ -74,6 +80,15 @@ def _is_allowlisted_local_provider_host(hostname: str) -> bool:
     )
 
 
+def _is_local_provider_network_address(
+    address: ipaddress.IPv4Address | ipaddress.IPv6Address,
+) -> bool:
+    return any(
+        address.version == network.version and address in network
+        for network in _LOCAL_PROVIDER_NETWORKS
+    )
+
+
 def _format_normalized_netloc(hostname: str, port: int, *, explicit_port: bool) -> str:
     host_part = f"[{hostname}]" if ":" in hostname else hostname
     if not explicit_port:
@@ -82,16 +97,14 @@ def _format_normalized_netloc(hostname: str, port: int, *, explicit_port: bool)
 
 
 def _validate_global_address(address: str, *, hostname: str | None = None) -> str:
-    """Validate that an IP address is globally routable, or explicitly allowed.
-
-    When ``ALLOW_LOCAL_LLM_PROVIDERS`` is enabled the address is accepted if:
-    - the IP is a loopback address, **or**
-    - the *original* hostname (before DNS resolution) is present in
-      ``ALLOWED_LLM_BASE_URL_HOSTS``.
-
-    This second condition is necessary because Docker container names (e.g.
-    ``ollama``) resolve to RFC-1918 private IPs that would otherwise be
-    rejected by the global-address check.
+    """Validate a globally routable address or an explicitly scoped local one.
+
+    ``ALLOW_LOCAL_LLM_PROVIDERS`` admits loopback addresses for local developer
+    runtimes. An exact, operator-allowlisted single-label provider hostname may
+    additionally resolve only into RFC 1918 IPv4 or RFC 4193 IPv6 unique-local
+    space. Link-local, reserved, unspecified, multicast, and other non-global
+    address classes never become reachable merely because a hostname is
+    allowlisted.
     """
     try:
         ip_address = ipaddress.ip_address(address)
@@ -102,7 +115,11 @@ def _validate_global_address(address: str, *, hostname: str | None = None) -> st
     if settings.ALLOW_LOCAL_LLM_PROVIDERS:
         if ip_address.is_loopback:
             is_allowed_local = True
-        elif hostname and _is_allowlisted_local_provider_host(hostname):
+        elif (
+            hostname
+            and _is_allowlisted_local_provider_host(hostname)
+            and _is_local_provider_network_address(ip_address)
+        ):
             is_allowed_local = True
 
     if not is_allowed_local:
@@ -130,8 +147,8 @@ def _resolve_all_global_addresses(hostname: str, port: int) -> tuple[str, ...]:
     addresses: list[str] = []
     seen_addresses: set[str] = set()
     for address_info in address_infos:
-        # Pass the original hostname so that Docker container names listed in
-        # ALLOWED_LLM_BASE_URL_HOSTS are matched before checking the resolved IP.
+        # Pass the original hostname so that explicitly allowlisted local-provider
+        # names can be bound to their permitted private container addresses.
         address = _validate_global_address(str(address_info[4][0]), hostname=hostname)
         if address not in seen_addresses:
             seen_addresses.add(address)
@@ -269,8 +286,7 @@ def __init__(self, hostname: str, port: int, addresses: tuple[str, ...]):
             raise ValueError(LLM_BASE_URL_NOT_ALLOWED)
         self._hostname = hostname
         self._port = port
-        # Re-validate each address; pass the hostname so Docker-container names
-        # in ALLOWED_LLM_BASE_URL_HOSTS are accepted.
+        # Re-validate each address against the same hostname-scoped local boundary.
         self._addresses = tuple(
             _validate_global_address(address, hostname=hostname)
             for address in addresses

backend/tests/test_data_document_authorization.py

@@ -0,0 +1,128 @@
+"""Authorization regressions for workspace document actions."""
+
+from __future__ import annotations
+
+import pytest
+from fastapi import HTTPException
+
+from api.auth import AuthContext
+from api.data import _get_workspace_document
+from db.models import Document
+
+
+class _ScalarResult:
+    """Minimal SQLAlchemy result surface used by the authorization helper."""
+
+    def __init__(self, value: Document | None) -> None:
+        self._value = value
+
+    def scalar_one_or_none(self) -> Document | None:
+        """Return the single simulated document result."""
+
+        return self._value
+
+
+class _OrganizationAwareSession:
+    """Evaluate document scope predicates against one in-memory document."""
+
+    def __init__(self, document: Document) -> None:
+        self.document = document
+
+    async def execute(self, statement):
+        """Return the document only when every emitted scope predicate matches."""
+
+        compiled = statement.compile()
+        rendered_scope = str(statement.whereclause)
+        values = tuple(compiled.params.values())
+        if self.document.document_id not in values:
+            return _ScalarResult(None)
+        if self.document.workspace_id not in values:
+            return _ScalarResult(None)
+
+        # A missing organization predicate reproduces the vulnerable behavior:
+        # a document from another tenant is still returned solely because both
+        # principals supplied the same workspace identifier.
+        if "workspace_documents.organization_id" not in rendered_scope:
+            return _ScalarResult(self.document)
+
+        if self.document.organization_id is None:
+            organization_matches = "IS NULL" in rendered_scope.upper()
+        else:
+            organization_matches = self.document.organization_id in values
+        return _ScalarResult(self.document if organization_matches else None)
+
+
+def _auth_context(organization_id: str | None) -> AuthContext:
+    return AuthContext(
+        user_id="member-a",
+        role="member",
+        organization_id=organization_id,
+        group_ids=(),
+        workspace_id="workspace-shared-identifier",
+    )
+
+
+@pytest.mark.asyncio
+async def test_workspace_document_lookup_rejects_cross_organization_idor() -> None:
+    """A reused workspace identifier must not cross the tenant boundary."""
+
+    document = Document(
+        document_id="doc-org-b",
+        workspace_id="workspace-shared-identifier",
+        organization_id="org-b",
+        document_name="private.md",
+        document_type="text/markdown",
+        document_content="tenant B evidence",
+        document_status="uploaded",
+    )
+    session = _OrganizationAwareSession(document)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await _get_workspace_document(session, _auth_context("org-a"), document.document_id)  # type: ignore[arg-type]
+
+    assert exc_info.value.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_workspace_document_lookup_preserves_same_organization_collaboration() -> None:
+    """Workspace members in the owning organization retain shared-document access."""
+
+    document = Document(
+        document_id="doc-org-a",
+        workspace_id="workspace-shared-identifier",
+        organization_id="org-a",
+        document_name="shared.md",
+        document_type="text/markdown",
+        document_content="shared tenant evidence",
+        document_status="uploaded",
+    )
+    session = _OrganizationAwareSession(document)
+
+    resolved = await _get_workspace_document(
+        session,  # type: ignore[arg-type]
+        _auth_context("org-a"),
+        document.document_id,
+    )
+
+    assert resolved is document
+
+
+@pytest.mark.asyncio
+async def test_personal_workspace_document_lookup_rejects_organization_document() -> None:
+    """Personal-scope sessions cannot read organization-owned workspace documents."""
+
+    document = Document(
+        document_id="doc-org-owned",
+        workspace_id="workspace-shared-identifier",
+        organization_id="org-a",
+        document_name="org.md",
+        document_type="text/markdown",
+        document_content="organization evidence",
+        document_status="uploaded",
+    )
+    session = _OrganizationAwareSession(document)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await _get_workspace_document(session, _auth_context(None), document.document_id)  # type: ignore[arg-type]
+
+    assert exc_info.value.status_code == 404