gemini-code-assist[bot] commented on code in PR #38992:
URL: https://github.com/apache/beam/pull/38992#discussion_r3424453358
##########
infra/enforcement/account_keys.py:
##########
@@ -106,6 +106,53 @@ def _denormalize_username(self, username: str) -> str:
return username.split(":", 1)[1].strip().lower()
return username
+ def _get_user_managed_keys_from_iam(self, account_email: str) -> List[str]:
+ """"
+ Retrieves the list of user-managed keys for a given service account
from IAM.
+
+ Args:
+ account_email (str): The email of the service account to retrieve
keys for.
+
+ Returns:
+ List[str]: A list of key IDs for the user-managed keys associated
with the service account
+ """
+ request = types.ListServiceAccountKeysRequest()
+ request.name =
f"projects/{self.project_id}/serviceAccounts/{account_email}"
+ request.key_types =
[types.ListServiceAccountKeysRequest.KeyType.USER_MANAGED]
+
+ try:
+ response =
self.service_account_client.list_service_account_keys(request=request)
+ return [key.name.split("/")[-1] for key in response.keys]
+ except Exception as e:
+ self.logger.error(f"Failed to retrieve keys for service account
'{account_email}': {e}")
+ return []
+
+ def _get_legal_keys_from_secret_manager(self, account_email: str) ->
List[str]:
+ """
+ Retrieves the list of legal keys for a given service account from
Secret Manager.
+
+ Args:
+ account_email (str): The email of the service account to retrieve
keys for.
+
+ Returns:
+ List[str]: A list of key IDs for the legal keys associated with
the service account.
+ """
+ legal_keys = []
+ parent = self.secret_client.secret_path(self.project_id,
self.secret_id)
+
+ try:
+ versions =
self.secret_client.list_secret_versions(request={"parent": parent})
+ for version in versions:
+ if version.state.name ==
secretmanager.SecretVersion.State.ENABLED:
+ response =
self.secret_client.access_secret_version(request={"name": version.name})
+ data_str = response.payload.data.decode("UTF-8")
+ key_id = data_str.split(":",1)[0]
+ legal_keys.append(key_id)
+ return legal_keys
+ except Exception as e:
+ self.logger.error(f"Failed to retrieve legal keys from Secret
Manager for account '{account_email}': {e}")
+ return []
Review Comment:
In `_get_legal_keys_from_secret_manager`, the parameter `account_email` is
defined but never used. Instead, `self.secret_id` is used to construct the
secret path. This is a critical bug because it will query the same secret
(whatever `self.secret_id` is set to) for every service account, rather than
querying the specific secret for each service account (e.g., `secret_name`
which is passed to this method).
To fix this, we should rename the parameter to `secret_name` to match its
usage and use it when constructing the secret path.
```python
def _get_legal_keys_from_secret_manager(self, secret_name: str) ->
List[str]:
"""
Retrieves the list of legal keys for a given service account from
Secret Manager.
Args:
secret_name (str): The name of the secret to retrieve keys for.
Returns:
List[str]: A list of key IDs for the legal keys associated with
the service account.
"""
legal_keys = []
parent = self.secret_client.secret_path(self.project_id, secret_name)
try:
versions =
self.secret_client.list_secret_versions(request={"parent": parent})
for version in versions:
if version.state.name ==
secretmanager.SecretVersion.State.ENABLED:
response =
self.secret_client.access_secret_version(request={"name": version.name})
data_str = response.payload.data.decode("UTF-8")
key_id = data_str.split(":",1)[0]
legal_keys.append(key_id)
return legal_keys
except Exception as e:
self.logger.error(f"Failed to retrieve legal keys from Secret
Manager for secret '{secret_name}': {e}")
return []
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
