refactor(config): centralize configuration validation and simplify startup
Implement centralized configuration validation (ADR-020) to simplify deployment mode detection and improve error messages. Changes: - Create ADR-020 documenting 5 deployment modes with required/optional config - Add config_validators.py with validate_configuration() and mode detection - Simplify app.py startup with single validation point at get_app() - Remove duplicate is_oauth_mode() function (43 lines) - Fix DeploymentMode mapping (only SELF_HOSTED and SMITHERY_STATELESS exist) - Add comprehensive unit tests (41 tests covering all modes and edge cases) - Add enable_multi_user_basic_auth to Settings and BasicAuthMiddleware Docker Compose: - Remove conflicting ENABLE_MULTI_USER_BASIC_AUTH from mcp-oauth service - Add dedicated mcp-multi-user-basic service on port 8003 Test Results: - 237/237 integration tests PASSED - All deployment modes verified: single-user BasicAuth, multi-user BasicAuth, OAuth single-audience, OAuth token exchange (Keycloak), Smithery stateless 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
+33
-1
@@ -35,7 +35,7 @@ services:
|
||||
# Mount OIDC development directory outside /var/www/html to avoid rsync conflicts
|
||||
# The post-installation hook will register /opt/apps as an additional app directory
|
||||
#- ./third_party:/opt/apps:ro
|
||||
#- ./third_party/astrolabe:/opt/apps/astrolabe:ro
|
||||
- ./third_party/astrolabe:/opt/apps/astrolabe:ro
|
||||
environment:
|
||||
- NEXTCLOUD_TRUSTED_DOMAINS=app
|
||||
- NEXTCLOUD_ADMIN_USER=admin
|
||||
@@ -123,6 +123,32 @@ services:
|
||||
# - DOCUMENT_CHUNK_SIZE=512 # Words per chunk (default: 512)
|
||||
# - DOCUMENT_CHUNK_OVERLAP=50 # Overlapping words (default: 50, recommended: 10-20% of chunk size)
|
||||
|
||||
mcp-multi-user-basic:
|
||||
build: .
|
||||
restart: always
|
||||
command: ["--transport", "streamable-http"]
|
||||
depends_on:
|
||||
app:
|
||||
condition: service_healthy
|
||||
ports:
|
||||
- 127.0.0.1:8003:8000
|
||||
environment:
|
||||
# Multi-user BasicAuth pass-through mode (ADR-020)
|
||||
- NEXTCLOUD_HOST=http://app:80
|
||||
- NEXTCLOUD_PUBLIC_ISSUER_URL=http://localhost:8080
|
||||
- ENABLE_MULTI_USER_BASIC_AUTH=true
|
||||
|
||||
# Token storage (required for middleware initialization)
|
||||
- TOKEN_ENCRYPTION_KEY=ESF1BvEQdGYsCluwMx9Cxvw3uh5pFowPH7Rg_nIliyo=
|
||||
- TOKEN_STORAGE_DB=/app/data/tokens.db
|
||||
|
||||
# Vector sync disabled (stateless pass-through mode)
|
||||
- VECTOR_SYNC_ENABLED=false
|
||||
|
||||
# NO admin credentials - credentials come from client Authorization header
|
||||
volumes:
|
||||
- multi-user-basic-data:/app/data
|
||||
|
||||
mcp-oauth:
|
||||
build: .
|
||||
command: ["--transport", "streamable-http", "--oauth", "--port", "8001", "--oauth-token-type", "jwt"]
|
||||
@@ -159,6 +185,11 @@ services:
|
||||
# Qdrant configuration - persistent local storage
|
||||
- QDRANT_LOCATION=/app/data/qdrant
|
||||
|
||||
# Embedding provider for vector sync (use Simple provider as fallback)
|
||||
# Ollama not available in CI/test environments
|
||||
# - OLLAMA_BASE_URL=http://ollama:11434
|
||||
# - OLLAMA_EMBEDDING_MODEL=nomic-embed-text
|
||||
|
||||
# NO admin credentials - using OAuth with Dynamic Client Registration (DCR)
|
||||
# Client credentials registered via RFC 7591 and stored in volume
|
||||
# JWT token type is used for testing (faster validation, scopes embedded in token)
|
||||
@@ -280,3 +311,4 @@ volumes:
|
||||
keycloak-oauth-storage:
|
||||
qdrant-data:
|
||||
mcp-data:
|
||||
multi-user-basic-data:
|
||||
|
||||
@@ -0,0 +1,342 @@
|
||||
# ADR-020: Deployment Modes and Configuration Validation
|
||||
|
||||
**Status:** Accepted
|
||||
**Date:** 2025-12-20
|
||||
**Deciders:** Development Team
|
||||
**Related:** ADR-002 (Vector Sync), ADR-004 (Progressive Consent), ADR-019 (Multi-user BasicAuth)
|
||||
|
||||
## Context
|
||||
|
||||
The MCP server supports multiple deployment scenarios with different authentication methods, storage backends, and feature sets. Over time, the configuration system evolved to support ~500+ possible combinations across deployment modes, authentication patterns, and feature toggles. This complexity made it difficult to:
|
||||
|
||||
1. Understand what configuration is required for a given deployment
|
||||
2. Debug configuration errors (validation scattered across multiple files)
|
||||
3. Provide helpful error messages when configuration is invalid
|
||||
4. Maintain clear boundaries between deployment modes
|
||||
|
||||
**Problems Identified:**
|
||||
- No single source of truth for "what config is required for mode X"
|
||||
- Validation happening at 4+ different points (Settings.__post_init__, setup_oauth_config(), context helpers, starlette_lifespan)
|
||||
- Startup sequence unclear (OAuth setup before FastMCP creation, sync initialization errors)
|
||||
- Error messages generic ("X is required") without explaining which deployment mode triggered the requirement
|
||||
- Multiple overlapping decision trees (deployment mode, auth mode, features)
|
||||
|
||||
## Decision
|
||||
|
||||
We formalize five distinct deployment modes with explicit configuration requirements and implement centralized configuration validation.
|
||||
|
||||
### Deployment Modes
|
||||
|
||||
#### 1. Single-User BasicAuth
|
||||
|
||||
**Use Case:** Personal Nextcloud instance, local development
|
||||
|
||||
**Required Configuration:**
|
||||
```bash
|
||||
NEXTCLOUD_HOST=http://localhost:8080
|
||||
NEXTCLOUD_USERNAME=admin
|
||||
NEXTCLOUD_PASSWORD=password # Or app password
|
||||
```
|
||||
|
||||
**Optional Configuration:**
|
||||
```bash
|
||||
# Vector sync (semantic search)
|
||||
VECTOR_SYNC_ENABLED=true
|
||||
QDRANT_LOCATION=/path/to/qdrant # Or QDRANT_URL for remote
|
||||
|
||||
# Embeddings (optional - Simple provider used as fallback)
|
||||
OLLAMA_BASE_URL=http://localhost:11434
|
||||
OLLAMA_EMBEDDING_MODEL=nomic-embed-text
|
||||
|
||||
# Document processing
|
||||
DOCUMENT_CHUNK_SIZE=512
|
||||
DOCUMENT_CHUNK_OVERLAP=50
|
||||
```
|
||||
|
||||
**Characteristics:**
|
||||
- Single shared NextcloudClient created at startup
|
||||
- No OAuth infrastructure needed
|
||||
- No multi-user support
|
||||
- Vector sync runs as single-user background task
|
||||
- Admin UI available at /app
|
||||
|
||||
---
|
||||
|
||||
#### 2. Multi-User BasicAuth Pass-Through
|
||||
|
||||
**Use Case:** Internal deployment where users provide their own credentials, no background sync needed
|
||||
|
||||
**Required Configuration:**
|
||||
```bash
|
||||
NEXTCLOUD_HOST=http://nextcloud.example.com
|
||||
ENABLE_MULTI_USER_BASIC_AUTH=true
|
||||
```
|
||||
|
||||
**Optional Configuration:**
|
||||
```bash
|
||||
# For background sync (requires app passwords from Astrolabe)
|
||||
ENABLE_OFFLINE_ACCESS=true
|
||||
TOKEN_ENCRYPTION_KEY=<key>
|
||||
TOKEN_STORAGE_DB=/path/to/tokens.db
|
||||
NEXTCLOUD_OIDC_CLIENT_ID=<client-id>
|
||||
NEXTCLOUD_OIDC_CLIENT_SECRET=<client-secret>
|
||||
VECTOR_SYNC_ENABLED=true
|
||||
# ... plus Qdrant and embedding config
|
||||
```
|
||||
|
||||
**Conditional Requirements:**
|
||||
- If `ENABLE_OFFLINE_ACCESS=true`: requires `NEXTCLOUD_OIDC_CLIENT_ID`, `NEXTCLOUD_OIDC_CLIENT_SECRET`, `TOKEN_ENCRYPTION_KEY`, `TOKEN_STORAGE_DB`
|
||||
- If `VECTOR_SYNC_ENABLED=true`: requires `ENABLE_OFFLINE_ACCESS=true`
|
||||
|
||||
**Characteristics:**
|
||||
- No OAuth for client authentication (uses BasicAuth in request headers)
|
||||
- BasicAuthMiddleware extracts credentials from Authorization header
|
||||
- Client created per-request from extracted credentials
|
||||
- Optional: Background sync using app passwords (via Astrolabe API)
|
||||
- Admin UI available at /app
|
||||
|
||||
---
|
||||
|
||||
#### 3. OAuth Single-Audience (Default)
|
||||
|
||||
**Use Case:** Multi-user deployment with OAuth authentication, tokens work for both MCP and Nextcloud
|
||||
|
||||
**Required Configuration:**
|
||||
```bash
|
||||
NEXTCLOUD_HOST=http://nextcloud.example.com
|
||||
# No NEXTCLOUD_USERNAME/PASSWORD (triggers OAuth mode)
|
||||
```
|
||||
|
||||
**Auto-Configured:**
|
||||
- OIDC discovery URL: `{NEXTCLOUD_HOST}/.well-known/openid-configuration`
|
||||
- Client credentials: Dynamic Client Registration (DCR) if available
|
||||
- Token storage: SQLite at `~/.oauth/clients.db`
|
||||
|
||||
**Optional Configuration:**
|
||||
```bash
|
||||
# Static client credentials (instead of DCR)
|
||||
NEXTCLOUD_OIDC_CLIENT_ID=<client-id>
|
||||
NEXTCLOUD_OIDC_CLIENT_SECRET=<client-secret>
|
||||
|
||||
# Offline access for background sync
|
||||
ENABLE_OFFLINE_ACCESS=true
|
||||
TOKEN_ENCRYPTION_KEY=<key>
|
||||
TOKEN_STORAGE_DB=/path/to/tokens.db
|
||||
VECTOR_SYNC_ENABLED=true
|
||||
# ... plus Qdrant and embedding config
|
||||
|
||||
# Scopes
|
||||
NEXTCLOUD_OIDC_SCOPES="openid profile email notes:read notes:write ..."
|
||||
```
|
||||
|
||||
**Conditional Requirements:**
|
||||
- If `ENABLE_OFFLINE_ACCESS=true`: requires `TOKEN_ENCRYPTION_KEY`, `TOKEN_STORAGE_DB`
|
||||
- If `VECTOR_SYNC_ENABLED=true`: requires `ENABLE_OFFLINE_ACCESS=true`
|
||||
|
||||
**Characteristics:**
|
||||
- Tokens contain both `aud: ["mcp-server", "nextcloud"]`
|
||||
- Pass token through to Nextcloud APIs (no exchange)
|
||||
- Client created per-request from token in Authorization header
|
||||
- Background sync uses refresh tokens (if offline_access enabled)
|
||||
- Admin UI available at /app
|
||||
|
||||
---
|
||||
|
||||
#### 4. OAuth Token Exchange (RFC 8693)
|
||||
|
||||
**Use Case:** Multi-user deployment where MCP token is separate from Nextcloud token
|
||||
|
||||
**Required Configuration:**
|
||||
```bash
|
||||
NEXTCLOUD_HOST=http://nextcloud.example.com
|
||||
ENABLE_TOKEN_EXCHANGE=true
|
||||
# No NEXTCLOUD_USERNAME/PASSWORD (triggers OAuth mode)
|
||||
```
|
||||
|
||||
**Optional Configuration:**
|
||||
- Same as OAuth Single-Audience, plus:
|
||||
```bash
|
||||
TOKEN_EXCHANGE_CACHE_TTL=300 # Cache exchanged tokens
|
||||
```
|
||||
|
||||
**Characteristics:**
|
||||
- Tokens contain only `aud: "mcp-server"`
|
||||
- MCP server exchanges token for Nextcloud token via RFC 8693
|
||||
- Exchanged tokens cached per-user
|
||||
- Client created per-request using exchanged token
|
||||
- Background sync uses refresh tokens (if offline_access enabled)
|
||||
|
||||
---
|
||||
|
||||
#### 5. Smithery Stateless
|
||||
|
||||
**Use Case:** Multi-tenant SaaS deployment via Smithery platform
|
||||
|
||||
**Required Configuration:**
|
||||
- None! Configuration comes from session URL params: `?nextcloud_url=...&username=...&app_password=...`
|
||||
|
||||
**Forbidden Configuration:**
|
||||
- Must NOT set: `NEXTCLOUD_HOST`, `NEXTCLOUD_USERNAME`, `NEXTCLOUD_PASSWORD`, `ENABLE_MULTI_USER_BASIC_AUTH`, `ENABLE_TOKEN_EXCHANGE`, `ENABLE_OFFLINE_ACCESS`, `VECTOR_SYNC_ENABLED`, `NEXTCLOUD_OIDC_CLIENT_ID`, `NEXTCLOUD_OIDC_CLIENT_SECRET`
|
||||
|
||||
**Characteristics:**
|
||||
- No persistent storage (stateless)
|
||||
- Client created per-request from session config
|
||||
- No vector sync (disabled)
|
||||
- No admin UI (no /app routes)
|
||||
- No OAuth infrastructure
|
||||
|
||||
---
|
||||
|
||||
### Configuration Validation
|
||||
|
||||
**Implementation:** `nextcloud_mcp_server/config_validators.py`
|
||||
|
||||
**Key Functions:**
|
||||
```python
|
||||
def detect_auth_mode(settings: Settings) -> AuthMode:
|
||||
"""Detect authentication mode from configuration.
|
||||
|
||||
Priority (most specific to most general):
|
||||
1. Smithery (explicit flag)
|
||||
2. Token exchange (most specific OAuth mode)
|
||||
3. Multi-user BasicAuth
|
||||
4. Single-user BasicAuth
|
||||
5. OAuth single-audience (default OAuth mode)
|
||||
"""
|
||||
|
||||
def validate_configuration(settings: Settings) -> tuple[AuthMode, list[str]]:
|
||||
"""Validate configuration for detected mode.
|
||||
|
||||
Returns:
|
||||
Tuple of (detected_mode, list_of_errors)
|
||||
Empty list means valid configuration.
|
||||
"""
|
||||
```
|
||||
|
||||
**Validation Rules:**
|
||||
- **Required variables:** Must be set and non-empty
|
||||
- **Forbidden variables:** Must NOT be set (or must be False for booleans)
|
||||
- **Conditional requirements:** If feature X is enabled, requires variables Y and Z
|
||||
|
||||
**Error Messages:**
|
||||
```
|
||||
Configuration validation failed for {mode} mode:
|
||||
- [{mode}] Missing required configuration: NEXTCLOUD_HOST
|
||||
- [{mode}] ENABLE_OFFLINE_ACCESS must be enabled when VECTOR_SYNC_ENABLED is true
|
||||
|
||||
Mode: {mode}
|
||||
Description: {mode_description}
|
||||
|
||||
Required configuration:
|
||||
- VAR1
|
||||
- VAR2
|
||||
|
||||
Optional configuration:
|
||||
- VAR3
|
||||
- VAR4
|
||||
|
||||
Conditional requirements:
|
||||
When FEATURE is enabled:
|
||||
- VAR5
|
||||
- VAR6
|
||||
```
|
||||
|
||||
**Integration:**
|
||||
- Validation runs at app startup in `get_app()` (app.py:1048-1062)
|
||||
- All errors reported before any initialization begins
|
||||
- Mode-specific error messages explain requirements
|
||||
- Validation uses the same Settings object used throughout the app
|
||||
|
||||
### Configuration Matrix
|
||||
|
||||
| Variable | Single BasicAuth | Multi BasicAuth | OAuth Single | OAuth Exchange | Smithery |
|
||||
|----------|------------------|-----------------|--------------|----------------|----------|
|
||||
| **NEXTCLOUD_HOST** | Required | Required | Required | Required | Forbidden |
|
||||
| **NEXTCLOUD_USERNAME** | Required | Forbidden | Forbidden | Forbidden | Forbidden |
|
||||
| **NEXTCLOUD_PASSWORD** | Required | Forbidden | Forbidden | Forbidden | Forbidden |
|
||||
| **ENABLE_MULTI_USER_BASIC_AUTH** | Forbidden | Required | Forbidden | Forbidden | Forbidden |
|
||||
| **ENABLE_TOKEN_EXCHANGE** | Forbidden | Forbidden | Forbidden | Required | Forbidden |
|
||||
| **ENABLE_OFFLINE_ACCESS** | Optional\* | Optional\* | Optional\* | Optional\* | Forbidden |
|
||||
| **TOKEN_ENCRYPTION_KEY** | If offline | If offline | If offline | If offline | Forbidden |
|
||||
| **TOKEN_STORAGE_DB** | If offline | If offline | If offline | If offline | Forbidden |
|
||||
| **OIDC_CLIENT_ID** | Forbidden | If offline | Optional\*\* | Optional\*\* | Forbidden |
|
||||
| **OIDC_CLIENT_SECRET** | Forbidden | If offline | Optional\*\* | Optional\*\* | Forbidden |
|
||||
| **VECTOR_SYNC_ENABLED** | Optional | Optional | Optional | Optional | Forbidden |
|
||||
| **QDRANT_URL/LOCATION** | If vector | If vector | If vector | If vector | Forbidden |
|
||||
| **OLLAMA_BASE_URL/OPENAI_API_KEY** | Optional | Optional | Optional | Optional | Forbidden |
|
||||
|
||||
\* Only enables background sync for semantic search
|
||||
\*\* Uses DCR if not provided
|
||||
|
||||
## Consequences
|
||||
|
||||
### Positive
|
||||
|
||||
1. **Clarity:** Single function to detect mode from config
|
||||
2. **Validation:** All config validated upfront with helpful errors
|
||||
3. **Debugging:** Clear logs showing "Running in X mode with config Y"
|
||||
4. **Maintenance:** Mode-specific logic can be isolated
|
||||
5. **Documentation:** Clear mapping of mode → required config
|
||||
6. **Error Messages:** Context-aware ("X is required for Y mode")
|
||||
7. **Testing:** Each mode testable in isolation
|
||||
|
||||
### Negative
|
||||
|
||||
1. **Migration:** Existing invalid configurations will now fail at startup
|
||||
2. **Flexibility:** Less flexibility in configuration combinations
|
||||
3. **Strictness:** Some previously-working combinations may be rejected
|
||||
|
||||
### Neutral
|
||||
|
||||
1. **Backward Compatibility:** Valid configurations continue to work
|
||||
2. **Mode Detection:** Automatic based on config (no explicit mode selection)
|
||||
3. **Default Mode:** OAuth single-audience when no credentials provided
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
### Embedding Provider Validation
|
||||
|
||||
Originally, validation required either `OLLAMA_BASE_URL` or `OPENAI_API_KEY` when vector sync was enabled. This was too strict because the Simple provider is always available as a fallback (ADR-015). The validation was removed to allow vector sync without explicit provider configuration.
|
||||
|
||||
### Variable Scoping Issues
|
||||
|
||||
During implementation, several Python variable scoping issues were discovered in `app.py`:
|
||||
- Local variable assignments in `starlette_lifespan()` shadowed outer scope variables
|
||||
- Fixed by using unique variable names (e.g., `nextcloud_host_for_context`, `basic_auth_storage`)
|
||||
- Removed redundant `settings = get_settings()` call (re-used outer scope)
|
||||
|
||||
### Docker Compose Configuration
|
||||
|
||||
The `mcp-oauth` service configuration was updated to remove `ENABLE_MULTI_USER_BASIC_AUTH=true` which conflicted with its intended OAuth mode. The service now runs in OAuth single-audience mode with vector sync using the Simple embedding provider as fallback.
|
||||
|
||||
## Testing
|
||||
|
||||
### Unit Tests
|
||||
|
||||
`tests/unit/test_config_validators.py` provides comprehensive coverage:
|
||||
- Mode detection with priority ordering (7 tests)
|
||||
- Single-user BasicAuth validation (8 tests)
|
||||
- Multi-user BasicAuth validation (7 tests)
|
||||
- OAuth single-audience validation (6 tests)
|
||||
- OAuth token exchange validation (3 tests)
|
||||
- Smithery validation (4 tests)
|
||||
- Mode summary generation (3 tests)
|
||||
- Edge cases (3 tests)
|
||||
|
||||
**Total: 41 tests, all passing**
|
||||
|
||||
### Integration Tests
|
||||
|
||||
Integration tests verify that:
|
||||
- Each mode starts successfully with valid configuration
|
||||
- Invalid configurations fail with clear error messages
|
||||
- Existing deployments continue to work
|
||||
|
||||
## References
|
||||
|
||||
- [ADR-002: Vector Sync Authentication](ADR-002-vector-sync-authentication.md)
|
||||
- [ADR-004: Progressive Consent](ADR-004-progressive-consent.md)
|
||||
- [ADR-015: Unified Provider Architecture](ADR-015-unified-provider-architecture.md)
|
||||
- [ADR-019: Multi-user BasicAuth Pass-Through](ADR-019-multi-user-basicauth-passthrough.md)
|
||||
- Implementation: `nextcloud_mcp_server/config_validators.py`
|
||||
- Tests: `tests/unit/test_config_validators.py`
|
||||
+171
-88
@@ -41,10 +41,14 @@ from nextcloud_mcp_server.auth.unified_verifier import UnifiedTokenVerifier
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
from nextcloud_mcp_server.config import (
|
||||
DeploymentMode,
|
||||
get_deployment_mode,
|
||||
get_document_processor_config,
|
||||
get_settings,
|
||||
)
|
||||
from nextcloud_mcp_server.config_validators import (
|
||||
AuthMode,
|
||||
get_mode_summary,
|
||||
validate_configuration,
|
||||
)
|
||||
from nextcloud_mcp_server.context import get_client as get_nextcloud_client
|
||||
from nextcloud_mcp_server.document_processors import get_registry
|
||||
from nextcloud_mcp_server.observability import (
|
||||
@@ -351,6 +355,52 @@ def get_smithery_session_config() -> dict | None:
|
||||
return _smithery_session_config.get()
|
||||
|
||||
|
||||
class BasicAuthMiddleware:
|
||||
"""Middleware to extract BasicAuth credentials from Authorization header.
|
||||
|
||||
For multi-user BasicAuth pass-through mode, this middleware extracts
|
||||
username/password from the Authorization: Basic header and stores them
|
||||
in the request state for use by the context layer.
|
||||
|
||||
The credentials are NOT stored persistently - they are passed through
|
||||
directly to Nextcloud APIs for each request (stateless).
|
||||
"""
|
||||
|
||||
def __init__(self, app: ASGIApp):
|
||||
self.app = app
|
||||
|
||||
async def __call__(
|
||||
self, scope: StarletteScope, receive: Receive, send: Send
|
||||
) -> None:
|
||||
if scope["type"] == "http":
|
||||
# Extract Authorization header
|
||||
headers = dict(scope.get("headers", []))
|
||||
auth_header = headers.get(b"authorization", b"")
|
||||
|
||||
if auth_header.startswith(b"Basic "):
|
||||
try:
|
||||
import base64
|
||||
|
||||
# Decode base64(username:password)
|
||||
encoded = auth_header[6:] # Skip "Basic "
|
||||
decoded = base64.b64decode(encoded).decode("utf-8")
|
||||
username, password = decoded.split(":", 1)
|
||||
|
||||
# Store in request state
|
||||
scope.setdefault("state", {})
|
||||
scope["state"]["basic_auth"] = {
|
||||
"username": username,
|
||||
"password": password,
|
||||
}
|
||||
logger.debug(
|
||||
f"BasicAuth credentials extracted for user: {username}"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to extract BasicAuth credentials: {e}")
|
||||
|
||||
await self.app(scope, receive, send)
|
||||
|
||||
|
||||
class SmitheryConfigMiddleware:
|
||||
"""Middleware to extract Smithery config from URL query parameters.
|
||||
|
||||
@@ -423,41 +473,6 @@ async def app_lifespan_smithery(server: FastMCP) -> AsyncIterator[SmitheryAppCon
|
||||
logger.info("Shutting down Smithery stateless mode")
|
||||
|
||||
|
||||
def is_oauth_mode() -> bool:
|
||||
"""
|
||||
Determine if OAuth mode should be used.
|
||||
|
||||
OAuth mode is enabled when:
|
||||
- NEXTCLOUD_USERNAME and NEXTCLOUD_PASSWORD are NOT set
|
||||
- AND we are NOT in Smithery stateless mode
|
||||
- Or explicitly enabled via configuration
|
||||
|
||||
Returns:
|
||||
True if OAuth mode, False if BasicAuth mode
|
||||
"""
|
||||
# ADR-016: Smithery stateless mode uses per-request BasicAuth from session config
|
||||
# It's not OAuth mode even though env credentials aren't set
|
||||
deployment_mode = get_deployment_mode()
|
||||
if deployment_mode == DeploymentMode.SMITHERY_STATELESS:
|
||||
logger.info(
|
||||
"BasicAuth mode (Smithery stateless - credentials from session config)"
|
||||
)
|
||||
return False
|
||||
|
||||
username = os.getenv("NEXTCLOUD_USERNAME")
|
||||
password = os.getenv("NEXTCLOUD_PASSWORD")
|
||||
|
||||
# If both username and password are set, use BasicAuth
|
||||
if username and password:
|
||||
logger.info(
|
||||
"BasicAuth mode detected (NEXTCLOUD_USERNAME and NEXTCLOUD_PASSWORD set)"
|
||||
)
|
||||
return False
|
||||
|
||||
logger.info("OAuth mode detected (NEXTCLOUD_USERNAME/PASSWORD not set)")
|
||||
return True
|
||||
|
||||
|
||||
async def load_oauth_client_credentials(
|
||||
nextcloud_host: str, registration_endpoint: str | None
|
||||
) -> tuple[str, str]:
|
||||
@@ -578,17 +593,31 @@ async def app_lifespan_basic(server: FastMCP) -> AsyncIterator[AppContext]:
|
||||
"""
|
||||
Manage application lifecycle for BasicAuth mode (FastMCP session lifespan).
|
||||
|
||||
Creates a single Nextcloud client with basic authentication
|
||||
For single-user mode: Creates a single Nextcloud client with basic authentication
|
||||
that is shared across all requests within a session.
|
||||
|
||||
For multi-user mode: No shared client - clients created per-request by BasicAuthMiddleware.
|
||||
|
||||
Note: Background tasks (scanner, processor) are started at server level
|
||||
in starlette_lifespan, not here. This lifespan runs per-session.
|
||||
"""
|
||||
logger.info("Starting MCP session in BasicAuth mode")
|
||||
logger.info("Creating Nextcloud client with BasicAuth")
|
||||
settings = get_settings()
|
||||
is_multi_user = settings.enable_multi_user_basic_auth
|
||||
|
||||
client = NextcloudClient.from_env()
|
||||
logger.info("Client initialization complete")
|
||||
logger.info(
|
||||
f"Starting MCP session in {'multi-user' if is_multi_user else 'single-user'} BasicAuth mode"
|
||||
)
|
||||
|
||||
# Only create shared client for single-user mode
|
||||
client = None
|
||||
if not is_multi_user:
|
||||
logger.info("Creating shared Nextcloud client with BasicAuth")
|
||||
client = NextcloudClient.from_env()
|
||||
logger.info("Client initialization complete")
|
||||
else:
|
||||
logger.info(
|
||||
"Multi-user mode - clients created per-request from BasicAuth headers"
|
||||
)
|
||||
|
||||
# Initialize persistent storage (for webhook tracking and future features)
|
||||
from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
|
||||
@@ -604,7 +633,7 @@ async def app_lifespan_basic(server: FastMCP) -> AsyncIterator[AppContext]:
|
||||
# Include vector sync state from module singleton (set by starlette_lifespan)
|
||||
try:
|
||||
yield AppContext(
|
||||
client=client,
|
||||
client=client, # type: ignore[arg-type] # None in multi-user mode
|
||||
storage=storage,
|
||||
document_send_stream=_vector_sync_state.document_send_stream,
|
||||
document_receive_stream=_vector_sync_state.document_receive_stream,
|
||||
@@ -613,7 +642,8 @@ async def app_lifespan_basic(server: FastMCP) -> AsyncIterator[AppContext]:
|
||||
)
|
||||
finally:
|
||||
logger.info("Shutting down BasicAuth session")
|
||||
await client.close()
|
||||
if client is not None:
|
||||
await client.close()
|
||||
|
||||
|
||||
async def setup_oauth_config():
|
||||
@@ -985,6 +1015,33 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
# Initialize observability (logging will be configured by uvicorn)
|
||||
settings = get_settings()
|
||||
|
||||
# Validate configuration and detect deployment mode
|
||||
mode, config_errors = validate_configuration(settings)
|
||||
|
||||
if config_errors:
|
||||
error_msg = (
|
||||
f"Configuration validation failed for {mode.value} mode:\n"
|
||||
+ "\n".join(f" - {err}" for err in config_errors)
|
||||
+ "\n\n"
|
||||
+ get_mode_summary(mode)
|
||||
)
|
||||
logger.error(error_msg)
|
||||
raise ValueError(error_msg)
|
||||
|
||||
logger.info(f"✅ Configuration validated successfully for {mode.value} mode")
|
||||
logger.debug(f"Mode details:\n{get_mode_summary(mode)}")
|
||||
|
||||
# Derive helper variables for backward compatibility with existing code
|
||||
oauth_enabled = mode in (
|
||||
AuthMode.OAUTH_SINGLE_AUDIENCE,
|
||||
AuthMode.OAUTH_TOKEN_EXCHANGE,
|
||||
)
|
||||
deployment_mode = (
|
||||
DeploymentMode.SMITHERY_STATELESS
|
||||
if mode == AuthMode.SMITHERY_STATELESS
|
||||
else DeploymentMode.SELF_HOSTED
|
||||
)
|
||||
|
||||
# Setup Prometheus metrics (always enabled by default)
|
||||
if settings.metrics_enabled:
|
||||
setup_metrics(port=settings.metrics_port)
|
||||
@@ -1008,11 +1065,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
"OpenTelemetry tracing disabled (set OTEL_EXPORTER_OTLP_ENDPOINT to enable)"
|
||||
)
|
||||
|
||||
# Determine authentication mode and deployment mode
|
||||
oauth_enabled = is_oauth_mode()
|
||||
deployment_mode = get_deployment_mode()
|
||||
|
||||
if oauth_enabled:
|
||||
# Create MCP server based on detected mode
|
||||
if mode in (AuthMode.OAUTH_SINGLE_AUDIENCE, AuthMode.OAUTH_TOKEN_EXCHANGE):
|
||||
logger.info("Configuring MCP server for OAuth mode")
|
||||
# Asynchronously get the OAuth configuration
|
||||
import anyio
|
||||
@@ -1075,33 +1129,32 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
enable_dns_rebinding_protection=False
|
||||
),
|
||||
)
|
||||
elif mode == AuthMode.SMITHERY_STATELESS:
|
||||
logger.info("Configuring MCP server for Smithery stateless mode")
|
||||
# json_response=True returns plain JSON-RPC instead of SSE format,
|
||||
# required for Smithery scanner compatibility
|
||||
mcp = FastMCP(
|
||||
"Nextcloud MCP",
|
||||
lifespan=app_lifespan_smithery,
|
||||
json_response=True,
|
||||
# Disable DNS rebinding protection for containerized deployments (k8s, Docker)
|
||||
# MCP 1.23+ auto-enables this for localhost, breaking k8s service DNS names
|
||||
transport_security=TransportSecuritySettings(
|
||||
enable_dns_rebinding_protection=False
|
||||
),
|
||||
)
|
||||
else:
|
||||
# ADR-016: Use Smithery lifespan for stateless mode, BasicAuth otherwise
|
||||
if deployment_mode == DeploymentMode.SMITHERY_STATELESS:
|
||||
logger.info("Configuring MCP server for Smithery stateless mode")
|
||||
# json_response=True returns plain JSON-RPC instead of SSE format,
|
||||
# required for Smithery scanner compatibility
|
||||
mcp = FastMCP(
|
||||
"Nextcloud MCP",
|
||||
lifespan=app_lifespan_smithery,
|
||||
json_response=True,
|
||||
# Disable DNS rebinding protection for containerized deployments (k8s, Docker)
|
||||
# MCP 1.23+ auto-enables this for localhost, breaking k8s service DNS names
|
||||
transport_security=TransportSecuritySettings(
|
||||
enable_dns_rebinding_protection=False
|
||||
),
|
||||
)
|
||||
else:
|
||||
logger.info("Configuring MCP server for BasicAuth mode")
|
||||
mcp = FastMCP(
|
||||
"Nextcloud MCP",
|
||||
lifespan=app_lifespan_basic,
|
||||
# Disable DNS rebinding protection for containerized deployments (k8s, Docker)
|
||||
# MCP 1.23+ auto-enables this for localhost, breaking k8s service DNS names
|
||||
transport_security=TransportSecuritySettings(
|
||||
enable_dns_rebinding_protection=False
|
||||
),
|
||||
)
|
||||
# BasicAuth modes (single-user or multi-user)
|
||||
logger.info(f"Configuring MCP server for {mode.value} mode")
|
||||
mcp = FastMCP(
|
||||
"Nextcloud MCP",
|
||||
lifespan=app_lifespan_basic,
|
||||
# Disable DNS rebinding protection for containerized deployments (k8s, Docker)
|
||||
# MCP 1.23+ auto-enables this for localhost, breaking k8s service DNS names
|
||||
transport_security=TransportSecuritySettings(
|
||||
enable_dns_rebinding_protection=False
|
||||
),
|
||||
)
|
||||
|
||||
@mcp.resource("nc://capabilities")
|
||||
async def nc_get_capabilities():
|
||||
@@ -1139,8 +1192,6 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
|
||||
# Register semantic search tools (cross-app feature)
|
||||
# ADR-016: Skip in Smithery stateless mode (no vector database)
|
||||
settings = get_settings()
|
||||
deployment_mode = get_deployment_mode()
|
||||
if deployment_mode == DeploymentMode.SMITHERY_STATELESS:
|
||||
logger.info("Skipping semantic search tools (Smithery stateless mode)")
|
||||
elif settings.vector_sync_enabled:
|
||||
@@ -1227,13 +1278,20 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
# Set OAuth context for OAuth login routes (ADR-004)
|
||||
if oauth_enabled:
|
||||
# Prepare OAuth config from setup_oauth_config closure variables
|
||||
# Get nextcloud_host from settings (it was validated as required)
|
||||
nextcloud_host_for_context = settings.nextcloud_host
|
||||
if not nextcloud_host_for_context:
|
||||
raise ValueError("NEXTCLOUD_HOST is required for OAuth mode")
|
||||
|
||||
mcp_server_url = os.getenv(
|
||||
"NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000"
|
||||
)
|
||||
nextcloud_resource_uri = os.getenv("NEXTCLOUD_RESOURCE_URI", nextcloud_host)
|
||||
nextcloud_resource_uri = os.getenv(
|
||||
"NEXTCLOUD_RESOURCE_URI", nextcloud_host_for_context
|
||||
)
|
||||
discovery_url = os.getenv(
|
||||
"OIDC_DISCOVERY_URL",
|
||||
f"{nextcloud_host}/.well-known/openid-configuration",
|
||||
f"{nextcloud_host_for_context}/.well-known/openid-configuration",
|
||||
)
|
||||
scopes = os.getenv("NEXTCLOUD_OIDC_SCOPES", "")
|
||||
|
||||
@@ -1247,7 +1305,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
"client_id": client_id, # From setup_oauth_config (DCR or static)
|
||||
"client_secret": client_secret, # From setup_oauth_config (DCR or static)
|
||||
"scopes": scopes,
|
||||
"nextcloud_host": nextcloud_host,
|
||||
"nextcloud_host": nextcloud_host_for_context,
|
||||
"nextcloud_resource_uri": nextcloud_resource_uri,
|
||||
"oauth_provider": oauth_provider,
|
||||
},
|
||||
@@ -1273,16 +1331,16 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
# BasicAuth mode - share storage with browser_app for webhook management
|
||||
from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
|
||||
|
||||
storage = RefreshTokenStorage.from_env()
|
||||
await storage.initialize()
|
||||
basic_auth_storage = RefreshTokenStorage.from_env()
|
||||
await basic_auth_storage.initialize()
|
||||
|
||||
app.state.storage = storage
|
||||
app.state.storage = basic_auth_storage
|
||||
|
||||
# Also share with browser_app for webhook routes
|
||||
for route in app.routes:
|
||||
if isinstance(route, Mount) and route.path == "/app":
|
||||
browser_app = cast(Starlette, route.app)
|
||||
browser_app.state.storage = storage
|
||||
browser_app.state.storage = basic_auth_storage
|
||||
logger.info(
|
||||
"Storage shared with browser_app for webhook management"
|
||||
)
|
||||
@@ -1292,7 +1350,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
# Scanner runs at server-level (once), not per-session
|
||||
import anyio as anyio_module
|
||||
|
||||
settings = get_settings()
|
||||
# Re-use settings from outer scope (already validated)
|
||||
|
||||
# Check if vector sync is enabled and determine the mode
|
||||
enable_offline_access_for_sync = os.getenv(
|
||||
@@ -1300,7 +1358,13 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
).lower() in ("true", "1", "yes")
|
||||
encryption_key = os.getenv("TOKEN_ENCRYPTION_KEY")
|
||||
|
||||
if settings.vector_sync_enabled and not oauth_enabled:
|
||||
# Multi-user BasicAuth uses OAuth-style background sync (with app passwords)
|
||||
# So skip single-user BasicAuth vector sync if in multi-user mode
|
||||
if (
|
||||
settings.vector_sync_enabled
|
||||
and not oauth_enabled
|
||||
and not settings.enable_multi_user_basic_auth
|
||||
):
|
||||
# BasicAuth mode - single user sync
|
||||
logger.info("Starting background vector sync tasks for BasicAuth mode")
|
||||
|
||||
@@ -1400,13 +1464,15 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
|
||||
elif (
|
||||
settings.vector_sync_enabled
|
||||
and oauth_enabled
|
||||
and (oauth_enabled or settings.enable_multi_user_basic_auth)
|
||||
and enable_offline_access_for_sync
|
||||
and refresh_token_storage
|
||||
and encryption_key
|
||||
):
|
||||
# OAuth mode with offline access - multi-user sync
|
||||
logger.info("Starting background vector sync tasks for OAuth mode")
|
||||
# Also used for multi-user BasicAuth mode (client auth is BasicAuth, background sync uses app passwords)
|
||||
mode_desc = "OAuth mode" if oauth_enabled else "Multi-user BasicAuth mode"
|
||||
logger.info(f"Starting background vector sync tasks for {mode_desc}")
|
||||
|
||||
from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
|
||||
from nextcloud_mcp_server.vector.oauth_sync import (
|
||||
@@ -1414,10 +1480,15 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
user_manager_task,
|
||||
)
|
||||
|
||||
# Get nextcloud_host (from settings - already validated)
|
||||
nextcloud_host_for_sync = settings.nextcloud_host
|
||||
if not nextcloud_host_for_sync:
|
||||
raise ValueError("NEXTCLOUD_HOST required for vector sync")
|
||||
|
||||
# Get OIDC discovery URL (same as used for OAuth setup)
|
||||
discovery_url = os.getenv(
|
||||
"OIDC_DISCOVERY_URL",
|
||||
f"{nextcloud_host}/.well-known/openid-configuration",
|
||||
f"{nextcloud_host_for_sync}/.well-known/openid-configuration",
|
||||
)
|
||||
|
||||
# Get client credentials from oauth_context (set by setup_oauth_config)
|
||||
@@ -1428,6 +1499,11 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
sync_client_id = oauth_config.get("client_id")
|
||||
sync_client_secret = oauth_config.get("client_secret")
|
||||
|
||||
# For multi-user BasicAuth mode, get OIDC credentials from environment
|
||||
if not sync_client_id or not sync_client_secret:
|
||||
sync_client_id = settings.oidc_client_id
|
||||
sync_client_secret = settings.oidc_client_secret
|
||||
|
||||
if not sync_client_id or not sync_client_secret:
|
||||
logger.error(
|
||||
"Cannot start OAuth vector sync: client credentials not found in oauth_context"
|
||||
@@ -2141,4 +2217,11 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
|
||||
app = SmitheryConfigMiddleware(app)
|
||||
logger.info("SmitheryConfigMiddleware enabled for query parameter config")
|
||||
|
||||
# Apply BasicAuthMiddleware for multi-user BasicAuth pass-through mode
|
||||
if settings.enable_multi_user_basic_auth:
|
||||
app = BasicAuthMiddleware(app)
|
||||
logger.info(
|
||||
"BasicAuthMiddleware enabled - multi-user BasicAuth pass-through mode active"
|
||||
)
|
||||
|
||||
return app
|
||||
|
||||
@@ -187,6 +187,11 @@ class Settings:
|
||||
enable_token_exchange: bool = False
|
||||
enable_offline_access: bool = False
|
||||
|
||||
# Multi-user BasicAuth pass-through mode (ADR-019 interim solution)
|
||||
# When enabled, MCP server extracts BasicAuth credentials from request headers
|
||||
# and passes them through to Nextcloud APIs (no storage, stateless)
|
||||
enable_multi_user_basic_auth: bool = False
|
||||
|
||||
# Token exchange cache settings
|
||||
token_exchange_cache_ttl: int = 300 # seconds (5 minutes default)
|
||||
|
||||
@@ -376,6 +381,10 @@ def get_settings() -> Settings:
|
||||
enable_offline_access=(
|
||||
os.getenv("ENABLE_OFFLINE_ACCESS", "false").lower() == "true"
|
||||
),
|
||||
# Multi-user BasicAuth pass-through mode
|
||||
enable_multi_user_basic_auth=(
|
||||
os.getenv("ENABLE_MULTI_USER_BASIC_AUTH", "false").lower() == "true"
|
||||
),
|
||||
# Token exchange cache settings
|
||||
token_exchange_cache_ttl=int(os.getenv("TOKEN_EXCHANGE_CACHE_TTL", "300")),
|
||||
# Token and webhook storage settings (encryption key optional for webhook-only usage)
|
||||
|
||||
@@ -0,0 +1,432 @@
|
||||
"""Configuration validation and mode detection for the MCP server.
|
||||
|
||||
This module provides:
|
||||
- Mode detection based on configuration
|
||||
- Configuration validation with clear error messages
|
||||
- Single source of truth for deployment mode requirements
|
||||
|
||||
See plan in /home/chris/.claude/plans/prancy-chasing-gizmo.md for architecture details.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
|
||||
from nextcloud_mcp_server.config import Settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AuthMode(Enum):
|
||||
"""Authentication mode for the MCP server.
|
||||
|
||||
Determines how users authenticate and how the server accesses Nextcloud.
|
||||
"""
|
||||
|
||||
SINGLE_USER_BASIC = "single_user_basic"
|
||||
MULTI_USER_BASIC = "multi_user_basic"
|
||||
OAUTH_SINGLE_AUDIENCE = "oauth_single"
|
||||
OAUTH_TOKEN_EXCHANGE = "oauth_exchange"
|
||||
SMITHERY_STATELESS = "smithery"
|
||||
|
||||
|
||||
@dataclass
|
||||
class ModeRequirements:
|
||||
"""Requirements for a deployment mode.
|
||||
|
||||
Attributes:
|
||||
required: Configuration variables that must be set
|
||||
optional: Configuration variables that may be set
|
||||
forbidden: Configuration variables that should not be set
|
||||
conditional: Additional requirements based on feature flags
|
||||
Format: {feature_flag: [required_vars]}
|
||||
description: Human-readable description of the mode
|
||||
"""
|
||||
|
||||
required: list[str]
|
||||
optional: list[str]
|
||||
forbidden: list[str]
|
||||
conditional: dict[str, list[str]]
|
||||
description: str
|
||||
|
||||
|
||||
# Mode requirements definition
|
||||
MODE_REQUIREMENTS: dict[AuthMode, ModeRequirements] = {
|
||||
AuthMode.SINGLE_USER_BASIC: ModeRequirements(
|
||||
required=["nextcloud_host", "nextcloud_username", "nextcloud_password"],
|
||||
optional=[
|
||||
"vector_sync_enabled",
|
||||
"qdrant_url",
|
||||
"qdrant_location",
|
||||
"ollama_base_url",
|
||||
"ollama_embedding_model",
|
||||
"openai_api_key",
|
||||
"openai_embedding_model",
|
||||
"document_chunk_size",
|
||||
"document_chunk_overlap",
|
||||
],
|
||||
forbidden=[
|
||||
"enable_multi_user_basic_auth",
|
||||
"enable_token_exchange",
|
||||
"oidc_client_id",
|
||||
"oidc_client_secret",
|
||||
],
|
||||
conditional={
|
||||
"vector_sync_enabled": [
|
||||
# Either qdrant_url OR qdrant_location (checked in Settings.__post_init__)
|
||||
# At least one embedding provider (ollama_base_url OR openai_api_key)
|
||||
],
|
||||
},
|
||||
description="Single-user deployment with BasicAuth credentials. "
|
||||
"Suitable for personal Nextcloud instances and local development.",
|
||||
),
|
||||
AuthMode.MULTI_USER_BASIC: ModeRequirements(
|
||||
required=["nextcloud_host", "enable_multi_user_basic_auth"],
|
||||
optional=[
|
||||
# Background sync with app passwords (via Astrolabe)
|
||||
"enable_offline_access",
|
||||
"token_encryption_key",
|
||||
"token_storage_db",
|
||||
"oidc_client_id",
|
||||
"oidc_client_secret",
|
||||
# Vector sync
|
||||
"vector_sync_enabled",
|
||||
"qdrant_url",
|
||||
"qdrant_location",
|
||||
"ollama_base_url",
|
||||
"ollama_embedding_model",
|
||||
"openai_api_key",
|
||||
"openai_embedding_model",
|
||||
],
|
||||
forbidden=[
|
||||
"nextcloud_username",
|
||||
"nextcloud_password",
|
||||
"enable_token_exchange",
|
||||
],
|
||||
conditional={
|
||||
"enable_offline_access": [
|
||||
"oidc_client_id",
|
||||
"oidc_client_secret",
|
||||
"token_encryption_key",
|
||||
"token_storage_db",
|
||||
],
|
||||
"vector_sync_enabled": [
|
||||
# Requires offline access for background sync
|
||||
"enable_offline_access",
|
||||
],
|
||||
},
|
||||
description="Multi-user deployment with BasicAuth pass-through. "
|
||||
"Users provide credentials in request headers. "
|
||||
"Optional background sync using app passwords stored via Astrolabe.",
|
||||
),
|
||||
AuthMode.OAUTH_SINGLE_AUDIENCE: ModeRequirements(
|
||||
required=["nextcloud_host"],
|
||||
optional=[
|
||||
# OAuth credentials (uses DCR if not provided)
|
||||
"oidc_client_id",
|
||||
"oidc_client_secret",
|
||||
"oidc_discovery_url",
|
||||
# Offline access
|
||||
"enable_offline_access",
|
||||
"token_encryption_key",
|
||||
"token_storage_db",
|
||||
# Vector sync
|
||||
"vector_sync_enabled",
|
||||
"qdrant_url",
|
||||
"qdrant_location",
|
||||
"ollama_base_url",
|
||||
"ollama_embedding_model",
|
||||
"openai_api_key",
|
||||
"openai_embedding_model",
|
||||
# Scopes
|
||||
"nextcloud_oidc_scopes",
|
||||
],
|
||||
forbidden=[
|
||||
"nextcloud_username",
|
||||
"nextcloud_password",
|
||||
"enable_token_exchange",
|
||||
"enable_multi_user_basic_auth",
|
||||
],
|
||||
conditional={
|
||||
"enable_offline_access": [
|
||||
"token_encryption_key",
|
||||
"token_storage_db",
|
||||
],
|
||||
"vector_sync_enabled": [
|
||||
"enable_offline_access", # Background sync requires refresh tokens
|
||||
],
|
||||
},
|
||||
description="OAuth multi-user deployment with single-audience tokens. "
|
||||
"Tokens work for both MCP server and Nextcloud APIs (pass-through). "
|
||||
"Uses Dynamic Client Registration if credentials not provided.",
|
||||
),
|
||||
AuthMode.OAUTH_TOKEN_EXCHANGE: ModeRequirements(
|
||||
required=["nextcloud_host", "enable_token_exchange"],
|
||||
optional=[
|
||||
# OAuth credentials
|
||||
"oidc_client_id",
|
||||
"oidc_client_secret",
|
||||
"oidc_discovery_url",
|
||||
# Token exchange settings
|
||||
"token_exchange_cache_ttl",
|
||||
# Offline access
|
||||
"enable_offline_access",
|
||||
"token_encryption_key",
|
||||
"token_storage_db",
|
||||
# Vector sync
|
||||
"vector_sync_enabled",
|
||||
"qdrant_url",
|
||||
"qdrant_location",
|
||||
"ollama_base_url",
|
||||
"ollama_embedding_model",
|
||||
"openai_api_key",
|
||||
"openai_embedding_model",
|
||||
],
|
||||
forbidden=[
|
||||
"nextcloud_username",
|
||||
"nextcloud_password",
|
||||
"enable_multi_user_basic_auth",
|
||||
],
|
||||
conditional={
|
||||
"enable_offline_access": [
|
||||
"token_encryption_key",
|
||||
"token_storage_db",
|
||||
],
|
||||
"vector_sync_enabled": [
|
||||
"enable_offline_access",
|
||||
],
|
||||
},
|
||||
description="OAuth multi-user deployment with token exchange (RFC 8693). "
|
||||
"MCP tokens are separate from Nextcloud tokens. "
|
||||
"Server exchanges MCP token for Nextcloud token on each request.",
|
||||
),
|
||||
AuthMode.SMITHERY_STATELESS: ModeRequirements(
|
||||
required=[], # All config from session URL params
|
||||
optional=[],
|
||||
forbidden=[
|
||||
"nextcloud_host",
|
||||
"nextcloud_username",
|
||||
"nextcloud_password",
|
||||
"enable_multi_user_basic_auth",
|
||||
"enable_token_exchange",
|
||||
"enable_offline_access",
|
||||
"vector_sync_enabled",
|
||||
"oidc_client_id",
|
||||
"oidc_client_secret",
|
||||
],
|
||||
conditional={},
|
||||
description="Stateless multi-tenant deployment for Smithery platform. "
|
||||
"Configuration comes from session URL parameters. "
|
||||
"No persistent storage, no OAuth, no vector sync.",
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def detect_auth_mode(settings: Settings) -> AuthMode:
|
||||
"""Detect authentication mode from configuration.
|
||||
|
||||
Mode detection priority (most specific to most general):
|
||||
1. Smithery (explicit flag)
|
||||
2. Token exchange (most specific OAuth mode)
|
||||
3. Multi-user BasicAuth
|
||||
4. Single-user BasicAuth
|
||||
5. OAuth single-audience (default OAuth mode)
|
||||
|
||||
Args:
|
||||
settings: Application settings
|
||||
|
||||
Returns:
|
||||
Detected AuthMode
|
||||
"""
|
||||
# Check for Smithery mode (explicit environment variable)
|
||||
# Note: This checks the environment directly, not settings
|
||||
# because Smithery mode has no settings-based config
|
||||
import os
|
||||
|
||||
if os.getenv("SMITHERY_DEPLOYMENT", "false").lower() == "true":
|
||||
return AuthMode.SMITHERY_STATELESS
|
||||
|
||||
# Check for token exchange (most specific OAuth mode)
|
||||
if settings.enable_token_exchange:
|
||||
return AuthMode.OAUTH_TOKEN_EXCHANGE
|
||||
|
||||
# Check for multi-user BasicAuth
|
||||
if settings.enable_multi_user_basic_auth:
|
||||
return AuthMode.MULTI_USER_BASIC
|
||||
|
||||
# Check for single-user BasicAuth (explicit credentials)
|
||||
if settings.nextcloud_username and settings.nextcloud_password:
|
||||
return AuthMode.SINGLE_USER_BASIC
|
||||
|
||||
# Default: OAuth single-audience mode
|
||||
# This is the safest multi-user mode (no credential storage)
|
||||
return AuthMode.OAUTH_SINGLE_AUDIENCE
|
||||
|
||||
|
||||
def validate_configuration(settings: Settings) -> tuple[AuthMode, list[str]]:
|
||||
"""Validate configuration for detected mode.
|
||||
|
||||
Args:
|
||||
settings: Application settings
|
||||
|
||||
Returns:
|
||||
Tuple of (detected_mode, list_of_errors)
|
||||
Empty list means valid configuration.
|
||||
"""
|
||||
mode = detect_auth_mode(settings)
|
||||
requirements = MODE_REQUIREMENTS[mode]
|
||||
errors: list[str] = []
|
||||
|
||||
logger.debug(f"Validating configuration for mode: {mode.value}")
|
||||
|
||||
# Check required variables
|
||||
for var in requirements.required:
|
||||
value = getattr(settings, var, None)
|
||||
if value is None or (isinstance(value, str) and not value.strip()):
|
||||
errors.append(
|
||||
f"[{mode.value}] Missing required configuration: {var.upper()}"
|
||||
)
|
||||
|
||||
# Check forbidden variables
|
||||
for var in requirements.forbidden:
|
||||
value = getattr(settings, var, None)
|
||||
# For bools, check if True (forbidden means must be False/unset)
|
||||
# For strings, check if non-empty
|
||||
is_set = False
|
||||
if isinstance(value, bool):
|
||||
is_set = value is True
|
||||
elif isinstance(value, str):
|
||||
is_set = bool(value.strip())
|
||||
elif value is not None:
|
||||
is_set = True
|
||||
|
||||
if is_set:
|
||||
errors.append(
|
||||
f"[{mode.value}] Forbidden configuration: {var.upper()} "
|
||||
f"should not be set in this mode"
|
||||
)
|
||||
|
||||
# Check conditional requirements
|
||||
for condition, required_vars in requirements.conditional.items():
|
||||
# Check if the condition is enabled
|
||||
condition_value = getattr(settings, condition, None)
|
||||
is_enabled = False
|
||||
|
||||
if isinstance(condition_value, bool):
|
||||
is_enabled = condition_value is True
|
||||
elif isinstance(condition_value, str):
|
||||
is_enabled = bool(condition_value.strip())
|
||||
elif condition_value is not None:
|
||||
is_enabled = True
|
||||
|
||||
if is_enabled:
|
||||
# Check that all required vars for this condition are set
|
||||
for var in required_vars:
|
||||
value = getattr(settings, var, None)
|
||||
|
||||
# For boolean requirements, check that they are True (not just set)
|
||||
if hasattr(Settings, var):
|
||||
field_type = type(getattr(Settings(), var, None))
|
||||
if field_type is bool:
|
||||
if value is not True:
|
||||
errors.append(
|
||||
f"[{mode.value}] {var.upper()} must be enabled when "
|
||||
f"{condition.upper()} is enabled"
|
||||
)
|
||||
continue
|
||||
|
||||
# For non-boolean requirements, check that they are set
|
||||
if value is None or (isinstance(value, str) and not value.strip()):
|
||||
errors.append(
|
||||
f"[{mode.value}] {var.upper()} is required when "
|
||||
f"{condition.upper()} is enabled"
|
||||
)
|
||||
|
||||
# Special validations for specific modes
|
||||
if mode == AuthMode.SINGLE_USER_BASIC:
|
||||
# Validate that NEXTCLOUD_HOST doesn't have trailing slash
|
||||
if settings.nextcloud_host and settings.nextcloud_host.endswith("/"):
|
||||
errors.append(
|
||||
f"[{mode.value}] NEXTCLOUD_HOST should not have trailing slash: "
|
||||
f"{settings.nextcloud_host}"
|
||||
)
|
||||
|
||||
if mode in [
|
||||
AuthMode.OAUTH_SINGLE_AUDIENCE,
|
||||
AuthMode.OAUTH_TOKEN_EXCHANGE,
|
||||
]:
|
||||
# If OAuth credentials not provided, DCR must be available
|
||||
# (This is a runtime check, not a config check, so we just warn)
|
||||
if not settings.oidc_client_id or not settings.oidc_client_secret:
|
||||
logger.info(
|
||||
f"[{mode.value}] OAuth credentials not configured. "
|
||||
"Will attempt Dynamic Client Registration (DCR) at startup."
|
||||
)
|
||||
|
||||
if mode == AuthMode.MULTI_USER_BASIC:
|
||||
# Validate that if offline access enabled, we have OAuth credentials
|
||||
if settings.enable_offline_access:
|
||||
if not settings.oidc_client_id or not settings.oidc_client_secret:
|
||||
errors.append(
|
||||
f"[{mode.value}] NEXTCLOUD_OIDC_CLIENT_ID and "
|
||||
"NEXTCLOUD_OIDC_CLIENT_SECRET are required when "
|
||||
"ENABLE_OFFLINE_ACCESS is enabled (for app password retrieval)"
|
||||
)
|
||||
|
||||
# Validate vector sync requirements
|
||||
if settings.vector_sync_enabled and not settings.enable_offline_access:
|
||||
errors.append(
|
||||
f"[{mode.value}] ENABLE_OFFLINE_ACCESS must be enabled when "
|
||||
"VECTOR_SYNC_ENABLED is true (background sync requires "
|
||||
"app passwords or refresh tokens)"
|
||||
)
|
||||
|
||||
# Note: Embedding provider validation removed - Simple provider is always
|
||||
# available as fallback (ADR-015). Users can optionally configure Ollama or OpenAI
|
||||
# for better quality embeddings.
|
||||
|
||||
return mode, errors
|
||||
|
||||
|
||||
def get_mode_summary(mode: AuthMode) -> str:
|
||||
"""Get human-readable summary of a deployment mode.
|
||||
|
||||
Args:
|
||||
mode: Deployment mode
|
||||
|
||||
Returns:
|
||||
Multi-line string describing the mode
|
||||
"""
|
||||
requirements = MODE_REQUIREMENTS[mode]
|
||||
|
||||
summary_lines = [
|
||||
f"Mode: {mode.value}",
|
||||
f"Description: {requirements.description}",
|
||||
"",
|
||||
"Required configuration:",
|
||||
]
|
||||
|
||||
if requirements.required:
|
||||
for var in requirements.required:
|
||||
summary_lines.append(f" - {var.upper()}")
|
||||
else:
|
||||
summary_lines.append(" (none - configured via session)")
|
||||
|
||||
summary_lines.append("")
|
||||
summary_lines.append("Optional configuration:")
|
||||
|
||||
if requirements.optional:
|
||||
for var in requirements.optional:
|
||||
summary_lines.append(f" - {var.upper()}")
|
||||
else:
|
||||
summary_lines.append(" (none)")
|
||||
|
||||
if requirements.conditional:
|
||||
summary_lines.append("")
|
||||
summary_lines.append("Conditional requirements:")
|
||||
for condition, vars in requirements.conditional.items():
|
||||
summary_lines.append(f" When {condition.upper()} is enabled:")
|
||||
for var in vars:
|
||||
summary_lines.append(f" - {var.upper()}")
|
||||
|
||||
return "\n".join(summary_lines)
|
||||
@@ -0,0 +1,578 @@
|
||||
"""Unit tests for configuration validation and mode detection.
|
||||
|
||||
Tests cover:
|
||||
- Mode detection logic
|
||||
- Configuration validation for each mode
|
||||
- Error message generation
|
||||
- Edge cases and boundary conditions
|
||||
"""
|
||||
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
|
||||
from nextcloud_mcp_server.config import Settings
|
||||
from nextcloud_mcp_server.config_validators import (
|
||||
AuthMode,
|
||||
detect_auth_mode,
|
||||
get_mode_summary,
|
||||
validate_configuration,
|
||||
)
|
||||
|
||||
|
||||
class TestModeDetection:
|
||||
"""Test auth mode detection from configuration."""
|
||||
|
||||
def test_smithery_mode_detection(self):
|
||||
"""Test Smithery mode is detected from environment variable."""
|
||||
settings = Settings()
|
||||
|
||||
with patch.dict(os.environ, {"SMITHERY_DEPLOYMENT": "true"}):
|
||||
mode = detect_auth_mode(settings)
|
||||
assert mode == AuthMode.SMITHERY_STATELESS
|
||||
|
||||
def test_token_exchange_mode_detection(self):
|
||||
"""Test token exchange mode is detected."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
enable_token_exchange=True,
|
||||
)
|
||||
|
||||
mode = detect_auth_mode(settings)
|
||||
assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE
|
||||
|
||||
def test_multi_user_basic_mode_detection(self):
|
||||
"""Test multi-user BasicAuth mode is detected."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
enable_multi_user_basic_auth=True,
|
||||
)
|
||||
|
||||
mode = detect_auth_mode(settings)
|
||||
assert mode == AuthMode.MULTI_USER_BASIC
|
||||
|
||||
def test_single_user_basic_mode_detection(self):
|
||||
"""Test single-user BasicAuth mode is detected."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
)
|
||||
|
||||
mode = detect_auth_mode(settings)
|
||||
assert mode == AuthMode.SINGLE_USER_BASIC
|
||||
|
||||
def test_oauth_single_audience_default(self):
|
||||
"""Test OAuth single-audience is default mode."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
)
|
||||
|
||||
mode = detect_auth_mode(settings)
|
||||
assert mode == AuthMode.OAUTH_SINGLE_AUDIENCE
|
||||
|
||||
def test_mode_priority_smithery_over_all(self):
|
||||
"""Test Smithery mode has highest priority."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
enable_token_exchange=True,
|
||||
enable_multi_user_basic_auth=True,
|
||||
)
|
||||
|
||||
with patch.dict(os.environ, {"SMITHERY_DEPLOYMENT": "true"}):
|
||||
mode = detect_auth_mode(settings)
|
||||
assert mode == AuthMode.SMITHERY_STATELESS
|
||||
|
||||
def test_mode_priority_token_exchange_over_basic(self):
|
||||
"""Test token exchange has priority over BasicAuth."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
enable_token_exchange=True,
|
||||
)
|
||||
|
||||
mode = detect_auth_mode(settings)
|
||||
assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE
|
||||
|
||||
|
||||
class TestSingleUserBasicValidation:
|
||||
"""Test validation for single-user BasicAuth mode."""
|
||||
|
||||
def test_valid_minimal_config(self):
|
||||
"""Test valid minimal single-user BasicAuth config."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.SINGLE_USER_BASIC
|
||||
assert len(errors) == 0
|
||||
|
||||
def test_valid_with_vector_sync(self):
|
||||
"""Test valid config with vector sync enabled."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
vector_sync_enabled=True,
|
||||
qdrant_location=":memory:",
|
||||
ollama_base_url="http://ollama:11434",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.SINGLE_USER_BASIC
|
||||
assert len(errors) == 0
|
||||
|
||||
def test_missing_required_host(self):
|
||||
"""Test error when NEXTCLOUD_HOST is missing."""
|
||||
settings = Settings(
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.SINGLE_USER_BASIC
|
||||
assert any("nextcloud_host" in err.lower() for err in errors)
|
||||
|
||||
def test_missing_required_username(self):
|
||||
"""Test that partial credentials fall back to OAuth mode."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
nextcloud_password="password", # Password without username
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
# Mode detection requires BOTH username AND password for single-user BasicAuth
|
||||
# If only one is present, it defaults to OAuth single-audience
|
||||
assert mode == AuthMode.OAUTH_SINGLE_AUDIENCE
|
||||
# In OAuth mode, having a password set is forbidden
|
||||
assert any("nextcloud_password" in err.lower() for err in errors)
|
||||
|
||||
def test_missing_required_password(self):
|
||||
"""Test that partial credentials fall back to OAuth mode."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
nextcloud_username="admin", # Username without password
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
# Mode detection requires BOTH username AND password for single-user BasicAuth
|
||||
# If only one is present, it defaults to OAuth single-audience
|
||||
assert mode == AuthMode.OAUTH_SINGLE_AUDIENCE
|
||||
# In OAuth mode, having a username set is forbidden
|
||||
assert any("nextcloud_username" in err.lower() for err in errors)
|
||||
|
||||
def test_forbidden_multi_user_basic_auth(self):
|
||||
"""Test error when ENABLE_MULTI_USER_BASIC_AUTH is set."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
enable_multi_user_basic_auth=True,
|
||||
)
|
||||
|
||||
# Note: This will detect as MULTI_USER_BASIC due to priority
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.MULTI_USER_BASIC
|
||||
# It will fail multi-user validation because username/password are forbidden
|
||||
assert len(errors) > 0
|
||||
|
||||
def test_forbidden_token_exchange(self):
|
||||
"""Test error when ENABLE_TOKEN_EXCHANGE is set."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
enable_token_exchange=True,
|
||||
)
|
||||
|
||||
# Note: This will detect as OAUTH_TOKEN_EXCHANGE due to priority
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE
|
||||
# It will fail OAuth validation
|
||||
|
||||
def test_vector_sync_without_embedding_provider_uses_fallback(self):
|
||||
"""Test that vector sync works with Simple provider fallback (no config needed)."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
vector_sync_enabled=True,
|
||||
qdrant_location=":memory:",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.SINGLE_USER_BASIC
|
||||
# Should pass - Simple provider is always available as fallback
|
||||
assert len(errors) == 0
|
||||
|
||||
|
||||
class TestMultiUserBasicValidation:
|
||||
"""Test validation for multi-user BasicAuth mode."""
|
||||
|
||||
def test_valid_minimal_config(self):
|
||||
"""Test valid minimal multi-user BasicAuth config."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
enable_multi_user_basic_auth=True,
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.MULTI_USER_BASIC
|
||||
assert len(errors) == 0
|
||||
|
||||
def test_valid_with_offline_access(self):
|
||||
"""Test valid config with offline access enabled."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
enable_multi_user_basic_auth=True,
|
||||
enable_offline_access=True,
|
||||
oidc_client_id="test-client",
|
||||
oidc_client_secret="test-secret",
|
||||
token_encryption_key="test-key-" + "a" * 32,
|
||||
token_storage_db="/tmp/tokens.db",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.MULTI_USER_BASIC
|
||||
assert len(errors) == 0
|
||||
|
||||
def test_missing_required_host(self):
|
||||
"""Test error when NEXTCLOUD_HOST is missing."""
|
||||
settings = Settings(
|
||||
enable_multi_user_basic_auth=True,
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.MULTI_USER_BASIC
|
||||
assert any("nextcloud_host" in err.lower() for err in errors)
|
||||
|
||||
def test_forbidden_username_password(self):
|
||||
"""Test error when NEXTCLOUD_USERNAME/PASSWORD are set."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
enable_multi_user_basic_auth=True,
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
# Multi-user BasicAuth has higher priority than single-user in detection
|
||||
# (explicit flags come before credentials)
|
||||
assert mode == AuthMode.MULTI_USER_BASIC
|
||||
# Should report errors for forbidden username/password
|
||||
assert any("nextcloud_username" in err.lower() for err in errors)
|
||||
assert any("nextcloud_password" in err.lower() for err in errors)
|
||||
|
||||
def test_offline_access_missing_oauth_credentials(self):
|
||||
"""Test error when offline access enabled but OAuth credentials missing."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
enable_multi_user_basic_auth=True,
|
||||
enable_offline_access=True,
|
||||
token_encryption_key="test-key-" + "a" * 32,
|
||||
token_storage_db="/tmp/tokens.db",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.MULTI_USER_BASIC
|
||||
assert any("oidc_client_id" in err.lower() for err in errors)
|
||||
|
||||
def test_offline_access_missing_encryption_key(self):
|
||||
"""Test error when offline access enabled but encryption key missing."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
enable_multi_user_basic_auth=True,
|
||||
enable_offline_access=True,
|
||||
oidc_client_id="test-client",
|
||||
oidc_client_secret="test-secret",
|
||||
token_storage_db="/tmp/tokens.db",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.MULTI_USER_BASIC
|
||||
assert any("token_encryption_key" in err.lower() for err in errors)
|
||||
|
||||
def test_vector_sync_requires_offline_access(self):
|
||||
"""Test error when vector sync enabled but offline access disabled."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
enable_multi_user_basic_auth=True,
|
||||
vector_sync_enabled=True,
|
||||
qdrant_location=":memory:",
|
||||
ollama_base_url="http://ollama:11434",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.MULTI_USER_BASIC
|
||||
assert any("enable_offline_access" in err.lower() for err in errors)
|
||||
|
||||
|
||||
class TestOAuthSingleAudienceValidation:
|
||||
"""Test validation for OAuth single-audience mode."""
|
||||
|
||||
def test_valid_minimal_config(self):
|
||||
"""Test valid minimal OAuth single-audience config."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.OAUTH_SINGLE_AUDIENCE
|
||||
assert len(errors) == 0
|
||||
|
||||
def test_valid_with_static_credentials(self):
|
||||
"""Test valid config with static OAuth credentials."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
oidc_client_id="test-client",
|
||||
oidc_client_secret="test-secret",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.OAUTH_SINGLE_AUDIENCE
|
||||
assert len(errors) == 0
|
||||
|
||||
def test_valid_with_offline_access(self):
|
||||
"""Test valid config with offline access."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
oidc_client_id="test-client",
|
||||
oidc_client_secret="test-secret",
|
||||
enable_offline_access=True,
|
||||
token_encryption_key="test-key-" + "a" * 32,
|
||||
token_storage_db="/tmp/tokens.db",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.OAUTH_SINGLE_AUDIENCE
|
||||
assert len(errors) == 0
|
||||
|
||||
def test_forbidden_username_password(self):
|
||||
"""Test that username/password trigger single-user mode instead."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
# This should detect as SINGLE_USER_BASIC
|
||||
assert mode == AuthMode.SINGLE_USER_BASIC
|
||||
|
||||
def test_offline_access_missing_encryption_key(self):
|
||||
"""Test error when offline access enabled but encryption key missing."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
enable_offline_access=True,
|
||||
token_storage_db="/tmp/tokens.db",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.OAUTH_SINGLE_AUDIENCE
|
||||
assert any("token_encryption_key" in err.lower() for err in errors)
|
||||
|
||||
def test_vector_sync_requires_offline_access(self):
|
||||
"""Test error when vector sync enabled but offline access disabled."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
vector_sync_enabled=True,
|
||||
qdrant_location=":memory:",
|
||||
ollama_base_url="http://ollama:11434",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.OAUTH_SINGLE_AUDIENCE
|
||||
assert any("enable_offline_access" in err.lower() for err in errors)
|
||||
|
||||
|
||||
class TestOAuthTokenExchangeValidation:
|
||||
"""Test validation for OAuth token exchange mode."""
|
||||
|
||||
def test_valid_minimal_config(self):
|
||||
"""Test valid minimal OAuth token exchange config."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
enable_token_exchange=True,
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE
|
||||
assert len(errors) == 0
|
||||
|
||||
def test_valid_with_credentials(self):
|
||||
"""Test valid config with OAuth credentials."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
enable_token_exchange=True,
|
||||
oidc_client_id="test-client",
|
||||
oidc_client_secret="test-secret",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE
|
||||
assert len(errors) == 0
|
||||
|
||||
def test_forbidden_username_password(self):
|
||||
"""Test error when username/password are set."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
enable_token_exchange=True,
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE
|
||||
assert any("nextcloud_username" in err.lower() for err in errors)
|
||||
assert any("nextcloud_password" in err.lower() for err in errors)
|
||||
|
||||
|
||||
class TestSmitheryValidation:
|
||||
"""Test validation for Smithery stateless mode."""
|
||||
|
||||
def test_valid_empty_config(self):
|
||||
"""Test valid empty config for Smithery mode."""
|
||||
settings = Settings()
|
||||
|
||||
with patch.dict(os.environ, {"SMITHERY_DEPLOYMENT": "true"}):
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.SMITHERY_STATELESS
|
||||
assert len(errors) == 0
|
||||
|
||||
def test_forbidden_nextcloud_host(self):
|
||||
"""Test error when NEXTCLOUD_HOST is set."""
|
||||
settings = Settings(
|
||||
nextcloud_host="http://localhost",
|
||||
)
|
||||
|
||||
with patch.dict(os.environ, {"SMITHERY_DEPLOYMENT": "true"}):
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.SMITHERY_STATELESS
|
||||
assert any("nextcloud_host" in err.lower() for err in errors)
|
||||
|
||||
def test_forbidden_credentials(self):
|
||||
"""Test error when credentials are set."""
|
||||
settings = Settings(
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
)
|
||||
|
||||
with patch.dict(os.environ, {"SMITHERY_DEPLOYMENT": "true"}):
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.SMITHERY_STATELESS
|
||||
assert any("nextcloud_username" in err.lower() for err in errors)
|
||||
|
||||
def test_forbidden_vector_sync(self):
|
||||
"""Test error when vector sync is enabled."""
|
||||
settings = Settings(
|
||||
vector_sync_enabled=True,
|
||||
)
|
||||
|
||||
with patch.dict(os.environ, {"SMITHERY_DEPLOYMENT": "true"}):
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
assert mode == AuthMode.SMITHERY_STATELESS
|
||||
assert any("vector_sync_enabled" in err.lower() for err in errors)
|
||||
|
||||
|
||||
class TestModeSummary:
|
||||
"""Test mode summary generation."""
|
||||
|
||||
def test_single_user_basic_summary(self):
|
||||
"""Test summary for single-user BasicAuth mode."""
|
||||
summary = get_mode_summary(AuthMode.SINGLE_USER_BASIC)
|
||||
|
||||
assert "single_user_basic" in summary
|
||||
assert "NEXTCLOUD_HOST" in summary
|
||||
assert "NEXTCLOUD_USERNAME" in summary
|
||||
assert "NEXTCLOUD_PASSWORD" in summary
|
||||
assert "VECTOR_SYNC_ENABLED" in summary
|
||||
|
||||
def test_smithery_summary(self):
|
||||
"""Test summary for Smithery mode."""
|
||||
summary = get_mode_summary(AuthMode.SMITHERY_STATELESS)
|
||||
|
||||
assert "smithery" in summary
|
||||
assert "session" in summary.lower()
|
||||
assert "(none" in summary # No required config
|
||||
|
||||
def test_oauth_token_exchange_summary(self):
|
||||
"""Test summary for OAuth token exchange mode."""
|
||||
summary = get_mode_summary(AuthMode.OAUTH_TOKEN_EXCHANGE)
|
||||
|
||||
assert "oauth_exchange" in summary
|
||||
assert "ENABLE_TOKEN_EXCHANGE" in summary
|
||||
assert "RFC 8693" in summary
|
||||
|
||||
|
||||
class TestEdgeCases:
|
||||
"""Test edge cases and boundary conditions."""
|
||||
|
||||
def test_empty_string_treated_as_missing(self):
|
||||
"""Test that empty strings are treated as missing values."""
|
||||
settings = Settings(
|
||||
nextcloud_host="", # Empty string
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
# Should fail because nextcloud_host is effectively missing
|
||||
assert any("nextcloud_host" in err.lower() for err in errors)
|
||||
|
||||
def test_whitespace_treated_as_missing(self):
|
||||
"""Test that whitespace-only strings are treated as missing."""
|
||||
settings = Settings(
|
||||
nextcloud_host=" ", # Whitespace only
|
||||
nextcloud_username="admin",
|
||||
nextcloud_password="password",
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
# Should fail because nextcloud_host is effectively missing
|
||||
assert any("nextcloud_host" in err.lower() for err in errors)
|
||||
|
||||
def test_multiple_errors_reported(self):
|
||||
"""Test that multiple errors are all reported."""
|
||||
settings = Settings(
|
||||
# Missing all required fields for single-user BasicAuth
|
||||
)
|
||||
|
||||
mode, errors = validate_configuration(settings)
|
||||
|
||||
# Should have errors for missing host (OAuth mode is default)
|
||||
assert len(errors) > 0
|
||||
Reference in New Issue
Block a user