Merge pull request #473 from cbcoutinho/fix/multi-user-basicauth-app-password-storage

fix(auth): Store app passwords locally for multi-user BasicAuth background sync
2026-01-14 20:52:12 +01:00
parent 7465e962d4 01ad2b3d21
commit 1f1dd94598
15 changed files with 2141 additions and 200 deletions
@@ -0,0 +1,16 @@
+#!/bin/bash
+# Configure MCP server URL for Astrolabe background sync
+# This URL is used by Astrolabe to send app passwords to the MCP server
+
+set -e
+
+# The MCP multi-user BasicAuth service runs on port 8000 inside the container
+# From Nextcloud's perspective (inside Docker network), we reach it via service name
+MCP_SERVER_URL="${MCP_SERVER_URL:-http://mcp-multi-user-basic:8000}"
+
+echo "Configuring MCP server URL: $MCP_SERVER_URL"
+
+# Set the mcp_server_url in config.php via occ
+php occ config:system:set mcp_server_url --value="$MCP_SERVER_URL"
+
+echo "MCP server URL configured successfully"
@@ -0,0 +1,357 @@
+# Webhook Management Guide
+
+This guide explains how to enable and disable webhooks for vector sync in each MCP server deployment mode. Webhooks enable near-real-time synchronization of content changes to the vector database, complementing the default polling-based sync.
+
+**Related ADRs:**
+- ADR-010: Webhook-Based Vector Sync
+- ADR-020: Deployment Modes and Configuration Validation
+
+## Prerequisites
+
+Before enabling webhooks, ensure:
+
+1. **Nextcloud 30+** with `webhook_listeners` app enabled
+2. **Astrolabe app** installed in Nextcloud (provides settings UI and credentials API)
+3. **MCP server** accessible from Nextcloud via HTTP(S)
+4. **Vector sync enabled** on the MCP server
+
+## Webhook Architecture Overview
+
+The webhook system has two components:
+
+1. **Webhook Registration** - Configuring Nextcloud to send change notifications to the MCP server
+2. **Background Sync Credentials** - Allowing the MCP server to access Nextcloud APIs on behalf of users
+
+Both must be configured for webhooks to function properly.
+
+## Deployment Mode Specifics
+
+### 1. Single-User BasicAuth
+
+**Configuration:**
+```bash
+NEXTCLOUD_HOST=http://localhost:8080
+NEXTCLOUD_USERNAME=admin
+NEXTCLOUD_PASSWORD=password
+VECTOR_SYNC_ENABLED=true
+```
+
+**Enable Webhooks:**
+1. Register webhooks using occ commands (requires Nextcloud admin):
+   ```bash
+   # Enable webhook_listeners app
+   php occ app:enable webhook_listeners
+
+   # Register webhooks for vector sync
+   php occ webhook_listeners:add \
+     --event "OCP\Files\Events\Node\NodeCreatedEvent" \
+     --uri "http://mcp-server:8000/webhooks/nextcloud" \
+     --method POST
+
+   # Repeat for other events (see Event Types below)
+   ```
+
+2. Optionally reduce polling frequency:
+   ```bash
+   VECTOR_SYNC_SCAN_INTERVAL=86400  # 24 hours
+   ```
+
+**Disable Webhooks:**
+```bash
+# List registered webhooks
+php occ webhook_listeners:list
+
+# Remove specific webhook by ID
+php occ webhook_listeners:remove <webhook-id>
+```
+
+**Notes:**
+- Simplest mode - admin credentials used for all operations
+- No per-user provisioning required
+- Background sync runs as the configured admin user
+
+---
+
+### 2. Multi-User BasicAuth Pass-Through
+
+**Configuration:**
+```bash
+NEXTCLOUD_HOST=http://nextcloud.example.com
+ENABLE_MULTI_USER_BASIC_AUTH=true
+ENABLE_BACKGROUND_OPERATIONS=true
+TOKEN_ENCRYPTION_KEY=<key>
+TOKEN_STORAGE_DB=/app/data/tokens.db
+VECTOR_SYNC_ENABLED=true
+# OAuth client for Astrolabe API access
+NEXTCLOUD_OIDC_CLIENT_ID=<client-id>
+NEXTCLOUD_OIDC_CLIENT_SECRET=<client-secret>
+```
+
+**Credential Architecture:**
+This mode uses **two separate credential mechanisms**:
+
+1. **OAuth Session** (for management API access, including webhooks):
+   - Obtained via browser OAuth flow (`/oauth/login`)
+   - Stores refresh token in MCP server's `tokens.db`
+   - Used for webhook registration/management APIs
+
+2. **App Password** (for background sync):
+   - Generated in Nextcloud Security settings
+   - Stored encrypted in Nextcloud's `oc_preferences` via Astrolabe
+   - Used by background scanners to access Nextcloud APIs
+
+**Enable Webhooks:**
+
+#### Step 1: Complete OAuth Login (for Management API)
+Users must authorize the MCP server to access their Nextcloud:
+
+1. Navigate to **Nextcloud Settings → Astrolabe** (Personal settings)
+2. Click **"Authorize via OAuth"** under "Option 1"
+3. Complete OAuth consent flow
+4. Verify the page shows "Background Sync Access: Active"
+
+#### Step 2: Configure App Password (for Background Sync)
+Since OAuth refresh tokens have short expiry, users should also configure an app password:
+
+1. Navigate to **Nextcloud Settings → Security**
+2. Generate a new app password (name it "Astrolabe" or "MCP Server")
+3. Return to **Nextcloud Settings → Astrolabe**
+4. Under "Option 2: App Password", paste the app password
+5. Click **Save**
+
+#### Step 3: Register Webhooks (Admin)
+Same as Single-User BasicAuth:
+```bash
+php occ webhook_listeners:add \
+  --event "OCP\Files\Events\Node\NodeCreatedEvent" \
+  --uri "http://mcp-server:8003/webhooks/nextcloud" \
+  --method POST
+```
+
+**Disable Webhooks:**
+
+*Per-User:*
+1. Navigate to **Nextcloud Settings → Astrolabe**
+2. Click **"Revoke Access"** (for OAuth tokens) or **"Revoke Access"** (for app password)
+
+*System-Wide:*
+```bash
+php occ webhook_listeners:remove <webhook-id>
+```
+
+**Troubleshooting:**
+
+If OAuth login fails with "Access forbidden - Your client is not authorized":
+1. Check if OAuth client is registered:
+   ```sql
+   SELECT id, name, client_identifier FROM oc_oidc_clients
+   WHERE dcr = 1 ORDER BY id DESC LIMIT 5;
+   ```
+2. Restart MCP server to trigger DCR re-registration
+3. Verify `NEXTCLOUD_OIDC_CLIENT_ID` and `NEXTCLOUD_OIDC_CLIENT_SECRET` are set
+
+If background sync fails with "User no longer provisioned":
+1. Verify app password is stored:
+   ```sql
+   SELECT userid, configkey FROM oc_preferences
+   WHERE appid = 'astrolabe' AND userid = 'username';
+   ```
+2. Ensure user completed **both** OAuth login AND app password setup
+
+---
+
+### 3. OAuth Single-Audience (Default OAuth Mode)
+
+**Configuration:**
+```bash
+NEXTCLOUD_HOST=http://nextcloud.example.com
+# No NEXTCLOUD_USERNAME/PASSWORD
+ENABLE_BACKGROUND_OPERATIONS=true
+TOKEN_ENCRYPTION_KEY=<key>
+TOKEN_STORAGE_DB=/app/data/tokens.db
+VECTOR_SYNC_ENABLED=true
+```
+
+**Enable Webhooks:**
+
+#### Step 1: User Provisioning
+Users authorize via OAuth with `offline_access` scope:
+
+1. MCP client initiates OAuth flow
+2. User consents to requested scopes including `offline_access`
+3. MCP server stores refresh token for background operations
+
+Alternatively, via Astrolabe UI:
+1. Navigate to **Nextcloud Settings → Astrolabe**
+2. Click **"Authorize via OAuth"**
+3. Complete consent flow
+
+#### Step 2: Register Webhooks (Admin)
+```bash
+php occ webhook_listeners:add \
+  --event "OCP\Files\Events\Node\NodeCreatedEvent" \
+  --uri "http://mcp-server:8001/webhooks/nextcloud" \
+  --method POST
+```
+
+**Disable Webhooks:**
+
+*Per-User:*
+- Via Astrolabe UI: Click "Disable Indexing" or "Disconnect"
+- Via MCP tool: Use `revoke_nextcloud_access` if available
+
+*System-Wide:*
+```bash
+php occ webhook_listeners:remove <webhook-id>
+```
+
+---
+
+### 4. OAuth Token Exchange (RFC 8693)
+
+**Configuration:**
+```bash
+NEXTCLOUD_HOST=http://nextcloud.example.com
+ENABLE_TOKEN_EXCHANGE=true
+ENABLE_BACKGROUND_OPERATIONS=true
+TOKEN_ENCRYPTION_KEY=<key>
+TOKEN_STORAGE_DB=/app/data/tokens.db
+VECTOR_SYNC_ENABLED=true
+```
+
+**Enable/Disable Webhooks:**
+Same process as OAuth Single-Audience. The token exchange happens transparently when the MCP server accesses Nextcloud APIs.
+
+---
+
+### 5. Smithery Stateless
+
+**Configuration:**
+- Configuration from session URL params
+- `VECTOR_SYNC_ENABLED=false` (required)
+
+**Webhooks:**
+**Not supported.** This mode is stateless with no persistent storage or background operations.
+
+---
+
+## Webhook Event Types
+
+Register these webhook events for full vector sync coverage:
+
+### File/Note Events
+```bash
+# Use BeforeNodeDeletedEvent for deletions (includes node.id)
+php occ webhook_listeners:add --event "OCP\Files\Events\Node\NodeCreatedEvent" --uri "$MCP_URL/webhooks/nextcloud"
+php occ webhook_listeners:add --event "OCP\Files\Events\Node\NodeWrittenEvent" --uri "$MCP_URL/webhooks/nextcloud"
+php occ webhook_listeners:add --event "OCP\Files\Events\Node\BeforeNodeDeletedEvent" --uri "$MCP_URL/webhooks/nextcloud"
+```
+
+### Calendar Events
+```bash
+php occ webhook_listeners:add --event "OCP\Calendar\Events\CalendarObjectCreatedEvent" --uri "$MCP_URL/webhooks/nextcloud"
+php occ webhook_listeners:add --event "OCP\Calendar\Events\CalendarObjectUpdatedEvent" --uri "$MCP_URL/webhooks/nextcloud"
+php occ webhook_listeners:add --event "OCP\Calendar\Events\CalendarObjectDeletedEvent" --uri "$MCP_URL/webhooks/nextcloud"
+```
+
+### Tables Events
+```bash
+php occ webhook_listeners:add --event "OCA\Tables\Event\RowAddedEvent" --uri "$MCP_URL/webhooks/nextcloud"
+php occ webhook_listeners:add --event "OCA\Tables\Event\RowUpdatedEvent" --uri "$MCP_URL/webhooks/nextcloud"
+php occ webhook_listeners:add --event "OCA\Tables\Event\RowDeletedEvent" --uri "$MCP_URL/webhooks/nextcloud"
+```
+
+## Webhook Presets (via Astrolabe UI)
+
+The Astrolabe app provides preset webhook configurations that can be enabled/disabled via the Admin settings UI:
+
+| Preset | Events Covered |
+|--------|----------------|
+| `notes_sync` | File create/update/delete for .md files |
+| `calendar_sync` | Calendar object events |
+| `tables_sync` | Tables row events |
+| `forms_sync` | Forms submission events |
+| `files_sync` | All file events (optional, high volume) |
+
+**Enable Presets:**
+1. Navigate to **Nextcloud Settings → Astrolabe** (Admin settings)
+2. Toggle desired presets in "Webhook Configuration"
+
+**Note:** Presets require the MCP server's management API to be accessible. The API uses OAuth bearer tokens from the user's session.
+
+## Security Considerations
+
+### Webhook Authentication
+Configure `WEBHOOK_SECRET` to require authentication for incoming webhooks:
+
+```bash
+# MCP Server
+WEBHOOK_SECRET=<generate-random-secret>
+
+# Nextcloud webhook registration
+php occ webhook_listeners:add \
+  --event "..." \
+  --uri "$MCP_URL/webhooks/nextcloud" \
+  --header "Authorization: Bearer <secret>"
+```
+
+### Token Storage
+- Refresh tokens and app passwords are encrypted using `TOKEN_ENCRYPTION_KEY`
+- Store the key securely (environment variable, secrets manager)
+- Different users have isolated credential storage
+
+## Monitoring
+
+### MCP Server Logs
+```bash
+# Docker
+docker compose logs mcp-multi-user-basic | grep -i webhook
+
+# Key log messages
+# - "Queued document from webhook: ..." - Success
+# - "Webhook authentication failed" - Auth error
+# - "User X no longer provisioned" - Missing credentials
+```
+
+### Nextcloud Logs
+```bash
+docker compose exec app cat /var/www/html/data/nextcloud.log | \
+  jq 'select(.message | contains("webhook"))' | tail
+```
+
+### Database Checks
+```sql
+-- Check registered webhooks
+SELECT * FROM oc_webhook_listeners;
+
+-- Check OAuth clients
+SELECT id, name, token_type FROM oc_oidc_clients WHERE dcr = 1;
+
+-- Check user credentials in Astrolabe
+SELECT userid, configkey FROM oc_preferences WHERE appid = 'astrolabe';
+```
+
+## Common Issues
+
+### "Access forbidden - Your client is not authorized to connect"
+**Cause:** OAuth client registration expired or not present in Nextcloud
+**Fix:** Restart MCP server to trigger DCR re-registration
+
+### "User X no longer provisioned, stopping scanner"
+**Cause:** Background sync credentials missing or expired
+**Fix:** User must complete credential provisioning (see mode-specific steps)
+
+### "Failed to fetch" in browser console during OAuth
+**Cause:** Network issue between browser and MCP server callback endpoint
+**Fix:** Verify MCP server is accessible at the configured `NEXTCLOUD_MCP_SERVER_URL`
+
+### Webhooks not firing
+**Causes:**
+1. `webhook_listeners` app not enabled
+2. Webhook not registered for the event type
+3. Background job workers not running
+**Fix:**
+```bash
+php occ app:enable webhook_listeners
+php occ background:cron  # or configure systemd cron
+```
@@ -0,0 +1,50 @@
+"""Add app_passwords table for multi-user BasicAuth mode
+
+This migration adds support for storing app passwords that are provisioned
+via Astrolabe's personal settings. This enables background sync in
+multi-user BasicAuth mode without requiring OAuth.
+
+Revision ID: 002
+Revises: 001
+Create Date: 2026-01-13 12:00:00.000000
+
+"""
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "002"
+down_revision = "001"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """Add app_passwords table for multi-user BasicAuth mode."""
+
+    # App passwords table for multi-user BasicAuth background sync
+    op.execute(
+        """
+        CREATE TABLE IF NOT EXISTS app_passwords (
+            user_id TEXT PRIMARY KEY,
+            encrypted_password BLOB NOT NULL,
+            created_at INTEGER NOT NULL,
+            updated_at INTEGER NOT NULL
+        )
+        """
+    )
+
+    # Index for efficient user lookups
+    op.execute(
+        """
+        CREATE INDEX IF NOT EXISTS idx_app_passwords_updated
+        ON app_passwords(updated_at)
+        """
+    )
+
+
+def downgrade() -> None:
+    """Drop app_passwords table."""
+
+    op.execute("DROP INDEX IF EXISTS idx_app_passwords_updated")
+    op.execute("DROP TABLE IF EXISTS app_passwords")
@@ -10,12 +10,18 @@ All endpoints use OAuth bearer token authentication via UnifiedTokenVerifier.
 The PHP app obtains tokens through PKCE flow and uses them to access these endpoints.
 """

+import base64
 import logging
+import re
 import time
+from collections import defaultdict
 from importlib.metadata import version
-from typing import Any
+from typing import TYPE_CHECKING, Any

 import httpx
+
+if TYPE_CHECKING:
+    from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
 from starlette.requests import Request
 from starlette.responses import JSONResponse

@@ -25,6 +31,23 @@ logger = logging.getLogger(__name__)
 # Get package version from metadata
 __version__ = version("nextcloud-mcp-server")

+# App password format regex (Nextcloud format: xxxxx-xxxxx-xxxxx-xxxxx-xxxxx)
+APP_PASSWORD_PATTERN = re.compile(
+    r"^[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}$"
+)
+
+# Timeout for Nextcloud API validation requests (seconds)
+NEXTCLOUD_VALIDATION_TIMEOUT = 10.0
+
+# Rate limiting configuration for app password provisioning
+# Limits: 5 attempts per user per hour
+RATE_LIMIT_MAX_ATTEMPTS = 5
+RATE_LIMIT_WINDOW_SECONDS = 3600  # 1 hour
+
+# In-memory rate limiter storage
+# Structure: {user_id: [(timestamp, success), ...]}
+_rate_limit_attempts: dict[str, list[tuple[float, bool]]] = defaultdict(list)
+
 # Track server start time for uptime calculation
 _server_start_time = time.time()

@@ -181,6 +204,141 @@ def _validate_query_string(query: str, max_length: int = 10000) -> None:
        raise ValueError(f"Query too long: maximum {max_length} characters")


+async def _get_app_password_storage(request: Request) -> "RefreshTokenStorage":
+    """Get or initialize RefreshTokenStorage for app password operations.
+
+    Checks app.state.storage first, then falls back to creating from environment.
+    This helper avoids repeated storage initialization logic across endpoints.
+
+    Args:
+        request: Starlette request with app state
+
+    Returns:
+        Initialized RefreshTokenStorage instance
+    """
+    from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
+
+    storage = getattr(request.app.state, "storage", None)
+
+    if not storage:
+        # Multi-user BasicAuth mode may not have oauth_context
+        # Initialize storage from environment
+        storage = RefreshTokenStorage.from_env()
+        await storage.initialize()
+
+    return storage
+
+
+def _check_rate_limit(user_id: str) -> tuple[bool, int]:
+    """Check if user is rate limited for app password operations.
+
+    Implements a sliding window rate limiter to prevent brute-force attacks
+    on the app password provisioning endpoint.
+
+    Args:
+        user_id: User identifier to check
+
+    Returns:
+        Tuple of (is_allowed, seconds_until_retry)
+        - is_allowed: True if request should be allowed
+        - seconds_until_retry: Seconds to wait if rate limited (0 if allowed)
+    """
+    current_time = time.time()
+    window_start = current_time - RATE_LIMIT_WINDOW_SECONDS
+
+    # Clean up old attempts outside the window
+    _rate_limit_attempts[user_id] = [
+        (ts, success)
+        for ts, success in _rate_limit_attempts[user_id]
+        if ts > window_start
+    ]
+
+    # Count recent attempts (both successful and failed)
+    recent_attempts = len(_rate_limit_attempts[user_id])
+
+    if recent_attempts >= RATE_LIMIT_MAX_ATTEMPTS:
+        # Find when the oldest attempt in the window will expire
+        oldest_attempt = min(ts for ts, _ in _rate_limit_attempts[user_id])
+        seconds_until_retry = int(
+            oldest_attempt + RATE_LIMIT_WINDOW_SECONDS - current_time
+        )
+        return False, max(1, seconds_until_retry)
+
+    return True, 0
+
+
+def _record_rate_limit_attempt(user_id: str, success: bool) -> None:
+    """Record an app password provisioning attempt for rate limiting.
+
+    Args:
+        user_id: User identifier
+        success: Whether the attempt was successful
+    """
+    _rate_limit_attempts[user_id].append((time.time(), success))
+
+
+def _extract_basic_auth(
+    request: Request, path_user_id: str
+) -> tuple[str, str, JSONResponse | None]:
+    """Extract and validate BasicAuth credentials from request.
+
+    Validates:
+    1. Authorization header is present and valid BasicAuth format
+    2. Username in credentials matches the path user_id
+
+    Args:
+        request: Starlette request with Authorization header
+        path_user_id: User ID from the URL path to verify against
+
+    Returns:
+        Tuple of (username, password, error_response)
+        - If successful: (username, password, None)
+        - If failed: ("", "", JSONResponse with error)
+    """
+    auth_header = request.headers.get("Authorization")
+
+    if not auth_header or not auth_header.startswith("Basic "):
+        return (
+            "",
+            "",
+            JSONResponse(
+                {"success": False, "error": "Missing BasicAuth credentials"},
+                status_code=401,
+            ),
+        )
+
+    try:
+        # Decode BasicAuth
+        encoded = auth_header.split(" ", 1)[1]
+        decoded = base64.b64decode(encoded).decode("utf-8")
+        username, password = decoded.split(":", 1)
+    except Exception:
+        return (
+            "",
+            "",
+            JSONResponse(
+                {"success": False, "error": "Invalid BasicAuth format"},
+                status_code=401,
+            ),
+        )
+
+    # Verify username matches path user_id
+    if username != path_user_id:
+        logger.warning(
+            f"Username mismatch in app password operation for path user {path_user_id}"
+        )
+        return (
+            "",
+            "",
+            JSONResponse(
+                {"success": False, "error": "Username does not match path user_id"},
+                status_code=403,
+            ),
+        )
+
+    return username, password, None
+
+
 async def get_server_status(request: Request) -> JSONResponse:
    """GET /api/v1/status - Server status and version.

