feat: add Alembic database migration system
Implements Alembic for managing token storage database schema versions. Migrations run automatically on startup with full backward compatibility. **Changes:** - Add Alembic dependency (1.14.0+) and SQLAlchemy (auto-installed) - Create migration infrastructure in alembic/ directory - Add initial migration (001) capturing current schema - Modify RefreshTokenStorage.initialize() to run migrations via anyio - Add CLI commands: db upgrade, current, history, downgrade, migrate - Add comprehensive migration documentation **Backward Compatibility:** - Pre-Alembic databases automatically stamped with revision 001 - No schema changes for existing databases - Automatic upgrade on first startup after update **Migration Strategy:** Three scenarios handled: 1. New database → Run migrations from scratch 2. Pre-Alembic database → Stamp with 001 (no changes) 3. Alembic-managed → Upgrade to latest **Architecture:** - Uses anyio.to_thread.run_sync() for structured concurrency - Alembic env.py runs with anyio.run() in worker thread - SQLite-friendly migration patterns documented - No ThreadPoolExecutor needed (anyio handles it) **CLI Usage:** ```bash nextcloud-mcp-server db upgrade # Upgrade to latest nextcloud-mcp-server db current # Show version nextcloud-mcp-server db history # View changelog nextcloud-mcp-server db downgrade # Rollback (with confirmation) nextcloud-mcp-server db migrate "description" # Create migration ``` **Testing:** - All 13 webhook storage tests pass - New/pre-Alembic database scenarios validated - anyio integration tested 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
co-authored by
Claude Sonnet 4.5
Claude Sonnet 4.5
parent
a4a34e46a8
commit
3fa376905c
@@ -376,16 +376,35 @@ async def oauth_login_callback(request: Request) -> RedirectResponse | HTMLRespo
|
||||
user_id = f"user-{secrets.token_hex(8)}"
|
||||
username = "unknown"
|
||||
|
||||
# Calculate refresh token expiration from token response
|
||||
refresh_expires_in = token_data.get("refresh_expires_in")
|
||||
refresh_expires_at = None
|
||||
if refresh_expires_in:
|
||||
import time
|
||||
|
||||
refresh_expires_at = int(time.time()) + refresh_expires_in
|
||||
logger.info(
|
||||
f"Refresh token expires in {refresh_expires_in}s (at timestamp {refresh_expires_at})"
|
||||
)
|
||||
|
||||
# Extract granted scopes
|
||||
granted_scopes = (
|
||||
token_data.get("scope", "").split() if token_data.get("scope") else None
|
||||
)
|
||||
|
||||
# Store refresh token (for background jobs ONLY)
|
||||
if refresh_token:
|
||||
logger.info(f"Storing refresh token for user_id: {user_id}")
|
||||
logger.info(f" State parameter (provisioning_client_id): {state[:16]}...")
|
||||
logger.info(f" Granted scopes: {granted_scopes}")
|
||||
logger.info(f" Expires at: {refresh_expires_at}")
|
||||
await storage.store_refresh_token(
|
||||
user_id=user_id,
|
||||
refresh_token=refresh_token,
|
||||
expires_at=None,
|
||||
expires_at=refresh_expires_at,
|
||||
flow_type="browser", # Browser-based login flow
|
||||
provisioning_client_id=state, # Store state for unified session lookup
|
||||
scopes=granted_scopes,
|
||||
)
|
||||
logger.info(f"✓ Refresh token stored successfully for user_id: {user_id}")
|
||||
logger.info(
|
||||
|
||||
@@ -517,12 +517,23 @@ async def oauth_callback_nextcloud(request: Request):
|
||||
token_data.get("scope", "").split() if token_data.get("scope") else None
|
||||
)
|
||||
|
||||
# Calculate refresh token expiration from token response
|
||||
refresh_expires_in = token_data.get("refresh_expires_in")
|
||||
refresh_expires_at = None
|
||||
if refresh_expires_in:
|
||||
import time
|
||||
|
||||
refresh_expires_at = int(time.time()) + refresh_expires_in
|
||||
logger.info(f" refresh_expires_in: {refresh_expires_in}s")
|
||||
logger.info(f" refresh_expires_at: {refresh_expires_at}")
|
||||
|
||||
logger.info("Storing refresh token:")
|
||||
logger.info(f" user_id: {user_id}")
|
||||
logger.info(" flow_type: flow2")
|
||||
logger.info(" token_audience: nextcloud")
|
||||
logger.info(f" provisioning_client_id: {state[:16]}...")
|
||||
logger.info(f" scopes: {granted_scopes}")
|
||||
logger.info(f" expires_at: {refresh_expires_at}")
|
||||
|
||||
await storage.store_refresh_token(
|
||||
user_id=user_id,
|
||||
@@ -531,7 +542,7 @@ async def oauth_callback_nextcloud(request: Request):
|
||||
token_audience="nextcloud",
|
||||
provisioning_client_id=state, # Store which client initiated provisioning
|
||||
scopes=granted_scopes,
|
||||
expires_at=None, # Refresh tokens typically don't expire
|
||||
expires_at=refresh_expires_at,
|
||||
)
|
||||
logger.info(f"✓ Stored Flow 2 master refresh token for user {user_id}")
|
||||
logger.info("=" * 60)
|
||||
|
||||
@@ -117,7 +117,14 @@ class RefreshTokenStorage:
|
||||
return cls(db_path=db_path, encryption_key=encryption_key)
|
||||
|
||||
async def initialize(self) -> None:
|
||||
"""Initialize database schema"""
|
||||
"""
|
||||
Initialize database schema using Alembic migrations.
|
||||
|
||||
This method handles three scenarios:
|
||||
1. New database: Run migrations from scratch
|
||||
2. Pre-Alembic database: Stamp with initial revision (no changes)
|
||||
3. Alembic-managed database: Upgrade to latest version
|
||||
"""
|
||||
if self._initialized:
|
||||
return
|
||||
|
||||
@@ -125,137 +132,59 @@ class RefreshTokenStorage:
|
||||
db_dir = Path(self.db_path).parent
|
||||
db_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Set restrictive permissions on database file
|
||||
# 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:
|
||||
await db.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', -- 'hybrid', 'flow1', 'flow2'
|
||||
token_audience TEXT DEFAULT 'nextcloud', -- 'mcp-server' or 'nextcloud'
|
||||
provisioned_at INTEGER, -- When Flow 2 was completed
|
||||
provisioning_client_id TEXT, -- Which MCP client initiated Flow 1
|
||||
scopes TEXT, -- JSON array of granted scopes
|
||||
-- Browser session profile cache
|
||||
user_profile TEXT, -- JSON cache of IdP user profile (for browser UI only)
|
||||
profile_cached_at INTEGER -- When profile was last cached
|
||||
# Check if database is managed by Alembic
|
||||
cursor = await db.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name='alembic_version'"
|
||||
)
|
||||
has_alembic = await cursor.fetchone() is not None
|
||||
|
||||
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
|
||||
|
||||
await db.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
|
||||
if has_schema:
|
||||
logger.info(
|
||||
f"Detected pre-Alembic database at {self.db_path}, "
|
||||
"stamping with initial revision"
|
||||
)
|
||||
else:
|
||||
logger.info(
|
||||
f"Initializing new database at {self.db_path} with migrations"
|
||||
)
|
||||
|
||||
# Run migrations in a worker thread using anyio.to_thread
|
||||
# This allows Alembic to run its own async operations in a separate context
|
||||
from anyio import to_thread
|
||||
|
||||
from nextcloud_mcp_server.migrations import stamp_database, upgrade_database
|
||||
|
||||
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(
|
||||
"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("Database initialized with migrations")
|
||||
else:
|
||||
# Alembic-managed database - upgrade to latest
|
||||
await to_thread.run_sync(upgrade_database, self.db_path, "head")
|
||||
logger.info("Database upgraded to latest version")
|
||||
|
||||
# Create index on audit logs for efficient queries
|
||||
await db.execute(
|
||||
"CREATE INDEX IF NOT EXISTS idx_audit_user_timestamp "
|
||||
"ON audit_logs(user_id, timestamp)"
|
||||
)
|
||||
|
||||
# OAuth client credentials storage
|
||||
await db.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
|
||||
)
|
||||
"""
|
||||
)
|
||||
|
||||
# OAuth flow sessions (ADR-004 Progressive Consent)
|
||||
await db.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', -- 'hybrid', 'flow1', 'flow2'
|
||||
requested_scopes TEXT, -- JSON array of requested scopes
|
||||
granted_scopes TEXT, -- JSON array of granted scopes
|
||||
is_provisioning BOOLEAN DEFAULT FALSE -- True if this is a Flow 2 provisioning session
|
||||
)
|
||||
"""
|
||||
)
|
||||
|
||||
# Create index for MCP authorization code lookups
|
||||
await db.execute(
|
||||
"CREATE INDEX IF NOT EXISTS idx_oauth_sessions_mcp_code "
|
||||
"ON oauth_sessions(mcp_authorization_code)"
|
||||
)
|
||||
|
||||
# Schema version tracking
|
||||
await db.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)
|
||||
await db.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
|
||||
)
|
||||
"""
|
||||
)
|
||||
|
||||
# Create indexes for efficient webhook queries
|
||||
await db.execute(
|
||||
"CREATE INDEX IF NOT EXISTS idx_webhooks_preset "
|
||||
"ON registered_webhooks(preset_id)"
|
||||
)
|
||||
await db.execute(
|
||||
"CREATE INDEX IF NOT EXISTS idx_webhooks_created "
|
||||
"ON registered_webhooks(created_at)"
|
||||
)
|
||||
|
||||
await db.commit()
|
||||
|
||||
# Set restrictive permissions after creation
|
||||
# Set restrictive permissions after initialization
|
||||
os.chmod(self.db_path, 0o600)
|
||||
|
||||
self._initialized = True
|
||||
|
||||
Reference in New Issue
Block a user