fix(data): robust CSV parser, migration run state machine, shell script hardening

backend/database/models.py

@@ -25,13 +25,68 @@ class Base(DeclarativeBase):
 
 
 class RunStatus(str, enum.Enum):
-    """Lifecycle states for a migration run."""
+    """Lifecycle states for a migration run.
+
+    State machine::
+
+        CREATED ──▶ UPLOADING ──▶ VALIDATING ──▶ READY ──▶ MIGRATING ──▶ COMPLETED
+           │            │             │            │           │
+           └────────────┴─────────────┴────────────┴───────────┴──▶ ERROR / FAILED
+
+    ``ERROR`` marks a run whose upload/validation broke mid-way (recoverable by
+    re-uploading or re-validating); ``FAILED`` marks a migration execution that
+    ran and did not succeed.
+    """
     CREATED = "created"
     UPLOADING = "uploading"
     VALIDATING = "validating"
+    READY = "ready"
     MIGRATING = "migrating"
     COMPLETED = "completed"
     FAILED = "failed"
+    ERROR = "error"
+
+
+class InvalidStateTransition(Exception):
+    """Raised when a migration run is moved to a state it cannot reach."""
+
+
+# Explicit allowed transitions. Same-state transitions are always allowed
+# (idempotent updates, e.g. uploading a second file while UPLOADING).
+ALLOWED_TRANSITIONS: dict[RunStatus, frozenset[RunStatus]] = {
+    RunStatus.CREATED: frozenset({
+        RunStatus.UPLOADING, RunStatus.VALIDATING, RunStatus.MIGRATING, RunStatus.ERROR,
+    }),
+    RunStatus.UPLOADING: frozenset({
+        RunStatus.VALIDATING, RunStatus.READY, RunStatus.MIGRATING,
+        RunStatus.ERROR, RunStatus.CREATED,
+    }),
+    RunStatus.VALIDATING: frozenset({
+        RunStatus.READY, RunStatus.ERROR, RunStatus.CREATED,
+    }),
+    RunStatus.READY: frozenset({
+        RunStatus.UPLOADING, RunStatus.VALIDATING, RunStatus.MIGRATING, RunStatus.ERROR,
+    }),
+    RunStatus.MIGRATING: frozenset({
+        RunStatus.COMPLETED, RunStatus.FAILED, RunStatus.ERROR, RunStatus.CREATED,
+    }),
+    RunStatus.COMPLETED: frozenset({
+        RunStatus.VALIDATING, RunStatus.MIGRATING, RunStatus.ERROR,
+    }),
+    RunStatus.FAILED: frozenset({
+        RunStatus.UPLOADING, RunStatus.VALIDATING, RunStatus.MIGRATING, RunStatus.ERROR,
+    }),
+    RunStatus.ERROR: frozenset({
+        RunStatus.CREATED, RunStatus.UPLOADING, RunStatus.VALIDATING, RunStatus.MIGRATING,
+    }),
+}
+
+
+def can_transition(current: RunStatus, target: RunStatus) -> bool:
+    """Return True when ``current`` → ``target`` is a legal state change."""
+    if current == target:
+        return True
+    return target in ALLOWED_TRANSITIONS.get(current, frozenset())
 
 
 class MigrationRun(Base):
@@ -65,6 +120,20 @@ class MigrationRun(Base):
         "UploadedFile", back_populates="migration_run", cascade="all, delete-orphan"
     )
 
+    def transition_to(self, target: "RunStatus") -> None:
+        """Move the run to ``target``, enforcing the lifecycle state machine.
+
+        Raises ``InvalidStateTransition`` for illegal moves (e.g. COMPLETED →
+        UPLOADING) so bugs surface immediately instead of corrupting run state.
+        """
+        current = self.status if isinstance(self.status, RunStatus) else RunStatus(self.status)
+        if not can_transition(current, target):
+            raise InvalidStateTransition(
+                f"Migration run {self.id}: illegal transition "
+                f"{current.value!r} → {target.value!r}"
+            )
+        self.status = target
+
     def __repr__(self) -> str:
         return f"<MigrationRun id={self.id} name={self.name!r} status={self.status}>"
 

backend/migration/build/csv_loader.sh

