Coverage for apps/integrations/settings_api.py: 75%
124 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-04 14:06 +0000
1"""Provider settings and credentials router (mounted at /api/settings).
3Implements:
4 GET /api/settings/providers -> list[ProviderDef]
5 GET /api/settings/{org_unit}/credentials/{provider} -> dict[str, str] (masked secrets)
6 PUT /api/settings/{org_unit}/credentials/{provider} -> dict[str, str] (masked secrets)
7"""
9from __future__ import annotations
11import json
12from dataclasses import dataclass
13from typing import Any
15from ninja import Body, Router
17from apps.accounts.auth import StaffAuth
18from apps.accounts.models import OrgUnit
19from apps.integrations.encryption import decrypt_secret, encrypt_secret
20from apps.integrations.models import OrgIntegration
21from apps.integrations.settings_schemas import ProviderDefOut
22from domain.result import AppError, DomainError
24router = Router(tags=["settings-integrations"])
27@dataclass(frozen=True)
28class CredentialField:
29 key: str
30 label: str
31 secret: bool
32 required: bool
35@dataclass(frozen=True)
36class ProviderDefinition:
37 provider: str
38 label: str
39 edit_permission: str
40 fields: tuple[CredentialField, ...]
43SETTINGS_PROVIDERS: tuple[ProviderDefinition, ...] = (
44 ProviderDefinition(
45 provider="docmagic",
46 label="DocMagic",
47 edit_permission="EDIT_DOCMAGIC_CREDENTIALS",
48 fields=(
49 CredentialField(key="accountId", label="Account ID", secret=False, required=True),
50 CredentialField(key="username", label="Username", secret=False, required=True),
51 CredentialField(key="password", label="Password", secret=True, required=True),
52 ),
53 ),
54 ProviderDefinition(
55 provider="du",
56 label="Fannie Mae Desktop Underwriter",
57 edit_permission="EDIT_AUS_CREDENTIALS",
58 fields=(
59 CredentialField(key="institutionId", label="DO/DU Institution ID", secret=False, required=True),
60 CredentialField(key="username", label="Username", secret=False, required=True),
61 CredentialField(key="password", label="Password", secret=True, required=True),
62 ),
63 ),
64 ProviderDefinition(
65 provider="lpa",
66 label="Freddie Mac Loan Product Advisor",
67 edit_permission="EDIT_AUS_CREDENTIALS",
68 fields=(
69 CredentialField(key="sellerServicerNumber", label="Seller/Servicer #", secret=False, required=True),
70 CredentialField(key="username", label="Username", secret=False, required=True),
71 CredentialField(key="password", label="Password", secret=True, required=True),
72 ),
73 ),
74 ProviderDefinition(
75 provider="credit",
76 label="Credit (MeridianLink / CRA)",
77 edit_permission="EDIT_CREDIT_CREDENTIALS",
78 fields=(
79 CredentialField(key="craId", label="CRA / Company ID", secret=False, required=True),
80 CredentialField(key="username", label="Username", secret=False, required=True),
81 CredentialField(key="password", label="Password", secret=True, required=True),
82 ),
83 ),
84 ProviderDefinition(
85 provider="lender",
86 label="Lender TPO / Pricing",
87 edit_permission="MANAGE_USER_ACCOUNT",
88 fields=(
89 CredentialField(key="lenderId", label="Lender ID", secret=False, required=True),
90 CredentialField(key="apiKey", label="API Key", secret=True, required=True),
91 ),
92 ),
93 ProviderDefinition(
94 provider="stripe",
95 label="Stripe Billing",
96 edit_permission="MANAGE_USER_ACCOUNT",
97 fields=(
98 CredentialField(key="publishableKey", label="Publishable Key", secret=False, required=True),
99 CredentialField(key="secretKey", label="Secret Key", secret=True, required=True),
100 ),
101 ),
102 ProviderDefinition(
103 provider="auth0",
104 label="Auth0",
105 edit_permission="MANAGE_USER_ACCOUNT",
106 fields=(
107 CredentialField(key="domain", label="Tenant Domain", secret=False, required=True),
108 CredentialField(key="clientId", label="Client ID", secret=False, required=True),
109 CredentialField(key="clientSecret", label="Client Secret", secret=True, required=True),
110 ),
111 ),
112 ProviderDefinition(
113 provider="rocket",
114 label="Rocket Pro TPO (SSO)",
115 edit_permission="MANAGE_USER_ACCOUNT",
116 fields=(
117 CredentialField(key="partnerId", label="Partner ID", secret=False, required=True),
118 CredentialField(key="ssoSecret", label="SSO Secret", secret=True, required=True),
119 ),
120 ),
121 ProviderDefinition(
122 provider="sendgrid",
123 label="SendGrid (Email + open tracking)",
124 edit_permission="MANAGE_USER_ACCOUNT",
125 fields=(
126 CredentialField(key="apiKey", label="API Key", secret=True, required=True),
127 CredentialField(key="fromEmail", label="From email", secret=False, required=True),
128 CredentialField(key="fromName", label="From name", secret=False, required=False),
129 ),
130 ),
131 ProviderDefinition(
132 provider="twilio",
133 label="Twilio (SMS + Voice / Robocall)",
134 edit_permission="MANAGE_USER_ACCOUNT",
135 fields=(
136 CredentialField(key="accountSid", label="Account SID", secret=False, required=True),
137 CredentialField(key="authToken", label="Auth Token", secret=True, required=True),
138 CredentialField(key="fromNumber", label="From number (SMS/voice)", secret=False, required=True),
139 ),
140 ),
141)
143_PROVIDERS_BY_CODE = {p.provider: p for p in SETTINGS_PROVIDERS}
146def _mask_secret(v: str) -> str:
147 if not v: 147 ↛ 148line 147 didn't jump to line 148 because the condition on line 147 was never true
148 return ""
149 if len(v) <= 4: 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true
150 return "••••"
151 return "••••••" + v[-4:]
154def _resolve_org(org_unit_str: str) -> OrgUnit:
155 if org_unit_str.isdigit(): 155 ↛ 156line 155 didn't jump to line 156 because the condition on line 155 was never true
156 org = OrgUnit.objects.filter(pk=int(org_unit_str)).first()
157 if org:
158 return org
159 org = OrgUnit.objects.filter(name__iexact=org_unit_str).first()
160 if org: 160 ↛ 162line 160 didn't jump to line 162 because the condition on line 160 was always true
161 return org
162 fallback = OrgUnit.objects.filter(is_headquarters=True).first() or OrgUnit.objects.first()
163 if fallback is None:
164 fallback = OrgUnit.objects.create(name=org_unit_str or "Headquarters", is_headquarters=True)
165 return fallback
168@router.get("/providers", response=list[ProviderDefOut], auth=StaffAuth())
169def list_providers(request):
170 """Return catalog of configurable provider cards."""
171 return [
172 {
173 "provider": p.provider,
174 "label": p.label,
175 "edit_permission": p.edit_permission,
176 "fields": [
177 {
178 "key": f.key,
179 "label": f.label,
180 "secret": f.secret,
181 "required": f.required,
182 }
183 for f in p.fields
184 ],
185 }
186 for p in SETTINGS_PROVIDERS
187 ]
190@router.get("/{org_unit}/credentials/{provider}", response=dict[str, str], auth=StaffAuth())
191def get_credentials(request, org_unit: str, provider: str):
192 """Fetch stored credentials for provider and org unit with secrets masked."""
193 p_def = _PROVIDERS_BY_CODE.get(provider)
194 if p_def is None: 194 ↛ 195line 194 didn't jump to line 195 because the condition on line 194 was never true
195 raise DomainError([AppError("NOT_FOUND", f"Unknown provider: {provider}.")])
197 if not request.principal.has(p_def.edit_permission):
198 raise DomainError(
199 [AppError("FORBIDDEN", f"You don't have permission to edit {p_def.label} credentials.")]
200 )
202 org = _resolve_org(org_unit)
203 oi = OrgIntegration.objects.filter(org_unit=org, provider=provider).first()
204 if oi is None:
205 return {}
207 config: dict[str, str] = dict(oi.config or {})
208 secrets: dict[str, str] = {}
209 if oi.secret_enc: 209 ↛ 217line 209 didn't jump to line 217 because the condition on line 209 was always true
210 decrypted = decrypt_secret(oi.secret_enc)
211 if decrypted: 211 ↛ 217line 211 didn't jump to line 217 because the condition on line 211 was always true
212 try:
213 secrets = json.loads(decrypted)
214 except Exception:
215 secrets = {"password": decrypted, "secret": decrypted, "apiKey": decrypted}
217 out: dict[str, str] = {}
218 for f in p_def.fields:
219 if f.secret:
220 val = secrets.get(f.key) or config.get(f.key)
221 if val is not None: 221 ↛ 218line 221 didn't jump to line 218 because the condition on line 221 was always true
222 out[f.key] = _mask_secret(str(val))
223 else:
224 val = config.get(f.key)
225 if val is not None: 225 ↛ 218line 225 didn't jump to line 218 because the condition on line 225 was always true
226 out[f.key] = str(val)
228 return out
231@router.put("/{org_unit}/credentials/{provider}", response=dict[str, str], auth=StaffAuth())
232def save_credentials(request, org_unit: str, provider: str, payload: dict[str, Any] = Body(...)):
233 """Save or update credentials for provider and org unit."""
234 p_def = _PROVIDERS_BY_CODE.get(provider)
235 if p_def is None: 235 ↛ 236line 235 didn't jump to line 236 because the condition on line 235 was never true
236 raise DomainError([AppError("NOT_FOUND", f"Unknown provider: {provider}.")])
238 if not request.principal.has(p_def.edit_permission): 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true
239 raise DomainError(
240 [AppError("FORBIDDEN", f"You don't have permission to edit {p_def.label} credentials.")]
241 )
243 # Validate required fields
244 missing = [
245 f"{f.label} is required"
246 for f in p_def.fields
247 if f.required and not str(payload.get(f.key, "")).strip()
248 ]
249 if missing:
250 raise DomainError([AppError("VALIDATION", msg) for msg in missing])
252 org = _resolve_org(org_unit)
253 oi, _ = OrgIntegration.objects.get_or_create(org_unit=org, provider=provider)
255 # Load existing secrets if updating
256 existing_secrets: dict[str, str] = {}
257 if oi.secret_enc: 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 decrypted = decrypt_secret(oi.secret_enc)
259 if decrypted:
260 try:
261 existing_secrets = json.loads(decrypted)
262 except Exception:
263 pass
265 new_config: dict[str, str] = {}
266 new_secrets: dict[str, str] = dict(existing_secrets)
268 for f in p_def.fields:
269 val = payload.get(f.key)
270 if val is None: 270 ↛ 271line 270 didn't jump to line 271 because the condition on line 270 was never true
271 continue
272 val_str = str(val).strip()
273 if f.secret:
274 # If value is masked ("••••"), retain existing secret
275 if val_str.startswith("••"): 275 ↛ 276line 275 didn't jump to line 276 because the condition on line 275 was never true
276 continue
277 new_secrets[f.key] = val_str
278 else:
279 new_config[f.key] = val_str
281 oi.config = new_config
282 if new_secrets: 282 ↛ 284line 282 didn't jump to line 284 because the condition on line 282 was always true
283 oi.secret_enc = encrypt_secret(json.dumps(new_secrets))
284 oi.enabled = True
285 oi.save()
287 # Return masked representation
288 out: dict[str, str] = {}
289 for f in p_def.fields:
290 if f.secret:
291 val = new_secrets.get(f.key)
292 if val is not None: 292 ↛ 289line 292 didn't jump to line 289 because the condition on line 292 was always true
293 out[f.key] = _mask_secret(val)
294 else:
295 val = new_config.get(f.key)
296 if val is not None: 296 ↛ 289line 296 didn't jump to line 289 because the condition on line 296 was always true
297 out[f.key] = val
299 return out