817 lines
29 KiB
Python
817 lines
29 KiB
Python
"""
|
|
Connection management endpoints — all existing cloud connection behaviors.
|
|
|
|
Preserves all existing endpoint URLs from api/cloud.py without change.
|
|
Introduces PATCH /api/cloud/connections/{connection_id} for display_name rename.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import secrets
|
|
import uuid
|
|
import urllib.parse
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from fastapi.responses import JSONResponse, RedirectResponse
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from api.cloud.schemas import ConnectionRenameRequest
|
|
from api.schemas import CloudConnectionOut
|
|
from config import settings
|
|
from db.models import CloudConnection, User
|
|
from deps.auth import get_regular_user
|
|
from deps.db import get_db
|
|
from deps.utils import get_client_ip, parse_uuid
|
|
from services.audit import write_audit_log
|
|
from services.rate_limiting import account_limiter
|
|
from storage.cloud_backend_factory import build_cloud_backend
|
|
from storage.cloud_utils import decrypt_credentials, encrypt_credentials, validate_cloud_url
|
|
|
|
router = APIRouter()
|
|
|
|
VALID_OAUTH_PROVIDERS = {"google_drive", "onedrive"}
|
|
VALID_WEBDAV_PROVIDERS = {"nextcloud", "webdav"}
|
|
VALID_CLOUD_PROVIDERS = VALID_OAUTH_PROVIDERS | VALID_WEBDAV_PROVIDERS
|
|
_VALID_BACKENDS = frozenset({"minio", *VALID_CLOUD_PROVIDERS})
|
|
|
|
_DISPLAY_NAMES = {
|
|
"google_drive": "Google Drive",
|
|
"onedrive": "OneDrive",
|
|
"nextcloud": "Nextcloud",
|
|
"webdav": "WebDAV server",
|
|
}
|
|
|
|
|
|
class WebDAVConnectRequest(BaseModel):
|
|
server_url: str
|
|
username: str
|
|
password: str
|
|
provider: str # "nextcloud" or "webdav"
|
|
|
|
|
|
class DefaultStorageRequest(BaseModel):
|
|
backend: str
|
|
|
|
|
|
def _master_key() -> bytes:
|
|
return settings.cloud_creds_key.encode()
|
|
|
|
|
|
def _oauth_redirect_uri(provider: str) -> str:
|
|
return f"{settings.backend_url}/api/cloud/oauth/callback/{provider}"
|
|
|
|
|
|
def _oauth_state_key(state_token: str) -> str:
|
|
return f"oauth_state:{state_token}"
|
|
|
|
|
|
def _settings_redirect(**params: str) -> RedirectResponse:
|
|
query = urllib.parse.urlencode(params)
|
|
return RedirectResponse(url=f"{settings.frontend_url}/settings?{query}", status_code=302)
|
|
|
|
|
|
def _unsupported_provider_error(provider: str, valid_providers: set[str]) -> HTTPException:
|
|
return HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Unsupported provider: {provider}. Valid providers: {sorted(valid_providers)}",
|
|
)
|
|
|
|
|
|
def _ensure_oauth_configured(provider: str) -> None:
|
|
if provider == "google_drive" and (not settings.google_client_id or not settings.google_client_secret):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Google Drive OAuth is not configured on this server. Set GOOGLE_CLIENT_ID and GOOGLE_CLIENT_SECRET in your environment.",
|
|
)
|
|
|
|
if provider == "onedrive" and (
|
|
not settings.onedrive_client_id
|
|
or not settings.onedrive_client_secret
|
|
or not settings.onedrive_tenant_id
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="OneDrive OAuth is not configured on this server. Set ONEDRIVE_CLIENT_ID, ONEDRIVE_CLIENT_SECRET, and ONEDRIVE_TENANT_ID in your environment.",
|
|
)
|
|
|
|
|
|
def _webdav_credentials(body: WebDAVConnectRequest) -> dict[str, str]:
|
|
return {
|
|
"server_url": body.server_url,
|
|
"username": body.username,
|
|
"password": body.password,
|
|
}
|
|
|
|
|
|
async def _insert_cloud_connection(
|
|
session: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
provider: str,
|
|
credentials_enc: str,
|
|
) -> CloudConnection:
|
|
"""Always insert a new connection row — never merges same-provider rows."""
|
|
conn = CloudConnection(
|
|
id=uuid.uuid4(),
|
|
user_id=user_id,
|
|
provider=provider,
|
|
display_name=_DISPLAY_NAMES.get(provider, provider),
|
|
credentials_enc=credentials_enc,
|
|
status="ACTIVE",
|
|
)
|
|
session.add(conn)
|
|
return conn
|
|
|
|
|
|
# Keep legacy alias used by OAuth callback (OAuth creates one connection per flow)
|
|
_upsert_cloud_connection = _insert_cloud_connection
|
|
|
|
|
|
async def _get_owned_connection(
|
|
session: AsyncSession,
|
|
connection_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
) -> CloudConnection:
|
|
"""Return the connection owned by user_id, or raise 404."""
|
|
conn = await session.get(CloudConnection, connection_id)
|
|
if conn is None or conn.user_id != user_id:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Connection not found")
|
|
return conn
|
|
|
|
|
|
async def _get_active_connection(
|
|
session: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
provider: str,
|
|
) -> CloudConnection:
|
|
result = await session.execute(
|
|
select(CloudConnection).where(
|
|
CloudConnection.user_id == user_id,
|
|
CloudConnection.provider == provider,
|
|
CloudConnection.status == "ACTIVE",
|
|
)
|
|
)
|
|
conn = result.scalar_one_or_none()
|
|
if conn is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="No active cloud connection found for this provider",
|
|
)
|
|
return conn
|
|
|
|
|
|
def _decrypt_connection(conn: CloudConnection, user_id: uuid.UUID) -> dict:
|
|
return decrypt_credentials(_master_key(), str(user_id), conn.credentials_enc)
|
|
|
|
|
|
async def _oauth_authorization_url(provider: str, redirect_uri: str, state_token: str) -> str:
|
|
if provider == "google_drive":
|
|
from google_auth_oauthlib.flow import Flow # lazy import
|
|
|
|
# D-17: Phase 13 requests the full `drive` scope so authorized users can
|
|
# operate on all existing Drive items, not just files created by this app.
|
|
# Consent text must make the expanded scope explicit.
|
|
flow = Flow.from_client_config(
|
|
{
|
|
"web": {
|
|
"client_id": settings.google_client_id,
|
|
"client_secret": settings.google_client_secret,
|
|
"auth_uri": "https://accounts.google.com/o/oauth2/auth",
|
|
"token_uri": "https://oauth2.googleapis.com/token",
|
|
}
|
|
},
|
|
scopes=["https://www.googleapis.com/auth/drive"],
|
|
redirect_uri=redirect_uri,
|
|
)
|
|
authorization_url, _ = flow.authorization_url(
|
|
access_type="offline",
|
|
prompt="consent",
|
|
state=state_token,
|
|
)
|
|
return authorization_url
|
|
|
|
if provider == "onedrive":
|
|
import msal # lazy import
|
|
|
|
app = msal.ConfidentialClientApplication(
|
|
settings.onedrive_client_id,
|
|
client_credential=settings.onedrive_client_secret,
|
|
authority=f"https://login.microsoftonline.com/{settings.onedrive_tenant_id}",
|
|
)
|
|
return await asyncio.to_thread(
|
|
app.get_authorization_request_url,
|
|
scopes=["Files.ReadWrite", "offline_access"],
|
|
redirect_uri=redirect_uri,
|
|
state=state_token,
|
|
)
|
|
|
|
raise ValueError(f"Unsupported OAuth provider: {provider}")
|
|
|
|
|
|
async def _oauth_credentials_from_code(provider: str, redirect_uri: str, code: Optional[str]) -> dict:
|
|
if provider == "google_drive":
|
|
from google_auth_oauthlib.flow import Flow # lazy import
|
|
|
|
# D-17: Broader Phase 13 drive scope (must match initiation scope above).
|
|
flow = Flow.from_client_config(
|
|
{
|
|
"web": {
|
|
"client_id": settings.google_client_id,
|
|
"client_secret": settings.google_client_secret,
|
|
"auth_uri": "https://accounts.google.com/o/oauth2/auth",
|
|
"token_uri": "https://oauth2.googleapis.com/token",
|
|
}
|
|
},
|
|
scopes=["https://www.googleapis.com/auth/drive"],
|
|
redirect_uri=redirect_uri,
|
|
)
|
|
await asyncio.to_thread(flow.fetch_token, code=code)
|
|
token_credentials = flow.credentials
|
|
return {
|
|
"access_token": token_credentials.token,
|
|
"refresh_token": token_credentials.refresh_token,
|
|
"token_uri": token_credentials.token_uri,
|
|
"client_id": token_credentials.client_id,
|
|
"client_secret": token_credentials.client_secret,
|
|
"expiry": token_credentials.expiry.isoformat() if token_credentials.expiry else None,
|
|
}
|
|
|
|
if provider == "onedrive":
|
|
import msal # lazy import
|
|
|
|
app = msal.ConfidentialClientApplication(
|
|
settings.onedrive_client_id,
|
|
client_credential=settings.onedrive_client_secret,
|
|
authority=f"https://login.microsoftonline.com/{settings.onedrive_tenant_id}",
|
|
)
|
|
result = await asyncio.to_thread(
|
|
app.acquire_token_by_authorization_code,
|
|
code,
|
|
scopes=["Files.ReadWrite", "offline_access"],
|
|
redirect_uri=redirect_uri,
|
|
)
|
|
if "error" in result:
|
|
raise ValueError(f"Token exchange failed: {result.get('error_description', result['error'])}")
|
|
return {
|
|
"access_token": result["access_token"],
|
|
"refresh_token": result.get("refresh_token", ""),
|
|
"token_uri": f"https://login.microsoftonline.com/{settings.onedrive_tenant_id}/oauth2/v2.0/token",
|
|
"client_id": settings.onedrive_client_id,
|
|
"client_secret": settings.onedrive_client_secret,
|
|
}
|
|
|
|
raise ValueError(f"Unsupported OAuth provider: {provider}")
|
|
|
|
|
|
def _connection_response(conn: CloudConnection) -> dict:
|
|
data = CloudConnectionOut.model_validate(conn).model_dump()
|
|
if conn.provider not in VALID_WEBDAV_PROVIDERS:
|
|
return data
|
|
|
|
try:
|
|
credentials = _decrypt_connection(conn, conn.user_id)
|
|
except Exception:
|
|
return data
|
|
|
|
data["server_url"] = credentials.get("server_url")
|
|
data["connection_username"] = credentials.get("username")
|
|
return data
|
|
|
|
|
|
# ── Endpoints ─────────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/oauth/initiate/{provider}")
|
|
@account_limiter.limit("100/minute")
|
|
async def oauth_initiate(
|
|
provider: str,
|
|
request: Request,
|
|
current_user: User = Depends(get_regular_user),
|
|
) -> dict:
|
|
"""Start an OAuth flow and return the provider authorization URL."""
|
|
request.state.current_user = current_user
|
|
|
|
if provider not in VALID_OAUTH_PROVIDERS:
|
|
raise _unsupported_provider_error(provider, VALID_OAUTH_PROVIDERS)
|
|
_ensure_oauth_configured(provider)
|
|
|
|
state_token = secrets.token_urlsafe(32)
|
|
await request.app.state.redis.setex(_oauth_state_key(state_token), 1800, str(current_user.id))
|
|
|
|
authorization_url = await _oauth_authorization_url(
|
|
provider,
|
|
redirect_uri=_oauth_redirect_uri(provider),
|
|
state_token=state_token,
|
|
)
|
|
return JSONResponse({"url": authorization_url})
|
|
|
|
|
|
@router.get("/oauth/callback/{provider}", response_class=RedirectResponse)
|
|
async def oauth_callback(
|
|
provider: str,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Exchange OAuth code, store encrypted credentials, and redirect to settings."""
|
|
state = request.query_params.get("state")
|
|
code = request.query_params.get("code")
|
|
error_param = request.query_params.get("error")
|
|
|
|
try:
|
|
if error_param:
|
|
raise ValueError(f"OAuth provider returned error: {error_param}")
|
|
|
|
if provider not in VALID_OAUTH_PROVIDERS:
|
|
raise ValueError(f"Unsupported OAuth provider: {provider}")
|
|
|
|
if not state:
|
|
raise ValueError("Missing OAuth state parameter")
|
|
|
|
redis_client = request.app.state.redis
|
|
state_key = _oauth_state_key(state)
|
|
stored_user_id = await redis_client.get(state_key)
|
|
|
|
if not stored_user_id:
|
|
return _settings_redirect(
|
|
cloud_error="Invalid or expired OAuth state. Please try connecting again."
|
|
)
|
|
|
|
await redis_client.delete(state_key)
|
|
|
|
if isinstance(stored_user_id, bytes):
|
|
stored_user_id = stored_user_id.decode("utf-8")
|
|
user_id = uuid.UUID(stored_user_id)
|
|
|
|
user = await session.get(User, user_id)
|
|
if user is None or not user.is_active:
|
|
raise ValueError("User not found or inactive")
|
|
|
|
credentials = await _oauth_credentials_from_code(provider, _oauth_redirect_uri(provider), code)
|
|
credentials_enc = encrypt_credentials(_master_key(), str(user_id), credentials)
|
|
conn = await _upsert_cloud_connection(session, user_id, provider, credentials_enc)
|
|
await session.flush()
|
|
|
|
await write_audit_log(
|
|
session,
|
|
event_type="cloud.connected",
|
|
user_id=user.id,
|
|
actor_id=user.id,
|
|
resource_id=conn.id,
|
|
ip_address=None,
|
|
metadata_={"provider": provider},
|
|
)
|
|
await session.commit()
|
|
|
|
return _settings_redirect(cloud_connected=provider)
|
|
|
|
except Exception as exc:
|
|
return _settings_redirect(cloud_error=str(exc))
|
|
|
|
|
|
@router.post("/connections/webdav", status_code=status.HTTP_201_CREATED)
|
|
@account_limiter.limit("100/minute")
|
|
async def connect_webdav(
|
|
body: WebDAVConnectRequest,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
) -> dict:
|
|
"""Connect a WebDAV or Nextcloud server."""
|
|
request.state.current_user = current_user
|
|
if body.provider not in VALID_WEBDAV_PROVIDERS:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"Unsupported WebDAV provider: {body.provider}. Valid values: {sorted(VALID_WEBDAV_PROVIDERS)}",
|
|
)
|
|
|
|
try:
|
|
validate_cloud_url(body.server_url)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"Invalid server URL: {exc}",
|
|
) from exc
|
|
|
|
try:
|
|
credentials = _webdav_credentials(body)
|
|
backend = build_cloud_backend(body.provider, credentials)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"Invalid server URL: {exc}",
|
|
) from exc
|
|
|
|
try:
|
|
ok = await backend.health_check()
|
|
if not ok:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="Connection test failed — check server URL and credentials",
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"Connection test failed — check server URL and credentials: {exc}",
|
|
) from exc
|
|
|
|
credentials_enc = encrypt_credentials(_master_key(), str(current_user.id), credentials)
|
|
|
|
conn = await _insert_cloud_connection(session, current_user.id, body.provider, credentials_enc)
|
|
await session.flush()
|
|
|
|
_ip = get_client_ip(request)
|
|
await write_audit_log(
|
|
session,
|
|
event_type="cloud.connected",
|
|
user_id=current_user.id,
|
|
actor_id=current_user.id,
|
|
resource_id=conn.id,
|
|
ip_address=_ip,
|
|
metadata_={"provider": body.provider},
|
|
)
|
|
await session.commit()
|
|
await session.refresh(conn)
|
|
|
|
return CloudConnectionOut.model_validate(conn).model_dump()
|
|
|
|
|
|
@router.get("/connections")
|
|
@account_limiter.limit("100/minute")
|
|
async def list_connections(
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
) -> dict:
|
|
"""List the current user's cloud connections."""
|
|
request.state.current_user = current_user
|
|
result = await session.execute(
|
|
select(CloudConnection).where(CloudConnection.user_id == current_user.id)
|
|
)
|
|
return {"items": [_connection_response(conn) for conn in result.scalars().all()]}
|
|
|
|
|
|
@router.get("/connections/{connection_id}/config")
|
|
@account_limiter.limit("100/minute")
|
|
async def get_connection_config(
|
|
request: Request,
|
|
connection_id: uuid.UUID,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
) -> dict:
|
|
"""Return non-secret WebDAV/Nextcloud connection fields."""
|
|
request.state.current_user = current_user
|
|
conn = await _get_owned_connection(session, connection_id, current_user.id)
|
|
|
|
if conn.provider not in VALID_WEBDAV_PROVIDERS:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Connection config is only available for WebDAV/Nextcloud connections",
|
|
)
|
|
|
|
try:
|
|
credentials = _decrypt_connection(conn, current_user.id)
|
|
except Exception:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="Failed to decrypt connection credentials",
|
|
)
|
|
|
|
return {
|
|
"id": str(conn.id),
|
|
"provider": conn.provider,
|
|
"server_url": credentials.get("server_url", ""),
|
|
"connection_username": credentials.get("username", ""),
|
|
}
|
|
|
|
|
|
class WebDAVCredentialUpdateRequest(BaseModel):
|
|
server_url: Optional[str] = None
|
|
username: Optional[str] = None
|
|
password: Optional[str] = None # omit to keep existing
|
|
|
|
|
|
@router.put("/connections/{connection_id}/credentials", status_code=status.HTTP_200_OK)
|
|
@account_limiter.limit("100/minute")
|
|
async def update_webdav_credentials(
|
|
connection_id: uuid.UUID,
|
|
body: WebDAVCredentialUpdateRequest,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
) -> dict:
|
|
"""Update credentials for an existing WebDAV/Nextcloud connection (owner-scoped).
|
|
|
|
Only changes the fields provided. Omitting password preserves the existing secret.
|
|
Re-validates the URL (SSRF gate) and runs a health-check before committing.
|
|
T-12-06-01: owner scope via _get_owned_connection.
|
|
T-12-06-04: re-runs validate_cloud_url and health check before storing.
|
|
"""
|
|
request.state.current_user = current_user
|
|
conn = await _get_owned_connection(session, connection_id, current_user.id)
|
|
|
|
if conn.provider not in VALID_WEBDAV_PROVIDERS:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Credential update is only available for WebDAV/Nextcloud connections",
|
|
)
|
|
|
|
# Decrypt current credentials so we can merge with the submitted partial update
|
|
try:
|
|
current_creds = _decrypt_connection(conn, current_user.id)
|
|
except Exception:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="Failed to decrypt existing credentials",
|
|
)
|
|
|
|
merged = {
|
|
"server_url": body.server_url if body.server_url is not None else current_creds.get("server_url", ""),
|
|
"username": body.username if body.username is not None else current_creds.get("username", ""),
|
|
"password": body.password if body.password else current_creds.get("password", ""),
|
|
}
|
|
|
|
try:
|
|
validate_cloud_url(merged["server_url"])
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"Invalid server URL: {exc}",
|
|
) from exc
|
|
|
|
try:
|
|
backend = build_cloud_backend(conn.provider, merged)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"Invalid credentials: {exc}",
|
|
) from exc
|
|
|
|
try:
|
|
ok = await backend.health_check()
|
|
if not ok:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="Connection test failed — check server URL and credentials",
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"Connection test failed: {exc}",
|
|
) from exc
|
|
|
|
conn.credentials_enc = encrypt_credentials(_master_key(), str(current_user.id), merged)
|
|
conn.status = "ACTIVE"
|
|
|
|
_ip = get_client_ip(request)
|
|
await write_audit_log(
|
|
session,
|
|
event_type="cloud.credentials_updated",
|
|
user_id=current_user.id,
|
|
actor_id=current_user.id,
|
|
resource_id=conn.id,
|
|
ip_address=_ip,
|
|
metadata_={"provider": conn.provider},
|
|
)
|
|
await session.commit()
|
|
await session.refresh(conn)
|
|
|
|
return CloudConnectionOut.model_validate(conn).model_dump()
|
|
|
|
|
|
@router.patch("/connections/{connection_id}", status_code=status.HTTP_200_OK)
|
|
@account_limiter.limit("100/minute")
|
|
async def rename_connection(
|
|
connection_id: uuid.UUID,
|
|
body: ConnectionRenameRequest,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
) -> dict:
|
|
"""Rename a cloud connection's display name.
|
|
|
|
Only the display_name field is accepted — mass assignment prevention.
|
|
T-12-01: owner check via _get_owned_connection.
|
|
"""
|
|
request.state.current_user = current_user
|
|
conn = await _get_owned_connection(session, connection_id, current_user.id)
|
|
conn.display_name_override = body.display_name
|
|
await session.commit()
|
|
await session.refresh(conn)
|
|
return {"id": str(conn.id), "display_name": body.display_name}
|
|
|
|
|
|
@router.post("/connections/{connection_id}/reconnect", status_code=status.HTTP_200_OK)
|
|
@account_limiter.limit("100/minute")
|
|
async def reconnect_connection(
|
|
connection_id: uuid.UUID,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
) -> dict:
|
|
"""Reconnect an AUTH_FAILED connection in-place.
|
|
|
|
CONN-01: Patches the existing CloudConnection row — no new row created.
|
|
CONN-02: Re-encrypts credentials (refreshed if possible, original otherwise).
|
|
CONN-03: Response never exposes raw credentials, tokens, or provider URLs.
|
|
D-14: Cached cloud items are preserved as stale (not deleted).
|
|
|
|
Ownership verified via service layer (T-13-01).
|
|
"""
|
|
request.state.current_user = current_user
|
|
|
|
from services.cloud_operations import reconnect_connection as _svc_reconnect
|
|
from services.cloud_items import ConnectionNotFound
|
|
from services.cloud_cache import invalidate_provider_cache # lazy import
|
|
|
|
try:
|
|
result = await _svc_reconnect(
|
|
session,
|
|
connection_id=connection_id,
|
|
user_id=current_user.id,
|
|
master_key=_master_key(),
|
|
)
|
|
except ConnectionNotFound:
|
|
raise HTTPException(status_code=404, detail="Connection not found")
|
|
|
|
# Invalidate provider cache so the next browse is fresh (D-14)
|
|
invalidate_provider_cache(str(current_user.id), result["provider"])
|
|
|
|
_ip = get_client_ip(request)
|
|
await write_audit_log(
|
|
session,
|
|
event_type="cloud.reconnect",
|
|
user_id=current_user.id,
|
|
actor_id=current_user.id,
|
|
resource_id=connection_id,
|
|
ip_address=_ip,
|
|
metadata_={"provider": result["provider"], "connection_id": str(connection_id)},
|
|
)
|
|
# CR-05: The service committed the credential update; this commit persists the
|
|
# audit row that was written to the session after that commit. Without this,
|
|
# the audit row is rolled back when the session closes and the reconnect event
|
|
# is silently lost.
|
|
await session.commit()
|
|
|
|
return result
|
|
|
|
|
|
@router.get("/connections/{connection_id}/health", status_code=status.HTTP_200_OK)
|
|
@account_limiter.limit("100/minute")
|
|
async def get_connection_health(
|
|
connection_id: uuid.UUID,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
) -> dict:
|
|
"""Return connection health status.
|
|
|
|
D-12: Connection health is available explicitly (not inferred from browse).
|
|
Response includes 'status': 'healthy' | 'degraded' | 'auth_failed' | 'offline'.
|
|
Response never exposes credentials (T-13-02).
|
|
|
|
Ownership verified via service layer (T-13-01).
|
|
"""
|
|
request.state.current_user = current_user
|
|
|
|
from services.cloud_operations import get_connection_health as _svc_health
|
|
from services.cloud_items import ConnectionNotFound
|
|
|
|
try:
|
|
return await _svc_health(
|
|
session,
|
|
connection_id=connection_id,
|
|
user_id=current_user.id,
|
|
)
|
|
except ConnectionNotFound:
|
|
raise HTTPException(status_code=404, detail="Connection not found")
|
|
|
|
|
|
@router.post("/connections/{connection_id}/test", status_code=status.HTTP_200_OK)
|
|
@account_limiter.limit("100/minute")
|
|
async def test_connection(
|
|
connection_id: uuid.UUID,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
) -> dict:
|
|
"""Run an explicit connection health probe and update connection status.
|
|
|
|
D-13: Explicit Test action triggers a real health probe against the provider.
|
|
No probing on every folder navigation. Updates the connection status row.
|
|
Response includes 'status' field (same vocabulary as health endpoint).
|
|
Response never exposes credentials (T-13-02).
|
|
|
|
Ownership verified via service layer (T-13-01).
|
|
"""
|
|
request.state.current_user = current_user
|
|
|
|
from services.cloud_operations import test_connection as _svc_test
|
|
from services.cloud_items import ConnectionNotFound
|
|
|
|
try:
|
|
return await _svc_test(
|
|
session,
|
|
connection_id=connection_id,
|
|
user_id=current_user.id,
|
|
master_key=_master_key(),
|
|
)
|
|
except ConnectionNotFound:
|
|
raise HTTPException(status_code=404, detail="Connection not found")
|
|
|
|
|
|
@router.delete("/connections/{connection_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
@account_limiter.limit("100/minute")
|
|
async def delete_connection(
|
|
connection_id: uuid.UUID,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
) -> None:
|
|
"""Disconnect a cloud connection.
|
|
|
|
D-16: Removes credentials_enc and all connection-scoped CloudItems.
|
|
Provider files are NOT deleted — only cached metadata is cleaned up.
|
|
|
|
Ownership verified via service layer (T-13-01).
|
|
"""
|
|
request.state.current_user = current_user
|
|
|
|
# Resolve connection first to get provider for cache invalidation and audit
|
|
conn = await _get_owned_connection(session, connection_id, current_user.id)
|
|
provider = conn.provider
|
|
|
|
from services.cloud_cache import invalidate_provider_cache # lazy import
|
|
from services.cloud_operations import disconnect_connection as _svc_disconnect
|
|
from services.cloud_items import ConnectionNotFound
|
|
|
|
invalidate_provider_cache(str(current_user.id), provider)
|
|
|
|
_ip = get_client_ip(request)
|
|
|
|
await write_audit_log(
|
|
session,
|
|
event_type="cloud.disconnected",
|
|
user_id=current_user.id,
|
|
actor_id=current_user.id,
|
|
resource_id=conn.id,
|
|
ip_address=_ip,
|
|
metadata_={"provider": provider},
|
|
)
|
|
|
|
# Use service layer for explicit cascade delete (D-16, T-13-12)
|
|
try:
|
|
await _svc_disconnect(
|
|
session,
|
|
connection_id=connection_id,
|
|
user_id=current_user.id,
|
|
)
|
|
except ConnectionNotFound:
|
|
# Connection was already deleted between the ownership check and the
|
|
# service call — this is idempotent, return 204 normally
|
|
pass
|
|
|
|
|
|
@router.get("/folders/{provider}/{folder_id:path}")
|
|
@account_limiter.limit("100/minute")
|
|
async def list_cloud_folders(
|
|
request: Request,
|
|
provider: str,
|
|
folder_id: str,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
) -> dict:
|
|
"""List folder contents for a connected cloud provider (compatibility route).
|
|
|
|
This endpoint is the original provider-keyed browse route preserved for
|
|
backward compatibility. The canonical route is GET /connections/{id}/items.
|
|
"""
|
|
request.state.current_user = current_user
|
|
if provider not in VALID_CLOUD_PROVIDERS:
|
|
raise _unsupported_provider_error(provider, VALID_CLOUD_PROVIDERS)
|
|
|
|
conn = await _get_active_connection(session, current_user.id, provider)
|
|
credentials = _decrypt_connection(conn, current_user.id)
|
|
|
|
from services.cloud_cache import get_cloud_folders_cached # lazy import
|
|
|
|
if provider == "google_drive":
|
|
from api.cloud.browse import _fetch_google_drive_folders
|
|
fetcher = lambda: _fetch_google_drive_folders(credentials, folder_id)
|
|
|
|
elif provider == "onedrive":
|
|
from api.cloud.browse import _fetch_onedrive_folders
|
|
fetcher = lambda: _fetch_onedrive_folders(credentials, folder_id)
|
|
|
|
else:
|
|
from api.cloud.browse import _fetch_webdav_folders
|
|
fetcher = lambda: _fetch_webdav_folders(provider, credentials, folder_id)
|
|
|
|
items = await get_cloud_folders_cached(str(current_user.id), provider, folder_id, fetcher)
|
|
|
|
return {"items": items}
|