fix: address 15 open issues — security, data integrity, tests, CI

.github/workflows/quality-gate.yml

@@ -41,6 +41,10 @@ jobs:
       - name: Component health check
         run: make health
 
+      # #169: test catalogue freshness — fails if docs/TEST_CASES.md is stale.
+      - name: Test catalogue check
+        run: python3 scripts/test_catalogue.py --check
+
       # Tier X (cross-engine) is deliberately absent: multi-engine support is
       # deferred by ADR-0001 (PostgreSQL first), so this pipeline is PostgreSQL
       # only. The scenario still exists and passes — run it with `make
@@ -165,6 +169,18 @@ jobs:
           PGDATABASE: te_mgmt_dev
         run: python3 scripts/test_report.py --strict
 
+      # #194: measure api/services coverage with per-loader floors.
+      - name: api/services coverage gate
+        env:
+          PGDATABASE: te_mgmt_dev
+        run: |
+          python -m pytest tests/ \
+            --cov=api/services \
+            --cov-report=json:coverage-services.json \
+            --cov-report=term-missing \
+            -q --no-header 2>&1 | tail -20
+          python3 scripts/check_coverage_floors.py coverage-services.json
+
       - name: Upload eval reports
         if: always()
         uses: actions/upload-artifact@v7

README.md

@@ -143,7 +143,7 @@ PostgreDataMigrationAppWithCSVLoader/
 │
 ├── env.dev.example                   ← dev environment template
 ├── docs/ENV_VARIABLES.md                  ← all environment variables documented
