Compare commits

..

3 Commits

Author SHA1 Message Date
github-actions[bot] 91d06acfb4 bump: version 0.57.49 → 0.57.50 2026-02-16 11:38:36 +00:00
Chris Coutinho 90874ca7cd Merge pull request #563 from cbcoutinho/renovate/ollama-1.x
chore(deps): update helm release ollama to v1.43.0
2026-02-16 12:37:47 +01:00
renovate-bot-cbcoutinho[bot] da8fed3382 chore(deps): update helm release ollama to v1.43.0 2026-02-16 11:16:48 +00:00
26 changed files with 53 additions and 437 deletions
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.commitizen]
name = "cz_conventional_commits"
version = "0.57.49"
version = "0.57.50"
tag_format = "nextcloud-mcp-server-$version"
version_scheme = "semver"
update_changelog_on_bump = true
+2
View File
@@ -14,6 +14,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Configurable resource limits
- Grafana dashboard annotations
## nextcloud-mcp-server-0.57.50 (2026-02-16)
## nextcloud-mcp-server-0.57.49 (2026-02-16)
### Refactor
+3 -3
View File
@@ -4,6 +4,6 @@ dependencies:
version: 1.16.3
- name: ollama
repository: https://otwld.github.io/ollama-helm
version: 1.42.0
digest: sha256:a9aef6e290f23b1ed961450e0635eb0bce42f8c52805276901a80df7c27473f6
generated: "2026-02-10T11:10:44.457881902Z"
version: 1.43.0
digest: sha256:533eda3fdb4bd92cdafee49bd7b0428fc87d21b509032442c04ed645900e464a
generated: "2026-02-16T11:16:41.257136832Z"
+2 -2
View File
@@ -2,7 +2,7 @@ apiVersion: v2
name: nextcloud-mcp-server
description: A Helm chart for Nextcloud MCP Server - enables AI assistants to interact with Nextcloud
type: application
version: 0.57.49
version: 0.57.50
appVersion: "0.63.5"
keywords:
- nextcloud
@@ -31,6 +31,6 @@ dependencies:
repository: https://qdrant.github.io/qdrant-helm
condition: qdrant.networkMode.deploySubchart
- name: ollama
version: "1.42.0"
version: "1.43.0"
repository: https://otwld.github.io/ollama-helm
condition: ollama.enabled
-52
View File
@@ -171,58 +171,6 @@ NEXTCLOUD_PASSWORD=your_app_password_or_password
---
## SSL/TLS Configuration (Optional)
If your Nextcloud instance uses a self-signed certificate or a private CA (common with reverse proxies like Traefik or Caddy), the MCP server will reject the connection by default. Use these settings to configure certificate verification.
### Custom CA Bundle (Recommended)
Point the server at your CA certificate file:
```dotenv
NEXTCLOUD_CA_BUNDLE=/etc/ssl/certs/my-ca.pem
```
With Docker, mount the certificate as a read-only volume:
```bash
docker run \
-v /path/to/my-ca.pem:/etc/ssl/certs/my-ca.pem:ro \
-e NEXTCLOUD_CA_BUNDLE=/etc/ssl/certs/my-ca.pem \
-e NEXTCLOUD_HOST=https://nextcloud.local \
--env-file .env \
ghcr.io/cbcoutinho/nextcloud-mcp-server:latest
```
### Disable Verification (Development Only)
> [!WARNING]
> Disabling TLS verification is insecure. Only use this for local development or testing.
```dotenv
NEXTCLOUD_VERIFY_SSL=false
```
### Environment Variables Reference
| Variable | Required | Default | Description |
|----------|----------|---------|-------------|
| `NEXTCLOUD_VERIFY_SSL` | ⚠️ Optional | `true` | Set to `false` to disable TLS certificate verification |
| `NEXTCLOUD_CA_BUNDLE` | ⚠️ Optional | - | Path to a PEM CA bundle file for custom certificate authorities |
### Scope
These settings apply to **all** outbound connections to Nextcloud and its OIDC endpoints, including:
- Nextcloud API calls (Notes, Calendar, Contacts, WebDAV, etc.)
- OIDC discovery and token endpoints
- OAuth client registration (DCR)
- Health checks
They do **not** affect connections to internal services (Ollama, Qdrant, Unstructured) which have their own SSL configuration.
---
## Semantic Search Configuration (Optional)
**New in v0.58.0:** Simplified semantic search configuration with automatic dependency resolution.
-13
View File
@@ -217,19 +217,6 @@ NEXTCLOUD_PASSWORD=
#CUSTOM_PROCESSOR_TIMEOUT=60
#CUSTOM_PROCESSOR_TYPES=application/pdf,image/jpeg,image/png
# ===== SSL/TLS =====
# For Nextcloud behind reverse proxies with self-signed or private CA certificates
#
# Disable TLS certificate verification (insecure, development only):
#NEXTCLOUD_VERIFY_SSL=false
#
# Use a custom CA bundle (path to PEM file):
#NEXTCLOUD_CA_BUNDLE=/etc/ssl/certs/my-ca.pem
#
# Docker example: mount the CA bundle as a volume
# docker run -v /path/to/ca.pem:/etc/ssl/certs/my-ca.pem:ro \
# -e NEXTCLOUD_CA_BUNDLE=/etc/ssl/certs/my-ca.pem ...
# ===== SECURITY & ADVANCED =====
# Cookie security (browser UI)
# Auto-detects from NEXTCLOUD_HOST protocol if not set
+2 -8
View File
@@ -21,8 +21,6 @@ import httpx
from starlette.requests import Request
from starlette.responses import JSONResponse
from ..http import nextcloud_httpx_client
if TYPE_CHECKING:
from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
@@ -254,9 +252,7 @@ async def provision_app_password(request: Request) -> JSONResponse:
# Validate app password against Nextcloud
try:
async with nextcloud_httpx_client(
timeout=NEXTCLOUD_VALIDATION_TIMEOUT
) as client:
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(
@@ -384,9 +380,7 @@ async def delete_app_password(request: Request) -> JSONResponse:
nextcloud_host = settings.nextcloud_host
try:
async with nextcloud_httpx_client(
timeout=NEXTCLOUD_VALIDATION_TIMEOUT
) as client:
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,
+5 -6
View File
@@ -10,6 +10,7 @@ All endpoints require OAuth bearer token authentication via UnifiedTokenVerifier
import logging
import httpx
from starlette.requests import Request
from starlette.responses import JSONResponse
@@ -19,8 +20,6 @@ from nextcloud_mcp_server.api.management import (
validate_token_and_get_user,
)
from ..http import nextcloud_httpx_client
logger = logging.getLogger(__name__)
@@ -58,7 +57,7 @@ async def get_installed_apps(request: Request) -> JSONResponse:
raise ValueError("Nextcloud host not configured")
# Create authenticated HTTP client
async with nextcloud_httpx_client(
async with httpx.AsyncClient(
base_url=nextcloud_host,
headers={"Authorization": f"Bearer {token}"},
timeout=30.0,
@@ -130,7 +129,7 @@ async def list_webhooks(request: Request) -> JSONResponse:
raise ValueError("Nextcloud host not configured")
# Create authenticated HTTP client
async with nextcloud_httpx_client(
async with httpx.AsyncClient(
base_url=nextcloud_host,
headers={"Authorization": f"Bearer {token}"},
timeout=30.0,
@@ -211,7 +210,7 @@ async def create_webhook(request: Request) -> JSONResponse:
raise ValueError("Nextcloud host not configured")
# Create authenticated HTTP client
async with nextcloud_httpx_client(
async with httpx.AsyncClient(
base_url=nextcloud_host,
headers={"Authorization": f"Bearer {token}"},
timeout=30.0,
@@ -287,7 +286,7 @@ async def delete_webhook(request: Request) -> JSONResponse:
raise ValueError("Nextcloud host not configured")
# Create authenticated HTTP client
async with nextcloud_httpx_client(
async with httpx.AsyncClient(
base_url=nextcloud_host,
headers={"Authorization": f"Bearer {token}"},
timeout=30.0,
+3 -4
View File
@@ -58,7 +58,6 @@ from nextcloud_mcp_server.config_validators import (
)
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.http import nextcloud_httpx_client
from nextcloud_mcp_server.observability import (
ObservabilityMiddleware,
setup_metrics,
@@ -691,7 +690,7 @@ async def setup_oauth_config():
logger.info(f"Performing OIDC discovery: {discovery_url}")
# Perform OIDC discovery
async with nextcloud_httpx_client(follow_redirects=True) as client:
async with httpx.AsyncClient(follow_redirects=True) as client:
response = await client.get(discovery_url)
response.raise_for_status()
discovery = response.json()
@@ -995,7 +994,7 @@ async def setup_oauth_config_for_multi_user_basic(
# Perform OIDC discovery
try:
async with nextcloud_httpx_client(
async with httpx.AsyncClient(
timeout=30.0, follow_redirects=True
) as http_client:
response = await http_client.get(discovery_url)
@@ -1976,7 +1975,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
# Try to connect to Nextcloud
start_time = time.time()
try:
async with nextcloud_httpx_client(timeout=2.0) as client:
async with httpx.AsyncClient(timeout=2.0) as client:
response = await client.get(f"{nextcloud_host}/status.php")
duration = time.time() - start_time
if response.status_code == 200:
@@ -9,7 +9,7 @@ import logging
import time
from typing import Optional
from ..http import nextcloud_httpx_client
import httpx
logger = logging.getLogger(__name__)
@@ -60,7 +60,7 @@ class AstrolabeClient:
# Discover token endpoint
discovery_url = f"{self.nextcloud_host}/.well-known/openid-configuration"
async with nextcloud_httpx_client() as client:
async with httpx.AsyncClient() as client:
logger.debug(f"Discovering token endpoint from {discovery_url}")
discovery_resp = await client.get(discovery_url)
discovery_resp.raise_for_status()
@@ -107,7 +107,7 @@ class AstrolabeClient:
token = await self.get_access_token()
url = f"{self.nextcloud_host}/apps/astrolabe/api/v1/background-sync/credentials/{user_id}"
async with nextcloud_httpx_client() as client:
async with httpx.AsyncClient() as client:
logger.debug(f"Retrieving app password for user: {user_id}")
response = await client.get(
@@ -22,8 +22,6 @@ from nextcloud_mcp_server.auth.userinfo_routes import (
_query_idp_userinfo,
)
from ..http import nextcloud_httpx_client
logger = logging.getLogger(__name__)
@@ -144,7 +142,7 @@ async def oauth_login(request: Request) -> RedirectResponse | JSONResponse:
)
# Fetch authorization endpoint
async with nextcloud_httpx_client() as http_client:
async with httpx.AsyncClient() as http_client:
response = await http_client.get(discovery_url)
response.raise_for_status()
discovery = response.json()
@@ -288,7 +286,7 @@ async def oauth_login_callback(request: Request) -> RedirectResponse | HTMLRespo
if code_verifier:
token_params["code_verifier"] = code_verifier
async with nextcloud_httpx_client() as http_client:
async with httpx.AsyncClient() as http_client:
response = await http_client.post(
oauth_client.token_endpoint,
data=token_params,
@@ -298,7 +296,7 @@ async def oauth_login_callback(request: Request) -> RedirectResponse | HTMLRespo
else:
# Integrated mode (Nextcloud OIDC)
discovery_url = oauth_config.get("discovery_url")
async with nextcloud_httpx_client() as http_client:
async with httpx.AsyncClient() as http_client:
response = await http_client.get(discovery_url)
response.raise_for_status()
discovery = response.json()
@@ -316,7 +314,7 @@ async def oauth_login_callback(request: Request) -> RedirectResponse | HTMLRespo
if code_verifier:
token_params["code_verifier"] = code_verifier
async with nextcloud_httpx_client() as http_client:
async with httpx.AsyncClient() as http_client:
response = await http_client.post(
token_endpoint,
data=token_params,
@@ -10,8 +10,6 @@ import httpx
from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
from ..http import nextcloud_httpx_client
logger = logging.getLogger(__name__)
@@ -134,7 +132,7 @@ async def register_client(
logger.info(f"Registering OAuth client with Nextcloud: {client_name}")
logger.debug(f"Registration endpoint: {registration_endpoint}")
async with nextcloud_httpx_client(timeout=30.0) as client:
async with httpx.AsyncClient(timeout=30.0) as client:
try:
response = await client.post(
registration_endpoint,
@@ -231,7 +229,7 @@ async def delete_client(
logger.info(f"Deleting OAuth client: {client_id[:16]}...")
logger.debug(f"Deletion endpoint: {deletion_endpoint}")
async with nextcloud_httpx_client(timeout=30.0) as http_client:
async with httpx.AsyncClient(timeout=30.0) as http_client:
for attempt in range(max_retries):
try:
# Prefer RFC 7592 Bearer token authentication
+1 -3
View File
@@ -18,8 +18,6 @@ from urllib.parse import urlencode, urlparse
import httpx
from ..http import nextcloud_httpx_client
logger = logging.getLogger(__name__)
@@ -109,7 +107,7 @@ class KeycloakOAuthClient:
async def _get_http_client(self) -> httpx.AsyncClient:
"""Get or create HTTP client"""
if self._http_client is None:
self._http_client = nextcloud_httpx_client(timeout=30.0)
self._http_client = httpx.AsyncClient(timeout=30.0)
return self._http_client
async def close(self) -> None:
+5 -6
View File
@@ -27,6 +27,7 @@ import time
from base64 import urlsafe_b64encode
from urllib.parse import urlencode
import httpx
import jwt
from starlette.requests import Request
from starlette.responses import JSONResponse, RedirectResponse
@@ -34,8 +35,6 @@ from starlette.responses import JSONResponse, RedirectResponse
from nextcloud_mcp_server.auth.client_registry import get_client_registry
from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
from ..http import nextcloud_httpx_client
logger = logging.getLogger(__name__)
@@ -219,7 +218,7 @@ async def oauth_authorize(request: Request) -> RedirectResponse | JSONResponse:
)
# Fetch authorization endpoint from discovery
async with nextcloud_httpx_client() as http_client:
async with httpx.AsyncClient() as http_client:
response = await http_client.get(discovery_url)
response.raise_for_status()
discovery = response.json()
@@ -355,7 +354,7 @@ async def oauth_authorize_nextcloud(
status_code=500,
)
async with nextcloud_httpx_client() as http_client:
async with httpx.AsyncClient() as http_client:
response = await http_client.get(discovery_url)
response.raise_for_status()
discovery = response.json()
@@ -463,7 +462,7 @@ async def oauth_callback_nextcloud(request: Request):
callback_uri = f"{mcp_server_url}/oauth/callback"
discovery_url = oauth_config.get("discovery_url")
async with nextcloud_httpx_client() as http_client:
async with httpx.AsyncClient() as http_client:
response = await http_client.get(discovery_url)
response.raise_for_status()
discovery = response.json()
@@ -483,7 +482,7 @@ async def oauth_callback_nextcloud(request: Request):
token_params["code_verifier"] = code_verifier
# Exchange code for tokens
async with nextcloud_httpx_client() as http_client:
async with httpx.AsyncClient() as http_client:
response = await http_client.post(
token_endpoint,
data=token_params,
+1 -3
View File
@@ -25,8 +25,6 @@ import jwt
from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
from nextcloud_mcp_server.auth.token_exchange import exchange_token_for_delegation
from ..http import nextcloud_httpx_client
logger = logging.getLogger(__name__)
@@ -138,7 +136,7 @@ class TokenBrokerService:
async def _get_http_client(self) -> httpx.AsyncClient:
"""Get or create HTTP client."""
if self._http_client is None:
self._http_client = nextcloud_httpx_client(
self._http_client = httpx.AsyncClient(
timeout=httpx.Timeout(30.0), follow_redirects=True
)
return self._http_client
+1 -2
View File
@@ -20,7 +20,6 @@ import httpx
import jwt
from ..config import get_settings
from ..http import nextcloud_httpx_client
from .storage import RefreshTokenStorage
logger = logging.getLogger(__name__)
@@ -69,7 +68,7 @@ class TokenExchangeService:
self.storage: Optional[RefreshTokenStorage] = None
# Create HTTP client
self.http_client = nextcloud_httpx_client(
self.http_client = httpx.AsyncClient(
timeout=30.0,
follow_redirects=True,
)
@@ -31,8 +31,6 @@ from nextcloud_mcp_server.observability.metrics import (
record_oauth_token_validation,
)
from ..http import nextcloud_httpx_client
logger = logging.getLogger(__name__)
@@ -63,7 +61,7 @@ class UnifiedTokenVerifier(TokenVerifier):
self.mode = "exchange" if settings.enable_token_exchange else "multi-audience"
# Common components for all modes
self.http_client = nextcloud_httpx_client(timeout=10.0)
self.http_client = httpx.AsyncClient(timeout=10.0)
# JWT verification support
self.jwks_client: PyJWKClient | None = None
+3 -4
View File
@@ -13,6 +13,7 @@ import traceback
from pathlib import Path
from typing import Any
import httpx
from jinja2 import Environment, FileSystemLoader
from starlette.authentication import requires
from starlette.requests import Request
@@ -21,8 +22,6 @@ from starlette.responses import HTMLResponse, JSONResponse
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.config import get_settings
from ..http import nextcloud_httpx_client
logger = logging.getLogger(__name__)
# Setup Jinja2 environment for templates
@@ -258,7 +257,7 @@ async def _get_userinfo_endpoint(oauth_ctx: dict[str, Any]) -> str | None:
return None
try:
async with nextcloud_httpx_client(timeout=10.0) as client:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(discovery_url)
response.raise_for_status()
discovery = response.json()
@@ -291,7 +290,7 @@ async def _query_idp_userinfo(
User info dictionary from IdP, or None if query fails
"""
try:
async with nextcloud_httpx_client(timeout=10.0) as client:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
userinfo_uri,
headers={"Authorization": f"Bearer {access_token_str}"},
+2 -4
View File
@@ -20,8 +20,6 @@ from nextcloud_mcp_server.server.webhook_presets import (
get_preset,
)
from ..http import nextcloud_httpx_client
logger = logging.getLogger(__name__)
@@ -142,7 +140,7 @@ async def _get_authenticated_client(request: Request) -> httpx.AsyncClient:
assert nextcloud_host is not None # Type narrowing for type checker
assert username is not None and password is not None # Type narrowing
return nextcloud_httpx_client(
return httpx.AsyncClient(
base_url=nextcloud_host,
auth=(username, password),
timeout=30.0,
@@ -165,7 +163,7 @@ async def _get_authenticated_client(request: Request) -> httpx.AsyncClient:
if not nextcloud_host:
raise RuntimeError("Nextcloud host not configured")
return nextcloud_httpx_client(
return httpx.AsyncClient(
base_url=nextcloud_host,
headers={"Authorization": f"Bearer {access_token}"},
timeout=30.0,
+2 -2
View File
@@ -4,6 +4,7 @@ import os
from httpx import (
AsyncBaseTransport,
AsyncClient,
AsyncHTTPTransport,
Auth,
BasicAuth,
Request,
@@ -12,7 +13,6 @@ from httpx import (
)
from ..controllers.notes_search import NotesSearchController
from ..http import nextcloud_httpx_transport
from .calendar import CalendarClient
from .contacts import ContactsClient
from .cookbook import CookbookClient
@@ -67,7 +67,7 @@ class NextcloudClient:
self._client = AsyncClient(
base_url=base_url,
auth=auth,
transport=AsyncDisableCookieTransport(nextcloud_httpx_transport()),
transport=AsyncDisableCookieTransport(AsyncHTTPTransport()),
event_hooks={"request": [log_request], "response": [log_response]},
timeout=Timeout(timeout=30, connect=5),
)
-3
View File
@@ -13,8 +13,6 @@ from icalendar import Alarm, Calendar, vRecur
from icalendar import Event as ICalEvent
from icalendar import Todo as ICalTodo
from ..config import get_nextcloud_ssl_verify
logger = logging.getLogger(__name__)
@@ -36,7 +34,6 @@ class CalendarClient:
url=f"{base_url}/remote.php/dav/",
username=username,
auth=auth,
ssl_verify_cert=get_nextcloud_ssl_verify(), # type: ignore[arg-type] # caldav types say bool|str but passes through to httpx which accepts SSLContext
)
self._calendar_home_url = f"{base_url}/remote.php/dav/calendars/{username}/"
+1 -44
View File
@@ -2,7 +2,6 @@ import logging
import logging.config
import os
import socket
import ssl
from dataclasses import dataclass
from enum import Enum
from typing import Any, Optional
@@ -182,10 +181,6 @@ class Settings:
nextcloud_username: Optional[str] = None
nextcloud_password: Optional[str] = None
# Nextcloud SSL/TLS settings
nextcloud_verify_ssl: bool = True
nextcloud_ca_bundle: Optional[str] = None
# ADR-005: Token Audience Validation (required for OAuth mode)
nextcloud_mcp_server_url: Optional[str] = None # MCP server URL (used as audience)
nextcloud_resource_uri: Optional[str] = None # Nextcloud resource identifier
@@ -257,25 +252,9 @@ class Settings:
log_include_trace_context: bool = True
def __post_init__(self):
"""Validate configuration and set defaults."""
"""Validate Qdrant configuration and set defaults."""
logger = logging.getLogger(__name__)
# Validate SSL/TLS configuration
if not self.nextcloud_verify_ssl:
logger.warning(
"NEXTCLOUD_VERIFY_SSL is disabled. "
"TLS certificate verification is turned off for all Nextcloud connections. "
"This is insecure and should only be used for development/testing."
)
if self.nextcloud_ca_bundle:
import os as _os
if not _os.path.isfile(self.nextcloud_ca_bundle):
raise ValueError(
f"NEXTCLOUD_CA_BUNDLE path does not exist: {self.nextcloud_ca_bundle}"
)
logger.info("Using custom CA bundle: %s", self.nextcloud_ca_bundle)
# Ensure mutual exclusivity
if self.qdrant_url and self.qdrant_location:
raise ValueError(
@@ -525,11 +504,6 @@ def get_settings() -> Settings:
nextcloud_host=os.getenv("NEXTCLOUD_HOST"),
nextcloud_username=os.getenv("NEXTCLOUD_USERNAME"),
nextcloud_password=os.getenv("NEXTCLOUD_PASSWORD"),
# Nextcloud SSL/TLS settings
nextcloud_verify_ssl=(
os.getenv("NEXTCLOUD_VERIFY_SSL", "true").lower() == "true"
),
nextcloud_ca_bundle=os.getenv("NEXTCLOUD_CA_BUNDLE"),
# ADR-005: Token Audience Validation
nextcloud_mcp_server_url=os.getenv("NEXTCLOUD_MCP_SERVER_URL"),
nextcloud_resource_uri=os.getenv("NEXTCLOUD_RESOURCE_URI"),
@@ -595,20 +569,3 @@ def get_settings() -> Settings:
log_include_trace_context=os.getenv("LOG_INCLUDE_TRACE_CONTEXT", "true").lower()
== "true",
)
def get_nextcloud_ssl_verify() -> bool | ssl.SSLContext:
"""Return the SSL verification setting for Nextcloud connections.
Returns:
- False if NEXTCLOUD_VERIFY_SSL=false (disable verification)
- ssl.SSLContext if NEXTCLOUD_CA_BUNDLE is set (custom CA)
- True otherwise (default system CA verification)
"""
settings = get_settings()
if not settings.nextcloud_verify_ssl:
return False
if settings.nextcloud_ca_bundle:
ctx = ssl.create_default_context(cafile=settings.nextcloud_ca_bundle)
return ctx
return True
-45
View File
@@ -1,45 +0,0 @@
"""Centralized HTTP client factory for Nextcloud connections.
All outbound connections to Nextcloud (API calls, OIDC endpoints) should use
these factories to ensure consistent SSL/TLS configuration from environment
variables (NEXTCLOUD_VERIFY_SSL, NEXTCLOUD_CA_BUNDLE).
"""
from typing import Any
import httpx
from .config import get_nextcloud_ssl_verify
def nextcloud_httpx_client(**kwargs: Any) -> httpx.AsyncClient:
"""Create an httpx.AsyncClient with Nextcloud SSL settings applied.
Reads NEXTCLOUD_VERIFY_SSL and NEXTCLOUD_CA_BUNDLE from the environment
via ``get_nextcloud_ssl_verify()``. Caller-supplied ``verify`` kwarg
takes precedence if explicitly provided.
Args:
**kwargs: Forwarded to ``httpx.AsyncClient()``.
Returns:
Configured ``httpx.AsyncClient``.
"""
kwargs.setdefault("verify", get_nextcloud_ssl_verify())
return httpx.AsyncClient(**kwargs)
def nextcloud_httpx_transport(**kwargs: Any) -> httpx.AsyncHTTPTransport:
"""Create an httpx.AsyncHTTPTransport with Nextcloud SSL settings applied.
Used by ``NextcloudClient`` which wraps the transport in
``AsyncDisableCookieTransport``.
Args:
**kwargs: Forwarded to ``httpx.AsyncHTTPTransport()``.
Returns:
Configured ``httpx.AsyncHTTPTransport``.
"""
kwargs.setdefault("verify", get_nextcloud_ssl_verify())
return httpx.AsyncHTTPTransport(**kwargs)
+2 -3
View File
@@ -11,6 +11,7 @@ import secrets
from typing import Optional
from urllib.parse import urlencode
import httpx
import jwt
from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.auth.provider import AccessToken
@@ -23,8 +24,6 @@ from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
from nextcloud_mcp_server.auth.userinfo_routes import _query_idp_userinfo
from ..http import nextcloud_httpx_client
logger = logging.getLogger(__name__)
@@ -70,7 +69,7 @@ async def extract_user_id_from_token(ctx: Context) -> str:
"OIDC_DISCOVERY_URI",
"http://localhost:8080/.well-known/openid-configuration",
)
async with nextcloud_httpx_client() as http_client:
async with httpx.AsyncClient() as http_client:
discovery_response = await http_client.get(oidc_discovery_uri)
discovery_response.raise_for_status()
discovery = discovery_response.json()
@@ -185,11 +185,7 @@ async def test_provision_app_password_success(temp_storage, mocker):
# Mock settings (imported locally in the function)
mocker.patch(
"nextcloud_mcp_server.config.get_settings",
return_value=MagicMock(
nextcloud_host="http://localhost:8080",
nextcloud_verify_ssl=True,
nextcloud_ca_bundle=None,
),
return_value=MagicMock(nextcloud_host="http://localhost:8080"),
)
# Mock httpx client for Nextcloud validation
@@ -234,11 +230,7 @@ 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",
nextcloud_verify_ssl=True,
nextcloud_ca_bundle=None,
),
return_value=MagicMock(nextcloud_host="http://localhost:8080"),
)
# Mock httpx client to return 401
@@ -357,11 +349,7 @@ async def test_delete_app_password_success(temp_storage, mocker):
# Mock settings (imported locally in the function)
mocker.patch(
"nextcloud_mcp_server.config.get_settings",
return_value=MagicMock(
nextcloud_host="http://localhost:8080",
nextcloud_verify_ssl=True,
nextcloud_ca_bundle=None,
),
return_value=MagicMock(nextcloud_host="http://localhost:8080"),
)
# Mock httpx client for Nextcloud validation
@@ -405,11 +393,7 @@ async def test_delete_app_password_not_found(temp_storage, mocker):
# Mock settings (imported locally in the function)
mocker.patch(
"nextcloud_mcp_server.config.get_settings",
return_value=MagicMock(
nextcloud_host="http://localhost:8080",
nextcloud_verify_ssl=True,
nextcloud_ca_bundle=None,
),
return_value=MagicMock(nextcloud_host="http://localhost:8080"),
)
# Mock httpx client for Nextcloud validation
@@ -448,11 +432,7 @@ 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",
nextcloud_verify_ssl=True,
nextcloud_ca_bundle=None,
),
return_value=MagicMock(nextcloud_host="http://localhost:8080"),
)
# Mock httpx client to return 401
@@ -522,11 +502,7 @@ 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",
nextcloud_verify_ssl=True,
nextcloud_ca_bundle=None,
),
return_value=MagicMock(nextcloud_host="http://localhost:8080"),
)
# Mock httpx client to return 401 (failed validation)
@@ -585,11 +561,7 @@ 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",
nextcloud_verify_ssl=True,
nextcloud_ca_bundle=None,
),
return_value=MagicMock(nextcloud_host="http://localhost:8080"),
)
# Mock httpx client to return 401
-178
View File
@@ -1,178 +0,0 @@
"""Tests for SSL/TLS configuration (NEXTCLOUD_VERIFY_SSL, NEXTCLOUD_CA_BUNDLE)."""
import logging
import os
import ssl
from unittest.mock import patch
import certifi
import httpx
import pytest
from nextcloud_mcp_server.config import Settings, get_nextcloud_ssl_verify, get_settings
from nextcloud_mcp_server.http import nextcloud_httpx_client, nextcloud_httpx_transport
class TestSSLSettings:
"""Test SSL/TLS fields on Settings dataclass."""
def test_defaults(self):
"""verify_ssl defaults to True, ca_bundle defaults to None."""
settings = Settings()
assert settings.nextcloud_verify_ssl is True
assert settings.nextcloud_ca_bundle is None
def test_verify_ssl_false_logs_warning(self, caplog):
caplog.set_level(logging.WARNING, logger="nextcloud_mcp_server.config")
Settings(nextcloud_verify_ssl=False)
assert "NEXTCLOUD_VERIFY_SSL is disabled" in caplog.text
def test_ca_bundle_nonexistent_path_raises(self):
with pytest.raises(ValueError, match="does not exist"):
Settings(nextcloud_ca_bundle="/nonexistent/path/ca.pem")
def test_ca_bundle_existing_path_logs_info(self, caplog, tmp_path):
ca_file = tmp_path / "ca.pem"
ca_file.write_text(
"-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----\n"
)
caplog.set_level(logging.INFO, logger="nextcloud_mcp_server.config")
settings = Settings(nextcloud_ca_bundle=str(ca_file))
assert settings.nextcloud_ca_bundle == str(ca_file)
assert "Using custom CA bundle" in caplog.text
class TestGetNextcloudSSLVerify:
"""Test the get_nextcloud_ssl_verify() helper function."""
def test_default_returns_true(self):
env = {
"NEXTCLOUD_VERIFY_SSL": "true",
}
with patch.dict(os.environ, env, clear=False):
# Clear any cached settings
result = get_nextcloud_ssl_verify()
assert result is True
def test_verify_false_returns_false(self):
env = {
"NEXTCLOUD_VERIFY_SSL": "false",
}
with patch.dict(os.environ, env, clear=False):
with patch(
"nextcloud_mcp_server.config.get_settings",
return_value=Settings(nextcloud_verify_ssl=False),
):
result = get_nextcloud_ssl_verify()
assert result is False
def test_ca_bundle_returns_ssl_context(self):
ca_bundle = certifi.where()
with patch(
"nextcloud_mcp_server.config.get_settings",
return_value=Settings(nextcloud_ca_bundle=ca_bundle),
):
result = get_nextcloud_ssl_verify()
assert isinstance(result, ssl.SSLContext)
def test_ca_bundle_ssl_context_has_loaded_certs(self):
"""SSLContext created from CA bundle should have loaded certificates."""
ca_bundle = certifi.where()
with patch(
"nextcloud_mcp_server.config.get_settings",
return_value=Settings(nextcloud_ca_bundle=ca_bundle),
):
result = get_nextcloud_ssl_verify()
assert isinstance(result, ssl.SSLContext)
stats = result.cert_store_stats()
assert stats["x509_ca"] > 0
def test_verify_false_takes_precedence_over_ca_bundle(self, tmp_path):
"""When verify_ssl=False, ca_bundle is ignored (False wins)."""
ca_file = tmp_path / "ca.pem"
ca_file.write_text(
"-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----\n"
)
with patch(
"nextcloud_mcp_server.config.get_settings",
return_value=Settings(
nextcloud_verify_ssl=False,
nextcloud_ca_bundle=str(ca_file),
),
):
result = get_nextcloud_ssl_verify()
assert result is False
class TestGetSettingsSSLEnvVars:
"""Test that get_settings() reads SSL env vars correctly."""
def test_verify_ssl_env_true(self):
env = {"NEXTCLOUD_VERIFY_SSL": "true"}
with patch.dict(os.environ, env, clear=False):
settings = get_settings()
assert settings.nextcloud_verify_ssl is True
def test_verify_ssl_env_false(self):
env = {"NEXTCLOUD_VERIFY_SSL": "false"}
with patch.dict(os.environ, env, clear=False):
settings = get_settings()
assert settings.nextcloud_verify_ssl is False
def test_verify_ssl_env_missing_defaults_true(self):
with patch.dict(os.environ, {}, clear=False):
# Remove NEXTCLOUD_VERIFY_SSL if it exists
os.environ.pop("NEXTCLOUD_VERIFY_SSL", None)
settings = get_settings()
assert settings.nextcloud_verify_ssl is True
def test_ca_bundle_env(self, tmp_path):
ca_file = tmp_path / "ca.pem"
ca_file.write_text(
"-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----\n"
)
env = {"NEXTCLOUD_CA_BUNDLE": str(ca_file)}
with patch.dict(os.environ, env, clear=False):
settings = get_settings()
assert settings.nextcloud_ca_bundle == str(ca_file)
class TestHTTPClientFactory:
"""Test that factory functions apply verify correctly."""
def test_client_applies_verify_true(self):
with patch(
"nextcloud_mcp_server.http.get_nextcloud_ssl_verify", return_value=True
):
client = nextcloud_httpx_client()
# httpx stores verify as an SSLConfig; check the _transport
assert isinstance(client, httpx.AsyncClient)
def test_client_applies_verify_false(self):
with patch(
"nextcloud_mcp_server.http.get_nextcloud_ssl_verify", return_value=False
):
client = nextcloud_httpx_client()
assert isinstance(client, httpx.AsyncClient)
def test_client_caller_override_takes_precedence(self):
"""Caller-supplied verify kwarg should not be overridden."""
with patch(
"nextcloud_mcp_server.http.get_nextcloud_ssl_verify", return_value=True
):
client = nextcloud_httpx_client(verify=False)
assert isinstance(client, httpx.AsyncClient)
def test_transport_applies_verify(self):
with patch(
"nextcloud_mcp_server.http.get_nextcloud_ssl_verify", return_value=False
):
transport = nextcloud_httpx_transport()
assert isinstance(transport, httpx.AsyncHTTPTransport)
def test_client_passes_extra_kwargs(self):
with patch(
"nextcloud_mcp_server.http.get_nextcloud_ssl_verify", return_value=True
):
client = nextcloud_httpx_client(timeout=5.0, follow_redirects=True)
assert isinstance(client, httpx.AsyncClient)