Coverage for app/routers/openstack_credentials.py: 96.88%

1"""Per-user OpenStack credential management. 

2 

3All endpoints scope to the caller (`current_user`) — there is no 

4`{user_id}` path parameter. This makes IDOR impossible by construction. 

5The masked response never returns identifier or secret material. 

6""" 

7from __future__ import annotations 

8 

9from fastapi import APIRouter, Depends, HTTPException, Response, status 

10from sqlalchemy.orm import Session 

11 

12from app.crud import deployments as crud_deployments 

13from app.crud import locks as crud_locks 

14from app.crud import openstack_credentials as crud_creds 

15from app.database import get_db 

16from app.models import User 

17from app.schemas import ( 

18 OpenStackCredentialFromYaml, 

19 OpenStackCredentialResponse, 

20 OpenStackCredentialUpsert, 

21) 

22from app.services import clouds_yaml_parser, openstack_validator 

23from app.utils.keycloak_auth import get_current_user_keycloak 

24 

25router = APIRouter() 

26 

27 

28def _lock_state(db: Session, user_id) -> tuple[bool, int]: 

29 n = crud_deployments.count_active_user_deployments(db, user_id) 

30 return (n > 0, n) 

31 

32 

33def _assert_unlocked(db: Session, user: User) -> None: 

34 """Refuse credential mutation while the user has non-destroyed deployments.""" 

35 is_locked, n = _lock_state(db, user.userId) 

36 if is_locked: 

37 raise HTTPException( 

38 status_code=status.HTTP_409_CONFLICT, 

39 detail={ 

40 "reason": "openstack_credentials_locked", 

41 "active_deployments": n, 

42 }, 

43 ) 

44 

45 

46def _to_response( 

47 row, 

48 *, 

49 has_credential: bool, 

50 is_locked: bool, 

51 active_deployments: int, 

52) -> OpenStackCredentialResponse: 

53 if row is None: 

54 return OpenStackCredentialResponse( 

55 has_credential=False, 

56 is_locked=is_locked, 

57 active_deployments=active_deployments, 

58 ) 

59 return OpenStackCredentialResponse( 

60 auth_type=row.auth_type, 

61 auth_url=row.auth_url, 

62 region_name=row.region_name, 

63 interface=row.interface, 

64 identity_api_version=row.identity_api_version, 

65 project_id=row.project_id, 

66 project_name=row.project_name, 

67 user_domain_name=row.user_domain_name, 

68 project_domain_name=row.project_domain_name, 

69 has_credential=has_credential, 

70 last_validated_at=row.last_validated_at, 

71 last_validation_error=row.last_validation_error, 

72 created_at=row.created_at, 

73 updated_at=row.updated_at, 

74 is_locked=is_locked, 

75 active_deployments=active_deployments, 

76 ) 

77 

78 

79@router.get( 

80 "/me/openstack-credentials", 

81 response_model=OpenStackCredentialResponse, 

82) 

83def get_my_credentials( 

84 db: Session = Depends(get_db), 

85 current_user: User = Depends(get_current_user_keycloak), 

86): 

87 """Always 200 — returns has_credential=False when none configured. 

88 

89 Lock state is included regardless so the frontend can render the 

90 correct guard UI without a second request. 

91 """ 

92 row = crud_creds.get_for_user(db, current_user.userId) 

93 is_locked, n = _lock_state(db, current_user.userId) 

94 return _to_response( 

95 row, 

96 has_credential=row is not None, 

97 is_locked=is_locked, 

98 active_deployments=n, 

99 ) 

100 

101 

102@router.put( 

103 "/me/openstack-credentials", 

104 response_model=OpenStackCredentialResponse, 

105) 

106def upsert_my_credentials( 

107 payload: OpenStackCredentialUpsert, 

108 db: Session = Depends(get_db), 

109 current_user: User = Depends(get_current_user_keycloak), 

110): 