@@ -510,6 +668,254 @@ async def revoke_user_access(request: Request) -> JSONResponse:
        )


+async def provision_app_password(request: Request) -> JSONResponse:
+    """POST /api/v1/users/{user_id}/app-password - Store app password for background sync.
+
+    This endpoint is used by Astrolabe (Nextcloud PHP app) to provision app passwords
+    for multi-user BasicAuth mode background sync.
+
+    The request must include BasicAuth credentials where:
+    - username: Nextcloud user ID (must match path user_id)
+    - password: The app password being provisioned
+
+    The MCP server validates the app password against Nextcloud before storing it.
+    This proves the user owns the password and has access to Nextcloud.
+
+    Security model:
+    - User identity is verified via BasicAuth against Nextcloud
+    - App password is encrypted before storage
+    - Only the user who owns the password can provision it
+    - Rate limited to prevent brute-force attacks
+    """
+    from nextcloud_mcp_server.config import get_settings
+
+    # Get user_id from path
+    path_user_id = request.path_params.get("user_id")
+    if not path_user_id:
+        return JSONResponse(
+            {"success": False, "error": "Missing user_id in path"},
+            status_code=400,
+        )
+
+    # Check rate limit before processing
+    is_allowed, retry_after = _check_rate_limit(path_user_id)
+    if not is_allowed:
+        logger.warning(
+            f"Rate limit exceeded for app password provisioning: {path_user_id}"
+        )
+        return JSONResponse(
+            {
+                "success": False,
+                "error": f"Rate limit exceeded. Try again in {retry_after} seconds.",
+            },
+            status_code=429,
+            headers={"Retry-After": str(retry_after)},
+        )
+
+    # Extract and validate BasicAuth credentials
+    username, app_password, error_response = _extract_basic_auth(request, path_user_id)
+    if error_response is not None:
+        _record_rate_limit_attempt(path_user_id, success=False)
+        return error_response
+
+    # Validate app password format
+    if not APP_PASSWORD_PATTERN.match(app_password):
+        _record_rate_limit_attempt(path_user_id, success=False)
+        return JSONResponse(
+            {"success": False, "error": "Invalid app password format"},
+            status_code=400,
+        )
+
+    # Get Nextcloud host from settings
+    settings = get_settings()
+    nextcloud_host = settings.nextcloud_host
+
+    if not nextcloud_host:
+        logger.error("NEXTCLOUD_HOST not configured")
+        return JSONResponse(
+            {"success": False, "error": "Server not configured"},
+            status_code=500,
+        )
+
+    # Validate app password against Nextcloud
+    try:
+        async with httpx.AsyncClient(timeout=NEXTCLOUD_VALIDATION_TIMEOUT) as client:
+            # Use OCS API to verify credentials
+            test_url = f"{nextcloud_host}/ocs/v1.php/cloud/user"
+            response = await client.get(
+                test_url,
+                auth=(username, app_password),
+                params={"format": "json"},
+                headers={"OCS-APIRequest": "true"},
+            )
+
+            if response.status_code != 200:
+                logger.warning(
+                    f"App password validation failed for user: HTTP {response.status_code}"
+                )
+                _record_rate_limit_attempt(path_user_id, success=False)
+                return JSONResponse(
+                    {"success": False, "error": "Invalid app password"},
+                    status_code=401,
+                )
+
+            # Verify the user ID from response matches
+            data = response.json()
+            ocs_user_id = data.get("ocs", {}).get("data", {}).get("id")
+            if ocs_user_id != username:
+                logger.warning("User ID mismatch in OCS response")
+                _record_rate_limit_attempt(path_user_id, success=False)
+                return JSONResponse(
+                    {"success": False, "error": "User ID mismatch"},
+                    status_code=403,
+                )
+
+    except httpx.RequestError as e:
+        logger.error(f"Failed to validate app password: {e}")
+        return JSONResponse(
+            {"success": False, "error": "Failed to validate credentials"},
+            status_code=500,
+        )
+
+    # Store the validated app password
+    try:
+        storage = await _get_app_password_storage(request)
+        await storage.store_app_password(username, app_password)
+
+        _record_rate_limit_attempt(path_user_id, success=True)
+        logger.info(f"Provisioned app password for user: {username}")
+
+        return JSONResponse(
+            {
+                "success": True,
+                "message": f"App password stored for {username}",
+            }
+        )
+
+    except Exception as e:
+        error_msg = _sanitize_error_for_client(e, "provision_app_password")
+        return JSONResponse(
+            {"success": False, "error": error_msg},
+            status_code=500,
+        )
+
+
+async def get_app_password_status(request: Request) -> JSONResponse:
+    """GET /api/v1/users/{user_id}/app-password - Check if user has provisioned app password.
+
+    Returns status of background sync access for multi-user BasicAuth mode.
+
+    Requires BasicAuth with the user's app password for authentication.
+    """
+    # Get user_id from path
+    path_user_id = request.path_params.get("user_id")
+    if not path_user_id:
+        return JSONResponse(
+            {"success": False, "error": "Missing user_id in path"},
+            status_code=400,
+        )
+
+    # Extract and validate BasicAuth credentials
+    username, _, error_response = _extract_basic_auth(request, path_user_id)
+    if error_response is not None:
+        return error_response
+
+    try:
+        storage = await _get_app_password_storage(request)
+        app_password = await storage.get_app_password(username)
+
+        return JSONResponse(
+            {
+                "success": True,
+                "user_id": username,
+                "has_app_password": app_password is not None,
+            }
+        )
+
+    except Exception as e:
+        error_msg = _sanitize_error_for_client(e, "get_app_password_status")
+        return JSONResponse(
+            {"success": False, "error": error_msg},
+            status_code=500,
+        )
+
+
+async def delete_app_password(request: Request) -> JSONResponse:
+    """DELETE /api/v1/users/{user_id}/app-password - Delete stored app password.
+
+    Removes the user's app password from MCP server storage.
+
+    Requires BasicAuth with the user's credentials.
+    """
+    from nextcloud_mcp_server.config import get_settings
+
+    # Get user_id from path
+    path_user_id = request.path_params.get("user_id")
+    if not path_user_id:
+        return JSONResponse(
+            {"success": False, "error": "Missing user_id in path"},
+            status_code=400,
+        )
+
+    # Extract and validate BasicAuth credentials
+    username, password, error_response = _extract_basic_auth(request, path_user_id)
+    if error_response is not None:
+        return error_response
+
+    # Validate credentials against Nextcloud
+    settings = get_settings()
+    nextcloud_host = settings.nextcloud_host
+
+    try:
+        async with httpx.AsyncClient(timeout=NEXTCLOUD_VALIDATION_TIMEOUT) as client:
+            test_url = f"{nextcloud_host}/ocs/v1.php/cloud/user"
+            response = await client.get(
+                test_url,
+                auth=(username, password),
+                params={"format": "json"},
+                headers={"OCS-APIRequest": "true"},
+            )
+
+            if response.status_code != 200:
+                return JSONResponse(
+                    {"success": False, "error": "Invalid credentials"},
+                    status_code=401,
+                )
+    except httpx.RequestError as e:
+        logger.error(f"Failed to validate credentials: {e}")
+        return JSONResponse(
+            {"success": False, "error": "Failed to validate credentials"},
+            status_code=500,
+        )
+
+    try:
+        storage = await _get_app_password_storage(request)
+        deleted = await storage.delete_app_password(username)
+
+        if deleted:
+            logger.info(f"Deleted app password for user: {username}")
+            return JSONResponse(
+                {
+                    "success": True,
+                    "message": f"App password deleted for {username}",
+                }
+            )
+        else:
+            return JSONResponse(
+                {
+                    "success": True,
+                    "message": "No app password found to delete",
+                }
+            )
+
+    except Exception as e:
+        error_msg = _sanitize_error_for_client(e, "delete_app_password")
+        return JSONResponse(
+            {"success": False, "error": error_msg},
+            status_code=500,
+        )
+
+
 async def get_installed_apps(request: Request) -> JSONResponse:
    """GET /api/v1/apps - Get list of installed Nextcloud apps.

@@ -2012,7 +2012,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
            checks["auth_mode"] = "multi_user_basic"
            checks["auth_configured"] = "ok"
            # Indicate if app passwords are supported (when offline_access enabled)
-            checks["supports_app_passwords"] = settings.enable_offline_access
+            checks["supports_app_passwords"] = get_settings().enable_offline_access
        elif mode == AuthMode.SINGLE_USER_BASIC:
            username = os.getenv("NEXTCLOUD_USERNAME")
            password = os.getenv("NEXTCLOUD_PASSWORD")
@@ -2029,9 +2029,9 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =

        # Check Qdrant status if using network mode (external Qdrant service)
        # In-memory and persistent modes use embedded Qdrant, no external service to check
-        vector_sync_enabled = (
-            os.getenv("VECTOR_SYNC_ENABLED", "false").lower() == "true"
-        )
+        # Note: get_settings() supports both ENABLE_SEMANTIC_SEARCH and VECTOR_SYNC_ENABLED
+        settings = get_settings()
+        vector_sync_enabled = settings.vector_sync_enabled
        qdrant_url = os.getenv("QDRANT_URL")  # Only set in network mode

        if vector_sync_enabled and qdrant_url:
@@ -2114,13 +2114,16 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
    if enable_management_apis:
        from nextcloud_mcp_server.api.management import (
            create_webhook,
+            delete_app_password,
            delete_webhook,
+            get_app_password_status,
            get_chunk_context,
            get_installed_apps,
            get_server_status,
            get_user_session,
            get_vector_sync_status,
            list_webhooks,
+            provision_app_password,
            revoke_user_access,
            unified_search,
            vector_search,
@@ -2148,6 +2151,28 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
                methods=["POST"],
            )
        )
+        # App password endpoints for multi-user BasicAuth mode
+        routes.append(
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                provision_app_password,
+                methods=["POST"],
+            )
+        )
+        routes.append(
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                get_app_password_status,
+                methods=["GET"],
+            )
+        )
+        routes.append(
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                delete_app_password,
+                methods=["DELETE"],
+            )
+        )
        routes.append(
            Route("/api/v1/vector-viz/search", vector_search, methods=["POST"])
        )
@@ -2166,6 +2191,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
        logger.info(
            "Management API endpoints enabled: /api/v1/status, /api/v1/vector-sync/status, "
            "/api/v1/users/{user_id}/session, /api/v1/users/{user_id}/revoke, "
+            "/api/v1/users/{user_id}/app-password, "
            "/api/v1/vector-viz/search, /api/v1/search, /api/v1/apps, "
            "/api/v1/webhooks"
        )
@@ -1240,6 +1240,180 @@ class RefreshTokenStorage:

        return deleted

+    # ============================================================================
+    # App Password Storage (multi-user BasicAuth mode)
+    # ============================================================================
+
+    async def store_app_password(
+        self,
+        user_id: str,
+        app_password: str,
+    ) -> None:
+        """
+        Store encrypted app password for background sync (multi-user BasicAuth mode).
+
+        Args:
+            user_id: Nextcloud user ID
+            app_password: Nextcloud app password to store
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        if not self.cipher:
+            raise RuntimeError(
+                "Encryption key not configured. "
+                "Set TOKEN_ENCRYPTION_KEY for app password storage."
+            )
+
+        encrypted_password = self.cipher.encrypt(app_password.encode())
+        now = int(time.time())
+
+        start_time = time.time()
+        try:
+            async with aiosqlite.connect(self.db_path) as db:
+                await db.execute(
+                    """
+                    INSERT OR REPLACE INTO app_passwords
+                    (user_id, encrypted_password, created_at, updated_at)
+                    VALUES (
+                        ?,
+                        ?,
+                        COALESCE((SELECT created_at FROM app_passwords WHERE user_id = ?), ?),
+                        ?
+                    )
+                    """,
+                    (user_id, encrypted_password, user_id, now, now),
+                )
+                await db.commit()
+
+            duration = time.time() - start_time
+            record_db_operation("sqlite", "insert", duration, "success")
+            logger.info(f"Stored app password for user {user_id}")
+
+        except Exception:
+            duration = time.time() - start_time
+            record_db_operation("sqlite", "insert", duration, "error")
+            raise
+
+        # Audit log
+        await self._audit_log(
+            event="store_app_password",
+            user_id=user_id,
+            auth_method="app_password",
+        )
+
+    async def get_app_password(self, user_id: str) -> Optional[str]:
+        """
+        Retrieve and decrypt app password for a user.
+
+        Args:
+            user_id: Nextcloud user ID
+
+        Returns:
+            Decrypted app password, or None if not found
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        if not self.cipher:
+            raise RuntimeError(
+                "Encryption key not configured. "
+                "Set TOKEN_ENCRYPTION_KEY for app password retrieval."
+            )
+
+        start_time = time.time()
+        try:
+            async with aiosqlite.connect(self.db_path) as db:
+                async with db.execute(
+                    "SELECT encrypted_password FROM app_passwords WHERE user_id = ?",
+                    (user_id,),
+                ) as cursor:
+                    row = await cursor.fetchone()
+
+            if not row:
+                logger.debug(f"No app password found for user {user_id}")
+                duration = time.time() - start_time
+                record_db_operation("sqlite", "select", duration, "success")
+                return None
+
+            encrypted_password = row[0]
+            decrypted_password = self.cipher.decrypt(encrypted_password).decode()
+
+            duration = time.time() - start_time
+            record_db_operation("sqlite", "select", duration, "success")
+            logger.debug(f"Retrieved app password for user {user_id}")
+
+            return decrypted_password
+
+        except Exception as e:
+            duration = time.time() - start_time
+            record_db_operation("sqlite", "select", duration, "error")
+            logger.error(f"Failed to decrypt app password for user {user_id}: {e}")
+            return None
+
+    async def delete_app_password(self, user_id: str) -> bool:
+        """
+        Delete app password for a user.
+
+        Args:
+            user_id: Nextcloud user ID
+
+        Returns:
+            True if password was deleted, False if not found
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        start_time = time.time()
+        try:
+            async with aiosqlite.connect(self.db_path) as db:
+                cursor = await db.execute(
+                    "DELETE FROM app_passwords WHERE user_id = ?",
+                    (user_id,),
+                )
+                await db.commit()
+                deleted = cursor.rowcount > 0
+
+            duration = time.time() - start_time
+            record_db_operation("sqlite", "delete", duration, "success")
+
+            if deleted:
+                logger.info(f"Deleted app password for user {user_id}")
+                await self._audit_log(
+                    event="delete_app_password",
+                    user_id=user_id,
+                    auth_method="app_password",
+                )
+            else:
+                logger.debug(f"No app password to delete for user {user_id}")
+
+            return deleted
+
+        except Exception:
+            duration = time.time() - start_time
+            record_db_operation("sqlite", "delete", duration, "error")
+            raise
+
+    async def get_all_app_password_user_ids(self) -> list[str]:
+        """
+        Get list of all user IDs with stored app passwords.
+
+        Returns:
+            List of user IDs
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        async with aiosqlite.connect(self.db_path) as db:
+            async with db.execute(
+                "SELECT user_id FROM app_passwords ORDER BY updated_at DESC"
+            ) as cursor:
+                rows = await cursor.fetchall()
+
+        user_ids = [row[0] for row in rows]
+        logger.debug(f"Found {len(user_ids)} users with app passwords")
+        return user_ids
+

 async def generate_encryption_key() -> str:
    """