@@ -89,6 +89,13 @@ done
 [[ -z "$CSV_FILE" ]] && { error "No CSV file specified."; usage; }
 [[ ! -f "$CSV_FILE" ]] && { error "File not found: $CSV_FILE"; exit 1; }
 
+# Validate the env token: it is interpolated into variable names by the engine
+# adapters (some via eval), so it must never contain shell metacharacters.
+if [[ ! "$TARGET_ENV" =~ ^[a-zA-Z_][a-zA-Z0-9_]*$ ]]; then
+   error "Invalid --env value: '${TARGET_ENV}'. Allowed: letters, digits, underscore."
+   exit 1
+fi
+
 # ── Load configuration ────────────────────────────────────────────────────────
 if [[ -f "$CONFIG_LOCAL" ]]; then
    source "$CONFIG_LOCAL"
@@ -118,6 +125,20 @@ TABLE_NAME="${TABLE_NAME%.CSV}"
 # Sanitise: lowercase, replace spaces/hyphens with underscores
 TABLE_NAME="$(echo "$TABLE_NAME" | tr '[:upper:]' '[:lower:]' | tr ' -' '__')"
 
+# ── Validate table identifier (blocks `;`, `--`, quotes, spaces, etc.) ───────
+# The table name is interpolated into generated SQL, so it must be a strict
+# SQL identifier: letters/digits/underscore, not starting with a digit.
+if [[ ! "$TABLE_NAME" =~ ^[a-zA-Z_][a-zA-Z0-9_]*$ ]]; then
+   error "Invalid table name derived from input: '${TABLE_NAME}'."
+   error "Allowed: letters, digits, underscore; must not start with a digit."
+   error "Rename the CSV file or pass a safe name with --table <name>."
+   exit 1
+fi
+if [[ "${#TABLE_NAME}" -gt 63 ]]; then
+   error "Table name '${TABLE_NAME}' exceeds PostgreSQL's 63-character identifier limit."
+   exit 1
+fi
+
 # ── Setup log directory and files ─────────────────────────────────────────────
 mkdir -p "$LOG_DIR"
 TIMESTAMP="$(date +%Y%m%d_%H%M%S)"

backend/migration/build/csv_utilise.sh

@@ -146,12 +146,21 @@ if [[ "$DB_ENGINE" != "postgresql" ]]; then
 fi
 
 # ── Resolve PostgreSQL connection details ────────────────────────────────────
+# Validate the env token first: it is used to build variable names, so it must
+# never contain shell metacharacters (blocks `--env 'x; rm -rf ~'` style abuse).
+if [[ ! "$TARGET_ENV" =~ ^[a-zA-Z_][a-zA-Z0-9_]*$ ]]; then
+   error "Invalid --env value: '${TARGET_ENV}'. Allowed: letters, digits, underscore."
+   exit 1
+fi
 E="${TARGET_ENV^^}"
 PG_HOST="${PGHOST:-${PG_HOST:-localhost}}"
 PG_PORT="${PGPORT:-${PG_PORT:-5432}}"
 PG_USER="${PGUSER:-${PG_SUPERUSER:-postgres}}"
-DB_NAME="$(eval echo "\$PG_DB_${E}")"
-SCHEMA="$(eval echo "\$PG_SCHEMA_${E}")"
+# Indirect expansion instead of eval — no code execution path.
+_db_var="PG_DB_${E}"
+_schema_var="PG_SCHEMA_${E}"
+DB_NAME="${!_db_var:-}"
+SCHEMA="${!_schema_var:-}"
 
 if [[ -z "$DB_NAME" || -z "$SCHEMA" ]]; then
    error "Could not resolve database / schema for env '${TARGET_ENV}'."

backend/services/csv_parser.py

