feat(mcp): per-user scope restrictions for BasicAuth mode

.dockerignore

@@ -4,6 +4,4 @@
 !README.md
 !uv.lock
 
-!nextcloud_mcp_server/**/*.py
-!nextcloud_mcp_server/**/*.html
-!nextcloud_mcp_server/auth/static/*
+!nextcloud_mcp_server/

Dockerfile

@@ -8,7 +8,7 @@ COPY --from=ghcr.io/astral-sh/uv:0.10.7@sha256:edd1fd89f3e5b005814cc8f777610445d
 RUN apt update && apt install --no-install-recommends --no-install-suggests -y \
     git \
     tesseract-ocr \
-    sqlite3 && apt clean
+    sqlite3 && apt clean && rm -rf /var/lib/apt/lists/*
 
 WORKDIR /app
 
@@ -20,6 +20,10 @@ COPY . .
 
 RUN uv sync --locked --no-dev --no-editable --no-cache
 
+RUN adduser --disabled-password --gecos "" --uid 10001 appuser && \
+    chown -R appuser:appuser /app
+USER appuser
+
 ENV PYTHONUNBUFFERED=1
 ENV VIRTUAL_ENV=/app/.venv
 ENV PATH=/app/.venv/bin:$PATH

Dockerfile.smithery

@@ -23,13 +23,17 @@ COPY --from=ghcr.io/astral-sh/uv:0.10.7@sha256:edd1fd89f3e5b005814cc8f777610445d
 # 1. git (required for caldav dependency from git)
 # 2. sqlite for development with token db
 RUN apt update && apt install --no-install-recommends --no-install-suggests -y \
-    git
+    git && apt clean && rm -rf /var/lib/apt/lists/*
 
 # Copy project files
 COPY . .
 
 RUN uv sync --locked --no-dev --no-editable --no-cache
 
+RUN adduser --disabled-password --gecos "" --uid 10001 appuser && \
+    chown -R appuser:appuser /app
+USER appuser
+
 # Set Smithery mode environment variables
 ENV SMITHERY_DEPLOYMENT=true
 ENV VECTOR_SYNC_ENABLED=false

README.md

@@ -119,6 +119,7 @@ The server supports four authentication modes:
 **Multi-User (BasicAuth Pass-Through):**
 - MCP clients send credentials via Authorization header
 - Server passes through to Nextcloud (stateless by default)
+- Optional per-user scope restrictions via `BASIC_AUTH_SCOPES_<USERNAME>` env vars
 - Optional offline access for background operations (`ENABLE_MULTI_USER_BASIC_AUTH=true`)
 - Best for: Multi-user setups without OAuth infrastructure
 

docs/authentication.md

@@ -170,6 +170,12 @@ When running in multi-user BasicAuth mode with `ENABLE_OFFLINE_ACCESS=true`, the
 # Enable multi-user BasicAuth
 ENABLE_MULTI_USER_BASIC_AUTH=true
 
+# Per-user scope restrictions (optional)
+# Omit a user's variable to grant full access (backward compatible)
+BASIC_AUTH_SCOPES_CLAUDE=files:read,files:write,notes:read
+BASIC_AUTH_SCOPES_N8N=calendar:read,calendar:write
+# BASIC_AUTH_SCOPES_ADMIN not set → full access
+
 # Enable hybrid mode (OAuth provisioning for management APIs)
 ENABLE_OFFLINE_ACCESS=true
 

nextcloud_mcp_server/app.py

@@ -97,6 +97,7 @@
 from nextcloud_mcp_server.config import (
     DeploymentMode,
     Settings,
+    get_basic_auth_scopes,
     get_document_processor_config,
     get_settings,
 )
@@ -456,6 +457,36 @@ async def __call__(
                     logger.debug(
                         f"BasicAuth credentials extracted for user: {username}"
                     )
+
+                    # Inject synthetic AccessToken if scopes are configured
+                    # for this user, enabling @require_scopes checks and
+                    # list_tools_filtered() to work in BasicAuth mode
+                    configured_scopes = get_basic_auth_scopes(username)
+                    if configured_scopes is not None:
+                        from mcp.server.auth.middleware.auth_context import (  # noqa: PLC0415
+                            auth_context_var,
+                        )
+                        from mcp.server.auth.middleware.bearer_auth import (  # noqa: PLC0415
+                            AuthenticatedUser,
+                        )
+                        from mcp.server.auth.provider import (  # noqa: PLC0415
+                            AccessToken,
+                        )
+
+                        synthetic_token = AccessToken(
+                            token="basic-auth-synthetic",
+                            client_id=f"basic-auth:{username}",
+                            scopes=configured_scopes,
+                        )
+                        ctx_token = auth_context_var.set(
+                            AuthenticatedUser(synthetic_token)
+                        )
+                        try:
+                            await self.app(scope, receive, send)
+                        finally:
+                            auth_context_var.reset(ctx_token)
+                        return
+
                 except Exception as e:
                     logger.warning(f"Failed to extract BasicAuth credentials: {e}")
 
@@ -1480,8 +1511,9 @@ async def nc_get_capabilities():
         logger.info("Registering Login Flow v2 auth tools")
         register_auth_tools(mcp)
 
-    # Override list_tools to filter based on user's token scopes (OAuth mode only)
-    if oauth_enabled:
+    # Override list_tools to filter based on user's token scopes
+    # Active in OAuth mode and when BasicAuth users have scope configuration
+    if oauth_enabled or settings.enable_multi_user_basic_auth:
         original_list_tools = mcp._tool_manager.list_tools
 
         def list_tools_filtered():
@@ -1525,9 +1557,8 @@ def list_tools_filtered():
 
         # Replace the tool manager's list_tools method
         mcp._tool_manager.list_tools = list_tools_filtered  # type: ignore[method-assign]
-        logger.info(
-            "Dynamic tool filtering enabled for OAuth mode (JWT and Bearer tokens)"
-        )
+        mode = "OAuth" if oauth_enabled else "BasicAuth"
+        logger.info(f"Dynamic tool filtering enabled for {mode} mode")
 
     mcp_app = mcp.streamable_http_app()
 

nextcloud_mcp_server/auth/__init__.py

@@ -12,6 +12,7 @@
     get_required_scopes,
     has_required_scopes,
     is_jwt_token,
+    require_resource_scopes,
     require_scopes,
 )
 from .unified_verifier import UnifiedTokenVerifier
@@ -23,6 +24,7 @@
     "ensure_oauth_client",
     "get_client_from_context",
     "require_scopes",
+    "require_resource_scopes",
     "ScopeAuthorizationError",
     "InsufficientScopeError",
     "check_scopes",

nextcloud_mcp_server/auth/browser_oauth_routes.py

@@ -28,6 +28,17 @@
 logger = logging.getLogger(__name__)
 
 
+def _validate_redirect_url(url: str, default: str = "/app") -> str:
+    """Validate redirect URL to prevent open redirects.
+
+    Only relative paths (starting with /) without embedded protocol are allowed.
+    Rejects absolute URLs, protocol-relative URLs, and other redirect tricks.
+    """
+    if not url.startswith("/") or "://" in url or url.startswith("//"):
+        return default
+    return url
+
+
 def _should_use_secure_cookies() -> bool:
     """Determine if cookies should have secure flag.
 
@@ -75,7 +86,7 @@ async def oauth_login(request: Request) -> RedirectResponse | JSONResponse:
     logger.info(f"oauth_login called - oauth_client: {oauth_client is not None}")
 
     # Get redirect URL from query params (default to /app)
-    next_url = request.query_params.get("next", "/app")
+    next_url = _validate_redirect_url(request.query_params.get("next", "/app"))
     logger.info(f"oauth_login - next_url: {next_url}")
 
     # Generate state for CSRF protection
@@ -442,7 +453,7 @@ async def oauth_login_callback(request: Request) -> RedirectResponse | HTMLRespo
 
     # Create response and set session cookie
     # Redirect to stored next_url (from OAuth session) or /app as default
-    response = RedirectResponse(next_url, status_code=302)
+    response = RedirectResponse(_validate_redirect_url(next_url), status_code=302)
     response.set_cookie(
         key="mcp_session",
         value=user_id,
@@ -465,7 +476,9 @@ async def oauth_logout(request: Request) -> RedirectResponse:
     Returns:
         302 redirect with cleared session cookie
     """
-    next_url = request.query_params.get("next", "/oauth/login")
+    next_url = _validate_redirect_url(
+        request.query_params.get("next", "/oauth/login"), default="/oauth/login"
+    )
 
     # TODO: Optionally revoke refresh token from storage
     # session_id = request.cookies.get("mcp_session")

nextcloud_mcp_server/auth/scope_authorization.py

@@ -475,6 +475,43 @@ async def create_note():
     return sorted(all_scopes)
 
 
+def require_resource_scopes(*required_scopes: str):
+    """Decorator to require scopes for MCP resource handlers.
+
+    Unlike require_scopes (which extracts Context from tool parameters),
+    this decorator reads the access token directly from the MCP SDK's
+    auth_context_var. This works for @mcp.resource() handlers that don't
+    receive a Context parameter.
+
+    When no access token is present (unconfigured BasicAuth), all access
+    is permitted for backward compatibility.
+
+    Args:
+        *required_scopes: Scope strings required for access
+
+    Raises:
+        InsufficientScopeError: If the token lacks required scopes
+    """
+
+    def decorator(func: Callable) -> Callable:
+        func._required_scopes = list(required_scopes)  # type: ignore[attr-defined]
+
+        @wraps(func)
+        async def wrapper(*args: Any, **kwargs: Any) -> Any:
+            access_token: AccessToken | None = get_access_token()
+            if access_token is None:
+                # No token → BasicAuth without scope config → allow all
+                return await func(*args, **kwargs)
+            missing = set(required_scopes) - set(access_token.scopes or [])
+            if missing:
+                raise InsufficientScopeError(sorted(missing))
+            return await func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator
+
+
 # ── Login Flow v2 helpers ────────────────────────────────────────────────
 
 # Scope cache: user_id → (expires_at, scopes)

nextcloud_mcp_server/client/calendar.py

@@ -13,6 +13,8 @@
 from icalendar import Alarm, Calendar, vDDDTypes, vRecur
 from icalendar import Event as ICalEvent
 from icalendar import Todo as ICalTodo
+
+from nextcloud_mcp_server.utils.xml import escape_xml
 from lxml import etree  # type: ignore[import-untyped]
 
 from ..config import get_nextcloud_ssl_verify
@@ -123,7 +125,8 @@ async def list_calendars(self) -> List[Dict[str, Any]]:
         result = []
 
         # Parse XML response
-        tree = etree.fromstring(response.raw.encode("utf-8"))
+        parser = etree.XMLParser(resolve_entities=False, no_network=True)
+        tree = etree.fromstring(response.raw.encode("utf-8"), parser=parser)
         ns = {
             "d": "DAV:",
             "cs": "http://calendarserver.org/ns/",
@@ -203,9 +206,9 @@ async def create_calendar(
 <mkcalendar xmlns="urn:ietf:params:xml:ns:caldav" xmlns:d="DAV:" xmlns:cs="http://calendarserver.org/ns/">
     <d:set>
         <d:prop>
-            <d:displayname>{display_name or calendar_name}</d:displayname>
-            <cs:calendar-color>{color}</cs:calendar-color>
-            <caldav:calendar-description xmlns:caldav="urn:ietf:params:xml:ns:caldav">{description}</caldav:calendar-description>
+            <d:displayname>{escape_xml(display_name or calendar_name)}</d:displayname>
+            <cs:calendar-color>{escape_xml(color)}</cs:calendar-color>
+            <caldav:calendar-description xmlns:caldav="urn:ietf:params:xml:ns:caldav">{escape_xml(description)}</caldav:calendar-description>
             <caldav:supported-calendar-component-set xmlns:caldav="urn:ietf:params:xml:ns:caldav">
                 <caldav:comp name="VEVENT"/>
                 <caldav:comp name="VTODO"/>

nextcloud_mcp_server/client/contacts.py

@@ -1,10 +1,12 @@
 """CardDAV client for NextCloud contacts operations."""
 
 import logging
-import xml.etree.ElementTree as ET
+import defusedxml.ElementTree as ET
 
 from pythonvCard4.vcard import Contact
 
+from nextcloud_mcp_server.utils.xml import escape_xml
+
 from .base import BaseNextcloudClient
 
 logger = logging.getLogger(__name__)
@@ -104,7 +106,7 @@ async def create_addressbook(self, *, name: str, display_name: str):
                         <d:collection/>
                         <c:addressbook/>
                     </d:resourcetype>
-                    <d:displayname>{display_name}</d:displayname>
+                    <d:displayname>{escape_xml(display_name)}</d:displayname>
                 </d:prop>
             </d:set>
         </d:mkcol>"""

nextcloud_mcp_server/client/cookbook.py

@@ -6,6 +6,8 @@
 
 from httpx import Timeout
 
+from nextcloud_mcp_server.utils.url import validate_external_url
+
 from .base import BaseNextcloudClient
 
 logger = logging.getLogger(__name__)
@@ -130,6 +132,7 @@ async def import_recipe(self, url: str) -> Dict[str, Any]:
         Returns:
             Full imported recipe data
         """
+        validate_external_url(url)
         logger.info(f"Importing recipe from URL: {url}")
         response = await self._make_request(
             "POST",

nextcloud_mcp_server/client/news.py

@@ -4,6 +4,8 @@
 from enum import IntEnum
 from typing import Any
 
+from nextcloud_mcp_server.utils.url import validate_external_url
+
 from .base import BaseNextcloudClient
 
 logger = logging.getLogger(__name__)
@@ -121,6 +123,7 @@ async def create_feed(
         Raises:
             HTTPStatusError: 409 if feed already exists, 422 if URL is invalid
         """
+        validate_external_url(url)
         body: dict[str, Any] = {"url": url}
         if folder_id is not None:
             body["folderId"] = folder_id