Coverage for app/crud/openstack_credentials.py: 100.00%

1"""CRUD for the per-user OpenStack credential row. 

2 

3Two flavours of `get_*` are exposed because the dispatch path and the 

4backend's own quotas endpoint have different needs: 

5 

6* `get_dispatch_envelope` — JSON-safe dict with **base64 ciphertext** 

7 for Celery transport. The backend never decrypts here. 

8* `get_decrypted_for_backend` — plaintext dict for the backend's own 

9 OpenStack calls (quotas). Plaintext lives only in this process's 

10 memory for the duration of the request. 

11 

12The plaintext envelope MUST NOT be passed to Celery. 

13""" 

14from __future__ import annotations 

15 

16import base64 

17from uuid import UUID 

18 

19from sqlalchemy.orm import Session 

20 

21from app.models import UserOpenStackCredential 

22from app.schemas import OpenStackCredentialUpsert 

23from app.utils import crypto 

24from app.utils.time import utcnow 

25 

26 

27class NoCredentialError(Exception): 

28 """Raised when a deploy is attempted before the user has uploaded creds.""" 

29 

30 

31def get_for_user(db: Session, user_id: UUID) -> UserOpenStackCredential | None: 

32 return ( 

33 db.query(UserOpenStackCredential) 

34 .filter(UserOpenStackCredential.userId == user_id) 

35 .first() 

36 ) 

37 

38 

39def upsert( 

40 db: Session, 

41 user_id: UUID, 

42 payload: OpenStackCredentialUpsert, 

43 validation_result: tuple[bool, str | None], 

44) -> UserOpenStackCredential: 

45 """Create or update the user's credential row. 

46 

47 `validation_result` comes from `services.openstack_validator.validate`. 

48 On success we stamp `last_validated_at` and clear the error; 

49 on failure we still persist (so the user can fix it via the UI) and 

50 record the message. 

51 """ 

52 ok, error = validation_result 

53 now = utcnow() 

54 

55 enc_id = crypto.encrypt(payload.identifier) 

56 enc_secret = crypto.encrypt(payload.secret) 

57 

58 row = get_for_user(db, user_id) 

59 if row is None: 

60 row = UserOpenStackCredential( 

61 userId=user_id, 

62 auth_type=payload.auth_type, 

63 auth_url=payload.auth_url, 

64 region_name=payload.region_name, 

65 interface=payload.interface or "public", 

66 identity_api_version=payload.identity_api_version or "3", 

67 project_id=payload.project_id, 

68 project_name=payload.project_name, 

69 user_domain_name=payload.user_domain_name, 

70 project_domain_name=payload.project_domain_name, 

71 encrypted_identifier=enc_id, 

72 encrypted_secret=enc_secret, 

73 last_validated_at=now if ok else None, 

74 last_validation_error=None if ok else error, 

75 ) 

76 db.add(row) 

77 else: 

78 row.auth_type = payload.auth_type 

79 row.auth_url = payload.auth_url 

80 row.region_name = payload.region_name 

81 row.interface = payload.interface or "public" 

82 row.identity_api_version = payload.identity_api_version or "3" 

83 row.project_id = payload.project_id 

84 row.project_name = payload.project_name 

85 row.user_domain_name = payload.user_domain_name 

86 row.project_domain_name = payload.project_domain_name 

87 row.encrypted_identifier = enc_id 

88 row.encrypted_secret = enc_secret 

89 row.last_validated_at = now if ok else row.last_validated_at 

90 row.last_validation_error = None if ok else error 

91 

92 db.commit() 

93 db.refresh(row) 

94 return row 

95 

96 

97def stamp_validation( 

98 db: Session, 

99 row: UserOpenStackCredential, 

100 validation_result: tuple[bool, str | None], 

101) -> UserOpenStackCredential: 

102 """Update validation metadata after a `/test` call without rotating ciphertext.""" 

103 ok, error = validation_result 

104 if ok: 

105 row.last_validated_at = utcnow() 

106 row.last_validation_error = None 

107 else: 

108 row.last_validation_error = error 

109 db.commit() 

110 db.refresh(row) 

111 return row 

112 

113 

114def delete(db: Session, user_id: UUID) -> bool: 

115 row = get_for_user(db, user_id) 

116 if row is None: 

117 return False 

118 db.delete(row) 

119 db.commit() 

120 return True 

121 

122 

123def _common_metadata(row: UserOpenStackCredential) -> dict: 

124 return { 

125 "auth_type": row.auth_type.value if hasattr(row.auth_type, "value") else row.auth_type, 

126 "auth_url": row.auth_url, 

127 "region_name": row.region_name, 

128 "interface": row.interface or "public", 

129 "identity_api_version": row.identity_api_version or "3", 

130 "project_id": row.project_id, 

131 "project_name": row.project_name, 

132 "user_domain_name": row.user_domain_name, 

133 "project_domain_name": row.project_domain_name, 

134 } 

135 

136 

137def get_dispatch_envelope(db: Session, user_id: UUID) -> dict: 

138 """Build the JSON-safe envelope shipped to the worker via Celery. 

139 

140 Ciphertext is base64-encoded straight from Postgres — no decryption 

141 hop in the backend. The worker decrypts in-process. 

142 """ 

143 row = get_for_user(db, user_id) 

144 if row is None: 

145 raise NoCredentialError(f"No OpenStack credential for user {user_id}") 

146 envelope = _common_metadata(row) 

147 envelope["encrypted_identifier_b64"] = base64.b64encode(row.encrypted_identifier).decode("ascii") 

148 envelope["encrypted_secret_b64"] = base64.b64encode(row.encrypted_secret).decode("ascii") 

149 return envelope 

150 

151 

152def get_decrypted_for_backend(db: Session, user_id: UUID) -> dict: 

153 """Plaintext dict for backend-only OpenStack calls (quotas). 

154 

155 NEVER pass this to Celery. Use `get_dispatch_envelope` for that path. 

156 """ 

157 row = get_for_user(db, user_id) 

158 if row is None: 

159 raise NoCredentialError(f"No OpenStack credential for user {user_id}") 

160 out = _common_metadata(row) 

161 out["identifier"] = crypto.decrypt(row.encrypted_identifier) 

162 out["secret"] = crypto.decrypt(row.encrypted_secret) 

163 return out