Remove the duplicate GCS signing cache from orch.py

DEVELOPING.md

@@ -112,7 +112,7 @@ links; raw HTML, images, and other block content are not displayed.
 
 ### The local loop (two terminals)
 
-The local backend talks to a real dev GCP project through your Application Default Credentials, so run `gcloud auth application-default login` once first.
+The local backend talks to a real dev GCP project through your Application Default Credentials. Because GCS URL signing (`/api/signUrl`, `/api/uploadUrl`) requires a service account identity, run `gcloud auth application-default login --impersonate-service-account=<SA_EMAIL>` (or set `GOOGLE_APPLICATION_CREDENTIALS` to a service account key file) once first.
 
 **Terminal 1 - local backend (the `/api` server):**
 

orch.py

@@ -65,17 +65,15 @@
 from werkzeug.exceptions import RequestEntityTooLarge
 from flask_cors import CORS
 from google.api_core import exceptions as google_exceptions
-from google.auth import compute_engine
-from google.auth import default as google_auth_default
 from google.auth.transport import requests as google_auth_requests
 from google.cloud import firestore
 from google.cloud.firestore_v1 import field_path
-from google.cloud import storage
 from google.oauth2 import id_token as google_id_token
 import orchestrator
 import transcription
 from util import database as util_database
 from util import errors as util_errors
+from util import gcs_wrapper
 from util import model_allowlist
 from util import submission_validation
 from werkzeug.security import safe_join
@@ -692,9 +690,8 @@ def get_status_handler() -> flask_response:
 # ---------------------------------------------------------------------------
 # Mediated data plane (ROLE='app'): signed-URL minting for GCS plus CRUD on
 # the UI Firestore database, so the SPA can run without direct Firestore or
-# Storage access. Module-level lazy singletons keep the per-request cost at
-# ~1 IAM signBlob RPC per unique path (the per-call construction in
-# util.gcs_wrapper.get_signed_url costs ~3 RPCs).
+# Storage access. Storage client and IAM signing credentials delegate to the
+# cached signing context in util.gcs_wrapper.
 # ---------------------------------------------------------------------------
 
 _SIGNED_GET_TTL = datetime.timedelta(hours=24)
@@ -732,17 +729,7 @@ def _is_allowed_upload_content_type(content_type: str) -> bool:
       _ALLOWED_UPLOAD_TYPE_PREFIXES
   )
 
-_storage_client = None
 _ui_db = None
-_signing_credentials = None
-
-
-def _get_storage_client() -> storage.Client:
-  """Returns the lazily-built module-level Cloud Storage client."""
-  global _storage_client
-  if _storage_client is None:
-    _storage_client = storage.Client()
-  return _storage_client
 
 
 def _get_ui_db() -> firestore.Client | None:
@@ -759,39 +746,19 @@ def _get_ui_db() -> firestore.Client | None:
   return _ui_db
 
 
-def _get_signing_credentials() -> compute_engine.IDTokenCredentials:
-  """Returns cached IAM-signing credentials, rebuilding them on expiry.
-
-  Inlines the flask-context mechanics of util.gcs_wrapper.get_signed_url:
-  the metadata-server credentials carry no private key, so URL signing
-  goes through the IAM signBlob API via compute_engine.IDTokenCredentials
-  (requires roles/iam.serviceAccountTokenCreator on the runtime SA).
-  """
-  global _signing_credentials
-  if _signing_credentials is None or _signing_credentials.expired:
-    auth_request = google_auth_requests.Request()
-    source_credentials, _ = google_auth_default()
-    source_credentials.refresh(auth_request)
-    _signing_credentials = compute_engine.IDTokenCredentials(
-        auth_request,
-        '',
-        service_account_email=source_credentials.service_account_email,  # pyright: ignore[reportAttributeAccessIssue]
-    )
-  return _signing_credentials
-
-
 def _signed_url(
     blob, method: str, expiration: datetime.timedelta, content_type=None
 ) -> str:
-  """Generates a V4 signed URL for the blob with the cached credentials."""
+  """Generates a V4 signed URL for the blob with cached IAM credentials."""
+  _, credentials = gcs_wrapper.get_signing_context()
   kwargs = {}
   if content_type is not None:
     kwargs['content_type'] = content_type
   return blob.generate_signed_url(
       version='v4',
       expiration=expiration,
       method=method,
-      credentials=_get_signing_credentials(),
+      credentials=credentials,
       **kwargs,
   )
 
@@ -999,7 +966,8 @@ def upload_url_handler() -> flask_response:
   if not bucket_name:
     return _json_error('gcsBucket not configured', 500)
   object_path = f'{prefix}/{file_name}'
-  blob = _get_storage_client().bucket(bucket_name).blob(object_path)
+  storage_client, _ = gcs_wrapper.get_signing_context()
+  blob = storage_client.bucket(bucket_name).blob(object_path)
   exists = blob.exists()
   upload_url = None
   if not exists:
@@ -1061,7 +1029,8 @@ def sign_url_handler() -> flask_response:
     if not ttl_arg.isdecimal() or not 1 <= int(ttl_arg) <= 86400:
       return _json_error('ttl must be 1-86400 seconds', 400)
     ttl = datetime.timedelta(seconds=int(ttl_arg))
