fix(batch-permit-job): use runtime auth for storage

jobs/batch-permit-validator/devops/vaults.gcp.env

@@ -5,7 +5,6 @@ DATABASE_PORT="op://database/$APP_ENV/strr-db/DATABASE_PORT"
 DATABASE_UNIX_SOCKET="op://database/$APP_ENV/strr-db/DATABASE_UNIX_SOCKET"
 GCP_CS_PROJECT_ID="op://buckets/$APP_ENV/strr/GCP_CS_PROJECT_ID"
 GCP_CS_SA_SCOPE="op://buckets/$APP_ENV/strr/GCP_CS_SA_SCOPE"
-GCP_AUTH_KEY="op://buckets/$APP_ENV/strr/GCP_AUTH_KEY"
 STR_DATA_API_CLIENT_ID="op://keycloak/$APP_ENV/str-data-api/STR_DATA_API_CLIENT_ID"
 STR_DATA_API_CLIENT_SECRET="op://keycloak/$APP_ENV/str-data-api/STR_DATA_API_CLIENT_SECRET"
 STR_DATA_API_TOKEN_URL="op://keycloak/$APP_ENV/str-data-api/STR_DATA_API_TOKEN_URL"

strr-api/src/strr_api/services/gcp_storage_service.py

@@ -40,6 +40,7 @@
 from datetime import timedelta
 
 from flask import current_app, has_app_context
+from google.auth.transport.requests import Request
 from google.cloud import storage
 from google.oauth2 import service_account
 
@@ -190,11 +191,27 @@ def upload_file(cls, file_type, file_contents, bucket_id, metadata=None):
     @classmethod
     def get_presigned_url(cls, bucket_id, blob_name, expiration_minutes):
         """Gets the presigned url for a file."""
-        bucket = cls.get_bucket(bucket_id)
+        storage_client = cls._create_storage_client()
+        bucket = storage_client.bucket(bucket_id)
         blob = bucket.blob(blob_name)
+        signed_url_kwargs = {}
+
+        credentials = storage_client._credentials  # pylint: disable=protected-access
+        if credentials and not isinstance(credentials, service_account.Credentials):
+            credentials.refresh(Request())
+            if service_account_email := getattr(credentials, "service_account_email", None):
+                signed_url_kwargs = {
+                    "service_account_email": service_account_email,
+                    "access_token": credentials.token,
+                }
 
         # Generate the signed URL
-        url = blob.generate_signed_url(version="v4", expiration=timedelta(minutes=expiration_minutes), method="GET")
+        url = blob.generate_signed_url(
+            version="v4",
+            expiration=timedelta(minutes=expiration_minutes),
+            method="GET",
+            **signed_url_kwargs,
+        )
 
         return url
 

strr-api/tests/unit/services/test_gcp_storage_service.py

@@ -187,15 +187,24 @@ def test_file_upload_and_presigned_url_succeed(mock_get_bucket, mock_uuid):
     mock_blob = MagicMock()
     mock_blob.generate_signed_url.return_value = "https://signed-url"
     mock_get_bucket.return_value.blob.return_value = mock_blob
+    mock_storage_client = MagicMock()
+    mock_storage_client._credentials = None
+    mock_storage_client.bucket.return_value.blob.return_value = mock_blob
 
     upload_key = GCPStorageService.upload_file("text/csv", b"file contents", "target-bucket")
-    url = GCPStorageService.get_presigned_url("target-bucket", "file-key", 10)
+    with patch(
+        "strr_api.services.gcp_storage_service.GCPStorageService._create_storage_client",
+        return_value=mock_storage_client,
+    ):
+        url = GCPStorageService.get_presigned_url("target-bucket", "file-key", 10)
 
     assert upload_key == "file-key"
     assert url == "https://signed-url"
     mock_uuid.assert_called_once()
-    mock_get_bucket.assert_any_call("target-bucket")
+    mock_get_bucket.assert_called_once_with("target-bucket")
     mock_get_bucket.return_value.blob.assert_any_call("file-key")
+    mock_storage_client.bucket.assert_called_once_with("target-bucket")
+    mock_storage_client.bucket.return_value.blob.assert_called_once_with("file-key")
     mock_blob.upload_from_string.assert_called_once_with(data=b"file contents", content_type="text/csv")
     mock_blob.generate_signed_url.assert_called_once_with(
         version="v4",
@@ -204,6 +213,34 @@ def test_file_upload_and_presigned_url_succeed(mock_get_bucket, mock_uuid):
     )
 
 
+@patch("strr_api.services.gcp_storage_service.Request")
+@patch("strr_api.services.gcp_storage_service.GCPStorageService._create_storage_client")
+def test_presigned_url_uses_adc_token_signing_kwargs(mock_create_client, mock_request):
+    """get_presigned_url supports keyless runtime credentials for signed URLs."""
+    mock_credentials = MagicMock()
+    mock_credentials.service_account_email = "sa-job@example.iam.gserviceaccount.com"
+    mock_credentials.token = "access-token"
+
+    mock_blob = MagicMock()
+    mock_blob.generate_signed_url.return_value = "https://signed-url"
+    mock_storage_client = MagicMock()
+    mock_storage_client._credentials = mock_credentials
+    mock_storage_client.bucket.return_value.blob.return_value = mock_blob
+    mock_create_client.return_value = mock_storage_client
+
+    url = GCPStorageService.get_presigned_url("target-bucket", "file-key", 10)
+
+    assert url == "https://signed-url"
+    mock_credentials.refresh.assert_called_once_with(mock_request.return_value)
+    mock_blob.generate_signed_url.assert_called_once_with(
+        version="v4",
+        expiration=timedelta(minutes=10),
+        method="GET",
+        service_account_email="sa-job@example.iam.gserviceaccount.com",
+        access_token="access-token",
+    )
+
+
 @patch("strr_api.services.gcp_storage_service.GCPStorageService.get_bucket")
 def test_upload_file_raises_external_service_exception_on_failure(mock_get_bucket):
     """upload_file wraps storage errors in ExternalServiceException."""