@@ -20,6 +20,7 @@ from starlette.requests import Request
 from starlette.responses import HTMLResponse, JSONResponse

 from nextcloud_mcp_server.client import NextcloudClient
+from nextcloud_mcp_server.config import get_settings

 logger = logging.getLogger(__name__)

@@ -106,9 +107,9 @@ async def _get_processing_status(request: Request) -> dict[str, Any] | None:
            "status": str,  # "syncing" or "idle"
        }
    """
-    # Check if vector sync is enabled
-    vector_sync_enabled = os.getenv("VECTOR_SYNC_ENABLED", "false").lower() == "true"
-    if not vector_sync_enabled:
+    # Check if vector sync is enabled (supports both old and new env var names)
+    settings = get_settings()
+    if not settings.vector_sync_enabled:
        return None

    try:
@@ -127,10 +128,8 @@ async def _get_processing_status(request: Request) -> dict[str, Any] | None:
        # Get Qdrant client and query indexed count
        indexed_count = 0
        try:
-            from nextcloud_mcp_server.config import get_settings
            from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client

-            settings = get_settings()
            qdrant_client = await get_qdrant_client()

            # Count documents in collection
@@ -634,7 +633,9 @@ async def user_info_html(request: Request) -> HTMLResponse:
        """

    # Check if vector sync is enabled (needed for Welcome tab)
