Coverage for app/services/openstack_auth.py: 100.00%

1"""Per-task OpenStack credential materialization. 

2 

3The worker receives an encrypted credential envelope on the Celery task args 

4(see backend `crud.openstack_credentials.get_dispatch_envelope`). This module 

5decrypts the envelope in-process and writes a `clouds.yaml` (mode 0600) into 

6the per-deployment workspace for the duration of one task. The file is 

7removed eagerly when the context manager exits — so, even on a crash, the 

8plaintext credential lives on disk only for the active phase of the build. 

9 

10The plaintext is never logged. The `creds` dict is wiped from memory in 

11`__exit__` so the GC can reclaim it quickly. 

12""" 

13 

14from __future__ import annotations 

15 

16import base64 

17import contextlib 

18import os 

19from typing import Any 

20 

21import yaml 

22 

23from ..utils.crypto import decrypt 

24 

25 

26class CredentialEnvelopeError(Exception): 

27 """Raised when the dispatched envelope is missing fields or fails to decrypt.""" 

28 

29 

30class PerTaskCloudsConfig: 

31 """Context manager: materialize a per-task `clouds.yaml`, then shred it. 

32 

33 Usage: 

34 with PerTaskCloudsConfig(envelope, work_dir=repo_path) as env: 

35 packer.run(env_vars=env) 

36 terraform.run(env_vars=env) 

37 """ 

38 

39 # Profile name written into the per-task ``clouds.yaml``. App 

40 # Terraform templates reference this via ``provider "openstack" { 

41 # cloud = "openstack" }`` (matching the OpenStack convention used 

42 # in their docs), so changing this requires updating every 

43 # template — keep it as ``"openstack"`` unless there's a strong 

44 # reason to switch. 

45 CLOUD_NAME = "openstack" 

46 

47 def __init__(self, envelope: dict[str, Any], work_dir: str): 

48 if not isinstance(envelope, dict): 

49 raise CredentialEnvelopeError("OpenStack credential envelope is missing or invalid") 

50 

51 try: 

52 id_b64 = envelope["encrypted_identifier_b64"] 

53 secret_b64 = envelope["encrypted_secret_b64"] 

54 auth_type = envelope["auth_type"] 

55 auth_url = envelope["auth_url"] 

56 except KeyError as e: 

57 raise CredentialEnvelopeError(f"OpenStack credential envelope missing field: {e}") 

58 

59 try: 

60 identifier = decrypt(base64.b64decode(id_b64.encode("ascii"))) 

61 secret = decrypt(base64.b64decode(secret_b64.encode("ascii"))) 

62 except Exception as e: 

63 # Don't surface InvalidToken text — only the type — to keep logs clean. 

64 raise CredentialEnvelopeError( 

65 f"Failed to decrypt OpenStack credential envelope ({type(e).__name__}). " 

66 "Check that backend and worker share the same CREDENTIAL_ENCRYPTION_KEY." 

67 ) 

68 

69 self._creds: dict[str, Any] | None = { 

70 "auth_type": auth_type, 

71 "auth_url": auth_url, 

72 "region_name": envelope.get("region_name"), 

73 "interface": envelope.get("interface") or "public", 

74 "identity_api_version": envelope.get("identity_api_version") or "3", 

75 "project_id": envelope.get("project_id"), 

76 "project_name": envelope.get("project_name"), 

77 "user_domain_name": envelope.get("user_domain_name"), 

78 "project_domain_name": envelope.get("project_domain_name") or envelope.get("user_domain_name"), 

79 "_identifier": identifier, 

80 "_secret": secret, 

81 } 

82 

83 os.makedirs(work_dir, exist_ok=True) 

84 self.path = os.path.join(work_dir, "clouds.yaml") 

85 

86 def _cloud_block(self) -> dict[str, Any]: 

87 c = self._creds 

88 if c is None: 

89 raise CredentialEnvelopeError("Credentials already shredded") 

90 

91 block: dict[str, Any] = { 

92 "auth_type": c["auth_type"], 

93 "auth": {"auth_url": c["auth_url"]}, 

94 "interface": c["interface"], 

95 "identity_api_version": c["identity_api_version"], 

96 } 