@@ -0,0 +1,222 @@
+"""Robust CSV parsing for uploaded migration files.
+
+``CsvParser`` centralises every defensive check that used to be scattered (or
+silently swallowed) in ``migration_service._parse_csv_metadata``:
+
+  * zero-byte / header-only files
+  * encoding fallback (UTF-8 with BOM → Latin-1) with an explicit warning
+  * ragged rows (data rows whose column count differs from the header)
+  * suspicious header names that could smuggle SQL fragments into generated
+    DDL (``;``, ``--``, quotes, comment markers, DROP/DELETE/... keywords)
+
+The parser never raises for malformed *content* — it returns a
+``CsvParseResult`` whose ``issues`` list explains everything found, so the
+caller decides whether to reject the file or store it with warnings.
+"""
+from __future__ import annotations
+
+import csv
+import io
+import json
+import re
+from dataclasses import dataclass, field
+from typing import Optional
+
+# Characters/sequences in a header that are never legitimate column names and
+# are classic SQL-injection vectors when identifiers are interpolated in DDL.
+_FORBIDDEN_HEADER_PATTERN = re.compile(r"""[;'"`\\]|--|/\*|\*/""")
+
+# Statement keywords that should never appear as a standalone word inside a
+# column header (e.g. ``name; DROP TABLE users``).
+_FORBIDDEN_KEYWORDS = re.compile(
+    r"\b(drop|delete|truncate|insert|update|alter|grant|revoke|exec|execute)\b",
+    re.IGNORECASE,
+)
+
+# Maximum header length we accept — anything longer is suspicious and would
+# exceed PostgreSQL's 63-byte identifier limit anyway.
+MAX_HEADER_LENGTH = 128
+
+
+@dataclass
+class CsvIssue:
+    """A single problem found while parsing a CSV file."""
+
+    severity: str  # "error" | "warning"
+    check: str     # machine-readable check name
+    message: str   # human-readable explanation
+
+    def as_dict(self) -> dict:
+        return {"severity": self.severity, "check": self.check, "message": self.message}
+
+
+@dataclass
+class CsvParseResult:
+    """Outcome of parsing one CSV payload."""
+
+    ok: bool = False
+    row_count: Optional[int] = None
+    column_count: Optional[int] = None
+    columns: Optional[list[str]] = None
+    encoding: Optional[str] = None
+    issues: list[CsvIssue] = field(default_factory=list)
+
+    @property
+    def errors(self) -> list[CsvIssue]:
+        return [i for i in self.issues if i.severity == "error"]
+
+    @property
+    def warnings(self) -> list[CsvIssue]:
+        return [i for i in self.issues if i.severity == "warning"]
+
+    def columns_json(self) -> Optional[str]:
+        """Column names serialised as JSON, matching the legacy DB format."""
+        return json.dumps(self.columns) if self.columns is not None else None
+
+
+class CsvParser:
+    """Defensive CSV parser used by the upload pipeline."""
+
+    def __init__(self, max_ragged_examples: int = 5) -> None:
+        # How many ragged-row line numbers to include in the issue message.
+        self.max_ragged_examples = max_ragged_examples
+
+    # -- decoding ----------------------------------------------------------
+
+    def _decode(self, content: bytes, result: CsvParseResult) -> Optional[str]:
+        """Decode bytes to text, preferring UTF-8 (BOM-aware), falling back
+        to Latin-1 with a warning. Returns None if undecodable."""
+        try:
+            result.encoding = "utf-8"
+            return content.decode("utf-8-sig")
+        except UnicodeDecodeError:
+            pass
+        try:
+            text = content.decode("latin-1")
+            result.encoding = "latin-1"
+            result.issues.append(CsvIssue(
+                severity="warning",
+                check="encoding_fallback",
+                message="File is not valid UTF-8; decoded as Latin-1. "
+                        "Non-ASCII characters may be misinterpreted.",
+            ))
+            return text
+        except UnicodeDecodeError:  # pragma: no cover — latin-1 never fails
+            result.issues.append(CsvIssue(
+                severity="error",
+                check="undecodable",
+                message="File could not be decoded as UTF-8 or Latin-1.",
+            ))
+            return None
+
+    # -- header checks -----------------------------------------------------
+
+    def _check_headers(self, headers: list[str], result: CsvParseResult) -> None:
+        for idx, header in enumerate(headers):
+            stripped = header.strip()
+            if not stripped:
+                result.issues.append(CsvIssue(
+                    severity="warning",
+                    check="empty_header",
+                    message=f"Column {idx + 1} has an empty header name.",
+                ))
+                continue
+            if len(stripped) > MAX_HEADER_LENGTH:
+                result.issues.append(CsvIssue(
+                    severity="error",
+                    check="header_too_long",
+                    message=f"Header {idx + 1} exceeds {MAX_HEADER_LENGTH} characters.",
+                ))
+            if _FORBIDDEN_HEADER_PATTERN.search(stripped) or _FORBIDDEN_KEYWORDS.search(stripped):
+                result.issues.append(CsvIssue(
+                    severity="error",
+                    check="suspicious_header",
+                    message=(
+                        f"Header {idx + 1} ({stripped[:64]!r}) contains characters or "
+                        "SQL keywords that are not allowed in column names."
+                    ),
+                ))
+
+    # -- main entry point ----------------------------------------------------
+
+    def parse(self, content: bytes, filename: str = "upload.csv") -> CsvParseResult:
+        """Parse CSV bytes and return a fully-populated ``CsvParseResult``.
+
+        Never raises for malformed content; every problem is reported in
+        ``result.issues`` and ``result.ok`` is False when any error exists.
+        """
+        result = CsvParseResult()
+
+        # 1. Zero-byte / whitespace-only guard
+        if not content or not content.strip():
+            result.issues.append(CsvIssue(
+                severity="error",
+                check="empty_file",
+                message=f"File {filename!r} is empty (0 bytes of data).",
+            ))
+            return result
+
+        # 2. Decode with fallback
+        text = self._decode(content, result)
+        if text is None:
+            return result
+
+        # 3. Header row
+        reader = csv.reader(io.StringIO(text))
+        try:
+            headers = next(reader, None)
+        except csv.Error as exc:
+            result.issues.append(CsvIssue(
+                severity="error",
+                check="malformed_csv",
+                message=f"CSV structure could not be parsed: {exc}",
+            ))
+            return result
+
+        if not headers or all(not h.strip() for h in headers):
+            result.issues.append(CsvIssue(
+                severity="error",
+                check="missing_header",
+                message="No header row found.",
+            ))
+            return result
+
+        result.columns = [h.strip() for h in headers]
+        result.column_count = len(headers)
+        self._check_headers(headers, result)
+
+        # 4. Data rows — count and detect ragged rows
+        expected = len(headers)
+        row_count = 0
+        ragged: list[int] = []
+        try:
+            for line_no, row in enumerate(reader, start=2):
+                if not row:  # skip completely blank lines
+                    continue
+                row_count += 1
+                if len(row) != expected:
+                    ragged.append(line_no)
+        except csv.Error as exc:
+            result.issues.append(CsvIssue(
+                severity="error",
+                check="malformed_csv",
+                message=f"CSV structure broke mid-file: {exc}",
+            ))
+            return result
+
+        result.row_count = row_count
+        if ragged:
+            sample = ", ".join(map(str, ragged[: self.max_ragged_examples]))
+            more = f" (+{len(ragged) - self.max_ragged_examples} more)" \
+                if len(ragged) > self.max_ragged_examples else ""
+            result.issues.append(CsvIssue(
+                severity="error",
+                check="ragged_rows",
+                message=(
+                    f"{len(ragged)} row(s) have a column count different from the "
+                    f"header ({expected} expected). Lines: {sample}{more}."
+                ),
+            ))
+
+        result.ok = not result.errors
+        return result

backend/services/execution_service.py

@@ -100,7 +100,7 @@ def execute_migration(db: Session, run_id: int) -> Optional[dict[str, Any]]:
     if not run:
         return None
 
-    run.status = RunStatus.MIGRATING
+    run.transition_to(RunStatus.MIGRATING)
     db.commit()
 
     files = (
@@ -110,7 +110,7 @@ def execute_migration(db: Session, run_id: int) -> Optional[dict[str, Any]]:
     )
 
     if not files:
-        run.status = RunStatus.CREATED
+        run.transition_to(RunStatus.CREATED)
         db.commit()
         return {
             "run_id": run_id,
@@ -178,7 +178,7 @@ def execute_migration(db: Session, run_id: int) -> Optional[dict[str, Any]]:
                 "rows_loaded": 0,
             })
 
-    run.status = RunStatus.COMPLETED if not has_error else RunStatus.FAILED
+    run.transition_to(RunStatus.COMPLETED if not has_error else RunStatus.FAILED)
     db.commit()
 
     return {