From 3c5bb14b31a88023025f913847d956f83976b296 Mon Sep 17 00:00:00 2001
From: David Gamez Diaz <1192523+davidgamez@users.noreply.github.com>
Date: Thu, 8 Oct 2026 16:32:57 -0400
Subject: [PATCH] add verifier script and endpoints. add debug info to failing
http requests
---
docs/OperationsAPI.yaml | 139 +++++++
.../batch_process_dataset/src/main.py | 41 +-
.../src/scripts/download_verifier.py | 108 +++--
.../tests/test_download_failure_recording.py | 113 +++++
.../helpers/tests/test_download_debug.py | 326 +++++++++++++++
functions-python/helpers/utils.py | 385 ++++++++++++++++--
.../operations_api/.openapi-generator/FILES | 1 +
.../impl/feeds_operations_impl.py | 76 ++++
.../test_download_debug_operations_impl.py | 130 ++++++
.../check_gtfs_feed_availability.py | 16 +-
liquibase/changelog.xml | 2 +
liquibase/changes/feat_download_debug.sql | 20 +
scripts/download-verifier.sh | 124 ++++++
13 files changed, 1394 insertions(+), 87 deletions(-)
create mode 100644 functions-python/batch_process_dataset/tests/test_download_failure_recording.py
create mode 100644 functions-python/helpers/tests/test_download_debug.py
create mode 100644 functions-python/operations_api/tests/feeds_operations/impl/test_download_debug_operations_impl.py
create mode 100644 liquibase/changes/feat_download_debug.sql
create mode 100755 scripts/download-verifier.sh
diff --git a/docs/OperationsAPI.yaml b/docs/OperationsAPI.yaml
index ffd2cdf10..8b7104e66 100644
--- a/docs/OperationsAPI.yaml
+++ b/docs/OperationsAPI.yaml
@@ -202,6 +202,35 @@ paths:
description: GTFS feed not found.
500:
description: Internal server error.
+ /v1/operations/gtfs_feeds/{id}/download_debug:
+ parameters:
+ - $ref: "#/components/parameters/feed_id_path_param"
+ post:
+ description: >
+ Downloads the GTFS feed through the same code path used by the nightly dataset
+ download and returns the request and response detail of that attempt. Credential
+ values in headers and URLs are redacted. Nothing is stored.
+ tags:
+ - "operations"
+ operationId: debugGtfsFeedDownload
+ security:
+ - Authentication: []
+ parameters:
+ - $ref: "#/components/parameters/download_debug_max_bytes"
+ - $ref: "#/components/parameters/download_debug_timeout_seconds"
+ responses:
+ 200:
+ description: Diagnostics for the download attempt.
+ content:
+ application/json:
+ schema:
+ $ref: "#/components/schemas/GtfsFeedDownloadDebugResponse"
+ 400:
+ description: Invalid request parameters.
+ 404:
+ description: GTFS feed not found.
+ 500:
+ description: Internal server error.
/v1/operations/gtfs_rt_feeds/{id}:
parameters:
- $ref: "#/components/parameters/feed_id_path_param"
@@ -1680,6 +1709,92 @@ components:
nullable: true
description: Machine-readable error category when the check failed.
example: timeout
+ GtfsFeedDownloadDebugResponse:
+ type: object
+ description: >
+ Request and response detail of a single feed download attempt. Credential values
+ in headers and URLs are replaced with a redaction marker.
+ required:
+ - feed_id
+ - producer_url
+ - success
+ properties:
+ feed_id:
+ type: string
+ description: Stable identifier of the GTFS feed.
+ example: mdb-123
+ producer_url:
+ type: string
+ description: Feed URL that was requested, with credentials redacted.
+ example: https://example.com/gtfs.zip
+ resolved_url:
+ type: string
+ nullable: true
+ description: URL actually requested after credentials were applied, with credentials redacted.
+ success:
+ type: boolean
+ description: Whether the download returned a 2xx response.
+ example: false
+ status_code:
+ type: integer
+ nullable: true
+ description: HTTP status code of the final response.
+ example: 403
+ latency_ms:
+ type: integer
+ nullable: true
+ description: Time to first response, in milliseconds.
+ example: 1840
+ request_headers:
+ type: object
+ nullable: true
+ additionalProperties:
+ type: string
+ description: Headers sent with the request, with credential values redacted.
+ response_headers:
+ type: object
+ nullable: true
+ additionalProperties:
+ type: string
+ description: Headers returned by the server, with credential values redacted.
+ redirect_urls:
+ type: array
+ nullable: true
+ items:
+ type: string
+ description: URLs the request was redirected through.
+ external_ip:
+ type: string
+ nullable: true
+ description: Egress IP address observed for the request, when it could be determined.
+ example: 34.95.1.1
+ content_type:
+ type: string
+ nullable: true
+ description: Normalised Content-Type of the response.
+ example: text/html
+ is_zip:
+ type: boolean
+ nullable: true
+ description: Whether the downloaded content is a ZIP file.
+ downloaded_bytes:
+ type: integer
+ nullable: true
+ description: Number of bytes read before the download completed or hit the byte cap.
+ example: 2038
+ truncated:
+ type: boolean
+ description: Whether the download stopped at the byte cap rather than at the end of the file.
+ example: false
+ error_type:
+ type: string
+ nullable: true
+ description: Machine-readable error category when the download failed.
+ example: ValueError
+ error_message:
+ type: string
+ nullable: true
+ description: Error detail when the download failed, with credentials redacted.
GtfsFeedContinuousCoverageResponse:
type: object
description: >
@@ -3576,6 +3691,30 @@ components:
minimum: 0
default: 0
example: 0
+ download_debug_max_bytes:
+ x-operation: true
+ name: max_bytes
+ in: query
+ description: Stop reading the feed after this many bytes. Keeps large feeds from exhausting the instance.
+ required: false
+ schema:
+ type: integer
+ minimum: 1024
+ maximum: 52428800
+ default: 52428800
+ example: 1048576
+ download_debug_timeout_seconds:
+ x-operation: true
+ name: timeout_seconds
+ in: query
+ description: Connect and read timeout for the download, in seconds.
+ required: false
+ schema:
+ type: integer
+ minimum: 1
+ maximum: 300
+ default: 60
+ example: 30
limit_query_param_availability_endpoint:
x-operation: true
name: limit
diff --git a/functions-python/batch_process_dataset/src/main.py b/functions-python/batch_process_dataset/src/main.py
index 2ff117741..f305e6b1d 100644
--- a/functions-python/batch_process_dataset/src/main.py
+++ b/functions-python/batch_process_dataset/src/main.py
@@ -42,6 +42,8 @@
from shared.dataset_service.main import DatasetTraceService, DatasetTrace, Status
from shared.helpers.logger import init_logger, get_logger
from shared.helpers.utils import (
+ FeedDownloadError,
+ build_availability_check,
download_and_get_hash,
get_hash_from_file,
download_from_gcs,
@@ -139,19 +141,40 @@ def download_content(self, temporary_file_path, feed_id):
"""
Downloads the content of a URL and return the hash of the file
"""
- file_hash = download_and_get_hash(
- self.producer_url,
- file_path=temporary_file_path,
- feed_id=feed_id,
- authentication_type=self.authentication_type,
- api_key_parameter_name=self.api_key_parameter_name,
- credentials=self.feed_credentials,
- logger=self.logger,
- )
+ try:
+ file_hash = download_and_get_hash(
+ self.producer_url,
+ file_path=temporary_file_path,
+ feed_id=feed_id,
+ authentication_type=self.authentication_type,
+ api_key_parameter_name=self.api_key_parameter_name,
+ credentials=self.feed_credentials,
+ logger=self.logger,
+ )
+ except FeedDownloadError as exc:
+ if feed_id:
+ self.record_download_failure(feed_id, exc.diagnostics)
+ raise
self.logger.info("hash is: %s", file_hash)
is_zip = zipfile.is_zipfile(temporary_file_path)
return file_hash, is_zip
+ @with_db_session
+ def record_download_failure(self, feed_id, diagnostics, db_session: Session = None):
+ """
+ Stores the diagnostics of a failed download in gtfs_feed_availability_check.
+ Never raises: a storage failure must not mask the original download error.
+ """
+ try:
+ db_session.add(
+ build_availability_check(
+ diagnostics, feed_id=feed_id, source="dataset_download"
+ )
+ )
+ db_session.commit()
+ except Exception as exc:
+ self.logger.error("Could not store download diagnostics: %s", exc)
+
def upload_dataset_zip_to_storage(
self,
source_file_path,
diff --git a/functions-python/batch_process_dataset/src/scripts/download_verifier.py b/functions-python/batch_process_dataset/src/scripts/download_verifier.py
index 08a8be08c..db2a0a534 100644
--- a/functions-python/batch_process_dataset/src/scripts/download_verifier.py
+++ b/functions-python/batch_process_dataset/src/scripts/download_verifier.py
@@ -1,28 +1,32 @@
+import argparse
+import json
import logging
import os
from main import DatasetProcessor
from gcp_storage_emulator.server import create_server
+from shared.helpers.utils import FeedDownloadError
HOST = "localhost"
PORT = 9023
BUCKET_NAME = "verifier"
-PRODUCER_URL = "https://example.com/dataset.zip" # Replace with actual producer URL
-def verify_download_content(producer_url: str):
+def verify_download_content(producer_url: str, feed_id=None):
"""
- Verifies the download_content is able to retrieve the file
- This is useful to simulate the download code locally and test issues related with user-agent and downloaded content.
- Not supported authenticated feeds currently.
- """
- logging.info("Verifying downloaded content... (not implemented)")
+ Verifies the download_content is able to retrieve the file.
+ This is useful to simulate the download code locally and test issues related with
+ user-agent and downloaded content.
+ Authenticated feeds are not supported currently.
- logging.info(f"Producer URL: {producer_url}")
+ Pass feed_id to apply the per-feed feed_download/http_headers override from the
+ config database, as the nightly run does.
+ """
+ logging.info("Producer URL: %s", producer_url)
processor = DatasetProcessor(
producer_url=producer_url,
- feed_id=None,
+ feed_id=feed_id,
feed_stable_id=None,
execution_id=None,
latest_hash=None,
@@ -32,20 +36,26 @@ def verify_download_content(producer_url: str):
public_hosted_datasets_url=None,
)
tempfile = processor.generate_temp_filename()
- logging.info(f"Temp filename: {tempfile}")
- file_hash, is_zip = processor.download_content(tempfile, "feed_id")
- logging.info(f"Downloaded file from {producer_url} is a valid ZIP file: {is_zip}")
- logging.info(f"File hash: {file_hash}")
+ logging.info("Temp filename: %s", tempfile)
+ try:
+ file_hash, is_zip = processor.download_content(tempfile, feed_id)
+ except FeedDownloadError as exc:
+ logging.error("Download failed")
+ logging.error(json.dumps(exc.diagnostics.as_dict(), indent=2, default=str))
+ raise
+ logging.info(
+ "Downloaded file from %s is a valid ZIP file: %s", producer_url, is_zip
+ )
+ logging.info("File hash: %s", file_hash)
def verify_upload_dataset(producer_url: str):
"""
Verifies the upload_dataset is able to upload the dataset to the GCP storage emulator.
- This is useful to simulate the upload code locally and test issues related with user-agent and uploaded content.
- This function also tests the DatasetProcessor class methods for generating a temporary filename
- and uploading the dataset.
- :param producer_url:
- :return:
+ This is useful to simulate the upload code locally and test issues related with
+ user-agent and uploaded content.
+ This function also tests the DatasetProcessor class methods for generating a temporary
+ filename and uploading the dataset.
"""
processor = DatasetProcessor(
producer_url=producer_url,
@@ -59,31 +69,55 @@ def verify_upload_dataset(producer_url: str):
public_hosted_datasets_url=None,
)
tempfile = processor.generate_temp_filename()
- logging.info(f"Temp filename: {tempfile}")
+ logging.info("Temp filename: %s", tempfile)
dataset_file = processor.transfer_dataset("feed_id_2126", False)
- logging.info(f"Dataset File: {dataset_file}")
+ logging.info("Dataset File: %s", dataset_file)
-if __name__ == "__main__":
+def parse_args():
+ parser = argparse.ArgumentParser(
+ description="Download a GTFS feed through the shared download code and print the "
+ "request and response detail of the attempt."
+ )
+ parser.add_argument("--url", required=True, help="Producer URL of the feed.")
+ parser.add_argument(
+ "--feed_id",
+ default=None,
+ help="Feed UUID, to apply the per-feed http_headers config override.",
+ )
+ parser.add_argument(
+ "--skip-upload",
+ action="store_true",
+ help="Only verify the download, skipping the GCP storage emulator.",
+ )
+ return parser.parse_args()
+
+
+def main():
logging.basicConfig(level=logging.INFO)
- # Replace with actual producer URL
+ args = parse_args()
+
+ os.environ["WORKING_DIR"] = "/tmp/verifier"
+ os.makedirs(os.environ["WORKING_DIR"], exist_ok=True)
+
+ if args.skip_upload:
+ verify_download_content(producer_url=args.url, feed_id=args.feed_id)
+ logging.info("Download content verification completed successfully.")
+ return
+
+ os.environ["STORAGE_EMULATOR_HOST"] = f"http://{HOST}:{PORT}"
+ server = create_server(
+ host=HOST, port=PORT, in_memory=False, default_bucket=BUCKET_NAME
+ )
+ server.start()
try:
- os.environ["STORAGE_EMULATOR_HOST"] = f"http://{HOST}:{PORT}"
- os.environ["WORKING_DIR"] = "/tmp/verifier"
- # create working dir if not exists
- if not os.path.exists(os.environ["WORKING_DIR"]):
- os.makedirs(os.environ["WORKING_DIR"])
-
- server = create_server(
- host=HOST, port=PORT, in_memory=False, default_bucket=BUCKET_NAME
- )
- server.start()
-
- verify_download_content(producer_url=PRODUCER_URL)
+ verify_download_content(producer_url=args.url, feed_id=args.feed_id)
logging.info("Download content verification completed successfully.")
- verify_upload_dataset(producer_url=PRODUCER_URL)
- except Exception as e:
- logging.error(f"Error verifying download content: {e}")
+ verify_upload_dataset(producer_url=args.url)
finally:
server.stop()
logging.info("Verification completed.")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/functions-python/batch_process_dataset/tests/test_download_failure_recording.py b/functions-python/batch_process_dataset/tests/test_download_failure_recording.py
new file mode 100644
index 000000000..f3582831d
--- /dev/null
+++ b/functions-python/batch_process_dataset/tests/test_download_failure_recording.py
@@ -0,0 +1,113 @@
+import unittest
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from main import DatasetProcessor
+from shared.helpers.utils import FeedDownloadError, FeedRequestDiagnostics
+
+
+def _processor(feed_id="feed-1"):
+ return DatasetProcessor(
+ producer_url="https://example.com/gtfs.zip",
+ feed_id=feed_id,
+ feed_stable_id="mdb-1",
+ execution_id=None,
+ latest_hash=None,
+ bucket_name=None,
+ authentication_type=0,
+ api_key_parameter_name=None,
+ public_hosted_datasets_url=None,
+ )
+
+
+def _diagnostics():
+ return FeedRequestDiagnostics(
+ request_url="https://example.com/gtfs.zip",
+ resolved_url="https://example.com/gtfs.zip",
+ request_headers={"User-Agent": "agent"},
+ response_headers={"Server": "cloudflare"},
+ status_code=403,
+ latency_ms=1840,
+ redirect_urls=[],
+ external_ip="34.1.2.3",
+ content_type="text/html",
+ is_zip=False,
+ error_type="ValueError",
+ error_message="Invalid HTTP response code: [403]",
+ )
+
+
+class TestDownloadFailureRecording(unittest.TestCase):
+ def test_failure_records_one_row_and_reraises(self):
+ processor = _processor()
+ diagnostics = _diagnostics()
+
+ with patch(
+ "main.download_and_get_hash",
+ side_effect=FeedDownloadError("boom", diagnostics),
+ ):
+ with patch.object(processor, "record_download_failure") as mock_record:
+ with pytest.raises(FeedDownloadError):
+ processor.download_content("/tmp/does-not-matter", "feed-1")
+
+ mock_record.assert_called_once()
+ args = mock_record.call_args[0]
+ self.assertEqual(args[0], "feed-1")
+ self.assertIs(args[1], diagnostics)
+
+ def test_success_records_nothing(self):
+ processor = _processor()
+
+ with patch("main.download_and_get_hash", return_value="abc123"):
+ with patch("main.zipfile.is_zipfile", return_value=True):
+ with patch.object(processor, "record_download_failure") as mock_record:
+ file_hash, is_zip = processor.download_content(
+ "/tmp/does-not-matter", "feed-1"
+ )
+
+ self.assertEqual(file_hash, "abc123")
+ self.assertTrue(is_zip)
+ mock_record.assert_not_called()
+
+ def test_no_feed_id_skips_recording(self):
+ """The local verifier runs without a feed id, so there is no row to attach."""
+ processor = _processor(feed_id=None)
+
+ with patch(
+ "main.download_and_get_hash",
+ side_effect=FeedDownloadError("boom", _diagnostics()),
+ ):
+ with patch.object(processor, "record_download_failure") as mock_record:
+ with pytest.raises(FeedDownloadError):
+ processor.download_content("/tmp/does-not-matter", None)
+
+ mock_record.assert_not_called()
+
+ def test_storage_failure_does_not_mask_download_error(self):
+ processor = _processor()
+
+ session = MagicMock()
+ session.commit.side_effect = Exception("db down")
+
+ with patch(
+ "main.download_and_get_hash",
+ side_effect=FeedDownloadError("boom", _diagnostics()),
+ ):
+ with patch("main.build_availability_check", return_value=MagicMock()):
+ with pytest.raises(FeedDownloadError):
+ processor.download_content("/tmp/does-not-matter", "feed-1")
+
+ def test_recorded_row_is_tagged_as_dataset_download(self):
+ processor = _processor()
+ diagnostics = _diagnostics()
+ session = MagicMock()
+
+ with patch("main.build_availability_check") as mock_build:
+ processor.record_download_failure("feed-1", diagnostics, db_session=session)
+
+ mock_build.assert_called_once_with(
+ diagnostics, feed_id="feed-1", source="dataset_download"
+ )
+ session.add.assert_called_once()
+ session.commit.assert_called_once()
diff --git a/functions-python/helpers/tests/test_download_debug.py b/functions-python/helpers/tests/test_download_debug.py
new file mode 100644
index 000000000..997eefe4c
--- /dev/null
+++ b/functions-python/helpers/tests/test_download_debug.py
@@ -0,0 +1,326 @@
+import os
+import unittest
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+import utils
+from utils import (
+ REDACTED,
+ FeedDownloadError,
+ FeedRequestDiagnostics,
+ build_availability_check,
+ download_feed,
+ get_external_ip,
+ sanitize_headers,
+ sanitize_url,
+)
+
+
+class TestSanitizeHeaders(unittest.TestCase):
+ def test_redacts_known_secret_headers(self):
+ result = sanitize_headers(
+ {
+ "User-Agent": "agent",
+ "Authorization": "Bearer abc",
+ "Cookie": "session=1",
+ "X-Api-Key": "secret",
+ }
+ )
+ self.assertEqual(result["User-Agent"], "agent")
+ self.assertEqual(result["Authorization"], REDACTED)
+ self.assertEqual(result["Cookie"], REDACTED)
+ self.assertEqual(result["X-Api-Key"], REDACTED)
+
+ def test_redacts_api_key_parameter_name_header(self):
+ result = sanitize_headers(
+ {"User-Agent": "agent", "my-feed-token": "secret"},
+ api_key_parameter_name="my-feed-token",
+ )
+ self.assertEqual(result["my-feed-token"], REDACTED)
+ self.assertEqual(result["User-Agent"], "agent")
+
+ def test_header_names_are_preserved(self):
+ result = sanitize_headers({"Authorization": "Bearer abc"})
+ self.assertIn("Authorization", result)
+
+ def test_none_returns_none(self):
+ self.assertIsNone(sanitize_headers(None))
+
+
+class TestSanitizeUrl(unittest.TestCase):
+ def test_redacts_api_key_query_parameter(self):
+ result = sanitize_url(
+ "https://example.com/gtfs.zip?operator=CE&token=secret",
+ api_key_parameter_name="token",
+ )
+ self.assertIn("operator=CE", result)
+ self.assertNotIn("secret", result)
+
+ def test_redacts_known_secret_query_parameters(self):
+ result = sanitize_url("https://example.com/gtfs.zip?api_key=secret&x=1")
+ self.assertIn("x=1", result)
+ self.assertNotIn("secret", result)
+
+ def test_strips_userinfo(self):
+ result = sanitize_url("https://user:password@example.com/gtfs.zip")
+ self.assertNotIn("password", result)
+ self.assertIn("example.com", result)
+
+ def test_url_without_credentials_is_unchanged(self):
+ url = "https://example.com/gtfs.zip"
+ self.assertEqual(sanitize_url(url), url)
+
+ def test_none_returns_none(self):
+ self.assertIsNone(sanitize_url(None))
+
+
+class TestGetExternalIp(unittest.TestCase):
+ def setUp(self):
+ utils._external_ip_cache = None
+
+ def tearDown(self):
+ utils._external_ip_cache = None
+
+ def test_returns_ip_on_success(self):
+ response = MagicMock(status=200, data=b"34.1.2.3\n")
+ pool = MagicMock()
+ pool.request.return_value = response
+ pool.__enter__ = lambda s: pool
+ pool.__exit__ = MagicMock(return_value=False)
+
+ with patch("utils.urllib3.PoolManager", return_value=pool):
+ self.assertEqual(get_external_ip(), "34.1.2.3")
+
+ def test_returns_none_when_lookup_fails(self):
+ with patch("utils.urllib3.PoolManager", side_effect=Exception("boom")):
+ self.assertIsNone(get_external_ip())
+
+ def test_result_is_cached(self):
+ response = MagicMock(status=200, data=b"34.1.2.3")
+ pool = MagicMock()
+ pool.request.return_value = response
+ pool.__enter__ = lambda s: pool
+ pool.__exit__ = MagicMock(return_value=False)
+
+ with patch("utils.urllib3.PoolManager", return_value=pool) as mock_pool:
+ get_external_ip()
+ get_external_ip()
+ self.assertEqual(mock_pool.call_count, 1)
+
+
+def _mock_response(status=200, headers=None, chunks=None):
+ response = MagicMock()
+ response.status = status
+ response.headers = headers if headers is not None else {}
+ response.retries = MagicMock(history=[])
+ response.__enter__ = lambda s: response
+ response.__exit__ = MagicMock(return_value=False)
+ data = list(chunks if chunks is not None else [b""])
+
+ def read(_size):
+ return data.pop(0) if data else b""
+
+ response.read.side_effect = read
+ return response
+
+
+def _mock_pool(response):
+ pool = MagicMock()
+ pool.request.return_value = response
+ pool.__enter__ = lambda s: pool
+ pool.__exit__ = MagicMock(return_value=False)
+ return pool
+
+
+@patch("shared.common.config_reader.get_config_value", return_value=None)
+class TestDownloadFeed(unittest.TestCase):
+ def setUp(self):
+ self.file_path = os.path.join(
+ os.path.dirname(__file__), "download_feed_test.bin"
+ )
+ utils._external_ip_cache = None
+
+ def tearDown(self):
+ if os.path.exists(self.file_path):
+ os.remove(self.file_path)
+ utils._external_ip_cache = None
+
+ def test_diagnostics_populated_on_success(self, _mock_config):
+ response = _mock_response(
+ status=200,
+ headers={"Content-Type": "application/zip", "Server": "nginx"},
+ chunks=[b"PK\x03\x04data", b""],
+ )
+ with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)):
+ result = download_feed(
+ "https://example.com/gtfs.zip", self.file_path, feed_id="feed-1"
+ )
+
+ diagnostics = result.diagnostics
+ self.assertTrue(diagnostics.success)
+ self.assertEqual(diagnostics.status_code, 200)
+ self.assertEqual(diagnostics.content_type, "application/zip")
+ self.assertEqual(diagnostics.downloaded_bytes, 8)
+ self.assertEqual(diagnostics.response_headers["Server"], "nginx")
+ self.assertIn("User-Agent", diagnostics.request_headers)
+ self.assertIsNotNone(diagnostics.latency_ms)
+ self.assertIsNone(diagnostics.error_type)
+ self.assertIsNotNone(result.file_hash)
+
+ def test_external_ip_not_looked_up_on_success(self, _mock_config):
+ response = _mock_response(chunks=[b"data", b""])
+ with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)):
+ with patch("utils.get_external_ip") as mock_ip:
+ download_feed("https://example.com/gtfs.zip", self.file_path)
+ mock_ip.assert_not_called()
+
+ def test_non_2xx_raises_with_diagnostics(self, _mock_config):
+ response = _mock_response(
+ status=403, headers={"Content-Type": "text/html", "Server": "cloudflare"}
+ )
+ with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)):
+ with patch("utils.get_external_ip", return_value="34.1.2.3"):
+ with pytest.raises(FeedDownloadError) as exc_info:
+ download_feed("https://example.com/gtfs.zip", self.file_path)
+
+ diagnostics = exc_info.value.diagnostics
+ self.assertFalse(diagnostics.success)
+ self.assertEqual(diagnostics.status_code, 403)
+ self.assertEqual(diagnostics.content_type, "text/html")
+ self.assertEqual(diagnostics.response_headers["Server"], "cloudflare")
+ self.assertEqual(diagnostics.external_ip, "34.1.2.3")
+ self.assertEqual(diagnostics.error_type, "ValueError")
+
+ def test_network_error_raises_with_diagnostics(self, _mock_config):
+ with patch("utils.urllib3.PoolManager", side_effect=Exception("Network error")):
+ with patch("utils.get_external_ip", return_value=None):
+ with pytest.raises(FeedDownloadError) as exc_info:
+ download_feed("https://example.com/gtfs.zip", self.file_path)
+
+ diagnostics = exc_info.value.diagnostics
+ self.assertFalse(diagnostics.success)
+ self.assertIsNone(diagnostics.status_code)
+ self.assertEqual(diagnostics.error_type, "Exception")
+ self.assertEqual(diagnostics.error_message, "Network error")
+
+ def test_credentials_never_appear_in_diagnostics(self, _mock_config):
+ response = _mock_response(status=403)
+ with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)):
+ with patch("utils.get_external_ip", return_value=None):
+ with pytest.raises(FeedDownloadError) as exc_info:
+ download_feed(
+ "https://example.com/gtfs.zip",
+ self.file_path,
+ authentication_type=1,
+ api_key_parameter_name="token",
+ credentials="super-secret",
+ )
+
+ serialized = str(exc_info.value.diagnostics.as_dict())
+ self.assertNotIn("super-secret", serialized)
+ self.assertIn(REDACTED, serialized)
+
+ def test_max_bytes_truncates_and_returns_no_hash(self, _mock_config):
+ response = _mock_response(chunks=[b"a" * 10, b"a" * 10, b""])
+ with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)):
+ result = download_feed(
+ "https://example.com/gtfs.zip",
+ self.file_path,
+ chunk_size=10,
+ max_bytes=10,
+ )
+
+ self.assertIsNone(result.file_hash)
+ self.assertEqual(result.diagnostics.downloaded_bytes, 10)
+
+ def test_redirect_urls_are_recorded(self, _mock_config):
+ response = _mock_response(chunks=[b"data", b""])
+ response.retries = MagicMock(
+ history=[MagicMock(redirect_location="https://cdn.example.com/gtfs.zip")]
+ )
+ with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)):
+ result = download_feed("https://example.com/gtfs.zip", self.file_path)
+
+ self.assertEqual(
+ result.diagnostics.redirect_urls, ["https://cdn.example.com/gtfs.zip"]
+ )
+
+
+class TestBuildAvailabilityCheck(unittest.TestCase):
+ def test_maps_diagnostics_onto_row(self):
+ diagnostics = FeedRequestDiagnostics(
+ request_url="https://example.com/gtfs.zip",
+ resolved_url="https://example.com/gtfs.zip",
+ request_headers={"User-Agent": "agent"},
+ response_headers={"Server": "cloudflare"},
+ status_code=403,
+ latency_ms=1840,
+ redirect_urls=["https://cdn.example.com/gtfs.zip"],
+ external_ip="34.1.2.3",
+ content_type="text/html",
+ is_zip=False,
+ error_type="ValueError",
+ error_message="Invalid HTTP response code: [403]",
+ )
+
+ row = build_availability_check(
+ diagnostics, feed_id="feed-1", source="dataset_download"
+ )
+
+ self.assertEqual(row.feed_id, "feed-1")
+ self.assertEqual(row.source, "dataset_download")
+ self.assertEqual(row.request_type, "http_get")
+ self.assertEqual(row.status_code, 403)
+ self.assertEqual(row.external_ip, "34.1.2.3")
+ self.assertEqual(row.response_headers, {"Server": "cloudflare"})
+ self.assertFalse(row.success)
+
+ def test_success_flag_follows_status_code(self):
+ diagnostics = FeedRequestDiagnostics(status_code=200)
+ row = build_availability_check(
+ diagnostics, feed_id="feed-1", source="dataset_download"
+ )
+ self.assertTrue(row.success)
+
+
+@patch("shared.common.config_reader.get_config_value", return_value=None)
+class TestErrorMessageSanitization(unittest.TestCase):
+ def setUp(self):
+ self.file_path = os.path.join(os.path.dirname(__file__), "err_test.bin")
+ utils._external_ip_cache = None
+
+ def tearDown(self):
+ if os.path.exists(self.file_path):
+ os.remove(self.file_path)
+ utils._external_ip_cache = None
+
+ def test_userinfo_in_producer_url_is_redacted_in_error(self, _mock_config):
+ url = "https://user:password@example.com/gtfs.zip"
+ with patch(
+ "utils.urllib3.PoolManager", side_effect=Exception(f"failed for {url}")
+ ):
+ with patch("utils.get_external_ip", return_value=None):
+ with pytest.raises(FeedDownloadError) as exc_info:
+ download_feed(url, self.file_path)
+
+ self.assertNotIn("password", exc_info.value.diagnostics.error_message)
+
+ def test_credentials_in_resolved_url_are_redacted_in_error(self, _mock_config):
+ def boom(*_args, **_kwargs):
+ raise Exception(
+ "failed for https://example.com/gtfs.zip?token=super-secret"
+ )
+
+ with patch("utils.urllib3.PoolManager", side_effect=boom):
+ with patch("utils.get_external_ip", return_value=None):
+ with pytest.raises(FeedDownloadError) as exc_info:
+ download_feed(
+ "https://example.com/gtfs.zip",
+ self.file_path,
+ authentication_type=1,
+ api_key_parameter_name="token",
+ credentials="super-secret",
+ )
+
+ self.assertNotIn("super-secret", exc_info.value.diagnostics.error_message)
diff --git a/functions-python/helpers/utils.py b/functions-python/helpers/utils.py
index 259cf06dd..c117566ff 100644
--- a/functions-python/helpers/utils.py
+++ b/functions-python/helpers/utils.py
@@ -18,10 +18,13 @@
import os
import ssl
import time
+import zipfile
import urllib3.exceptions
+from dataclasses import asdict, dataclass
from datetime import date, datetime, timezone
from logging import Logger
from typing import Optional
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
import requests
import urllib3
@@ -142,6 +145,41 @@ def create_feed_ssl_context(trusted_certs: bool = False):
return ctx
+def get_feed_credentials(stable_id: str) -> Optional[str]:
+ """Return the API credential for a feed from the FEEDS_CREDENTIALS env var, or None."""
+ try:
+ import json
+
+ feeds_credentials = json.loads(os.getenv("FEEDS_CREDENTIALS", "{}"))
+ return feeds_credentials.get(stable_id, None)
+ except Exception as exc:
+ logging.warning("Could not parse FEEDS_CREDENTIALS: %s", exc)
+ return None
+
+
+def _get_http_headers_override(feed_id: Optional[str]) -> Optional[dict]:
+ """Return the feed_download/http_headers config override, or None.
+
+ Returns None when the config database is unreachable so that callers fall back to
+ the default headers instead of failing. This keeps the local verifier script usable
+ without a database.
+ """
+ try:
+ from shared.common.config_reader import get_config_value
+
+ return get_config_value(
+ namespace="feed_download", key="http_headers", feed_id=feed_id
+ )
+ except Exception as exc:
+ logging.warning(
+ "Could not read feed_download/http_headers for feed %s, "
+ "falling back to default headers: %s",
+ feed_id,
+ exc,
+ )
+ return None
+
+
def build_feed_request_params(
url: str,
feed_id: Optional[str] = None,
@@ -161,11 +199,7 @@ def build_feed_request_params(
Returns:
(headers, resolved_url) ready to pass to any HTTP method.
"""
- from shared.common.config_reader import get_config_value
-
- headers = get_config_value(
- namespace="feed_download", key="http_headers", feed_id=feed_id
- )
+ headers = _get_http_headers_override(feed_id)
if headers is None:
headers = {
"User-Agent": "Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) "
@@ -249,6 +283,201 @@ def _sanitize_error_message(
return message
+REDACTED = "***REDACTED***"
+
+_SECRET_HEADER_NAMES = frozenset(
+ {
+ "authorization",
+ "proxy-authorization",
+ "cookie",
+ "set-cookie",
+ "x-api-key",
+ "api-key",
+ "apikey",
+ "token",
+ "access-token",
+ "x-access-token",
+ "x-auth-token",
+ }
+)
+
+
+def _is_secret_name(name: str, api_key_parameter_name: Optional[str] = None) -> bool:
+ """Return True when a header or query parameter name carries a credential.
+
+ Underscores and hyphens are treated as equivalent so that both api_key and api-key
+ are matched.
+ """
+ lowered = name.strip().lower()
+ if lowered.replace("_", "-") in _SECRET_HEADER_NAMES:
+ return True
+ if not api_key_parameter_name:
+ return False
+ return lowered.replace("_", "-") == api_key_parameter_name.strip().lower().replace(
+ "_", "-"
+ )
+
+
+def sanitize_headers(
+ headers: Optional[dict], api_key_parameter_name: Optional[str] = None
+) -> Optional[dict]:
+ """Return a copy of headers with credential values replaced by REDACTED.
+
+ Names are always preserved, only values are redacted.
+ """
+ if headers is None:
+ return None
+ return {
+ name: (
+ REDACTED if _is_secret_name(name, api_key_parameter_name) else str(value)
+ )
+ for name, value in headers.items()
+ }
+
+
+def sanitize_url(
+ url: Optional[str], api_key_parameter_name: Optional[str] = None
+) -> Optional[str]:
+ """Return url with credential query-parameter values and userinfo redacted."""
+ if not url:
+ return url
+ try:
+ parsed = urlsplit(url)
+ except ValueError:
+ return REDACTED
+
+ netloc = parsed.netloc
+ if "@" in netloc:
+ netloc = f"{REDACTED}@{netloc.rsplit('@', 1)[1]}"
+
+ query = parsed.query
+ if query:
+ query = urlencode(
+ [
+ (
+ name,
+ (
+ REDACTED
+ if _is_secret_name(name, api_key_parameter_name)
+ else value
+ ),
+ )
+ for name, value in parse_qsl(query, keep_blank_values=True)
+ ],
+ safe="*",
+ )
+
+ return urlunsplit((parsed.scheme, netloc, parsed.path, query, parsed.fragment))
+
+
+_EXTERNAL_IP_URL = "https://checkip.amazonaws.com"
+_external_ip_cache: Optional[str] = None
+
+
+def get_external_ip(timeout_seconds: int = 3) -> Optional[str]:
+ """Return the egress IP address of this process, or None if it cannot be determined.
+
+ The result is cached for the lifetime of the process. Never raises.
+ """
+ global _external_ip_cache
+ if _external_ip_cache is not None:
+ return _external_ip_cache
+ try:
+ with urllib3.PoolManager() as http:
+ response = http.request(
+ "GET",
+ _EXTERNAL_IP_URL,
+ timeout=urllib3.Timeout(connect=timeout_seconds, read=timeout_seconds),
+ retries=False,
+ )
+ if response.status == 200:
+ _external_ip_cache = response.data.decode("utf-8").strip() or None
+ except Exception as exc:
+ logging.warning("Could not determine external IP address: %s", exc)
+ return _external_ip_cache
+
+
+@dataclass
+class FeedRequestDiagnostics:
+ """Request and response detail for a feed download attempt, safe to log and store."""
+
+ request_url: Optional[str] = None
+ resolved_url: Optional[str] = None
+ request_headers: Optional[dict] = None
+ response_headers: Optional[dict] = None
+ status_code: Optional[int] = None
+ latency_ms: Optional[int] = None
+ redirect_urls: Optional[list] = None
+ external_ip: Optional[str] = None
+ content_type: Optional[str] = None
+ is_zip: Optional[bool] = None
+ downloaded_bytes: Optional[int] = None
+ error_type: Optional[str] = None
+ error_message: Optional[str] = None
+
+ @property
+ def success(self) -> bool:
+ return (
+ self.error_type is None
+ and self.status_code is not None
+ and 200 <= self.status_code < 300
+ )
+
+ def as_dict(self) -> dict:
+ return asdict(self)
+
+
+class FeedDownloadError(Exception):
+ """Raised when a feed download fails, carrying the diagnostics of the attempt."""
+
+ def __init__(self, message: str, diagnostics: FeedRequestDiagnostics):
+ super().__init__(message)
+ self.diagnostics = diagnostics
+
+
+@dataclass
+class FeedDownloadResult:
+ """Outcome of a successful feed download."""
+
+ file_hash: Optional[str]
+ is_zip: Optional[bool]
+ diagnostics: FeedRequestDiagnostics
+
+
+def build_availability_check(
+ diagnostics: FeedRequestDiagnostics,
+ feed_id: str,
+ source: str,
+ request_type: str = "http_get",
+ checked_at: Optional[datetime] = None,
+):
+ """Map diagnostics onto a GtfsFeedAvailabilityCheck row.
+
+ source is one of 'availability_check' or 'dataset_download'.
+ """
+ from shared.database_gen.sqlacodegen_models import GtfsFeedAvailabilityCheck
+
+ return GtfsFeedAvailabilityCheck(
+ feed_id=feed_id,
+ checked_at=checked_at or datetime.now(timezone.utc),
+ request_url=diagnostics.request_url,
+ resolved_url=diagnostics.resolved_url,
+ request_type=request_type,
+ status_code=diagnostics.status_code,
+ latency_ms=diagnostics.latency_ms,
+ error_message=diagnostics.error_message,
+ error_type=diagnostics.error_type,
+ success=diagnostics.success,
+ content_type=diagnostics.content_type,
+ is_zip=diagnostics.is_zip,
+ source=source,
+ request_headers=diagnostics.request_headers,
+ response_headers=diagnostics.response_headers,
+ redirect_urls=diagnostics.redirect_urls,
+ external_ip=diagnostics.external_ip,
+ )
+
+
def _execute_http_request(
method: str,
url: str,
@@ -410,6 +639,7 @@ def perform_request(
feed_id=feed_id,
checked_at=checked_at,
request_url=producer_url,
+ resolved_url=sanitize_url(resolved_url, api_key_parameter_name),
request_type=request_type,
status_code=status_code,
latency_ms=latency_ms,
@@ -418,10 +648,16 @@ def perform_request(
success=success,
content_type=content_type,
is_zip=is_zip,
+ source="availability_check",
+ request_headers=sanitize_headers(headers, api_key_parameter_name),
+ response_headers=sanitize_headers(dict(resp_headers) if resp_headers else None),
+ redirect_urls=[
+ sanitize_url(u, api_key_parameter_name) for u in (redirect_urls or [])
+ ],
)
-def download_and_get_hash(
+def download_feed(
url,
file_path,
hash_algorithm="sha256",
@@ -432,48 +668,143 @@ def download_and_get_hash(
credentials=None,
logger=None,
trusted_certs=False, # If True, disables SSL verification
-):
- """
- Downloads the content of a URL and stores it in a file and returns the hash of the file
+ max_bytes: Optional[int] = None,
+ timeout_seconds: Optional[int] = None,
+) -> FeedDownloadResult:
+ """Download a feed to file_path and return its hash along with request diagnostics.
+
+ max_bytes stops the download once that many bytes have been read, which lets callers
+ inspect the response without storing the whole dataset. The returned hash is None in
+ that case because the file is incomplete.
+
+ Raises FeedDownloadError carrying a FeedRequestDiagnostics on any failure.
"""
logger = logger or logging.getLogger(__name__)
- try:
- hash_object = hashlib.new(hash_algorithm)
+ producer_url = url
+ resolved_url = url
+ hash_object = hashlib.new(hash_algorithm)
+ diagnostics = FeedRequestDiagnostics(
+ request_url=sanitize_url(producer_url, api_key_parameter_name)
+ )
+ truncated = False
+ try:
ctx = create_feed_ssl_context(trusted_certs=trusted_certs)
- headers, url = build_feed_request_params(
- url,
+ headers, resolved_url = build_feed_request_params(
+ producer_url,
feed_id=feed_id,
authentication_type=authentication_type,
api_key_parameter_name=api_key_parameter_name,
credentials=credentials,
)
+ diagnostics.resolved_url = sanitize_url(resolved_url, api_key_parameter_name)
+ diagnostics.request_headers = sanitize_headers(headers, api_key_parameter_name)
+
+ timeout = (
+ urllib3.Timeout(connect=timeout_seconds, read=timeout_seconds)
+ if timeout_seconds
+ else None
+ )
+ request_kwargs = {"timeout": timeout} if timeout else {}
with urllib3.PoolManager(ssl_context=ctx) as http:
+ start = time.monotonic()
with http.request(
- "GET", url, preload_content=False, headers=headers, redirect=True
+ "GET",
+ resolved_url,
+ preload_content=False,
+ headers=headers,
+ redirect=True,
+ **request_kwargs,
) as r, open(file_path, "wb") as out_file:
- if 200 <= r.status < 300:
- logger.info(f"HTTP response code: [{r.status}]")
- while True:
- data = r.read(chunk_size)
- if not data:
- break
- hash_object.update(data)
- out_file.write(data)
- r.release_conn()
- else:
+ diagnostics.latency_ms = int((time.monotonic() - start) * 1000)
+ diagnostics.status_code = r.status
+ diagnostics.response_headers = sanitize_headers(dict(r.headers))
+ diagnostics.redirect_urls = [
+ sanitize_url(h.redirect_location, api_key_parameter_name)
+ for h in (r.retries.history or [])
+ if h.redirect_location
+ ]
+ diagnostics.content_type = _parse_content_type(
+ r.headers.get("Content-Type")
+ )
+
+ if not 200 <= r.status < 300:
raise ValueError(f"Invalid HTTP response code: [{r.status}]")
- return hash_object.hexdigest()
- except Exception as e:
+
+ logger.info(f"HTTP response code: [{r.status}]")
+ total = 0
+ while True:
+ data = r.read(chunk_size)
+ if not data:
+ break
+ hash_object.update(data)
+ out_file.write(data)
+ total += len(data)
+ if max_bytes is not None and total >= max_bytes:
+ truncated = True
+ break
+ r.release_conn()
+ diagnostics.downloaded_bytes = total
+
+ diagnostics.is_zip = zipfile.is_zipfile(file_path) or (
+ _is_zip_from_content_type(diagnostics.content_type) or False
+ )
+ return FeedDownloadResult(
+ file_hash=None if truncated else hash_object.hexdigest(),
+ is_zip=diagnostics.is_zip,
+ diagnostics=diagnostics,
+ )
+ except Exception as exc:
+ diagnostics.error_type = type(exc).__name__
+ # Collapse the credential-bearing resolved URL onto the producer URL, then the
+ # producer URL onto its sanitized form, so neither can leak through the message.
+ message = _sanitize_error_message(str(exc), resolved_url, producer_url)
+ if message and producer_url != diagnostics.request_url:
+ message = message.replace(producer_url, diagnostics.request_url)
+ diagnostics.error_message = message
+ diagnostics.external_ip = get_external_ip()
+ logger.warning(
+ "Feed download failed for %s: %s",
+ diagnostics.request_url,
+ diagnostics.as_dict(),
+ )
if os.path.exists(file_path):
try:
os.remove(file_path)
except Exception:
logger.error(f"Delete file: [{file_path}]")
+ raise FeedDownloadError(str(exc), diagnostics) from exc
- raise e
+
+def download_and_get_hash(
+ url,
+ file_path,
+ hash_algorithm="sha256",
+ chunk_size=8192,
+ feed_id=None,
+ authentication_type=0,
+ api_key_parameter_name=None,
+ credentials=None,
+ logger=None,
+ trusted_certs=False, # If True, disables SSL verification
+):
+ """
+ Downloads the content of a URL and stores it in a file and returns the hash of the file
+ """
+ return download_feed(
+ url,
+ file_path,
+ hash_algorithm=hash_algorithm,
+ chunk_size=chunk_size,
+ feed_id=feed_id,
+ authentication_type=authentication_type,
+ api_key_parameter_name=api_key_parameter_name,
+ credentials=credentials,
+ logger=logger,
+ trusted_certs=trusted_certs,
+ ).file_hash
def create_http_task(
diff --git a/functions-python/operations_api/.openapi-generator/FILES b/functions-python/operations_api/.openapi-generator/FILES
index e317b1b7f..d0fed2e31 100644
--- a/functions-python/operations_api/.openapi-generator/FILES
+++ b/functions-python/operations_api/.openapi-generator/FILES
@@ -45,6 +45,7 @@ src/feeds_gen/models/gtfs_feed_continuous_coverage.py
src/feeds_gen/models/gtfs_feed_continuous_coverage_boundary.py
src/feeds_gen/models/gtfs_feed_continuous_coverage_file.py
src/feeds_gen/models/gtfs_feed_continuous_coverage_response.py
+src/feeds_gen/models/gtfs_feed_download_debug_response.py
src/feeds_gen/models/gtfs_feed_validation_notice.py
src/feeds_gen/models/gtfs_feed_validation_report.py
src/feeds_gen/models/gtfs_feed_validation_reports_response.py
diff --git a/functions-python/operations_api/src/feeds_operations/impl/feeds_operations_impl.py b/functions-python/operations_api/src/feeds_operations/impl/feeds_operations_impl.py
index c4520249d..8edaefcd8 100644
--- a/functions-python/operations_api/src/feeds_operations/impl/feeds_operations_impl.py
+++ b/functions-python/operations_api/src/feeds_operations/impl/feeds_operations_impl.py
@@ -18,6 +18,9 @@
from fastapi.responses import JSONResponse
import logging
+import os
+import tempfile
+import uuid
from datetime import datetime
from typing import Annotated, Final, Optional
@@ -33,6 +36,9 @@
from feeds_gen.models.gtfs_feed_availability_response import (
GtfsFeedAvailabilityResponse,
)
+from feeds_gen.models.gtfs_feed_download_debug_response import (
+ GtfsFeedDownloadDebugResponse,
+)
from feeds_gen.models.operation_create_request_gtfs_feed import (
OperationCreateRequestGtfsFeed,
)
@@ -67,6 +73,11 @@
GtfsFeedAvailabilityCheckImpl,
)
from shared.helpers.pub_sub import get_execution_id, trigger_dataset_download
+from shared.helpers.utils import (
+ FeedDownloadError,
+ download_feed,
+ get_feed_credentials,
+)
from shared.helpers.query_helper import (
query_feed_by_stable_id,
get_feeds_query,
@@ -299,6 +310,71 @@ def get_gtfs_feed_availability(
checks=[GtfsFeedAvailabilityCheckImpl.from_orm(c) for c in checks],
)
+ @with_db_session
+ def debug_gtfs_feed_download(
+ self,
+ id: Annotated[
+ StrictStr, Field(description="The feed ID of the requested feed.")
+ ],
+ max_bytes: Optional[int] = 52428800,
+ timeout_seconds: Optional[int] = 60,
+ db_session: Session = None,
+ ) -> GtfsFeedDownloadDebugResponse:
+ """Download the feed through the shared download path and return the diagnostics."""
+ gtfs_feed = (
+ db_session.query(Gtfsfeed).filter(Gtfsfeed.stable_id == id).one_or_none()
+ )
+ if gtfs_feed is None:
+ raise HTTPException(status_code=404, detail="GTFS feed not found")
+ if not gtfs_feed.producer_url:
+ raise HTTPException(status_code=400, detail="GTFS feed has no producer URL")
+
+ temp_file = os.path.join(
+ tempfile.gettempdir(), f"download_debug_{uuid.uuid4()}.zip"
+ )
+ truncated = False
+ try:
+ try:
+ result = download_feed(
+ gtfs_feed.producer_url,
+ file_path=temp_file,
+ feed_id=gtfs_feed.id,
+ authentication_type=gtfs_feed.authentication_type or 0,
+ api_key_parameter_name=gtfs_feed.api_key_parameter_name,
+ credentials=get_feed_credentials(gtfs_feed.stable_id),
+ max_bytes=max_bytes,
+ timeout_seconds=timeout_seconds,
+ )
+ diagnostics = result.diagnostics
+ truncated = result.file_hash is None
+ except FeedDownloadError as exc:
+ diagnostics = exc.diagnostics
+ finally:
+ if os.path.exists(temp_file):
+ try:
+ os.remove(temp_file)
+ except OSError as exc:
+ logging.warning("Could not remove %s: %s", temp_file, exc)
+
+ return GtfsFeedDownloadDebugResponse(
+ feed_id=gtfs_feed.stable_id,
+ producer_url=diagnostics.request_url,
+ resolved_url=diagnostics.resolved_url,
+ success=diagnostics.success,
+ status_code=diagnostics.status_code,
+ latency_ms=diagnostics.latency_ms,
+ request_headers=diagnostics.request_headers,
+ response_headers=diagnostics.response_headers,
+ redirect_urls=diagnostics.redirect_urls,
+ external_ip=diagnostics.external_ip,
+ content_type=diagnostics.content_type,
+ is_zip=diagnostics.is_zip,
+ downloaded_bytes=diagnostics.downloaded_bytes,
+ truncated=truncated,
+ error_type=diagnostics.error_type,
+ error_message=diagnostics.error_message,
+ )
+
@with_db_session
def get_gtfs_rt_feed(
self,
diff --git a/functions-python/operations_api/tests/feeds_operations/impl/test_download_debug_operations_impl.py b/functions-python/operations_api/tests/feeds_operations/impl/test_download_debug_operations_impl.py
new file mode 100644
index 000000000..7afc517e7
--- /dev/null
+++ b/functions-python/operations_api/tests/feeds_operations/impl/test_download_debug_operations_impl.py
@@ -0,0 +1,130 @@
+from unittest.mock import patch
+
+import pytest
+from fastapi import HTTPException
+
+from conftest import feed_mdb_40
+from feeds_operations.impl.feeds_operations_impl import OperationsApiImpl
+from shared.database.database import with_db_session
+from shared.database_gen.sqlacodegen_models import GtfsFeedAvailabilityCheck
+from shared.helpers.utils import FeedDownloadError, FeedDownloadResult
+from shared.helpers.utils import FeedRequestDiagnostics
+
+
+def _diagnostics(**overrides):
+ defaults = dict(
+ request_url="https://example.com/gtfs.zip",
+ resolved_url="https://example.com/gtfs.zip",
+ request_headers={"User-Agent": "agent"},
+ response_headers={"Server": "cloudflare"},
+ status_code=200,
+ latency_ms=120,
+ redirect_urls=[],
+ content_type="application/zip",
+ is_zip=True,
+ downloaded_bytes=2038,
+ )
+ defaults.update(overrides)
+ return FeedRequestDiagnostics(**defaults)
+
+
+@pytest.mark.asyncio
+async def test_download_debug_success():
+ """Returns the diagnostics of a successful download."""
+ api = OperationsApiImpl()
+ diagnostics = _diagnostics()
+ result = FeedDownloadResult(
+ file_hash="abc123", is_zip=True, diagnostics=diagnostics
+ )
+
+ with patch(
+ "feeds_operations.impl.feeds_operations_impl.download_feed",
+ return_value=result,
+ ):
+ response = api.debug_gtfs_feed_download(id=feed_mdb_40.stable_id)
+
+ assert response.feed_id == feed_mdb_40.stable_id
+ assert response.success is True
+ assert response.status_code == 200
+ assert response.is_zip is True
+ assert response.downloaded_bytes == 2038
+ assert response.truncated is False
+ assert response.response_headers == {"Server": "cloudflare"}
+
+
+@pytest.mark.asyncio
+async def test_download_debug_failure_returns_diagnostics():
+ """A failed download returns 200 with the failure detail rather than raising."""
+ api = OperationsApiImpl()
+ diagnostics = _diagnostics(
+ status_code=403,
+ content_type="text/html",
+ is_zip=False,
+ downloaded_bytes=None,
+ external_ip="34.1.2.3",
+ error_type="ValueError",
+ error_message="Invalid HTTP response code: [403]",
+ )
+
+ with patch(
+ "feeds_operations.impl.feeds_operations_impl.download_feed",
+ side_effect=FeedDownloadError("boom", diagnostics),
+ ):
+ response = api.debug_gtfs_feed_download(id=feed_mdb_40.stable_id)
+
+ assert response.success is False
+ assert response.status_code == 403
+ assert response.error_type == "ValueError"
+ assert response.external_ip == "34.1.2.3"
+
+
+@pytest.mark.asyncio
+async def test_download_debug_truncated_when_capped():
+ """A download stopped at the byte cap is reported as truncated."""
+ api = OperationsApiImpl()
+ result = FeedDownloadResult(
+ file_hash=None, is_zip=None, diagnostics=_diagnostics(downloaded_bytes=1024)
+ )
+
+ with patch(
+ "feeds_operations.impl.feeds_operations_impl.download_feed",
+ return_value=result,
+ ):
+ response = api.debug_gtfs_feed_download(
+ id=feed_mdb_40.stable_id, max_bytes=1024
+ )
+
+ assert response.truncated is True
+
+
+@pytest.mark.asyncio
+async def test_download_debug_not_found():
+ """Returns 404 for an unknown feed ID."""
+ api = OperationsApiImpl()
+ with pytest.raises(HTTPException) as exc_info:
+ api.debug_gtfs_feed_download(id="mdb-9999")
+ assert exc_info.value.status_code == 404
+
+
+@with_db_session
+def _count_availability_checks(db_session=None) -> int:
+ return db_session.query(GtfsFeedAvailabilityCheck).count()
+
+
+@pytest.mark.asyncio
+async def test_download_debug_does_not_persist():
+ """The endpoint stores nothing in gtfs_feed_availability_check."""
+ api = OperationsApiImpl()
+ result = FeedDownloadResult(
+ file_hash="abc123", is_zip=True, diagnostics=_diagnostics()
+ )
+
+ before = _count_availability_checks()
+
+ with patch(
+ "feeds_operations.impl.feeds_operations_impl.download_feed",
+ return_value=result,
+ ):
+ api.debug_gtfs_feed_download(id=feed_mdb_40.stable_id)
+
+ assert _count_availability_checks() == before
diff --git a/functions-python/tasks_executor/src/tasks/feed_availability/check_gtfs_feed_availability.py b/functions-python/tasks_executor/src/tasks/feed_availability/check_gtfs_feed_availability.py
index 8e54d6577..371f52f51 100644
--- a/functions-python/tasks_executor/src/tasks/feed_availability/check_gtfs_feed_availability.py
+++ b/functions-python/tasks_executor/src/tasks/feed_availability/check_gtfs_feed_availability.py
@@ -14,7 +14,6 @@
# limitations under the License.
#
import logging
-import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
@@ -28,7 +27,7 @@
Gtfsfeed,
GtfsFeedAvailabilityCheck,
)
-from shared.helpers.utils import perform_request
+from shared.helpers.utils import get_feed_credentials, perform_request
DEFAULT_CONCURRENCY: int = 15
DEFAULT_TIMEOUT_SECONDS: int = 20
@@ -36,18 +35,6 @@
DEFAULT_FALLBACK_TO_GET: bool = True
-def get_feed_credentials(stable_id: str) -> Optional[str]:
- """Return the API credential for a feed from the FEEDS_CREDENTIALS env var, or None."""
- try:
- import json
-
- feeds_credentials = json.loads(os.getenv("FEEDS_CREDENTIALS", "{}"))
- return feeds_credentials.get(stable_id, None)
- except Exception as exc:
- logging.warning("Could not parse FEEDS_CREDENTIALS: %s", exc)
- return None
-
-
def get_parameters(payload: dict):
dry_run = payload.get("dry_run", True)
skip_db_update = payload.get("skip_db_update", False)
@@ -286,6 +273,7 @@ def check_gtfs_feed_availability(
error_message=str(exc),
error_type=type(exc).__name__,
success=False,
+ source="availability_check",
)
results.append(check)
batch.append(check)
diff --git a/liquibase/changelog.xml b/liquibase/changelog.xml
index 336fa821d..9a93a0916 100644
--- a/liquibase/changelog.xml
+++ b/liquibase/changelog.xml
@@ -144,6 +144,8 @@
+
+
diff --git a/liquibase/changes/feat_download_debug.sql b/liquibase/changes/feat_download_debug.sql
new file mode 100644
index 000000000..55a9893f5
--- /dev/null
+++ b/liquibase/changes/feat_download_debug.sql
@@ -0,0 +1,20 @@
+-- Add request/response debugging columns to gtfs_feed_availability_check.
+-- source distinguishes rows written by the daily availability task from rows written
+-- when a dataset download fails in batch_process_dataset.
+-- request_headers, response_headers and resolved_url are sanitized before being stored:
+-- credential-bearing header values and query parameters are replaced with a redaction marker.
+-- external_ip is the egress IP observed for the request, captured best-effort on failures only.
+ALTER TABLE gtfs_feed_availability_check
+ ADD COLUMN IF NOT EXISTS source TEXT,
+ ADD COLUMN IF NOT EXISTS request_headers JSONB,
+ ADD COLUMN IF NOT EXISTS response_headers JSONB,
+ ADD COLUMN IF NOT EXISTS redirect_urls JSONB,
+ ADD COLUMN IF NOT EXISTS external_ip TEXT,
+ ADD COLUMN IF NOT EXISTS resolved_url TEXT;
+
+UPDATE gtfs_feed_availability_check
+SET source = 'availability_check'
+WHERE source IS NULL;
+
+CREATE INDEX IF NOT EXISTS idx_gtfs_feed_availability_check_source_checked_at
+ ON gtfs_feed_availability_check (source, checked_at);
diff --git a/scripts/download-verifier.sh b/scripts/download-verifier.sh
new file mode 100755
index 000000000..2a129a0fd
--- /dev/null
+++ b/scripts/download-verifier.sh
@@ -0,0 +1,124 @@
+#!/bin/bash
+#
+#
+# MobilityData 2026
+#
+# 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.
+#
+#
+
+# Downloads a GTFS feed using the same shared code as the nightly dataset download and
+# prints the request and response detail of the attempt, with credentials redacted.
+# No database is required: when the config database is unreachable the download falls back
+# to the default headers.
+
+# Usage:
+# download-verifier.sh --url [--feed_id ] [--with_upload] [--no_install_venv]
+# Example:
+# download-verifier.sh --url https://example.com/gtfs.zip
+
+# relative path
+SCRIPT_PATH="$(dirname -- "${BASH_SOURCE[0]}")"
+FUNCTION_PATH="$SCRIPT_PATH/../functions-python/batch_process_dataset"
+
+# function printing usage
+display_usage() {
+ printf "\nThis script downloads a GTFS feed and prints the request and response detail"
+ printf "\nScript Usage:\n"
+ echo "Usage: $0 [options]"
+ echo "Options:"
+ echo " -h|--help Display help content."
+ echo " --url Producer URL of the feed to download. Required."
+ echo " --feed_id Feed UUID, to apply the per-feed http_headers config override."
+ echo " --with_upload Also verify the upload path using the GCP storage emulator."
+ echo " --no_install_venv Do not reinstall the python virtual environment."
+ exit 1
+}
+
+url=''
+feed_id=''
+with_upload=false
+no_install_venv=false
+
+while [[ $# -gt 0 ]]; do
+ key="$1"
+
+ case $key in
+ -h | --help)
+ display_usage
+ exit 0
+ ;;
+ --url)
+ url="$2"
+ shift # past argument
+ shift # past value
+ ;;
+ --feed_id)
+ feed_id="$2"
+ shift # past argument
+ shift # past value
+ ;;
+ --with_upload)
+ with_upload=true
+ shift # past argument
+ ;;
+ --no_install_venv)
+ no_install_venv=true
+ shift # past argument
+ ;;
+ *) # unknown option
+ shift # past argument
+ ;;
+ esac
+done
+
+if [ -z "$url" ]; then
+ printf "\nERROR: --url is required\n"
+ display_usage
+ exit 1
+fi
+
+if [ ! -d "$FUNCTION_PATH/src" ]; then
+ printf "\nERROR: function's folder not found at location: %s\n" "$FUNCTION_PATH/src"
+ exit 1
+fi
+
+# Link the shared code into src/shared so the function's imports resolve.
+printf "\nINFO: linking shared code\n"
+"$SCRIPT_PATH"/function-python-setup.sh --function_name batch_process_dataset
+
+if [ "$no_install_venv" != "true" ]; then
+ printf "\nINFO: installing python virtual environment\n"
+ pushd "$FUNCTION_PATH" >/dev/null
+ rm -rf venv
+ pip3 install --disable-pip-version-check virtualenv >/dev/null
+ python3 -m virtualenv venv >/dev/null
+ venv/bin/python -m pip install --disable-pip-version-check \
+ -r requirements.txt -r requirements_dev.txt >/dev/null
+ popd >/dev/null
+else
+ printf "\nINFO: skipping python virtual environment installation\n"
+fi
+
+export PYTHONPATH="$FUNCTION_PATH/src"
+
+args=(--url "$url")
+if [ -n "$feed_id" ]; then
+ args+=(--feed_id "$feed_id")
+fi
+if [ "$with_upload" != "true" ]; then
+ args+=(--skip-upload)
+fi
+
+printf "\nINFO: downloading %s\n" "$url"
+"$FUNCTION_PATH"/venv/bin/python "$FUNCTION_PATH/src/scripts/download_verifier.py" "${args[@]}"