feat(db): serve requests as a runtime role that can't change the audit or upload records (M3d, #107)

api/config.py

@@ -9,6 +9,11 @@ class Settings:
     PG_USER: str = os.environ.get("PGUSER", "postgres")
     PG_PASSWORD: str = os.environ.get("PGPASSWORD", "")
     PG_DATABASE: str = os.environ.get("PGDATABASE", "te_mgmt_dev")
+    # M3d (#107): the runtime role the API's request pool connects as. It may only
+    # read and insert in the append-only tables. PG_USER stays the owner, used at
+    # startup for migrations and grants. Unset: the pool connects as PG_USER.
+    APP_DB_USER: str = os.environ.get("APP_DB_USER", "")
+    APP_DB_PASSWORD: str = os.environ.get("APP_DB_PASSWORD", "")
 
     # Schema where dynamically created per-CSV tables + the registry live.
     UPLOADS_SCHEMA: str = os.environ.get("CSV_UPLOADS_SCHEMA", "csv_uploads")

api/db.py

@@ -2,6 +2,7 @@
 
 import logging
 import time
+from contextlib import contextmanager
 from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeout
 from pathlib import Path
 
@@ -39,8 +40,8 @@ def init_pool(max_attempts: int = 30, base_delay: float = 1.0) -> None:
                 maxconn=_MAXCONN,
                 host=settings.PG_HOST,
                 port=settings.PG_PORT,
-                user=settings.PG_USER,
-                password=settings.PG_PASSWORD,
+                user=settings.APP_DB_USER or settings.PG_USER,
+                password=settings.APP_DB_PASSWORD if settings.APP_DB_USER else settings.PG_PASSWORD,
                 dbname=settings.PG_DATABASE,
             )
             if attempt > 1:
@@ -176,8 +177,67 @@ def bootstrap() -> None:
                 settings.PG_USER, settings.PG_HOST, settings.PG_PORT,
                 settings.PG_DATABASE)
     command.upgrade(cfg, "head")
+    with owner_connect() as conn, conn.cursor() as cur:
+        for schema, create in ((settings.UPLOADS_SCHEMA, True), (settings.CONTROL_SCHEMA, False),
+                               (settings.TE_SCHEMA, False)):
+            grant_runtime(cur, schema, create=create)
     _migrate_existing_tenants()
     logger.info("Alembic migrations complete")
+    if not settings.APP_DB_USER:
+        logger.warning("APP_DB_USER is not set: the API connects as the owner %r, so the database "
+                       "doesn't stop it changing the audit log (M3d). See docs/QUICKSTART.md.",
+                       settings.PG_USER)
+
+
+@contextmanager
+def owner_connect():
+    """A connection as the owner (PG_USER): for migrations, provisioning and grants, never requests.
+    Closed on exit; the caller commits."""
+    conn = psycopg2.connect(host=settings.PG_HOST, port=settings.PG_PORT, user=settings.PG_USER,
+                            password=settings.PG_PASSWORD, dbname=settings.PG_DATABASE)
+    try:
+        yield conn
+    finally:
+        conn.close()
+
+
+# Tables the runtime role may only read and insert (M3, #107). Triggers also stop the owner.
+APPEND_ONLY = ("audit_log", "uploads", "upload_blobs", "source_rows", "alembic_version")
+
+
+def grant_runtime(cur, schema: str, create: bool) -> None:
+    """Give the runtime role (APP_DB_USER) what requests need in one schema (M3d, #107).
+
+    Read and write on ordinary tables; read and insert only on APPEND_ONLY.
+    ``create`` lets it create, and so own and drop, the per-upload csv_* tables;
+    ones the owner created before M3d are handed over. Commits. No-op without
+    APP_DB_USER or if the schema doesn't exist.
+    """
+    role = settings.APP_DB_USER
+    if not role:
+        return
+    cur.execute("SELECT 1 FROM pg_namespace WHERE nspname = %s", (schema,))
+    if cur.fetchone() is None:
+        return
+    s, r = sql.Identifier(schema), sql.Identifier(role)
+    cur.execute(sql.SQL("GRANT USAGE{} ON SCHEMA {} TO {}").format(
+        sql.SQL(", CREATE") if create else sql.SQL(""), s, r))
+    cur.execute(sql.SQL("GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA {} TO {}").format(s, r))
+    cur.execute(sql.SQL("GRANT USAGE, SELECT, UPDATE ON ALL SEQUENCES IN SCHEMA {} TO {}").format(s, r))
+    cur.execute("SELECT relname FROM pg_class WHERE relnamespace = %s::regnamespace AND relname = ANY(%s)",
+                (schema, list(APPEND_ONLY)))
+    for (table,) in cur.fetchall():
+        cur.execute(sql.SQL("REVOKE UPDATE, DELETE, TRUNCATE ON {}.{} FROM {}").format(
+            s, sql.Identifier(table), r))
+    cur.execute("SELECT to_regclass(%s)", (f'"{schema}".csv_files',))
+    if create and cur.fetchone()[0] is not None:
+        cur.execute(sql.SQL(
+            "SELECT c.relname FROM {}.csv_files f JOIN pg_class c ON c.relname = f.table_name "
+            "AND c.relnamespace = %s::regnamespace WHERE f.mode = 'dynamic' "
+            "AND pg_get_userbyid(c.relowner) <> %s").format(s), (schema, role))
+        for (table,) in cur.fetchall():
+            cur.execute(sql.SQL("ALTER TABLE {}.{} OWNER TO {}").format(s, sql.Identifier(table), r))
+    cur.connection.commit()
 
 
 def alembic_ini_path() -> Path:
