fix: convert CA bundle path to ssl.SSLContext to avoid httpx deprecation warning
httpx emits a DeprecationWarning when verify=<str> is passed, recommending ssl.SSLContext instead. This affected both our httpx client factories and the caldav library passthrough. Changed get_nextcloud_ssl_verify() to return bool | ssl.SSLContext instead of bool | str by constructing an SSLContext when NEXTCLOUD_CA_BUNDLE is set. All downstream consumers (httpx, caldav) natively accept ssl.SSLContext. Also fixed app password endpoint tests that used overly broad MagicMock (auto-generated truthy nextcloud_ca_bundle attribute). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -2,6 +2,7 @@ import logging
|
||||
import logging.config
|
||||
import os
|
||||
import socket
|
||||
import ssl
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from typing import Any, Optional
|
||||
@@ -596,17 +597,18 @@ def get_settings() -> Settings:
|
||||
)
|
||||
|
||||
|
||||
def get_nextcloud_ssl_verify() -> bool | str:
|
||||
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)
|
||||
- CA bundle path if NEXTCLOUD_CA_BUNDLE is set (custom CA)
|
||||
- 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:
|
||||
return settings.nextcloud_ca_bundle
|
||||
ctx = ssl.create_default_context(cafile=settings.nextcloud_ca_bundle)
|
||||
return ctx
|
||||
return True
|
||||
|
||||
@@ -185,7 +185,11 @@ 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"),
|
||||
return_value=MagicMock(
|
||||
nextcloud_host="http://localhost:8080",
|
||||
nextcloud_verify_ssl=True,
|
||||
nextcloud_ca_bundle=None,
|
||||
),
|
||||
)
|
||||
|
||||
# Mock httpx client for Nextcloud validation
|
||||
@@ -230,7 +234,11 @@ 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"),
|
||||
return_value=MagicMock(
|
||||
nextcloud_host="http://localhost:8080",
|
||||
nextcloud_verify_ssl=True,
|
||||
nextcloud_ca_bundle=None,
|
||||
),
|
||||
)
|
||||
|
||||
# Mock httpx client to return 401
|
||||
@@ -349,7 +357,11 @@ 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"),
|
||||
return_value=MagicMock(
|
||||
nextcloud_host="http://localhost:8080",
|
||||
nextcloud_verify_ssl=True,
|
||||
nextcloud_ca_bundle=None,
|
||||
),
|
||||
)
|
||||
|
||||
# Mock httpx client for Nextcloud validation
|
||||
@@ -393,7 +405,11 @@ 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"),
|
||||
return_value=MagicMock(
|
||||
nextcloud_host="http://localhost:8080",
|
||||
nextcloud_verify_ssl=True,
|
||||
nextcloud_ca_bundle=None,
|
||||
),
|
||||
)
|
||||
|
||||
# Mock httpx client for Nextcloud validation
|
||||
@@ -432,7 +448,11 @@ 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"),
|
||||
return_value=MagicMock(
|
||||
nextcloud_host="http://localhost:8080",
|
||||
nextcloud_verify_ssl=True,
|
||||
nextcloud_ca_bundle=None,
|
||||
),
|
||||
)
|
||||
|
||||
# Mock httpx client to return 401
|
||||
@@ -502,7 +522,11 @@ 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"),
|
||||
return_value=MagicMock(
|
||||
nextcloud_host="http://localhost:8080",
|
||||
nextcloud_verify_ssl=True,
|
||||
nextcloud_ca_bundle=None,
|
||||
),
|
||||
)
|
||||
|
||||
# Mock httpx client to return 401 (failed validation)
|
||||
@@ -561,7 +585,11 @@ 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"),
|
||||
return_value=MagicMock(
|
||||
nextcloud_host="http://localhost:8080",
|
||||
nextcloud_verify_ssl=True,
|
||||
nextcloud_ca_bundle=None,
|
||||
),
|
||||
)
|
||||
|
||||
# Mock httpx client to return 401
|
||||
|
||||
@@ -2,8 +2,10 @@
|
||||
|
||||
import logging
|
||||
import os
|
||||
import ssl
|
||||
from unittest.mock import patch
|
||||
|
||||
import certifi
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
@@ -64,17 +66,26 @@ class TestGetNextcloudSSLVerify:
|
||||
result = get_nextcloud_ssl_verify()
|
||||
assert result is False
|
||||
|
||||
def test_ca_bundle_returns_path(self, tmp_path):
|
||||
ca_file = tmp_path / "ca.pem"
|
||||
ca_file.write_text(
|
||||
"-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----\n"
|
||||
)
|
||||
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=str(ca_file)),
|
||||
return_value=Settings(nextcloud_ca_bundle=ca_bundle),
|
||||
):
|
||||
result = get_nextcloud_ssl_verify()
|
||||
assert result == str(ca_file)
|
||||
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)."""
|
||||
|
||||
Reference in New Issue
Block a user