-├── docs/API_INTEGRATION.md                ← frontend ↔ API endpoint mapping
+├── docs/API_INTEGRATION.md                ← API design notes: auth, known gaps, migrations
 ├── docs/QUICKSTART.md                     ← 10-minute setup guide
 └── docs/VCRM.md                           ← business requirement traceability
 ```

api/ratelimit.py

@@ -88,17 +88,16 @@ def reset(self) -> None:
 
 
 def _client_id(request: Request) -> str:
-    """Identify the caller.
+    """Identify the caller by a trusted attribute.
 
-    Prefers the API key (a deployed caller's stable identity) over the socket
-    address, because several callers can share one NAT address. Falls back to
-    the peer IP. ``X-Forwarded-For`` is deliberately ignored: it is
-    client-controlled, so trusting it would let anyone bypass the limit by
-    varying the header.
+    Uses the peer IP as the rate-limit identity. The API key is NOT used here
+    because it is checked before authentication: an attacker could rotate
+    arbitrary X-API-Key values to get a fresh rate-limit bucket each time,
+    bypassing brute-force protection (#183).
+
+    ``X-Forwarded-For`` is deliberately ignored: it is client-controlled, so
+    trusting it would let anyone bypass the limit by varying the header.
     """
-    api_key = request.headers.get("x-api-key")
-    if api_key:
-        return f"key:{api_key[:8]}"
     client = request.client
     return f"ip:{client.host}" if client and client.host else "ip:unknown"
 

api/routers/csv_routes.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import base64
+import binascii
 import logging
 from urllib.parse import quote
 
@@ -35,13 +37,22 @@ class PreviewRequest(BaseModel):
     content: str = Field(min_length=1)
 
 
+class HeaderRename(BaseModel):
+    original: str = Field(max_length=255)
+    renamed: str = Field(default="", max_length=128)  # blank keeps the original
+
+
 class UploadRequest(BaseModel):
     fileName: str = Field(min_length=1, max_length=255)
     content: str = Field(min_length=1)
     types: list[str] | None = None
     overwrite: bool = False
     mode: str = "dynamic"  # "dynamic" | "te"
     targetTable: str | None = None  # required when mode == "te"
+    headerMapping: list[HeaderRename] | None = Field(default=None, max_length=1000)  # dynamic mode only (#186)
+    # The file's bytes as the browser read them, base64. Kept as the preserved
+    # original so its hash matches the uploader's copy; `content` is still what loads (#185).
+    contentBase64: str | None = None
 
 
 @router.post("/preview", dependencies=[Depends(allow("preview"))])
@@ -69,21 +80,40 @@ def preview(req: PreviewRequest) -> dict:
     return result
 
 
+def _decode_original(b64: str | None) -> bytes | None:
+    if b64 is None:
+        return None
+    # base64 inflates by 4/3; refuse before decoding rather than after.
+    if len(b64) > settings.MAX_UPLOAD_BYTES * 4 // 3 + 4:
+        raise HTTPException(413, "File too large")
+    try:
+        return base64.b64decode(b64, validate=True)
+    except (binascii.Error, ValueError):
+        raise HTTPException(422, "contentBase64 is not valid base64") from None
+
+
 @router.post("/upload", dependencies=[Depends(allow("upload"))])
-def upload(req: UploadRequest) -> dict:
+def upload(req: UploadRequest, session: dict | None = Depends(tenant_scope)) -> dict:
     if len(req.content.encode("utf-8")) > settings.MAX_UPLOAD_BYTES:
         raise HTTPException(413, "File too large")
+    original = _decode_original(req.contentBase64)
     if req.mode == "te":
         if not req.targetTable:
             raise HTTPException(422, "targetTable is required when mode is 'te'")
-        result = upload_te(req.fileName, req.content, req.targetTable)
+        if req.headerMapping:
+            # T&E columns must match the fixed schema exactly; a rename could
+            # only be ignored, which is the silent failure #186 describes.
+            raise HTTPException(422, "headerMapping is only supported in 'dynamic' mode")
+        result = upload_te(req.fileName, req.content, req.targetTable, session=session)
     elif req.mode == "dynamic":
-        result = upload_dynamic(req.fileName, req.content, req.types, req.overwrite)
+        mapping = [m.model_dump() for m in req.headerMapping] if req.headerMapping else None
+        result = upload_dynamic(req.fileName, req.content, req.types, req.overwrite,
+                                session=session, header_mapping=mapping)
     else:
         raise HTTPException(422, f"Unknown mode '{req.mode}' — use 'dynamic' or 'te'")
 
     # M3b (#107): keep the original and its outcome, whatever the outcome was.
-    upload_id = record_upload(req.fileName, req.content, req.mode, result)
+    upload_id = record_upload(req.fileName, req.content, req.mode, result, original=original)
     result["uploadId"] = None if upload_id is None else str(upload_id)
 
     # O-04: record the load so audit_log covers creations, not just deletions.

api/services/csv_parse.py

@@ -133,21 +133,7 @@ def sanitize_columns(headers: list[str]) -> list[str]:
     counter: dict[str, int] = {}
     out: list[str] = []
     for idx, h in enumerate(headers):
-        base = (h or f"column_{idx + 1}").lower().strip()
-        base = re.sub(r"[^a-z0-9_]+", "_", base)
-        # Check reserved names (all of which have a leading underscore, e.g.
-        # "_id") before stripping underscores — otherwise "_id" strips to
-        # "id" first and never matches _RESERVED, silently losing the
-        # collision guard against the real system columns.
-        if base in _RESERVED:
-            base = f"{base}_col"
-        base = base.strip("_")
-        if not base:
-            base = f"column_{idx + 1}"
-        if base[0].isdigit():
-            base = f"col_{base}"
-        if len(base) > 55:
-            base = base[:55]
+        base = _base_column_name(h, idx)
         name = base
         while name in assigned:
             counter[base] = counter.get(base, 0) + 1
@@ -157,6 +143,62 @@ def sanitize_columns(headers: list[str]) -> list[str]:
     return out
 
 
+def _base_column_name(header: str, idx: int) -> str:
+    """One header's safe identifier, before de-duplication against its siblings."""
+    base = (header or f"column_{idx + 1}").lower().strip()
+    base = re.sub(r"[^a-z0-9_]+", "_", base)
+    # Check reserved names (all of which have a leading underscore, e.g.
+    # "_id") before stripping underscores — otherwise "_id" strips to
+    # "id" first and never matches _RESERVED, silently losing the
+    # collision guard against the real system columns.
+    if base in _RESERVED:
+        base = f"{base}_col"
+    base = base.strip("_")
+    if not base:
+        base = f"column_{idx + 1}"
+    if base[0].isdigit():
+        base = f"col_{base}"
+    if len(base) > 55:
+        base = base[:55]
+    return base
+
+
+def apply_header_mapping(
+    headers: list[str], mapping: list[dict] | None
+) -> tuple[list[str], str | None]:
+    """Apply user-confirmed header renames (#186) to the raw header row.
+
+    ``mapping`` is a list of ``{"original": ..., "renamed": ...}``. A blank
+    ``renamed`` keeps the original. Returns ``(headers, error)``; on error the
+    headers are returned unchanged and nothing should be loaded. A rename is
+    refused, rather than silently bent into ``name_2``, when it would collide
+    with another column — the user asked for that exact name.
+    """
+    if not mapping:
+        return headers, None
+
+    renames: dict[str, str] = {}
+    for entry in mapping:
+        original = entry["original"]
+        if original in renames:
+            return headers, f'Header mapping lists "{original}" more than once.'
+        if original not in headers:
+            return headers, f'Header mapping refers to "{original}", which is not a header in this file.'
+        renames[original] = (entry.get("renamed") or "").strip()
+
+    mapped = [renames.get(h) or h for h in headers]
+    renamed_idx = {i for i, h in enumerate(headers) if renames.get(h)}
+    by_base: dict[str, list[int]] = {}
+    for i, h in enumerate(mapped):
+        by_base.setdefault(_base_column_name(h, i), []).append(i)
+    for base, idxs in by_base.items():
+        if len(idxs) > 1 and renamed_idx.intersection(idxs):
+            clash = next(i for i in idxs if i in renamed_idx)
+            return headers, (f'Renaming a column to "{mapped[clash]}" would give two columns the '
+                             f'name "{base}". Choose a different name.')
+    return mapped, None
+
+
 def extra_cells(raw: list[str], width: int) -> list[str]:
     """Non-empty cells beyond the header width (#82).
 

api/services/destructive_ops.py

@@ -0,0 +1,57 @@
+"""Centralised authorisation for destructive database operations (#175, #170).
+
+Every code path that drops a user-data table must go through one of these
+functions.  This prevents the auth bypass where upload-with-overwrite or
+T&E mode-switch silently dropped tables without checking delete permission.
+"""
+
+from __future__ import annotations
+
+from fastapi import HTTPException
+from psycopg2 import sql
+
+from api.auth import PERMISSIONS
+
+
+def authorized_drop_table(cur, session: dict | None, schema: str, table_name: str) -> None:
+    """Drop a table only if the caller holds delete_file permission.
+
+    ``session`` is ``None`` for API-key (non-session) requests, which are
+    allowed through (they have full access until per-key roles exist).
+    Raises HTTP 403 if the session's role lacks ``delete_file``.
+    """
+    _require_delete(session)
+    cur.execute(
+        sql.SQL("DROP TABLE IF EXISTS {}.{}").format(
+            sql.Identifier(schema), sql.Identifier(table_name)
+        )
+    )
+
+
+def safe_cleanup_prior(cur, session: dict | None, schema: str,
+                       table_name: str, mode: str) -> None:
+    """Drop a prior dynamic-mode table during overwrite, with auth check.
+
+    Only drops when the prior entry is dynamic mode and the table name starts
+    with ``csv_`` (T&E-mode entries point at shared tables that must not be
+    dropped here).
+    """
+    _require_delete(session)
+    if mode == "dynamic" and table_name.startswith("csv_"):
+        cur.execute(
+            sql.SQL("DROP TABLE IF EXISTS {}.{}").format(
+                sql.Identifier(schema), sql.Identifier(table_name)
+            )
+        )
+
+
+def _require_delete(session: dict | None) -> None:
+    """Raise 403 if the session's role cannot delete files."""
+    if session is None:
+        return
+    roles = PERMISSIONS["delete_file"]
+    if session["role"] not in roles:
+        raise HTTPException(
+            status_code=403,
+            detail=f"Role '{session['role']}' may not delete or overwrite files",
+        )

api/services/dynamic_loader.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import hashlib
+import json
 import logging
 import time
 
@@ -20,8 +21,10 @@
 from api.config import settings
 from api.tenancy import uploads_schema
 from api.db import Conn
+from api.services.destructive_ops import safe_cleanup_prior
 from api.services.csv_parse import (
     ALLOWED_TYPES,
+    apply_header_mapping,
     cast_value,
     extra_cells,
     parse_csv,
@@ -54,6 +57,9 @@ def upload_dynamic(
     content: str,
     types: list[str] | None,
     overwrite: bool,
+    *,
+    session: dict | None = None,
+    header_mapping: list[dict] | None = None,
 ) -> dict:
     logs: list[dict] = []
     _log(logs, "receive", f'Received "{file_name}" ({len(content)} chars)')
@@ -85,6 +91,17 @@ def upload_dynamic(
             "logs": logs,
         }
 
+    # #186: honour confirmed header renames; the column name in the table is
+    # the renamed one, not the CSV's original.
+    mapped_headers, mapping_error = apply_header_mapping(rows[0], header_mapping)
+    if mapping_error:
+        _log(logs, "header_mapping", mapping_error, "error")
+        return {"status": "error", "reason": "invalid_header_mapping", "message": mapping_error, "logs": logs}
+    if header_mapping:
+        renamed_count = sum(1 for a, b in zip(rows[0], mapped_headers) if a != b)
+        _log(logs, "header_mapping", f"Renamed {renamed_count} column header(s)", count=renamed_count)
+        rows = [mapped_headers] + rows[1:]
+
     # BUG-027 guard: reject files above the configured row cap before doing any
     # per-row work. `rows` includes the header, so subtract one.
     data_row_count = len(rows) - 1
@@ -107,7 +124,7 @@ def upload_dynamic(
     schema = uploads_schema()
 
     try:
-        return _do_upload(file_name, types, overwrite, logs, schema, rows, file_hash)
+        return _do_upload(file_name, types, overwrite, logs, schema, rows, file_hash, session=session)
     except psycopg2.errors.UniqueViolation:
         # #81: an identical concurrent upload won a race that _do_upload could
         # not settle itself (e.g. both creating csv_<hash> at once). Nothing
@@ -142,6 +159,8 @@ def _do_upload(
     schema: str,
     rows: list[list[str]],
     file_hash: str,
+    *,
+    session: dict | None = None,
 ) -> dict:
     replaced_file_name: str | None = None
     with Conn() as conn:
@@ -168,16 +187,7 @@ def _do_upload(
                         "logs": logs,
                     }
                 _log(logs, "overwrite", f'Overwriting previous upload "{name_match[1]}"', "warn")
-                # table_name is only a bare identifier owned by this schema for
-                # dynamic-mode rows (csv_<hash>); TE-mode rows store "schema.table"
-                # pointing at a shared T&E table that must never be dropped here.
-                # Same guard as te_loader._cleanup_prior_registry_entries.
-                if name_match[5] == "dynamic" and name_match[2].startswith("csv_"):
-                    cur.execute(
-                        sql.SQL("DROP TABLE IF EXISTS {}.{}").format(
-                            sql.Identifier(schema), sql.Identifier(name_match[2])
-                        )
-                    )
+                safe_cleanup_prior(cur, session, schema, name_match[2], name_match[5])
                 cur.execute(
                     sql.SQL("DELETE FROM {}.csv_files WHERE id = %s").format(sql.Identifier(schema)),
                     (name_match[0],),
@@ -206,13 +216,7 @@ def _do_upload(
                         "logs": logs,
                     }
                 _log(logs, "overwrite", f'Overwriting previous content match "{content_match[1]}"', "warn")
-                # Same TE-vs-dynamic guard as above.
-                if content_match[4] == "dynamic" and content_match[2].startswith("csv_"):
-                    cur.execute(
-                        sql.SQL("DROP TABLE IF EXISTS {}.{}").format(
-                            sql.Identifier(schema), sql.Identifier(content_match[2])
-                        )
-                    )
+                safe_cleanup_prior(cur, session, schema, content_match[2], content_match[4])
                 cur.execute(
                     sql.SQL("DELETE FROM {}.csv_files WHERE id = %s").format(sql.Identifier(schema)),
                     (content_match[0],),
@@ -301,10 +305,11 @@ def _do_upload(
                     values.append(val)
                 if failed:
                     continue
-                # BUG-022: ASCII unit separator (U+001F) between cells prevents
-                # ["ab","cd"] and ["a","bcd"] from hashing to the same value.
-                row_hash = hashlib.sha256(
-                    "\x1f".join(raw_joined).encode("utf-8")
+                # #178: JSON serialization prevents ambiguous concatenation.
+                # U+001F join was vulnerable: cells containing U+001F caused
+                # distinct rows to hash identically (silent data loss).
+                row_hash = "v2:" + hashlib.sha256(
+                    json.dumps(raw_joined, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
                 ).hexdigest()
                 if row_hash in seen:
                     duplicates += 1