From 2799f9188f20770cc8d14540a0443a70dca11632 Mon Sep 17 00:00:00 2001 From: jakeross Date: Tue, 18 Aug 2026 20:11:52 -0700 Subject: [PATCH] fix(ingestion): supply GCP credentials in a runtime that has none With the import fixed, database_connectivity reaches the Cloud SQL connector and fails on DefaultCredentialsError. Dagster+ Serverless runs outside GCP, so there is no metadata server and google.auth.default() finds nothing. The service account key travels as a Dagster+ secret and is written to a file at runtime, because GOOGLE_APPLICATION_CREDENTIALS names a path rather than holding a value. The file is mode 600 in the process temporary directory, which the container discards with the run. Called before importing db.engine rather than after: that module builds its connector at import time and resolves credentials right then, so doing it afterwards would be too late. The dlt pipeline calls it too -- gcsfs resolves credentials the same way, so the raw zone would have failed identically once the loader got that far. Existing credentials are never shadowed, so a developer's gcloud login is used as-is. An unset key is not an error: locally google.auth finds its own, and in Serverless it fails loudly, which is correct in both cases. A path supplied instead of the key itself is rejected with a message saying so, since that mistake would otherwise surface deep inside google.auth. Co-Authored-By: Claude Opus 5 --- automated_ingestion/defs/resources.py | 10 +++ .../scripts/set_code_location_env.sh | 13 ++- automated_ingestion/shared/credentials.py | 83 +++++++++++++++++ .../sources/san_acacia/dlt_pipeline.py | 8 ++ automated_ingestion/tests/test_credentials.py | 88 +++++++++++++++++++ 5 files changed, 201 insertions(+), 1 deletion(-) create mode 100644 automated_ingestion/shared/credentials.py create mode 100644 automated_ingestion/tests/test_credentials.py diff --git a/automated_ingestion/defs/resources.py b/automated_ingestion/defs/resources.py index 6e11316d8..25172bfa1 100644 --- a/automated_ingestion/defs/resources.py +++ b/automated_ingestion/defs/resources.py @@ -39,6 +39,16 @@ class OcotilloDatabase(ConfigurableResource): @contextmanager def session(self) -> Iterator[object]: """Yield a SQLAlchemy session, rolled back and closed on the way out.""" + # Credentials first: db.engine builds its Cloud SQL connector at import + # time, and the connector resolves Application Default Credentials right + # then. Serverless has none until they are written to disk, so doing this + # afterwards would be too late. + from automated_ingestion.shared.credentials import ( + ensure_application_default_credentials, + ) + + ensure_application_default_credentials() + # Imported lazily: importing db.engine builds an engine from the # environment at import time, which should happen when a run asks for a # session, not when Dagster loads the code location to list assets. diff --git a/automated_ingestion/scripts/set_code_location_env.sh b/automated_ingestion/scripts/set_code_location_env.sh index 7d11f647e..e4e872d8c 100755 --- a/automated_ingestion/scripts/set_code_location_env.sh +++ b/automated_ingestion/scripts/set_code_location_env.sh @@ -21,6 +21,7 @@ # # Usage: # ./automated_ingestion/scripts/set_code_location_env.sh storage +# ./automated_ingestion/scripts/set_code_location_env.sh credentials # ./automated_ingestion/scripts/set_code_location_env.sh vendor # ./automated_ingestion/scripts/set_code_location_env.sh database # @@ -53,6 +54,16 @@ storage) set_var INGESTION_GCS_BUCKET ocotillo-ingestion-production --scope full set_var INGESTION_GCS_BUCKET ocotillo-ingestion-staging --scope branch ;; +credentials) + : "${INGESTION_GCP_CREDENTIALS_JSON:?export the service account key JSON, not a path}" + # Serverless runs outside GCP, so there is no metadata server and nothing + # supplies Application Default Credentials. Both the Cloud SQL connector and + # gcsfs need them. Mint the key with: + # gcloud iam service-accounts keys create /dev/stdout \ + # --iam-account ocotillo-ingestion@waterdatainitiative-271000.iam.gserviceaccount.com + echo "GCP credentials (key JSON read from this shell, not echoed):" + set_var INGESTION_GCP_CREDENTIALS_JSON --from-local-env + ;; vendor) : "${DIVERHUB_USERNAME:?export it first, see the header}" : "${DIVERHUB_PASSWORD:?export it first, see the header}" @@ -89,7 +100,7 @@ database) fi ;; *) - echo "usage: $0 {storage|vendor|database}" >&2 + echo "usage: $0 {storage|credentials|vendor|database}" >&2 exit 64 ;; esac diff --git a/automated_ingestion/shared/credentials.py b/automated_ingestion/shared/credentials.py new file mode 100644 index 000000000..cc108ba84 --- /dev/null +++ b/automated_ingestion/shared/credentials.py @@ -0,0 +1,83 @@ +# =============================================================================== +# Copyright 2026 ross +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +""" +Application Default Credentials for a runtime that has none. + +Dagster+ Serverless runs outside GCP, so there is no metadata server to supply +credentials. Anything reaching Google -- the Cloud SQL connector for the loader, +gcsfs for the raw zone -- calls ``google.auth.default()`` and fails with +``DefaultCredentialsError`` unless something has put credentials on disk first. + +The service account key therefore travels as a Dagster+ secret and is written to +a file here, because ``GOOGLE_APPLICATION_CREDENTIALS`` names a path rather than +holding a value. The file lands in the process's temporary directory, which the +container discards when the run ends. +""" + +import json +import os +import tempfile + +CREDENTIALS_ENV_VAR = "INGESTION_GCP_CREDENTIALS_JSON" +"""Service account key JSON, as a Dagster+ secret. Never committed.""" + +_ADC_ENV_VAR = "GOOGLE_APPLICATION_CREDENTIALS" + +_written_path: str | None = None + + +def ensure_application_default_credentials() -> str | None: + """Materialize ADC from the environment, returning the path if written. + + Idempotent, and does nothing when credentials already exist -- locally that + means a developer's gcloud login is used as-is rather than being shadowed. + """ + global _written_path + + existing = os.environ.get(_ADC_ENV_VAR, "").strip() + if existing: + return existing + if _written_path is not None: + return _written_path + + raw = os.environ.get(CREDENTIALS_ENV_VAR, "").strip() + if not raw: + # No key configured. Leave google.auth to its own discovery, which + # succeeds on a developer machine and fails loudly in Serverless -- the + # right outcome in both cases. + return None + + try: + parsed = json.loads(raw) + except json.JSONDecodeError as exc: + raise RuntimeError( + f"{CREDENTIALS_ENV_VAR} is set but is not valid JSON. It must hold the " + "service account key itself, not a path to one." + ) from exc + + handle = tempfile.NamedTemporaryFile( + mode="w", suffix=".json", prefix="ingestion-adc-", delete=False + ) + with handle as fh: + json.dump(parsed, fh) + os.chmod(handle.name, 0o600) + + os.environ[_ADC_ENV_VAR] = handle.name + _written_path = handle.name + return handle.name + + +# ============= EOF ============================================= diff --git a/automated_ingestion/sources/san_acacia/dlt_pipeline.py b/automated_ingestion/sources/san_acacia/dlt_pipeline.py index c7b9e7319..25b08a270 100644 --- a/automated_ingestion/sources/san_acacia/dlt_pipeline.py +++ b/automated_ingestion/sources/san_acacia/dlt_pipeline.py @@ -173,6 +173,14 @@ def _approved_timestamps( def build_pipeline(environment: str) -> Any: """A dlt pipeline writing parquet to the raw zone for one environment.""" + # gcsfs resolves Application Default Credentials the same way the Cloud SQL + # connector does, and Serverless supplies none of its own. + from automated_ingestion.shared.credentials import ( + ensure_application_default_credentials, + ) + + ensure_application_default_credentials() + return dlt.pipeline( pipeline_name=f"san_acacia_{environment}", destination=dlt.destinations.filesystem( diff --git a/automated_ingestion/tests/test_credentials.py b/automated_ingestion/tests/test_credentials.py new file mode 100644 index 000000000..c7e788f88 --- /dev/null +++ b/automated_ingestion/tests/test_credentials.py @@ -0,0 +1,88 @@ +# =============================================================================== +# Copyright 2026 ross +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +""" +Credential materialization for a runtime with no metadata server. + +The failure this prevents is not subtle -- DefaultCredentialsError -- but it +only appears in Serverless, so the tests stand in for a deployment. +""" + +import json +import os + +import pytest + +from automated_ingestion.shared import credentials +from automated_ingestion.shared.credentials import ( + CREDENTIALS_ENV_VAR, + ensure_application_default_credentials, +) + +KEY = {"type": "service_account", "project_id": "waterdatainitiative-271000"} + + +@pytest.fixture(autouse=True) +def _clean(monkeypatch): + monkeypatch.setattr(credentials, "_written_path", None) + monkeypatch.delenv("GOOGLE_APPLICATION_CREDENTIALS", raising=False) + monkeypatch.delenv(CREDENTIALS_ENV_VAR, raising=False) + + +def test_existing_credentials_are_left_alone(monkeypatch): + # A developer's gcloud login must not be shadowed by a key in the + # environment; whatever is already configured wins. + monkeypatch.setenv("GOOGLE_APPLICATION_CREDENTIALS", "/existing/adc.json") + monkeypatch.setenv(CREDENTIALS_ENV_VAR, json.dumps(KEY)) + assert ensure_application_default_credentials() == "/existing/adc.json" + + +def test_no_key_configured_is_not_an_error(monkeypatch): + # Locally this is normal -- google.auth finds its own credentials. In + # Serverless it fails later, loudly, which is the correct outcome. + assert ensure_application_default_credentials() is None + assert "GOOGLE_APPLICATION_CREDENTIALS" not in os.environ + + +def test_key_is_written_and_pointed_at(monkeypatch): + monkeypatch.setenv(CREDENTIALS_ENV_VAR, json.dumps(KEY)) + path = ensure_application_default_credentials() + assert path and os.path.exists(path) + assert os.environ["GOOGLE_APPLICATION_CREDENTIALS"] == path + with open(path) as fh: + assert json.load(fh) == KEY + + +def test_key_file_is_not_world_readable(monkeypatch): + monkeypatch.setenv(CREDENTIALS_ENV_VAR, json.dumps(KEY)) + path = ensure_application_default_credentials() + assert oct(os.stat(path).st_mode)[-3:] == "600" + + +def test_repeated_calls_write_once(monkeypatch): + monkeypatch.setenv(CREDENTIALS_ENV_VAR, json.dumps(KEY)) + first = ensure_application_default_credentials() + assert ensure_application_default_credentials() == first + + +def test_a_path_instead_of_a_key_is_rejected(monkeypatch): + # Setting the variable to a filename is the obvious mistake, and it would + # otherwise fail much later inside google.auth. + monkeypatch.setenv(CREDENTIALS_ENV_VAR, "/path/to/key.json") + with pytest.raises(RuntimeError, match="not valid JSON"): + ensure_application_default_credentials() + + +# ============= EOF =============================================