PowerOn/gateway
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fixes infomaniac different than in doc

Browse source
This commit is contained in:
ValueOn AG 2026-04-29 00:57:28 +02:00
parent b405cebdec
commit 9816f13ae9
2 changed files with 84 additions and 166 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

133
modules/connectors/providerInfomaniak/connectorInfomaniak.py
View file

@ -187,76 +187,67 @@ async def resolveOwnerIdentity(token: str) -> InfomaniakOwnerIdentity:
)
async def resolveAccessibleAccountIds(token: str) -> List[int]:
"""Return every Infomaniak account_id the PAT has access to.
async def listAccessibleDrives(token: str) -> List[Dict[str, Any]]:
"""Return every kDrive the PAT can reach (admin OR user role).
Hits ``GET /1/accounts`` -- the only Infomaniak endpoint that lists
*all* account_ids of a token in one call. Requires the PAT scope
``accounts`` (Infomaniak responds 403 with
``code: 'all_scopes', context: {scopes: ['accounts']}`` if missing).
Hits ``GET /2/drive/init?with=drives`` -- the only PAT-friendly
endpoint that enumerates a user's drives **independently of the
Drive-Manager admin role**. The plain ``/2/drive?account_id=...``
listing is filtered to drives where the caller is an admin and
therefore returns an empty array for everyone with ``role: user``,
even though the same user can read/write the drive's files via
``/2/drive/{driveId}/...``.
The kSuite account_id from PIM (``resolveOwnerIdentity``) is **not**
sufficient for kDrive: a standalone or free-tier kDrive lives on a
different account_id than its kSuite counterpart. ``/2/drive`` is
queried per account_id, so we resolve them all here and union the
drive listings in :class:`KdriveAdapter`.
The endpoint requires only the ``drive`` PAT scope -- no
``accounts``, no ``user_info``, no admin permission. Each entry
matches the shape documented for ``GET /2/drive/{drive_id}``
(``id``, ``name``, ``account_id``, ``role``, ...).
Raises :class:`InfomaniakIdentityError` when the PAT does not carry
the ``accounts`` scope or the response is malformed.
the ``drive`` scope or the response is malformed.
"""
payload = await _infomaniakGet(token, "/1/accounts")
payload = await _infomaniakGet(token, "/2/drive/init?with=drives")
if isinstance(payload, dict) and payload.get("error"):
raise InfomaniakIdentityError(
"Could not list Infomaniak accounts. The PAT must carry the "
"'accounts' scope so kDrive can discover the owning account "
f"(/1/accounts said: {payload['error']})."
"Could not list Infomaniak kDrives. The PAT must carry the "
f"'drive' scope (/2/drive/init said: {payload['error']})."
)
data = _unwrapData(payload)
if not isinstance(data, list):
if not isinstance(data, dict):
raise InfomaniakIdentityError(
"Unexpected /1/accounts response shape (expected a list)."
"Unexpected /2/drive/init response shape (expected an object)."
)
accountIds: List[int] = []
for entry in data:
if not isinstance(entry, dict):
continue
accountId = entry.get("id")
if isinstance(accountId, int):
accountIds.append(accountId)
if not accountIds:
drives = data.get("drives") or []
if not isinstance(drives, list):
raise InfomaniakIdentityError(
"/1/accounts returned no accounts -- the PAT cannot reach any "
"Infomaniak account."
"Unexpected /2/drive/init response: 'drives' is not a list."
)
return accountIds
return [d for d in drives if isinstance(d, dict) and d.get("id")]
class KdriveAdapter(ServiceAdapter):
"""kDrive ServiceAdapter -- browse drives, folders, files within all
accounts the PAT can reach.
"""kDrive ServiceAdapter -- browse drives, folders, files.
Infomaniak's ``/2/drive`` listing endpoint requires the integer
``account_id`` of the *drive-owning* account as a query arg. A user
may own kDrives in several accounts (typically a kSuite account
plus a standalone / free-tier kDrive account), and the kSuite
account_id from PIM does **not** cover the standalone case.
Drive enumeration goes through :func:`listAccessibleDrives` which
calls ``/2/drive/init?with=drives``. That endpoint returns every
drive the PAT can read regardless of the Drive-Manager admin role
-- unlike the documented ``/2/drive?account_id=...`` listing which
silently returns an empty array for users with ``role: 'user'``
(the most common case for kSuite members).
The only PAT-friendly way to enumerate every account_id is
:func:`resolveAccessibleAccountIds` (``GET /1/accounts`` with the
``accounts`` scope). This adapter therefore resolves the full
account list once per instance and unions the ``/2/drive`` listing
across all of them in :meth:`_listDrives`.
The drive list is cached on the adapter instance so each browse
pays for one ``/2/drive/init`` call at most.
"""
def __init__(self, accessToken: str):
self._token = accessToken
self._accountIds: Optional[List[int]] = None
self._drives: Optional[List[Dict[str, Any]]] = None
async def _ensureAccountIds(self) -> List[int]:
if self._accountIds is not None:
return self._accountIds
self._accountIds = await resolveAccessibleAccountIds(self._token)
return self._accountIds
async def _ensureDrives(self) -> List[Dict[str, Any]]:
if self._drives is not None:
return self._drives
self._drives = await listAccessibleDrives(self._token)
return self._drives
async def browse(
self,
@ -278,41 +269,23 @@ class KdriveAdapter(ServiceAdapter):
return await self._listChildren(driveId, fileId=fileId, limit=limit)
async def _listDrives(self) -> List[ExternalEntry]:
accountIds = await self._ensureAccountIds()
seen: set = set()
drives = await self._ensureDrives()
entries: List[ExternalEntry] = []
for accountId in accountIds:
result = await _infomaniakGet(
self._token, f"/2/drive?account_id={accountId}"
)
if isinstance(result, dict) and result.get("error"):
logger.warning(
f"kDrive list-drives for account {accountId} failed: {result['error']}"
)
for drive in drives:
driveId = str(drive.get("id", ""))
if not driveId:
continue
data = _unwrapData(result)
drives: List[Dict[str, Any]]
if isinstance(data, list):
drives = [d for d in data if isinstance(d, dict)]
elif isinstance(data, dict):
drives = data.get("drives", {}).get("accounts", []) or []
else:
drives = []
for drive in drives:
driveId = str(drive.get("id", ""))
if not driveId or driveId in seen:
continue
seen.add(driveId)
entries.append(ExternalEntry(
name=drive.get("name") or driveId,
path=f"/{driveId}",
isFolder=True,
metadata={
"id": driveId,
"kind": "drive",
"accountId": accountId,
},
))
entries.append(ExternalEntry(
name=drive.get("name") or driveId,
path=f"/{driveId}",
isFolder=True,
metadata={
"id": driveId,
"kind": "drive",
"accountId": drive.get("account_id"),
"role": drive.get("role"),
},
))
return entries
async def _listChildren(

117
modules/routes/routeSecurityInfomaniak.py
View file

@ -7,7 +7,6 @@ The user must create a Personal Access Token (PAT) at
https://manager.infomaniak.com/v3/ng/accounts/token/list with the API
scopes:
- ``accounts`` -> account discovery (REQUIRED for kDrive)
- ``drive`` -> kDrive (active adapter)
- ``workspace:calendar`` -> Calendar (active adapter)
- ``workspace:contact`` -> Contacts (active adapter)
@ -15,14 +14,16 @@ scopes:
Validation strategy
-------------------
The submit endpoint validates the PAT in three deterministic steps,
each addressing exactly one scope:
The submit endpoint validates the PAT in two deterministic steps,
each addressing one scope:
1. ``resolveAccessibleAccountIds(pat)`` -> ``GET /1/accounts`` proves
the ``accounts`` scope is on the PAT. Without this scope, kDrive
cannot enumerate the owning account_ids (a standalone or free-tier
kDrive lives on a *different* account_id than its kSuite
counterpart, so the kSuite account_id from PIM is not enough).
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
@ -30,10 +31,6 @@ each addressing exactly one scope:
that at least one of ``workspace:calendar`` / ``workspace:contact``
is on the PAT (the connection would otherwise be blank in the UI).
3. ``GET /2/drive?account_id={firstAccountId}`` is the final scope
probe -- 200 means the ``drive`` scope is present. 401/403 means
the scope is missing.
Mail has no separate probe: its scope is recorded in ``grantedScopes``
so a future adapter can pick it up without re-issuing the token.
"""
@ -42,7 +39,6 @@ from fastapi import APIRouter, HTTPException, Request, status, Depends, Path, Bo
import logging
from typing import Dict, Any
import hashlib
import httpx
from modules.interfaces.interfaceDbApp import getInterface
from modules.datamodels.datamodelUam import AuthAuthority, User, ConnectionStatus, UserConnection
@ -52,7 +48,7 @@ from modules.shared.timeUtils import getUtcTimestamp, createExpirationTimestamp
from modules.shared.i18nRegistry import apiRouteContext
from modules.connectors.providerInfomaniak.connectorInfomaniak import (
resolveOwnerIdentity,
resolveAccessibleAccountIds,
listAccessibleDrives,
InfomaniakIdentityError,
)
@ -60,8 +56,6 @@ routeApiMsg = apiRouteContext("routeSecurityInfomaniak")
logger = logging.getLogger(__name__)
INFOMANIAK_API_BASE = "https://api.infomaniak.com"
# 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
@ -81,55 +75,6 @@ router = APIRouter(
)
async def _probeDriveScope(client: httpx.AsyncClient, pat: str, accountId: int) -> None:
"""Confirm the ``drive`` scope is on the PAT.
Issues ``GET /2/drive?account_id={accountId}`` -- a clean 200 means
both the ``drive`` scope is present and the resolved ``account_id``
is correct. 401/403 means the scope is missing; anything else means
Infomaniak is misbehaving and we refuse to persist.
"""
url = f"{INFOMANIAK_API_BASE}/2/drive?account_id={accountId}"
try:
resp = await client.get(
url,
headers={"Authorization": f"Bearer {pat}", "Accept": "application/json"},
)
except httpx.HTTPError as e:
logger.error(f"Infomaniak drive-probe network error ({url}): {e}")
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=routeApiMsg("Could not reach Infomaniak to validate the token"),
)
if resp.status_code == 200:
return
if resp.status_code in (401, 403):
logger.warning(
f"Infomaniak drive-probe rejected PAT ({url}): "
f"{resp.status_code} {resp.text[:200]}"
)
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'). Recommended for upcoming "
"services: 'workspace:mail'."
),
)
logger.error(
f"Infomaniak drive-probe unexpected response ({url}): "
f"{resp.status_code} {resp.text[:200]}"
)
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=routeApiMsg(
"Infomaniak drive-probe returned an unexpected response."
),
)
@router.post("/connections/{connectionId}/token")
@limiter.limit("10/minute")
async def submit_infomaniak_token(
@ -143,18 +88,18 @@ async def submit_infomaniak_token(
Body:
{ "token": "<personal-access-token from Infomaniak Manager>" }
Validation order (all three must succeed before persisting):
1. ``resolveAccessibleAccountIds(pat)`` -> proves the
``accounts`` scope is on the PAT (required for kDrive
account discovery).
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.
3. ``/2/drive?account_id=<first>`` -> proves the ``drive``
scope is on the PAT.
account_id for the connection UI label (proves at least one
of ``workspace:calendar`` / ``workspace:contact`` is present).
No data derived from the PAT is stored as adapter state -- both
account list and owner identity are re-resolved lazily by the
adapters at request time.
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():
@ -177,19 +122,19 @@ async def submit_infomaniak_token(
)
try:
accountIds = await resolveAccessibleAccountIds(pat)
drives = await listAccessibleDrives(pat)
except InfomaniakIdentityError as e:
logger.warning(
f"Infomaniak token submit for connection {connectionId} could not "
f"list accounts: {e}"
f"list drives: {e}"
)
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=routeApiMsg(
"Token rejected by Infomaniak (missing scope 'accounts'). "
"kDrive needs the 'accounts' scope to discover the owning "
"Infomaniak account. Required scopes: 'accounts', 'drive', "
"'workspace:calendar', 'workspace:contact'."
"Token rejected by Infomaniak (missing scope 'drive'). "
"Required scopes: 'drive' (kDrive) and "
"'workspace:calendar' (or 'workspace:contact'). Mail "
"scope 'workspace:mail' is reserved."
),
)
@ -209,9 +154,6 @@ async def submit_infomaniak_token(
),
)
async with httpx.AsyncClient(timeout=15.0, follow_redirects=False) as client:
await _probeDriveScope(client, pat, accountIds[0])
tokenFingerprint = "pat-" + hashlib.sha256(pat.encode("utf-8")).hexdigest()[:8]
username = identity["displayName"] or f"infomaniak-{tokenFingerprint}"
expiresAt = createExpirationTimestamp(_INFOMANIAK_TOKEN_EXPIRES_IN_SEC)
@ -223,7 +165,6 @@ async def submit_infomaniak_token(
connection.externalId = str(identity["accountId"])
connection.externalUsername = username
connection.grantedScopes = [
"accounts",
"drive",
"workspace:mail",
"workspace:calendar",
@ -244,11 +185,15 @@ async def submit_infomaniak_token(
)
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"accessibleAccounts={accountIds})"
f"accessibleDrives={driveSummary})"
)
return {