Coverage for src/gcp_attest/credentials.py: 49%

1import os 

2from typing import Literal 

3from typing import TYPE_CHECKING 

4from typing import TypedDict 

5 

6if TYPE_CHECKING: 

7 from google.auth import impersonated_credentials 

8 

9 

10SCOPES = ["https://www.googleapis.com/auth/cloud-platform"] 

11 

12 

13class JsonCredentialFormat(TypedDict): 

14 type: Literal["json"] 

15 subject_token_field_name: str 

16 

17 

18class UrlCredentialSource(TypedDict): 

19 url: str 

20 headers: dict[str, str] 

21 format: JsonCredentialFormat 

22 

23 

24class CredentialsConfig(TypedDict): 

25 universe_domain: str 

26 type: str 

27 audience: str 

28 subject_token_type: str 

29 token_url: str 

30 credential_source: UrlCredentialSource 

31 token_info_url: str 

32 

33 

34def _get_forgejo_credentials_config(audience: str) -> CredentialsConfig | None: 

35 """Return credential config for current Forgejo Actions run. 

36 

37 :param audience: URL of a Google Cloud Workload Identity Pool 

38 """ 

39 from urllib.parse import urlsplit, parse_qs, urlencode 

40 

41 forgejo_token_request_url = os.environ.get("ACTIONS_ID_TOKEN_REQUEST_URL") 

42 if forgejo_token_request_url is None: 

43 # Not in actions 

44 return None 

45 try: 

46 bearer = os.environ["ACTIONS_ID_TOKEN_REQUEST_TOKEN"] 

47 except KeyError as e: 

48 raise RuntimeError("Actions ID token request bearer not found") from e 

49 

50 parts = urlsplit(forgejo_token_request_url) 

51 query = parse_qs(parts.query) 

52 query["audience"] = [audience] 

53 forgejo_token_request_url = parts._replace( 

54 query=urlencode(query, doseq=True) 

55 ) 

56 

57 return { 

58 "universe_domain": "googleapis.com", 

59 "type": "external_account", 

60 "audience": audience.removeprefix("https:"), 

61 "subject_token_type": "urn:ietf:params:oauth:token-type:jwt", 

62 "token_url": "https://sts.googleapis.com/v1/token", 

63 "credential_source": { 

64 "url": forgejo_token_request_url.geturl(), 

65 "headers": { 

66 "Authorization": f"Bearer {bearer}", 

67 }, 

68 "format": {"type": "json", "subject_token_field_name": "value"}, 

69 }, 

70 "token_info_url": "https://sts.googleapis.com/v1/introspect", 

71 } 

72 

73 

74def _get_forgejo_actions_credentials( 

75 service_account: str | None = None, 

76 pool_url: str | None = None, 

77) -> impersonated_credentials.Credentials | None: 

78 from google.auth import identity_pool 

79 from google.auth import impersonated_credentials 

80 

81 if service_account is None: 81 ↛ 83line 81 didn't jump to line 83 because the condition on line 81 was always true

82 service_account = os.environ.get("GCP_ATTEST_SERVICE_ACCOUNT_EMAIL") 

83 if service_account is None: 83 ↛ 88line 83 didn't jump to line 88 because the condition on line 83 was always true

84 # TODO: better error handling 

85 # raise RuntimeError("No service account email found") 

86 return None 

87 

88 if pool_url is None: 

89 pool_url = os.environ.get("GCP_ATTEST_IDENTITY_POOL_URL") 

90 if pool_url is None: 

91 raise RuntimeError("No URL for Workload Identity Pool found") 

92 

93 cred_config = _get_forgejo_credentials_config(pool_url) 

94 if cred_config is None: 

95 return None 

96 

97 source_credentials = identity_pool.Credentials.from_info(cred_config) 

98 return impersonated_credentials.Credentials( 

99 source_credentials, 

100 service_account, 

101 target_scopes=SCOPES, 

102 ) 

103 

104 

105def _get_ambient_gcp_credentials() -> ( 

106 impersonated_credentials.Credentials | None 

107): 

108 from google.auth import default as default_credentials 

109 from google.auth import impersonated_credentials 

110 from google.oauth2 import service_account 

111 

112 credentials, _ = default_credentials(scopes=SCOPES) 

113 if isinstance(credentials, service_account.Credentials): 113 ↛ 114line 113 didn't jump to line 114 because the condition on line 113 was never true

114 return impersonated_credentials.Credentials( 

115 credentials, 

116 credentials.service_account_email, 

117 target_scopes=SCOPES, 

118 ) 

119 elif isinstance(credentials, impersonated_credentials.Credentials): 119 ↛ 122line 119 didn't jump to line 122 because the condition on line 119 was always true

120 return credentials 

121 else: 

122 return None 

123 

124 

125def get_id_token_credentials( 

126 audience: str, 

127 credentials: impersonated_credentials.Credentials | None = None, 

128) -> impersonated_credentials.IDTokenCredentials: 

129 """Return IDTokenCredentials for Google Cloud. 

130 

131 If no credentials are provided, will try to get ambient credentials. 

132 Currently supported: 

133 

134 - Forgejo Actions 

135 - Google Application Default Credentials 

136 

137 :param audience: ID token 'aud' claim 

138 :param credentials: impersonated GCP crednetials to use to issue ID Tokens 

139 """ 

140 from google.auth import impersonated_credentials 

141 

142 if credentials is None: 142 ↛ 146line 142 didn't jump to line 146 because the condition on line 142 was always true

143 # Try Forgejo 

144 credentials = _get_forgejo_actions_credentials() 

145 

146 if credentials is None: 146 ↛ 150line 146 didn't jump to line 150 because the condition on line 146 was always true

147 # Try Google 

148 credentials = _get_ambient_gcp_credentials() 

149 

150 if credentials is None: 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true

151 raise RuntimeError("Could not detect any ambient credentials") 

152 

153 return impersonated_credentials.IDTokenCredentials( 

154 credentials, 

155 target_audience=audience, 

156 include_email=True, 

157 )