feat(integrations): add Dropbox connection test and global-credential sharing

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>

app/api/dropbox.py

@@ -23,6 +23,81 @@ logger = logging.getLogger(__name__)
 router = APIRouter()
 
 
+@router.get("/dropbox/global-authorize-url")
+@require_login
+async def dropbox_global_authorize_url(request: Request):
+    """Return the Dropbox OAuth authorization URL using the global app credentials.
+
+    This endpoint is used when ``DROPBOX_ALLOW_GLOBAL_CREDENTIALS_FOR_INTEGRATIONS``
+    is enabled so that users can authorize their personal Dropbox integration without
+    needing to supply their own app key/secret.  Only the public ``app_key`` is
+    embedded in the URL; the ``app_secret`` is never sent to the browser.
+    """
+    if not settings.dropbox_allow_global_credentials_for_integrations:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Global credentials for integrations are not enabled",
+        )
+    if not settings.dropbox_app_key or not settings.dropbox_app_secret:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="Global Dropbox credentials are not configured",
+        )
+    redirect_uri = str(request.base_url).rstrip("/") + "/dropbox-callback"
+    authorize_url = (
+        "https://www.dropbox.com/oauth2/authorize"
+        f"?client_id={settings.dropbox_app_key}"
+        "&response_type=code"
+        "&token_access_type=offline"
+        f"&redirect_uri={redirect_uri}"
+    )
+    return {"authorize_url": authorize_url}
+
+
+@router.post("/dropbox/exchange-token-global")
+@require_login
+async def exchange_dropbox_token_global(
+    request: Request,
+    code: Annotated[str, Form(...)],
+    redirect_uri: Annotated[str, Form(...)],
+):
+    """Exchange an authorization code using the global Dropbox app credentials.
+
+    Used when ``DROPBOX_ALLOW_GLOBAL_CREDENTIALS_FOR_INTEGRATIONS`` is enabled so
+    that the ``app_secret`` is never exposed to the browser.  Only the OAuth code
+    and redirect URI need to be supplied by the client.
+    """
+    if not settings.dropbox_allow_global_credentials_for_integrations:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Global credentials for integrations are not enabled",
+        )
+    if not settings.dropbox_app_key or not settings.dropbox_app_secret:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="Global Dropbox credentials are not configured",
+        )
+
+    token_url = "https://api.dropboxapi.com/oauth2/token"
+    payload = {
+        "client_id": settings.dropbox_app_key,
+        "client_secret": settings.dropbox_app_secret,
+        "code": code,
+        "redirect_uri": redirect_uri,
+        "grant_type": "authorization_code",
+    }
+
+    token_data = exchange_oauth_token(provider_name="Dropbox", token_url=token_url, payload=payload)
+
+    return {
+        "refresh_token": token_data["refresh_token"],
+        "access_token": token_data["access_token"],
+        "expires_in": token_data.get("expires_in", 14400),
+        # Return the public app_key so the callback can store it in the integration
+        "app_key": settings.dropbox_app_key,
+    }
+
+
 @router.post("/dropbox/exchange-token")
 @require_login
 async def exchange_dropbox_token(

app/api/integrations.py

@@ -32,6 +32,20 @@ from app.utils.encryption import decrypt_value, encrypt_value
 from app.utils.subscription import get_tier, get_user_tier_id
 from app.utils.user_scope import get_current_owner_id
 
+# Optional Dropbox SDK — imported at module level so tests can patch it cleanly.
+try:
+    import dropbox as dbx_lib
+    from dropbox.exceptions import AuthError as _DropboxAuthError
+    from dropbox.exceptions import BadInputError as _DropboxBadInputError
+except ImportError:  # pragma: no cover
+    dbx_lib = None  # type: ignore[assignment]
+
+    class _DropboxAuthError(Exception):  # type: ignore[no-redef]
+        """Stub — only used when the dropbox package is missing."""
+
+    class _DropboxBadInputError(Exception):  # type: ignore[no-redef]
+        """Stub — only used when the dropbox package is missing."""
+
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/integrations", tags=["integrations"])
 
@@ -550,6 +564,44 @@ def _test_s3_connection(config: dict[str, Any] | None, credentials: dict[str, An
         return {"success": False, "message": "S3 connection failed"}
 
 
+def _test_dropbox_connection(config: dict[str, Any] | None, credentials: dict[str, Any] | None) -> dict[str, Any]:
+    """Test a Dropbox connection by verifying OAuth credentials via the Dropbox API."""
+    if dbx_lib is None:
+        return {"success": False, "message": "dropbox package is not installed"}  # pragma: no cover
+
+    creds = credentials or {}
+    app_key = creds.get("app_key", "")
+    app_secret = creds.get("app_secret", "")
+    refresh_token = creds.get("refresh_token", "")
+
+    if not refresh_token:
+        return {"success": False, "message": "Missing required credential: refresh_token"}
+    if not app_key or not app_secret:
+        return {"success": False, "message": "Missing required credentials: app_key and app_secret"}
+
+    try:
+        dbx = dbx_lib.Dropbox(
+            app_key=app_key,
+            app_secret=app_secret,
+            oauth2_refresh_token=refresh_token,
+        )
+        account = dbx.users_get_current_account()
+        display_name = getattr(account, "name", None)
+        name_str = ""
+        if display_name:
+            name_str = f" ({getattr(display_name, 'display_name', '') or ''})"
+        return {"success": True, "message": f"Dropbox connection successful{name_str}"}
+    except _DropboxAuthError as exc:
+        logger.warning("Dropbox auth error: %s", exc)
+        return {"success": False, "message": "Dropbox authentication failed — check app_key, app_secret, and refresh_token"}
+    except _DropboxBadInputError as exc:
+        logger.warning("Dropbox bad input error: %s", exc)
+        return {"success": False, "message": "Dropbox connection failed — invalid credentials format"}
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("Dropbox connection error: %s", exc)
+        return {"success": False, "message": "Dropbox connection failed — check credentials and network connectivity"}
+
+
 def _test_webdav_connection(config: dict[str, Any] | None, credentials: dict[str, Any] | None) -> dict[str, Any]:
     """Test a WebDAV/Nextcloud connection by issuing an HTTP PROPFIND."""
     import urllib.request
@@ -596,6 +648,7 @@ def _test_webdav_connection(config: dict[str, Any] | None, credentials: dict[str
 
 
 _CONNECTION_TESTERS: dict[str, Any] = {
+    IntegrationType.DROPBOX: _test_dropbox_connection,
     IntegrationType.IMAP: _test_imap_connection,
     IntegrationType.S3: _test_s3_connection,
     IntegrationType.WEBDAV: _test_webdav_connection,