From d55bc672c1f8784e957cc2b36e976e3455a4b492 Mon Sep 17 00:00:00 2001 From: badry-dev Date: Tue, 1 Sep 2026 20:45:35 +0000 Subject: [PATCH 1/9] fix(runner): keep upstream values out of debug logs, bound row-error staging, fix dedupe attribution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Debug logging emitted whole upstream records and values (sample record, mapped record, date-detection row and value). Log column names, presence and type only; imported rows can carry PII and DEBUG logs leave the DB boundary. - Validation-stage row errors are flushed every ROW_ERROR_FLUSH_SIZE (1000) staged entries so a run with thousands of invalid rows no longer holds every TaskRunLog in the session until the single commit. - Batch de-duplication carries the caller's absolute row index through routing (duplicate reports and fallback errors pointed at compressed indices) and no longer folds rows that lack a key value into one 'duplicate' — they take the normal route as before. - Tests: keyless rows not collapsed, duplicate row_index is absolute, and a real CASE ... ELSE exercise (one None column preserved while a sibling column updates). Claude-Session: https://claude.ai/code/session_018bq3je2VeLdrF71FThQVvU --- backend/app/services/runner.py | 46 +++++++++++++----- backend/tests/unit/test_upsert_logic.py | 63 +++++++++++++++++++++++++ 2 files changed, 96 insertions(+), 13 deletions(-) diff --git a/backend/app/services/runner.py b/backend/app/services/runner.py index f09b576..4f85211 100644 --- a/backend/app/services/runner.py +++ b/backend/app/services/runner.py @@ -24,6 +24,10 @@ _SAFE_SQL_IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_$#]{0,127}$") +# Row-error staging is flushed to the DB every N entries (see run_import step 8). +ROW_ERROR_FLUSH_SIZE = 1000 + + def _redact_cursor(value): """Same redaction policy as workers.tasks._redact_cursor — see that function for rationale. Duplicated here to avoid a circular import @@ -268,13 +272,16 @@ async def run_import( # Debug: Log source field names and sample values if flattened_records: sample_record = flattened_records[0] + # Names and types only: upstream rows can carry PII and DEBUG logs + # may be retained or shipped outside the database boundary. logger.debug(f"Sample flattened record keys: {list(sample_record.keys())}") - logger.debug(f"Sample flattened record: {sample_record}") for mapping in column_mappings: source_val = sample_record.get(mapping.source_field) logger.debug( f"Mapping: {mapping.source_field} -> {mapping.dest_column} " - f"| transforms={mapping.transform_rules} | source_value={repr(source_val)}" + f"| transforms={mapping.transform_rules} " + f"| source_present={source_val is not None} " + f"| source_type={type(source_val).__name__}" ) mapped_records = mapper.map_rows(flattened_records, column_mappings) @@ -282,7 +289,7 @@ async def run_import( # Debug: Log mapped results if mapped_records: sample_mapped = mapped_records[0] - logger.debug(f"Sample mapped record: {sample_mapped}") + logger.debug(f"Sample mapped record columns: {list(sample_mapped.keys())}") # Step 7: Validate rows log_step(db, task_run_id, "VALIDATE", f"Validating {len(mapped_records)} records") @@ -294,7 +301,10 @@ async def run_import( logger.info(f"Validation complete: {len(valid_rows)} valid, {len(invalid_rows)} invalid") - # Step 8: Log validation errors + # Step 8: Log validation errors. log_row_error() only stages rows; flush + # in chunks so a run with thousands of invalid rows doesn't hold every + # TaskRunLog in the session until the single commit below. + staged_errors = 0 for idx, invalid_item in enumerate(invalid_rows): errors = invalid_item["errors"] for error in errors: @@ -307,6 +317,9 @@ async def run_import( error_message=error.message, source_value=str(error.value) if error.value is not None else None, ) + staged_errors += 1 + if staged_errors % ROW_ERROR_FLUSH_SIZE == 0: + db.flush() task_run.error_count = len(invalid_rows) db.commit() @@ -535,7 +548,7 @@ def _build_insert_statement( # Debug: Log sample row if sample_row: - logger.debug(f"Sample row for date detection: {sample_row}") + logger.debug(f"Sample row columns for date detection: {list(sample_row.keys())}") # Build placeholders, wrapping date strings with TO_DATE() placeholders = [] @@ -548,7 +561,7 @@ def _build_insert_statement( value = sample_row[column] if isinstance(value, str) and re.match(r"^\d{4}-\d{2}-\d{2}$", value): is_date_string = True - logger.debug(f"Detected date column: {column} = {value}") + logger.debug(f"Detected date column: {column}") if is_date_string: # Wrap with TO_DATE() for Oracle DATE columns @@ -751,10 +764,17 @@ def _process_upsert_batch( # clauses silently dropped by SQL) or both to_insert (constraint failure # or duplicate rows). First occurrence wins; duplicates are counted as # skipped with an attribution entry. + # Rows are carried as (absolute_row_index, row) from here on so error + # attribution keeps pointing at the caller's row numbers after duplicates + # are dropped. Rows missing any key value cannot collide on a key: they + # skip deduplication and take the normal route (no match -> insert). seen_keys = set() - deduped_batch = [] + indexed_batch: list[tuple[int, dict]] = [] for idx, row in enumerate(batch): key_tuple = tuple(row.get(key) for key in upsert_keys) + if any(value is None for value in key_tuple): + indexed_batch.append((batch_offset + idx, row)) + continue if key_tuple in seen_keys: results["skipped"] += 1 results["error_details"].append( @@ -766,8 +786,8 @@ def _process_upsert_batch( ) continue seen_keys.add(key_tuple) - deduped_batch.append(row) - batch = deduped_batch + indexed_batch.append((batch_offset + idx, row)) + batch = [row for _, row in indexed_batch] if not batch: return results @@ -825,20 +845,20 @@ def _process_upsert_batch( to_update = [] to_skip = [] - for idx, row in enumerate(batch): + for abs_idx, row in indexed_batch: row_key_tuple = tuple(row.get(key) for key in upsert_keys) existing = existing_rows.get(row_key_tuple) if existing is None: # New record - to_insert.append((batch_offset + idx, row)) + to_insert.append((abs_idx, row)) elif task.skip_column and task.skip_value is not None and _should_skip(task, existing): # Record exists and matches the configured skip condition — # third parties may have marked this row processed; never # overwrite it. - to_skip.append((batch_offset + idx, row)) + to_skip.append((abs_idx, row)) else: - to_update.append((batch_offset + idx, row)) + to_update.append((abs_idx, row)) # Steps 3+4: Bulk UPDATE + INSERT inside a savepoint, so a partially # applied bulk statement is rolled back completely before falling back diff --git a/backend/tests/unit/test_upsert_logic.py b/backend/tests/unit/test_upsert_logic.py index 01867e5..2daf152 100644 --- a/backend/tests/unit/test_upsert_logic.py +++ b/backend/tests/unit/test_upsert_logic.py @@ -497,6 +497,34 @@ def test_duplicate_keys_first_row_wins(self, task): "duplicate upsert key" in str(d.get("error", "")) for d in results["error_details"] ) + def test_rows_without_key_values_are_not_collapsed(self, task): + """Rows lacking the upsert key cannot collide on it; they must not be + folded into one 'duplicate' — the pre-dedupe behaviour (insert) stands.""" + db = MagicMock() + db.execute.return_value.fetchall.return_value = [] + + rows = [{"name": "A"}, {"name": "B"}] + results = process_rows_with_upsert(db, task, 1, rows) + + assert results["skipped"] == 0 + assert results["inserted"] == 2 + assert not any("duplicate" in str(d.get("error", "")) for d in results["error_details"]) + + def test_duplicate_report_uses_the_callers_row_index(self, task): + db = MagicMock() + db.execute.return_value.fetchall.return_value = [] + + rows = [ + {"employee_id": 1, "name": "First"}, + {"employee_id": 1, "name": "Dup"}, # index 1 + {"employee_id": 2, "name": "Other"}, # must still be attributed as index 2 + ] + results = process_rows_with_upsert(db, task, 1, rows) + + dups = [d for d in results["error_details"] if "duplicate" in str(d.get("error", ""))] + assert [d["row_index"] for d in dups] == [1] + assert results["inserted"] == 2 + class TestBatchSkipCondition: """Regression tests for the unimplemented skip condition in the batch @@ -558,3 +586,38 @@ def test_row_marked_processed_is_skipped(self, sqlite_dest, task_with_skip): if __name__ == "__main__": pytest.main([__file__, "-v"]) + + +def test_none_column_preserved_while_sibling_column_updates(sqlite_dest_factory): + """Exercise the ``ELSE `` branch of the bulk CASE update. + + A row with one None column and one non-None column must emit a SET clause + (so the CASE actually executes) and still keep the stored value for the + None column. + """ + from unittest.mock import MagicMock + + from sqlalchemy import text + + from app.services.runner import _bulk_update_rows + + session = sqlite_dest_factory( + "CREATE TABLE EMPLOYEES (employee_id INTEGER PRIMARY KEY, name TEXT, dept TEXT)" + ) + session.execute( + text("INSERT INTO EMPLOYEES (employee_id, name, dept) VALUES (1, 'Original', 'OldDept')") + ) + session.commit() + + task = MagicMock() + task.dest_table = "EMPLOYEES" + task.upsert_keys = ["employee_id"] + + count = _bulk_update_rows( + session, task, [(0, {"employee_id": 1, "name": None, "dept": "NewDept"})], ["employee_id"] + ) + session.commit() + + row = session.execute(text("SELECT name, dept FROM EMPLOYEES WHERE employee_id = 1")).fetchone() + assert count == 1 + assert tuple(row) == ("Original", "NewDept") From a3ed0a8d67c7121cde4825e8c720a448ca0d81a3 Mon Sep 17 00:00:00 2001 From: badry-dev Date: Tue, 1 Sep 2026 20:45:35 +0000 Subject: [PATCH 2/9] fix(scheduler): persist the auto-pause before removing the job; require threshold >= 1 _handle_dispatch_failure set is_active=False, removed the APScheduler job, then committed. A failed commit rolled the flag back while the in-memory job was already gone, so dispatch stopped silently until a restart. Commit first; a job-removal failure is logged and does not undo the persisted pause. SCHEDULE_MAX_CONSECUTIVE_FAILURES is constrained to >= 1: 0 or a negative value paused a schedule on its very first failure. test_config_env: the default-APP_ENV subprocess test passes _env_file=None so a developer's backend/.env cannot mask the default, and _make_app reloads app.main again after restoring settings so later imports do not see the production-configured module. Threshold bound tests added. Claude-Session: https://claude.ai/code/session_018bq3je2VeLdrF71FThQVvU --- backend/app/core/config.py | 6 +++-- backend/app/services/scheduler.py | 27 ++++++++++++++----- backend/tests/unit/test_config_env.py | 22 +++++++++++++++- backend/tests/unit/test_scheduler.py | 38 +++++++++++++++++++++++++++ 4 files changed, 83 insertions(+), 10 deletions(-) diff --git a/backend/app/core/config.py b/backend/app/core/config.py index bd96af8..d2eea2f 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -1,4 +1,4 @@ -from pydantic import field_validator +from pydantic import Field, field_validator from pydantic_settings import BaseSettings @@ -45,7 +45,9 @@ class Settings(BaseSettings): ALLOWED_SOURCE_HOSTS: str = "" # Auto-pause a schedule after this many consecutive dispatch failures. - SCHEDULE_MAX_CONSECUTIVE_FAILURES: int = 5 + # Auto-pause a schedule after this many consecutive dispatch failures. + # 0 or negative would pause on the very first failure, so require >= 1. + SCHEDULE_MAX_CONSECUTIVE_FAILURES: int = Field(default=5, ge=1) @property def destination_sqlalchemy_url(self) -> str: diff --git a/backend/app/services/scheduler.py b/backend/app/services/scheduler.py index 0a631f7..5949e66 100644 --- a/backend/app/services/scheduler.py +++ b/backend/app/services/scheduler.py @@ -176,20 +176,33 @@ def _handle_dispatch_failure(self, task_id: int, exc: Exception): current = task_schedule.consecutive_failures or 0 task_schedule.consecutive_failures = current + 1 - if current + 1 >= threshold and task_schedule.is_active: + paused = current + 1 >= threshold and task_schedule.is_active + if paused: task_schedule.is_active = False - self.remove_schedule(task_id) - logger.error( - f"Auto-paused schedule for task {task_id}: " - f"{current + 1} consecutive dispatch failures " - f"(threshold={threshold}). Use the resume endpoint to reactivate." - ) + # Persist FIRST. If the commit fails, the rollback below leaves + # the schedule active in the DB while its in-memory job would + # already be gone — dispatch would silently stop until a restart. self.db.commit() logger.error( f"Scheduled task {task_id} dispatch failure " f"(consecutive_failures={current + 1})" ) + + if paused: + try: + self.remove_schedule(task_id) + except Exception as remove_exc: + # DB says paused; the job will not be reloaded on restart. + logger.error( + f"Auto-paused task {task_id} in the database but failed to " + f"remove its scheduler job: {remove_exc}" + ) + logger.error( + f"Auto-paused schedule for task {task_id}: " + f"{current + 1} consecutive dispatch failures " + f"(threshold={threshold}). Use the resume endpoint to reactivate." + ) except Exception as counter_exc: logger.error( f"Failed to persist consecutive_failures for task {task_id}: {counter_exc}" diff --git a/backend/tests/unit/test_config_env.py b/backend/tests/unit/test_config_env.py index 45c6589..7b28a0e 100644 --- a/backend/tests/unit/test_config_env.py +++ b/backend/tests/unit/test_config_env.py @@ -30,7 +30,12 @@ def test_missing_app_env_defaults_to_dev(self): from pathlib import Path backend = Path(__file__).resolve().parents[2] - code = "from app.core.config import Settings;print(Settings(SECRET_KEY='x').APP_ENV)" + # _env_file=None: a developer's backend/.env must not be able to supply APP_ENV + # and mask the built-in default this test exists to exercise. + code = ( + "from app.core.config import Settings;" + "print(Settings(_env_file=None, SECRET_KEY='x').APP_ENV)" + ) out = subprocess.run( [sys.executable, "-c", code], env=env, @@ -61,6 +66,10 @@ def _make_app(self, app_env): finally: settings.APP_ENV = original settings.API_TOKEN = original_token + # Reload again so app.main.app reflects the restored settings for + # any later test that imports it; the caller already holds the + # instance built above. + importlib.reload(main_module) def test_docs_enabled_outside_production(self): from fastapi.testclient import TestClient @@ -77,3 +86,14 @@ def test_docs_disabled_in_production_canonical_value(self): client = TestClient(app) assert client.get("/docs").status_code == 404 assert client.get("/openapi.json").status_code == 404 + + +class TestScheduleFailureThreshold: + def test_default_is_five(self): + assert Settings(SECRET_KEY="x").SCHEDULE_MAX_CONSECUTIVE_FAILURES == 5 + + @pytest.mark.parametrize("bad", [0, -1]) + def test_non_positive_threshold_rejected(self, bad): + # 0 or negative would auto-pause a schedule on its very first failure. + with pytest.raises(ValueError): + Settings(SECRET_KEY="x", SCHEDULE_MAX_CONSECUTIVE_FAILURES=bad) diff --git a/backend/tests/unit/test_scheduler.py b/backend/tests/unit/test_scheduler.py index fced82d..15bf914 100644 --- a/backend/tests/unit/test_scheduler.py +++ b/backend/tests/unit/test_scheduler.py @@ -164,6 +164,44 @@ def test_below_threshold_does_not_pause(self, scheduler): assert schedule.is_active is True mock_remove.assert_not_called() + def test_auto_pause_is_persisted_before_the_job_is_removed(self, scheduler): + """If the commit failed after the job was gone, the DB would still say + active while nothing dispatches — so the order must be commit, then remove.""" + sched, db = scheduler + schedule = _make_schedule(consecutive_failures=4, is_active=True) + db.query.return_value.filter.return_value.first.return_value = schedule + order: list[str] = [] + db.commit.side_effect = lambda: order.append("commit") + + with ( + patch("app.services.scheduler.enqueue_run", side_effect=RuntimeError("down")), + patch.object(sched, "remove_schedule", side_effect=lambda tid: order.append("remove")), + patch("app.core.config.settings") as mock_settings, + ): + mock_settings.SCHEDULE_MAX_CONSECUTIVE_FAILURES = 5 + sched._execute_scheduled_task(7, "Nightly import") + + assert order == ["commit", "remove"] + assert schedule.is_active is False + + def test_job_removal_failure_keeps_the_persisted_pause(self, scheduler): + sched, db = scheduler + schedule = _make_schedule(consecutive_failures=4, is_active=True) + db.query.return_value.filter.return_value.first.return_value = schedule + + with ( + patch("app.services.scheduler.enqueue_run", side_effect=RuntimeError("down")), + patch.object(sched, "remove_schedule", side_effect=RuntimeError("apscheduler gone")), + patch("app.core.config.settings") as mock_settings, + ): + mock_settings.SCHEDULE_MAX_CONSECUTIVE_FAILURES = 5 + sched._execute_scheduled_task(7, "Nightly import") # must not raise + + assert schedule.is_active is False + db.commit.assert_called_once() + # Only the initial rollback that precedes the counter transaction. + db.rollback.assert_called_once() + if __name__ == "__main__": pytest.main([__file__, "-v"]) From 5065effcef368ed8e6af3466afea293d0a5085bc Mon Sep 17 00:00:00 2001 From: badry-dev Date: Tue, 1 Sep 2026 20:45:35 +0000 Subject: [PATCH 3/9] fix(http): async SSRF check with 403 on fetch-time blocks; validate ports on every path; parser-based URL redaction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - column_mappings: the standalone preview awaited nothing and called the synchronous validate_url, resolving DNS on the event loop; use validate_url_async. Both preview endpoints now return 403 when the fetch-time re-validation raises SSRFBlockedError — it subclasses ValueError and surfaced as 400 'Invalid JSON'. - url_guard: the invalid-port check ran only on the hostname-resolution branch. It now runs for literal IPs and for resolve=False (config-time) too, so http://8.8.8.8:99999/ is an SSRFBlockedError, not a ValueError. - api_connector._redact_url_for_log: rebuilt on urlsplit/urlunsplit; drops query, fragment and userinfo and strips control characters so a crafted URL cannot forge log lines. Claude-Session: https://claude.ai/code/session_018bq3je2VeLdrF71FThQVvU --- backend/app/api/v1/routes/column_mappings.py | 17 ++++++++++-- backend/app/core/url_guard.py | 16 ++++++----- backend/app/services/api_connector.py | 27 ++++++++++++------- .../tests/integration/test_api_endpoints.py | 25 +++++++++++++++++ backend/tests/unit/test_url_guard.py | 9 +++++++ backend/tests/unit/test_url_redaction.py | 26 ++++++++++++++++++ 6 files changed, 102 insertions(+), 18 deletions(-) create mode 100644 backend/tests/unit/test_url_redaction.py diff --git a/backend/app/api/v1/routes/column_mappings.py b/backend/app/api/v1/routes/column_mappings.py index 3559ba1..cc3d70b 100644 --- a/backend/app/api/v1/routes/column_mappings.py +++ b/backend/app/api/v1/routes/column_mappings.py @@ -16,7 +16,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session -from app.core.url_guard import SSRFBlockedError, validate_url +from app.core.url_guard import SSRFBlockedError, validate_url_async from app.db.models.column_mapping import ColumnMapping from app.db.models.task import Task from app.db.schemas.column_mapping import ( @@ -390,6 +390,12 @@ async def preview_fields( except HTTPException: # Re-raise deliberate HTTP errors instead of rewriting them to 400. raise + except SSRFBlockedError as e: + # Raised by the fetch-time re-validation inside fetch_json. It + # subclasses ValueError, so it must be handled before that branch or + # it surfaces as a misleading 400 "Invalid JSON". + logger.warning(f"SSRF guard rejected preview fetch: {e}") + raise HTTPException(status_code=403, detail=str(e)) except ValueError as e: logger.warning(f"Invalid JSON for task {task_id}: {str(e)}") raise HTTPException(status_code=400, detail=f"Invalid JSON: {str(e)}") @@ -506,7 +512,8 @@ async def preview_fields_standalone(request: PreviewFieldsRequest): # SSRF guard (C4): reject private/loopback/link-local targets. try: - validate_url(request.url) + # Async variant: DNS resolution must not block the event loop. + await validate_url_async(request.url) except SSRFBlockedError as e: logger.warning(f"SSRF guard rejected standalone preview URL: {e}") raise HTTPException(status_code=403, detail=str(e)) @@ -592,6 +599,12 @@ async def preview_fields_standalone(request: PreviewFieldsRequest): # Re-raise deliberate HTTP errors (400 missing sample_json, 403 SSRF # block) instead of letting the broad handler rewrite them to 400. raise + except SSRFBlockedError as e: + # Raised by the fetch-time re-validation inside fetch_json. It + # subclasses ValueError, so it must be handled before that branch or + # it surfaces as a misleading 400 "Invalid JSON". + logger.warning(f"SSRF guard rejected preview fetch: {e}") + raise HTTPException(status_code=403, detail=str(e)) except ValueError as e: logger.warning(f"Invalid JSON: {str(e)}") raise HTTPException(status_code=400, detail=f"Invalid JSON: {str(e)}") diff --git a/backend/app/core/url_guard.py b/backend/app/core/url_guard.py index 0c40961..1d09c3c 100644 --- a/backend/app/core/url_guard.py +++ b/backend/app/core/url_guard.py @@ -61,6 +61,15 @@ def validate_url(url: str, *, resolve: bool = True) -> str: if not host: raise SSRFBlockedError("URL must include a host") + # urlparse raises ValueError on attribute access for out-of-range ports + # (http://example.com:99999/). Check it up front so literal-IP URLs and + # config-time validation (resolve=False) reject it too, as an + # SSRFBlockedError rather than a 500. + try: + port = parsed.port + except ValueError as e: + raise SSRFBlockedError(f"URL has an invalid port: {e}") from e + # Literal IPs are checked directly; hostnames are resolved. try: candidate_addrs = [ipaddress.ip_address(host)] @@ -71,13 +80,6 @@ def validate_url(url: str, *, resolve: bool = True) -> str: host_allowed = host.lower() in allowed if not candidate_addrs and resolve: - # urlparse raises ValueError on attribute access for out-of-range - # ports (http://example.com:99999/) — surface it as SSRFBlockedError, - # not a 500. - try: - port = parsed.port - except ValueError as e: - raise SSRFBlockedError(f"URL has an invalid port: {e}") from e try: infos = socket.getaddrinfo(host, port or None) candidate_addrs = [ipaddress.ip_address(info[4][0]) for info in infos] diff --git a/backend/app/services/api_connector.py b/backend/app/services/api_connector.py index 8b3f68c..f7a02b5 100644 --- a/backend/app/services/api_connector.py +++ b/backend/app/services/api_connector.py @@ -2,7 +2,9 @@ import base64 import datetime as _dt import email.utils +import re from typing import Any +from urllib.parse import urlsplit, urlunsplit import httpx from loguru import logger @@ -49,16 +51,23 @@ def mask_headers(headers: dict) -> dict: return masked +_LOG_CONTROL_CHARS_RE = re.compile(r"[\x00-\x1f\x7f]") + + def _redact_url_for_log(url: str) -> str: - """Strip query strings AND userinfo (may embed api keys / tokens) for log - output. All URL logging must go through this helper.""" - base = url.split("?", 1)[0] - scheme, sep, rest = base.partition("://") - if sep and "@" in rest.split("/", 1)[0]: - # https://user:secret@host/path -> https://host/path - _, _, tail = rest.partition("@") - return f"{scheme}://{tail}" - return base + """Reduce a URL to scheme://host[:port]/path for log output. + + Drops the query string and fragment (may embed api keys / tokens) and any + userinfo (user:secret@), then strips control characters so a crafted URL + cannot forge log lines. All URL logging must go through this helper. + """ + try: + parts = urlsplit(str(url)) + except ValueError: + return "" + netloc = parts.netloc.rpartition("@")[2] # drop user:secret@ + redacted = urlunsplit((parts.scheme, netloc, parts.path, "", "")) + return _LOG_CONTROL_CHARS_RE.sub("", redacted) def _parse_retry_after(header_value: str | None) -> float | None: diff --git a/backend/tests/integration/test_api_endpoints.py b/backend/tests/integration/test_api_endpoints.py index 91d9cf4..dfa68b2 100644 --- a/backend/tests/integration/test_api_endpoints.py +++ b/backend/tests/integration/test_api_endpoints.py @@ -451,3 +451,28 @@ def test_root_endpoint(client: TestClient): if __name__ == "__main__": pytest.main([__file__, "-v"]) + + +# ============================================================================ +# Standalone field preview: fetch-time SSRF block must surface as 403 +# ============================================================================ + + +def test_standalone_preview_fetch_time_ssrf_block_is_403(client: TestClient): + """SSRFBlockedError subclasses ValueError; before the dedicated handler it + came back as 400 "Invalid JSON", hiding that the URL was blocked.""" + from unittest.mock import patch + + from app.core.url_guard import SSRFBlockedError + + path = "/api/v1/preview-fields-standalone" + with patch( + "app.api.v1.routes.column_mappings.fetch_sample_response", + side_effect=SSRFBlockedError("Host 'api.example.com' resolves to a non-public address"), + ): + resp = client.post( + path, + json={"use_auto_fetch": True, "method": "GET", "url": "https://api.example.com/x"}, + ) + assert resp.status_code == 403 + assert "non-public" in resp.json()["detail"] diff --git a/backend/tests/unit/test_url_guard.py b/backend/tests/unit/test_url_guard.py index 99f27f4..2257ee8 100644 --- a/backend/tests/unit/test_url_guard.py +++ b/backend/tests/unit/test_url_guard.py @@ -90,3 +90,12 @@ async def test_fetch_json_rejects_invalid_port_as_ssrf_error(self): # validation failure), not escape as ValueError -> HTTP 500. with pytest.raises(SSRFBlockedError): await fetch_json("GET", "http://example.com:99999/") + + +class TestInvalidPort: + @pytest.mark.parametrize("url", ["http://8.8.8.8:99999/", "https://example.com:70000/x"]) + def test_out_of_range_port_rejected_without_resolving(self, url): + # Must be an SSRFBlockedError on every path, including literal IPs and + # config-time validation (resolve=False), not a ValueError/500. + with pytest.raises(SSRFBlockedError, match="invalid port"): + validate_url(url, resolve=False) diff --git a/backend/tests/unit/test_url_redaction.py b/backend/tests/unit/test_url_redaction.py new file mode 100644 index 0000000..58f7e59 --- /dev/null +++ b/backend/tests/unit/test_url_redaction.py @@ -0,0 +1,26 @@ +"""Tests for the log-safe URL redaction helper.""" + +import pytest + +from app.services.api_connector import _redact_url_for_log + + +@pytest.mark.parametrize( + "raw,expected", + [ + ("https://api.example.com/v1/users?api_key=SECRET", "https://api.example.com/v1/users"), + ("https://user:s3cret@api.example.com/v1/users", "https://api.example.com/v1/users"), + ("https://api.example.com:8443/v1/users#frag", "https://api.example.com:8443/v1/users"), + ("https://user:s3cret@[::1]:8080/p?x=1", "https://[::1]:8080/p"), + ("https://api.example.com/v1/users", "https://api.example.com/v1/users"), + ], +) +def test_query_userinfo_and_fragment_are_dropped(raw, expected): + assert _redact_url_for_log(raw) == expected + + +def test_control_characters_cannot_forge_log_lines(): + forged = "https://api.example.com/v1\r\n2026-01-01 ERROR fake line?token=abc" + redacted = _redact_url_for_log(forged) + assert "\r" not in redacted and "\n" not in redacted + assert "token" not in redacted From cb5c5bf59c14d037929cd8e4233ef1a778ad60eb Mon Sep 17 00:00:00 2001 From: badry-dev Date: Tue, 1 Sep 2026 20:45:35 +0000 Subject: [PATCH 4/9] fix(tasks): enforce the upsert-keys invariant at create and on the effective update state Field validators do not run for omitted fields, so {"upsert_enabled": true} without upsert_keys passed TaskCreate and the runner silently fell back to plain inserts. The rule is now a model validator on TaskCreate. TaskUpdate rejects upsert_enabled=true with an explicit empty key list, and update_task validates the effective state (request merged over stored values) whenever the request touches either field, so re-enabling with stored keys still works and unrelated partial updates of legacy rows are not blocked. _validate_source_url branches on the parsed scheme: a relative endpoint such as /api/items?redirect=https://cb.example.com contains '://' but is not an absolute URL and was rejected. get_task_stats: the non-SQLite average-duration fallback also filters started_at IS NOT NULL; a run with a null start crashed it. Claude-Session: https://claude.ai/code/session_018bq3je2VeLdrF71FThQVvU --- backend/app/api/v1/routes/tasks.py | 20 ++++++- backend/app/db/schemas/task.py | 43 ++++++++++----- .../tests/unit/test_tasks_partial_update.py | 54 +++++++++++++++++++ 3 files changed, 103 insertions(+), 14 deletions(-) diff --git a/backend/app/api/v1/routes/tasks.py b/backend/app/api/v1/routes/tasks.py index c4b032f..5aac852 100644 --- a/backend/app/api/v1/routes/tasks.py +++ b/backend/app/api/v1/routes/tasks.py @@ -197,6 +197,20 @@ def update_task(task_id: int, payload: TaskUpdate, db: Session = Depends(get_db) if "connection_id" in update_data: _require_existing_connection(update_data["connection_id"]) + # Upsert invariant on the EFFECTIVE state (request merged over stored + # values): {"upsert_enabled": true} alone must not leave a task with no + # keys, or the runner silently degrades to plain inserts. Only enforced + # when the request touches the upsert fields, so an unrelated partial + # update of a legacy row is not blocked. + if "upsert_enabled" in update_data or "upsert_keys" in update_data: + effective_upsert = update_data.get("upsert_enabled", task.upsert_enabled) + effective_keys = update_data.get("upsert_keys", task.upsert_keys) + if effective_upsert and not effective_keys: + raise HTTPException( + status_code=400, + detail="upsert_enabled requires at least one column in upsert_keys", + ) + # Encrypt secrets. Empty string = explicit clear; None (explicit null) # also clears; omitted = untouched (key not in update_data). if "api_key" in update_data: @@ -573,7 +587,11 @@ def get_task_stats(task_id: int, db: Session = Depends(get_db)): else: pairs = ( db.query(TaskRun.started_at, TaskRun.ended_at) - .filter(TaskRun.task_id == task_id, TaskRun.ended_at.isnot(None)) + .filter( + TaskRun.task_id == task_id, + TaskRun.started_at.isnot(None), + TaskRun.ended_at.isnot(None), + ) .all() ) durations = [(p.ended_at - p.started_at).total_seconds() for p in pairs] diff --git a/backend/app/db/schemas/task.py b/backend/app/db/schemas/task.py index 73ef7bd..dc06c08 100644 --- a/backend/app/db/schemas/task.py +++ b/backend/app/db/schemas/task.py @@ -1,6 +1,7 @@ import re from datetime import datetime from typing import Any, Literal +from urllib.parse import urlparse from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator @@ -45,6 +46,10 @@ def _validate_source_url(url: str | None) -> str | None: """ if url is None or "://" not in url: return url + if not urlparse(url.strip()).scheme: + # Relative endpoint path whose query value happens to contain "://" + # (e.g. /api/items?redirect=https://cb.example.com); resolved elsewhere. + return url try: return validate_url(url, resolve=False) except SSRFBlockedError as e: @@ -246,16 +251,25 @@ def validate_dest_table_identifier(cls, v: str): @field_validator("upsert_keys") @classmethod - def validate_upsert_key_identifiers(cls, v: list[str] | None, info): - """Normalize entries to safe SQL identifiers and require at least one - key when upsert is enabled.""" + def validate_upsert_key_identifiers(cls, v: list[str] | None): + """Normalize entries to safe SQL identifiers.""" if v: v = [_validate_sql_identifier(k, "upsert_keys entry") for k in v] - upsert_enabled = info.data.get("upsert_enabled") - if upsert_enabled and (not v or len(v) == 0): - raise ValueError("upsert_enabled requires at least one column in upsert_keys") return v + @model_validator(mode="after") + def validate_upsert_requires_keys(self): + """upsert_enabled needs at least one key. + + A model validator (not a field validator on upsert_keys) so the rule + also runs when upsert_keys is omitted entirely — otherwise + {"upsert_enabled": true} alone was accepted and the runner silently + fell back to plain inserts. + """ + if self.upsert_enabled and not self.upsert_keys: + raise ValueError("upsert_enabled requires at least one column in upsert_keys") + return self + @field_validator("skip_column") @classmethod def validate_skip_column_identifier(cls, v: str | None): @@ -327,14 +341,17 @@ def validate_skip_column_identifier_update(cls, v: str | None): _validate_sql_identifier(v, "skip_column") return v - @field_validator("upsert_keys") - @classmethod - def validate_upsert_keys(cls, v: list[str] | None, info): - """When enabling upsert, keys must be present in the same request.""" - upsert_enabled = info.data.get("upsert_enabled") - if upsert_enabled and not v: + @model_validator(mode="after") + def validate_upsert_keys_when_supplied(self): + """Enabling upsert while explicitly sending an empty key list is invalid. + + Runs as a model validator so it also fires when upsert_keys is set to + null/[] together with upsert_enabled. When upsert_keys is omitted the + stored keys apply; update_task() validates that effective state. + """ + if self.upsert_enabled and "upsert_keys" in self.model_fields_set and not self.upsert_keys: raise ValueError("upsert_enabled=true requires at least one column in upsert_keys") - return v + return self @model_validator(mode="after") def reject_explicit_nulls_for_required_fields(self): diff --git a/backend/tests/unit/test_tasks_partial_update.py b/backend/tests/unit/test_tasks_partial_update.py index 834bf02..73d8b67 100644 --- a/backend/tests/unit/test_tasks_partial_update.py +++ b/backend/tests/unit/test_tasks_partial_update.py @@ -160,3 +160,57 @@ def test_create_rejects_non_http_scheme(self, client): if __name__ == "__main__": pytest.main([__file__, "-v"]) + + +class TestUpsertInvariant: + """upsert_enabled without any key must be rejected, at create and update time.""" + + def test_create_enabled_without_keys_is_422(self, client): + resp = client.post( + "/api/v1/tasks/", + json={ + "name": "No Keys", + "connection_id": "conn-1", + "endpoint_path": "https://api.example.com/users", + "dest_table": "users", + "upsert_enabled": True, + }, + ) + assert resp.status_code == 422 + + def test_enable_without_stored_or_supplied_keys_is_400(self, client): + task_id = _create_task(client, "Enable No Keys")["id"] + resp = client.put(f"/api/v1/tasks/{task_id}", json={"upsert_enabled": True}) + assert resp.status_code == 400 + assert "upsert_keys" in resp.json()["detail"] + + def test_enable_with_supplied_keys_then_toggle_uses_stored_keys(self, client, test_db): + task_id = _create_task(client, "Enable With Keys")["id"] + resp = client.put( + f"/api/v1/tasks/{task_id}", json={"upsert_enabled": True, "upsert_keys": ["id"]} + ) + assert resp.status_code == 200, resp.text + resp = client.put(f"/api/v1/tasks/{task_id}", json={"upsert_enabled": False}) + assert resp.status_code == 200, resp.text + # Stored keys satisfy the invariant when re-enabling without resending them. + resp = client.put(f"/api/v1/tasks/{task_id}", json={"upsert_enabled": True}) + assert resp.status_code == 200, resp.text + row = test_db.query(Task).filter(Task.id == task_id).first() + assert row.upsert_enabled is True + + def test_enable_with_explicit_empty_keys_is_422(self, client): + task_id = _create_task(client, "Enable Empty Keys")["id"] + resp = client.put( + f"/api/v1/tasks/{task_id}", json={"upsert_enabled": True, "upsert_keys": []} + ) + assert resp.status_code == 422 + + +class TestRelativeEndpointPath: + def test_relative_path_with_absolute_url_in_query_is_kept(self, client): + """A relative endpoint whose query embeds an absolute URL is not an SSRF target.""" + task_id = _create_task(client, "Relative Path")["id"] + path = "/api/items?redirect=https://cb.example.com/done" + resp = client.put(f"/api/v1/tasks/{task_id}", json={"endpoint_path": path}) + assert resp.status_code == 200, resp.text + assert resp.json()["endpoint_path"] == path From 9df83f24cee3e34d3adf94d3da10e2cb3ee0df85 Mon Sep 17 00:00:00 2001 From: badry-dev Date: Tue, 1 Sep 2026 20:45:35 +0000 Subject: [PATCH 5/9] chore(ci,deps): least-privilege token, Node 22 with engines field, idna floor, quoted Redis healthcheck MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ci.yml: top-level permissions: contents: read (nothing in the workflow writes); frontend jobs on Node 22 — Node 20 is EOL and Vite 8 requires ^20.19.0 || >=22.12.0 — matched by an engines.node field in package.json. - requirements.txt: idna>=3.15 (PYSEC-2026-215); httpx leaves it unconstrained. - docker-compose.yml: quote $REDIS_PASSWORD in the redis-cli healthcheck. Claude-Session: https://claude.ai/code/session_018bq3je2VeLdrF71FThQVvU --- .github/workflows/ci.yml | 12 ++++++++---- backend/requirements.txt | 2 ++ docker-compose.yml | 2 +- frontend/package.json | 3 +++ 4 files changed, 14 insertions(+), 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 49fbb98..c537df9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,5 +1,9 @@ name: CI +# Least-privilege token for every job; nothing here writes to the repo. +permissions: + contents: read + on: push: branches: [main] @@ -118,7 +122,7 @@ jobs: - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0 with: - node-version: "20" + node-version: "22" cache: npm cache-dependency-path: frontend/package-lock.json @@ -144,7 +148,7 @@ jobs: - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0 with: - node-version: "20" + node-version: "22" cache: npm cache-dependency-path: frontend/package-lock.json @@ -167,7 +171,7 @@ jobs: - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0 with: - node-version: "20" + node-version: "22" cache: npm cache-dependency-path: frontend/package-lock.json @@ -197,7 +201,7 @@ jobs: - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0 with: - node-version: "20" + node-version: "22" cache: npm cache-dependency-path: frontend/package-lock.json diff --git a/backend/requirements.txt b/backend/requirements.txt index 8872b0e..4fb4157 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -7,6 +7,8 @@ SQLAlchemy==2.0.36 alembic==1.13.3 oracledb==3.4.2 httpx==0.27.2 +# httpx leaves idna unconstrained; <3.15 is vulnerable (PYSEC-2026-215 / GHSA-65pc-fj4g-8rjx) +idna>=3.15 jsonpath-ng==1.6.1 celery==5.4.0 redis==5.0.8 diff --git a/docker-compose.yml b/docker-compose.yml index 54fccae..ac36d19 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -13,7 +13,7 @@ services: # For the authenticated healthcheck ($$REDIS_PASSWORD at runtime) REDIS_PASSWORD: ${REDIS_PASSWORD:?REDIS_PASSWORD required in .env} healthcheck: - test: ["CMD-SHELL", "redis-cli -a $$REDIS_PASSWORD ping | grep PONG"] + test: ["CMD-SHELL", "redis-cli -a \"$$REDIS_PASSWORD\" ping | grep PONG"] interval: 10s timeout: 5s retries: 5 diff --git a/frontend/package.json b/frontend/package.json index a2a1b11..13d7824 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -3,6 +3,9 @@ "private": true, "version": "0.1.0", "type": "module", + "engines": { + "node": "^20.19.0 || >=22.12.0" + }, "scripts": { "dev": "vite", "build": "tsc -b && vite build", From 3263b37ed101d50e2d5a11ef0d0d25d6068e296c Mon Sep 17 00:00:00 2001 From: badry-dev Date: Tue, 1 Sep 2026 20:45:35 +0000 Subject: [PATCH 6/9] test(storage): run lock-contention workers from the backend dir and kill them on timeout The worker processes 'import app', so derive cwd from __file__ instead of os.getcwd(); terminate all workers if communicate() times out so a hung worker cannot leave the suite waiting. Claude-Session: https://claude.ai/code/session_018bq3je2VeLdrF71FThQVvU --- .../tests/unit/test_connection_storage_atomic.py | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/backend/tests/unit/test_connection_storage_atomic.py b/backend/tests/unit/test_connection_storage_atomic.py index 6e9170d..df82347 100644 --- a/backend/tests/unit/test_connection_storage_atomic.py +++ b/backend/tests/unit/test_connection_storage_atomic.py @@ -1,6 +1,7 @@ """Tests for atomic, locked connections-file storage (v1.4 H4).""" import os +from pathlib import Path import pytest @@ -104,6 +105,9 @@ def test_interleaved_read_modify_write_preserves_both(self, tmp_path): from app.core.config import settings as app_settings parent_key = app_settings.ENCRYPTION_KEY or os.environ.get("ENCRYPTION_KEY", "") + # Workers `import app`, so run them from the backend dir regardless of + # where pytest was launched. + backend_dir = str(Path(__file__).resolve().parents[2]) procs = [] for i in range(n): procs.append( @@ -117,13 +121,20 @@ def test_interleaved_read_modify_write_preserves_both(self, tmp_path): str(ready_dir), str(n), ], - cwd=os.getcwd(), + cwd=backend_dir, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env={**os.environ, "ENCRYPTION_KEY": parent_key}, ) ) - outs = [p.communicate(timeout=60) for p in procs] + outs = [] + for p in procs: + try: + outs.append(p.communicate(timeout=60)) + except subprocess.TimeoutExpired: + for q in procs: + q.kill() + raise failed = [ (p.returncode, err.decode()[-500:]) for p, (_, err) in zip(procs, outs) From a9527a206a01ab8825b439d8e7b2912a1964b1f6 Mon Sep 17 00:00:00 2001 From: badry-dev Date: Tue, 1 Sep 2026 20:45:35 +0000 Subject: [PATCH 7/9] fix(frontend): say when the row-error list is truncated The card title reported the uncapped row_errors_total while the table rendered only the capped list. Show 'showing N of M' when truncated and document the row_errors_limit parameter and its bounds on the type. Claude-Session: https://claude.ai/code/session_018bq3je2VeLdrF71FThQVvU --- frontend/src/pages/RunDetail.tsx | 8 +++++++- frontend/src/types/index.ts | 3 ++- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/frontend/src/pages/RunDetail.tsx b/frontend/src/pages/RunDetail.tsx index 88b19c3..841a928 100644 --- a/frontend/src/pages/RunDetail.tsx +++ b/frontend/src/pages/RunDetail.tsx @@ -157,7 +157,13 @@ export function RunDetail() { {/* Row Errors */} {run.row_errors && run.row_errors.length > 0 && ( - + run.row_errors.length + ? `Row Errors (showing ${run.row_errors.length} of ${run.row_errors_total})` + : `Row Errors (${run.row_errors.length})` + } + > ({ ...err, key: idx }))} diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 1d9f77b..4c6d1eb 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -266,7 +266,8 @@ export interface TaskRun { replay_of_run_id?: number | null execution_logs?: TaskLog[] row_errors?: TaskRunLog[] - // Uncapped count of row errors (row_errors list is capped at 500 by the API) + // Uncapped count of row errors. row_errors itself is capped by the + // row_errors_limit query parameter (default 500, max 5000). row_errors_total?: number } From 48d8541683dba492a9fd2067f4badc51063db040 Mon Sep 17 00:00:00 2001 From: badry-dev Date: Tue, 1 Sep 2026 21:01:27 +0000 Subject: [PATCH 8/9] fix(http): let SSRFBlockedError survive fetch_sample_response so previews return 403 fetch_sample_response ended with 'except Exception: raise ValueError(...)', which flattened a fetch-time SSRFBlockedError raised by fetch_json into a plain ValueError. The 403 handlers added in the preview endpoints were therefore unreachable for real blocks (saved-task previews, and standalone previews when DNS changes between the pre-check and the fetch); the endpoint test masked this by mocking fetch_sample_response itself. Re-raise SSRFBlockedError unchanged inside the helper, patch fetch_json in the endpoint test so the error must cross the wrapper, and add a direct unit test on the helper. Claude-Session: https://claude.ai/code/session_018bq3je2VeLdrF71FThQVvU --- backend/app/services/api_connector.py | 4 ++++ backend/tests/integration/test_api_endpoints.py | 8 ++++++-- backend/tests/unit/test_url_guard.py | 17 +++++++++++++++++ 3 files changed, 27 insertions(+), 2 deletions(-) diff --git a/backend/app/services/api_connector.py b/backend/app/services/api_connector.py index f7a02b5..04c4b32 100644 --- a/backend/app/services/api_connector.py +++ b/backend/app/services/api_connector.py @@ -573,6 +573,10 @@ async def fetch_sample_response( logger.info(f"Fetched sample API response: {type(response_data).__name__}") return response_data + except SSRFBlockedError: + # Preserve the SSRF signal: callers map it to 403. The catch-all below + # would flatten it into a generic ValueError (a misleading 400). + raise except httpx.HTTPStatusError as e: # Status + reason only: upstream bodies can contain PII, so neither # the log line nor the client-facing message includes an excerpt. diff --git a/backend/tests/integration/test_api_endpoints.py b/backend/tests/integration/test_api_endpoints.py index dfa68b2..613c117 100644 --- a/backend/tests/integration/test_api_endpoints.py +++ b/backend/tests/integration/test_api_endpoints.py @@ -460,14 +460,18 @@ def test_root_endpoint(client: TestClient): def test_standalone_preview_fetch_time_ssrf_block_is_403(client: TestClient): """SSRFBlockedError subclasses ValueError; before the dedicated handler it - came back as 400 "Invalid JSON", hiding that the URL was blocked.""" + came back as 400 "Invalid JSON", hiding that the URL was blocked. + + The block is raised from fetch_json, i.e. underneath fetch_sample_response, + so it must survive that helper's catch-all wrapper to reach the route. + """ from unittest.mock import patch from app.core.url_guard import SSRFBlockedError path = "/api/v1/preview-fields-standalone" with patch( - "app.api.v1.routes.column_mappings.fetch_sample_response", + "app.services.api_connector.fetch_json", side_effect=SSRFBlockedError("Host 'api.example.com' resolves to a non-public address"), ): resp = client.post( diff --git a/backend/tests/unit/test_url_guard.py b/backend/tests/unit/test_url_guard.py index 2257ee8..ad2d0fc 100644 --- a/backend/tests/unit/test_url_guard.py +++ b/backend/tests/unit/test_url_guard.py @@ -99,3 +99,20 @@ def test_out_of_range_port_rejected_without_resolving(self, url): # config-time validation (resolve=False), not a ValueError/500. with pytest.raises(SSRFBlockedError, match="invalid port"): validate_url(url, resolve=False) + + +class TestFetchSampleResponsePreservesSsrfError: + @pytest.mark.asyncio + async def test_ssrf_block_is_not_flattened_into_value_error(self): + from unittest.mock import patch + + from app.services.api_connector import fetch_sample_response + + with patch( + "app.services.api_connector.fetch_json", + side_effect=SSRFBlockedError("Host 'x' resolves to a non-public address"), + ): + with pytest.raises(SSRFBlockedError): + await fetch_sample_response( + method="GET", url="https://api.example.com/x", auth_type="none" + ) From 5faccc7ca8e395551f10f674988ec70c0ccc097f Mon Sep 17 00:00:00 2001 From: badry-dev Date: Tue, 1 Sep 2026 21:04:06 +0000 Subject: [PATCH 9/9] fix(http): strip URL path parameters from log redaction; tighten three test isolations - _redact_url_for_log kept parts.path verbatim, so ;jsessionid=SECRET style path parameters reached every log line built with the helper. Path parameters are now dropped segment by segment; regression case added. - The fetch-time SSRF endpoint test stubs validate_url_async explicitly so the fetch-time path is what is exercised, independent of the conftest bypass. - The default-threshold test deletes SCHEDULE_MAX_CONSECUTIVE_FAILURES from the environment and passes _env_file=None so neither can mask the default. - The lock-contention test calls communicate() after kill() on workers that are still running, draining pipes and populating return codes. Claude-Session: https://claude.ai/code/session_018bq3je2VeLdrF71FThQVvU --- backend/app/services/api_connector.py | 4 +++- backend/tests/integration/test_api_endpoints.py | 15 +++++++++++---- backend/tests/unit/test_config_env.py | 6 ++++-- .../tests/unit/test_connection_storage_atomic.py | 4 +++- backend/tests/unit/test_url_redaction.py | 4 ++++ 5 files changed, 25 insertions(+), 8 deletions(-) diff --git a/backend/app/services/api_connector.py b/backend/app/services/api_connector.py index 04c4b32..9f97645 100644 --- a/backend/app/services/api_connector.py +++ b/backend/app/services/api_connector.py @@ -66,7 +66,9 @@ def _redact_url_for_log(url: str) -> str: except ValueError: return "" netloc = parts.netloc.rpartition("@")[2] # drop user:secret@ - redacted = urlunsplit((parts.scheme, netloc, parts.path, "", "")) + # Drop ;param=value path parameters segment by segment (e.g. ;jsessionid=...). + path = "/".join(segment.split(";", 1)[0] for segment in parts.path.split("/")) + redacted = urlunsplit((parts.scheme, netloc, path, "", "")) return _LOG_CONTROL_CHARS_RE.sub("", redacted) diff --git a/backend/tests/integration/test_api_endpoints.py b/backend/tests/integration/test_api_endpoints.py index 613c117..f9dd0d4 100644 --- a/backend/tests/integration/test_api_endpoints.py +++ b/backend/tests/integration/test_api_endpoints.py @@ -465,14 +465,21 @@ def test_standalone_preview_fetch_time_ssrf_block_is_403(client: TestClient): The block is raised from fetch_json, i.e. underneath fetch_sample_response, so it must survive that helper's catch-all wrapper to reach the route. """ - from unittest.mock import patch + from unittest.mock import AsyncMock, patch from app.core.url_guard import SSRFBlockedError path = "/api/v1/preview-fields-standalone" - with patch( - "app.services.api_connector.fetch_json", - side_effect=SSRFBlockedError("Host 'api.example.com' resolves to a non-public address"), + with ( + # The pre-fetch check must pass so the fetch-time path is what is exercised. + patch( + "app.api.v1.routes.column_mappings.validate_url_async", + new=AsyncMock(return_value="https://api.example.com/x"), + ), + patch( + "app.services.api_connector.fetch_json", + side_effect=SSRFBlockedError("Host 'api.example.com' resolves to a non-public address"), + ), ): resp = client.post( path, diff --git a/backend/tests/unit/test_config_env.py b/backend/tests/unit/test_config_env.py index 7b28a0e..6909153 100644 --- a/backend/tests/unit/test_config_env.py +++ b/backend/tests/unit/test_config_env.py @@ -89,8 +89,10 @@ def test_docs_disabled_in_production_canonical_value(self): class TestScheduleFailureThreshold: - def test_default_is_five(self): - assert Settings(SECRET_KEY="x").SCHEDULE_MAX_CONSECUTIVE_FAILURES == 5 + def test_default_is_five(self, monkeypatch): + monkeypatch.delenv("SCHEDULE_MAX_CONSECUTIVE_FAILURES", raising=False) + settings = Settings(_env_file=None, SECRET_KEY="x") + assert settings.SCHEDULE_MAX_CONSECUTIVE_FAILURES == 5 @pytest.mark.parametrize("bad", [0, -1]) def test_non_positive_threshold_rejected(self, bad): diff --git a/backend/tests/unit/test_connection_storage_atomic.py b/backend/tests/unit/test_connection_storage_atomic.py index df82347..a3f94fa 100644 --- a/backend/tests/unit/test_connection_storage_atomic.py +++ b/backend/tests/unit/test_connection_storage_atomic.py @@ -133,7 +133,9 @@ def test_interleaved_read_modify_write_preserves_both(self, tmp_path): outs.append(p.communicate(timeout=60)) except subprocess.TimeoutExpired: for q in procs: - q.kill() + if q.poll() is None: + q.kill() + q.communicate() # drain pipes, populate returncode raise failed = [ (p.returncode, err.decode()[-500:]) diff --git a/backend/tests/unit/test_url_redaction.py b/backend/tests/unit/test_url_redaction.py index 58f7e59..d678948 100644 --- a/backend/tests/unit/test_url_redaction.py +++ b/backend/tests/unit/test_url_redaction.py @@ -13,6 +13,10 @@ ("https://api.example.com:8443/v1/users#frag", "https://api.example.com:8443/v1/users"), ("https://user:s3cret@[::1]:8080/p?x=1", "https://[::1]:8080/p"), ("https://api.example.com/v1/users", "https://api.example.com/v1/users"), + ( + "https://api.example.com/v1;jsessionid=SECRET/users;v=2?x=1", + "https://api.example.com/v1/users", + ), ], ) def test_query_userinfo_and_fragment_are_dropped(raw, expected):