Chris CoutinhoandClaude Opus 4.7
f2b7bf132f
Round-2 fixes after the bot review on PR #798 plus two user follow-ups (self-signed Postgres support; asyncpg should be a PyPI extra). Folded into the same PR rather than a follow-up since the work is still unmerged. Security -------- - Mask database credentials in all 5 log call sites (storage.py × 4, migrations.py × 1) via a new `mask_db_password()` helper in config.py. Uses SQLAlchemy's `make_url(...).render_as_string(hide_password=True)` with a regex fallback so the masking path never raises. - New `tests/unit/test_storage_logging.py` asserts a sentinel password never appears in `caplog` during `RefreshTokenStorage.initialize()`. Distribution ------------ - `asyncpg` moved to `[project.optional-dependencies] postgres` so a vanilla `pip install nextcloud-mcp-server` no longer pulls in the ~5 MB C extension. The Docker image runs `uv sync --extra postgres`, so containerized deployments are unchanged. - When `DATABASE_URL=postgresql+asyncpg://...` is set on a venv missing the extra, `RefreshTokenStorage.initialize()` raises a friendly RuntimeError pointing at `[postgres]` rather than the generic ModuleNotFoundError. TLS for the Postgres backend ---------------------------- - New `DATABASE_VERIFY_SSL` + `DATABASE_CA_BUNDLE` env vars mirror the existing `NEXTCLOUD_VERIFY_SSL` / `NEXTCLOUD_CA_BUNDLE` pattern (validators in Settings.__post_init__, `get_database_ssl()` helper alongside `get_nextcloud_ssl_verify()`). `DATABASE_VERIFY_SSL=false` wins over `DATABASE_CA_BUNDLE` for incident-response convenience. - Default is **None** rather than True — keeps PR #798's behavior intact for cluster-internal Postgres that runs without TLS. Operators opt into verify-full or supply a private CA. ADR-026 records the reasoning vs the Nextcloud HTTPS default. - Engine factory in `storage.py` passes `ssl` via `connect_args` only when `get_database_ssl()` returns non-None; otherwise asyncpg's default (`prefer`) applies. - Storage logs which TLS mode is active at INFO (no secret material). Configurable connection pool ---------------------------- - `DATABASE_POOL_SIZE` (default 10) and `DATABASE_MAX_OVERFLOW` (default 20) replace the hardcoded engine values. With many replicas this can blow past managed-Postgres `max_connections=100`; tune down for large fleets. - gte-1 / gte-0 validators in __post_init__ reject 0/negative pool sizes at startup with the offending value in the error. Consistency polish ------------------ - Migration 006: convert raw `op.execute("ALTER TABLE ... ADD COLUMN")` to `op.batch_alter_table(...).add_column(sa.Column("nonce", sa.Text))` for stylistic consistency with the rewritten 001-005. Downgrade now drops the column instead of being a no-op. - `registered_webhooks.created_at` standardized from `sa.Float` to `sa.BigInteger` (all other `*_at` columns); `store_webhook()` casts `time.time()` → `int`. - `is_sqlite_url()` made case-insensitive. Testing ------- - New `tests/integration/test_storage_postgres.py::test_cleanup_expired_roundtrip` exercises `cleanup_expired_tokens`, `cleanup_expired_sessions`, and `cleanup_expired_browser_sessions` — relies on DELETE rowcount, historically dialect-tricky. - `tests/unit/test_ssl_config.py` extended with `TestDatabaseSSLSettings` + `TestGetDatabaseSSL` classes (9 new tests) mirroring the existing Nextcloud SSL tests one-for-one. Docs ---- - `docs/configuration.md` Centralized-Storage section grew the four new env vars + a homelab example with a private CA. - `docs/ADR-026` grew Distribution, TLS, and `alembic/env.py` async-pattern subsections explaining the non-obvious design choices. Helm chart counterpart in cbcoutinho/helm-charts PR #34 (separate commit on `feat/nextcloud-mcp-server-database-url`). Verification ------------ - `uv run pytest tests/unit/` — 1025 passed. - `TEST_DATABASE_URL=... uv run pytest tests/integration/test_storage_postgres.py -m postgres` — 6 passed (including new cleanup test). - `uv run ruff check && uv run ruff format --check && uv run ty check -- nextcloud_mcp_server` — clean. Tracked on Astrolabe Cloud POC board, card #99. --- _This PR was generated with the help of AI, and reviewed by a Human_ Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
280 lines
10 KiB
Python
280 lines
10 KiB
Python
"""Tests for SSL/TLS configuration.
|
|
|
|
Covers two parallel patterns:
|
|
|
|
- ``NEXTCLOUD_VERIFY_SSL`` / ``NEXTCLOUD_CA_BUNDLE`` for the httpx
|
|
client talking to Nextcloud.
|
|
- ``DATABASE_VERIFY_SSL`` / ``DATABASE_CA_BUNDLE`` for the asyncpg
|
|
driver talking to a centralized Postgres backend (ADR-026).
|
|
|
|
The DB-side helper has a different default (``None`` instead of
|
|
``True``) because asyncpg's default ``prefer`` is the right back-compat
|
|
posture for cluster-internal Postgres — see ``get_database_ssl()``.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import ssl
|
|
from unittest.mock import patch
|
|
|
|
import certifi
|
|
import httpx
|
|
import pytest
|
|
|
|
from nextcloud_mcp_server.config import (
|
|
Settings,
|
|
_reload_config,
|
|
get_database_ssl,
|
|
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):
|
|
_reload_config()
|
|
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):
|
|
_reload_config()
|
|
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):
|
|
_reload_config()
|
|
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)
|
|
_reload_config()
|
|
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):
|
|
_reload_config()
|
|
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)
|
|
|
|
|
|
class TestDatabaseSSLSettings:
|
|
"""Test DATABASE_VERIFY_SSL / DATABASE_CA_BUNDLE fields on Settings (ADR-026)."""
|
|
|
|
def test_defaults(self):
|
|
"""Default is None / None — preserves PR #798's asyncpg ``prefer``."""
|
|
settings = Settings()
|
|
assert settings.database_verify_ssl is None
|
|
assert settings.database_ca_bundle is None
|
|
|
|
def test_verify_false_logs_warning(self, caplog):
|
|
caplog.set_level(logging.WARNING, logger="nextcloud_mcp_server.config")
|
|
Settings(database_verify_ssl=False)
|
|
assert "DATABASE_VERIFY_SSL is disabled" in caplog.text
|
|
|
|
def test_ca_bundle_nonexistent_path_raises(self):
|
|
with pytest.raises(ValueError, match="DATABASE_CA_BUNDLE path does not exist"):
|
|
Settings(database_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(database_ca_bundle=str(ca_file))
|
|
assert "custom CA bundle for Postgres backend" in caplog.text
|
|
|
|
|
|
class TestGetDatabaseSSL:
|
|
"""Test the get_database_ssl() helper (ADR-026)."""
|
|
|
|
def test_both_unset_returns_none(self):
|
|
"""The asyncpg-default opt-out path — no `ssl` kwarg passed."""
|
|
with patch(
|
|
"nextcloud_mcp_server.config.get_settings",
|
|
return_value=Settings(),
|
|
):
|
|
assert get_database_ssl() is None
|
|
|
|
def test_verify_true_returns_true(self):
|
|
with patch(
|
|
"nextcloud_mcp_server.config.get_settings",
|
|
return_value=Settings(database_verify_ssl=True),
|
|
):
|
|
assert get_database_ssl() is True
|
|
|
|
def test_verify_false_returns_false(self):
|
|
with patch(
|
|
"nextcloud_mcp_server.config.get_settings",
|
|
return_value=Settings(database_verify_ssl=False),
|
|
):
|
|
assert get_database_ssl() 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(database_ca_bundle=ca_bundle),
|
|
):
|
|
result = get_database_ssl()
|
|
assert isinstance(result, ssl.SSLContext)
|
|
assert result.cert_store_stats()["x509_ca"] > 0
|
|
|
|
def test_verify_false_wins_over_ca_bundle(self, tmp_path):
|
|
"""False is the explicit-opt-out and must override a stale bundle path."""
|
|
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(
|
|
database_verify_ssl=False,
|
|
database_ca_bundle=str(ca_file),
|
|
),
|
|
):
|
|
assert get_database_ssl() is False
|