"""WebDAV-backed MCP server, scoped to one configured vault directory.""" from __future__ import annotations import json import os import xml.etree.ElementTree as ET from collections import deque from contextlib import asynccontextmanager from pathlib import Path from urllib.parse import quote, unquote, urljoin, urlsplit import httpx from mcp.server import MCPServer from mcp.server.auth.middleware.auth_context import get_access_token from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions, RevocationOptions from mcp.server.auth.routes import create_protected_resource_routes from mcp.server.transport_security import RequestBodyLimitMiddleware, TransportSecuritySettings from mcp.types import ToolAnnotations from pydantic import AnyHttpUrl from starlette.applications import Starlette from starlette.requests import Request from starlette.responses import JSONResponse from starlette.routing import Mount, Route from oauth import LocalOAuthProvider, make_login_routes DAV_NS = "{DAV:}" MAX_FILE_BYTES = 256 * 1024 MAX_PROPFIND_BYTES = 2 * 1024 * 1024 MAX_SEARCH_DIRECTORIES = 500 MAX_SEARCH_FILES = 1500 HTTP_TIMEOUT = httpx.Timeout(30.0, connect=10.0) class WebDAVError(Exception): pass def _secret(name: str, *, required: bool = True) -> str: file_name = os.getenv(f"{name}_FILE") if file_name: try: value = Path(file_name).read_bytes().rstrip(b"\r\n").decode("utf-8") except OSError as exc: raise RuntimeError(f"Cannot read secret file configured by {name}_FILE") from exc else: value = os.getenv(name, "") if required and not value: raise RuntimeError(f"Set {name} or {name}_FILE before starting the service") return value def _https_origin(name: str, default: str) -> str: value = os.getenv(name, default).strip().rstrip("/") parsed = urlsplit(value) if parsed.scheme != "https" and parsed.hostname not in {"localhost", "127.0.0.1", "::1"}: raise RuntimeError(f"{name} must use HTTPS") if not parsed.scheme or not parsed.netloc or parsed.username or parsed.password or parsed.query or parsed.fragment: raise RuntimeError(f"{name} must be an absolute URL without credentials, query, or fragment") return value WEBDAV_URL = _https_origin("WEBDAV_URL", "https://dav.ryankegel.com") WEBDAV_USERNAME = os.getenv("WEBDAV_USERNAME", "").strip() WEBDAV_PASSWORD = _secret("WEBDAV_PASSWORD") if not WEBDAV_USERNAME: raise RuntimeError("Set WEBDAV_USERNAME before starting the service") WEBDAV_ROOT = os.getenv("WEBDAV_ROOT", "/Notes/Personal") if not WEBDAV_ROOT.startswith("/"): raise RuntimeError("WEBDAV_ROOT must be an absolute WebDAV path, for example /Notes/Personal") _root_raw = WEBDAV_ROOT.strip("/") _root_parts = _root_raw.split("/") if _root_raw else [] if not _root_parts or any(part in {"", ".", ".."} for part in _root_parts): raise RuntimeError("WEBDAV_ROOT must be a non-root path without dot segments") for _part in _root_parts: _decoded_part = _part for _ in range(4): _decoded_part = unquote(_decoded_part) if _decoded_part in {".", ".."} or "/" in _decoded_part or "\\" in _decoded_part: raise RuntimeError("WEBDAV_ROOT cannot contain encoded path separators or dot segments") if _decoded_part == _part: break MCP_PUBLIC_URL = _https_origin("MCP_PUBLIC_URL", "https://mcp.example.com") if urlsplit(MCP_PUBLIC_URL).path not in {"", "/"}: raise RuntimeError("MCP_PUBLIC_URL must be the HTTPS origin only, without a path") MCP_RESOURCE_URL = f"{MCP_PUBLIC_URL}/mcp" MCP_AUTH_USERNAME = os.getenv("MCP_AUTH_USERNAME", "").strip() MCP_AUTH_PASSWORD = _secret("MCP_AUTH_PASSWORD") if not MCP_AUTH_USERNAME: raise RuntimeError("Set MCP_AUTH_USERNAME before starting the service") if len(MCP_AUTH_PASSWORD.encode("utf-8")) < 16: raise RuntimeError("MCP_AUTH_PASSWORD must contain at least 16 bytes") MCP_ALLOWED_HOSTS = [item.strip() for item in os.getenv("MCP_ALLOWED_HOSTS", "").split(",") if item.strip()] if not MCP_ALLOWED_HOSTS: raise RuntimeError("Set MCP_ALLOWED_HOSTS to the public MCP host name") MCP_ALLOWED_ORIGINS = [item.strip() for item in os.getenv("MCP_ALLOWED_ORIGINS", "").split(",") if item.strip()] READ_ONLY = os.getenv("MCP_READ_ONLY", "true").strip().lower() not in {"0", "false", "no"} DATABASE_PATH = os.getenv("OAUTH_DATABASE_PATH", "/data/oauth.sqlite3") OAUTH_SCOPES = ["notes:read"] + ([] if READ_ONLY else ["notes:write"]) _root_encoded = "/".join(quote(part, safe="-._~") for part in _root_parts) _WEBDAV_ROOT_URL = f"{WEBDAV_URL.rstrip('/')}/{_root_encoded}" _DAV_ORIGIN = urlsplit(WEBDAV_URL) _ROOT_PATH_UNQUOTED = unquote(urlsplit(_WEBDAV_ROOT_URL).path).rstrip("/") def _clean_relative_path(path: str, *, allow_empty: bool = True) -> list[str]: if not isinstance(path, str): raise WebDAVError("Path must be text") if len(path) > 2048: raise WebDAVError("Path is too long") if path.startswith("/") or "\\" in path or "\x00" in path: raise WebDAVError("Use a relative path inside the configured vault") raw = path[:-1] if path.endswith("/") else path if not raw: if allow_empty: return [] raise WebDAVError("A file path is required") parts = raw.split("/") if any(not part for part in parts): raise WebDAVError("Path cannot contain empty segments") for part in parts: decoded = part for _ in range(4): decoded = unquote(decoded) if decoded in {".", ".."} or "/" in decoded or "\\" in decoded: raise WebDAVError("Path cannot escape the configured vault") if decoded == part: break if any(ord(char) < 32 or ord(char) == 127 for char in decoded): raise WebDAVError("Path cannot contain control characters") return parts def _path_url(path: str, *, collection: bool = False) -> str: parts = _clean_relative_path(path) encoded = "/".join(quote(part, safe="-._~") for part in parts) url = _WEBDAV_ROOT_URL + (f"/{encoded}" if encoded else "") if collection: return url.rstrip("/") + "/" return url def _same_origin(url: str) -> bool: parsed = urlsplit(url) try: base_port = _DAV_ORIGIN.port or (443 if _DAV_ORIGIN.scheme.lower() == "https" else 80) parsed_port = parsed.port or (443 if parsed.scheme.lower() == "https" else 80) except ValueError: return False return ( parsed.scheme.lower() == _DAV_ORIGIN.scheme.lower() and (parsed.hostname or "").lower() == (_DAV_ORIGIN.hostname or "").lower() and parsed_port == base_port ) def _href_relative(href: str) -> str | None: absolute = urljoin(WEBDAV_URL.rstrip("/") + "/", href) parsed = urlsplit(absolute) if not _same_origin(absolute) or parsed.query or parsed.fragment: return None candidate = unquote(parsed.path).rstrip("/") if candidate == _ROOT_PATH_UNQUOTED: return "" prefix = _ROOT_PATH_UNQUOTED + "/" if not candidate.startswith(prefix): return None relative = candidate[len(prefix):] try: return "/".join(_clean_relative_path(relative, allow_empty=False)) except WebDAVError: return None def _dav_client() -> httpx.AsyncClient: return httpx.AsyncClient( auth=httpx.BasicAuth(WEBDAV_USERNAME, WEBDAV_PASSWORD), timeout=HTTP_TIMEOUT, follow_redirects=False, headers={"User-Agent": "dav-mcp/1.0"}, ) async def _limited_response_body(response: httpx.Response, max_bytes: int) -> bytes: result = bytearray() async for chunk in response.aiter_bytes(): if len(result) + len(chunk) > max_bytes: raise WebDAVError(f"Response exceeds the {max_bytes}-byte size limit") result.extend(chunk) return bytes(result) async def _propfind(client: httpx.AsyncClient, path: str, *, depth: str = "1") -> tuple[bytes, int]: body = ( '' '' '' ) try: async with client.stream( "PROPFIND", _path_url(path, collection=True), headers={"Depth": depth, "Content-Type": "application/xml; charset=utf-8"}, content=body, ) as response: if response.status_code == 404: raise WebDAVError(f"Folder not found inside the vault: {path or '/'}") if response.status_code not in {200, 207}: raise WebDAVError(f"WebDAV PROPFIND failed with HTTP {response.status_code}") return await _limited_response_body(response, MAX_PROPFIND_BYTES), response.status_code except httpx.HTTPError as exc: raise WebDAVError(f"WebDAV request failed: {type(exc).__name__}") from exc def _parse_propfind(xml_body: bytes) -> list[dict[str, object]]: try: root = ET.fromstring(xml_body) except ET.ParseError as exc: raise WebDAVError("WebDAV returned invalid directory XML") from exc items: list[dict[str, object]] = [] for response in root.findall(f"{DAV_NS}response"): href_node = response.find(f"{DAV_NS}href") if href_node is None or not href_node.text: continue rel_path = _href_relative(href_node.text) if rel_path is None: continue props = None for propstat in response.findall(f"{DAV_NS}propstat"): status_node = propstat.find(f"{DAV_NS}status") if status_node is not None and status_node.text and " 200 " not in status_node.text: continue props = propstat.find(f"{DAV_NS}prop") if props is not None: break if props is None: continue resource_type = props.find(f"{DAV_NS}resourcetype") is_directory = resource_type is not None and resource_type.find(f"{DAV_NS}collection") is not None length_node = props.find(f"{DAV_NS}getcontentlength") modified_node = props.find(f"{DAV_NS}getlastmodified") etag_node = props.find(f"{DAV_NS}getetag") try: size = int(length_node.text) if length_node is not None and length_node.text else None except ValueError: size = None items.append( { "path": rel_path, "name": rel_path.rsplit("/", 1)[-1] if rel_path else "/", "is_directory": is_directory, "size": size, "last_modified": modified_node.text if modified_node is not None else None, "etag": etag_node.text if etag_node is not None else None, } ) return items async def _list_directory(client: httpx.AsyncClient, path: str) -> list[dict[str, object]]: base = "/".join(_clean_relative_path(path)) xml_body, _ = await _propfind(client, base, depth="1") all_items = _parse_propfind(xml_body) target = next((item for item in all_items if item["path"] == base), None) if target is not None and not target["is_directory"]: raise WebDAVError(f"Not a folder: {base or '/'}") children = [] for item in all_items: item_path = str(item["path"]) if item_path == base: continue parent = item_path.rsplit("/", 1)[0] if "/" in item_path else "" if parent == base: children.append(item) children.sort(key=lambda item: (not bool(item["is_directory"]), str(item["name"]).casefold())) return children async def _walk_directories(client: httpx.AsyncClient, start: str): queue = deque([(start, 0)]) visited: set[str] = set() while queue and len(visited) < MAX_SEARCH_DIRECTORIES: current, depth = queue.popleft() if current in visited: continue visited.add(current) entries = await _list_directory(client, current) yield current, entries if depth < 32: for entry in entries: if entry["is_directory"]: queue.append((str(entry["path"]), depth + 1)) def _write_scope_ok() -> bool: access = get_access_token() return bool(access and "notes:write" in access.scopes) oauth_provider = LocalOAuthProvider( issuer_url=MCP_PUBLIC_URL, resource_url=MCP_RESOURCE_URL, database_path=DATABASE_PATH, username=MCP_AUTH_USERNAME, password=MCP_AUTH_PASSWORD, writable=not READ_ONLY, ) mcp = MCPServer( "webdav-personal-notes", title="Personal Obsidian Vault", description="Browse, search, and read the WebDAV vault at /Notes/Personal.", instructions=( "All paths are relative to the configured /Notes/Personal vault. " "Use list_directory, search_files, or search_notes to discover notes before reading. " "The server never exposes paths outside the configured vault." ), auth_server_provider=oauth_provider, auth=AuthSettings( issuer_url=AnyHttpUrl(MCP_PUBLIC_URL), resource_server_url=AnyHttpUrl(MCP_RESOURCE_URL), required_scopes=["notes:read"], validate_token_resource=True, client_registration_options=ClientRegistrationOptions( enabled=True, valid_scopes=OAUTH_SCOPES, default_scopes=OAUTH_SCOPES, ), revocation_options=RevocationOptions(enabled=True), ), ) @mcp.tool( description="List files and subfolders immediately inside a vault folder. Use an empty path for the vault root.", annotations=ToolAnnotations(read_only_hint=True, open_world_hint=True), ) async def list_directory(path: str = "") -> str: """List direct children of a relative vault folder.""" try: async with _dav_client() as client: entries = await _list_directory(client, path) return json.dumps(entries, ensure_ascii=False, indent=2) except (WebDAVError, httpx.HTTPError) as exc: return f"Error: {exc}" @mcp.tool( description="Read a UTF-8 text or Markdown file from the vault. Returns the file text and ETag; files are limited to 256 KiB.", annotations=ToolAnnotations(read_only_hint=True, open_world_hint=True), ) async def read_file(path: str) -> str: """Read a text note by its relative path.""" try: parts = _clean_relative_path(path, allow_empty=False) relative = "/".join(parts) async with _dav_client() as client: async with client.stream("GET", _path_url(relative)) as response: if response.status_code == 404: raise WebDAVError(f"File not found inside the vault: {relative}") if response.status_code >= 300: raise WebDAVError(f"WebDAV GET failed with HTTP {response.status_code}") data = await _limited_response_body(response, MAX_FILE_BYTES) etag = response.headers.get("ETag") text = data.decode("utf-8-sig", errors="replace") return json.dumps({"path": relative, "etag": etag, "content": text}, ensure_ascii=False) except (WebDAVError, httpx.HTTPError) as exc: return f"Error: {exc}" @mcp.tool( description="Find files and folders by a case-insensitive path or filename substring. Searches recursively from the supplied folder.", annotations=ToolAnnotations(read_only_hint=True, open_world_hint=True), ) async def search_files(query: str, path: str = "", limit: int = 100) -> str: """Search file and folder paths recursively.""" try: _clean_relative_path(path) if not query or len(query) > 256: raise WebDAVError("Search query must contain 1 to 256 characters") if not 1 <= limit <= 200: raise WebDAVError("limit must be between 1 and 200") needle = query.casefold() matches: list[dict[str, object]] = [] dirs_visited = 0 truncated = False async with _dav_client() as client: async for _, entries in _walk_directories(client, path): dirs_visited += 1 for entry in entries: if needle in str(entry["path"]).casefold(): matches.append(entry) if len(matches) >= limit: truncated = True break if truncated: break if dirs_visited >= MAX_SEARCH_DIRECTORIES: truncated = True return json.dumps({"matches": matches, "truncated": truncated}, ensure_ascii=False, indent=2) except (WebDAVError, httpx.HTTPError) as exc: return f"Error: {exc}" @mcp.tool( description="Search note text recursively. Searches Markdown and text files only; each file is limited to 256 KiB and the crawl is capped.", annotations=ToolAnnotations(read_only_hint=True, open_world_hint=True), ) async def search_notes(query: str, path: str = "", limit: int = 20) -> str: """Search Markdown and text file contents for a phrase.""" try: _clean_relative_path(path) if not query or len(query) > 256: raise WebDAVError("Search query must contain 1 to 256 characters") if not 1 <= limit <= 100: raise WebDAVError("limit must be between 1 and 100") needle = query.casefold() results: list[dict[str, str]] = [] scanned_files = 0 dirs_visited = 0 truncated = False async with _dav_client() as client: async for _, entries in _walk_directories(client, path): dirs_visited += 1 for entry in entries: if entry["is_directory"]: continue file_path = str(entry["path"]) if not file_path.casefold().endswith((".md", ".markdown", ".txt")): continue scanned_files += 1 if scanned_files > MAX_SEARCH_FILES: truncated = True break try: async with client.stream("GET", _path_url(file_path)) as response: if response.status_code == 404: continue if response.status_code >= 300: raise WebDAVError(f"WebDAV GET failed with HTTP {response.status_code}") data = await _limited_response_body(response, MAX_FILE_BYTES) except httpx.HTTPError as exc: raise WebDAVError(f"WebDAV request failed: {type(exc).__name__}") from exc except WebDAVError as exc: if "size limit" in str(exc): truncated = True continue raise content = data.decode("utf-8-sig", errors="replace") match_at = content.casefold().find(needle) if match_at >= 0: start = max(0, match_at - 180) end = min(len(content), match_at + len(query) + 180) snippet = content[start:end].replace("\r", " ").replace("\n", " ") results.append({"path": file_path, "snippet": snippet}) if len(results) >= limit: truncated = True break if truncated: break if dirs_visited >= MAX_SEARCH_DIRECTORIES or scanned_files >= MAX_SEARCH_FILES: truncated = True return json.dumps({"matches": results, "scanned_files": scanned_files, "truncated": truncated}, ensure_ascii=False, indent=2) except (WebDAVError, httpx.HTTPError) as exc: return f"Error: {exc}" if not READ_ONLY: @mcp.tool( description=( "Create or replace a UTF-8 text file in the vault. New files are create-only by default. " "To replace a note, set overwrite=true and pass the ETag returned by read_file as expected_etag." ), annotations=ToolAnnotations( read_only_hint=False, destructive_hint=True, idempotent_hint=False, open_world_hint=True ), ) async def write_file( path: str, content: str, overwrite: bool = False, expected_etag: str | None = None, ) -> str: """Create a note, or replace it only when an ETag confirms the version read.""" if not _write_scope_ok(): return "Error: This OAuth connection lacks notes:write. Reconnect and authorize write access." try: parts = _clean_relative_path(path, allow_empty=False) relative = "/".join(parts) if path.endswith("/"): raise WebDAVError("write_file requires a file path, not a folder path") if overwrite and not expected_etag: raise WebDAVError("expected_etag is required when overwrite=true; read the file first") if not overwrite and expected_etag: raise WebDAVError("expected_etag is only used when overwrite=true") data = content.encode("utf-8") if len(data) > MAX_FILE_BYTES: raise WebDAVError(f"File exceeds the {MAX_FILE_BYTES}-byte write limit") headers = {"Content-Type": "text/markdown; charset=utf-8"} headers["If-Match" if overwrite else "If-None-Match"] = expected_etag if overwrite else "*" async with _dav_client() as client: async with client.stream("PUT", _path_url(relative), headers=headers, content=data) as response: if response.status_code == 412: raise WebDAVError("The file changed or already exists. Read it again and retry deliberately.") if response.status_code not in {200, 201, 204}: raise WebDAVError(f"WebDAV PUT failed with HTTP {response.status_code}") etag = response.headers.get("ETag") return json.dumps({"saved": relative, "etag": etag, "bytes_written": len(data)}, ensure_ascii=False) except (WebDAVError, httpx.HTTPError) as exc: return f"Error: {exc}" @mcp.tool( description="Create one new folder inside the vault. The parent folder must already exist.", annotations=ToolAnnotations( read_only_hint=False, destructive_hint=False, idempotent_hint=True, open_world_hint=True ), ) async def create_directory(path: str) -> str: """Create a single directory using WebDAV MKCOL.""" if not _write_scope_ok(): return "Error: This OAuth connection lacks notes:write. Reconnect and authorize write access." try: parts = _clean_relative_path(path, allow_empty=False) relative = "/".join(parts) async with _dav_client() as client: async with client.stream("MKCOL", _path_url(relative, collection=True)) as response: if response.status_code == 405: raise WebDAVError("That folder already exists") if response.status_code not in {200, 201, 204}: raise WebDAVError(f"WebDAV MKCOL failed with HTTP {response.status_code}") return f"Created folder: {relative}" except (WebDAVError, httpx.HTTPError) as exc: return f"Error: {exc}" async def health(_: Request) -> JSONResponse: return JSONResponse({"status": "ok"}) transport_security = TransportSecuritySettings( allowed_hosts=MCP_ALLOWED_HOSTS, allowed_origins=MCP_ALLOWED_ORIGINS, ) mcp_http_app = mcp.streamable_http_app(transport_security=transport_security) _resource_metadata_path = "/.well-known/oauth-protected-resource/mcp" mcp_http_app.router.routes = [ route for route in mcp_http_app.router.routes if getattr(route, "path", None) != _resource_metadata_path ] mcp_http_app.router.routes.extend( create_protected_resource_routes( resource_url=AnyHttpUrl(MCP_RESOURCE_URL), authorization_servers=[AnyHttpUrl(MCP_PUBLIC_URL)], scopes_supported=OAUTH_SCOPES, resource_name="Personal Obsidian Vault", ) ) for _route in make_login_routes(oauth_provider): if getattr(_route, "path", None) == "/login/callback": _route.app = RequestBodyLimitMiddleware(_route.app, 16 * 1024) mcp_http_app.router.routes.append(_route) @asynccontextmanager async def lifespan(_: Starlette): await oauth_provider.initialize() async with mcp.session_manager.run(): yield app = Starlette( routes=[Route("/healthz", health, methods=["GET"]), Mount("/", app=mcp_http_app)], lifespan=lifespan, )