-    vector_sync_enabled = os.getenv("VECTOR_SYNC_ENABLED", "false").lower() == "true"
+    # Note: get_settings() supports both ENABLE_SEMANTIC_SEARCH and VECTOR_SYNC_ENABLED
+    settings = get_settings()
+    vector_sync_enabled = settings.vector_sync_enabled

    # Render template
    template = _jinja_env.get_template("user_info.html")
@@ -637,7 +637,9 @@ def configure_deck_tools(mcp: FastMCP):

    @mcp.tool(
        title="Remove Label from Deck Card",
-        annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True),
+        annotations=ToolAnnotations(
+            destructiveHint=True, idempotentHint=True, openWorldHint=True
+        ),
    )
    @require_scopes("deck:write")
    @instrument_tool
@@ -692,7 +694,9 @@ def configure_deck_tools(mcp: FastMCP):

    @mcp.tool(
        title="Unassign User from Deck Card",
-        annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True),
+        annotations=ToolAnnotations(
+            destructiveHint=True, idempotentHint=True, openWorldHint=True
+        ),
    )
    @require_scopes("deck:write")
    @instrument_tool
@@ -1,7 +1,6 @@
 """Semantic search MCP tools using vector database."""

 import logging
-import os

 import anyio
 from httpx import RequestError
@@ -658,12 +657,11 @@ def configure_semantic_tools(mcp: FastMCP):
        after creating or updating content across all indexed apps.
        """

-        # Check if vector sync is enabled
-        vector_sync_enabled = (
-            os.getenv("VECTOR_SYNC_ENABLED", "false").lower() == "true"
-        )
+        # Check if vector sync is enabled (supports both old and new env var names)
+        from nextcloud_mcp_server.config import get_settings

-        if not vector_sync_enabled:
+        settings = get_settings()
+        if not settings.vector_sync_enabled:
            return VectorSyncStatusResponse(
                indexed_count=0,
                pending_count=0,
@@ -8,8 +8,8 @@ Manages background vector sync for multi-user deployments:
 Authentication strategies are mutually exclusive by deployment mode:

 Multi-user BasicAuth mode (ENABLE_MULTI_USER_BASIC_AUTH=true):
- Uses app passwords obtained via Astrolabe Management API
- Users provision via Astrolabe personal settings
+- Uses app passwords stored locally in MCP server's database
+- Users provision via Astrolabe personal settings, which sends to MCP API
 - OAuth is NOT used

 OAuth mode (with external IdP like Keycloak):
@@ -33,7 +33,6 @@ from anyio.streams.memory import (
 )
 from httpx import BasicAuth

-from nextcloud_mcp_server.auth.astrolabe_client import AstrolabeClient
 from nextcloud_mcp_server.client import NextcloudClient
 from nextcloud_mcp_server.config import get_settings
 from nextcloud_mcp_server.vector.scanner import DocumentTask, scan_user_documents
@@ -71,15 +70,18 @@ class UserSyncState:
 async def get_user_client_basic_auth(
    user_id: str,
    nextcloud_host: str,
+    storage: "RefreshTokenStorage | None" = None,
 ) -> NextcloudClient:
    """Get an authenticated NextcloudClient using app password (BasicAuth mode).

    For multi-user BasicAuth deployments where users provision app passwords
-    via Astrolabe personal settings. OAuth is NOT used in this mode.
+    via Astrolabe personal settings. The app password is stored locally in the
+    MCP server's database after being provisioned through the management API.

    Args:
        user_id: User identifier
        nextcloud_host: Nextcloud base URL
+        storage: Optional RefreshTokenStorage instance (created from env if not provided)

    Returns:
        Authenticated NextcloudClient with BasicAuth
