Coverage for app/services/openstack_validator.py: 79.41%

1"""Authorize an OpenStack credential payload against the target Keystone. 

2 

3Used by the upsert and `/test` endpoints. The 15-second socket boundary 

4keeps a stuck Keystone from blocking the FastAPI worker; openstacksdk 

5otherwise has no consistent timeout knob across releases. 

6""" 

7from __future__ import annotations 

8 

9import socket 

10 

11import openstack 

12from openstack import exceptions as os_exc 

13 

14from app.models import OpenStackAuthType 

15from app.schemas import OpenStackCredentialUpsert 

16 

17_TIMEOUT_SECONDS = 15 

18 

19 

20def _build_connect_kwargs(payload: OpenStackCredentialUpsert) -> dict: 

21 base = { 

22 "auth_url": payload.auth_url, 

23 "region_name": payload.region_name, 

24 "interface": payload.interface or "public", 

25 "identity_api_version": payload.identity_api_version or "3", 

26 } 

27 if payload.auth_type == OpenStackAuthType.APPLICATION_CREDENTIAL: 

28 base.update({ 

29 "auth_type": "v3applicationcredential", 

30 "application_credential_id": payload.identifier, 

31 "application_credential_secret": payload.secret, 

32 }) 

33 else: 

34 base.update({ 

35 "auth_type": "password", 

36 "username": payload.identifier, 

37 "password": payload.secret, 

38 "project_id": payload.project_id, 

39 "project_name": payload.project_name, 

40 "user_domain_name": payload.user_domain_name, 

41 "project_domain_name": payload.project_domain_name or payload.user_domain_name, 

42 }) 

43 # openstack.connect tolerates None values for keys it doesn't need; keep them. 

44 return base 

45 

46 

47def validate(payload: OpenStackCredentialUpsert) -> tuple[bool, str | None]: 

48 """Try to authorize. Returns (ok, error_message). 

49 

50 The error message is short and human-readable — safe to surface in the 

51 UI. Never echoes the secret back. 

52 """ 

53 prev_default_timeout = socket.getdefaulttimeout() 

54 socket.setdefaulttimeout(_TIMEOUT_SECONDS) 

55 try: 

56 conn = openstack.connect(**_build_connect_kwargs(payload)) 

57 # Force a token round-trip; .authorize() returns the token string. 

58 conn.authorize() 

59 return True, None 

60 except os_exc.HttpException as e: 

61 status = getattr(e, "status_code", None) or getattr(e, "http_status", None) 

62 if status in (401, 403): 

63 return False, "Invalid credentials" 

64 if status == 404: 

65 return False, "Project or domain not found" 

66 return False, f"OpenStack rejected request (HTTP {status or '?'})" 

67 except (TimeoutError, socket.gaierror) as e: 

68 return False, f"Could not reach auth_url: {e}" 

69 except os_exc.SDKException as e: 

70 return False, f"OpenStack SDK error: {type(e).__name__}" 

71 except Exception as e: 

72 # Unknown error — return the type only, never the message (might 

73 # contain the request body). 

74 return False, f"Unexpected error: {type(e).__name__}" 

75 finally: 

76 socket.setdefaulttimeout(prev_default_timeout)