-  bucket = _get_storage_client().bucket(bucket_name)
+  storage_client, _ = gcs_wrapper.get_signing_context()
+  bucket = storage_client.bucket(bucket_name)
   urls = {
       path: _signed_url(bucket.blob(path), 'GET', ttl) for path in paths
   }

test/test_frontdoor_data.py

@@ -17,10 +17,9 @@
 Follows the fixture pattern of test/test_frontdoor.py: env vars are set
 via monkeypatch BEFORE (re)importing orch, and orchestrator's import-time
 side effects are neutralised by the shared orchestrator_module fixture.
-The GCS and UI-Firestore
-clients are replaced with in-memory fakes by presetting orch's lazy
-module-level singletons, and the cached signing-credentials factory is
-stubbed so no test performs network I/O.
+The UI-Firestore client is replaced with an in-memory fake by presetting
+orch's lazy module-level singleton, and GCS storage/signing context is
+stubbed on util.gcs_wrapper so no test performs network I/O.
 """
 
 import copy
@@ -37,6 +36,7 @@
 
 # Reused module-scoped fixture; pytest picks it up from this namespace.
 from test.test_frontdoor import orchestrator_module  # noqa: F401  pylint: disable=unused-import
+from util import gcs_wrapper
 from util import model_allowlist
 
 _REPO = pathlib.Path(__file__).resolve().parent.parent
@@ -376,8 +376,11 @@ def _load_app(monkeypatch, **env):
   fake_db = FakeUiDb()
   fake_storage = FakeStorageClient()
   monkeypatch.setattr(orch, '_ui_db', fake_db)
-  monkeypatch.setattr(orch, '_storage_client', fake_storage)
-  monkeypatch.setattr(orch, '_get_signing_credentials', lambda: None)
+  monkeypatch.setattr(
+      gcs_wrapper,
+      'get_signing_context',
+      lambda: (fake_storage, None),
+  )
   monkeypatch.setitem(orch.config, 'gcsBucket', _BUCKET)
   return orch, fake_db, fake_storage
 
@@ -2041,8 +2044,11 @@ def test_data_endpoints_fail_soft_without_firestore_db_ui(
   assert response.get_json() == {'error': 'FIRESTORE_DB_UI not configured'}
 
   # The storage endpoints do not need the UI database.
-  monkeypatch.setattr(orch, '_storage_client', FakeStorageClient())
-  monkeypatch.setattr(orch, '_get_signing_credentials', lambda: None)
+  monkeypatch.setattr(
+      gcs_wrapper,
+      'get_signing_context',
+      lambda: (FakeStorageClient(), None),
+  )
   monkeypatch.setitem(orch.config, 'gcsBucket', _BUCKET)
   assert client.get('/api/signUrl?path=remix-input/a.png').status_code == 200
 
@@ -2061,3 +2067,37 @@ def test_worker_role_has_no_data_routes(monkeypatch, orchestrator_module):
   assert client.get('/api/projects').status_code == 404
   assert client.post('/api/uploadUrl', json={}).status_code == 404
   assert client.get('/api/config').status_code == 404
+
+
+def test_orch_signed_url_routes_through_gcs_wrapper_without_second_cache(
+    monkeypatch, orchestrator_module
+):
+  """orch delegates signed-URL generation to util.gcs_wrapper with no second cache."""
+  del orchestrator_module
+  orch, _, fake_storage = _load_app(monkeypatch)
+
+  # Assert orch has no duplicate cache state or functions
+  assert not hasattr(orch, '_storage_client')
+  assert not hasattr(orch, '_signing_credentials')
+  assert not hasattr(orch, '_get_signing_credentials')
+
+  blob = fake_storage.bucket(_BUCKET).blob('remix-input/asset.png')
+  sentinel_creds = object()
+  called_with_creds = []
+
+  def fake_get_signing_context():
+    return fake_storage, sentinel_creds
+
+  def fake_generate_signed_url(credentials=None, **kwargs):
+    del kwargs
+    called_with_creds.append(credentials)
+    return 'https://signed.example/url'
+
+  monkeypatch.setattr(
+      gcs_wrapper, 'get_signing_context', fake_get_signing_context
+  )
+  monkeypatch.setattr(blob, 'generate_signed_url', fake_generate_signed_url)
+
+  url = orch._signed_url(blob, 'GET', datetime.timedelta(hours=1))
+  assert url == 'https://signed.example/url'
+  assert called_with_creds == [sentinel_creds]

test/test_gcs_wrapper.py

@@ -15,6 +15,7 @@
 """Tests for the Cloud Storage wrapper."""
 
 import datetime
+import threading
 from unittest import mock
 
 import pytest
@@ -113,7 +114,10 @@ def test_get_signed_url_flask_context_caches_client_and_credentials_until_expire
 
   mock_cred = mock.Mock()
   mock_cred.service_account_email = 'sa@example.com'
-  future_expiry = datetime.datetime.now() + datetime.timedelta(hours=1)
+  future_expiry = (
+      datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)
+      + datetime.timedelta(hours=1)
+  )
   mock_cred.expiry = future_expiry
   mock_default = mock.Mock(return_value=(mock_cred, 'project-id'))
 
@@ -140,9 +144,119 @@ def test_get_signed_url_flask_context_caches_client_and_credentials_until_expire
 
     # Now expire credentials via real expiry; refresh must be called a second time
     gcs_wrapper._CACHED_SIGNING_CREDENTIALS.expiry = (
-        datetime.datetime.now() - datetime.timedelta(minutes=5)
+        datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)
+        - datetime.timedelta(minutes=5)
     )
     url3 = gcs_wrapper.get_signed_url('bucket', 'c.mp4', flask_context=True)
     assert url3 == 'https://storage.googleapis.com/signed'
     assert client_factory.call_count == 1
     assert mock_cred.refresh.call_count == 2
+
+
+def test_signing_context_uses_refreshed_service_account_email():
+  """Signer identities must use the email resolved by credential refresh."""
+  gcs_wrapper._CACHED_STORAGE_CLIENT = None
+  gcs_wrapper._CACHED_CLIENT_FACTORY = None
+  gcs_wrapper._CACHED_SIGNING_CREDENTIALS = None
+  gcs_wrapper._CACHED_AUTH_FACTORY = None
+
+  mock_client = mock.Mock()
+  mock_cred = mock.Mock()
+  mock_cred.service_account_email = 'default'
+  mock_cred.expiry = (
+      datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)
+      + datetime.timedelta(hours=1)
+  )
+
+  def resolve_service_account(_):
+    mock_cred.service_account_email = 'resolved@example.com'
+
+  mock_cred.refresh.side_effect = resolve_service_account
+  mock_default = mock.Mock(return_value=(mock_cred, 'project-id'))
+
+  with mock.patch(
+      'util.gcs_wrapper.storage.Client', return_value=mock_client
+  ), mock.patch(
+      'util.gcs_wrapper.default', mock_default
+  ), mock.patch(
+      'util.gcs_wrapper.iam.Signer'
+  ) as signer_factory, mock.patch(
+      'util.gcs_wrapper.compute_engine.IDTokenCredentials'
+  ) as id_token_factory:
+    gcs_wrapper.get_signing_context()
+
+  assert signer_factory.call_args.args[2] == 'resolved@example.com'
+  assert id_token_factory.call_args.kwargs[
+      'service_account_email'
+  ] == 'resolved@example.com'
+
+
+def test_signing_context_built_once_across_concurrent_callers():
+  mock_client = mock.Mock()
+  mock_cred = mock.Mock()
+  mock_cred.service_account_email = 'sa@example.com'
+  mock_cred.expiry = (
+      datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)
+      + datetime.timedelta(hours=1)
+  )
+  mock_default = mock.Mock(return_value=(mock_cred, 'project-id'))
+
+  with mock.patch(
+      'util.gcs_wrapper.storage.Client', return_value=mock_client
+  ) as client_factory, mock.patch(
+      'util.gcs_wrapper.default', mock_default
+  ), mock.patch(
+      'util.gcs_wrapper.iam.Signer'
+  ) as signer_factory:
+    num_threads = 10
+    barrier = threading.Barrier(num_threads)
+    results = [None] * num_threads
+
+    def worker(idx):
+      barrier.wait()
+      results[idx] = gcs_wrapper.get_signing_context()
+
+    threads = [
+        threading.Thread(target=worker, args=(i,)) for i in range(num_threads)
+    ]
+    for t in threads:
+      t.start()
+    for t in threads:
+      t.join()
+
+    assert client_factory.call_count == 1
+    assert mock_default.call_count == 1
+    assert mock_cred.refresh.call_count == 1
+    assert signer_factory.call_count == 1
+
+    expected_client, expected_creds = results[0]
+    for client, creds in results:
+      assert client is expected_client
+      assert creds is expected_creds
+
+
+def test_non_service_account_adc_raises_actionable_error():
+  # Simulates an end-user OAuth credential from `gcloud auth application-default login`
+  # which has no service_account_email attribute.
+  user_cred = mock.NonCallableMock(spec=['token', 'refresh'])
+  assert not hasattr(user_cred, 'service_account_email')
+
+  mock_client = mock.Mock()
+  mock_default = mock.Mock(return_value=(user_cred, 'project-id'))
+
+  with mock.patch(
+      'util.gcs_wrapper.storage.Client', return_value=mock_client
+  ) as client_factory, mock.patch(
+      'util.gcs_wrapper.default', mock_default
+  ):
+    # Must raise a clear, actionable RuntimeError, not an unhelpful AttributeError.
+    with pytest.raises(RuntimeError) as exc_info:
+      gcs_wrapper.get_signed_url('bucket', 'file.mp4', flask_context=True)
+
+    err_msg = str(exc_info.value)
+    assert 'service_account_email' in err_msg
+    assert 'gcloud auth application-default login' in err_msg
+    assert '--impersonate-service-account' in err_msg
+    assert 'DEVELOPING.md' in err_msg
+    assert client_factory.call_count == 1
+    assert mock_default.call_count == 1