@@ -188,7 +248,7 @@ def _migrate_existing_tenants() -> None:
     """Bring every tenant schema to head too (decision 6: migrations run across N schemas)."""
     from api.tenancy import migrate_tenant
 
-    with Conn() as conn, conn.cursor() as cur:
+    with owner_connect() as conn, conn.cursor() as cur:
         cur.execute(sql.SQL("SELECT slug FROM {}.tenants ORDER BY id").format(
             sql.Identifier(settings.CONTROL_SCHEMA)))
         slugs = [r[0] for r in cur.fetchall()]

api/tenancy.py

@@ -66,12 +66,13 @@ def migrate_tenant(slug: str) -> None:
             command.upgrade(cfg, "head")
         finally:
             settings.UPLOADS_SCHEMA = original
-    from api.db import Conn
+    from api.db import grant_runtime, owner_connect
 
-    with Conn() as conn:
+    with owner_connect() as conn:
         with conn.cursor() as cur:
             provision_te_tables(cur, schema_for(slug), settings.TE_SCHEMA)
-        conn.commit()
+            conn.commit()
+            grant_runtime(cur, schema_for(slug), create=True)
 
 
 def provision_te_tables(cur, schema: str, template: str) -> int:

docs/QUICKSTART.md

@@ -118,6 +118,25 @@ It prompts for a password of at least 12 characters, or reads it from
 `TE_ADMIN_PASSWORD`. Sign in with tenant `dev`, username `admin` and that
 password.
 
