fix: address open review findings from the v1.4 stack (#12–#15)

.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
 

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)}")

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]

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:

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]

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):

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,25 @@ 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 "<unparseable-url>"
+    netloc = parts.netloc.rpartition("@")[2]  # drop user:secret@
+    # 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)
 
 
 def _parse_retry_after(header_value: str | None) -> float | None:
@@ -564,6 +575,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.