diff --git a/PROGRESS.md b/PROGRESS.md index 6041dfc..d127e03 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -58,6 +58,7 @@ - [x] search_files matches % and _ in the query literally instead of as database wildcards (2026-09-26) - [x] update_circle_member_level explains when Circles cannot transfer ownership (SQLite or PostgreSQL before nextcloud/circles#2916) instead of passing on the database error (2026-09-26) - [x] Team folders (Nextcloud 35 with the Team folders app): create_circle makes one only with team_folder=true; delete_circle, leave_circle and delete_collective(delete_team) refuse to delete a team folder and its files unless delete_team_folder=true (2026-09-29) +- [x] Multi-user mode (NEXTCLOUD_MCP_MULTIUSER=true, HTTP only): every request brings its own Basic login (app password); per-login client pool (LRU + idle TTL); X-Nextcloud-MCP-Permissions lowers the level per request; initialize/tools/list without a login for client discovery (2026-10-07) ### In Progress @@ -123,7 +124,8 @@ | Cospend | 16 | 41 | | Flow | 5 | 58 | | Pagination | — | 28 | -| **Total** | **223** | **1827** | +| Multi-User | — | 64 | +| **Total** | **223** | **1891** | The test counts are what pytest collects, recomputed with `python scripts/sync_progress.py --write`, which also fails when a test file is not assigned to a row. diff --git a/README.md b/README.md index 4c7d152..d60716b 100644 --- a/README.md +++ b/README.md @@ -114,6 +114,8 @@ export NEXTCLOUD_MCP_UPLOAD_ROOT= # unset (default). If set to an absolute # inside that directory (symlinks resolved). ``` +To serve many Nextcloud accounts from one HTTP server, see [Multi-User Mode](#multi-user-mode-http-only). + ### Getting an App Password 1. Log into your Nextcloud instance @@ -170,6 +172,66 @@ nc-mcp-server --transport http # Listens on http://0.0.0.0:8100 by default ``` +### Multi-User Mode (HTTP only) + +By default the server acts as the one account in `NEXTCLOUD_USER`. With `NEXTCLOUD_MCP_MULTIUSER=true`, +one HTTP server can act for many Nextcloud accounts instead: the server holds no credentials, and every +request brings its own login. + +```bash +export NEXTCLOUD_URL=https://your-nextcloud.example.com +export NEXTCLOUD_MCP_MULTIUSER=true +export NEXTCLOUD_MCP_PERMISSIONS=write # the highest level any request may use +# NEXTCLOUD_USER and NEXTCLOUD_PASSWORD must NOT be set +nc-mcp-server --transport http +``` + +| Variable | Default | Meaning | +|----------|---------|---------| +| `NEXTCLOUD_MCP_MULTIUSER` | `false` | Take the Nextcloud login from each HTTP request. Needs `--transport http`. | +| `NEXTCLOUD_MCP_MULTIUSER_CACHE_SIZE` | `64` | Number of per-login clients kept (least recently used ones are dropped first). | +| `NEXTCLOUD_MCP_MULTIUSER_TTL` | `900` | Seconds an unused login's client is kept. | + +**Authentication.** Every request needs `Authorization: Basic base64(:)`. The login +can be the user ID or another login name, such as the e-mail address. The first request of a login asks +Nextcloud for the user ID (`GET /ocs/v2.php/cloud/user`); this also checks the credentials. The server +answers before any tool runs: + +| Situation | HTTP status | +|-----------|-------------| +| No `Authorization` header (except discovery, see below), malformed or repeated `Authorization` header, or a login that Nextcloud rejects (401/403) | `401` with `WWW-Authenticate: Basic` | +| Invalid or repeated `X-Nextcloud-MCP-Permissions` header | `400` | +| Nextcloud not reachable, a 5xx or an unexpected answer while the login is checked | `502` | +| `GET` or `DELETE` without a login | `405` | + +There is no fallback account: a request that has no valid login never runs as anyone. + +**Discovery without a login.** Some MCP clients list the tools at startup, before they know which user +will call them. A `POST` **without** an `Authorization` header is allowed when all of its JSON-RPC messages +are `initialize`, `ping`, `tools/list`, `notifications/initialized` or `notifications/cancelled`. The answer is +the static tool list, and no Nextcloud call happens. `tools/call` and all other methods still need a login. +A request with a wrong `Authorization` header is refused; it is never treated as discovery. + +**Client pool.** The server keeps one Nextcloud client per login (LRU with an idle TTL, see the variables +above). Two logins never share a client. The cache key is an HMAC of login and password with a random key +per process, so the pool does not keep the password or a plain hash of it as a key. A client that is +dropped from the pool is closed after a grace period, so a tool call that is still running can finish. + +**Per-request permission cap.** A request can lower, but never raise, the server's permission level with +the header `X-Nextcloud-MCP-Permissions: read|write|destructive`. Example: a read-only service account that +uses the same server. + +**Security notes.** + +- Each request is authenticated on its own. The MCP session ID is not a credential, and the user of a + request comes only from that request's header. +- Basic Auth sends the app password with every request. Use TLS (for example a reverse proxy) or a private + network between the MCP client and this server. +- Anyone who can reach the server with a valid app password can use it as that user, up to the server's + permission level. Give each user an app password of their own, and revoke it in Nextcloud to remove access. +- `NEXTCLOUD_MCP_UPLOAD_ROOT` is not available in this mode, because it would give every user access to the + server's local files. + ### Stdio Mode (default) ```bash diff --git a/pyproject.toml b/pyproject.toml index da2ee5a..75accbe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -112,6 +112,9 @@ reportUnusedFunction = false reportPrivateUsage = false executionEnvironments = [ { root = "tests/test_client_retry.py", reportAttributeAccessIssue = false, reportUnknownMemberType = false }, + # starlette's TestClient (httpx) has no complete types here + { root = "tests/test_multiuser.py", reportUnknownMemberType = false, reportUnknownVariableType = false, reportUnknownArgumentType = false }, + { root = "tests/integration/test_multiuser.py", reportUnknownMemberType = false, reportUnknownVariableType = false, reportUnknownArgumentType = false }, { root = "src/nc_mcp_server/tools/calendar.py", reportUnknownMemberType = false, reportUnknownVariableType = false, reportUnknownArgumentType = false }, { root = "src/nc_mcp_server/tools/contacts.py", reportUnknownMemberType = false, reportUnknownVariableType = false, reportUnknownArgumentType = false }, { root = "src/nc_mcp_server/tools/tasks.py", reportUnknownMemberType = false, reportUnknownVariableType = false, reportUnknownArgumentType = false }, diff --git a/scripts/sync_progress.py b/scripts/sync_progress.py index c8d0203..6829c27 100644 --- a/scripts/sync_progress.py +++ b/scripts/sync_progress.py @@ -56,6 +56,7 @@ "Circles": ["integration/test_circles.py", "test_circles.py"], "Cospend": ["integration/test_cospend.py"], "Flow": ["integration/test_flows.py", "test_flows.py"], + "Multi-User": ["integration/test_multiuser.py", "test_multiuser.py"], } diff --git a/src/nc_mcp_server/__main__.py b/src/nc_mcp_server/__main__.py index 84db39d..a9bfc2c 100644 --- a/src/nc_mcp_server/__main__.py +++ b/src/nc_mcp_server/__main__.py @@ -2,7 +2,10 @@ import argparse -from .server import create_server +import uvicorn + +from .config import Config +from .server import create_http_app, create_server def main() -> None: @@ -15,10 +18,16 @@ def main() -> None: ) args = parser.parse_args() - mcp = create_server() + config = Config.from_env() + if config.multiuser and args.transport != "http": + parser.error("NEXTCLOUD_MCP_MULTIUSER=true needs --transport http (logins come from HTTP headers)") + mcp = create_server(config) if args.transport == "http": - mcp.run(transport="streamable-http") + if config.multiuser: + uvicorn.run(create_http_app(mcp, config), host=config.host, port=config.port, log_level="info") + else: + mcp.run(transport="streamable-http") else: mcp.run(transport="stdio") diff --git a/src/nc_mcp_server/client.py b/src/nc_mcp_server/client.py index 25e4280..3dca126 100644 --- a/src/nc_mcp_server/client.py +++ b/src/nc_mcp_server/client.py @@ -307,7 +307,7 @@ async def _send(self, session: niquests.AsyncSession, method: str, url: str, **k def _build_session(self) -> niquests.AsyncSession: kwargs: dict[str, object] = { - "auth": (self._config.user, self._config.password), + "auth": (self._config.auth_login, self._config.password), "timeout": 30, "headers": { "OCS-APIRequest": "true", diff --git a/src/nc_mcp_server/config.py b/src/nc_mcp_server/config.py index fa2dbd7..96bd0e4 100644 --- a/src/nc_mcp_server/config.py +++ b/src/nc_mcp_server/config.py @@ -25,6 +25,11 @@ class Config: NEXTCLOUD_MCP_UPLOAD_ROOT: Absolute path to a local directory. When set, enables the upload_file_from_path tool, restricted to files under this directory (symlinks are resolved before the containment check). Unset by default — tool disabled. + NEXTCLOUD_MCP_MULTIUSER: Set to 'true' to serve many Nextcloud accounts from one HTTP + server. Every request then carries its own ``Authorization: Basic `` + header; NEXTCLOUD_USER and NEXTCLOUD_PASSWORD must not be set. Requires --transport http. + NEXTCLOUD_MCP_MULTIUSER_CACHE_SIZE: Clients kept for recent logins (default: 64). + NEXTCLOUD_MCP_MULTIUSER_TTL: Seconds an unused login's client is kept (default: 900). """ nextcloud_url: str = field(default="") @@ -36,6 +41,17 @@ class Config: retry_max: int = field(default=3) is_app_password: bool = field(default=False) upload_root: str = field(default="") + multiuser: bool = field(default=False) + multiuser_cache_size: int = field(default=64) + multiuser_ttl: float = field(default=900.0) + # Login name for Basic Auth when it differs from the user ID (e.g. a login by e-mail address). + # Empty means the user ID. Set per login in multi-user mode; the user ID builds DAV paths. + login: str = field(default="") + + @property + def auth_login(self) -> str: + """The name sent with Basic Auth: the login name if known, else the user ID.""" + return self.login or self.user @classmethod def from_env(cls) -> "Config": @@ -59,13 +75,10 @@ def from_env(cls) -> "Config": except ValueError: raise ValueError(f"Invalid NEXTCLOUD_MCP_RETRY_MAX='{retry_raw}'. Expected integer >= 0.") from None - app_pw_raw = os.environ.get("NEXTCLOUD_MCP_APP_PASSWORD", "").strip().lower() - if app_pw_raw in ("", "false", "0", "no"): - is_app_password = False - elif app_pw_raw in ("true", "1", "yes"): - is_app_password = True - else: - raise ValueError(f"Invalid NEXTCLOUD_MCP_APP_PASSWORD='{app_pw_raw}'. Expected: true/false, 1/0, yes/no.") + is_app_password = _env_bool("NEXTCLOUD_MCP_APP_PASSWORD") + multiuser = _env_bool("NEXTCLOUD_MCP_MULTIUSER") + cache_size = _env_number("NEXTCLOUD_MCP_MULTIUSER_CACHE_SIZE", "64", int, minimum=1) + ttl = _env_number("NEXTCLOUD_MCP_MULTIUSER_TTL", "900", float, minimum=1) upload_root_raw = os.environ.get("NEXTCLOUD_MCP_UPLOAD_ROOT", "").strip() if upload_root_raw: @@ -88,6 +101,9 @@ def from_env(cls) -> "Config": retry_max=max(0, retry_max), is_app_password=is_app_password, upload_root=upload_root, + multiuser=multiuser, + multiuser_cache_size=int(cache_size), + multiuser_ttl=float(ttl), ) def validate(self) -> None: @@ -95,6 +111,21 @@ def validate(self) -> None: missing: list[str] = [] if not self.nextcloud_url: missing.append("NEXTCLOUD_URL") + if self.multiuser: + # No fallback account: a request without its own login must fail, not run as someone else. + if self.user or self.password: + raise ValueError( + "NEXTCLOUD_MCP_MULTIUSER=true takes the login from each request; " + "unset NEXTCLOUD_USER and NEXTCLOUD_PASSWORD." + ) + if self.upload_root: + raise ValueError("NEXTCLOUD_MCP_UPLOAD_ROOT is not supported with NEXTCLOUD_MCP_MULTIUSER=true.") + if missing: + raise ValueError( + f"Missing required environment variables: {', '.join(missing)}. " + f"Set them before starting the MCP server." + ) + return if not self.user: missing.append("NEXTCLOUD_USER") if not self.password: @@ -104,3 +135,23 @@ def validate(self) -> None: f"Missing required environment variables: {', '.join(missing)}. " f"Set them before starting the MCP server." ) + + +def _env_bool(name: str) -> bool: + raw = os.environ.get(name, "").strip().lower() + if raw in ("", "false", "0", "no"): + return False + if raw in ("true", "1", "yes"): + return True + raise ValueError(f"Invalid {name}='{raw}'. Expected: true/false, 1/0, yes/no.") + + +def _env_number(name: str, default: str, kind: type[int] | type[float], minimum: float) -> float: + raw = os.environ.get(name, default).strip() + try: + value = kind(raw) + except ValueError: + raise ValueError(f"Invalid {name}='{raw}'. Expected a number >= {minimum:g}.") from None + if value < minimum: + raise ValueError(f"Invalid {name}='{raw}'. Expected a number >= {minimum:g}.") + return value diff --git a/src/nc_mcp_server/multiuser.py b/src/nc_mcp_server/multiuser.py new file mode 100644 index 0000000..65739e5 --- /dev/null +++ b/src/nc_mcp_server/multiuser.py @@ -0,0 +1,473 @@ +"""Multi-user mode: one HTTP server, one Nextcloud login per request. + +With ``NEXTCLOUD_MCP_MULTIUSER=true`` the server holds no account of its own. Every HTTP request +must carry ``Authorization: Basic ``. :class:`MultiUserAuthMiddleware` checks +the header before the MCP layer sees the request, resolves the login to its Nextcloud user ID +(one OCS call per new login, which also proves the credentials), and pins the matching client to +the request. Tools keep calling :func:`nc_mcp_server.state.get_client` and +:func:`nc_mcp_server.state.get_config`; in this mode those return the request's own client and a +config whose ``user`` is the request's user ID. + +Rules: + +* No header, a malformed header or credentials Nextcloud rejects -> HTTP 401, nothing runs. +* Exception: discovery without any ``Authorization`` header. A POST whose JSON-RPC messages are all + in :data:`DISCOVERY_METHODS` (``initialize``, ``ping``, ``tools/list``, notifications) runs + without a login. Some clients list tools at startup, before they know who will call them; the + answer is the same static tool list for everyone and no Nextcloud call happens. Everything else + (``tools/call`` above all) still needs a login. A request that sends a header that is wrong is + refused, never downgraded to discovery. GET/DELETE without a login get 405 (no SSE stream). +* Clients are cached per login (LRU, idle TTL). The cache key is an HMAC of login and password + with a per-process random key, so neither the password nor a plain hash of it is kept as a key. +* A request can lower, never raise, the server's permission level with the + ``X-Nextcloud-MCP-Permissions`` header (e.g. ``read`` for a read-only service account). +* The identity comes from each request's header, never from the MCP session: a session ID is not + a credential, and two requests in one session may run as different logins. +* On ASGI lifespan shutdown the pool closes every cached client. +""" + +import asyncio +import base64 +import binascii +import contextlib +import contextvars +import dataclasses +import hashlib +import hmac +import json +import logging +import secrets +import time +from collections import OrderedDict +from collections.abc import Awaitable, Callable, Generator +from typing import Any, cast + +import niquests +from mcp.server.lowlevel.server import request_ctx + +from .client import NextcloudClient +from .config import Config +from .permissions import PermissionLevel, set_cap_provider + +log = logging.getLogger(__name__) + +PERMISSIONS_HEADER = "x-nextcloud-mcp-permissions" +# A replaced or evicted client may still serve a tool call that started before; close it later. +CLOSE_GRACE_SECONDS = 600.0 +_MAX_HEADER_BYTES = 4096 +# Upper bound for a discovery body read before the login check; real ones are a few hundred bytes. +_MAX_DISCOVERY_BODY_BYTES = 64 * 1024 +# JSON-RPC methods that run without a login: protocol handshake and the static tool list. +DISCOVERY_METHODS = frozenset( + { + "initialize", + "ping", + "tools/list", + "notifications/initialized", + "notifications/cancelled", + } +) +_AMBIGUOUS = "\x00ambiguous" + + +class AuthenticationRequiredError(Exception): + """A tool ran without the login of a request (multi-user mode).""" + + +@dataclasses.dataclass(frozen=True) +class Credentials: + """A Nextcloud login taken from one request. ``repr`` never shows the password.""" + + login: str + password: str = dataclasses.field(repr=False) + + +@dataclasses.dataclass +class Binding: + """What a request runs as: its client, its config and its permission cap.""" + + client: NextcloudClient + config: Config + permission_cap: PermissionLevel | None = None + + +def parse_basic_auth(value: str | None) -> Credentials | None: + """Parse an ``Authorization`` header value. Returns None unless it is a usable Basic login.""" + if not value or len(value) > _MAX_HEADER_BYTES: + return None + scheme, _, token = value.strip().partition(" ") + if scheme.lower() != "basic" or not token.strip(): + return None + try: + decoded = base64.b64decode(token.strip(), validate=True).decode("utf-8") + except (binascii.Error, UnicodeDecodeError, ValueError): + return None + login, sep, password = decoded.partition(":") + if not sep or not login or not password: + return None + if any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in login + password): + return None + return Credentials(login=login, password=password) + + +def parse_permission_cap(value: str | None) -> PermissionLevel | None: + """Parse ``X-Nextcloud-MCP-Permissions``. Raises ValueError for unknown values.""" + if value is None or not value.strip(): + return None + return PermissionLevel(value.strip().lower()) + + +def lowest(a: PermissionLevel, b: PermissionLevel | None) -> PermissionLevel: + """The lower of two permission levels (``b`` None means no cap).""" + if b is None: + return a + return b if a.includes(b) else a + + +# User-ID lookup for a login: (base config, credentials) -> user ID. Raises LoginRejectedError. +UserIdLookup = Callable[[Config, Credentials], Awaitable[str]] + + +class LoginRejectedError(Exception): + """Nextcloud refused the login (HTTP 401/403).""" + + +class NextcloudUnavailableError(Exception): + """Nextcloud could not be asked (network error, 5xx, unexpected answer).""" + + +async def lookup_user_id(base: Config, creds: Credentials) -> str: + """Ask Nextcloud who the login belongs to (``GET /ocs/v2.php/cloud/user``).""" + url = f"{base.nextcloud_url}/ocs/v2.php/cloud/user" + try: + async with niquests.AsyncSession( + auth=(creds.login, creds.password), + timeout=15, + headers={"OCS-APIRequest": "true", "Accept": "application/json"}, + ) as session: + response = await session.get(url, params={"format": "json"}) + except (OSError, niquests.RequestException) as exc: + raise NextcloudUnavailableError(type(exc).__name__) from None + code = response.status_code or 0 + if code in (401, 403): + raise LoginRejectedError(str(code)) + if not response.ok: + raise NextcloudUnavailableError(f"HTTP {code}") + try: + user_id = response.json()["ocs"]["data"]["id"] + except (ValueError, KeyError, TypeError): + raise NextcloudUnavailableError("unexpected answer") from None + if not isinstance(user_id, str) or not user_id: + raise NextcloudUnavailableError("unexpected answer") + return user_id + + +@dataclasses.dataclass +class _Entry: + client: NextcloudClient + config: Config + last_used: float + + +class ClientPool: + """Clients per login: LRU with an idle TTL. Never shares a client between two logins.""" + + def __init__( + self, + base: Config, + *, + lookup: UserIdLookup = lookup_user_id, + clock: Callable[[], float] = time.monotonic, + close_grace: float = CLOSE_GRACE_SECONDS, + ) -> None: + self._base = base + self._lookup = lookup + self._clock = clock + self._close_grace = close_grace + self._key = secrets.token_bytes(32) + self._entries: OrderedDict[str, _Entry] = OrderedDict() + self._pending: dict[str, asyncio.Future[_Entry]] = {} + # Retired clients waiting out the grace period, so close() can close them right away. + self._closing: dict[asyncio.Task[None], NextcloudClient] = {} + + def _cache_key(self, creds: Credentials) -> str: + message = creds.login.encode("utf-8") + b"\0" + creds.password.encode("utf-8") + return hmac.new(self._key, message, hashlib.sha256).hexdigest() + + def __len__(self) -> int: + return len(self._entries) + + async def get(self, creds: Credentials) -> tuple[NextcloudClient, Config]: + """The client and config for a login; the first use of a login asks Nextcloud for its user ID. + + Raises LoginRejectedError or NextcloudUnavailableError when the login cannot be used. + """ + self._expire() + key = self._cache_key(creds) + entry = self._entries.get(key) + if entry is not None: + entry.last_used = self._clock() + self._entries.move_to_end(key) + return entry.client, entry.config + # Concurrent first requests of one login share a single lookup. + future = self._pending.get(key) + if future is None: + future = asyncio.get_running_loop().create_future() + self._pending[key] = future + try: + user_id = await self._lookup(self._base, creds) + config = dataclasses.replace( + self._base, + user=user_id, + login=creds.login, + password=creds.password, + is_app_password=True, + ) + entry = _Entry(client=NextcloudClient(config), config=config, last_used=self._clock()) + except BaseException as exc: + future.set_exception(exc) + # Retrieve it so an unawaited future does not log "exception was never retrieved". + future.exception() + raise + finally: + self._pending.pop(key, None) + future.set_result(entry) + self._entries[key] = entry + self._evict_overflow() + return entry.client, entry.config + entry = await asyncio.shield(future) + return entry.client, entry.config + + def _expire(self) -> None: + cutoff = self._clock() - self._base.multiuser_ttl + for key in [k for k, e in self._entries.items() if e.last_used <= cutoff]: + self._retire(self._entries.pop(key)) + + def _evict_overflow(self) -> None: + while len(self._entries) > self._base.multiuser_cache_size: + _, entry = self._entries.popitem(last=False) + self._retire(entry) + + def _retire(self, entry: _Entry) -> None: + async def close_later() -> None: + await asyncio.sleep(self._close_grace) + await entry.client.close() + + try: + task = asyncio.get_running_loop().create_task(close_later()) + except RuntimeError: + return + self._closing[task] = entry.client + task.add_done_callback(lambda t: self._closing.pop(t, None)) + + async def close(self) -> None: + """Close every client now, cached or retired (server shutdown).""" + clients = list(self._closing.values()) + for task in list(self._closing): + task.cancel() + self._closing.clear() + clients.extend(entry.client for entry in self._entries.values()) + self._entries = OrderedDict() + for client in clients: + try: + await client.close() + except Exception: + log.warning("Closing a Nextcloud client failed", exc_info=True) + + +# The binding of the running request. Set by the middleware (or bind() in tests and scripts). +_binding: contextvars.ContextVar[Binding | None] = contextvars.ContextVar("nc_mcp_binding", default=None) +_SCOPE_KEY = "nc_mcp_binding" + + +def current_binding() -> Binding: + """The binding of the running request. Raises AuthenticationRequiredError without one. + + Looks first at the MCP request context (the Starlette request the tool call came with), then at + the context variable. The MCP layer may run tools in a task created outside the HTTP request, so + the request's own scope is the reliable source. + """ + binding = _binding_from_request_context() + if binding is None: + binding = _binding.get() + if binding is None: + raise AuthenticationRequiredError( + "No Nextcloud login for this call. In multi-user mode every request needs its own " + "'Authorization: Basic ' header." + ) + return binding + + +def _binding_from_request_context() -> Binding | None: + try: + ctx = request_ctx.get() + except LookupError: + return None + request = getattr(ctx, "request", None) + scope = getattr(request, "scope", None) + if not isinstance(scope, dict): + return None + state = cast(dict[str, Any], scope).get("state") + if not isinstance(state, dict): + return None + binding = cast(dict[str, Any], state).get(_SCOPE_KEY) + return binding if isinstance(binding, Binding) else None + + +def current_cap(level: PermissionLevel) -> PermissionLevel: + """``level`` lowered by the running request's permission cap, if a multi-user request is running.""" + binding = _binding_from_request_context() or _binding.get() + return level if binding is None else lowest(level, binding.permission_cap) + + +set_cap_provider(current_cap) + + +@contextlib.contextmanager +def bind(binding: Binding) -> Generator[None]: + """Run code as one login (tests, scripts). The middleware does the same per request.""" + token = _binding.set(binding) + try: + yield + finally: + _binding.reset(token) + + +ASGIApp = Callable[[dict[str, Any], Callable[..., Awaitable[Any]], Callable[..., Awaitable[Any]]], Awaitable[None]] + + +class MultiUserAuthMiddleware: + """ASGI middleware: no valid login, no MCP beyond discovery. Pins the login's client to the request.""" + + def __init__(self, app: ASGIApp, pool: ClientPool, server_level: PermissionLevel) -> None: + self.app = app + self.pool = pool + self.server_level = server_level + + async def __call__(self, scope: dict[str, Any], receive: Callable[..., Any], send: Callable[..., Any]) -> None: + if scope.get("type") != "http": + if scope.get("type") == "lifespan": + send = self._closing_pool_on_shutdown(send) + await self.app(scope, receive, send) + return + headers = _headers(scope) + if "authorization" not in headers: + await self._without_login(scope, receive, send) + return + creds = parse_basic_auth(headers.get("authorization")) + if creds is None: + await _reply(send, 401, "Authentication required: send 'Authorization: Basic '.") + return + try: + cap = parse_permission_cap(headers.get(PERMISSIONS_HEADER)) + except ValueError: + await _reply(send, 400, "Invalid X-Nextcloud-MCP-Permissions: use read, write or destructive.") + return + try: + client, config = await self.pool.get(creds) + except LoginRejectedError: + await _reply(send, 401, "Nextcloud rejected the login.") + return + except NextcloudUnavailableError as exc: + log.warning("Login check failed: Nextcloud unavailable (%s)", exc) + await _reply(send, 502, "Nextcloud is not reachable.") + return + binding = Binding(client=client, config=config, permission_cap=lowest(self.server_level, cap)) + state = scope.setdefault("state", {}) + state[_SCOPE_KEY] = binding + token = _binding.set(binding) + try: + await self.app(scope, receive, send) + finally: + _binding.reset(token) + + def _closing_pool_on_shutdown(self, send: Callable[..., Any]) -> Callable[..., Awaitable[None]]: + """A lifespan ``send`` that closes the pool's clients before the app reports shutdown complete.""" + + async def wrapped(message: dict[str, Any]) -> None: + if message.get("type") == "lifespan.shutdown.complete": + await self.pool.close() + await send(message) + + return wrapped + + async def _without_login( + self, scope: dict[str, Any], receive: Callable[..., Any], send: Callable[..., Any] + ) -> None: + """No Authorization header: only discovery runs (see module docstring), without a client.""" + method = scope.get("method") + if method != "POST": + await _reply(send, 405, "Method not allowed without a login.", allow=b"POST") + return + body = await _read_body(receive, _MAX_DISCOVERY_BODY_BYTES) + if body is None or not is_discovery_body(body): + await _reply(send, 401, "Authentication required: send 'Authorization: Basic '.") + return + await self.app(scope, _replay(body, receive), send) + + +def is_discovery_body(body: bytes) -> bool: + """True if ``body`` is a JSON-RPC message (or non-empty batch) whose methods are all discovery.""" + try: + payload = json.loads(body) + except (ValueError, UnicodeDecodeError): + return False + messages = cast(list[Any], payload) if isinstance(payload, list) else [payload] + if not messages: + return False + return all(isinstance(m, dict) and cast(dict[str, Any], m).get("method") in DISCOVERY_METHODS for m in messages) + + +async def _read_body(receive: Callable[..., Any], limit: int) -> bytes | None: + """Read the whole request body; None if it is larger than ``limit`` or the client went away.""" + chunks: list[bytes] = [] + size = 0 + while True: + message = await receive() + if message.get("type") != "http.request": + return None + chunk = cast(bytes, message.get("body", b"")) + size += len(chunk) + if size > limit: + return None + chunks.append(chunk) + if not message.get("more_body", False): + return b"".join(chunks) + + +def _replay(body: bytes, receive: Callable[..., Any]) -> Callable[..., Awaitable[dict[str, Any]]]: + """A ``receive`` that hands the already read body to the app once, then waits on the real one.""" + sent = False + + async def replay() -> dict[str, Any]: + nonlocal sent + if not sent: + sent = True + return {"type": "http.request", "body": body, "more_body": False} + return cast(dict[str, Any], await receive()) + + return replay + + +def _headers(scope: dict[str, Any]) -> dict[str, str]: + out: dict[str, str] = {} + raw_headers = cast(list[tuple[bytes, bytes]], scope.get("headers") or []) + for raw_name, raw_value in raw_headers: + name = raw_name.decode("latin-1").lower() + # A repeated Authorization or permissions header is ambiguous: refuse rather than pick one. + if name in out and name in ("authorization", PERMISSIONS_HEADER): + out[name] = _AMBIGUOUS + continue + out[name] = raw_value.decode("latin-1") + return out + + +async def _reply(send: Callable[..., Any], status: int, message: str, allow: bytes | None = None) -> None: + body = ('{"error": "' + message.replace('"', "'") + '"}').encode("utf-8") + headers = [(b"content-type", b"application/json"), (b"content-length", str(len(body)).encode("ascii"))] + if status == 401: + headers.append((b"www-authenticate", b'Basic realm="nc-mcp-server", charset="UTF-8"')) + if allow is not None: + headers.append((b"allow", allow)) + await send({"type": "http.response.start", "status": status, "headers": headers}) + await send({"type": "http.response.body", "body": body}) diff --git a/src/nc_mcp_server/permissions.py b/src/nc_mcp_server/permissions.py index 9f8a39c..1c62fe2 100644 --- a/src/nc_mcp_server/permissions.py +++ b/src/nc_mcp_server/permissions.py @@ -54,9 +54,21 @@ def set_permission_level(level: PermissionLevel) -> None: _current_level = level +# Lowers the level for the running call (multi-user mode registers one, see multiuser.py). +_cap_provider: Callable[[PermissionLevel], PermissionLevel] | None = None + + +def set_cap_provider(provider: Callable[[PermissionLevel], PermissionLevel] | None) -> None: + """Register the function that lowers the global level for the running call.""" + global _cap_provider + _cap_provider = provider + + def get_permission_level() -> PermissionLevel: - """Get the current global permission level.""" - return _current_level + """Get the permission level of this call: the global level, lowered by a multi-user request's cap.""" + if _cap_provider is None: + return _current_level + return _cap_provider(_current_level) def require_permission(level: PermissionLevel) -> Callable[[F], F]: diff --git a/src/nc_mcp_server/server.py b/src/nc_mcp_server/server.py index 72b8fe8..9d2f885 100644 --- a/src/nc_mcp_server/server.py +++ b/src/nc_mcp_server/server.py @@ -1,9 +1,12 @@ """MCP server — registers all tools and manages the Nextcloud client lifecycle.""" +from typing import Any + from mcp.server.fastmcp import FastMCP from .client import NextcloudClient from .config import Config +from .multiuser import ClientPool, MultiUserAuthMiddleware from .permissions import set_permission_level from .state import get_client, get_config, set_state from .tools import ( @@ -34,7 +37,7 @@ versions, ) -__all__ = ["create_server", "get_client", "get_config"] +__all__ = ["create_http_app", "create_server", "get_client", "get_config"] def create_server(config: Config | None = None) -> FastMCP: @@ -50,7 +53,8 @@ def create_server(config: Config | None = None) -> FastMCP: config = Config.from_env() config.validate() - set_state(NextcloudClient(config), config) + # Multi-user mode: no client of its own; each request brings its login (multiuser.py). + set_state(None if config.multiuser else NextcloudClient(config), config) set_permission_level(config.permission_level) mcp = FastMCP( @@ -87,3 +91,12 @@ def create_server(config: Config | None = None) -> FastMCP: users.register(mcp) return mcp + + +def create_http_app(mcp: FastMCP, config: Config) -> Any: + """The Streamable HTTP ASGI app; in multi-user mode wrapped in the login middleware.""" + app = mcp.streamable_http_app() + if not config.multiuser: + return app + pool = ClientPool(config) + return MultiUserAuthMiddleware(app, pool, config.permission_level) diff --git a/src/nc_mcp_server/state.py b/src/nc_mcp_server/state.py index f0a4c4d..5ed8b2a 100644 --- a/src/nc_mcp_server/state.py +++ b/src/nc_mcp_server/state.py @@ -1,28 +1,45 @@ -"""Global state — holds the Nextcloud client and config singletons.""" +"""Global state — holds the Nextcloud client and config singletons. + +In multi-user mode (``NEXTCLOUD_MCP_MULTIUSER=true``) there is no global client: ``get_client`` and +``get_config`` return the client and config of the login that sent the running request +(see :mod:`nc_mcp_server.multiuser`). +""" from .client import NextcloudClient from .config import Config +from .multiuser import current_binding _client: NextcloudClient | None = None _config: Config | None = None def get_client() -> NextcloudClient: - """Get the global Nextcloud client. Raises if server not initialized.""" + """Get the Nextcloud client of this call. Raises if server not initialized.""" + if _config is not None and _config.multiuser: + return current_binding().client if _client is None: raise RuntimeError("Server not initialized. Call create_server() first.") return _client def get_config() -> Config: - """Get the global config. Raises if server not initialized.""" + """Get the config of this call. Raises if server not initialized.""" + if _config is None: + raise RuntimeError("Server not initialized. Call create_server() first.") + if _config.multiuser: + return current_binding().config + return _config + + +def get_server_config() -> Config: + """The server's own config (in multi-user mode: without any login).""" if _config is None: raise RuntimeError("Server not initialized. Call create_server() first.") return _config -def set_state(client: NextcloudClient, config: Config) -> None: - """Set the global state. Called once at server startup.""" +def set_state(client: NextcloudClient | None, config: Config) -> None: + """Set the global state. Called once at server startup (client is None in multi-user mode).""" global _client, _config _client = client _config = config diff --git a/src/nc_mcp_server/tools/files.py b/src/nc_mcp_server/tools/files.py index 3f382e4..f3661cf 100644 --- a/src/nc_mcp_server/tools/files.py +++ b/src/nc_mcp_server/tools/files.py @@ -23,7 +23,7 @@ ) from ..client import NextcloudClient from ..permissions import PermissionLevel, require_permission -from ..state import get_client, get_config +from ..state import get_client, get_config, get_server_config _IMAGE_MIME_TYPES = {"image/png", "image/jpeg", "image/gif", "image/webp", "image/bmp", "image/svg+xml"} _MAX_IMAGE_SIZE = 10 * 1024 * 1024 @@ -445,5 +445,5 @@ def register(mcp: FastMCP) -> None: _register_read_tools(mcp) _register_write_tools(mcp) _register_destructive_tools(mcp) - if get_config().upload_root: + if get_server_config().upload_root: _register_upload_from_path_tool(mcp) diff --git a/tests/integration/test_multiuser.py b/tests/integration/test_multiuser.py new file mode 100644 index 0000000..c2a82f3 --- /dev/null +++ b/tests/integration/test_multiuser.py @@ -0,0 +1,167 @@ +"""Multi-user mode against a real Nextcloud: two logins in parallel, never swapped; no login, no access. + +Runs the real Streamable HTTP app with the login middleware (as ``--transport http`` does with +NEXTCLOUD_MCP_MULTIUSER=true). The users get app passwords from Nextcloud itself. +""" + +import base64 +import contextlib +import json +import secrets +import threading +from collections.abc import AsyncGenerator +from typing import Any + +import niquests +import pytest +from starlette.testclient import TestClient + +import nc_mcp_server.state as state_module +from nc_mcp_server.client import NextcloudClient +from nc_mcp_server.config import Config +from nc_mcp_server.permissions import PermissionLevel +from nc_mcp_server.server import create_http_app, create_server + +from .conftest import _get_integration_config + +pytestmark = pytest.mark.integration + + +class User: + def __init__(self, user_id: str, login: str, app_password: str) -> None: + self.user_id = user_id + self.login = login + self.app_password = app_password + self.marker = f"marker-{user_id}.txt" + + def header(self) -> dict[str, str]: + token = base64.b64encode(f"{self.login}:{self.app_password}".encode()).decode() + return {"Authorization": f"Basic {token}"} + + +async def _app_password(base_url: str, login: str, password: str) -> str: + async with niquests.AsyncSession(auth=(login, password), headers={"OCS-APIRequest": "true"}) as session: + response = await session.get(f"{base_url}/ocs/v2.php/core/getapppassword", params={"format": "json"}) + assert response.ok, f"getapppassword: HTTP {response.status_code}" + return response.json()["ocs"]["data"]["apppassword"] + + +@pytest.fixture +async def two_users() -> AsyncGenerator[tuple[User, User]]: + """alice logs in with her user ID, bob with his e-mail address (login name != user ID).""" + admin_config = _get_integration_config() + admin = NextcloudClient(admin_config) + base = admin_config.nextcloud_url + suffix = secrets.token_hex(4) + created: list[str] = [] + users: list[User] = [] + try: + for name, by_mail in (("mu-alice", False), ("mu-bob", True)): + user_id = f"{name}-{suffix}" + password = f"Mcp-{secrets.token_hex(10)}!" + email = f"{user_id}@example.com" + await admin.ocs_post("cloud/users", data={"userid": user_id, "password": password, "email": email}) + created.append(user_id) + login = email if by_mail else user_id + app_password = await _app_password(base, login, password) + user = User(user_id, login, app_password) + own = NextcloudClient( + Config(nextcloud_url=base, user=user_id, login=login, password=app_password, is_app_password=True) + ) + try: + await own.dav_put(user.marker, user_id.encode(), content_type="text/plain") + finally: + await own.close() + users.append(user) + yield users[0], users[1] + finally: + for user_id in created: + with contextlib.suppress(Exception): # best-effort cleanup + await admin.ocs_delete(f"cloud/users/{user_id}") + await admin.close() + + +@pytest.fixture +def mu_client() -> Any: + base = _get_integration_config() + config = Config(nextcloud_url=base.nextcloud_url, multiuser=True, permission_level=PermissionLevel.DESTRUCTIVE) + config.validate() + mcp = create_server(config) + with TestClient(create_http_app(mcp, config)) as client: + yield client + state_module.set_state(None, Config()) + + +def _call(client: TestClient, tool: str, arguments: dict[str, Any], headers: dict[str, str]) -> tuple[int, Any]: + body = {"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": tool, "arguments": arguments}} + response = client.post( + "/mcp", + json=body, + headers={"Accept": "application/json, text/event-stream", **headers}, + ) + if response.status_code != 200: + return response.status_code, None + for line in response.text.splitlines(): + if line.startswith("data:"): + return 200, json.loads(line[5:]) + return 200, response.json() + + +def _names(payload: Any) -> set[str]: + assert payload["result"].get("isError") is not True, payload + data = json.loads(payload["result"]["content"][0]["text"]) + return {entry["path"].strip("/") for entry in data["data"]} + + +class TestMultiUser: + def test_no_header_is_401(self, mu_client: TestClient) -> None: + status, _ = _call(mu_client, "list_directory", {"path": "/"}, {}) + assert status == 401 + + def test_wrong_app_password_is_401(self, mu_client: TestClient, two_users: tuple[User, User]) -> None: + alice, _ = two_users + wrong = User(alice.user_id, alice.login, "wrong-" + secrets.token_hex(8)) + status, _ = _call(mu_client, "list_directory", {"path": "/"}, wrong.header()) + assert status == 401 + + def test_each_login_sees_own_files(self, mu_client: TestClient, two_users: tuple[User, User]) -> None: + for user in two_users: + status, payload = _call(mu_client, "list_directory", {"path": "/"}, user.header()) + assert status == 200 + names = _names(payload) + others = {u.marker for u in two_users if u is not user} + assert user.marker in names + assert not names & others + + def test_parallel_logins_never_swap(self, mu_client: TestClient, two_users: tuple[User, User]) -> None: + errors: list[str] = [] + + def worker(user: User, other: User) -> None: + for _ in range(8): + status, payload = _call(mu_client, "list_directory", {"path": "/"}, user.header()) + if status != 200: + errors.append(f"{user.user_id}: HTTP {status}") + continue + names = _names(payload) + if user.marker not in names or other.marker in names: + errors.append(f"{user.user_id} saw {sorted(names)}") + + alice, bob = two_users + threads = [ + threading.Thread(target=worker, args=args) + for args in ((alice, bob), (bob, alice), (alice, bob), (bob, alice)) + ] + for t in threads: + t.start() + for t in threads: + t.join() + assert errors == [] + + def test_read_cap_blocks_writes(self, mu_client: TestClient, two_users: tuple[User, User]) -> None: + alice, _ = two_users + headers = {**alice.header(), "X-Nextcloud-MCP-Permissions": "read"} + status, payload = _call(mu_client, "create_directory", {"path": "mu-should-not-exist"}, headers) + assert status == 200 + assert payload["result"]["isError"] is True + status, payload = _call(mu_client, "list_directory", {"path": "/"}, alice.header()) + assert "mu-should-not-exist" not in _names(payload) diff --git a/tests/test_multiuser.py b/tests/test_multiuser.py new file mode 100644 index 0000000..6c6ccae --- /dev/null +++ b/tests/test_multiuser.py @@ -0,0 +1,537 @@ +"""Tests for multi-user mode (NEXTCLOUD_MCP_MULTIUSER): header parsing, client pool, middleware.""" + +import asyncio +import base64 +import json +import threading +from typing import Any + +import pytest +from mcp.server.fastmcp import FastMCP +from starlette.testclient import TestClient + +import nc_mcp_server.state as state_module +from nc_mcp_server.config import Config +from nc_mcp_server.multiuser import ( + AuthenticationRequiredError, + Binding, + ClientPool, + Credentials, + LoginRejectedError, + NextcloudUnavailableError, + bind, + lowest, + parse_basic_auth, + parse_permission_cap, +) +from nc_mcp_server.permissions import PermissionLevel, get_permission_level, require_permission +from nc_mcp_server.server import create_http_app, create_server +from nc_mcp_server.state import get_client, get_config + + +def basic(login: str, password: str) -> str: + return "Basic " + base64.b64encode(f"{login}:{password}".encode()).decode() + + +def mu_config(**overrides: Any) -> Config: + values: dict[str, Any] = { + "nextcloud_url": "http://nc.invalid", + "multiuser": True, + "permission_level": PermissionLevel.DESTRUCTIVE, + } + values.update(overrides) + return Config(**values) + + +class TestParseBasicAuth: + def test_valid(self) -> None: + creds = parse_basic_auth(basic("alice", "pw:with:colons")) + assert creds == Credentials("alice", "pw:with:colons") + + def test_scheme_case_insensitive(self) -> None: + assert parse_basic_auth("basic " + base64.b64encode(b"a:b").decode()) == Credentials("a", "b") + + @pytest.mark.parametrize( + "value", + [ + None, + "", + "Bearer abc", + "Basic", + "Basic !!!notbase64", + "Basic " + base64.b64encode(b"no-colon").decode(), + "Basic " + base64.b64encode(b":pw").decode(), + "Basic " + base64.b64encode(b"user:").decode(), + "Basic " + base64.b64encode(b"us\ner:pw").decode(), + "Basic " + base64.b64encode(b"\xff\xfe:pw").decode(), + "Basic " + "A" * 5000, + ], + ) + def test_rejects(self, value: str | None) -> None: + assert parse_basic_auth(value) is None + + def test_repr_hides_password(self) -> None: + assert "secret" not in repr(Credentials("alice", "secret")) + + +class TestPermissionCap: + def test_parse(self) -> None: + assert parse_permission_cap(None) is None + assert parse_permission_cap(" ") is None + assert parse_permission_cap("READ") is PermissionLevel.READ + + def test_parse_invalid(self) -> None: + with pytest.raises(ValueError, match="admin"): + parse_permission_cap("admin") + + def test_lowest_never_raises(self) -> None: + assert lowest(PermissionLevel.READ, PermissionLevel.DESTRUCTIVE) is PermissionLevel.READ + assert lowest(PermissionLevel.DESTRUCTIVE, PermissionLevel.READ) is PermissionLevel.READ + assert lowest(PermissionLevel.WRITE, None) is PermissionLevel.WRITE + + +class TestConfig: + def test_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("NEXTCLOUD_URL", "http://cloud.example.com") + monkeypatch.setenv("NEXTCLOUD_MCP_MULTIUSER", "true") + monkeypatch.setenv("NEXTCLOUD_MCP_MULTIUSER_CACHE_SIZE", "5") + monkeypatch.setenv("NEXTCLOUD_MCP_MULTIUSER_TTL", "30") + monkeypatch.delenv("NEXTCLOUD_USER", raising=False) + monkeypatch.delenv("NEXTCLOUD_PASSWORD", raising=False) + config = Config.from_env() + assert config.multiuser is True + assert config.multiuser_cache_size == 5 + assert config.multiuser_ttl == 30.0 + config.validate() + + def test_default_off(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("NEXTCLOUD_MCP_MULTIUSER", raising=False) + assert Config.from_env().multiuser is False + + @pytest.mark.parametrize( + ("name", "value"), [("NEXTCLOUD_MCP_MULTIUSER", "maybe"), ("NEXTCLOUD_MCP_MULTIUSER_TTL", "0")] + ) + def test_invalid_values(self, monkeypatch: pytest.MonkeyPatch, name: str, value: str) -> None: + monkeypatch.setenv(name, value) + with pytest.raises(ValueError, match=name): + Config.from_env() + + def test_validate_refuses_fallback_account(self) -> None: + with pytest.raises(ValueError, match="unset NEXTCLOUD_USER"): + mu_config(user="admin", password="admin").validate() + + def test_validate_needs_url(self) -> None: + with pytest.raises(ValueError, match="NEXTCLOUD_URL"): + mu_config(nextcloud_url="").validate() + + def test_auth_login(self) -> None: + assert Config(user="uid").auth_login == "uid" + assert Config(user="uid", login="a@example.com").auth_login == "a@example.com" + + +class FakeLookup: + """User-ID lookup without Nextcloud: login 'bad' is rejected, 'down' is unavailable.""" + + def __init__(self, delay: float = 0.0) -> None: + self.calls: list[str] = [] + self.delay = delay + + async def __call__(self, base: Config, creds: Credentials) -> str: + self.calls.append(creds.login) + if self.delay: + await asyncio.sleep(self.delay) + if creds.login == "bad": + raise LoginRejectedError("401") + if creds.login == "down": + raise NextcloudUnavailableError("HTTP 503") + return "uid-" + creds.login.split("@")[0] + + +class FakeClock: + def __init__(self) -> None: + self.t = 1000.0 + + def __call__(self) -> float: + return self.t + + +class TestClientPool: + async def test_one_client_per_login(self) -> None: + pool = ClientPool(mu_config(), lookup=FakeLookup()) + a1, ca = await pool.get(Credentials("alice", "pa")) + b1, cb = await pool.get(Credentials("bob", "pb")) + a2, _ = await pool.get(Credentials("alice", "pa")) + assert a1 is a2 + assert a1 is not b1 + assert (ca.user, ca.auth_login, ca.password) == ("uid-alice", "alice", "pa") + assert (cb.user, cb.auth_login, cb.password) == ("uid-bob", "bob", "pb") + assert ca.is_app_password is True + await pool.close() + + async def test_other_password_is_other_client(self) -> None: + pool = ClientPool(mu_config(), lookup=FakeLookup()) + a1, _ = await pool.get(Credentials("alice", "old")) + a2, c2 = await pool.get(Credentials("alice", "new")) + assert a1 is not a2 + assert c2.password == "new" + await pool.close() + + async def test_login_name_differs_from_user_id(self) -> None: + pool = ClientPool(mu_config(), lookup=FakeLookup()) + _, config = await pool.get(Credentials("alice@example.com", "pa")) + assert config.user == "uid-alice" + assert config.auth_login == "alice@example.com" + await pool.close() + + async def test_concurrent_first_use_shares_lookup(self) -> None: + lookup = FakeLookup(delay=0.05) + pool = ClientPool(mu_config(), lookup=lookup) + results = await asyncio.gather(*(pool.get(Credentials("alice", "pa")) for _ in range(5))) + assert lookup.calls == ["alice"] + assert len({id(c) for c, _ in results}) == 1 + await pool.close() + + async def test_rejected_login_not_cached(self) -> None: + lookup = FakeLookup() + pool = ClientPool(mu_config(), lookup=lookup) + for _ in range(2): + with pytest.raises(LoginRejectedError): + await pool.get(Credentials("bad", "x")) + assert lookup.calls == ["bad", "bad"] + assert len(pool) == 0 + await pool.close() + + async def test_lru_eviction(self) -> None: + lookup = FakeLookup() + pool = ClientPool(mu_config(multiuser_cache_size=2), lookup=lookup, close_grace=0) + await pool.get(Credentials("a", "1")) + await pool.get(Credentials("b", "1")) + await pool.get(Credentials("a", "1")) # a is now the most recent + await pool.get(Credentials("c", "1")) # evicts b + assert len(pool) == 2 + await pool.get(Credentials("a", "1")) + await pool.get(Credentials("b", "1")) + assert lookup.calls == ["a", "b", "c", "b"] + await pool.close() + + async def test_idle_ttl(self) -> None: + clock = FakeClock() + lookup = FakeLookup() + pool = ClientPool(mu_config(multiuser_ttl=60), lookup=lookup, clock=clock, close_grace=0) + await pool.get(Credentials("a", "1")) + clock.t += 30 + await pool.get(Credentials("a", "1")) # refreshes last use + clock.t += 59 + await pool.get(Credentials("a", "1")) + clock.t += 61 + await pool.get(Credentials("a", "1")) + assert lookup.calls == ["a", "a"] + await pool.close() + + async def test_evicted_client_closed_after_grace(self) -> None: + pool = ClientPool(mu_config(multiuser_cache_size=1), lookup=FakeLookup(), close_grace=0) + client, _ = await pool.get(Credentials("a", "1")) + closed = asyncio.Event() + original = client.close + + async def close() -> None: + closed.set() + await original() + + client.close = close # type: ignore[method-assign] + await pool.get(Credentials("b", "1")) + await asyncio.wait_for(closed.wait(), 1) + await pool.close() + + async def test_close_closes_cached_and_retired_clients(self) -> None: + pool = ClientPool(mu_config(multiuser_cache_size=1), lookup=FakeLookup(), close_grace=3600) + closed: list[str] = [] + for login in ("a", "b"): + client, _ = await pool.get(Credentials(login, "1")) + original = client.close + + async def close(name: str = login, original: Any = original) -> None: + closed.append(name) + await original() + + client.close = close # type: ignore[method-assign] + assert len(pool) == 1 # "a" was evicted and waits out the grace period + await pool.close() + assert sorted(closed) == ["a", "b"] + assert len(pool) == 0 + + +class TestStateInMultiUserMode: + def test_no_call_without_login(self) -> None: + create_server(mu_config()) + try: + with pytest.raises(AuthenticationRequiredError): + get_client() + with pytest.raises(AuthenticationRequiredError): + get_config() + finally: + state_module.set_state(None, Config()) + + async def test_bind(self) -> None: + create_server(mu_config()) + pool = ClientPool(mu_config(), lookup=FakeLookup()) + try: + client, config = await pool.get(Credentials("alice", "pa")) + with bind(Binding(client=client, config=config, permission_cap=PermissionLevel.READ)): + assert get_client() is client + assert get_config().user == "uid-alice" + assert get_permission_level() is PermissionLevel.READ + with pytest.raises(AuthenticationRequiredError): + get_client() + finally: + await pool.close() + state_module.set_state(None, Config()) + + +def _add_probe_tools(mcp: FastMCP) -> None: + @mcp.tool() + async def probe_whoami() -> str: + """Test tool: which login does this call run as?""" + config = get_config() + await asyncio.sleep(0.01) # let concurrent requests interleave + return json.dumps({"user": config.user, "login": config.auth_login, "client": id(get_client())}) + + @mcp.tool() + @require_permission(PermissionLevel.WRITE) + async def probe_write() -> str: + """Test tool that needs write permission.""" + return "written" + + +def _call(client: TestClient, tool: str, headers: dict[str, str]) -> tuple[int, Any]: + body = {"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": tool, "arguments": {}}} + response = client.post( + "/mcp", + json=body, + headers={"Accept": "application/json, text/event-stream", "Content-Type": "application/json", **headers}, + ) + if response.status_code != 200: + return response.status_code, response.json() + for line in response.text.splitlines(): + if line.startswith("data:"): + return 200, json.loads(line[5:]) + return 200, response.json() + + +def _result_text(payload: Any) -> str: + return payload["result"]["content"][0]["text"] + + +@pytest.fixture +def mu_app_lookup(monkeypatch: pytest.MonkeyPatch) -> Any: + config = mu_config() + mcp = create_server(config) + _add_probe_tools(mcp) + lookup = FakeLookup() + monkeypatch.setattr("nc_mcp_server.server.ClientPool", lambda cfg: ClientPool(cfg, lookup=lookup)) # pyright: ignore[reportUnknownLambdaType] + app = create_http_app(mcp, config) + with TestClient(app) as client: + yield client, lookup + state_module.set_state(None, Config()) + + +@pytest.fixture +def mu_app(mu_app_lookup: Any) -> TestClient: + return mu_app_lookup[0] + + +class TestMiddleware: + def test_missing_header_is_401(self, mu_app: TestClient) -> None: + response = mu_app.post("/mcp", json={}) + assert response.status_code == 401 + assert response.headers["www-authenticate"].startswith("Basic") + + def test_malformed_header_is_401(self, mu_app: TestClient) -> None: + status, _ = _call(mu_app, "probe_whoami", {"Authorization": "Bearer xyz"}) + assert status == 401 + + def test_rejected_login_is_401(self, mu_app: TestClient) -> None: + status, _ = _call(mu_app, "probe_whoami", {"Authorization": basic("bad", "x")}) + assert status == 401 + + def test_nextcloud_down_is_502(self, mu_app: TestClient) -> None: + status, _ = _call(mu_app, "probe_whoami", {"Authorization": basic("down", "x")}) + assert status == 502 + + def test_invalid_permission_header_is_400(self, mu_app: TestClient) -> None: + status, _ = _call( + mu_app, "probe_whoami", {"Authorization": basic("alice", "pa"), "X-Nextcloud-MCP-Permissions": "root"} + ) + assert status == 400 + + def test_duplicate_authorization_is_401(self, mu_app: TestClient) -> None: + body = {"jsonrpc": "2.0", "id": 1, "method": "tools/list"} + response = mu_app.post( + "/mcp", + json=body, + headers=[ # pyright: ignore[reportArgumentType] + ("Authorization", basic("alice", "pa")), + ("Authorization", basic("bob", "pb")), + ("Accept", "application/json, text/event-stream"), + ], + ) + assert response.status_code == 401 + + def test_tool_runs_as_request_login(self, mu_app: TestClient) -> None: + status, payload = _call(mu_app, "probe_whoami", {"Authorization": basic("alice@example.com", "pa")}) + assert status == 200 + assert json.loads(_result_text(payload))["user"] == "uid-alice" + assert json.loads(_result_text(payload))["login"] == "alice@example.com" + + def test_permission_cap_lowers_level(self, mu_app: TestClient) -> None: + status, payload = _call(mu_app, "probe_write", {"Authorization": basic("alice", "pa")}) + assert (status, _result_text(payload)) == (200, "written") + status, payload = _call( + mu_app, "probe_write", {"Authorization": basic("alice", "pa"), "X-Nextcloud-MCP-Permissions": "read"} + ) + assert status == 200 + assert payload["result"]["isError"] is True + assert "requires 'write' permission" in _result_text(payload) + + def test_parallel_logins_never_swap(self, mu_app: TestClient) -> None: + errors: list[str] = [] + seen: dict[str, set[int]] = {"alice": set(), "bob": set()} + + def worker(login: str) -> None: + for _ in range(15): + status, payload = _call(mu_app, "probe_whoami", {"Authorization": basic(login, "pw-" + login)}) + if status != 200: + errors.append(f"{login}: HTTP {status}") + continue + got = json.loads(_result_text(payload)) + if got["user"] != "uid-" + login or got["login"] != login: + errors.append(f"{login} got {got}") + seen[login].add(got["client"]) + + threads = [threading.Thread(target=worker, args=(name,)) for name in ("alice", "bob", "alice", "bob")] + for t in threads: + t.start() + for t in threads: + t.join() + assert errors == [] + assert len(seen["alice"]) == 1 + assert len(seen["bob"]) == 1 + assert seen["alice"] != seen["bob"] + + +_MCP_HEADERS = {"Accept": "application/json, text/event-stream", "Content-Type": "application/json"} +_INIT = { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "t", "version": "0"}}, +} + + +def _payload(response: Any) -> Any: + for line in response.text.splitlines(): + if line.startswith("data:"): + return json.loads(line[5:]) + return response.json() + + +class TestLifespan: + def test_shutdown_closes_pool(self, monkeypatch: pytest.MonkeyPatch) -> None: + config = mu_config() + mcp = create_server(config) + _add_probe_tools(mcp) + pools: list[ClientPool] = [] + + def make_pool(cfg: Config) -> ClientPool: + pool = ClientPool(cfg, lookup=FakeLookup()) + pools.append(pool) + return pool + + monkeypatch.setattr("nc_mcp_server.server.ClientPool", make_pool) + closed = threading.Event() + try: + with TestClient(create_http_app(mcp, config)) as client: + status, _ = _call(client, "probe_whoami", {"Authorization": basic("alice", "pa")}) + assert status == 200 + assert len(pools[0]) == 1 + + original = pools[0].close + + async def close() -> None: + closed.set() + await original() + + pools[0].close = close # type: ignore[method-assign] + assert closed.is_set() + assert len(pools[0]) == 0 + finally: + state_module.set_state(None, Config()) + + +class TestDiscoveryWithoutLogin: + def test_initialize_and_tools_list_without_login(self, mu_app_lookup: Any) -> None: + client, lookup = mu_app_lookup + response = client.post("/mcp", json=_INIT, headers=_MCP_HEADERS) + assert response.status_code == 200 + assert _payload(response)["result"]["serverInfo"]["name"] == "nc-mcp-server" + note = {"jsonrpc": "2.0", "method": "notifications/initialized"} + assert client.post("/mcp", json=note, headers=_MCP_HEADERS).status_code == 202 + response = client.post("/mcp", json={"jsonrpc": "2.0", "id": 2, "method": "tools/list"}, headers=_MCP_HEADERS) + assert response.status_code == 200 + names = {t["name"] for t in _payload(response)["result"]["tools"]} + assert {"list_directory", "get_file", "probe_whoami"} <= names + ping = client.post("/mcp", json={"jsonrpc": "2.0", "id": 3, "method": "ping"}, headers=_MCP_HEADERS) + assert ping.status_code == 200 + assert lookup.calls == [] # no Nextcloud call, no client + + def test_tools_call_without_login_is_401(self, mu_app_lookup: Any) -> None: + client, lookup = mu_app_lookup + status, _ = _call(client, "probe_whoami", {}) + assert status == 401 + assert lookup.calls == [] + + @pytest.mark.parametrize( + "body", + [ + [{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}, {"jsonrpc": "2.0", "id": 2, "method": "tools/call"}], + [], + {"jsonrpc": "2.0", "id": 1, "method": "resources/read", "params": {"uri": "x"}}, + {"jsonrpc": "2.0", "id": 1}, + "tools/list", + ], + ) + def test_other_bodies_without_login_are_401(self, mu_app_lookup: Any, body: Any) -> None: + client, _ = mu_app_lookup + response = client.post("/mcp", json=body, headers=_MCP_HEADERS) + assert response.status_code == 401 + + def test_invalid_json_without_login_is_401(self, mu_app_lookup: Any) -> None: + client, _ = mu_app_lookup + response = client.post("/mcp", content=b"{not json", headers=_MCP_HEADERS) + assert response.status_code == 401 + + def test_oversized_body_without_login_is_401(self, mu_app_lookup: Any) -> None: + client, _ = mu_app_lookup + body = {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {"pad": "x" * 70_000}} + response = client.post("/mcp", json=body, headers=_MCP_HEADERS) + assert response.status_code == 401 + + def test_get_and_delete_without_login_are_405(self, mu_app_lookup: Any) -> None: + client, _ = mu_app_lookup + for method in ("GET", "DELETE"): + response = client.request(method, "/mcp", headers=_MCP_HEADERS) + assert response.status_code == 405 + assert response.headers["allow"] == "POST" + + @pytest.mark.parametrize("auth", ["", "Bearer xyz", "Basic !!!"]) + def test_wrong_header_is_never_downgraded_to_discovery(self, mu_app_lookup: Any, auth: str) -> None: + client, _ = mu_app_lookup + body = {"jsonrpc": "2.0", "id": 1, "method": "tools/list"} + response = client.post("/mcp", json=body, headers={**_MCP_HEADERS, "Authorization": auth}) + assert response.status_code == 401 + + def test_rejected_login_is_not_downgraded(self, mu_app_lookup: Any) -> None: + client, _ = mu_app_lookup + body = {"jsonrpc": "2.0", "id": 1, "method": "tools/list"} + response = client.post("/mcp", json=body, headers={**_MCP_HEADERS, "Authorization": basic("bad", "x")}) + assert response.status_code == 401