+### Set up the runtime database role (once, M3d)
+
+The API answers requests as a separate database role, `te_app`. It can only read
+and add to the audit log and the stored uploads, never change or delete them
+(#107). The owner (`PGUSER`, usually `postgres`) is used only at startup. Stop the
+API first, then run this once from the same folder:
+
+```powershell
+cd "C:\Users\User\OneDrive\Desktop\Migration using ai\PostgreDataMigrationAppWithCSVLoader"
+python scripts\create_app_role.py
+```
+
+It connects as the owner with the `PG*` values from step 3, so run it in a
+window where they're set (`PGPORT` 5433 and `PGPASSWORD` for PostgreSQL 18).
+It creates the role with a random password and writes the role's settings to
+`%USERPROFILE%\.te-migration\app-db.env`. It never prints the password.
+`start-api.ps1` loads that file. Until you run it, the API prints
+`APP_DB_USER not set` and runs as the owner. Running it again sets a new password.
+
 ### Change or reset a password (#117)
 
 - **Your own password:** sign in, then click the key icon in the header.

docs/VCRM.md

@@ -143,7 +143,7 @@ Status as assessed on 2026-10-26 at `main` 032b522. None of these is "Verified"
 | ID | Status | Reason | Planned verification | Milestone |
 |----|--------|--------|----------------------|-----------|
 | BR-23 | ✅ | Every upload attempt gets an `uploads` row (M3b, #107): id, original filename, SHA-256, uploader, timestamp and outcome (imported, partly_imported, failed, duplicate, rejected). The tenant is the schema the row lives in (decision 6), not a column. Registry UNIQUE hash and name from M1 (#87). `test_preserved_uploads.py` | Integration + DB constraint tests | M1, M3 |
-| BR-24 | ⚠️ Partial | Original bytes (`upload_blobs`) and parsed rows (`source_rows`) kept for every upload, including failed ones and files later overwritten or deleted (M3b, #107). Triggers reject UPDATE, DELETE and TRUNCATE (`test_preserved_uploads.py`, 6 tests). Still to do: the least-privilege runtime role (M3d); edits kept separate arrive with M4 | DB negative tests | M3 |
+| BR-24 | ⚠️ Partial | Original bytes (`upload_blobs`) and parsed rows (`source_rows`) kept for every upload, including failed ones and files later overwritten or deleted (M3b, #107). Triggers reject UPDATE, DELETE and TRUNCATE (`test_preserved_uploads.py`, 7 tests), and the API's runtime role has no UPDATE, DELETE or TRUNCATE grant on them (M3d, `test_runtime_role.py`). Still to do: show edits kept separate once they exist (M4) | DB negative tests | M3 |
 | BR-25 | ⚠️ Partial | 10-row preview sample; 200-row cap on table view | API paging + e2e | M4 |
 | BR-26 | ❌ Not started | No code | API + e2e | M4 |
 | BR-27 | ❌ Not started | No code | API + DB tests | M4 |
@@ -154,7 +154,7 @@ Status as assessed on 2026-10-26 at `main` 032b522. None of these is "Verified"
 | BR-32 | ❌ Not started | No code | API + e2e | M6 |
 | BR-33 | ⚠️ Partial | Single-transaction load; no staging or lineage | Integration incl. failure injection | M6 |
 | BR-34 | ⚠️ Partial | Total, inserted, duplicate and failed counts persisted per import (M1: #90, `tests/test_import_counts.py`); an upload where every row fails is reported as failed and saves nothing, and partial failures are shown on the files list (#123, `tests/test_failed_uploads.py`); over-wide rows reported, not dropped (M1: #88, `tests/test_ragged_rows.py`); excluded and edited rows not yet tracked (M6) | Integration | M1, M6 |
-| BR-35 | ⚠️ Partial | Export formula injection neutralised (M1: #86, `export.test.ts`). Audit log append-only (M3a, #107): triggers reject UPDATE, DELETE and TRUNCATE for every role; actor recorded; hash chain checked by `GET /api/audit/verify` (`test_audit_append_only.py`, 7 tests). Still to do: the least-privilege runtime role (M3d) and report filters (M7) | DB negative tests + unit | M1, M3, M7 |
+| BR-35 | ⚠️ Partial | Export formula injection neutralised (M1: #86, `export.test.ts`). Audit log append-only (M3a, #107): triggers reject UPDATE, DELETE and TRUNCATE for every role; actor recorded; hash chain checked by `GET /api/audit/verify` (`test_audit_append_only.py`, 7 tests). The API's runtime role can only read and insert it, and can't disable its triggers (M3d, `test_runtime_role.py`). Still to do: report filters, including stored failure reasons (M7) | DB negative tests + unit | M1, M3, M7 |
 | BR-36 | ⚠️ Partial | Done in M2:<br>• local accounts with lockout (#98, #103)<br>• OIDC (#105)<br>• per-tenant schemas, T&E tables included (#100, #104)<br>• role matrix in `docs/ROLES.md` (#101)<br><br>Tests: `tests/test_accounts.py`, `test_tenant_isolation.py`, `test_roles.py`, `test_oidc.py`, `test_password_reset.py`. Each failed with its control removed.<br>• frontend sign-in; the browser no longer holds the shared key (#115)<br>• password change, admin reset and break-glass reset (#117; `tests/test_password_reset.py`)<br><br>Open: the shared key still grants non-tenant access to scripts and services; security review (#112); real-IdP OIDC check (#113) | Cross-tenant negative tests; role matrix tests | M2 |
 | BR-37 | ⚠️ Partial | No external calls found; no automated check; no offline doc (#60) | Egress-blocked run + offline dry run | M7 |
 | BR-38 | ❌ Not started | No container bundle for the application yet (`infra/Dockerfile` is the schema and evals runner only) | Offline smoke test 7/7 on Windows, macOS and Linux | #128 |

docs/offline-install.md

@@ -66,8 +66,8 @@ The script runs six steps:
 1. Checks every file against `SHA256SUMS`.
 2. Installs the Ubuntu packages, using the bundle as the only apt source. The VM's own apt sources are ignored.
 3. Copies the app to `/opt/te-migration`, creates the `te-app` system user, and builds a Python virtual environment from `wheels/` without network access.
-4. Deploys database `te_mgmt_prod` with schema `te_prod` and its login role `te_prod_user`, using a generated password.
-5. Writes `/opt/te-migration/env` (mode 600) with that password and a generated API key, and starts the `te-api` (port 8000) and `te-frontend` (port 3000) services under systemd. The API key is for scripts only. The frontend holds no key; people sign in.
+4. Deploys database `te_mgmt_prod` with schema `te_prod` and its owner role `te_prod_user`. It also creates the runtime role `te_prod_runtime`, which serves requests and can only read and insert the audit and upload tables (M3d, #107). Both get generated passwords.
+5. Writes `/opt/te-migration/env` (mode 600) with both passwords and a generated API key, and starts the `te-api` (port 8000) and `te-frontend` (port 3000) services under systemd. The API key is for scripts only. The frontend holds no key; people sign in.
 6. Creates the first tenant and its tenant admin, then runs the smoke test.
 
 During step 2, apt may print `W: Download is performed unsandboxed as root`. This only means apt's sandbox user can't read the bundle folder, and it's harmless.

scripts/create_app_role.py

@@ -0,0 +1,71 @@
+"""Create or update the API's runtime database role (M3d, #107).
+
+Run once from the repo root, with the owner's PG* environment set (the same
+settings start-api.ps1 uses):
+    python scripts/create_app_role.py
+
+What it does:
+- creates the login role (default te_app) with a new random password, or gives
+  an existing one a new password;
+- lets the owner hand tables to it (GRANT te_app TO <owner>);
+- brings every schema to the current migration and applies the grants: read and
+  write on ordinary tables, read and insert only on the audit and upload tables;
+- writes APP_DB_USER and APP_DB_PASSWORD to an env file only you can read
+  (default ~/.te-migration/app-db.env), which start-api.ps1 loads.
+
+The password is never printed. Set TE_APP_DB_PASSWORD to choose it yourself.
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import secrets
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from psycopg2 import sql  # noqa: E402
+
+from api import db  # noqa: E402
+from api.config import settings  # noqa: E402
+
+DEFAULT_ENV_FILE = Path.home() / ".te-migration" / "app-db.env"
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument("--role", default="te_app")
+    ap.add_argument("--env-file", type=Path, default=DEFAULT_ENV_FILE)
+    args = ap.parse_args()
+    if args.role == settings.PG_USER:
+        print("The runtime role must differ from the owner (PGUSER).", file=sys.stderr)
+        return 2
+    password = os.environ.get("TE_APP_DB_PASSWORD") or secrets.token_urlsafe(32)
+
+    with db.owner_connect() as conn, conn.cursor() as cur:
+        role, pw = sql.Identifier(args.role), sql.Literal(password)
+        cur.execute("SELECT 1 FROM pg_roles WHERE rolname = %s", (args.role,))
+        verb = "ALTER" if cur.fetchone() else "CREATE"
+        cur.execute(sql.SQL(verb + " ROLE {} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE PASSWORD {}")
+                    .format(role, pw))
+        cur.execute(sql.SQL("GRANT {} TO {}").format(role, sql.Identifier(settings.PG_USER)))
+        cur.execute(sql.SQL("GRANT CONNECT ON DATABASE {} TO {}").format(
+            sql.Identifier(settings.PG_DATABASE), role))
+        conn.commit()
+
+    settings.APP_DB_USER, settings.APP_DB_PASSWORD = args.role, password
+    db.bootstrap()   # migrations, then grants on every schema, tenants included
+
+    args.env_file.parent.mkdir(parents=True, exist_ok=True)
+    fd = os.open(args.env_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+    with os.fdopen(fd, "w", encoding="utf-8") as f:
+        f.write(f"APP_DB_USER={args.role}\nAPP_DB_PASSWORD={password}\n")
+    print(f"Role '{args.role}' is ready ({'created' if verb == 'CREATE' else 'new password set'}). Its settings are in {args.env_file}; "
+          "start-api.ps1 loads them. Restart the API.")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/offline/install.sh

@@ -43,6 +43,14 @@ sed "s/__INJECT_AT_DEPLOY__/$DB_PASS/" "$DEST/app/build/environments/env_prod.ex
   > "$DEST/app/build/environments/env_prod.sql"
 ( cd "$DEST/app" && sudo -u postgres env PGHOST=/var/run/postgresql bash build/deploy_all.sh prod >/dev/null )
 rm "$DEST/app/build/environments/env_prod.sql"   # it holds the password
+# M3d (#107): a separate runtime role for requests; it can only read and insert the
+# audit and upload tables. te_prod_user stays the owner and runs migrations at startup.
+APP_DB_PASS="$(openssl rand -hex 24)"
+sudo -u postgres psql -q -v ON_ERROR_STOP=1 -d te_mgmt_prod -c "SET client_min_messages = warning" \
+  -c "DO \$\$BEGIN IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'te_prod_runtime') THEN CREATE ROLE te_prod_runtime LOGIN; END IF; END\$\$" \
+  -c "ALTER ROLE te_prod_runtime LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE PASSWORD '$APP_DB_PASS'" \
+  -c "GRANT te_prod_runtime TO te_prod_user" \
+  -c "GRANT CONNECT ON DATABASE te_mgmt_prod TO te_prod_runtime" >/dev/null
 
 echo "== 5/6 configuration and services"
 cat > "$DEST/env" <<ENV
@@ -53,6 +61,8 @@ PGPORT=5432
 PGUSER=te_prod_user
 PGPASSWORD=$DB_PASS
 PGDATABASE=te_mgmt_prod
+APP_DB_USER=te_prod_runtime
+APP_DB_PASSWORD=$APP_DB_PASS
 TE_SCHEMA=te_prod
 CORS_ORIGINS=http://$API_HOST:3000
 PORT=3000

scripts/start-api.ps1

@@ -20,6 +20,19 @@ if (-not $env:PGPASSWORD) {
         [Runtime.InteropServices.Marshal]::SecureStringToBSTR($sec))
 }
 
+# M3d (#107): requests use the runtime role made by scripts\create_app_role.py,
+# which can't change the audit log. The owner above is only used at startup.
+$appDbEnv = Join-Path $env:USERPROFILE ".te-migration\app-db.env"
+if ((-not $env:APP_DB_USER) -and (Test-Path $appDbEnv)) {
+    foreach ($line in Get-Content $appDbEnv) {
+        $k, $v = $line -split "=", 2
+        if ($k) { Set-Item -Path "env:$k" -Value $v }
+    }
+}
+if (-not $env:APP_DB_USER) {
+    Write-Host "APP_DB_USER not set - requests run as the owner. Run once: python scripts\create_app_role.py" -ForegroundColor Yellow
+}
+
 if (-not $env:API_KEY) {
     Write-Host "API_KEY not set - every endpoint is unauthenticated (fine for local dev)." -ForegroundColor Yellow
 }