fix(auth): Store app passwords locally for multi-user BasicAuth background sync

Previously, the multi-user BasicAuth mode attempted to retrieve app passwords
via OAuth client_credentials grant, which Nextcloud OIDC doesn't support.

This fix implements local storage for app passwords:
- Add app_passwords table via Alembic migration (002)
- Add store/get/delete methods to RefreshTokenStorage
- Add management API endpoints for app password provisioning:
  - POST /api/v1/users/{user_id}/app-password
  - GET /api/v1/users/{user_id}/app-password
  - DELETE /api/v1/users/{user_id}/app-password
- Update oauth_sync.py to read from local storage
- Update Astrolabe to send app passwords to MCP server after validation
- Add app-hook to configure mcp_server_url in Nextcloud

The flow is now:
1. User creates app password in Nextcloud Security settings
2. User enters it in Astrolabe Personal Settings
3. Astrolabe validates against Nextcloud, then sends to MCP server
4. MCP server stores encrypted app password locally
5. Background sync uses locally stored password

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-01-13 15:44:11 +01:00
parent 546f0c0674
commit e486e92f91
7 changed files with 691 additions and 32 deletions
+174
View File
@@ -1240,6 +1240,180 @@ class RefreshTokenStorage:
return deleted
# ============================================================================
# App Password Storage (multi-user BasicAuth mode)
# ============================================================================
async def store_app_password(
self,
user_id: str,
app_password: str,
) -> None:
"""
Store encrypted app password for background sync (multi-user BasicAuth mode).
Args:
user_id: Nextcloud user ID
app_password: Nextcloud app password to store
"""
if not self._initialized:
await self.initialize()
if not self.cipher:
raise RuntimeError(
"Encryption key not configured. "
"Set TOKEN_ENCRYPTION_KEY for app password storage."
)
encrypted_password = self.cipher.encrypt(app_password.encode())
now = int(time.time())
start_time = time.time()
try:
async with aiosqlite.connect(self.db_path) as db:
await db.execute(
"""
INSERT OR REPLACE INTO app_passwords
(user_id, encrypted_password, created_at, updated_at)
VALUES (
?,
?,
COALESCE((SELECT created_at FROM app_passwords WHERE user_id = ?), ?),
?
)
""",
(user_id, encrypted_password, user_id, now, now),
)
await db.commit()
duration = time.time() - start_time
record_db_operation("sqlite", "insert", duration, "success")
logger.info(f"Stored app password for user {user_id}")
except Exception:
duration = time.time() - start_time
record_db_operation("sqlite", "insert", duration, "error")
raise
# Audit log
await self._audit_log(
event="store_app_password",
user_id=user_id,
auth_method="app_password",
)
async def get_app_password(self, user_id: str) -> Optional[str]:
"""
Retrieve and decrypt app password for a user.
Args:
user_id: Nextcloud user ID
Returns:
Decrypted app password, or None if not found
"""
if not self._initialized:
await self.initialize()
if not self.cipher:
raise RuntimeError(
"Encryption key not configured. "
"Set TOKEN_ENCRYPTION_KEY for app password retrieval."
)
start_time = time.time()
try:
async with aiosqlite.connect(self.db_path) as db:
async with db.execute(
"SELECT encrypted_password FROM app_passwords WHERE user_id = ?",
(user_id,),
) as cursor:
row = await cursor.fetchone()
if not row:
logger.debug(f"No app password found for user {user_id}")
duration = time.time() - start_time
record_db_operation("sqlite", "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")
logger.debug(f"Retrieved app password for user {user_id}")
return decrypted_password
except Exception as e:
duration = time.time() - start_time
record_db_operation("sqlite", "select", duration, "error")
logger.error(f"Failed to decrypt app password for user {user_id}: {e}")
return None
async def delete_app_password(self, user_id: str) -> bool:
"""
Delete app password for a user.
Args:
user_id: Nextcloud user ID
Returns:
True if password was deleted, False if not found
"""
if not self._initialized:
await self.initialize()
start_time = time.time()
try:
async with aiosqlite.connect(self.db_path) as db:
cursor = await db.execute(
"DELETE FROM app_passwords WHERE user_id = ?",
(user_id,),
)
await db.commit()
deleted = cursor.rowcount > 0
duration = time.time() - start_time
record_db_operation("sqlite", "delete", duration, "success")
if deleted:
logger.info(f"Deleted app password for user {user_id}")
await self._audit_log(
event="delete_app_password",
user_id=user_id,
auth_method="app_password",
)
else:
logger.debug(f"No app password to delete for user {user_id}")
return deleted
except Exception:
duration = time.time() - start_time
record_db_operation("sqlite", "delete", duration, "error")
raise
async def get_all_app_password_user_ids(self) -> list[str]:
"""
Get list of all user IDs with stored app passwords.
Returns:
List of user IDs
"""
if not self._initialized:
await self.initialize()
async with aiosqlite.connect(self.db_path) as db:
async with db.execute(
"SELECT user_id FROM app_passwords ORDER BY updated_at DESC"
) as cursor:
rows = await cursor.fetchall()
user_ids = [row[0] for row in rows]
logger.debug(f"Found {len(user_ids)} users with app passwords")
return user_ids
async def generate_encryption_key() -> str:
"""