# Copyright (c) 2025 Patrick Motsch # All rights reserved. """Infomaniak Personal-Access-Token onboarding for data connections. Infomaniak does NOT support OAuth scopes for kDrive/kSuite data access. The user must create a Personal Access Token (PAT) at https://manager.infomaniak.com/v3/ng/accounts/token/list with the API scopes: - ``drive`` -> kDrive (active adapter) - ``workspace:calendar`` -> Calendar (active adapter) - ``workspace:contact`` -> Contacts (active adapter) - ``workspace:mail`` -> Mail (adapter pending; scope reserved) Validation strategy ------------------- The submit endpoint validates the PAT in two deterministic steps, each addressing one scope: 1. ``listAccessibleDrives(pat)`` -> ``GET /2/drive/init?with=drives`` proves the ``drive`` scope is on the PAT and -- as a side effect -- confirms the user has at least one accessible kDrive. This is the *only* listing endpoint that returns drives where the user has ``role: 'user'`` (the documented ``/2/drive?account_id=...`` listing is filtered to admin-only drives and would silently return ``[]`` for a standard kSuite member). 2. ``resolveOwnerIdentity(pat)`` -> PIM Calendar (preferred) or PIM Contacts (fallback) yields the user's display name + their kSuite account_id, used purely for connection labelling. This also proves that at least one of ``workspace:calendar`` / ``workspace:contact`` is on the PAT (the connection would otherwise be blank in the UI). Mail has no separate probe: its scope is recorded in ``grantedScopes`` so a future adapter can pick it up without re-issuing the token. """ from fastapi import APIRouter, HTTPException, Request, status, Depends, Path, Body import logging from typing import Dict, Any import hashlib from modules.interfaces.interfaceDbApp import getInterface from modules.datamodels.datamodelUam import AuthAuthority, User, ConnectionStatus, UserConnection from modules.datamodels.datamodelSecurity import Token, TokenPurpose from modules.auth import getCurrentUser, limiter from modules.shared.timeUtils import getUtcTimestamp, createExpirationTimestamp from modules.shared.i18nRegistry import apiRouteContext from modules.connectors.providerInfomaniak.connectorInfomaniak import ( resolveOwnerIdentity, listAccessibleDrives, InfomaniakIdentityError, ) routeApiMsg = apiRouteContext("routeSecurityInfomaniak") logger = logging.getLogger(__name__) # Infomaniak PATs do not expire unless the user sets an explicit lifetime in # the Manager (up to 30 years). We persist a 10-year horizon so the central # tokenStatus helper does not flag the connection as "no token". Mirrors # ClickUp. _INFOMANIAK_TOKEN_EXPIRES_IN_SEC = 10 * 365 * 24 * 3600 router = APIRouter( prefix="/api/infomaniak", tags=["Security Infomaniak"], responses={ 404: {"description": "Not found"}, 400: {"description": "Bad request"}, 401: {"description": "Unauthorized"}, 500: {"description": "Internal server error"}, }, ) @router.post("/connections/{connectionId}/token") @limiter.limit("10/minute") async def submit_infomaniak_token( request: Request, connectionId: str = Path(..., description="UserConnection id"), body: Dict[str, Any] = Body(..., description="{ 'token': '' }"), currentUser: User = Depends(getCurrentUser), ) -> Dict[str, Any]: """Validate and persist an Infomaniak Personal Access Token (PAT). Body: { "token": "" } Validation order (both must succeed before persisting): 1. ``listAccessibleDrives(pat)`` -> proves the ``drive`` scope is on the PAT and confirms the user can see at least one kDrive (uses ``/2/drive/init?with=drives`` so users with ``role: 'user'`` are also covered). 2. ``resolveOwnerIdentity(pat)`` -> display name + kSuite account_id for the connection UI label (proves at least one of ``workspace:calendar`` / ``workspace:contact`` is present). No PAT-derived data is stored as adapter state -- both the drive list and the owner identity are re-resolved lazily by the adapters at request time. """ pat = (body or {}).get("token") if not isinstance(pat, str) or not pat.strip(): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=routeApiMsg("Missing 'token' in request body"), ) pat = pat.strip() interface = getInterface(currentUser) connection = None for conn in interface.getUserConnections(currentUser.id): if conn.id == connectionId and conn.authority == AuthAuthority.INFOMANIAK: connection = conn break if not connection: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=routeApiMsg("Infomaniak connection not found"), ) try: drives = await listAccessibleDrives(pat) except InfomaniakIdentityError as e: logger.warning( f"Infomaniak token submit for connection {connectionId} could not " f"list drives: {e}" ) raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=routeApiMsg( "Token rejected by Infomaniak (missing scope 'drive'). " "Required scopes: 'drive' (kDrive) and " "'workspace:calendar' (or 'workspace:contact'). Mail " "scope 'workspace:mail' is reserved." ), ) try: identity = await resolveOwnerIdentity(pat) except InfomaniakIdentityError as e: logger.warning( f"Infomaniak token submit for connection {connectionId} could not " f"resolve owner identity: {e}" ) raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=routeApiMsg( "Could not derive your Infomaniak account from the token. " "Please ensure the PAT carries 'workspace:calendar' or " "'workspace:contact' so we can identify your account." ), ) tokenFingerprint = "pat-" + hashlib.sha256(pat.encode("utf-8")).hexdigest()[:8] username = identity["displayName"] or f"infomaniak-{tokenFingerprint}" expiresAt = createExpirationTimestamp(_INFOMANIAK_TOKEN_EXPIRES_IN_SEC) try: connection.status = ConnectionStatus.ACTIVE connection.lastChecked = getUtcTimestamp() connection.expiresAt = expiresAt connection.externalId = str(identity["accountId"]) connection.externalUsername = username connection.grantedScopes = [ "drive", "workspace:mail", "workspace:calendar", "workspace:contact", ] interface.db.recordModify(UserConnection, connectionId, connection.model_dump()) token = Token( userId=currentUser.id, authority=AuthAuthority.INFOMANIAK, connectionId=connectionId, tokenPurpose=TokenPurpose.DATA_CONNECTION, tokenAccess=pat, tokenRefresh=None, tokenType="bearer", expiresAt=expiresAt, createdAt=getUtcTimestamp(), ) interface.saveConnectionToken(token) driveSummary = [ {"id": d.get("id"), "name": d.get("name"), "role": d.get("role")} for d in drives ] logger.info( f"Infomaniak PAT stored for connection {connectionId} " f"(user {currentUser.id}, externalUsername={username}, " f"kSuiteAccountId={identity['accountId']}, " f"accessibleDrives={driveSummary})" ) return { "id": connection.id, "status": "connected", "type": "infomaniak", "externalUsername": username, "externalEmail": None, "lastChecked": connection.lastChecked, } except HTTPException: raise except Exception as e: logger.error( f"Error persisting Infomaniak token for connection {connectionId}: {e}", exc_info=True, ) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=routeApiMsg("Failed to store Infomaniak token"), )