111 """Auto-validates against Keystone before persisting. 

112 

113 On a successful authorize we record `last_validated_at`. On failure 

114 we still persist the row (so the user can fix it via the UI) and 

115 record the human-readable error message. 

116 """ 

117 crud_locks.acquire_user_xact_lock(db, current_user.userId) 

118 _assert_unlocked(db, current_user) 

119 result = openstack_validator.validate(payload) 

120 row = crud_creds.upsert(db, current_user.userId, payload, result) 

121 is_locked, n = _lock_state(db, current_user.userId) 

122 return _to_response(row, has_credential=True, is_locked=is_locked, active_deployments=n) 

123 

124 

125@router.put( 

126 "/me/openstack-credentials/from-yaml", 

127 response_model=OpenStackCredentialResponse, 

128) 

129def upsert_my_credentials_from_yaml( 

130 body: OpenStackCredentialFromYaml, 

131 db: Session = Depends(get_db), 

132 current_user: User = Depends(get_current_user_keycloak), 

133): 

134 crud_locks.acquire_user_xact_lock(db, current_user.userId) 

135 _assert_unlocked(db, current_user) 

136 payload = clouds_yaml_parser.parse(body.clouds_yaml, body.cloud_name) 

137 result = openstack_validator.validate(payload) 

138 row = crud_creds.upsert(db, current_user.userId, payload, result) 

139 is_locked, n = _lock_state(db, current_user.userId) 

140 return _to_response(row, has_credential=True, is_locked=is_locked, active_deployments=n) 

141 

142 

143@router.post( 

144 "/me/openstack-credentials/test", 

145 response_model=OpenStackCredentialResponse, 

146) 

147def test_my_credentials( 

148 db: Session = Depends(get_db), 

149 current_user: User = Depends(get_current_user_keycloak), 

150): 

151 """Re-authorize the stored credential and refresh validation metadata. 

152 

153 Allowed even while locked — purely read-only against Keystone. 

154 """ 

155 row = crud_creds.get_for_user(db, current_user.userId) 

156 if row is None: 

157 raise HTTPException( 

158 status_code=status.HTTP_404_NOT_FOUND, 

159 detail="No OpenStack credentials configured", 

160 ) 

161 plaintext = crud_creds.get_decrypted_for_backend(db, current_user.userId) 

162 payload = OpenStackCredentialUpsert( 

163 auth_type=row.auth_type, 

164 auth_url=row.auth_url, 

165 region_name=row.region_name, 

166 interface=row.interface, 

167 identity_api_version=row.identity_api_version, 

168 project_id=row.project_id, 

169 project_name=row.project_name, 

170 user_domain_name=row.user_domain_name, 

171 project_domain_name=row.project_domain_name, 

172 identifier=plaintext["identifier"], 

173 secret=plaintext["secret"], 

174 ) 

175 result = openstack_validator.validate(payload) 

176 row = crud_creds.stamp_validation(db, row, result) 

177 is_locked, n = _lock_state(db, current_user.userId) 

178 return _to_response(row, has_credential=True, is_locked=is_locked, active_deployments=n) 

179 

180 

181@router.delete( 

182 "/me/openstack-credentials", 

183 status_code=status.HTTP_204_NO_CONTENT, 

184 response_class=Response, 

185) 

186def delete_my_credentials( 

187 db: Session = Depends(get_db), 

188 current_user: User = Depends(get_current_user_keycloak), 

189): 

190 crud_locks.acquire_user_xact_lock(db, current_user.userId) 

191 _assert_unlocked(db, current_user) 

192 deleted = crud_creds.delete(db, current_user.userId) 

193 if not deleted: 

194 raise HTTPException( 

195 status_code=status.HTTP_404_NOT_FOUND, 

196 detail="No OpenStack credentials configured", 

197 ) 

198 return Response(status_code=status.HTTP_204_NO_CONTENT)