Coverage for apps/integrations/admin_api.py: 61%
112 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-04 17:24 +0000
1"""Integrations admin router (mounted at /api/admin).
3Org-integration writes are gated by the provider category's credential permission
4(EDIT_CREDIT/AUS/DOCMAGIC_CREDENTIALS), resolved dynamically from the request body; a user's
5own credentials need only authentication, another user's need MANAGE_USER_ACCOUNT; lenders need
6MANAGE_USER_ACCOUNT. Secrets are accepted on write and never returned (design D2).
7"""
9from __future__ import annotations
11from ninja import Query, Router
13from apps.accounts.auth import StaffAuth, requires
14from apps.accounts.models import OrgUnit, User
15from apps.integrations.catalog import PROVIDERS, get_provider
16from apps.integrations.encryption import encrypt_secret
17from apps.integrations.models import Lender, OrgIntegration, UserCredential
18from apps.integrations.schemas import (
19 CatalogEntryOut,
20 LenderCreate,
21 LenderOut,
22 LenderPatch,
23 OrgIntegrationIn,
24 OrgIntegrationOut,
25 UserCredentialIn,
26 UserCredentialOut,
27)
28from domain.result import AppError, DomainError
30router = Router(tags=["admin-integrations"])
32_CATEGORY_PERM = {
33 "credit": "EDIT_CREDIT_CREDENTIALS",
34 "aus": "EDIT_AUS_CREDENTIALS",
35 "docmagic": "EDIT_DOCMAGIC_CREDENTIALS",
36}
37_ANY_CRED_PERM = (
38 "EDIT_CREDIT_CREDENTIALS",
39 "EDIT_AUS_CREDENTIALS",
40 "EDIT_DOCMAGIC_CREDENTIALS",
41 "MANAGE_USER_ACCOUNT",
42)
45def _require(principal, perm: str) -> None:
46 if not principal.has(perm):
47 raise DomainError([AppError("FORBIDDEN", f"Missing permission: {perm}")])
50def _perm_for_provider(provider_code: str) -> str:
51 provider = get_provider(provider_code)
52 if provider is None: 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true
53 raise DomainError([AppError("VALIDATION", f"Unknown provider: {provider_code}")])
54 return _CATEGORY_PERM.get(provider.category, "MANAGE_USER_ACCOUNT")
57def _org_out(oi: OrgIntegration) -> dict:
58 return {
59 "id": oi.id,
60 "org_unit_id": oi.org_unit_id,
61 "provider": oi.provider,
62 "enabled": oi.enabled,
63 "config": oi.config,
64 "has_secret": bool(oi.secret_enc),
65 }
68def _cred_out(uc: UserCredential) -> dict:
69 return {
70 "id": uc.id,
71 "user_id": uc.user_id,
72 "provider": uc.provider,
73 "config": uc.config,
74 "has_secret": bool(uc.secret_enc),
75 }
78# --- Catalog -------------------------------------------------------------------
79@router.get("/integrations/catalog", response=list[CatalogEntryOut], auth=StaffAuth())
80def catalog(request):
81 return [
82 {"code": p.code, "category": p.category, "scope": p.scope, "label": p.label}
83 for p in PROVIDERS.values()
84 ]
87# --- Org integrations ----------------------------------------------------------
88@router.get("/org-integrations", response=list[OrgIntegrationOut], auth=StaffAuth())
89def list_org_integrations(request, org_unit_id: int | None = Query(None)):
90 if not any(request.principal.has(p) for p in _ANY_CRED_PERM): 90 ↛ 91line 90 didn't jump to line 91 because the condition on line 90 was never true
91 raise DomainError([AppError("FORBIDDEN", "Missing a credential-management permission")])
92 qs = OrgIntegration.objects.all()
93 if org_unit_id is not None: 93 ↛ 95line 93 didn't jump to line 95 because the condition on line 93 was always true
94 qs = qs.filter(org_unit_id=org_unit_id)
95 return [_org_out(oi) for oi in qs]
98@router.put("/org-integrations", response=OrgIntegrationOut, auth=StaffAuth())
99def upsert_org_integration(request, payload: OrgIntegrationIn):
100 _require(request.principal, _perm_for_provider(payload.provider))
101 org = OrgUnit.objects.filter(pk=payload.org_unit_id).first()
102 if org is None: 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true
103 raise DomainError([AppError("NOT_FOUND", "Org unit not found")])
104 oi, _ = OrgIntegration.objects.get_or_create(org_unit=org, provider=payload.provider)
105 oi.enabled = payload.enabled
106 if payload.config is not None: 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true
107 oi.config = payload.config
108 if payload.secret is not None: 108 ↛ 110line 108 didn't jump to line 110 because the condition on line 108 was always true
109 oi.secret_enc = encrypt_secret(payload.secret)
110 oi.save()
111 return _org_out(oi)
114@router.delete("/org-integrations/{item_id}", response={204: None}, auth=StaffAuth())
115def delete_org_integration(request, item_id: int):
116 oi = OrgIntegration.objects.filter(pk=item_id).first()
117 if oi is None:
118 raise DomainError([AppError("NOT_FOUND", "Integration not found")])
119 _require(request.principal, _perm_for_provider(oi.provider))
120 oi.delete()
121 return 204, None
124# --- User credentials: self ----------------------------------------------------
125@router.get("/me/credentials", response=list[UserCredentialOut], auth=StaffAuth())
126def list_my_credentials(request):
127 return [_cred_out(c) for c in UserCredential.objects.filter(user_id=request.principal.user_id)]
130@router.put("/me/credentials", response=UserCredentialOut, auth=StaffAuth())
131def upsert_my_credential(request, payload: UserCredentialIn):
132 if get_provider(payload.provider) is None: 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true
133 raise DomainError([AppError("VALIDATION", f"Unknown provider: {payload.provider}")])
134 uc, _ = UserCredential.objects.get_or_create(
135 user_id=request.principal.user_id, provider=payload.provider
136 )
137 if payload.config is not None: 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true
138 uc.config = payload.config
139 if payload.secret is not None: 139 ↛ 141line 139 didn't jump to line 141 because the condition on line 139 was always true
140 uc.secret_enc = encrypt_secret(payload.secret)
141 uc.save()
142 return _cred_out(uc)
145@router.delete("/me/credentials/{item_id}", response={204: None}, auth=StaffAuth())
146def delete_my_credential(request, item_id: int):
147 uc = UserCredential.objects.filter(pk=item_id, user_id=request.principal.user_id).first()
148 if uc is None:
149 raise DomainError([AppError("NOT_FOUND", "Credential not found")])
150 uc.delete()
151 return 204, None
154# --- User credentials: admin for another user ----------------------------------
155@router.get(
156 "/users/{user_id}/credentials",
157 response=list[UserCredentialOut],
158 auth=requires("MANAGE_USER_ACCOUNT"),
159)
160def list_user_credentials(request, user_id: int):
161 if not User.objects.filter(pk=user_id).exists():
162 raise DomainError([AppError("NOT_FOUND", "User not found")])
163 return [_cred_out(c) for c in UserCredential.objects.filter(user_id=user_id)]
166# --- Lenders -------------------------------------------------------------------
167@router.get("/lenders", response=list[LenderOut], auth=requires("MANAGE_USER_ACCOUNT"))
168def list_lenders(request, org_unit_id: int | None = Query(None)):
169 qs = Lender.objects.all()
170 if org_unit_id is not None: 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true
171 qs = qs.filter(org_unit_id=org_unit_id)
172 return list(qs)
175@router.post("/lenders", response={201: LenderOut}, auth=requires("MANAGE_USER_ACCOUNT"))
176def create_lender(request, payload: LenderCreate):
177 data = payload.dict()
178 org_unit_id = data.pop("org_unit_id")
179 if not OrgUnit.objects.filter(pk=org_unit_id).exists(): 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true
180 raise DomainError([AppError("NOT_FOUND", "Org unit not found")])
181 return 201, Lender.objects.create(org_unit_id=org_unit_id, **data)
184@router.patch("/lenders/{item_id}", response=LenderOut, auth=requires("MANAGE_USER_ACCOUNT"))
185def update_lender(request, item_id: int, payload: LenderPatch):
186 lender = Lender.objects.filter(pk=item_id).first()
187 if lender is None:
188 raise DomainError([AppError("NOT_FOUND", "Lender not found")])
189 for field_name, value in payload.dict(exclude_unset=True).items():
190 setattr(lender, field_name, value)
191 lender.save()
192 return lender
195@router.delete("/lenders/{item_id}", response={204: None}, auth=requires("MANAGE_USER_ACCOUNT"))
196def delete_lender(request, item_id: int):
197 lender = Lender.objects.filter(pk=item_id).first()
198 if lender is None:
199 raise DomainError([AppError("NOT_FOUND", "Lender not found")])
200 lender.delete()
201 return 204, None