Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion DEVELOPING.md
Original file line number Diff line number Diff line change
Expand Up @@ -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):**

Expand Down
51 changes: 10 additions & 41 deletions orch.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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:
Expand All @@ -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,
)

Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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
}
Expand Down
56 changes: 48 additions & 8 deletions test/test_frontdoor_data.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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

Expand All @@ -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]
134 changes: 132 additions & 2 deletions test/test_gcs_wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
"""Tests for the Cloud Storage wrapper."""

import datetime
import threading
from unittest import mock

import pytest
Expand Down Expand Up @@ -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'))

Expand All @@ -140,9 +144,135 @@ 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


def test_unresolved_default_service_account_email_after_refresh_raises():
mock_client = mock.Mock()
mock_cred = mock.Mock()
mock_cred.service_account_email = 'default'
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):
with pytest.raises(
RuntimeError,
match="credential refresh produced 'default'",
):
gcs_wrapper.get_signing_context()
Loading
Loading