diff --git a/docker-compose.yml b/docker-compose.yml index cddd583a..1d7077a3 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -352,6 +352,29 @@ services: profiles: - qdrant + # Centralized Postgres backend for token storage. Used by: + # 1. Integration tests gated on @pytest.mark.postgres — they read + # TEST_DATABASE_URL=postgresql+asyncpg://mcp:mcp@localhost:5433/mcp. + # 2. Manual smoke testing of the HA (multi-replica) deployment story. + # Bring up with: docker compose --profile postgres up -d postgres-test + # See ADR-026 for the pluggable-database-backend design. + postgres-test: + image: docker.io/library/postgres:16-alpine@sha256:16bc17c64a573ef34162af9298258d1aec548232985b33ed7b1eac33ba35c229 + restart: unless-stopped + environment: + POSTGRES_USER: mcp + POSTGRES_PASSWORD: mcp + POSTGRES_DB: mcp + ports: + - 127.0.0.1:5433:5432 + healthcheck: + test: ["CMD-SHELL", "pg_isready -U mcp -d mcp"] + interval: 5s + timeout: 3s + retries: 10 + profiles: + - postgres + volumes: nextcloud: db: diff --git a/docs/ADR-026-pluggable-database-backend.md b/docs/ADR-026-pluggable-database-backend.md new file mode 100644 index 00000000..3e39b76c --- /dev/null +++ b/docs/ADR-026-pluggable-database-backend.md @@ -0,0 +1,169 @@ +# ADR-026: Pluggable database backend (DATABASE_URL) + +## Status + +Accepted — 2026-05-16 + +## Context + +`RefreshTokenStorage` (in `nextcloud_mcp_server/auth/storage.py`) holds all +of the MCP server's persistent state: refresh tokens, OAuth client +credentials, OAuth sessions, browser sessions, app passwords, login-flow +sessions, audit logs, and webhook registrations. Until this ADR it was +backed by a single SQLite file, with the path configured by +`TOKEN_STORAGE_DB`. + +This works well for single-user deployments but blocks horizontal scaling +in Kubernetes: + +- Every pod needs its own PVC (ReadWriteOnce) or a ReadWriteMany volume. +- Tokens stored on pod A are invisible to pod B, so a Service can only + route traffic to one pod at a time. +- Restart / re-deploy cycles either drop the volume (token loss) or + require coordinated PVC handling. +- Backup, encryption-at-rest, and multi-region replication become + per-pod concerns rather than centrally managed DB concerns. + +We needed a way for pods to be stateless and share a centralized store +without giving up the zero-config SQLite path that single-user installs and +local development rely on. + +## Decision + +Introduce a `DATABASE_URL` setting that accepts any SQLAlchemy async URL, +with `sqlite+aiosqlite:///...` remaining the default. The runtime keeps a +single linear migration history and a single `RefreshTokenStorage` class — +the backend is selected purely by the URL. + +### Resolution order + +`get_database_url()` (in `nextcloud_mcp_server/config.py`) returns: + +1. `DATABASE_URL` if set — wins over everything. +2. Otherwise `sqlite+aiosqlite:///{get_token_db_path()}`, so the legacy + `TOKEN_STORAGE_DB` env var and the process-local ephemeral tempfile + fallback both keep working unchanged. + +### Why SQLAlchemy Core + async engine, not an ABC with parallel drivers + +Two alternatives were considered: + +| Option | Why rejected | +|---|---| +| Define a `Storage` ABC with `SQLiteStorage` (aiosqlite) and `PostgresStorage` (asyncpg) implementations | Doubles the surface area — every schema change has to land in two backends, with two sets of migrations, two SQL dialects, two upsert idioms. Diverges over time. | +| Switch to a full SQLAlchemy ORM (declarative models) | Larger refactor; the existing explicit-SQL style is intentional and well-understood by reviewers. | +| **Keep `RefreshTokenStorage` and put SQLAlchemy Core under it** *(chosen)* | One method body per operation, one migration history (Alembic is already SQLAlchemy-based). The URL drives dialect, pool, and DDL. | + +A thin compatibility shim (`_DBConn` / `_Cursor` / `_Row` in `storage.py`) +adapts the `async with aiosqlite.connect(...) as db: async with +db.execute(...) as cursor: ...` idiom to SQLAlchemy `AsyncEngine` / +`AsyncConnection`. Existing method bodies needed only their connection +context-manager swapped; `?` placeholders are rewritten to named binds on +the fly. The seven `INSERT OR REPLACE` statements were rewritten as +portable `INSERT ... ON CONFLICT (...) DO UPDATE` (SQLite ≥ 3.24, Postgres +≥ 9.5; we already require SQLite ≥ 3.35 elsewhere). + +### Encryption stays in Python (Fernet), not the DB + +The DB only ever sees ciphertext for sensitive columns +(`encrypted_token`, `encrypted_client_secret`, `encrypted_password`, +`encrypted_poll_token`). The Fernet key remains a `TOKEN_ENCRYPTION_KEY` +env var, applied in Python before INSERT and after SELECT. This means: + +- Switching backends does not invalidate or re-key existing data. +- Postgres-level features like `pgcrypto` are not required. +- Operators rotating the encryption key still go through the existing + Python path. + +### DDL portability + +All Alembic migrations were rewritten from raw `op.execute("CREATE TABLE +...")` strings to `op.create_table()` / `op.create_index()` calls with +SQLAlchemy types. Notable choices: + +- All `*_at` / expiration / timestamp columns use `sa.BigInteger` — + Postgres `INTEGER` is 32-bit and unix epochs are already past that range. + SQLite treats `BIGINT` and `INTEGER` identically (dynamic typing) so + this is backwards compatible. +- `BLOB` → `sa.LargeBinary` (becomes `BYTEA` on Postgres). +- `BOOLEAN DEFAULT FALSE` → `sa.Boolean, server_default=sa.false()`. +- Existing SQLite deployments are at revision `006` and skip the + rewritten migrations entirely — content rewrites are safe. + +### No data migration, no shipped Postgres + +Two scope decisions worth recording: + +1. **Clean cutover, no SQLite → Postgres data migration tool.** Tokens + are reissued on the next login; webhooks re-register on the next sync + tick. Acceptable because the ephemeral-default already implies this, + and the data being preserved (audit logs, OAuth sessions) is either + short-lived or reconstructible. +2. **Bring-your-own database.** The MCP server consumes a + `DATABASE_URL`; it does not provision Postgres itself. Operators use + CNPG, RDS, the project's existing Helm chart with a sub-chart, etc. + The `postgres-test` service in `docker-compose.yml` exists only for + integration tests and manual HA smoke testing — it is gated on the + `postgres` profile and is not the recommended production pattern. + +### CLI changes + +The `nextcloud-mcp-server db {upgrade,downgrade,current,history}` commands +gain a `--database-url / -u` flag (env `DATABASE_URL`) alongside the +existing `--database-path / -d` (env `TOKEN_STORAGE_DB`). `-u` wins over +`-d`; both fall back to `get_database_url()`. + +## Consequences + +### Positive + +- MCP server pods become stateless. A Kubernetes Deployment can run with + `replicas: 3` behind a Service, with all pods pointed at the same + Postgres URL — tokens written by pod A are immediately visible to pod B. +- Centralized DB operations (backup, restore, replication, encryption at + rest, monitoring) are handled by the operator's existing Postgres + infrastructure rather than duplicated per-pod. +- No regression for single-user / local-development / docker-compose + installs — the SQLite tempfile path is unchanged and remains the default + when `DATABASE_URL` is unset. +- Test coverage doubles automatically: every test that uses the + `temp_storage` fixture now runs against both SQLite and Postgres when + `TEST_DATABASE_URL` is exported. + +### Negative + +- One more thing operators have to think about for HA deployments + (Postgres connection string, credentials secret, network policies). +- Adds SQLAlchemy + asyncpg to the runtime dependency set. SQLAlchemy was + already transitively present via Alembic; asyncpg is genuinely new. +- The compatibility shim in `storage.py` is a small piece of bespoke code + that future contributors need to understand. The alternative — rewriting + every method body to SQLAlchemy idioms — was rejected as too risky for + this PR but might be revisited. + +### Neutral + +- The Alembic migration history was content-rewritten but its revision + graph is unchanged (still `001 → 006`), so existing SQLite deployments + do not re-run anything. +- `TOKEN_STORAGE_DB` still works exactly as before; deployments that + already set it require no changes. + +## Related + +- [ADR-022 Login Flow v2](ADR-022-deployment-mode-consolidation.md) — + defines the per-user app password storage that this ADR centralizes. +- [ADR-002 Vector sync authentication](ADR-002-vector-sync-authentication.md) + — explains the offline-access tokens that benefit most from HA storage. + +## Verification + +1. `uv run pytest tests/unit/` — SQLite path unchanged (1012 tests). +2. `docker compose --profile postgres up -d postgres-test` then + `TEST_DATABASE_URL=postgresql+asyncpg://mcp:mcp@localhost:5433/mcp uv run pytest tests/unit/test_app_password_storage.py tests/unit/test_webhook_storage.py` + — every test runs once per backend. +3. Manual end-to-end smoke against `mcp-login-flow` with a Postgres URL + (commands in `/home/chris/.claude/plans/spicy-enchanting-flurry.md` → + Verification). +4. k8s HA validation (after merge in `homelab-argocd`): `replicas: 3`, + confirm session continuity through the Service. diff --git a/docs/configuration.md b/docs/configuration.md index 63577c9c..0beb1150 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -118,6 +118,48 @@ See [Login Flow v2](login-flow-v2.md) for full setup, scope reference, and troub --- +## Centralized Token Storage (DATABASE_URL, Optional) + +By default the MCP server stores tokens / sessions / app passwords in a +local SQLite file (`TOKEN_STORAGE_DB`, falling back to a per-process +tempfile). For HA Kubernetes deployments where you need multiple +stateless pods to share state, point the server at a centralized +database via `DATABASE_URL`. + +```env +# Centralized Postgres backend (HA k8s deployments) +DATABASE_URL=postgresql+asyncpg://mcp:secret@postgres.svc.cluster.local:5432/mcp +TOKEN_ENCRYPTION_KEY= +``` + +| Variable | Required | Description | +|----------|----------|-------------| +| `DATABASE_URL` | Optional | SQLAlchemy async URL for any supported backend. When set, wins over `TOKEN_STORAGE_DB`. Primary supported targets: `postgresql+asyncpg://...` (recommended for HA) and `sqlite+aiosqlite:///...` (development). | +| `TOKEN_STORAGE_DB` | Optional | Legacy SQLite-only path. Used when `DATABASE_URL` is unset. Falls back to a per-process ephemeral tempfile when both are unset. | + +Notes: + +- **Bring-your-own DB.** The MCP server doesn't provision the database; + it just consumes the URL. Use CNPG, RDS, your existing Helm chart's + Postgres sub-chart, etc. +- **Encryption stays in the app.** `TOKEN_ENCRYPTION_KEY` (Fernet) is + applied in Python; the database only ever sees ciphertext for + sensitive columns. You don't need `pgcrypto`. +- **Schema is managed automatically.** On startup the server runs + Alembic migrations against the configured backend. Existing SQLite + deployments are stamped at the current revision and skip re-execution. +- **No data migration tool.** Moving from SQLite to Postgres is a clean + cutover — tokens are reissued on the next login, webhooks + re-register on the next sync tick. +- **Testing a Postgres backend locally:** `docker compose --profile + postgres up -d postgres-test` then export + `DATABASE_URL=postgresql+asyncpg://mcp:mcp@localhost:5433/mcp`. + +See [ADR-026 Pluggable database backend](ADR-026-pluggable-database-backend.md) +for the architecture rationale. + +--- + ## 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. diff --git a/nextcloud_mcp_server/alembic/versions/20251217_2200_001_initial_schema.py b/nextcloud_mcp_server/alembic/versions/20251217_2200_001_initial_schema.py index 3f1110c0..36eed2c5 100644 --- a/nextcloud_mcp_server/alembic/versions/20251217_2200_001_initial_schema.py +++ b/nextcloud_mcp_server/alembic/versions/20251217_2200_001_initial_schema.py @@ -8,12 +8,19 @@ This migration creates the initial database schema including: - registered_webhooks: Webhook registration tracking (both OAuth and BasicAuth) - schema_version: Legacy schema version tracking (deprecated, use alembic_version) +Uses Alembic's portable schema-DDL helpers (``op.create_table`` / +``op.create_index``) with SQLAlchemy types so the DDL is emitted correctly +for both SQLite (BLOB / INTEGER PRIMARY KEY AUTOINCREMENT) and Postgres +(BYTEA / SERIAL). See ADR-026. + Revision ID: 001 Revises: Create Date: 2025-12-17 22:00:00.000000 """ +import sqlalchemy as sa + from alembic import op # revision identifiers, used by Alembic. @@ -24,143 +31,104 @@ depends_on = None def upgrade() -> None: - """Create initial database schema.""" + """Create initial database schema. - # Refresh tokens table (OAuth mode only, for background jobs) - op.execute( - """ - CREATE TABLE IF NOT EXISTS refresh_tokens ( - user_id TEXT PRIMARY KEY, - encrypted_token BLOB NOT NULL, - expires_at INTEGER, - created_at INTEGER NOT NULL, - updated_at INTEGER NOT NULL, - -- ADR-004 Progressive Consent fields - flow_type TEXT DEFAULT 'hybrid', - token_audience TEXT DEFAULT 'nextcloud', - provisioned_at INTEGER, - provisioning_client_id TEXT, - scopes TEXT, - -- Browser session profile cache - user_profile TEXT, - profile_cached_at INTEGER - ) - """ + All ``*_at`` / expiration / timestamp columns use :class:`sa.BigInteger` + so Postgres allocates BIGINT (8-byte) and unix epoch values don't + overflow the 32-bit int32 INTEGER range on long-lived sessions. SQLite + treats BIGINT and INTEGER identically (dynamic typing), so this is + backwards compatible. + """ + + op.create_table( + "refresh_tokens", + sa.Column("user_id", sa.Text, primary_key=True), + sa.Column("encrypted_token", sa.LargeBinary, nullable=False), + sa.Column("expires_at", sa.BigInteger), + sa.Column("created_at", sa.BigInteger, nullable=False), + sa.Column("updated_at", sa.BigInteger, nullable=False), + # ADR-004 Progressive Consent fields + sa.Column("flow_type", sa.Text, server_default="hybrid"), + sa.Column("token_audience", sa.Text, server_default="nextcloud"), + sa.Column("provisioned_at", sa.BigInteger), + sa.Column("provisioning_client_id", sa.Text), + sa.Column("scopes", sa.Text), + # Browser session profile cache + sa.Column("user_profile", sa.Text), + sa.Column("profile_cached_at", sa.BigInteger), ) - # Audit logs table (both OAuth and BasicAuth modes) - op.execute( - """ - CREATE TABLE IF NOT EXISTS audit_logs ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - timestamp INTEGER NOT NULL, - event TEXT NOT NULL, - user_id TEXT NOT NULL, - resource_type TEXT, - resource_id TEXT, - auth_method TEXT, - hostname TEXT - ) - """ + op.create_table( + "audit_logs", + sa.Column("id", sa.Integer, primary_key=True, autoincrement=True), + sa.Column("timestamp", sa.BigInteger, nullable=False), + sa.Column("event", sa.Text, nullable=False), + sa.Column("user_id", sa.Text, nullable=False), + sa.Column("resource_type", sa.Text), + sa.Column("resource_id", sa.Text), + sa.Column("auth_method", sa.Text), + sa.Column("hostname", sa.Text), + ) + op.create_index("idx_audit_user_timestamp", "audit_logs", ["user_id", "timestamp"]) + + op.create_table( + "oauth_clients", + sa.Column("id", sa.Integer, primary_key=True, autoincrement=False), + sa.Column("client_id", sa.Text, nullable=False, unique=True), + sa.Column("encrypted_client_secret", sa.LargeBinary, nullable=False), + sa.Column("client_id_issued_at", sa.BigInteger, nullable=False), + sa.Column("client_secret_expires_at", sa.BigInteger, nullable=False), + sa.Column("redirect_uris", sa.Text, nullable=False), + sa.Column("encrypted_registration_access_token", sa.LargeBinary), + sa.Column("registration_client_uri", sa.Text), + sa.Column("created_at", sa.BigInteger, nullable=False), + sa.Column("updated_at", sa.BigInteger, nullable=False), ) - # Index on audit logs for efficient queries - op.execute( - """ - CREATE INDEX IF NOT EXISTS idx_audit_user_timestamp - ON audit_logs(user_id, timestamp) - """ + op.create_table( + "oauth_sessions", + sa.Column("session_id", sa.Text, primary_key=True), + sa.Column("client_id", sa.Text), + sa.Column("client_redirect_uri", sa.Text, nullable=False), + sa.Column("state", sa.Text), + sa.Column("code_challenge", sa.Text), + sa.Column("code_challenge_method", sa.Text), + sa.Column("mcp_authorization_code", sa.Text, unique=True), + sa.Column("idp_access_token", sa.Text), + sa.Column("idp_refresh_token", sa.Text), + sa.Column("user_id", sa.Text), + sa.Column("created_at", sa.BigInteger, nullable=False), + sa.Column("expires_at", sa.BigInteger, nullable=False), + # ADR-004 Progressive Consent fields + sa.Column("flow_type", sa.Text, server_default="hybrid"), + sa.Column("requested_scopes", sa.Text), + sa.Column("granted_scopes", sa.Text), + sa.Column("is_provisioning", sa.Boolean, server_default=sa.false()), + ) + op.create_index( + "idx_oauth_sessions_mcp_code", + "oauth_sessions", + ["mcp_authorization_code"], ) - # OAuth client credentials storage (OAuth mode only) - op.execute( - """ - CREATE TABLE IF NOT EXISTS oauth_clients ( - id INTEGER PRIMARY KEY, - client_id TEXT UNIQUE NOT NULL, - encrypted_client_secret BLOB NOT NULL, - client_id_issued_at INTEGER NOT NULL, - client_secret_expires_at INTEGER NOT NULL, - redirect_uris TEXT NOT NULL, - encrypted_registration_access_token BLOB, - registration_client_uri TEXT, - created_at INTEGER NOT NULL, - updated_at INTEGER NOT NULL - ) - """ + # Legacy schema-version table; superseded by alembic_version. Retained + # so pre-Alembic databases that get stamped into the migration chain + # still match the schema fingerprint they had on disk. + op.create_table( + "schema_version", + sa.Column("version", sa.Integer, primary_key=True, autoincrement=False), + sa.Column("applied_at", sa.Float, nullable=False), ) - # OAuth flow sessions (ADR-004 Progressive Consent) - op.execute( - """ - CREATE TABLE IF NOT EXISTS oauth_sessions ( - session_id TEXT PRIMARY KEY, - client_id TEXT, - client_redirect_uri TEXT NOT NULL, - state TEXT, - code_challenge TEXT, - code_challenge_method TEXT, - mcp_authorization_code TEXT UNIQUE, - idp_access_token TEXT, - idp_refresh_token TEXT, - user_id TEXT, - created_at INTEGER NOT NULL, - expires_at INTEGER NOT NULL, - -- ADR-004 Progressive Consent fields - flow_type TEXT DEFAULT 'hybrid', - requested_scopes TEXT, - granted_scopes TEXT, - is_provisioning BOOLEAN DEFAULT FALSE - ) - """ - ) - - # Index for MCP authorization code lookups - op.execute( - """ - CREATE INDEX IF NOT EXISTS idx_oauth_sessions_mcp_code - ON oauth_sessions(mcp_authorization_code) - """ - ) - - # Legacy schema version tracking table - # NOTE: This is deprecated in favor of Alembic's alembic_version table - # Kept for backward compatibility with pre-Alembic databases - op.execute( - """ - CREATE TABLE IF NOT EXISTS schema_version ( - version INTEGER PRIMARY KEY, - applied_at REAL NOT NULL - ) - """ - ) - - # Registered webhooks tracking (both BasicAuth and OAuth modes) - op.execute( - """ - CREATE TABLE IF NOT EXISTS registered_webhooks ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - webhook_id INTEGER NOT NULL UNIQUE, - preset_id TEXT NOT NULL, - created_at REAL NOT NULL - ) - """ - ) - - # Indexes for efficient webhook queries - op.execute( - """ - CREATE INDEX IF NOT EXISTS idx_webhooks_preset - ON registered_webhooks(preset_id) - """ - ) - - op.execute( - """ - CREATE INDEX IF NOT EXISTS idx_webhooks_created - ON registered_webhooks(created_at) - """ + op.create_table( + "registered_webhooks", + sa.Column("id", sa.Integer, primary_key=True, autoincrement=True), + sa.Column("webhook_id", sa.Integer, nullable=False, unique=True), + sa.Column("preset_id", sa.Text, nullable=False), + sa.Column("created_at", sa.Float, nullable=False), ) + op.create_index("idx_webhooks_preset", "registered_webhooks", ["preset_id"]) + op.create_index("idx_webhooks_created", "registered_webhooks", ["created_at"]) def downgrade() -> None: @@ -170,16 +138,13 @@ def downgrade() -> None: Use with extreme caution. """ - # Drop indexes first - op.execute("DROP INDEX IF EXISTS idx_webhooks_created") - op.execute("DROP INDEX IF EXISTS idx_webhooks_preset") - op.execute("DROP INDEX IF EXISTS idx_oauth_sessions_mcp_code") - op.execute("DROP INDEX IF EXISTS idx_audit_user_timestamp") - - # Drop tables - op.execute("DROP TABLE IF EXISTS registered_webhooks") - op.execute("DROP TABLE IF EXISTS schema_version") - op.execute("DROP TABLE IF EXISTS oauth_sessions") - op.execute("DROP TABLE IF EXISTS oauth_clients") - op.execute("DROP TABLE IF EXISTS audit_logs") - op.execute("DROP TABLE IF EXISTS refresh_tokens") + op.drop_index("idx_webhooks_created", table_name="registered_webhooks") + op.drop_index("idx_webhooks_preset", table_name="registered_webhooks") + op.drop_table("registered_webhooks") + op.drop_table("schema_version") + op.drop_index("idx_oauth_sessions_mcp_code", table_name="oauth_sessions") + op.drop_table("oauth_sessions") + op.drop_table("oauth_clients") + op.drop_index("idx_audit_user_timestamp", table_name="audit_logs") + op.drop_table("audit_logs") + op.drop_table("refresh_tokens") diff --git a/nextcloud_mcp_server/alembic/versions/20260113_1200_002_add_app_passwords.py b/nextcloud_mcp_server/alembic/versions/20260113_1200_002_add_app_passwords.py index 9e7b36c0..25aa17f4 100644 --- a/nextcloud_mcp_server/alembic/versions/20260113_1200_002_add_app_passwords.py +++ b/nextcloud_mcp_server/alembic/versions/20260113_1200_002_add_app_passwords.py @@ -10,6 +10,8 @@ Create Date: 2026-01-13 12:00:00.000000 """ +import sqlalchemy as sa + from alembic import op # revision identifiers, used by Alembic. @@ -22,29 +24,19 @@ depends_on = None def upgrade() -> None: """Add app_passwords table for multi-user BasicAuth mode.""" - # App passwords table for multi-user BasicAuth background sync - op.execute( - """ - CREATE TABLE IF NOT EXISTS app_passwords ( - user_id TEXT PRIMARY KEY, - encrypted_password BLOB NOT NULL, - created_at INTEGER NOT NULL, - updated_at INTEGER NOT NULL - ) - """ - ) - - # Index for efficient user lookups - op.execute( - """ - CREATE INDEX IF NOT EXISTS idx_app_passwords_updated - ON app_passwords(updated_at) - """ + op.create_table( + "app_passwords", + sa.Column("user_id", sa.Text, primary_key=True), + sa.Column("encrypted_password", sa.LargeBinary, nullable=False), + # BigInteger to keep unix epochs in range on Postgres (see 001). + sa.Column("created_at", sa.BigInteger, nullable=False), + sa.Column("updated_at", sa.BigInteger, nullable=False), ) + op.create_index("idx_app_passwords_updated", "app_passwords", ["updated_at"]) def downgrade() -> None: """Drop app_passwords table.""" - op.execute("DROP INDEX IF EXISTS idx_app_passwords_updated") - op.execute("DROP TABLE IF EXISTS app_passwords") + op.drop_index("idx_app_passwords_updated", table_name="app_passwords") + op.drop_table("app_passwords") diff --git a/nextcloud_mcp_server/alembic/versions/20260227_1200_003_add_scopes_and_login_flow.py b/nextcloud_mcp_server/alembic/versions/20260227_1200_003_add_scopes_and_login_flow.py index 5f14e678..1f049155 100644 --- a/nextcloud_mcp_server/alembic/versions/20260227_1200_003_add_scopes_and_login_flow.py +++ b/nextcloud_mcp_server/alembic/versions/20260227_1200_003_add_scopes_and_login_flow.py @@ -12,6 +12,8 @@ Create Date: 2026-02-27 12:00:00.000000 """ +import sqlalchemy as sa + from alembic import op # revision identifiers, used by Alembic. @@ -24,72 +26,37 @@ depends_on = None def upgrade() -> None: """Add scopes/username to app_passwords and create login_flow_sessions.""" - # Add scopes column (nullable JSON array, NULL = all scopes allowed) - op.execute( - """ - ALTER TABLE app_passwords ADD COLUMN scopes TEXT - """ - ) + # Nullable scope columns on the existing app_passwords table. + op.add_column("app_passwords", sa.Column("scopes", sa.Text)) + op.add_column("app_passwords", sa.Column("username", sa.Text)) - # Add username column (Nextcloud loginName from Login Flow v2) - op.execute( - """ - ALTER TABLE app_passwords ADD COLUMN username TEXT - """ + op.create_table( + "login_flow_sessions", + sa.Column("user_id", sa.Text, primary_key=True), + sa.Column("encrypted_poll_token", sa.LargeBinary, nullable=False), + sa.Column("poll_endpoint", sa.Text, nullable=False), + sa.Column("requested_scopes", sa.Text), + # BigInteger to keep unix epochs in range on Postgres (see 001). + sa.Column("created_at", sa.BigInteger, nullable=False), + sa.Column("expires_at", sa.BigInteger, nullable=False), ) - - # Login Flow v2 session tracking - op.execute( - """ - CREATE TABLE IF NOT EXISTS login_flow_sessions ( - user_id TEXT PRIMARY KEY, - encrypted_poll_token BLOB NOT NULL, - poll_endpoint TEXT NOT NULL, - requested_scopes TEXT, - created_at INTEGER NOT NULL, - expires_at INTEGER NOT NULL - ) - """ - ) - - # Index for efficient cleanup of expired sessions - op.execute( - """ - CREATE INDEX IF NOT EXISTS idx_login_flow_sessions_expires - ON login_flow_sessions(expires_at) - """ + op.create_index( + "idx_login_flow_sessions_expires", + "login_flow_sessions", + ["expires_at"], ) def downgrade() -> None: - """Drop login_flow_sessions and remove added columns.""" + """Drop login_flow_sessions and remove added columns. - op.execute("DROP INDEX IF EXISTS idx_login_flow_sessions_expires") - op.execute("DROP TABLE IF EXISTS login_flow_sessions") + ``batch_alter_table`` handles SQLite's pre-3.35 lack of ``DROP COLUMN`` + by recreating the table; on Postgres it issues a native ``DROP COLUMN``. + """ - # SQLite doesn't support DROP COLUMN before 3.35.0 - # Recreate app_passwords without the new columns - op.execute( - """ - CREATE TABLE app_passwords_backup ( - user_id TEXT PRIMARY KEY, - encrypted_password BLOB NOT NULL, - created_at INTEGER NOT NULL, - updated_at INTEGER NOT NULL - ) - """ - ) - op.execute( - """ - INSERT INTO app_passwords_backup (user_id, encrypted_password, created_at, updated_at) - SELECT user_id, encrypted_password, created_at, updated_at FROM app_passwords - """ - ) - op.execute("DROP TABLE app_passwords") - op.execute("ALTER TABLE app_passwords_backup RENAME TO app_passwords") - op.execute( - """ - CREATE INDEX IF NOT EXISTS idx_app_passwords_updated - ON app_passwords(updated_at) - """ - ) + op.drop_index("idx_login_flow_sessions_expires", table_name="login_flow_sessions") + op.drop_table("login_flow_sessions") + + with op.batch_alter_table("app_passwords") as batch_op: + batch_op.drop_column("username") + batch_op.drop_column("scopes") diff --git a/nextcloud_mcp_server/alembic/versions/20260502_1500_005_add_browser_sessions.py b/nextcloud_mcp_server/alembic/versions/20260502_1500_005_add_browser_sessions.py index b85f20d1..a8aaea7d 100644 --- a/nextcloud_mcp_server/alembic/versions/20260502_1500_005_add_browser_sessions.py +++ b/nextcloud_mcp_server/alembic/versions/20260502_1500_005_add_browser_sessions.py @@ -10,6 +10,8 @@ Revises: 004 Create Date: 2026-05-02 15:00:00.000000 """ +import sqlalchemy as sa + from alembic import op revision = "005" @@ -19,31 +21,19 @@ depends_on = None def upgrade() -> None: - op.execute( - """ - CREATE TABLE IF NOT EXISTS browser_sessions ( - session_id TEXT PRIMARY KEY, - user_id TEXT NOT NULL, - created_at INTEGER NOT NULL, - expires_at INTEGER NOT NULL - ) - """ - ) - op.execute( - """ - CREATE INDEX IF NOT EXISTS idx_browser_sessions_user - ON browser_sessions(user_id) - """ - ) - op.execute( - """ - CREATE INDEX IF NOT EXISTS idx_browser_sessions_expires - ON browser_sessions(expires_at) - """ + op.create_table( + "browser_sessions", + sa.Column("session_id", sa.Text, primary_key=True), + sa.Column("user_id", sa.Text, nullable=False), + # BigInteger to keep unix epochs in range on Postgres (see 001). + sa.Column("created_at", sa.BigInteger, nullable=False), + sa.Column("expires_at", sa.BigInteger, nullable=False), ) + op.create_index("idx_browser_sessions_user", "browser_sessions", ["user_id"]) + op.create_index("idx_browser_sessions_expires", "browser_sessions", ["expires_at"]) def downgrade() -> None: - op.execute("DROP INDEX IF EXISTS idx_browser_sessions_expires") - op.execute("DROP INDEX IF EXISTS idx_browser_sessions_user") - op.execute("DROP TABLE IF EXISTS browser_sessions") + op.drop_index("idx_browser_sessions_expires", table_name="browser_sessions") + op.drop_index("idx_browser_sessions_user", table_name="browser_sessions") + op.drop_table("browser_sessions") diff --git a/nextcloud_mcp_server/auth/storage.py b/nextcloud_mcp_server/auth/storage.py index 4090aa25..c31d9efe 100644 --- a/nextcloud_mcp_server/auth/storage.py +++ b/nextcloud_mcp_server/auth/storage.py @@ -31,22 +31,232 @@ import os import socket import sqlite3 import time +from contextlib import asynccontextmanager from pathlib import Path from typing import Any -import aiosqlite import anyio import httpx +import sqlalchemy as sa from anyio import to_thread from cryptography.fernet import Fernet +from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, create_async_engine +from sqlalchemy.pool import NullPool -from nextcloud_mcp_server.config import get_token_db_path, is_ephemeral_token_db +from nextcloud_mcp_server.config import ( + get_database_url, + is_ephemeral_token_db, + is_sqlite_url, +) from nextcloud_mcp_server.migrations import stamp_database, upgrade_database from nextcloud_mcp_server.observability.metrics import record_db_operation logger = logging.getLogger(__name__) +def _qmark_to_named(sql: str) -> tuple[str, list[str]]: + """Rewrite ``?`` positional placeholders to ``:p0, :p1, ...`` named binds. + + SQLAlchemy's :func:`text` only supports named bind parameters, so the + aiosqlite-style call sites (which use ``?``) are translated as they + cross the shim. The rewriter preserves ``?`` characters inside SQL + string literals; comments are not currently respected but the storage + layer doesn't put ``?`` inside comments. + """ + out: list[str] = [] + names: list[str] = [] + i = 0 + in_str = False + quote = "" + n = 0 + while i < len(sql): + ch = sql[i] + if in_str: + out.append(ch) + if ch == quote: + # SQL string escapes ('' or "") — stay in string mode. + if i + 1 < len(sql) and sql[i + 1] == quote: + out.append(quote) + i += 2 + continue + in_str = False + i += 1 + continue + if ch in ("'", '"'): + in_str = True + quote = ch + out.append(ch) + i += 1 + continue + if ch == "?": + name = f"p{n}" + out.append(f":{name}") + names.append(name) + n += 1 + i += 1 + continue + out.append(ch) + i += 1 + return "".join(out), names + + +class _Row: + """Hybrid tuple/dict row mirroring ``aiosqlite.Row`` semantics. + + The legacy SQLite-direct call sites use a mix of access patterns: + positional unpacking (``a, b, c = row``), indexed access (``row[0]``), + and dict-like access (``row["col"]``, ``dict(row)``) when + ``db.row_factory = aiosqlite.Row`` is set. To avoid touching every call + site, every row returned by :class:`_Cursor` is wrapped in this hybrid + object so all three patterns keep working. + """ + + __slots__ = ("_values", "_mapping") + + def __init__(self, values: tuple, mapping: dict) -> None: + self._values = values + self._mapping = mapping + + def __getitem__(self, key): + if isinstance(key, (int, slice)): + return self._values[key] + return self._mapping[key] + + def __iter__(self): + return iter(self._values) + + def __len__(self) -> int: + return len(self._values) + + def keys(self): + return self._mapping.keys() + + def values(self): + return self._mapping.values() + + def items(self): + return self._mapping.items() + + +def _wrap_row(row) -> _Row | None: + if row is None: + return None + return _Row(tuple(row), dict(row._mapping)) + + +def _wrap_rows(rows) -> list[_Row]: + """Wrap a list of SQLAlchemy rows; iterator never yields ``None``.""" + return [_Row(tuple(r), dict(r._mapping)) for r in rows] + + +class _Cursor: + """aiosqlite-compatible cursor view over a SQLAlchemy CursorResult. + + Existing storage methods iterate cursors via ``async with db.execute(...) + as cursor: row = await cursor.fetchone()``. SQLAlchemy returns a + synchronous :class:`Result` from an async ``execute``; this shim adds the + async context-manager / async-fetch surface so call sites are unchanged. + + ``rowcount`` is captured eagerly at construction time. ``lastrowid`` is + intentionally NOT exposed: accessing ``CursorResult.lastrowid`` on the + asyncpg dialect consumes the result buffer, which would silently turn + every subsequent ``fetchall()`` into an empty list (a real bug hit + during the Postgres port). + """ + + __slots__ = ("_result", "rowcount") + + def __init__(self, result: sa.CursorResult) -> None: + self._result = result + # ``rowcount`` is -1 for SELECTs in SQLAlchemy; existing code only + # reads it after writes (DELETE/UPDATE) where it is accurate. + self.rowcount = result.rowcount + + async def fetchone(self) -> _Row | None: + return _wrap_row(self._result.fetchone()) + + async def fetchall(self) -> list[_Row]: + return _wrap_rows(self._result.fetchall()) + + async def __aenter__(self) -> "_Cursor": + return self + + async def __aexit__(self, *exc: object) -> None: + # SQLAlchemy Result closes when the connection closes; no-op here. + return None + + +class _ExecuteCtx: + """Hybrid awaitable + async context manager for ``db.execute(...)``. + + Aiosqlite call sites use both forms interchangeably:: + + cursor = await db.execute(sql, params) + async with db.execute(sql, params) as cursor: ... + + so the return value must be awaitable (resolves to a cursor) AND a + one-shot async context manager (executes on ``__aenter__`` and returns + the cursor). This wrapper provides both surfaces without executing the + SQL twice — the cursor is cached after the first resolution. + """ + + __slots__ = ("_conn", "_sql", "_params", "_cursor") + + def __init__(self, conn: AsyncConnection, sql: str, params: tuple | list) -> None: + self._conn = conn + self._sql = sql + self._params = params + self._cursor: _Cursor | None = None + + async def _resolve(self) -> _Cursor: + if self._cursor is not None: + return self._cursor + text_sql, names = _qmark_to_named(self._sql) + if len(names) != len(self._params): + raise ValueError( + f"Placeholder count mismatch: SQL has {len(names)} '?' " + f"but got {len(self._params)} params" + ) + bind = dict(zip(names, self._params, strict=True)) + result = await self._conn.execute(sa.text(text_sql), bind) + self._cursor = _Cursor(result) + return self._cursor + + def __await__(self): + return self._resolve().__await__() + + async def __aenter__(self) -> _Cursor: + return await self._resolve() + + async def __aexit__(self, *exc: object) -> None: + return None + + +class _DBConn: + """aiosqlite-compatible wrapper around a SQLAlchemy AsyncConnection. + + Provides ``execute`` (with ``?`` placeholders, returning a hybrid + awaitable/context-manager :class:`_ExecuteCtx`) and ``commit`` so the + existing storage method bodies need no churn beyond swapping the + connection context-manager. Wraps results in :class:`_Cursor` for the + fetchone/fetchall/rowcount surface the call sites already use. + + ``row_factory`` is accepted as a setter for source compatibility with + aiosqlite call sites but is ignored: every row is already wrapped in + :class:`_Row` so dict-style access works unconditionally. + """ + + def __init__(self, conn: AsyncConnection) -> None: + self._conn = conn + self.row_factory = None # set by aiosqlite-shaped call sites; ignored + + def execute(self, sql: str, params: tuple | list = ()) -> _ExecuteCtx: + return _ExecuteCtx(self._conn, sql, params) + + async def commit(self) -> None: + await self._conn.commit() + + class RefreshTokenStorage: """Persistent storage for MCP server state (tokens, webhooks, and future features). @@ -65,17 +275,41 @@ class RefreshTokenStorage: Token-related operations require TOKEN_ENCRYPTION_KEY, but webhook operations do not. """ - def __init__(self, db_path: str, encryption_key: bytes | None = None): + def __init__( + self, + database_url: str | None = None, + encryption_key: bytes | None = None, + *, + db_path: str | None = None, + ): """ Initialize persistent storage. Args: - db_path: Path to SQLite database file + database_url: SQLAlchemy URL (``sqlite+aiosqlite:///...`` or + ``postgresql+asyncpg://...``). When omitted, falls back to + :func:`get_database_url` (honors ``DATABASE_URL`` env, then + ``TOKEN_STORAGE_DB``). encryption_key: Optional Fernet encryption key (32 bytes, base64-encoded). - Required for token storage operations, not required for webhook tracking. + Required for token storage operations, not required for webhook tracking. + db_path: Deprecated SQLite-only constructor argument retained for + tests that pass a tempfile path. Internally converted to + ``sqlite+aiosqlite:///{db_path}``. """ - self.db_path = db_path + if database_url is None and db_path is not None: + database_url = f"sqlite+aiosqlite:///{db_path}" + if database_url is None: + database_url = get_database_url() + self.database_url = database_url + # Legacy attribute retained for sqlite-only code paths (file perms, + # ephemeral tempfile detection, log messages). Empty string for + # non-sqlite URLs so accidental file ops fail loudly. + self.db_path = ( + database_url.split("///", 1)[1] if is_sqlite_url(database_url) else "" + ) self.cipher = Fernet(encryption_key) if encryption_key else None + self.engine: AsyncEngine | None = None + self._dialect: str = "unknown" self._initialized = False @classmethod @@ -84,10 +318,14 @@ class RefreshTokenStorage: Create storage instance from environment variables. Environment variables: - TOKEN_STORAGE_DB: Path to database file. If unset, a per-process - tempfile is allocated and deleted at interpreter exit — - tokens are ephemeral and wiped on restart. Set this to a - filesystem path to persist tokens across restarts. + DATABASE_URL: SQLAlchemy URL for any supported backend. Wins + over ``TOKEN_STORAGE_DB`` when set. Use + ``postgresql+asyncpg://user:pw@host/db`` for HA k8s + deployments. See ADR-026. + TOKEN_STORAGE_DB: Legacy SQLite-only path. If unset and + ``DATABASE_URL`` is also unset, a per-process tempfile is + allocated and deleted at interpreter exit — tokens are + ephemeral and wiped on restart. TOKEN_ENCRYPTION_KEY: Optional base64-encoded Fernet key (required for token storage) Returns: @@ -97,13 +335,17 @@ class RefreshTokenStorage: If TOKEN_ENCRYPTION_KEY is not set, token storage operations will fail, but webhook tracking will still work. """ - db_path = get_token_db_path() - if is_ephemeral_token_db(db_path): - logger.info( - "Using ephemeral token storage at %s " - "(set TOKEN_STORAGE_DB to persist tokens across restarts)", - db_path, - ) + database_url = get_database_url() + if is_sqlite_url(database_url): + sqlite_path = database_url.split("///", 1)[1] + if is_ephemeral_token_db(sqlite_path): + logger.info( + "Using ephemeral token storage at %s " + "(set DATABASE_URL or TOKEN_STORAGE_DB to persist tokens across restarts)", + sqlite_path, + ) + else: + logger.info("Using centralized token storage at %s", database_url) encryption_key_b64 = os.getenv("TOKEN_ENCRYPTION_KEY") encryption_key = None @@ -130,7 +372,7 @@ class RefreshTokenStorage: "but webhook tracking will still work" ) - return cls(db_path=db_path, encryption_key=encryption_key) + return cls(database_url=database_url, encryption_key=encryption_key) async def initialize(self) -> None: """ @@ -151,7 +393,9 @@ class RefreshTokenStorage: if self._initialized: return - if sqlite3.sqlite_version_info < (3, 35): + is_sqlite = is_sqlite_url(self.database_url) + + if is_sqlite and sqlite3.sqlite_version_info < (3, 35): raise RuntimeError( "SQLite >= 3.35 is required (DELETE ... RETURNING is used " "by delete_browser_session); detected " @@ -159,63 +403,85 @@ class RefreshTokenStorage: "image with a newer bundled libsqlite3." ) - # Ensure directory exists - db_dir = Path(self.db_path).parent - db_dir.mkdir(parents=True, exist_ok=True) + if is_sqlite: + # File-permission hardening + parent dir creation is sqlite-only; + # centralized backends manage their own filesystem. + db_dir = Path(self.db_path).parent + db_dir.mkdir(parents=True, exist_ok=True) + if Path(self.db_path).exists(): + os.chmod(self.db_path, 0o600) - # Set restrictive permissions on database file if it exists - if Path(self.db_path).exists(): - os.chmod(self.db_path, 0o600) - - # Check database state and run appropriate migration strategy - async with aiosqlite.connect(self.db_path) as db: - # Check if database is managed by Alembic - cursor = await db.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name='alembic_version'" + # Create the shared async engine for the chosen backend. SQLite uses + # NullPool (per-call connections, matches the prior aiosqlite-direct + # behavior); Postgres uses the default pool with pre-ping so dropped + # connections from idle k8s networks are retried transparently. + if is_sqlite: + self.engine = create_async_engine( + self.database_url, + poolclass=NullPool, + connect_args={"check_same_thread": False}, + future=True, ) - has_alembic = await cursor.fetchone() is not None + else: + self.engine = create_async_engine( + self.database_url, + pool_size=10, + max_overflow=20, + pool_pre_ping=True, + future=True, + ) + self._dialect = self.engine.dialect.name - if not has_alembic: - # Check if this is a pre-Alembic database with existing schema - cursor = await db.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name='refresh_tokens'" - ) - has_schema = await cursor.fetchone() is not None + # Check database state with the SQLAlchemy inspector so the legacy + # ``sqlite_master`` lookup works against either backend. + def _inspect(sync_conn: sa.Connection) -> tuple[bool, bool]: + insp = sa.inspect(sync_conn) + tables = set(insp.get_table_names()) + return ("alembic_version" in tables), ("refresh_tokens" in tables) - if has_schema: - logger.info( - "Detected pre-Alembic database at %s, stamping with initial revision", - self.db_path, - ) - else: - logger.info( - "Initializing new database at %s with migrations", self.db_path - ) + async with self.engine.connect() as conn: + has_alembic, has_schema = await conn.run_sync(_inspect) - # Run migrations in a worker thread using anyio.to_thread - # This allows Alembic to run its own async operations in a separate context if not has_alembic: if has_schema: - # Stamp existing database without running migrations - await to_thread.run_sync(stamp_database, self.db_path, "001") + logger.info( + "Detected pre-Alembic database at %s, stamping with initial revision", + self.database_url, + ) + await to_thread.run_sync(stamp_database, self.database_url, "001") logger.info( "Pre-Alembic database stamped successfully. " "Future schema changes will use migrations." ) else: - # New database - run migrations - await to_thread.run_sync(upgrade_database, self.db_path, "head") + logger.info( + "Initializing new database at %s with migrations", + self.database_url, + ) + await to_thread.run_sync(upgrade_database, self.database_url, "head") logger.info("Database initialized with migrations") else: - # Alembic-managed database - upgrade to latest - await to_thread.run_sync(upgrade_database, self.db_path, "head") + await to_thread.run_sync(upgrade_database, self.database_url, "head") logger.info("Database upgraded to latest version") - # Set restrictive permissions after initialization - os.chmod(self.db_path, 0o600) + if is_sqlite: + os.chmod(self.db_path, 0o600) self._initialized = True - logger.info("Initialized refresh token storage at %s", self.db_path) + logger.info("Initialized refresh token storage at %s", self.database_url) + + @asynccontextmanager + async def _db(self): + """Open a backend-agnostic connection. + + Yields a :class:`_DBConn` that mimics aiosqlite's API (``execute`` with + ``?`` placeholders, ``commit``, cursor with ``fetchone`` / + ``fetchall`` / ``rowcount``) so the existing storage method bodies + work against either SQLite or Postgres without per-call rewrites. + """ + assert self.engine is not None, "RefreshTokenStorage.initialize() not called" + async with self.engine.connect() as conn: + yield _DBConn(conn) async def store_refresh_token( self, @@ -260,20 +526,30 @@ class RefreshTokenStorage: start_time = time.time() try: - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: + # ON CONFLICT DO UPDATE preserves ``created_at`` (it's not + # listed in the update clause) so the original + # COALESCE(...)-based INSERT OR REPLACE semantics are kept. await db.execute( """ - INSERT OR REPLACE INTO refresh_tokens + INSERT INTO refresh_tokens (user_id, encrypted_token, expires_at, created_at, updated_at, flow_type, token_audience, provisioned_at, provisioning_client_id, scopes) - VALUES (?, ?, ?, COALESCE((SELECT created_at FROM refresh_tokens WHERE user_id = ?), ?), ?, - ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (user_id) DO UPDATE SET + encrypted_token = EXCLUDED.encrypted_token, + expires_at = EXCLUDED.expires_at, + updated_at = EXCLUDED.updated_at, + flow_type = EXCLUDED.flow_type, + token_audience = EXCLUDED.token_audience, + provisioned_at = EXCLUDED.provisioned_at, + provisioning_client_id = EXCLUDED.provisioning_client_id, + scopes = EXCLUDED.scopes """, ( user_id, encrypted_token, expires_at, - user_id, now, now, flow_type, @@ -285,7 +561,7 @@ class RefreshTokenStorage: ) await db.commit() duration = time.time() - start_time - record_db_operation("sqlite", "insert", duration, "success") + record_db_operation(self._dialect, "insert", duration, "success") logger.info( f"Stored refresh token for user {user_id}" @@ -293,7 +569,7 @@ class RefreshTokenStorage: ) except Exception: duration = time.time() - start_time - record_db_operation("sqlite", "insert", duration, "error") + record_db_operation(self._dialect, "insert", duration, "error") raise # Audit log @@ -322,7 +598,7 @@ class RefreshTokenStorage: profile_json = json.dumps(profile_data) now = int(time.time()) - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: await db.execute( """ UPDATE refresh_tokens @@ -351,7 +627,7 @@ class RefreshTokenStorage: if not self._initialized: await self.initialize() - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: async with db.execute( """ SELECT user_profile, profile_cached_at @@ -407,7 +683,7 @@ class RefreshTokenStorage: start_time = time.time() try: - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: async with db.execute( """ SELECT encrypted_token, expires_at, flow_type, token_audience, @@ -421,7 +697,7 @@ class RefreshTokenStorage: if not row: logger.debug("No refresh token found for user %s", user_id) duration = time.time() - start_time - record_db_operation("sqlite", "select", duration, "success") + record_db_operation(self._dialect, "select", duration, "success") return None ( @@ -443,7 +719,7 @@ class RefreshTokenStorage: ) await self.delete_refresh_token(user_id) duration = time.time() - start_time - record_db_operation("sqlite", "select", duration, "success") + record_db_operation(self._dialect, "select", duration, "success") return None decrypted_token = self.cipher.decrypt(encrypted_token).decode() @@ -456,7 +732,7 @@ class RefreshTokenStorage: ) duration = time.time() - start_time - record_db_operation("sqlite", "select", duration, "success") + record_db_operation(self._dialect, "select", duration, "success") return { "refresh_token": decrypted_token, @@ -470,7 +746,7 @@ class RefreshTokenStorage: } except Exception as e: duration = time.time() - start_time - record_db_operation("sqlite", "select", duration, "error") + record_db_operation(self._dialect, "select", duration, "error") logger.error("Failed to decrypt refresh token for user %s: %s", user_id, e) return None @@ -502,7 +778,7 @@ class RefreshTokenStorage: "TOKEN_ENCRYPTION_KEY is not set — token storage operations unavailable" ) - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: async with db.execute( """ SELECT user_id, encrypted_token, expires_at, flow_type, token_audience, @@ -582,7 +858,7 @@ class RefreshTokenStorage: start_time = time.time() try: - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute( "DELETE FROM refresh_tokens WHERE user_id = ?", (user_id,), @@ -591,7 +867,7 @@ class RefreshTokenStorage: deleted = cursor.rowcount > 0 duration = time.time() - start_time - record_db_operation("sqlite", "delete", duration, "success") + record_db_operation(self._dialect, "delete", duration, "success") if deleted: logger.info("Deleted refresh token for user %s", user_id) @@ -606,7 +882,7 @@ class RefreshTokenStorage: return deleted except Exception: duration = time.time() - start_time - record_db_operation("sqlite", "delete", duration, "error") + record_db_operation(self._dialect, "delete", duration, "error") raise async def get_all_user_ids(self) -> list[str]: @@ -619,7 +895,7 @@ class RefreshTokenStorage: if not self._initialized: await self.initialize() - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: async with db.execute( "SELECT user_id FROM refresh_tokens ORDER BY updated_at DESC" ) as cursor: @@ -641,7 +917,7 @@ class RefreshTokenStorage: now = int(time.time()) - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute( "DELETE FROM refresh_tokens WHERE expires_at IS NOT NULL AND expires_at < ?", (now,), @@ -700,18 +976,25 @@ class RefreshTokenStorage: redirect_uris_json = json.dumps(redirect_uris) now = int(time.time()) - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: + # Singleton row pinned at id=1; ON CONFLICT preserves the + # original ``created_at`` because it's omitted from the update. await db.execute( """ - INSERT OR REPLACE INTO oauth_clients + INSERT INTO oauth_clients (id, client_id, encrypted_client_secret, client_id_issued_at, client_secret_expires_at, redirect_uris, encrypted_registration_access_token, registration_client_uri, created_at, updated_at) - VALUES ( - 1, ?, ?, ?, ?, ?, ?, ?, - COALESCE((SELECT created_at FROM oauth_clients WHERE id = 1), ?), - ? - ) + VALUES (1, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (id) DO UPDATE SET + client_id = EXCLUDED.client_id, + encrypted_client_secret = EXCLUDED.encrypted_client_secret, + client_id_issued_at = EXCLUDED.client_id_issued_at, + client_secret_expires_at = EXCLUDED.client_secret_expires_at, + redirect_uris = EXCLUDED.redirect_uris, + encrypted_registration_access_token = EXCLUDED.encrypted_registration_access_token, + registration_client_uri = EXCLUDED.registration_client_uri, + updated_at = EXCLUDED.updated_at """, ( client_id, @@ -768,7 +1051,7 @@ class RefreshTokenStorage: "TOKEN_ENCRYPTION_KEY is not set — token storage operations unavailable" ) - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: async with db.execute( """ SELECT client_id, encrypted_client_secret, client_id_issued_at, @@ -841,7 +1124,7 @@ class RefreshTokenStorage: if not self._initialized: await self.initialize() - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute("DELETE FROM oauth_clients WHERE id = 1") await db.commit() deleted = cursor.rowcount > 0 @@ -868,7 +1151,7 @@ class RefreshTokenStorage: if not self._initialized: await self.initialize() - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: async with db.execute( "SELECT client_secret_expires_at FROM oauth_clients WHERE id = 1" ) as cursor: @@ -902,7 +1185,7 @@ class RefreshTokenStorage: hostname = socket.gethostname() timestamp = int(time.time()) - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: await db.execute( """ INSERT INTO audit_logs @@ -955,8 +1238,7 @@ class RefreshTokenStorage: query += " ORDER BY timestamp DESC LIMIT ?" params.append(limit) - async with aiosqlite.connect(self.db_path) as db: - db.row_factory = aiosqlite.Row + async with self._db() as db: async with db.execute(query, params) as cursor: rows = await cursor.fetchall() @@ -1001,7 +1283,7 @@ class RefreshTokenStorage: now = int(time.time()) expires_at = now + ttl_seconds - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: await db.execute( """ INSERT INTO oauth_sessions @@ -1042,8 +1324,7 @@ class RefreshTokenStorage: if not self._initialized: await self.initialize() - async with aiosqlite.connect(self.db_path) as db: - db.row_factory = aiosqlite.Row + async with self._db() as db: async with db.execute( "SELECT * FROM oauth_sessions WHERE session_id = ?", (session_id,) ) as cursor: @@ -1074,8 +1355,7 @@ class RefreshTokenStorage: if not self._initialized: await self.initialize() - async with aiosqlite.connect(self.db_path) as db: - db.row_factory = aiosqlite.Row + async with self._db() as db: async with db.execute( "SELECT * FROM oauth_sessions WHERE mcp_authorization_code = ?", (mcp_authorization_code,), @@ -1134,7 +1414,7 @@ class RefreshTokenStorage: params.append(session_id) - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute( f""" UPDATE oauth_sessions @@ -1161,7 +1441,7 @@ class RefreshTokenStorage: if not self._initialized: await self.initialize() - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute( "DELETE FROM oauth_sessions WHERE session_id = ?", (session_id,) ) @@ -1185,7 +1465,7 @@ class RefreshTokenStorage: now = int(time.time()) - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute( "DELETE FROM oauth_sessions WHERE expires_at < ?", (now,) ) @@ -1220,12 +1500,16 @@ class RefreshTokenStorage: now = int(time.time()) expires_at = now + ttl_seconds - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: await db.execute( """ - INSERT OR REPLACE INTO browser_sessions + INSERT INTO browser_sessions (session_id, user_id, created_at, expires_at) VALUES (?, ?, ?, ?) + ON CONFLICT (session_id) DO UPDATE SET + user_id = EXCLUDED.user_id, + created_at = EXCLUDED.created_at, + expires_at = EXCLUDED.expires_at """, (session_id, user_id, now, expires_at), ) @@ -1257,8 +1541,7 @@ class RefreshTokenStorage: if not self._initialized: await self.initialize() - async with aiosqlite.connect(self.db_path) as db: - db.row_factory = aiosqlite.Row + async with self._db() as db: async with db.execute( "SELECT user_id, expires_at FROM browser_sessions WHERE session_id = ?", (session_id,), @@ -1285,7 +1568,7 @@ class RefreshTokenStorage: # concurrent delete that empties the row between SELECT and DELETE # (PR #758 round-3 review). user_id: str | None = None - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: async with db.execute( "DELETE FROM browser_sessions WHERE session_id = ? RETURNING user_id", (session_id,), @@ -1319,7 +1602,7 @@ class RefreshTokenStorage: now = int(time.time()) - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute( "DELETE FROM browser_sessions WHERE expires_at < ?", (now,) ) @@ -1346,9 +1629,15 @@ class RefreshTokenStorage: if not self._initialized: await self.initialize() - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: await db.execute( - "INSERT OR REPLACE INTO registered_webhooks (webhook_id, preset_id, created_at) VALUES (?, ?, ?)", + """ + INSERT INTO registered_webhooks (webhook_id, preset_id, created_at) + VALUES (?, ?, ?) + ON CONFLICT (webhook_id) DO UPDATE SET + preset_id = EXCLUDED.preset_id, + created_at = EXCLUDED.created_at + """, (webhook_id, preset_id, time.time()), ) await db.commit() @@ -1368,7 +1657,7 @@ class RefreshTokenStorage: if not self._initialized: await self.initialize() - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute( "SELECT webhook_id FROM registered_webhooks WHERE preset_id = ?", (preset_id,), @@ -1390,7 +1679,7 @@ class RefreshTokenStorage: if not self._initialized: await self.initialize() - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute( "DELETE FROM registered_webhooks WHERE webhook_id = ?", (webhook_id,) ) @@ -1412,7 +1701,7 @@ class RefreshTokenStorage: if not self._initialized: await self.initialize() - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute( "SELECT webhook_id, preset_id, created_at FROM registered_webhooks ORDER BY created_at DESC" ) @@ -1436,7 +1725,7 @@ class RefreshTokenStorage: if not self._initialized: await self.initialize() - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute( "DELETE FROM registered_webhooks WHERE preset_id = ?", (preset_id,) ) @@ -1478,29 +1767,27 @@ class RefreshTokenStorage: start_time = time.time() try: - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: await db.execute( """ - INSERT OR REPLACE INTO app_passwords + INSERT INTO app_passwords (user_id, encrypted_password, created_at, updated_at) - VALUES ( - ?, - ?, - COALESCE((SELECT created_at FROM app_passwords WHERE user_id = ?), ?), - ? - ) + VALUES (?, ?, ?, ?) + ON CONFLICT (user_id) DO UPDATE SET + encrypted_password = EXCLUDED.encrypted_password, + updated_at = EXCLUDED.updated_at """, - (user_id, encrypted_password, user_id, now, now), + (user_id, encrypted_password, now, now), ) await db.commit() duration = time.time() - start_time - record_db_operation("sqlite", "insert", duration, "success") + record_db_operation(self._dialect, "insert", duration, "success") logger.info("Stored app password for user %s", user_id) except Exception: duration = time.time() - start_time - record_db_operation("sqlite", "insert", duration, "error") + record_db_operation(self._dialect, "insert", duration, "error") raise # Audit log @@ -1531,7 +1818,7 @@ class RefreshTokenStorage: start_time = time.time() try: - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: async with db.execute( "SELECT encrypted_password FROM app_passwords WHERE user_id = ?", (user_id,), @@ -1541,21 +1828,21 @@ class RefreshTokenStorage: if not row: logger.debug("No app password found for user %s", user_id) duration = time.time() - start_time - record_db_operation("sqlite", "select", duration, "success") + record_db_operation(self._dialect, "select", duration, "success") return None encrypted_password = row[0] decrypted_password = self.cipher.decrypt(encrypted_password).decode() duration = time.time() - start_time - record_db_operation("sqlite", "select", duration, "success") + record_db_operation(self._dialect, "select", duration, "success") logger.debug("Retrieved app password for user %s", user_id) return decrypted_password except Exception as e: duration = time.time() - start_time - record_db_operation("sqlite", "select", duration, "error") + record_db_operation(self._dialect, "select", duration, "error") logger.error("Failed to decrypt app password for user %s: %s", user_id, e) return None @@ -1574,7 +1861,7 @@ class RefreshTokenStorage: start_time = time.time() try: - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute( "DELETE FROM app_passwords WHERE user_id = ?", (user_id,), @@ -1583,7 +1870,7 @@ class RefreshTokenStorage: deleted = cursor.rowcount > 0 duration = time.time() - start_time - record_db_operation("sqlite", "delete", duration, "success") + record_db_operation(self._dialect, "delete", duration, "success") if deleted: logger.info("Deleted app password for user %s", user_id) @@ -1599,7 +1886,7 @@ class RefreshTokenStorage: except Exception: duration = time.time() - start_time - record_db_operation("sqlite", "delete", duration, "error") + record_db_operation(self._dialect, "delete", duration, "error") raise async def get_all_app_password_user_ids(self) -> list[str]: @@ -1612,7 +1899,7 @@ class RefreshTokenStorage: if not self._initialized: await self.initialize() - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: async with db.execute( "SELECT user_id FROM app_passwords ORDER BY updated_at DESC" ) as cursor: @@ -1732,24 +2019,21 @@ class RefreshTokenStorage: start_time = time.time() try: - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: await db.execute( """ - INSERT OR REPLACE INTO app_passwords + INSERT INTO app_passwords (user_id, encrypted_password, created_at, updated_at, scopes, username) - VALUES ( - ?, - ?, - COALESCE((SELECT created_at FROM app_passwords WHERE user_id = ?), ?), - ?, - ?, - ? - ) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT (user_id) DO UPDATE SET + encrypted_password = EXCLUDED.encrypted_password, + updated_at = EXCLUDED.updated_at, + scopes = EXCLUDED.scopes, + username = EXCLUDED.username """, ( user_id, encrypted_password, - user_id, now, now, scopes_json, @@ -1759,7 +2043,7 @@ class RefreshTokenStorage: await db.commit() duration = time.time() - start_time - record_db_operation("sqlite", "insert", duration, "success") + record_db_operation(self._dialect, "insert", duration, "success") logger.info( "Stored scoped app password for user %s (scopes=%s, username=%s)", user_id, @@ -1769,7 +2053,7 @@ class RefreshTokenStorage: except Exception: duration = time.time() - start_time - record_db_operation("sqlite", "insert", duration, "error") + record_db_operation(self._dialect, "insert", duration, "error") raise await self._audit_log( @@ -1799,7 +2083,7 @@ class RefreshTokenStorage: start_time = time.time() try: - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: async with db.execute( """ SELECT encrypted_password, scopes, username, created_at, updated_at @@ -1812,7 +2096,7 @@ class RefreshTokenStorage: if not row: logger.debug("No app password found for user %s", user_id) duration = time.time() - start_time - record_db_operation("sqlite", "select", duration, "success") + record_db_operation(self._dialect, "select", duration, "success") return None encrypted_password, scopes_json, username, created_at, updated_at = row @@ -1820,7 +2104,7 @@ class RefreshTokenStorage: scopes = json.loads(scopes_json) if scopes_json else None duration = time.time() - start_time - record_db_operation("sqlite", "select", duration, "success") + record_db_operation(self._dialect, "select", duration, "success") return { "app_password": decrypted_password, @@ -1832,7 +2116,7 @@ class RefreshTokenStorage: except Exception: duration = time.time() - start_time - record_db_operation("sqlite", "select", duration, "error") + record_db_operation(self._dialect, "select", duration, "error") raise async def update_app_password_scopes(self, user_id: str, scopes: list[str]) -> bool: @@ -1852,7 +2136,7 @@ class RefreshTokenStorage: now = int(time.time()) start_time = time.time() try: - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute( "UPDATE app_passwords SET scopes = ?, updated_at = ? WHERE user_id = ?", (scopes_json, now, user_id), @@ -1861,7 +2145,7 @@ class RefreshTokenStorage: updated = cursor.rowcount > 0 duration = time.time() - start_time - record_db_operation("sqlite", "update", duration, "success") + record_db_operation(self._dialect, "update", duration, "success") if updated: await self._audit_log( @@ -1874,7 +2158,7 @@ class RefreshTokenStorage: except Exception: duration = time.time() - start_time - record_db_operation("sqlite", "update", duration, "error") + record_db_operation(self._dialect, "update", duration, "error") raise # ── Login Flow v2: Session Tracking ────────────────────────────────── @@ -1913,13 +2197,19 @@ class RefreshTokenStorage: start_time = time.time() try: - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: await db.execute( """ - INSERT OR REPLACE INTO login_flow_sessions + INSERT INTO login_flow_sessions (user_id, encrypted_poll_token, poll_endpoint, requested_scopes, created_at, expires_at) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT (user_id) DO UPDATE SET + encrypted_poll_token = EXCLUDED.encrypted_poll_token, + poll_endpoint = EXCLUDED.poll_endpoint, + requested_scopes = EXCLUDED.requested_scopes, + created_at = EXCLUDED.created_at, + expires_at = EXCLUDED.expires_at """, ( user_id, @@ -1933,12 +2223,12 @@ class RefreshTokenStorage: await db.commit() duration = time.time() - start_time - record_db_operation("sqlite", "insert", duration, "success") + record_db_operation(self._dialect, "insert", duration, "success") logger.info("Stored login flow session for user %s", user_id) except Exception: duration = time.time() - start_time - record_db_operation("sqlite", "insert", duration, "error") + record_db_operation(self._dialect, "insert", duration, "error") raise async def get_login_flow_session(self, user_id: str) -> dict[str, Any] | None: @@ -1965,7 +2255,7 @@ class RefreshTokenStorage: now = int(time.time()) start_time = time.time() try: - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: async with db.execute( """ SELECT encrypted_poll_token, poll_endpoint, requested_scopes, @@ -1979,7 +2269,7 @@ class RefreshTokenStorage: if not row: duration = time.time() - start_time - record_db_operation("sqlite", "select", duration, "success") + record_db_operation(self._dialect, "select", duration, "success") return None encrypted_token, poll_endpoint, scopes_json, created_at, expires_at = row @@ -1987,7 +2277,7 @@ class RefreshTokenStorage: requested_scopes = json.loads(scopes_json) if scopes_json else None duration = time.time() - start_time - record_db_operation("sqlite", "select", duration, "success") + record_db_operation(self._dialect, "select", duration, "success") return { "poll_token": poll_token, @@ -1999,7 +2289,7 @@ class RefreshTokenStorage: except Exception as e: duration = time.time() - start_time - record_db_operation("sqlite", "select", duration, "error") + record_db_operation(self._dialect, "select", duration, "error") logger.error( "Failed to retrieve login flow session for user %s: %s", user_id, e ) @@ -2019,7 +2309,7 @@ class RefreshTokenStorage: start_time = time.time() try: - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute( "DELETE FROM login_flow_sessions WHERE user_id = ?", (user_id,), @@ -2028,7 +2318,7 @@ class RefreshTokenStorage: deleted = cursor.rowcount > 0 duration = time.time() - start_time - record_db_operation("sqlite", "delete", duration, "success") + record_db_operation(self._dialect, "delete", duration, "success") if deleted: logger.info("Deleted login flow session for user %s", user_id) @@ -2042,7 +2332,7 @@ class RefreshTokenStorage: except Exception: duration = time.time() - start_time - record_db_operation("sqlite", "delete", duration, "error") + record_db_operation(self._dialect, "delete", duration, "error") raise async def delete_expired_login_flow_sessions(self) -> int: @@ -2057,7 +2347,7 @@ class RefreshTokenStorage: now = int(time.time()) start_time = time.time() try: - async with aiosqlite.connect(self.db_path) as db: + async with self._db() as db: cursor = await db.execute( "DELETE FROM login_flow_sessions WHERE expires_at <= ?", (now,), @@ -2066,7 +2356,7 @@ class RefreshTokenStorage: count = cursor.rowcount duration = time.time() - start_time - record_db_operation("sqlite", "delete", duration, "success") + record_db_operation(self._dialect, "delete", duration, "success") if count > 0: logger.info("Cleaned up %s expired login flow sessions", count) @@ -2080,7 +2370,7 @@ class RefreshTokenStorage: except Exception: duration = time.time() - start_time - record_db_operation("sqlite", "delete", duration, "error") + record_db_operation(self._dialect, "delete", duration, "error") raise diff --git a/nextcloud_mcp_server/cli.py b/nextcloud_mcp_server/cli.py index ee6b045c..15e7de1f 100644 --- a/nextcloud_mcp_server/cli.py +++ b/nextcloud_mcp_server/cli.py @@ -5,8 +5,8 @@ import click import uvicorn from nextcloud_mcp_server.config import ( + get_database_url, get_settings, - get_token_db_path, is_ephemeral_token_db, ) from nextcloud_mcp_server.migrations import ( @@ -287,27 +287,67 @@ def db(): pass -def _warn_if_ephemeral(database_path: str) -> None: - if is_ephemeral_token_db(database_path): +def _resolve_db_url(database_url: str | None, database_path: str | None) -> str: + """Pick the database URL for a CLI subcommand. + + Priority: explicit ``--database-url`` > legacy ``--database-path`` + (treated as a SQLite file) > :func:`get_database_url` (honors + ``DATABASE_URL`` env or falls back to the ephemeral SQLite tempfile). + """ + if database_url: + return database_url + if database_path: + return f"sqlite+aiosqlite:///{database_path}" + return get_database_url() + + +def _warn_if_ephemeral(database_url: str) -> None: + """Warn when the resolved URL is the per-process SQLite tempfile.""" + if not database_url.startswith( + "sqlite+aiosqlite:///" + ) and not database_url.startswith("sqlite:///"): + return + path = database_url.split("///", 1)[1] + if is_ephemeral_token_db(path): click.echo( click.style( - f"⚠ Using ephemeral tempfile {database_path}; changes " - "will be lost on exit. Pass --database-path or set " - "TOKEN_STORAGE_DB to operate on a persistent database.", + f"⚠ Using ephemeral tempfile {path}; changes " + "will be lost on exit. Pass --database-url / --database-path " + "or set DATABASE_URL / TOKEN_STORAGE_DB to operate on a " + "persistent database.", fg="yellow", ), err=True, ) +def _db_target_options(fn): + """Attach the shared ``--database-url`` / ``--database-path`` options. + + Using a decorator factory rather than ``**kwargs`` dict-expansion so + static type checkers (ty) see ``click.option`` called with literal + keyword arguments, which is the only form it's typed to accept. + """ + fn = click.option( + "--database-path", + "-d", + envvar="TOKEN_STORAGE_DB", + default=None, + help="SQLite database file path. Equivalent to " + "--database-url sqlite+aiosqlite:///.", + )(fn) + fn = click.option( + "--database-url", + "-u", + envvar="DATABASE_URL", + default=None, + help="SQLAlchemy URL (e.g. postgresql+asyncpg://...). Wins over --database-path.", + )(fn) + return fn + + @db.command() -@click.option( - "--database-path", - "-d", - envvar="TOKEN_STORAGE_DB", - default=None, - help="Path to token storage database (can also use TOKEN_STORAGE_DB env var)", -) +@_db_target_options @click.option( "--revision", "-r", @@ -315,7 +355,7 @@ def _warn_if_ephemeral(database_path: str) -> None: show_default=True, help="Target revision (default: head for latest)", ) -def upgrade(database_path: str | None, revision: str): +def upgrade(database_url: str | None, database_path: str | None, revision: str): """Upgrade database to a specific revision. \b @@ -323,17 +363,17 @@ def upgrade(database_path: str | None, revision: str): # Upgrade to latest version $ nextcloud-mcp-server db upgrade - # Upgrade to specific revision - $ nextcloud-mcp-server db upgrade --revision 001 + # Upgrade a Postgres backend + $ nextcloud-mcp-server db upgrade -u postgresql+asyncpg://mcp:mcp@db/mcp - # Use custom database path + # Use custom SQLite path $ nextcloud-mcp-server db upgrade -d /path/to/tokens.db """ - database_path = database_path or get_token_db_path() - _warn_if_ephemeral(database_path) + url = _resolve_db_url(database_url, database_path) + _warn_if_ephemeral(url) try: click.echo(f"Upgrading database to revision: {revision}") - upgrade_database(database_path, revision) + upgrade_database(url, revision) click.echo(click.style("✓ Database upgraded successfully", fg="green")) except Exception as e: click.echo(click.style(f"✗ Upgrade failed: {e}", fg="red"), err=True) @@ -341,13 +381,7 @@ def upgrade(database_path: str | None, revision: str): @db.command() -@click.option( - "--database-path", - "-d", - envvar="TOKEN_STORAGE_DB", - default=None, - help="Path to token storage database", -) +@_db_target_options @click.option( "--revision", "-r", @@ -358,27 +392,16 @@ def upgrade(database_path: str | None, revision: str): @click.confirmation_option( prompt="Are you sure you want to downgrade the database? This may result in data loss." ) -def downgrade(database_path: str | None, revision: str): +def downgrade(database_url: str | None, database_path: str | None, revision: str): """Downgrade database to a specific revision. WARNING: This may result in data loss! Use with caution. - - \b - Examples: - # Downgrade by one version - $ nextcloud-mcp-server db downgrade - - # Downgrade to specific revision - $ nextcloud-mcp-server db downgrade --revision 001 - - # Downgrade to base (empty database) - $ nextcloud-mcp-server db downgrade --revision base """ - database_path = database_path or get_token_db_path() - _warn_if_ephemeral(database_path) + url = _resolve_db_url(database_url, database_path) + _warn_if_ephemeral(url) try: click.echo(f"Downgrading database to revision: {revision}") - downgrade_database(database_path, revision) + downgrade_database(url, revision) click.echo(click.style("✓ Database downgraded successfully", fg="green")) except Exception as e: click.echo(click.style(f"✗ Downgrade failed: {e}", fg="red"), err=True) @@ -386,24 +409,13 @@ def downgrade(database_path: str | None, revision: str): @db.command() -@click.option( - "--database-path", - "-d", - envvar="TOKEN_STORAGE_DB", - default=None, - help="Path to token storage database", -) -def current(database_path: str | None): - """Show current database revision. - - \b - Example: - $ nextcloud-mcp-server db current - """ - database_path = database_path or get_token_db_path() - _warn_if_ephemeral(database_path) +@_db_target_options +def current(database_url: str | None, database_path: str | None): + """Show current database revision.""" + url = _resolve_db_url(database_url, database_path) + _warn_if_ephemeral(url) try: - revision = get_current_revision(database_path) + revision = get_current_revision(url) if revision: click.echo(f"Current revision: {click.style(revision, fg='cyan')}") else: @@ -420,25 +432,14 @@ def current(database_path: str | None): @db.command() -@click.option( - "--database-path", - "-d", - envvar="TOKEN_STORAGE_DB", - default=None, - help="Path to token storage database", -) -def history(database_path: str | None): - """Show migration history. - - \b - Example: - $ nextcloud-mcp-server db history - """ - database_path = database_path or get_token_db_path() - _warn_if_ephemeral(database_path) +@_db_target_options +def history(database_url: str | None, database_path: str | None): + """Show migration history.""" + url = _resolve_db_url(database_url, database_path) + _warn_if_ephemeral(url) try: click.echo("Migration history:") - show_migration_history(database_path) + show_migration_history(url) except Exception as e: click.echo(click.style(f"✗ Failed to show history: {e}", fg="red"), err=True) raise click.ClickException(str(e)) diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 85a1d6df..7a3c3b4c 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -56,6 +56,10 @@ _DEFAULTS: dict[str, Any] = { # None = ephemeral per-process tempfile (see get_token_db_path()). # Set TOKEN_STORAGE_DB to persist tokens across restarts. "token_storage_db": None, + # Centralized backend (any SQLAlchemy URL). Wins over TOKEN_STORAGE_DB + # when set. Use postgresql+asyncpg://user:pw@host/db for HA k8s + # deployments so pods can be stateless. See ADR-026. + "database_url": None, # Webhook delivery authentication (ADR-010): when set, registrations # tell NC to add `Authorization: Bearer ` to webhook deliveries # and the receiver rejects unauthenticated requests. @@ -279,6 +283,30 @@ def is_ephemeral_token_db(path: str) -> bool: return path == _ephemeral_db_path +def get_database_url() -> str: + """Resolve the SQLAlchemy database URL for token storage. + + Priority: + 1. ``DATABASE_URL`` if set — any SQLAlchemy URL is accepted; the primary + supported backends are ``postgresql+asyncpg://...`` for HA k8s + deployments and ``sqlite+aiosqlite:///...`` for development. + 2. Otherwise build ``sqlite+aiosqlite:///{get_token_db_path()}`` so the + legacy ``TOKEN_STORAGE_DB`` env var and the ephemeral-tempfile + fallback both keep working unchanged. + """ + explicit = _dynaconf.get("DATABASE_URL") + if explicit: + return str(explicit) + return f"sqlite+aiosqlite:///{get_token_db_path()}" + + +def is_sqlite_url(url: str) -> bool: + """Return True for SQLite SQLAlchemy URLs (used to gate sqlite-only logic + like file-permission hardening and ``sqlite_master`` legacy lookups). + """ + return url.startswith("sqlite") + + LOGGING_CONFIG = { "version": 1, "disable_existing_loggers": False, diff --git a/nextcloud_mcp_server/migrations.py b/nextcloud_mcp_server/migrations.py index 15b5274c..6b2cff41 100644 --- a/nextcloud_mcp_server/migrations.py +++ b/nextcloud_mcp_server/migrations.py @@ -3,22 +3,49 @@ This module provides helper functions for managing Alembic database migrations programmatically. It enables automatic migration on application startup and provides CLI integration. + +All helpers accept a SQLAlchemy URL (``sqlite+aiosqlite:///...`` or +``postgresql+asyncpg://...``). When called without an explicit URL they fall +back to :func:`nextcloud_mcp_server.config.get_database_url`. """ import logging -import sqlite3 from pathlib import Path from alembic.config import Config +from sqlalchemy import create_engine, inspect, text import nextcloud_mcp_server.alembic as alembic_package from alembic import command -from nextcloud_mcp_server.config import get_token_db_path +from nextcloud_mcp_server.config import get_database_url logger = logging.getLogger(__name__) -def get_alembic_config(database_path: str | Path | None = None) -> Config: +def _coerce_url(database_url: str | Path | None) -> str: + """Accept either a URL string, a Path (legacy SQLite path), or None. + + A bare ``Path`` is interpreted as a SQLite database file for backward + compatibility with the prior path-based API. + """ + if database_url is None: + return get_database_url() + if isinstance(database_url, Path): + return f"sqlite+aiosqlite:///{database_url.resolve()}" + return database_url + + +def _to_sync_url(database_url: str) -> str: + """Map an async driver URL to its sync equivalent for blocking inspection. + + SQLAlchemy's :func:`inspect` and :func:`create_engine` used below are + synchronous APIs. The runtime uses async drivers (``aiosqlite``, + ``asyncpg``) but Alembic and these utility queries don't need them. + """ + return database_url.replace("+aiosqlite", "").replace("+asyncpg", "") + + +def get_alembic_config(database_url: str | Path | None = None) -> Config: """ Get Alembic configuration for programmatic use. @@ -26,145 +53,102 @@ def get_alembic_config(database_path: str | Path | None = None) -> Config: package location instead of alembic.ini file. Args: - database_path: Path to SQLite database file. If None, resolves via - config.get_token_db_path() (ephemeral tempfile unless - TOKEN_STORAGE_DB is set). + database_url: SQLAlchemy URL. If None, resolves via + :func:`get_database_url` (DATABASE_URL env var, falling back + to the ephemeral SQLite tempfile under ``TOKEN_STORAGE_DB``). + For backward compatibility a ``Path`` is treated as a SQLite + file path. Returns: - Alembic Config object configured for the specified database + Alembic Config object configured for the resolved URL. """ - # Use package location (works in both editable and installed modes) if alembic_package.__file__ is None: raise RuntimeError("alembic package __file__ is None") script_location = Path(alembic_package.__file__).parent - # Create config programmatically (no alembic.ini needed at runtime) config = Config() config.set_main_option("script_location", str(script_location)) - config.set_main_option("path_separator", "os") # Suppress deprecation warning + config.set_main_option("path_separator", "os") - # Set database URL - if database_path: - db_path = Path(database_path).resolve() - else: - db_path = Path(get_token_db_path()).resolve() - - url = f"sqlite+aiosqlite:///{db_path}" + url = _coerce_url(database_url) config.set_main_option("sqlalchemy.url", url) logger.debug("Alembic script location: %s", script_location) - logger.debug("Database: %s", db_path) + logger.debug("Database URL: %s", url) return config def upgrade_database( - database_path: str | Path | None = None, revision: str = "head" + database_url: str | Path | None = None, revision: str = "head" ) -> None: - """ - Upgrade database to a specific revision. - - Args: - database_path: Path to SQLite database file - revision: Target revision (default: "head" for latest) - """ - config = get_alembic_config(database_path) + """Upgrade database to a specific revision (default: latest).""" + config = get_alembic_config(database_url) logger.info("Upgrading database to revision: %s", revision) command.upgrade(config, revision) logger.info("Database upgrade completed successfully") def downgrade_database( - database_path: str | Path | None = None, revision: str = "-1" + database_url: str | Path | None = None, revision: str = "-1" ) -> None: - """ - Downgrade database to a specific revision. - - Args: - database_path: Path to SQLite database file - revision: Target revision (default: "-1" for previous version) - """ - config = get_alembic_config(database_path) + """Downgrade database to a specific revision (default: previous).""" + config = get_alembic_config(database_url) logger.warning("Downgrading database to revision: %s", revision) command.downgrade(config, revision) logger.info("Database downgrade completed successfully") -def get_current_revision(database_path: str | Path | None = None) -> str | None: +def get_current_revision(database_url: str | Path | None = None) -> str | None: """ - Get the current database revision by directly querying the alembic_version table. + Get the current database revision by reading the ``alembic_version`` table. - Args: - database_path: Path to SQLite database file - - Returns: - Current revision ID or None if not versioned + Returns ``None`` when the database does not exist or has no + ``alembic_version`` table (i.e. has never been migrated). """ + url = _to_sync_url(_coerce_url(database_url)) - if database_path is None: - database_path = get_token_db_path() - - db_path = Path(database_path).resolve() - - if not db_path.exists(): - logger.debug("Database does not exist: %s", db_path) - return None - - try: - # Query alembic_version table directly - conn = sqlite3.connect(str(db_path)) - cursor = conn.cursor() - - # Check if alembic_version table exists - cursor.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name='alembic_version'" - ) - has_table = cursor.fetchone() is not None - - if not has_table: - conn.close() + if url.startswith("sqlite:///"): + path = url[len("sqlite:///") :] + if path and not Path(path).exists(): + logger.debug("Database does not exist: %s", path) return None - # Get current version - cursor.execute("SELECT version_num FROM alembic_version") - row = cursor.fetchone() - conn.close() - - return row[0] if row else None - + try: + engine = create_engine(url, future=True) + try: + inspector = inspect(engine) + if not inspector.has_table("alembic_version"): + return None + with engine.connect() as conn: + row = conn.execute( + text("SELECT version_num FROM alembic_version") + ).fetchone() + return row[0] if row else None + finally: + engine.dispose() except Exception as e: logger.error("Failed to get current revision: %s", e) return None def stamp_database( - database_path: str | Path | None = None, revision: str = "head" + database_url: str | Path | None = None, revision: str = "head" ) -> None: """ Stamp database with a specific revision without running migrations. - This is useful for marking existing databases that were created before - Alembic was introduced. It tells Alembic "this database is at revision X" - without actually running the migration. - - Args: - database_path: Path to SQLite database file - revision: Revision to stamp (default: "head" for latest) + Useful for marking pre-Alembic databases as already at a known revision. """ - config = get_alembic_config(database_path) + config = get_alembic_config(database_url) logger.info("Stamping database with revision: %s", revision) command.stamp(config, revision) logger.info("Database stamped successfully") -def show_migration_history(database_path: str | Path | None = None) -> None: - """ - Display migration history. - - Args: - database_path: Path to SQLite database file - """ - config = get_alembic_config(database_path) +def show_migration_history(database_url: str | Path | None = None) -> None: + """Display migration history.""" + config = get_alembic_config(database_url) command.history(config, verbose=True) @@ -178,7 +162,9 @@ def create_migration(message: str, autogenerate: bool = False) -> None: Note: Since we don't use SQLAlchemy models, autogenerate will be disabled - and migrations must be written manually. + and migrations must be written manually using portable Alembic + operations (``op.create_table``, ``op.add_column`` …) rather than + raw SQL so they work on both SQLite and Postgres. """ config = get_alembic_config() logger.info("Creating new migration: %s", message) diff --git a/pyproject.toml b/pyproject.toml index 1e5c205d..9bd6a67e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,6 +45,8 @@ dependencies = [ "openai>=2.8.1", "dynaconf>=3.2.13,<4.0", "mistralai>=2.4.5", + "sqlalchemy[asyncio]>=2.0", + "asyncpg>=0.29", ] classifiers = [ "Development Status :: 4 - Beta", @@ -78,6 +80,7 @@ markers = [ "keycloak: OAuth tests that utilize keycloak external identity provider", "login_flow: Login Flow v2 integration tests (ADR-022)", "multi_user_basic: Multi-user BasicAuth pass-through tests (ADR-020)", + "postgres: Tests requiring the docker-compose postgres-test service (ADR-026)", ] testpaths = [ "tests", diff --git a/tests/fixtures/__init__.py b/tests/fixtures/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/fixtures/storage_backend.py b/tests/fixtures/storage_backend.py new file mode 100644 index 00000000..22a7f2e1 --- /dev/null +++ b/tests/fixtures/storage_backend.py @@ -0,0 +1,97 @@ +"""Shared pytest fixture for parametrizing storage tests over backends. + +Tests for ``RefreshTokenStorage`` are exercised against every backend that is +available in the current environment: + +- ``sqlite`` — always available; uses a per-test tempfile. +- ``postgres`` — opt-in. Bring up the test instance with:: + + docker compose --profile postgres up -d postgres-test + + and export the URL so the fixture picks it up:: + + export TEST_DATABASE_URL=postgresql+asyncpg://mcp:mcp@localhost:5433/mcp + + When ``TEST_DATABASE_URL`` is unset (or the host is unreachable), the + Postgres parametrization is skipped automatically so the suite still runs + cleanly without Docker. + +Each Postgres test runs against an isolated schema that is dropped and +recreated between tests, mirroring the per-tempfile isolation that the +SQLite path gets for free. +""" + +from __future__ import annotations + +import os +from typing import Any + +import pytest + + +def _postgres_url() -> str | None: + """Resolve the Postgres URL for tests, or ``None`` when opted out.""" + return os.environ.get("TEST_DATABASE_URL") or None + + +def _postgres_reachable(url: str) -> bool: + """Return ``True`` if the configured Postgres accepts TCP connections. + + A lightweight socket probe is used rather than a full DB handshake so + we don't have to ship a sync Postgres driver (psycopg2) just for the + test gate — asyncpg only works inside an event loop. + """ + import socket + from urllib.parse import urlparse + + try: + parsed = urlparse(url) + host = parsed.hostname or "localhost" + port = parsed.port or 5432 + with socket.create_connection((host, port), timeout=1.0): + return True + except OSError: + return False + + +def _backend_params() -> list[Any]: + """Build the pytest parametrize list, gating Postgres on availability.""" + params: list[Any] = [pytest.param("sqlite", id="sqlite")] + url = _postgres_url() + if url and _postgres_reachable(url): + params.append(pytest.param(url, id="postgres")) + return params + + +@pytest.fixture(params=_backend_params()) +def storage_backend(request): + """Yield ``{"kind": ..., "url": ..., "reset": }`` per backend. + + For ``sqlite`` the test fixture builds its own tempfile path; only the + ``kind`` discriminator is used. For ``postgres`` the URL is forwarded + and a ``reset()`` coroutine is provided so test fixtures can wipe the + schema between parametrized runs. + """ + if request.param == "sqlite": + yield {"kind": "sqlite"} + return + + url = request.param + + async def reset() -> None: + # Use a fresh async engine so we don't fight an async connection + # the test might still be holding open at teardown time. asyncpg + # is the only driver we ship for Postgres, so the reset path stays + # event-loop-only (no psycopg2 dependency required). + from sqlalchemy import text + from sqlalchemy.ext.asyncio import create_async_engine + + engine = create_async_engine(url, future=True) + try: + async with engine.begin() as conn: + await conn.execute(text("DROP SCHEMA public CASCADE")) + await conn.execute(text("CREATE SCHEMA public")) + finally: + await engine.dispose() + + yield {"kind": "postgres", "url": url, "reset": reset} diff --git a/tests/integration/test_storage_postgres.py b/tests/integration/test_storage_postgres.py new file mode 100644 index 00000000..8cdee460 --- /dev/null +++ b/tests/integration/test_storage_postgres.py @@ -0,0 +1,164 @@ +"""End-to-end Postgres backend smoke for RefreshTokenStorage (ADR-026). + +Exercises every storage method touched by the SQLAlchemy / asyncpg port +against a fresh Postgres schema. The test is opt-in: it requires the +``postgres-test`` docker-compose service to be running and +``TEST_DATABASE_URL`` to be exported. + +Bring up the dependency once:: + + docker compose --profile postgres up -d postgres-test + export TEST_DATABASE_URL=postgresql+asyncpg://mcp:mcp@localhost:5433/mcp + +Then run:: + + uv run pytest tests/integration/test_storage_postgres.py -v -m postgres + +When ``TEST_DATABASE_URL`` is unset (or the service is unreachable) the +test is skipped so the full suite still passes locally without Docker. +""" + +from __future__ import annotations + +import os +import socket +from urllib.parse import urlparse + +import pytest +from cryptography.fernet import Fernet + +from nextcloud_mcp_server.auth.storage import RefreshTokenStorage + +pytestmark = [pytest.mark.integration, pytest.mark.postgres] + + +def _postgres_url() -> str | None: + return os.environ.get("TEST_DATABASE_URL") or None + + +def _reachable(url: str) -> bool: + parsed = urlparse(url) + try: + with socket.create_connection( + (parsed.hostname or "localhost", parsed.port or 5432), timeout=1.0 + ): + return True + except OSError: + return False + + +@pytest.fixture +def postgres_url() -> str: + url = _postgres_url() + if not url: + pytest.skip( + "TEST_DATABASE_URL not set — run " + "`docker compose --profile postgres up -d postgres-test` and export " + "TEST_DATABASE_URL=postgresql+asyncpg://mcp:mcp@localhost:5433/mcp" + ) + if not _reachable(url): + pytest.skip(f"Postgres at {url} is not reachable") + return url + + +@pytest.fixture +async def reset_schema(postgres_url: str): + """Drop+recreate the public schema before and after each test.""" + from sqlalchemy import text + from sqlalchemy.ext.asyncio import create_async_engine + + async def _reset() -> None: + engine = create_async_engine(postgres_url, future=True) + try: + async with engine.begin() as conn: + await conn.execute(text("DROP SCHEMA public CASCADE")) + await conn.execute(text("CREATE SCHEMA public")) + finally: + await engine.dispose() + + await _reset() + yield + await _reset() + + +@pytest.fixture +async def storage(postgres_url: str, reset_schema): + key = Fernet.generate_key() + s = RefreshTokenStorage(database_url=postgres_url, encryption_key=key) + await s.initialize() + yield s + + +async def test_refresh_token_roundtrip(storage: RefreshTokenStorage): + """Store + retrieve + upsert + delete a refresh token end-to-end.""" + await storage.store_refresh_token( + user_id="alice", refresh_token="rt-1", expires_at=9_999_999_999 + ) + tok = await storage.get_refresh_token("alice") + assert tok is not None + assert tok["refresh_token"] == "rt-1" + assert tok["expires_at"] == 9_999_999_999 + + # Upsert preserves user_id, swaps token contents. + await storage.store_refresh_token( + user_id="alice", refresh_token="rt-2", expires_at=9_999_999_999 + ) + tok = await storage.get_refresh_token("alice") + assert tok is not None and tok["refresh_token"] == "rt-2" + + assert await storage.delete_refresh_token("alice") is True + assert await storage.get_refresh_token("alice") is None + + +async def test_app_password_roundtrip(storage: RefreshTokenStorage): + """Store + retrieve + replace + delete a scoped app password.""" + await storage.store_app_password(user_id="bob", app_password="pw-1") + assert await storage.get_app_password("bob") == "pw-1" + + # Replace path exercises the ON CONFLICT DO UPDATE on the singleton row. + await storage.store_app_password(user_id="bob", app_password="pw-2") + assert await storage.get_app_password("bob") == "pw-2" + + assert await storage.delete_app_password("bob") is True + assert await storage.get_app_password("bob") is None + + +async def test_oauth_session_lifecycle(storage: RefreshTokenStorage): + """Cover the ADR-004 progressive-consent session table.""" + await storage.store_oauth_session( + session_id="sess-1", + client_redirect_uri="http://localhost:12345/callback", + mcp_authorization_code="mcp-code-abc", + flow_type="hybrid", + ttl_seconds=600, + ) + fetched = await storage.get_oauth_session("sess-1") + assert fetched is not None + assert fetched["mcp_authorization_code"] == "mcp-code-abc" + + by_code = await storage.get_oauth_session_by_mcp_code("mcp-code-abc") + assert by_code is not None and by_code["session_id"] == "sess-1" + + +async def test_webhook_tracking(storage: RefreshTokenStorage): + """Tracks webhook ↔ preset mappings via ON CONFLICT upserts.""" + await storage.store_webhook(webhook_id=101, preset_id="notes_sync") + await storage.store_webhook(webhook_id=202, preset_id="notes_sync") + await storage.store_webhook(webhook_id=303, preset_id="calendar_sync") + + assert sorted(await storage.get_webhooks_by_preset("notes_sync")) == [101, 202] + assert await storage.get_webhooks_by_preset("calendar_sync") == [303] + + # Re-storing the same webhook_id is a no-op upsert. + await storage.store_webhook(webhook_id=101, preset_id="notes_sync") + assert sorted(await storage.get_webhooks_by_preset("notes_sync")) == [101, 202] + + assert await storage.delete_webhook(webhook_id=101) is True + assert await storage.get_webhooks_by_preset("notes_sync") == [202] + + +async def test_audit_log_capture(storage: RefreshTokenStorage): + """Audit events from upstream methods land in audit_logs.""" + await storage.store_app_password(user_id="carol", app_password="x") + logs = await storage.get_audit_logs(user_id="carol", limit=10) + assert any(entry["event"] == "store_app_password" for entry in logs) diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 29b3b160..51ba62a8 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -2,6 +2,11 @@ import pytest +# Re-export the parametrized storage backend fixture so it's auto-discovered +# by every unit test that names it as a parameter, without each test module +# having to import it explicitly. +from tests.fixtures.storage_backend import storage_backend # noqa: F401 + @pytest.fixture(autouse=True) def _reload_dynaconf_after_test(): diff --git a/tests/unit/test_app_password_storage.py b/tests/unit/test_app_password_storage.py index 37431974..ed2f425f 100644 --- a/tests/unit/test_app_password_storage.py +++ b/tests/unit/test_app_password_storage.py @@ -3,6 +3,12 @@ Unit tests for App Password Storage functionality. Tests the app password methods in RefreshTokenStorage for multi-user BasicAuth mode background sync. + +These tests are parametrized over both supported backends so the storage +layer is exercised against SQLite (default, always runs) and Postgres (gated +on ``TEST_DATABASE_URL``; bring up ``docker compose --profile postgres up +-d postgres-test`` and export ``TEST_DATABASE_URL=postgresql+asyncpg://mcp:mcp@localhost:5433/mcp`` +to opt in). """ import tempfile @@ -23,15 +29,32 @@ def encryption_key(): @pytest.fixture -async def temp_storage(encryption_key): - """Create temporary storage instance with encryption for testing.""" - with tempfile.TemporaryDirectory() as tmpdir: - db_path = Path(tmpdir) / "test_app_passwords.db" +async def temp_storage(encryption_key, storage_backend): + """Create a storage instance backed by either SQLite or Postgres. + + The ``storage_backend`` fixture is parametrized by pytest, so every test + that uses ``temp_storage`` runs once per backend that is available in + the current environment. + """ + if storage_backend["kind"] == "sqlite": + with tempfile.TemporaryDirectory() as tmpdir: + db_path = Path(tmpdir) / "test_app_passwords.db" + storage = RefreshTokenStorage( + db_path=str(db_path), encryption_key=encryption_key + ) + await storage.initialize() + yield storage + else: storage = RefreshTokenStorage( - db_path=str(db_path), encryption_key=encryption_key + database_url=storage_backend["url"], encryption_key=encryption_key ) await storage.initialize() - yield storage + try: + yield storage + finally: + # Each test gets an isolated schema; tear it down so the next + # parametrized run starts clean. + await storage_backend["reset"]() async def test_store_app_password(temp_storage): diff --git a/tests/unit/test_webhook_storage.py b/tests/unit/test_webhook_storage.py index e07d485e..be41347f 100644 --- a/tests/unit/test_webhook_storage.py +++ b/tests/unit/test_webhook_storage.py @@ -3,6 +3,9 @@ Unit tests for Webhook Storage functionality. Tests the webhook tracking methods in RefreshTokenStorage without requiring real database connections or network calls. + +Runs against both SQLite and Postgres backends — see the docstring on +``tests.fixtures.storage_backend`` for opt-in instructions. """ import tempfile @@ -17,14 +20,23 @@ pytestmark = pytest.mark.unit @pytest.fixture -async def temp_storage(): - """Create temporary storage instance for testing.""" - with tempfile.TemporaryDirectory() as tmpdir: - db_path = Path(tmpdir) / "test_webhooks.db" - # No encryption key needed for webhook tracking - storage = RefreshTokenStorage(db_path=str(db_path), encryption_key=None) +async def temp_storage(storage_backend): + """Create a storage instance backed by either SQLite or Postgres.""" + if storage_backend["kind"] == "sqlite": + with tempfile.TemporaryDirectory() as tmpdir: + db_path = Path(tmpdir) / "test_webhooks.db" + storage = RefreshTokenStorage(db_path=str(db_path), encryption_key=None) + await storage.initialize() + yield storage + else: + storage = RefreshTokenStorage( + database_url=storage_backend["url"], encryption_key=None + ) await storage.initialize() - yield storage + try: + yield storage + finally: + await storage_backend["reset"]() async def test_store_webhook(temp_storage): diff --git a/uv.lock b/uv.lock index b41bbc0d..c0bb2d9d 100644 --- a/uv.lock +++ b/uv.lock @@ -226,6 +226,54 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d2/39/e7eaf1799466a4aef85b6a4fe7bd175ad2b1c6345066aa33f1f58d4b18d0/asttokens-3.0.1-py3-none-any.whl", hash = "sha256:15a3ebc0f43c2d0a50eeafea25e19046c68398e487b9f1f5b517f7c0f40f976a", size = 27047, upload-time = "2025-11-15T16:43:16.109Z" }, ] +[[package]] +name = "asyncpg" +version = "0.31.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/fe/cc/d18065ce2380d80b1bcce927c24a2642efd38918e33fd724bc4bca904877/asyncpg-0.31.0.tar.gz", hash = "sha256:c989386c83940bfbd787180f2b1519415e2d3d6277a70d9d0f0145ac73500735", size = 993667, upload-time = "2025-11-24T23:27:00.812Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/08/17/cc02bc49bc350623d050fa139e34ea512cd6e020562f2a7312a7bcae4bc9/asyncpg-0.31.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:eee690960e8ab85063ba93af2ce128c0f52fd655fdff9fdb1a28df01329f031d", size = 643159, upload-time = "2025-11-24T23:25:36.443Z" }, + { url = "https://files.pythonhosted.org/packages/a4/62/4ded7d400a7b651adf06f49ea8f73100cca07c6df012119594d1e3447aa6/asyncpg-0.31.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:2657204552b75f8288de08ca60faf4a99a65deef3a71d1467454123205a88fab", size = 638157, upload-time = "2025-11-24T23:25:37.89Z" }, + { url = "https://files.pythonhosted.org/packages/d6/5b/4179538a9a72166a0bf60ad783b1ef16efb7960e4d7b9afe9f77a5551680/asyncpg-0.31.0-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:a429e842a3a4b4ea240ea52d7fe3f82d5149853249306f7ff166cb9948faa46c", size = 2918051, upload-time = "2025-11-24T23:25:39.461Z" }, + { url = "https://files.pythonhosted.org/packages/e6/35/c27719ae0536c5b6e61e4701391ffe435ef59539e9360959240d6e47c8c8/asyncpg-0.31.0-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:c0807be46c32c963ae40d329b3a686356e417f674c976c07fa49f1b30303f109", size = 2972640, upload-time = "2025-11-24T23:25:41.512Z" }, + { url = "https://files.pythonhosted.org/packages/43/f4/01ebb9207f29e645a64699b9ce0eefeff8e7a33494e1d29bb53736f7766b/asyncpg-0.31.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:e5d5098f63beeae93512ee513d4c0c53dc12e9aa2b7a1af5a81cddf93fe4e4da", size = 2851050, upload-time = "2025-11-24T23:25:43.153Z" }, + { url = "https://files.pythonhosted.org/packages/3e/f4/03ff1426acc87be0f4e8d40fa2bff5c3952bef0080062af9efc2212e3be8/asyncpg-0.31.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:37fc6c00a814e18eef51833545d1891cac9aa69140598bb076b4cd29b3e010b9", size = 2962574, upload-time = "2025-11-24T23:25:44.942Z" }, + { url = "https://files.pythonhosted.org/packages/c7/39/cc788dfca3d4060f9d93e67be396ceec458dfc429e26139059e58c2c244d/asyncpg-0.31.0-cp311-cp311-win32.whl", hash = "sha256:5a4af56edf82a701aece93190cc4e094d2df7d33f6e915c222fb09efbb5afc24", size = 521076, upload-time = "2025-11-24T23:25:46.486Z" }, + { url = "https://files.pythonhosted.org/packages/28/fc/735af5384c029eb7f1ca60ccb8fa95521dbdaeef788edf4cecfc604c3cab/asyncpg-0.31.0-cp311-cp311-win_amd64.whl", hash = "sha256:480c4befbdf079c14c9ca43c8c5e1fe8b6296c96f1f927158d4f1e750aacc047", size = 584980, upload-time = "2025-11-24T23:25:47.938Z" }, + { url = "https://files.pythonhosted.org/packages/2a/a6/59d0a146e61d20e18db7396583242e32e0f120693b67a8de43f1557033e2/asyncpg-0.31.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:b44c31e1efc1c15188ef183f287c728e2046abb1d26af4d20858215d50d91fad", size = 662042, upload-time = "2025-11-24T23:25:49.578Z" }, + { url = "https://files.pythonhosted.org/packages/36/01/ffaa189dcb63a2471720615e60185c3f6327716fdc0fc04334436fbb7c65/asyncpg-0.31.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:0c89ccf741c067614c9b5fc7f1fc6f3b61ab05ae4aaa966e6fd6b93097c7d20d", size = 638504, upload-time = "2025-11-24T23:25:51.501Z" }, + { url = "https://files.pythonhosted.org/packages/9f/62/3f699ba45d8bd24c5d65392190d19656d74ff0185f42e19d0bbd973bb371/asyncpg-0.31.0-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:12b3b2e39dc5470abd5e98c8d3373e4b1d1234d9fbdedf538798b2c13c64460a", size = 3426241, upload-time = "2025-11-24T23:25:53.278Z" }, + { url = "https://files.pythonhosted.org/packages/8c/d1/a867c2150f9c6e7af6462637f613ba67f78a314b00db220cd26ff559d532/asyncpg-0.31.0-cp312-cp312-manylinux_2_28_x86_64.whl", hash = "sha256:aad7a33913fb8bcb5454313377cc330fbb19a0cd5faa7272407d8a0c4257b671", size = 3520321, upload-time = "2025-11-24T23:25:54.982Z" }, + { url = "https://files.pythonhosted.org/packages/7a/1a/cce4c3f246805ecd285a3591222a2611141f1669d002163abef999b60f98/asyncpg-0.31.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:3df118d94f46d85b2e434fd62c84cb66d5834d5a890725fe625f498e72e4d5ec", size = 3316685, upload-time = "2025-11-24T23:25:57.43Z" }, + { url = "https://files.pythonhosted.org/packages/40/ae/0fc961179e78cc579e138fad6eb580448ecae64908f95b8cb8ee2f241f67/asyncpg-0.31.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:bd5b6efff3c17c3202d4b37189969acf8927438a238c6257f66be3c426beba20", size = 3471858, upload-time = "2025-11-24T23:25:59.636Z" }, + { url = "https://files.pythonhosted.org/packages/52/b2/b20e09670be031afa4cbfabd645caece7f85ec62d69c312239de568e058e/asyncpg-0.31.0-cp312-cp312-win32.whl", hash = "sha256:027eaa61361ec735926566f995d959ade4796f6a49d3bde17e5134b9964f9ba8", size = 527852, upload-time = "2025-11-24T23:26:01.084Z" }, + { url = "https://files.pythonhosted.org/packages/b5/f0/f2ed1de154e15b107dc692262395b3c17fc34eafe2a78fc2115931561730/asyncpg-0.31.0-cp312-cp312-win_amd64.whl", hash = "sha256:72d6bdcbc93d608a1158f17932de2321f68b1a967a13e014998db87a72ed3186", size = 597175, upload-time = "2025-11-24T23:26:02.564Z" }, + { url = "https://files.pythonhosted.org/packages/95/11/97b5c2af72a5d0b9bc3fa30cd4b9ce22284a9a943a150fdc768763caf035/asyncpg-0.31.0-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:c204fab1b91e08b0f47e90a75d1b3c62174dab21f670ad6c5d0f243a228f015b", size = 661111, upload-time = "2025-11-24T23:26:04.467Z" }, + { url = "https://files.pythonhosted.org/packages/1b/71/157d611c791a5e2d0423f09f027bd499935f0906e0c2a416ce712ba51ef3/asyncpg-0.31.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:54a64f91839ba59008eccf7aad2e93d6e3de688d796f35803235ea1c4898ae1e", size = 636928, upload-time = "2025-11-24T23:26:05.944Z" }, + { url = "https://files.pythonhosted.org/packages/2e/fc/9e3486fb2bbe69d4a867c0b76d68542650a7ff1574ca40e84c3111bb0c6e/asyncpg-0.31.0-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:c0e0822b1038dc7253b337b0f3f676cadc4ac31b126c5d42691c39691962e403", size = 3424067, upload-time = "2025-11-24T23:26:07.957Z" }, + { url = "https://files.pythonhosted.org/packages/12/c6/8c9d076f73f07f995013c791e018a1cd5f31823c2a3187fc8581706aa00f/asyncpg-0.31.0-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:bef056aa502ee34204c161c72ca1f3c274917596877f825968368b2c33f585f4", size = 3518156, upload-time = "2025-11-24T23:26:09.591Z" }, + { url = "https://files.pythonhosted.org/packages/ae/3b/60683a0baf50fbc546499cfb53132cb6835b92b529a05f6a81471ab60d0c/asyncpg-0.31.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:0bfbcc5b7ffcd9b75ab1558f00db2ae07db9c80637ad1b2469c43df79d7a5ae2", size = 3319636, upload-time = "2025-11-24T23:26:11.168Z" }, + { url = "https://files.pythonhosted.org/packages/50/dc/8487df0f69bd398a61e1792b3cba0e47477f214eff085ba0efa7eac9ce87/asyncpg-0.31.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:22bc525ebbdc24d1261ecbf6f504998244d4e3be1721784b5f64664d61fbe602", size = 3472079, upload-time = "2025-11-24T23:26:13.164Z" }, + { url = "https://files.pythonhosted.org/packages/13/a1/c5bbeeb8531c05c89135cb8b28575ac2fac618bcb60119ee9696c3faf71c/asyncpg-0.31.0-cp313-cp313-win32.whl", hash = "sha256:f890de5e1e4f7e14023619399a471ce4b71f5418cd67a51853b9910fdfa73696", size = 527606, upload-time = "2025-11-24T23:26:14.78Z" }, + { url = "https://files.pythonhosted.org/packages/91/66/b25ccb84a246b470eb943b0107c07edcae51804912b824054b3413995a10/asyncpg-0.31.0-cp313-cp313-win_amd64.whl", hash = "sha256:dc5f2fa9916f292e5c5c8b2ac2813763bcd7f58e130055b4ad8a0531314201ab", size = 596569, upload-time = "2025-11-24T23:26:16.189Z" }, + { url = "https://files.pythonhosted.org/packages/3c/36/e9450d62e84a13aea6580c83a47a437f26c7ca6fa0f0fd40b6670793ea30/asyncpg-0.31.0-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:f6b56b91bb0ffc328c4e3ed113136cddd9deefdf5f79ab448598b9772831df44", size = 660867, upload-time = "2025-11-24T23:26:17.631Z" }, + { url = "https://files.pythonhosted.org/packages/82/4b/1d0a2b33b3102d210439338e1beea616a6122267c0df459ff0265cd5807a/asyncpg-0.31.0-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:334dec28cf20d7f5bb9e45b39546ddf247f8042a690bff9b9573d00086e69cb5", size = 638349, upload-time = "2025-11-24T23:26:19.689Z" }, + { url = "https://files.pythonhosted.org/packages/41/aa/e7f7ac9a7974f08eff9183e392b2d62516f90412686532d27e196c0f0eeb/asyncpg-0.31.0-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:98cc158c53f46de7bb677fd20c417e264fc02b36d901cc2a43bd6cb0dc6dbfd2", size = 3410428, upload-time = "2025-11-24T23:26:21.275Z" }, + { url = "https://files.pythonhosted.org/packages/6f/de/bf1b60de3dede5c2731e6788617a512bc0ebd9693eac297ee74086f101d7/asyncpg-0.31.0-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:9322b563e2661a52e3cdbc93eed3be7748b289f792e0011cb2720d278b366ce2", size = 3471678, upload-time = "2025-11-24T23:26:23.627Z" }, + { url = "https://files.pythonhosted.org/packages/46/78/fc3ade003e22d8bd53aaf8f75f4be48f0b460fa73738f0391b9c856a9147/asyncpg-0.31.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:19857a358fc811d82227449b7ca40afb46e75b33eb8897240c3839dd8b744218", size = 3313505, upload-time = "2025-11-24T23:26:25.235Z" }, + { url = "https://files.pythonhosted.org/packages/bf/e9/73eb8a6789e927816f4705291be21f2225687bfa97321e40cd23055e903a/asyncpg-0.31.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:ba5f8886e850882ff2c2ace5732300e99193823e8107e2c53ef01c1ebfa1e85d", size = 3434744, upload-time = "2025-11-24T23:26:26.944Z" }, + { url = "https://files.pythonhosted.org/packages/08/4b/f10b880534413c65c5b5862f79b8e81553a8f364e5238832ad4c0af71b7f/asyncpg-0.31.0-cp314-cp314-win32.whl", hash = "sha256:cea3a0b2a14f95834cee29432e4ddc399b95700eb1d51bbc5bfee8f31fa07b2b", size = 532251, upload-time = "2025-11-24T23:26:28.404Z" }, + { url = "https://files.pythonhosted.org/packages/d3/2d/7aa40750b7a19efa5d66e67fc06008ca0f27ba1bd082e457ad82f59aba49/asyncpg-0.31.0-cp314-cp314-win_amd64.whl", hash = "sha256:04d19392716af6b029411a0264d92093b6e5e8285ae97a39957b9a9c14ea72be", size = 604901, upload-time = "2025-11-24T23:26:30.34Z" }, + { url = "https://files.pythonhosted.org/packages/ce/fe/b9dfe349b83b9dee28cc42360d2c86b2cdce4cb551a2c2d27e156bcac84d/asyncpg-0.31.0-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:bdb957706da132e982cc6856bb2f7b740603472b54c3ebc77fe60ea3e57e1bd2", size = 702280, upload-time = "2025-11-24T23:26:32Z" }, + { url = "https://files.pythonhosted.org/packages/6a/81/e6be6e37e560bd91e6c23ea8a6138a04fd057b08cf63d3c5055c98e81c1d/asyncpg-0.31.0-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:6d11b198111a72f47154fa03b85799f9be63701e068b43f84ac25da0bda9cb31", size = 682931, upload-time = "2025-11-24T23:26:33.572Z" }, + { url = "https://files.pythonhosted.org/packages/a6/45/6009040da85a1648dd5bc75b3b0a062081c483e75a1a29041ae63a0bf0dc/asyncpg-0.31.0-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:18c83b03bc0d1b23e6230f5bf8d4f217dc9bc08644ce0502a9d91dc9e634a9c7", size = 3581608, upload-time = "2025-11-24T23:26:35.638Z" }, + { url = "https://files.pythonhosted.org/packages/7e/06/2e3d4d7608b0b2b3adbee0d0bd6a2d29ca0fc4d8a78f8277df04e2d1fd7b/asyncpg-0.31.0-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:e009abc333464ff18b8f6fd146addffd9aaf63e79aa3bb40ab7a4c332d0c5e9e", size = 3498738, upload-time = "2025-11-24T23:26:37.275Z" }, + { url = "https://files.pythonhosted.org/packages/7d/aa/7d75ede780033141c51d83577ea23236ba7d3a23593929b32b49db8ed36e/asyncpg-0.31.0-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:3b1fbcb0e396a5ca435a8826a87e5c2c2cc0c8c68eb6fadf82168056b0e53a8c", size = 3401026, upload-time = "2025-11-24T23:26:39.423Z" }, + { url = "https://files.pythonhosted.org/packages/ba/7a/15e37d45e7f7c94facc1e9148c0e455e8f33c08f0b8a0b1deb2c5171771b/asyncpg-0.31.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:8df714dba348efcc162d2adf02d213e5fab1bd9f557e1305633e851a61814a7a", size = 3429426, upload-time = "2025-11-24T23:26:41.032Z" }, + { url = "https://files.pythonhosted.org/packages/13/d5/71437c5f6ae5f307828710efbe62163974e71237d5d46ebd2869ea052d10/asyncpg-0.31.0-cp314-cp314t-win32.whl", hash = "sha256:1b41f1afb1033f2b44f3234993b15096ddc9cd71b21a42dbd87fc6a57b43d65d", size = 614495, upload-time = "2025-11-24T23:26:42.659Z" }, + { url = "https://files.pythonhosted.org/packages/3c/d7/8fb3044eaef08a310acfe23dae9a8e2e07d305edc29a53497e52bc76eca7/asyncpg-0.31.0-cp314-cp314t-win_amd64.whl", hash = "sha256:bd4107bb7cdd0e9e65fae66a62afd3a249663b844fa34d479f6d5b3bef9c04c3", size = 706062, upload-time = "2025-11-24T23:26:44.086Z" }, +] + [[package]] name = "attrs" version = "25.4.0" @@ -2129,6 +2177,7 @@ dependencies = [ { name = "aiosqlite" }, { name = "alembic" }, { name = "anthropic" }, + { name = "asyncpg" }, { name = "authlib" }, { name = "boto3" }, { name = "caldav" }, @@ -2159,6 +2208,7 @@ dependencies = [ { name = "pythonvcard4" }, { name = "qdrant-client" }, { name = "recurring-ical-events" }, + { name = "sqlalchemy", extra = ["asyncio"] }, { name = "starlette" }, ] @@ -2184,6 +2234,7 @@ requires-dist = [ { name = "aiosqlite", specifier = ">=0.20.0" }, { name = "alembic", specifier = ">=1.14.0" }, { name = "anthropic", specifier = ">=0.42.0" }, + { name = "asyncpg", specifier = ">=0.29" }, { name = "authlib", specifier = ">=1.6.5" }, { name = "boto3", specifier = ">=1.35.0" }, { name = "caldav", specifier = ">=3.0.1,<4.0" }, @@ -2214,6 +2265,7 @@ requires-dist = [ { name = "pythonvcard4", specifier = ">=0.2.0" }, { name = "qdrant-client", specifier = ">=1.17.0" }, { name = "recurring-ical-events", specifier = ">=3.8.0,<4.0" }, + { name = "sqlalchemy", extras = ["asyncio"], specifier = ">=2.0" }, { name = "starlette", specifier = "<1.0" }, ] @@ -3944,6 +3996,11 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/bf/e1/3ccb13c643399d22289c6a9786c1a91e3dcbb68bce4beb44926ac2c557bf/sqlalchemy-2.0.45-py3-none-any.whl", hash = "sha256:5225a288e4c8cc2308dbdd874edad6e7d0fd38eac1e9e5f23503425c8eee20d0", size = 1936672, upload-time = "2025-12-09T21:54:52.608Z" }, ] +[package.optional-dependencies] +asyncio = [ + { name = "greenlet" }, +] + [[package]] name = "sse-starlette" version = "3.0.3"