97 if c["region_name"]: 

98 block["region_name"] = c["region_name"] 

99 

100 auth = block["auth"] 

101 if c["auth_type"] == "v3applicationcredential": 

102 auth["application_credential_id"] = c["_identifier"] 

103 auth["application_credential_secret"] = c["_secret"] 

104 else: 

105 auth["username"] = c["_identifier"] 

106 auth["password"] = c["_secret"] 

107 if c["project_id"]: 

108 auth["project_id"] = c["project_id"] 

109 if c["project_name"]: 

110 auth["project_name"] = c["project_name"] 

111 if c["user_domain_name"]: 

112 auth["user_domain_name"] = c["user_domain_name"] 

113 if c["project_domain_name"]: 

114 auth["project_domain_name"] = c["project_domain_name"] 

115 return block 

116 

117 def _env_vars(self) -> dict[str, str]: 

118 """Mirror selected creds into OS_* env vars for tools that ignore clouds.yaml.""" 

119 c = self._creds or {} 

120 env: dict[str, str] = {} 

121 if c.get("auth_url"): 

122 env["OS_AUTH_URL"] = c["auth_url"] 

123 if c.get("region_name"): 

124 env["OS_REGION_NAME"] = c["region_name"] 

125 if c.get("interface"): 

126 env["OS_INTERFACE"] = c["interface"] 

127 if c.get("identity_api_version"): 

128 env["OS_IDENTITY_API_VERSION"] = c["identity_api_version"] 

129 

130 if c.get("auth_type") == "v3applicationcredential": 

131 env["OS_AUTH_TYPE"] = "v3applicationcredential" 

132 env["OS_APPLICATION_CREDENTIAL_ID"] = c["_identifier"] 

133 env["OS_APPLICATION_CREDENTIAL_SECRET"] = c["_secret"] 

134 else: 

135 env["OS_AUTH_TYPE"] = "password" 

136 env["OS_USERNAME"] = c["_identifier"] 

137 env["OS_PASSWORD"] = c["_secret"] 

138 if c.get("project_id"): 

139 env["OS_PROJECT_ID"] = c["project_id"] 

140 if c.get("project_name"): 

141 env["OS_PROJECT_NAME"] = c["project_name"] 

142 if c.get("user_domain_name"): 

143 env["OS_USER_DOMAIN_NAME"] = c["user_domain_name"] 

144 if c.get("project_domain_name"): 

145 env["OS_PROJECT_DOMAIN_NAME"] = c["project_domain_name"] 

146 return env 

147 

148 def __enter__(self) -> dict[str, str]: 

149 # O_EXCL: refuse to overwrite a stale file from a previous task that 

150 # crashed before reaching its `__exit__`. mode 0600 is enforced by 

151 # the syscall, not by a separate chmod (no race window). 

152 flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL 

153 try: 

154 fd = os.open(self.path, flags, 0o600) 

155 except FileExistsError: 

156 # A stale file means a prior task in this same workspace died. Clean and retry. 

157 with contextlib.suppress(FileNotFoundError): 

158 os.remove(self.path) 

159 fd = os.open(self.path, flags, 0o600) 

160 

161 try: 

162 with os.fdopen(fd, "w") as f: 

163 yaml.safe_dump( 

164 {"clouds": {self.CLOUD_NAME: self._cloud_block()}}, 

165 f, 

166 default_flow_style=False, 

167 ) 

168 except Exception: 

169 with contextlib.suppress(FileNotFoundError): 

170 os.remove(self.path) 

171 raise 

172 

173 env = { 

174 "OS_CLIENT_CONFIG_FILE": self.path, 

175 "OS_CLOUD": self.CLOUD_NAME, 

176 } 

177 env.update(self._env_vars()) 

178 return env 

179 

180 def __exit__(self, exc_type, exc, tb) -> None: 

181 # Shred file eagerly. The repo workspace cleanup in tasks.py:finally 

182 # would also remove it, but doing it here narrows the window. 

183 with contextlib.suppress(FileNotFoundError): 

184 os.remove(self.path) 

185 # Drop plaintext references; let GC reclaim them quickly. 

186 self._creds = None