@@ -87,21 +89,15 @@ async def get_user_client_basic_auth(
    Raises:
        NotProvisionedError: If user has not provisioned an app password
    """
-    settings = get_settings()
+    from nextcloud_mcp_server.auth.storage import RefreshTokenStorage

-    if not settings.oidc_client_id or not settings.oidc_client_secret:
-        raise NotProvisionedError(
-            "Astrolabe client credentials not configured. "
-            "Set OIDC_CLIENT_ID and OIDC_CLIENT_SECRET for app password retrieval."
-        )
+    # Get or create storage instance
+    if storage is None:
+        storage = RefreshTokenStorage.from_env()
+        await storage.initialize()

-    astrolabe = AstrolabeClient(
-        nextcloud_host=nextcloud_host,
-        client_id=settings.oidc_client_id,
-        client_secret=settings.oidc_client_secret,
-    )
-
-    app_password = await astrolabe.get_user_app_password(user_id)
+    # Retrieve app password from local storage
+    app_password = await storage.get_app_password(user_id)

    if not app_password:
        raise NotProvisionedError(
@@ -419,8 +415,15 @@ async def user_manager_task(

    while not shutdown_event.is_set():
        try:
-            # Get current provisioned users
-            provisioned_users = set(await refresh_token_storage.get_all_user_ids())
+            # Get current provisioned users based on mode
+            if use_basic_auth:
+                # BasicAuth mode: query app_passwords table
+                provisioned_users = set(
+                    await refresh_token_storage.get_all_app_password_user_ids()
+                )
+            else:
+                # OAuth mode: query refresh_tokens table
+                provisioned_users = set(await refresh_token_storage.get_all_user_ids())
            active_users = set(user_states.keys())

            # Start scanners for new users
@@ -1,18 +1,21 @@
-"""Integration tests for app password provisioning via Astrolabe.
+"""Integration tests for app password provisioning via management API.

 Tests the complete flow for multi-user BasicAuth mode:
-1. User stores app password via Astrolabe API
-2. MCP server retrieves it via OAuth client credentials
-3. Background sync uses it to access Nextcloud (NOT OAuth refresh tokens)
+1. User stores app password via management API endpoint
+2. MCP server stores it locally (encrypted)
+3. Background sync uses locally stored password to access Nextcloud

 These tests verify that BasicAuth and OAuth are completely separate concerns
 with no fallback between them.
 """

-import pytest
+import tempfile
+from pathlib import Path

-from nextcloud_mcp_server.auth.astrolabe_client import AstrolabeClient
-from nextcloud_mcp_server.config import get_settings
+import pytest
+from cryptography.fernet import Fernet
+
+from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
 from nextcloud_mcp_server.vector.oauth_sync import (
    NotProvisionedError,
    get_user_client,
@@ -21,140 +24,60 @@ from nextcloud_mcp_server.vector.oauth_sync import (
 )


-@pytest.mark.integration
-async def test_astrolabe_client_initialization():
-    """Test AstrolabeClient can be instantiated."""
-    client = AstrolabeClient(
-        nextcloud_host="http://localhost:8080",
-        client_id="test-client",
-        client_secret="test-secret",
-    )
+@pytest.fixture
+def encryption_key():
+    """Generate a test encryption key."""
+    return Fernet.generate_key().decode()

-    assert client is not None
-    assert client.nextcloud_host == "http://localhost:8080"
-    assert client.client_id == "test-client"
-    assert client.client_secret == "test-secret"
-    assert client._token_cache is None
+
+@pytest.fixture
+async def temp_storage(encryption_key):
+    """Create temporary storage instance with encryption for testing."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        db_path = Path(tmpdir) / "test_provisioning.db"
+        storage = RefreshTokenStorage(
+            db_path=str(db_path), encryption_key=encryption_key
+        )
+        await storage.initialize()
+        yield storage


@pytest.mark.integration
-async def test_astrolabe_client_get_access_token_requires_oidc():
-    """Test that getting access token requires OIDC discovery endpoint."""
-    client = AstrolabeClient(
-        nextcloud_host="http://localhost:8080",
-        client_id="test-client",
-        client_secret="test-secret",
-    )
+async def test_basic_auth_mode_uses_local_storage(temp_storage, mocker):
+    """Test that BasicAuth mode uses locally stored app passwords.

-    # This will fail without proper OIDC setup, which is expected
-    # The test verifies the client follows the OAuth client credentials flow
-    try:
-        token = await client.get_access_token()
-        # If we get here, OIDC is configured
-        assert token is not None
-    except Exception as e:
-        # Expected if OIDC not fully configured for test client
-        # 400/401/403/404 all indicate the flow is working but credentials are invalid
-        assert any(code in str(e) for code in ["400", "401", "403", "404"])
-
-
-@pytest.mark.integration
-async def test_get_user_app_password_returns_none_for_unconfigured_user():
-    """Test that get_user_app_password returns None for users without app passwords."""
-    # This requires valid OAuth client credentials
-    settings = get_settings()
-
-    if not settings.oidc_client_id or not settings.oidc_client_secret:
-        pytest.skip("OAuth client credentials not configured")
-
-    client = AstrolabeClient(
-        nextcloud_host=settings.nextcloud_host or "http://localhost:8080",
-        client_id=settings.oidc_client_id,
-        client_secret=settings.oidc_client_secret,
-    )
-
-    # Try to get app password for a user that hasn't provisioned one
-    try:
-        app_password = await client.get_user_app_password("nonexistent_user")
-        # Should return None for unconfigured user (404 response)
-        assert app_password is None
-    except Exception as e:
-        # May fail with auth error if OAuth not fully configured
-        assert any(code in str(e) for code in ["400", "401", "403", "404"])
-
-
-@pytest.mark.integration
-async def test_basic_auth_mode_uses_app_password_only(mocker):
-    """Test that BasicAuth mode uses ONLY app passwords, NOT OAuth tokens.
-
-    In multi-user BasicAuth mode, OAuth refresh tokens are NOT used.
-    This is a complete separation of concerns.
+    In multi-user BasicAuth mode, app passwords are stored locally
+    in the MCP server's database after being provisioned via the API.
    """
-    # Mock settings to have client credentials
-    mock_settings = mocker.MagicMock()
-    mock_settings.oidc_client_id = "test-client-id"
-    mock_settings.oidc_client_secret = "test-client-secret"
-    mocker.patch(
-        "nextcloud_mcp_server.vector.oauth_sync.get_settings",
-        return_value=mock_settings,
-    )
+    # Store an app password in local storage
+    await temp_storage.store_app_password("test_user", "JHWzB-ZYgLZ-3qBDj-ZQe5o-LdKpB")

-    # Mock AstrolabeClient to return an app password
-    mock_astrolabe = mocker.AsyncMock()
-    mock_astrolabe.get_user_app_password.return_value = "test-app-password-12345"
-
-    mocker.patch(
-        "nextcloud_mcp_server.vector.oauth_sync.AstrolabeClient",
-        return_value=mock_astrolabe,
-    )
-
-    # Call get_user_client in BasicAuth mode
-    _client = await get_user_client(
+    # Call get_user_client_basic_auth with local storage
+    client = await get_user_client_basic_auth(
        user_id="test_user",
-        token_broker=None,  # No token broker needed for BasicAuth mode
        nextcloud_host="http://localhost:8080",
-        use_basic_auth=True,
+        storage=temp_storage,
    )

-    # Verify app password was requested
-    mock_astrolabe.get_user_app_password.assert_called_once_with("test_user")
-
-    # Verify client was created successfully with correct username
-    assert _client is not None
-    assert _client.username == "test_user"
+    # Verify client was created with correct credentials
+    assert client is not None
+    assert client.username == "test_user"


@pytest.mark.integration
-async def test_basic_auth_mode_raises_error_without_app_password(mocker):
+async def test_basic_auth_mode_raises_error_without_app_password(temp_storage):
    """Test that BasicAuth mode raises NotProvisionedError if no app password.

    There is NO fallback to OAuth - if no app password, user must provision one.
    """
-    # Mock settings to have client credentials
-    mock_settings = mocker.MagicMock()
-    mock_settings.oidc_client_id = "test-client-id"
-    mock_settings.oidc_client_secret = "test-client-secret"
-    mocker.patch(
-        "nextcloud_mcp_server.vector.oauth_sync.get_settings",
-        return_value=mock_settings,
-    )
+    # Don't store any app password

-    # Mock AstrolabeClient to return None (no app password)
-    mock_astrolabe = mocker.AsyncMock()
-    mock_astrolabe.get_user_app_password.return_value = None
-
-    mocker.patch(
-        "nextcloud_mcp_server.vector.oauth_sync.AstrolabeClient",
-        return_value=mock_astrolabe,
-    )
-
-    # Call get_user_client in BasicAuth mode - should raise NotProvisionedError
+    # Call get_user_client_basic_auth - should raise NotProvisionedError
    with pytest.raises(NotProvisionedError) as exc_info:
-        await get_user_client(
+        await get_user_client_basic_auth(
            user_id="test_user",
-            token_broker=None,
            nextcloud_host="http://localhost:8080",
-            use_basic_auth=True,
+            storage=temp_storage,
        )

    # Verify error message mentions app password provisioning
@@ -162,6 +85,33 @@ async def test_basic_auth_mode_raises_error_without_app_password(mocker):
    assert "test_user" in str(exc_info.value)


+@pytest.mark.integration
+async def test_get_user_client_dispatches_to_basic_auth(temp_storage, mocker):
+    """Test that get_user_client dispatches to BasicAuth mode correctly."""
+    # Store an app password
+    await temp_storage.store_app_password("alice", "aaaaa-bbbbb-ccccc-ddddd-eeeee")
+
+    # Mock RefreshTokenStorage.from_env at the source module
+    mocker.patch(
+        "nextcloud_mcp_server.auth.storage.RefreshTokenStorage.from_env",
+        return_value=temp_storage,
+    )
+    # Also mock initialize since from_env returns an uninitialized instance
+    mocker.patch.object(temp_storage, "initialize", return_value=None)
+
+    # Call get_user_client in BasicAuth mode
+    client = await get_user_client(
+        user_id="alice",
+        token_broker=None,  # No token broker needed for BasicAuth mode
+        nextcloud_host="http://localhost:8080",
+        use_basic_auth=True,
+    )
+
+    # Verify client was created successfully
+    assert client is not None
+    assert client.username == "alice"
+
+
@pytest.mark.integration
 async def test_oauth_mode_uses_refresh_token_only(mocker):
    """Test that OAuth mode uses ONLY refresh tokens, NOT app passwords.
@@ -183,7 +133,7 @@ async def test_oauth_mode_uses_refresh_token_only(mocker):
        use_basic_auth=False,  # OAuth mode
    )

-    # Verify token broker was called (NOT Astrolabe)
+    # Verify token broker was called
    mock_token_broker.get_background_token.assert_called_once()


@@ -213,38 +163,6 @@ async def test_oauth_mode_raises_error_without_token(mocker):
    assert "test_user" in str(exc_info.value)


-@pytest.mark.integration
-async def test_get_user_client_basic_auth_function(mocker):
-    """Test the dedicated get_user_client_basic_auth function."""
-    # Mock settings to have client credentials
-    mock_settings = mocker.MagicMock()
-    mock_settings.oidc_client_id = "test-client-id"
-    mock_settings.oidc_client_secret = "test-client-secret"
-    mocker.patch(
-        "nextcloud_mcp_server.vector.oauth_sync.get_settings",
-        return_value=mock_settings,
-    )
-
-    # Mock AstrolabeClient
-    mock_astrolabe = mocker.AsyncMock()
-    mock_astrolabe.get_user_app_password.return_value = "xxxxx-xxxxx-xxxxx-xxxxx-xxxxx"
-
-    mocker.patch(
-        "nextcloud_mcp_server.vector.oauth_sync.AstrolabeClient",
-        return_value=mock_astrolabe,
-    )
-
-    # Call dedicated function
-    client = await get_user_client_basic_auth(
-        user_id="alice",
-        nextcloud_host="http://localhost:8080",
-    )
-
-    assert client is not None
-    assert client.username == "alice"
-    mock_astrolabe.get_user_app_password.assert_called_once_with("alice")
-
-
@pytest.mark.integration
 async def test_get_user_client_oauth_function(mocker):
    """Test the dedicated get_user_client_oauth function."""
@@ -276,3 +194,69 @@ async def test_oauth_mode_requires_token_broker():
            nextcloud_host="http://localhost:8080",
            use_basic_auth=False,  # OAuth mode
        )
+
+
+@pytest.mark.integration
+async def test_multiple_users_basic_auth_mode(temp_storage, mocker):
+    """Test that multiple users can be provisioned independently."""
+    # Store app passwords for multiple users
+    users = {
+        "alice": "aaaaa-aaaaa-aaaaa-aaaaa-aaaaa",
+        "bob": "bbbbb-bbbbb-bbbbb-bbbbb-bbbbb",
+        "charlie": "ccccc-ccccc-ccccc-ccccc-ccccc",
+    }
+
+    for user_id, password in users.items():
+        await temp_storage.store_app_password(user_id, password)
+
+    # Verify each user can get a client
+    for user_id in users.keys():
+        client = await get_user_client_basic_auth(
+            user_id=user_id,
+            nextcloud_host="http://localhost:8080",
+            storage=temp_storage,
+        )
+        assert client is not None
+        assert client.username == user_id
+
+
+@pytest.mark.integration
+async def test_get_all_provisioned_users(temp_storage):
+    """Test that we can list all provisioned users for BasicAuth mode."""
+    # Store app passwords for multiple users
+    await temp_storage.store_app_password("alice", "aaaaa-aaaaa-aaaaa-aaaaa-aaaaa")
+    await temp_storage.store_app_password("bob", "bbbbb-bbbbb-bbbbb-bbbbb-bbbbb")
+
+    # Get all provisioned users
+    user_ids = await temp_storage.get_all_app_password_user_ids()
+
+    assert len(user_ids) == 2
+    assert "alice" in user_ids
+    assert "bob" in user_ids
+
+
+@pytest.mark.integration
+async def test_revoke_app_password(temp_storage):
+    """Test that deleting app password revokes background access."""
+    # Provision user
+    await temp_storage.store_app_password("alice", "aaaaa-aaaaa-aaaaa-aaaaa-aaaaa")
+
+    # Verify user is provisioned
+    user_ids = await temp_storage.get_all_app_password_user_ids()
+    assert "alice" in user_ids
+
+    # Revoke access
+    deleted = await temp_storage.delete_app_password("alice")
+    assert deleted is True
+
+    # Verify user is no longer provisioned
+    user_ids = await temp_storage.get_all_app_password_user_ids()
+    assert "alice" not in user_ids
+
+    # Verify get_user_client now raises NotProvisionedError
+    with pytest.raises(NotProvisionedError):
+        await get_user_client_basic_auth(
+            user_id="alice",
+            nextcloud_host="http://localhost:8080",
+            storage=temp_storage,
+        )
@@ -89,8 +89,13 @@ async def test_create_operations_not_idempotent(nc_mcp_client: ClientSession):
    """Verify create operations are marked as non-idempotent."""
    tools = await nc_mcp_client.list_tools()

+    # Exceptions: operations that are actually idempotent
+    # - calendar_create_meeting: creates or returns existing meeting
+    # - nc_webdav_create_directory: MKCOL returns 405 if exists (same end state)
+    idempotent_exceptions = {"calendar_create_meeting", "nc_webdav_create_directory"}
+
    for tool in tools.tools:
-        if "create" in tool.name.lower() and "calendar_create_meeting" not in tool.name:
+        if "create" in tool.name.lower() and tool.name not in idempotent_exceptions:
            assert tool.annotations is not None, f"Tool {tool.name} missing annotations"
            assert tool.annotations.idempotentHint is not True, (
                f"Create tool {tool.name} should not be idempotent (creates new resources)"
@@ -0,0 +1,227 @@
+"""
+Unit tests for App Password Storage functionality.
+
+Tests the app password methods in RefreshTokenStorage for multi-user
+BasicAuth mode background sync.
+"""
+
+import tempfile
+from pathlib import Path
+
+import pytest
+from cryptography.fernet import Fernet
+
+from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
+
+pytestmark = pytest.mark.unit
+
+
+@pytest.fixture
+def encryption_key():
+    """Generate a test encryption key."""
+    return Fernet.generate_key().decode()
+
+
+@pytest.fixture
+async def temp_storage(encryption_key):
+    """Create temporary storage instance with encryption for testing."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        db_path = Path(tmpdir) / "test_app_passwords.db"
+        storage = RefreshTokenStorage(
+            db_path=str(db_path), encryption_key=encryption_key
+        )
+        await storage.initialize()
+        yield storage
+
+
+async def test_store_app_password(temp_storage):
+    """Test storing an app password."""
+    await temp_storage.store_app_password(
+        user_id="testuser",
+        app_password="JHWzB-ZYgLZ-3qBDj-ZQe5o-LdKpB",
+    )
+
+    # Verify it can be retrieved
+    retrieved = await temp_storage.get_app_password("testuser")
+    assert retrieved == "JHWzB-ZYgLZ-3qBDj-ZQe5o-LdKpB"
+
+
+async def test_store_app_password_replaces_existing(temp_storage):
+    """Test that storing a new app password replaces the existing one."""
+    await temp_storage.store_app_password(
+        user_id="testuser",
+        app_password="aaaaa-bbbbb-ccccc-ddddd-eeeee",
+    )
+    await temp_storage.store_app_password(
+        user_id="testuser",
+        app_password="fffff-ggggg-hhhhh-iiiii-jjjjj",
+    )
+
+    retrieved = await temp_storage.get_app_password("testuser")
+    assert retrieved == "fffff-ggggg-hhhhh-iiiii-jjjjj"
+
+
+async def test_get_app_password_nonexistent(temp_storage):
+    """Test retrieving app password for non-existent user."""
+    retrieved = await temp_storage.get_app_password("nonexistent")
+    assert retrieved is None
+
+
+async def test_delete_app_password(temp_storage):
+    """Test deleting an app password."""
+    await temp_storage.store_app_password(
+        user_id="testuser",
+        app_password="JHWzB-ZYgLZ-3qBDj-ZQe5o-LdKpB",
+    )
+
+    deleted = await temp_storage.delete_app_password("testuser")
+    assert deleted is True
+
+    # Verify it's gone
+    retrieved = await temp_storage.get_app_password("testuser")
+    assert retrieved is None
+
+
+async def test_delete_app_password_nonexistent(temp_storage):
+    """Test deleting non-existent app password."""
+    deleted = await temp_storage.delete_app_password("nonexistent")
+    assert deleted is False
+
+
+async def test_get_all_app_password_user_ids(temp_storage):
+    """Test listing all users with app passwords."""
+    await temp_storage.store_app_password("alice", "aaaaa-aaaaa-aaaaa-aaaaa-aaaaa")
+    await temp_storage.store_app_password("bob", "bbbbb-bbbbb-bbbbb-bbbbb-bbbbb")
+    await temp_storage.store_app_password("charlie", "ccccc-ccccc-ccccc-ccccc-ccccc")
+
+    user_ids = await temp_storage.get_all_app_password_user_ids()
+    assert len(user_ids) == 3
+    assert "alice" in user_ids
+    assert "bob" in user_ids
+    assert "charlie" in user_ids
+
+
+async def test_get_all_app_password_user_ids_empty(temp_storage):
+    """Test listing users when none have app passwords."""
+    user_ids = await temp_storage.get_all_app_password_user_ids()
+    assert len(user_ids) == 0
+
+
+async def test_app_password_encryption(encryption_key):
+    """Test that app passwords are encrypted at rest."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        db_path = Path(tmpdir) / "test_encryption.db"
+        storage = RefreshTokenStorage(
+            db_path=str(db_path), encryption_key=encryption_key
+        )
+        await storage.initialize()
+
+        # Store a password
+        test_password = "JHWzB-ZYgLZ-3qBDj-ZQe5o-LdKpB"
+        await storage.store_app_password("testuser", test_password)
+
+        # Read directly from database to verify encryption
+        import aiosqlite
+
+        async with aiosqlite.connect(str(db_path)) as db:
+            async with db.execute(
+                "SELECT encrypted_password FROM app_passwords WHERE user_id = ?",
+                ("testuser",),
+            ) as cursor:
+                row = await cursor.fetchone()
+
+        # The stored value should be encrypted (not plain text)
+        encrypted_bytes = row[0]
+        assert encrypted_bytes != test_password.encode()
+        # Encrypted data should be longer due to Fernet overhead
+        assert len(encrypted_bytes) > len(test_password)
+
+
+async def test_app_password_requires_encryption_key():
+    """Test that app password operations require encryption key."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        db_path = Path(tmpdir) / "test_no_key.db"
+        storage = RefreshTokenStorage(db_path=str(db_path), encryption_key=None)
+        await storage.initialize()
+
+        # Storing should fail without encryption key
+        with pytest.raises(RuntimeError, match="Encryption key not configured"):
+            await storage.store_app_password(
+                "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+            )
+
+        # Getting should also fail without encryption key
+        with pytest.raises(RuntimeError, match="Encryption key not configured"):
+            await storage.get_app_password("testuser")
+
+
+async def test_multiple_users_independence(temp_storage):
+    """Test that different users maintain independent app passwords."""
+    users = ["alice", "bob", "charlie", "diana"]
+
+    # Store unique passwords for each user
+    for i, user in enumerate(users):
+        password = (
+            f"{user[0]}{user[0]}{user[0]}{user[0]}{user[0]}-" * 4
+            + f"{user[0]}{user[0]}{user[0]}{user[0]}{user[0]}"
+        )
+        await temp_storage.store_app_password(user, password)
+
+    # Verify each user has their correct password
+    for user in users:
+        expected = (
+            f"{user[0]}{user[0]}{user[0]}{user[0]}{user[0]}-" * 4
+            + f"{user[0]}{user[0]}{user[0]}{user[0]}{user[0]}"
+        )
+        retrieved = await temp_storage.get_app_password(user)
+        assert retrieved == expected
+
+    # Delete one user's password
+    await temp_storage.delete_app_password("bob")
+
+    # Verify other users unchanged
+    for user in ["alice", "charlie", "diana"]:
+        retrieved = await temp_storage.get_app_password(user)
+        assert retrieved is not None
+
+    # Verify bob's password is gone
+    assert await temp_storage.get_app_password("bob") is None
+
+
+async def test_app_password_with_special_characters(temp_storage):
+    """Test storing passwords with various alphanumeric patterns."""
+    # Nextcloud app passwords use alphanumeric characters
+    passwords = [
+        "AAAAA-BBBBB-CCCCC-DDDDD-EEEEE",  # uppercase
+        "aaaaa-bbbbb-ccccc-ddddd-eeeee",  # lowercase
+        "12345-67890-12345-67890-12345",  # numbers
+        "aB1cD-eF2gH-iJ3kL-mN4oP-qR5sT",  # mixed
+    ]
+
+    for i, password in enumerate(passwords):
+        user = f"user{i}"
+        await temp_storage.store_app_password(user, password)
+        retrieved = await temp_storage.get_app_password(user)
+        assert retrieved == password
+
+
+async def test_decryption_with_wrong_key(encryption_key):
+    """Test that decryption fails with wrong key."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        db_path = Path(tmpdir) / "test_wrong_key.db"
+
+        # Store with original key
+        storage1 = RefreshTokenStorage(
+            db_path=str(db_path), encryption_key=encryption_key
+        )
+        await storage1.initialize()
+        await storage1.store_app_password("testuser", "JHWzB-ZYgLZ-3qBDj-ZQe5o-LdKpB")
+
+        # Try to read with different key
+        wrong_key = Fernet.generate_key().decode()
+        storage2 = RefreshTokenStorage(db_path=str(db_path), encryption_key=wrong_key)
+        await storage2.initialize()
+
+        # Decryption should fail and return None (graceful handling)
+        retrieved = await storage2.get_app_password("testuser")
+        assert retrieved is None
@@ -0,0 +1,624 @@
+"""
+Unit tests for Management API app password endpoints.
+
+Tests the REST API endpoints for multi-user BasicAuth mode app password management:
+- POST /api/v1/users/{user_id}/app-password - Provision app password
+- GET /api/v1/users/{user_id}/app-password - Check status
+- DELETE /api/v1/users/{user_id}/app-password - Delete app password
+"""
+
+import base64
+import tempfile
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from cryptography.fernet import Fernet
+from starlette.applications import Starlette
+from starlette.routing import Route
+from starlette.testclient import TestClient
+
+from nextcloud_mcp_server.api import management
+from nextcloud_mcp_server.api.management import (
+    delete_app_password,
+    get_app_password_status,
+    provision_app_password,
+)
+from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
+
+pytestmark = pytest.mark.unit
+
+
+@pytest.fixture(autouse=True)
+def clear_rate_limit():
+    """Clear rate limit state before each test."""
+    management._rate_limit_attempts.clear()
+    yield
+    management._rate_limit_attempts.clear()
+
+
+@pytest.fixture
+def encryption_key():
+    """Generate a test encryption key."""
+    return Fernet.generate_key().decode()
+
+
+@pytest.fixture
+async def temp_storage(encryption_key):
+    """Create temporary storage instance with encryption for testing."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        db_path = Path(tmpdir) / "test_management.db"
+        storage = RefreshTokenStorage(
+            db_path=str(db_path), encryption_key=encryption_key
+        )
+        await storage.initialize()
+        yield storage
+
+
+def create_basic_auth_header(username: str, password: str) -> str:
+    """Create BasicAuth header value."""
+    credentials = f"{username}:{password}"
+    encoded = base64.b64encode(credentials.encode()).decode()
+    return f"Basic {encoded}"
+
+
+def create_test_app(storage):
+    """Create a test Starlette app with the management endpoints."""
+    app = Starlette(
+        routes=[
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                provision_app_password,
+                methods=["POST"],
+            ),
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                get_app_password_status,
+                methods=["GET"],
+            ),
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                delete_app_password,
+                methods=["DELETE"],
+            ),
+        ]
+    )
+    app.state.storage = storage
+    return app
+
+
+async def test_provision_app_password_missing_auth():
+    """Test that missing auth returns 401."""
+    app = Starlette(
+        routes=[
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                provision_app_password,
+                methods=["POST"],
+            ),
+        ]
+    )
+
+    client = TestClient(app)
+    response = client.post("/api/v1/users/testuser/app-password")
+
+    assert response.status_code == 401
+    assert "Missing BasicAuth" in response.json()["error"]
+
+
+async def test_provision_app_password_invalid_auth_format():
+    """Test that invalid auth format returns 401."""
+    app = Starlette(
+        routes=[
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                provision_app_password,
+                methods=["POST"],
+            ),
+        ]
+    )
+
+    client = TestClient(app)
+    response = client.post(
+        "/api/v1/users/testuser/app-password",
+        headers={"Authorization": "Basic invalid-not-base64!!!"},
+    )
+
+    assert response.status_code == 401
+    assert "Invalid BasicAuth" in response.json()["error"]
+
+
+async def test_provision_app_password_username_mismatch():
+    """Test that username mismatch returns 403."""
+    app = Starlette(
+        routes=[
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                provision_app_password,
+                methods=["POST"],
+            ),
+        ]
+    )
+
+    client = TestClient(app)
+    # Try to provision for "testuser" but auth as "otheruser"
+    response = client.post(
+        "/api/v1/users/testuser/app-password",
+        headers={
+            "Authorization": create_basic_auth_header(
+                "otheruser", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+            )
+        },
+    )
+
+    assert response.status_code == 403
+    assert "does not match" in response.json()["error"]
+
+
+async def test_provision_app_password_invalid_format():
+    """Test that invalid app password format returns 400."""
+    app = Starlette(
+        routes=[
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                provision_app_password,
+                methods=["POST"],
+            ),
+        ]
+    )
+
+    client = TestClient(app)
+    # Use invalid password format (not xxxxx-xxxxx-xxxxx-xxxxx-xxxxx)
+    response = client.post(
+        "/api/v1/users/testuser/app-password",
+        headers={
+            "Authorization": create_basic_auth_header("testuser", "invalid-password")
+        },
+    )
+
+    assert response.status_code == 400
+    assert "Invalid app password format" in response.json()["error"]
+
+
+async def test_provision_app_password_success(temp_storage, mocker):
+    """Test successful app password provisioning."""
+    # Mock settings (imported locally in the function)
+    mocker.patch(
+        "nextcloud_mcp_server.config.get_settings",
+        return_value=MagicMock(nextcloud_host="http://localhost:8080"),
+    )
+
+    # Mock httpx client for Nextcloud validation
+    mock_response = MagicMock()
+    mock_response.status_code = 200
+    mock_response.json.return_value = {"ocs": {"data": {"id": "testuser"}}}
+
+    mock_client = AsyncMock()
+    mock_client.get = AsyncMock(return_value=mock_response)
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock()
+
+    mocker.patch(
+        "nextcloud_mcp_server.api.management.httpx.AsyncClient",
+        return_value=mock_client,
+    )
+
+    # Create app with storage
+    app = create_test_app(temp_storage)
+
+    client = TestClient(app)
+    response = client.post(
+        "/api/v1/users/testuser/app-password",
+        headers={
+            "Authorization": create_basic_auth_header(
+                "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+            )
+        },
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert "stored" in data["message"].lower()
+
+    # Verify password was stored
+    stored_password = await temp_storage.get_app_password("testuser")
+    assert stored_password == "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+
+
+async def test_provision_app_password_nextcloud_validation_fails(mocker):
+    """Test that failed Nextcloud validation returns 401."""
+    mocker.patch(
+        "nextcloud_mcp_server.config.get_settings",
+        return_value=MagicMock(nextcloud_host="http://localhost:8080"),
+    )
+
+    # Mock httpx client to return 401
+    mock_response = MagicMock()
+    mock_response.status_code = 401
+
+    mock_client = AsyncMock()
+    mock_client.get = AsyncMock(return_value=mock_response)
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock()
+
+    mocker.patch(
+        "nextcloud_mcp_server.api.management.httpx.AsyncClient",
+        return_value=mock_client,
+    )
+
+    app = Starlette(
+        routes=[
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                provision_app_password,
+                methods=["POST"],
+            ),
+        ]
+    )
+
+    client = TestClient(app)
+    response = client.post(
+        "/api/v1/users/testuser/app-password",
+        headers={
+            "Authorization": create_basic_auth_header(
+                "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+            )
+        },
+    )
+
+    assert response.status_code == 401
+    assert "Invalid app password" in response.json()["error"]
+
+
+async def test_get_app_password_status_provisioned(temp_storage, mocker):
+    """Test checking status when app password is provisioned."""
+    # Store an app password
+    await temp_storage.store_app_password("testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee")
+
+    app = create_test_app(temp_storage)
+
+    client = TestClient(app)
+    response = client.get(
+        "/api/v1/users/testuser/app-password",
+        headers={
+            "Authorization": create_basic_auth_header(
+                "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+            )
+        },
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["user_id"] == "testuser"
+    assert data["has_app_password"] is True
+
+
+async def test_get_app_password_status_not_provisioned(temp_storage, mocker):
+    """Test checking status when app password is not provisioned."""
+    app = create_test_app(temp_storage)
+
+    client = TestClient(app)
+    response = client.get(
+        "/api/v1/users/testuser/app-password",
+        headers={
+            "Authorization": create_basic_auth_header(
+                "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+            )
+        },
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["user_id"] == "testuser"
+    assert data["has_app_password"] is False
+
+
+async def test_get_app_password_status_username_mismatch():
+    """Test that username mismatch returns 403 for status check."""
+    app = Starlette(
+        routes=[
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                get_app_password_status,
+                methods=["GET"],
+            ),
+        ]
+    )
+
+    client = TestClient(app)
+    response = client.get(
+        "/api/v1/users/testuser/app-password",
+        headers={
+            "Authorization": create_basic_auth_header(
+                "otheruser", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+            )
+        },
+    )
+
+    assert response.status_code == 403
+
+
+async def test_delete_app_password_success(temp_storage, mocker):
+    """Test successful app password deletion."""
+    # Store an app password
+    await temp_storage.store_app_password("testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee")
+
+    # Mock settings (imported locally in the function)
+    mocker.patch(
+        "nextcloud_mcp_server.config.get_settings",
+        return_value=MagicMock(nextcloud_host="http://localhost:8080"),
+    )
+
+    # Mock httpx client for Nextcloud validation
+    mock_response = MagicMock()
+    mock_response.status_code = 200
+
+    mock_client = AsyncMock()
+    mock_client.get = AsyncMock(return_value=mock_response)
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock()
+
+    mocker.patch(
+        "nextcloud_mcp_server.api.management.httpx.AsyncClient",
+        return_value=mock_client,
+    )
+
+    app = create_test_app(temp_storage)
+
+    client = TestClient(app)
+    response = client.delete(
+        "/api/v1/users/testuser/app-password",
+        headers={
+            "Authorization": create_basic_auth_header(
+                "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+            )
+        },
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert "deleted" in data["message"].lower()
+
+    # Verify password was removed
+    stored_password = await temp_storage.get_app_password("testuser")
+    assert stored_password is None
+
+
+async def test_delete_app_password_not_found(temp_storage, mocker):
+    """Test deleting non-existent app password."""
+    # Mock settings (imported locally in the function)
+    mocker.patch(
+        "nextcloud_mcp_server.config.get_settings",
+        return_value=MagicMock(nextcloud_host="http://localhost:8080"),
+    )
+
+    # Mock httpx client for Nextcloud validation
+    mock_response = MagicMock()
+    mock_response.status_code = 200
+
+    mock_client = AsyncMock()
+    mock_client.get = AsyncMock(return_value=mock_response)
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock()
+
+    mocker.patch(
+        "nextcloud_mcp_server.api.management.httpx.AsyncClient",
+        return_value=mock_client,
+    )
+
+    app = create_test_app(temp_storage)
+
+    client = TestClient(app)
+    response = client.delete(
+        "/api/v1/users/testuser/app-password",
+        headers={
+            "Authorization": create_basic_auth_header(
+                "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+            )
+        },
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert "no app password found" in data["message"].lower()
+
+
+async def test_delete_app_password_invalid_credentials(mocker):
+    """Test that invalid credentials returns 401 for deletion."""
+    mocker.patch(
+        "nextcloud_mcp_server.config.get_settings",
+        return_value=MagicMock(nextcloud_host="http://localhost:8080"),
+    )
+
+    # Mock httpx client to return 401
+    mock_response = MagicMock()
+    mock_response.status_code = 401
+
+    mock_client = AsyncMock()
+    mock_client.get = AsyncMock(return_value=mock_response)
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock()
+
+    mocker.patch(
+        "nextcloud_mcp_server.api.management.httpx.AsyncClient",
+        return_value=mock_client,
+    )
+
+    app = Starlette(
+        routes=[
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                delete_app_password,
+                methods=["DELETE"],
+            ),
+        ]
+    )
+
+    client = TestClient(app)
+    response = client.delete(
+        "/api/v1/users/testuser/app-password",
+        headers={
+            "Authorization": create_basic_auth_header(
+                "testuser", "wrong-password-xxxxx"
+            )
+        },
+    )
+
+    assert response.status_code == 401
+    assert "Invalid credentials" in response.json()["error"]
+
+
+async def test_delete_app_password_username_mismatch():
+    """Test that username mismatch returns 403 for deletion."""
+    app = Starlette(
+        routes=[
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                delete_app_password,
+                methods=["DELETE"],
+            ),
+        ]
+    )
+
+    client = TestClient(app)
+    response = client.delete(
+        "/api/v1/users/testuser/app-password",
+        headers={
+            "Authorization": create_basic_auth_header(
+                "otheruser", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+            )
+        },
+    )
+
+    assert response.status_code == 403
+
+
+async def test_provision_app_password_rate_limiting(mocker):
+    """Test that rate limiting blocks excessive provisioning attempts."""
+    mocker.patch(
+        "nextcloud_mcp_server.config.get_settings",
+        return_value=MagicMock(nextcloud_host="http://localhost:8080"),
+    )
+
+    # Mock httpx client to return 401 (failed validation)
+    mock_response = MagicMock()
+    mock_response.status_code = 401
+
+    mock_client = AsyncMock()
+    mock_client.get = AsyncMock(return_value=mock_response)
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock()
+
+    mocker.patch(
+        "nextcloud_mcp_server.api.management.httpx.AsyncClient",
+        return_value=mock_client,
+    )
+
+    app = Starlette(
+        routes=[
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                provision_app_password,
+                methods=["POST"],
+            ),
+        ]
+    )
+
+    client = TestClient(app)
+
+    # Make 5 failed attempts (should all return 401)
+    for i in range(5):
+        response = client.post(
+            "/api/v1/users/testuser/app-password",
+            headers={
+                "Authorization": create_basic_auth_header(
+                    "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+                )
+            },
+        )
+        assert response.status_code == 401, f"Attempt {i + 1} should return 401"
+
+    # 6th attempt should be rate limited (429)
+    response = client.post(
+        "/api/v1/users/testuser/app-password",
+        headers={
+            "Authorization": create_basic_auth_header(
+                "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+            )
+        },
+    )
+    assert response.status_code == 429
+    assert "Rate limit exceeded" in response.json()["error"]
+    assert "Retry-After" in response.headers
+
+
+async def test_rate_limiting_is_per_user(mocker):
+    """Test that rate limiting is applied per user, not globally."""
+    mocker.patch(
+        "nextcloud_mcp_server.config.get_settings",
+        return_value=MagicMock(nextcloud_host="http://localhost:8080"),
+    )
+
+    # Mock httpx client to return 401
+    mock_response = MagicMock()
+    mock_response.status_code = 401
+
+    mock_client = AsyncMock()
+    mock_client.get = AsyncMock(return_value=mock_response)
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock()
+
+    mocker.patch(
+        "nextcloud_mcp_server.api.management.httpx.AsyncClient",
+        return_value=mock_client,
+    )
+
+    app = Starlette(
+        routes=[
+            Route(
+                "/api/v1/users/{user_id}/app-password",
+                provision_app_password,
+                methods=["POST"],
+            ),
+        ]
+    )
+
+    client = TestClient(app)
+
+    # Make 5 failed attempts for user1 (hits rate limit)
+    for _ in range(5):
+        client.post(
+            "/api/v1/users/user1/app-password",
+            headers={
+                "Authorization": create_basic_auth_header(
+                    "user1", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+                )
+            },
+        )
+
+    # user1 should be rate limited
+    response = client.post(
+        "/api/v1/users/user1/app-password",
+        headers={
+            "Authorization": create_basic_auth_header(
+                "user1", "aaaaa-bbbbb-ccccc-ddddd-eeeee"
+            )
+        },
+    )
+    assert response.status_code == 429
+
+    # user2 should NOT be rate limited (different user)
+    response = client.post(
+        "/api/v1/users/user2/app-password",
+        headers={
+            "Authorization": create_basic_auth_header(
+                "user2", "bbbbb-ccccc-ddddd-eeeee-fffff"
+            )
+        },
+    )
+    assert response.status_code == 401  # Fails validation, but not rate limited
@@ -94,24 +94,90 @@ class CredentialsController extends Controller {
 			], Http::STATUS_UNAUTHORIZED);
 		}

-		// Store encrypted app password
+		// Store encrypted app password locally in Nextcloud
 		try {
 			$this->tokenStorage->storeBackgroundSyncPassword($userId, $appPassword);
-			$this->logger->info("Successfully stored app password for user: $userId");
-
-			return new JSONResponse([
-				'success' => true,
-				'message' => 'App password saved successfully'
-			], Http::STATUS_OK);
+			$this->logger->info("Stored app password locally for user: $userId");
 		} catch (\Exception $e) {
-			$this->logger->error("Failed to store app password for user $userId", [
+			$this->logger->error("Failed to store app password locally for user $userId", [
 				'error' => $e->getMessage()
 			]);
 			return new JSONResponse([
 				'success' => false,
-				'error' => 'Failed to save app password'
+				'error' => 'Failed to save app password locally'
 			], Http::STATUS_INTERNAL_SERVER_ERROR);
 		}
+
+		// Send app password to MCP server for background sync
+		// Get MCP server URL from system config (set in config.php)
+		$mcpServerUrl = $this->config->getSystemValue('mcp_server_url', '');
+		if (empty($mcpServerUrl)) {
+			$this->logger->warning("MCP server URL not configured, app password stored locally only");
+			return new JSONResponse([
+				'success' => true,
+				'partial_success' => true,
+				'local_storage' => true,
+				'mcp_sync' => false,
+				'message' => 'App password saved locally (MCP server not configured)'
+			], Http::STATUS_OK);
+		}
+
+		try {
+			$httpClient = $this->httpClientService->newClient();
+
+			// Send to MCP server with BasicAuth (user proves ownership of password)
+			$mcpEndpoint = rtrim($mcpServerUrl, '/') . '/api/v1/users/' . urlencode($userId) . '/app-password';
+
+			$this->logger->debug("Sending app password to MCP server: $mcpEndpoint");
+
+			$response = $httpClient->post($mcpEndpoint, [
+				'auth' => [$userId, $appPassword],
+				'headers' => [
+					'Content-Type' => 'application/json',
+					'Accept' => 'application/json',
+				],
+				'timeout' => 10,
+			]);
+
+			$statusCode = $response->getStatusCode();
+			$body = json_decode($response->getBody(), true);
+
+			if ($statusCode === 200 && ($body['success'] ?? false)) {
+				$this->logger->info("Successfully provisioned app password to MCP server for user: $userId");
+				return new JSONResponse([
+					'success' => true,
+					'partial_success' => false,
+					'local_storage' => true,
+					'mcp_sync' => true,
+					'message' => 'App password saved successfully'
+				], Http::STATUS_OK);
+			} else {
+				$error = $body['error'] ?? 'Unknown error';
+				$this->logger->error("MCP server rejected app password for user $userId: $error");
+				// Return partial success since it was stored locally but MCP sync failed
+				return new JSONResponse([
+					'success' => true,
+					'partial_success' => true,
+					'local_storage' => true,
+					'mcp_sync' => false,
+					'message' => 'App password saved locally (MCP server sync failed)',
+					'mcp_error' => $error
+				], Http::STATUS_OK);
+			}
+		} catch (\Exception $e) {
+			$this->logger->error("Failed to send app password to MCP server for user $userId", [
+				'error' => $e->getMessage()
+			]);
+			// Return partial success since it was stored locally but MCP was unreachable
+			return new JSONResponse([
+				'success' => true,
+				'partial_success' => true,
+				'local_storage' => true,
+				'mcp_sync' => false,
+				'message' => 'App password saved locally (MCP server unreachable)',
+				'mcp_error' => $e->getMessage()
+			], Http::STATUS_OK);
+		}
 	}

 	/**