diff --git a/.superpowers/sdd/task-2-report.md b/.superpowers/sdd/task-2-report.md index f9061d8c..62dd7afa 100644 --- a/.superpowers/sdd/task-2-report.md +++ b/.superpowers/sdd/task-2-report.md @@ -28,3 +28,124 @@ git diff --check The local-vector and preprocess service secret declarations remain for Task 3, which converts those services to the same bundle helper. Documentation and smoke command migration is reserved for Task 4. + +--- + +# Task 2 report — PostgreSQL session repository + +## Scope delivered + +- Added `PostgresSessionRepository`, implementing the Task 1 repository contract with a + direct PostgreSQL SQLAlchemy connection, transaction-local RLS context, UUIDv4 validation, + current artifacts (including `cte_sql:`), append-only decisions, preferences, and + content-free deletion tombstones. +- Added `tht session migrate --database-url URL [--status] --json` and a checksum-protected, + advisory-transaction-locked migration runner. +- Added server session configuration selection. `session_storage.connection` uses direct + PostgreSQL TLS modes `verify-ca` or `verify-full`; it does not use PostgREST. +- Updated packaging and `.gitignore` so session migrations are present in the built wheel. +- Did not alter Task 3 workflow commands, Pi gate code, or backend code. + +## TDD evidence + +### RED + +Command: + +```sh +cd harness && .venv/bin/pytest tests/test_postgres_session_repository.py tests/test_session_migrate_cmd.py -q +``` + +Result before production implementation: `1 failed, 4 errors in 3.89s`. + +- Four setup errors were `ModuleNotFoundError: No module named + 'tht.session.postgres_repository'`. +- The migration CLI test failed because `tht session migrate` did not exist (`No such command + 'migrate'`). + +### GREEN + +Initial focused suite after implementation: `5 passed in 4.18s`. + +Final focused verification: + +```sh +cd harness && .venv/bin/pytest \ + tests/test_session_repository.py \ + tests/test_postgres_session_repository.py \ + tests/test_session_migrate_cmd.py \ + tests/test_vector_migration_packaging.py -q +``` + +Result: `12 passed in 5.80s`. + +Changed-file lint verification: + +```sh +cd harness && .venv/bin/ruff check \ + tht/session/postgres_repository.py tht/migrations/sessions tht/config.py \ + tht/session/repository.py tht/cli/session_cmd.py \ + tests/test_postgres_session_repository.py tests/test_session_migrate_cmd.py \ + tests/test_vector_migration_packaging.py +``` + +Result: `All checks passed!`. + +## Migration and role policy choices + +`001_schema.sql` creates only private `thoth_sessions` tables: + +- `principals` and `principal_preferences`; +- `sessions`, with `session_artifacts` and `review_decisions` cascading on session deletion; +- `audit_log`, which deliberately has no content/detail/metadata column and keeps only action, + session UUID, actor identity, owner identity, and timestamp. + +`002_security.sql` creates separate `thoth_sessions_runtime` and +`thoth_sessions_migrator` group roles, explicitly `NOLOGIN NOBYPASSRLS NOSUPERUSER`, revokes +public access, gives the runtime role only the operations required by the adapter, and enables +and forces RLS on every table. Owner/admin policies read only transaction-local settings: +`thoth_sessions.actor_issuer`, `thoth_sessions.actor_subject`, and +`thoth_sessions.is_admin`. The adapter starts every operation in a transaction, switches to the +restricted runtime role, sets those settings with `set_config(..., true)`, and uses advisory +transaction locks for migrations and per-session mutations. + +The runtime role remains a `NOLOGIN` group role by design. Deployment must provision a dedicated +non-superuser LOGIN role and grant it membership, for example: + +```sql +CREATE ROLE thoth_sessions_app LOGIN NOINHERIT PASSWORD ''; +GRANT thoth_sessions_runtime TO thoth_sessions_app; +``` + +This avoids embedding an environment-specific login name or credential in versioned SQL. The +new integration test proves that this non-superuser membership path can create and read a +session while the adapter executes as `thoth_sessions_runtime`. + +## Security/self-review + +- Owner isolation and admin cross-owner reads run against disposable PostgreSQL containers, + not Supabase. +- No table or column includes `embedding`; repository code imports no embedding/vector code; + the regression test writes a session artifact under a monkeypatched embedding sentinel. +- An unauthorized owner receives the same `SessionError` as an absent session, preserving the + future backend's 404 mapping boundary. +- The audit row is inserted before deleting the parent session, so cascades remove all artifact + and decision content while the tombstone survives. +- A security review found and this task fixed the initial `.gitignore` rule that would have + excluded `migrations/sessions/*.sql` from Git/wheels. The wheel test now asserts both session + migration files and checks both the existing vector CLI and the new session CLI. +- The review also highlighted runtime login provisioning. It is covered by a non-superuser + regression test and documented above; concrete credential/role deployment belongs to Task 7. + +## Remaining concerns + +- Full `harness/.venv/bin/pytest -q` could not complete in this execution environment: the + runner terminated the command after roughly 30 seconds. Captured output reached 44% with no + failures before termination; `pgrep` confirmed no pytest process remained. The Task 2 focused + suites above completed successfully. +- `harness/.venv/bin/ruff check .` currently reports 34 pre-existing violations in unrelated + test files (for example unused imports in `tests/l0/test_db_connection.py` and semicolon style + in `tests/test_phase_effective.py`). The changed-file Ruff command is clean. +- Task 7 must safely provision the dedicated runtime login/membership and inject its TLS + credentials/CA; this task intentionally does not create a deployment-specific LOGIN role or + password. diff --git a/harness/.gitignore b/harness/.gitignore index af38b5f0..cdd01399 100644 --- a/harness/.gitignore +++ b/harness/.gitignore @@ -3,6 +3,9 @@ __pycache__/ .env .venv/ sessions/ +# Keep the versioned database migration package; only runtime session directories are ignored. +!tht/migrations/sessions/ +!tht/migrations/sessions/*.sql indexes/ artifacts/ # Per-customer active workspace (points at the customer repo via -c or symlink). diff --git a/harness/pyproject.toml b/harness/pyproject.toml index 30166fdb..2ca5241e 100644 --- a/harness/pyproject.toml +++ b/harness/pyproject.toml @@ -34,7 +34,7 @@ dev = [ include = ["tht*"] [tool.setuptools.package-data] -tht = ["migrations/vector/*.sql"] +tht = ["migrations/vector/*.sql", "migrations/sessions/*.sql"] [tool.ruff] line-length = 100 diff --git a/harness/tests/test_postgres_session_repository.py b/harness/tests/test_postgres_session_repository.py new file mode 100644 index 00000000..5206f704 --- /dev/null +++ b/harness/tests/test_postgres_session_repository.py @@ -0,0 +1,173 @@ +import uuid +from datetime import UTC, datetime + +import pytest +from sqlalchemy import create_engine, text +from testcontainers.postgres import PostgresContainer + +from tht.decisions import DecisionInput +from tht.session.models import PrincipalContext, SessionManifest +from tht.session.store import SessionError + + +@pytest.fixture(scope="module") +def database_url(): + with PostgresContainer("postgres:16-alpine") as postgres: + yield postgres.get_connection_url() + + +@pytest.fixture(scope="module", autouse=True) +def migrated(database_url): + from tht.session.postgres_repository import migrate + + migrate(database_url) + + +def _manifest(session_id: str) -> SessionManifest: + return SessionManifest( + id=session_id, + created_at=datetime(2026, 7, 16, 10, 0, tzinfo=UTC), + question="Which patients had an ablation?", + database="testdb", + schema="public", + ) + + +def _repository(database_url, subject: str, *, is_admin: bool = False): + from tht.session.postgres_repository import PostgresSessionRepository + + return PostgresSessionRepository( + database_url, + PrincipalContext(issuer="portal", subject=subject, is_admin=is_admin), + ) + + +def test_owner_can_read_own_snapshot_but_not_another_owners(database_url): + alice = _repository(database_url, "alice") + bob = _repository(database_url, "bob") + session_id = str(uuid.uuid4()) + + alice.create(_manifest(session_id)) + alice.write_artifact(session_id, "cte_sql:eligible_patients", "SELECT 1") + alice.append_decisions( + session_id, [DecisionInput(type="sql_approved", subject="phase:7", detail="SELECT 1")] + ) + + snapshot = alice.get(session_id) + assert snapshot.principal == alice.principal + assert snapshot.artifacts == {"cte_sql:eligible_patients": "SELECT 1"} + assert snapshot.decisions[0].seq == 1 + with pytest.raises(SessionError, match="Sessione non trovata"): + bob.get(session_id) + + +def test_admin_can_read_another_owners_session(database_url): + owner = _repository(database_url, "owner") + admin = _repository(database_url, "admin", is_admin=True) + session_id = str(uuid.uuid4()) + owner.create(_manifest(session_id)) + + assert admin.get(session_id).manifest.id == session_id + + +def test_non_superuser_runtime_login_can_assume_the_restricted_runtime_role(database_url): + admin = create_engine(database_url) + runtime_url = admin.url.set( + username="thoth_sessions_test_login", password="runtime-test-only" + ) + try: + with admin.begin() as connection: + connection.exec_driver_sql( + "CREATE ROLE thoth_sessions_test_login LOGIN NOINHERIT PASSWORD 'runtime-test-only'" + ) + connection.exec_driver_sql("GRANT thoth_sessions_runtime TO thoth_sessions_test_login") + repository = _repository( + runtime_url.render_as_string(hide_password=False), "runtime-user" + ) + session_id = str(uuid.uuid4()) + + repository.create(_manifest(session_id)) + + assert repository.get(session_id).manifest.id == session_id + repository.close() + finally: + with admin.begin() as connection: + connection.exec_driver_sql("DROP ROLE IF EXISTS thoth_sessions_test_login") + admin.dispose() + + +def test_delete_cascades_content_and_leaves_content_free_tombstone(database_url): + repository = _repository(database_url, "alice") + session_id = str(uuid.uuid4()) + repository.create(_manifest(session_id)) + repository.write_artifact(session_id, "sql_final", "SELECT confidential_value") + repository.append_decisions( + session_id, + [DecisionInput(type="sql_approved", subject="phase:7", detail="SELECT confidential_value")], + ) + + repository.delete(session_id) + + with pytest.raises(SessionError, match="Sessione non trovata"): + repository.get(session_id) + engine = create_engine(database_url) + try: + with engine.connect() as connection: + assert connection.execute( + text("SELECT count(*) FROM thoth_sessions.session_artifacts WHERE session_id = :session_id"), + {"session_id": session_id}, + ).scalar_one() == 0 + assert connection.execute( + text("SELECT count(*) FROM thoth_sessions.review_decisions WHERE session_id = :session_id"), + {"session_id": session_id}, + ).scalar_one() == 0 + columns = connection.execute( + text( + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema = 'thoth_sessions' AND table_name = 'audit_log'" + ) + ).scalars().all() + tombstone = connection.execute( + text( + "SELECT action, session_id, actor_issuer, actor_subject " + "FROM thoth_sessions.audit_log WHERE session_id = :session_id" + ), + {"session_id": session_id}, + ).one() + finally: + engine.dispose() + assert (tombstone[0], str(tombstone[1]), *tombstone[2:]) == ( + "session_deleted", + session_id, + "portal", + "alice", + ) + assert not {"content", "artifact_content", "detail", "metadata"} & set(columns) + + +def test_session_schema_does_not_create_or_invoke_embeddings(database_url, monkeypatch): + import tht.session.postgres_repository as repository_module + + monkeypatch.setattr( + repository_module, + "_embed", + lambda *_: pytest.fail("session persistence must not invoke embeddings"), + raising=False, + ) + repository = _repository(database_url, "alice") + session_id = str(uuid.uuid4()) + repository.create(_manifest(session_id)) + repository.write_artifact(session_id, "evidence", '{"sources": []}') + + engine = create_engine(database_url) + try: + with engine.connect() as connection: + columns = connection.execute( + text( + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema = 'thoth_sessions'" + ) + ).scalars().all() + finally: + engine.dispose() + assert all("embedding" not in column for column in columns) diff --git a/harness/tests/test_session_migrate_cmd.py b/harness/tests/test_session_migrate_cmd.py new file mode 100644 index 00000000..e784a380 --- /dev/null +++ b/harness/tests/test_session_migrate_cmd.py @@ -0,0 +1,28 @@ +import json + +from testcontainers.postgres import PostgresContainer +from typer.testing import CliRunner + +from tht.cli import app + + +def test_session_migrate_status_is_pristine_and_idempotent(): + with PostgresContainer("postgres:16-alpine") as postgres: + database_url = postgres.get_connection_url() + runner = CliRunner() + + before = runner.invoke( + app, ["session", "migrate", "--database-url", database_url, "--status", "--json"] + ) + assert before.exit_code == 0, before.output + assert json.loads(before.stdout) == {"applied": [], "drifted": [], "pending": ["001", "002"]} + assert before.stderr == "" + + first = runner.invoke(app, ["session", "migrate", "--database-url", database_url, "--json"]) + second = runner.invoke(app, ["session", "migrate", "--database-url", database_url, "--json"]) + + expected = {"applied": ["001", "002"], "drifted": [], "pending": []} + assert first.exit_code == 0, first.output + assert second.exit_code == 0, second.output + assert json.loads(first.stdout) == expected + assert json.loads(second.stdout) == expected diff --git a/harness/tests/test_vector_migration_packaging.py b/harness/tests/test_vector_migration_packaging.py index 24544c97..03addb39 100644 --- a/harness/tests/test_vector_migration_packaging.py +++ b/harness/tests/test_vector_migration_packaging.py @@ -6,7 +6,7 @@ import zipfile from pathlib import Path -def test_built_wheel_installs_vector_migrations_and_discovers_cli(tmp_path): +def test_built_wheel_installs_migrations_and_discovers_cli(tmp_path): harness = Path(__file__).parents[1] wheelhouse = tmp_path / "wheelhouse" target = tmp_path / "site" @@ -33,6 +33,8 @@ def test_built_wheel_installs_vector_migrations_and_discovers_cli(tmp_path): names = set(archive.namelist()) assert "tht/migrations/vector/001_extensions.sql" in names assert "tht/migrations/vector/003_roles.sql" in names + assert "tht/migrations/sessions/001_schema.sql" in names + assert "tht/migrations/sessions/002_security.sql" in names subprocess.run( [sys.executable, "-m", "pip", "install", "--no-deps", "--target", str(target), wheel], @@ -47,6 +49,8 @@ def test_built_wheel_installs_vector_migrations_and_discovers_cli(tmp_path): "-c", "from typer.testing import CliRunner; from tht.cli import app; " "r=CliRunner().invoke(app, ['vector','migrate','--help']); " + "assert r.exit_code == 0, r.output; " + "r=CliRunner().invoke(app, ['session','migrate','--help']); " "print(r.output); raise SystemExit(r.exit_code)", ], env=env, diff --git a/harness/tht/cli/session_cmd.py b/harness/tht/cli/session_cmd.py index e582e552..86f10ff5 100644 --- a/harness/tht/cli/session_cmd.py +++ b/harness/tht/cli/session_cmd.py @@ -9,6 +9,39 @@ from tht.cli.schema_cmd import _load_config_or_exit session_app = typer.Typer(help="Sessioni (directory artefatti)") +@session_app.command("migrate") +def migrate_cmd( + database_url: str = typer.Option( + ..., "--database-url", envvar="THT_SESSIONS_ADMIN_URL", help="Admin PostgreSQL URL." + ), + status_only: bool = typer.Option(False, "--status", help="Inspect without applying."), + json_output: bool = typer.Option(False, "--json", help="Emit pristine JSON."), +) -> None: + """Apply or inspect the private PostgreSQL session schema migrations.""" + from tht.session.postgres_repository import MigrationError, migrate, migration_status + + try: + status = migration_status(database_url) if status_only else migrate(database_url) + except (MigrationError, ValueError) as exc: + if json_output: + typer.echo(json.dumps({"error": str(exc)}, sort_keys=True)) + else: + typer.echo(f"ERROR: {exc}", err=True) + raise typer.Exit(code=1) from None + payload = { + "applied": [item.version for item in status.applied], + "drifted": [item.version for item in status.drifted], + "pending": [item.version for item in status.pending], + } + if json_output: + typer.echo(json.dumps(payload, sort_keys=True)) + else: + typer.echo( + f"Applied: {len(status.applied)}; pending: {len(status.pending)}; " + f"drifted: {len(status.drifted)}" + ) + + def session_dir(cfg, session_id: str) -> Path: return cfg.paths.sessions / session_id diff --git a/harness/tht/config.py b/harness/tht/config.py index 62603506..a4743613 100644 --- a/harness/tht/config.py +++ b/harness/tht/config.py @@ -78,6 +78,22 @@ class DatabaseConfig(BaseModel): model_config = {"populate_by_name": True} +class SessionPostgresConfig(DatabaseConfig): + """TLS-verified direct session storage; its login must be granted thoth_sessions_runtime. + + Provision the dedicated LOGIN role and membership out of band so deployment credentials + never appear in the versioned migration pack. + """ + + sslmode: Literal["verify-ca", "verify-full"] = "verify-full" + sslrootcert: Path | None = None + + +class SessionStorageConfig(BaseModel): + type: Literal["postgres_direct"] + connection: SessionPostgresConfig + + class RestConfig(BaseModel): """Accesso al DWH via Supabase/PostgREST. base_url es. https://host/dwh/ .""" @@ -292,6 +308,7 @@ class Config(BaseModel): _config_source: str = PrivateAttr(default="direct") dwh: DwhResourceConfig vectors: VectorResourceConfig | None = None + session_storage: SessionStorageConfig | None = None roots: WorkspaceRoots = WorkspaceRoots() # Compatibility views retained until all call sites consume typed resources. database: DatabaseConfig diff --git a/harness/tht/migrations/sessions/001_schema.sql b/harness/tht/migrations/sessions/001_schema.sql new file mode 100644 index 00000000..684d19d4 --- /dev/null +++ b/harness/tht/migrations/sessions/001_schema.sql @@ -0,0 +1,60 @@ +CREATE SCHEMA IF NOT EXISTS thoth_sessions; +REVOKE ALL ON SCHEMA thoth_sessions FROM PUBLIC; + +CREATE TABLE IF NOT EXISTS thoth_sessions.principals ( + id bigserial PRIMARY KEY, + issuer text NOT NULL, + subject text NOT NULL, + display_name text, + created_at timestamptz NOT NULL DEFAULT pg_catalog.now(), + updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(), + UNIQUE (issuer, subject) +); + +CREATE TABLE IF NOT EXISTS thoth_sessions.principal_preferences ( + principal_id bigint PRIMARY KEY REFERENCES thoth_sessions.principals(id) ON DELETE CASCADE, + preferences jsonb NOT NULL DEFAULT '{}'::jsonb, + updated_at timestamptz NOT NULL DEFAULT pg_catalog.now() +); + +CREATE TABLE IF NOT EXISTS thoth_sessions.sessions ( + id uuid PRIMARY KEY, + principal_id bigint NOT NULL REFERENCES thoth_sessions.principals(id), + manifest jsonb NOT NULL, + created_at timestamptz NOT NULL DEFAULT pg_catalog.now(), + updated_at timestamptz NOT NULL DEFAULT pg_catalog.now() +); +CREATE INDEX IF NOT EXISTS sessions_principal_id_created_at_idx +ON thoth_sessions.sessions (principal_id, created_at DESC); + +CREATE TABLE IF NOT EXISTS thoth_sessions.session_artifacts ( + session_id uuid NOT NULL REFERENCES thoth_sessions.sessions(id) ON DELETE CASCADE, + artifact_key text NOT NULL, + content text NOT NULL, + updated_at timestamptz NOT NULL DEFAULT pg_catalog.now(), + PRIMARY KEY (session_id, artifact_key) +); + +CREATE TABLE IF NOT EXISTS thoth_sessions.review_decisions ( + session_id uuid NOT NULL REFERENCES thoth_sessions.sessions(id) ON DELETE CASCADE, + seq integer NOT NULL CHECK (seq > 0), + ts timestamptz NOT NULL, + phase integer, + type text NOT NULL, + subject text NOT NULL, + detail text NOT NULL DEFAULT '', + rationale text NOT NULL DEFAULT '', + retracts integer, + PRIMARY KEY (session_id, seq) +); + +CREATE TABLE IF NOT EXISTS thoth_sessions.audit_log ( + id bigserial PRIMARY KEY, + occurred_at timestamptz NOT NULL DEFAULT pg_catalog.now(), + action text NOT NULL, + session_id uuid NOT NULL, + actor_issuer text NOT NULL, + actor_subject text NOT NULL, + owner_issuer text NOT NULL, + owner_subject text NOT NULL +); diff --git a/harness/tht/migrations/sessions/002_security.sql b/harness/tht/migrations/sessions/002_security.sql new file mode 100644 index 00000000..8a123e60 --- /dev/null +++ b/harness/tht/migrations/sessions/002_security.sql @@ -0,0 +1,126 @@ +DO $roles$ +BEGIN + IF NOT EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = 'thoth_sessions_runtime') THEN + CREATE ROLE thoth_sessions_runtime NOLOGIN NOBYPASSRLS NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT; + END IF; + IF NOT EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = 'thoth_sessions_migrator') THEN + CREATE ROLE thoth_sessions_migrator NOLOGIN NOBYPASSRLS NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT; + END IF; +END +$roles$; + +ALTER ROLE thoth_sessions_runtime NOLOGIN NOBYPASSRLS NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT; +ALTER ROLE thoth_sessions_migrator NOLOGIN NOBYPASSRLS NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT; + +REVOKE ALL ON SCHEMA thoth_sessions FROM PUBLIC; +REVOKE ALL ON ALL TABLES IN SCHEMA thoth_sessions FROM PUBLIC; +REVOKE ALL ON ALL SEQUENCES IN SCHEMA thoth_sessions FROM PUBLIC; +REVOKE ALL ON SCHEMA thoth_sessions FROM thoth_sessions_runtime, thoth_sessions_migrator; +REVOKE ALL ON ALL TABLES IN SCHEMA thoth_sessions FROM thoth_sessions_runtime, thoth_sessions_migrator; +REVOKE ALL ON ALL SEQUENCES IN SCHEMA thoth_sessions FROM thoth_sessions_runtime, thoth_sessions_migrator; + +GRANT USAGE ON SCHEMA thoth_sessions TO thoth_sessions_runtime; +GRANT SELECT, INSERT, UPDATE ON thoth_sessions.principals TO thoth_sessions_runtime; +GRANT SELECT, INSERT, UPDATE ON thoth_sessions.principal_preferences TO thoth_sessions_runtime; +GRANT SELECT, INSERT, UPDATE, DELETE ON thoth_sessions.sessions TO thoth_sessions_runtime; +GRANT SELECT, INSERT, UPDATE ON thoth_sessions.session_artifacts TO thoth_sessions_runtime; +GRANT SELECT, INSERT ON thoth_sessions.review_decisions TO thoth_sessions_runtime; +GRANT SELECT, INSERT ON thoth_sessions.audit_log TO thoth_sessions_runtime; +GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA thoth_sessions TO thoth_sessions_runtime; + +ALTER TABLE thoth_sessions.principals ENABLE ROW LEVEL SECURITY; +ALTER TABLE thoth_sessions.principals FORCE ROW LEVEL SECURITY; +ALTER TABLE thoth_sessions.principal_preferences ENABLE ROW LEVEL SECURITY; +ALTER TABLE thoth_sessions.principal_preferences FORCE ROW LEVEL SECURITY; +ALTER TABLE thoth_sessions.sessions ENABLE ROW LEVEL SECURITY; +ALTER TABLE thoth_sessions.sessions FORCE ROW LEVEL SECURITY; +ALTER TABLE thoth_sessions.session_artifacts ENABLE ROW LEVEL SECURITY; +ALTER TABLE thoth_sessions.session_artifacts FORCE ROW LEVEL SECURITY; +ALTER TABLE thoth_sessions.review_decisions ENABLE ROW LEVEL SECURITY; +ALTER TABLE thoth_sessions.review_decisions FORCE ROW LEVEL SECURITY; +ALTER TABLE thoth_sessions.audit_log ENABLE ROW LEVEL SECURITY; +ALTER TABLE thoth_sessions.audit_log FORCE ROW LEVEL SECURITY; + +CREATE POLICY principals_owner_or_admin ON thoth_sessions.principals + FOR ALL + USING ( + pg_catalog.current_setting('thoth_sessions.is_admin', true) = 'true' + OR (issuer = pg_catalog.current_setting('thoth_sessions.actor_issuer', true) + AND subject = pg_catalog.current_setting('thoth_sessions.actor_subject', true)) + ) + WITH CHECK ( + pg_catalog.current_setting('thoth_sessions.is_admin', true) = 'true' + OR (issuer = pg_catalog.current_setting('thoth_sessions.actor_issuer', true) + AND subject = pg_catalog.current_setting('thoth_sessions.actor_subject', true)) + ); + +CREATE POLICY preferences_owner_or_admin ON thoth_sessions.principal_preferences + FOR ALL + USING ( + EXISTS ( + SELECT 1 FROM thoth_sessions.principals p + WHERE p.id = principal_preferences.principal_id + AND (pg_catalog.current_setting('thoth_sessions.is_admin', true) = 'true' + OR (p.issuer = pg_catalog.current_setting('thoth_sessions.actor_issuer', true) + AND p.subject = pg_catalog.current_setting('thoth_sessions.actor_subject', true))) + ) + ) + WITH CHECK ( + EXISTS ( + SELECT 1 FROM thoth_sessions.principals p + WHERE p.id = principal_preferences.principal_id + AND (pg_catalog.current_setting('thoth_sessions.is_admin', true) = 'true' + OR (p.issuer = pg_catalog.current_setting('thoth_sessions.actor_issuer', true) + AND p.subject = pg_catalog.current_setting('thoth_sessions.actor_subject', true))) + ) + ); + +CREATE POLICY sessions_owner_or_admin ON thoth_sessions.sessions + FOR ALL + USING ( + EXISTS ( + SELECT 1 FROM thoth_sessions.principals p + WHERE p.id = sessions.principal_id + AND (pg_catalog.current_setting('thoth_sessions.is_admin', true) = 'true' + OR (p.issuer = pg_catalog.current_setting('thoth_sessions.actor_issuer', true) + AND p.subject = pg_catalog.current_setting('thoth_sessions.actor_subject', true))) + ) + ) + WITH CHECK ( + EXISTS ( + SELECT 1 FROM thoth_sessions.principals p + WHERE p.id = sessions.principal_id + AND (pg_catalog.current_setting('thoth_sessions.is_admin', true) = 'true' + OR (p.issuer = pg_catalog.current_setting('thoth_sessions.actor_issuer', true) + AND p.subject = pg_catalog.current_setting('thoth_sessions.actor_subject', true))) + ) + ); + +CREATE POLICY artifacts_owner_or_admin ON thoth_sessions.session_artifacts + FOR ALL + USING ( + EXISTS (SELECT 1 FROM thoth_sessions.sessions s WHERE s.id = session_artifacts.session_id) + ) + WITH CHECK ( + EXISTS (SELECT 1 FROM thoth_sessions.sessions s WHERE s.id = session_artifacts.session_id) + ); + +CREATE POLICY decisions_owner_or_admin ON thoth_sessions.review_decisions + FOR ALL + USING ( + EXISTS (SELECT 1 FROM thoth_sessions.sessions s WHERE s.id = review_decisions.session_id) + ) + WITH CHECK ( + EXISTS (SELECT 1 FROM thoth_sessions.sessions s WHERE s.id = review_decisions.session_id) + ); + +CREATE POLICY audit_admin_read ON thoth_sessions.audit_log + FOR SELECT + USING (pg_catalog.current_setting('thoth_sessions.is_admin', true) = 'true'); +CREATE POLICY audit_actor_write ON thoth_sessions.audit_log + FOR INSERT + WITH CHECK ( + pg_catalog.current_setting('thoth_sessions.is_admin', true) = 'true' + OR (actor_issuer = pg_catalog.current_setting('thoth_sessions.actor_issuer', true) + AND actor_subject = pg_catalog.current_setting('thoth_sessions.actor_subject', true)) + ); diff --git a/harness/tht/session/postgres_repository.py b/harness/tht/session/postgres_repository.py new file mode 100644 index 00000000..6febba4a --- /dev/null +++ b/harness/tht/session/postgres_repository.py @@ -0,0 +1,483 @@ +"""PostgreSQL implementation of the durable, user-owned session repository.""" + +from __future__ import annotations + +import hashlib +import json +import re +import uuid +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import UTC, datetime +from importlib.resources import files +from importlib.resources.abc import Traversable +from pathlib import Path +from typing import Iterator, Sequence + +from sqlalchemy import Engine, create_engine, text +from sqlalchemy.engine import URL, make_url +from sqlalchemy.exc import SQLAlchemyError + +from tht.decisions import DecisionInput, DecisionRecord +from tht.session.models import PrincipalContext, SessionManifest, SessionSnapshot +from tht.session.store import SessionError + +MIGRATIONS_DIR = files("tht").joinpath("migrations", "sessions") +_MIGRATION_NAME = re.compile(r"^(?P\d+)_(?P[a-z0-9_]+)\.sql$") +_SAFE_CTE_NAME = re.compile(r"[A-Za-z0-9_-]+\Z") +_ARTIFACT_KEYS = { + "question", + "schema_linking", + "evidence", + "sql_final", + "validation_report", + "retrieval_pack", + "cte_tests", +} +_LOCK_KEY = 8_420_613_069_444_020_731 +_RUNTIME_ROLE = "thoth_sessions_runtime" + + +class MigrationError(RuntimeError): + """Raised when session schema migration discovery or application is unsafe.""" + + +@dataclass(frozen=True) +class Migration: + version: str + name: str + path: Traversable + checksum: str + + +@dataclass(frozen=True) +class MigrationStatus: + applied: tuple[Migration, ...] + pending: tuple[Migration, ...] + drifted: tuple[Migration, ...] + + +def _migration_source(directory: Traversable | Path | str) -> Traversable: + return Path(directory) if isinstance(directory, (str, Path)) else directory + + +def _discover(directory: Traversable | Path | str = MIGRATIONS_DIR) -> tuple[Migration, ...]: + source = _migration_source(directory) + seen_versions: set[int] = set() + parsed = [] + for path in (item for item in source.iterdir() if item.name.endswith(".sql")): + match = _MIGRATION_NAME.fullmatch(path.name) + if match is None: + raise MigrationError(f"Invalid migration filename: {path.name}") + version = match.group("version") + numeric_version = int(version) + if numeric_version in seen_versions: + raise MigrationError(f"Duplicate migration version: {numeric_version}") + seen_versions.add(numeric_version) + parsed.append((numeric_version, version, match.group("name"), path)) + if not parsed: + raise MigrationError(f"No migrations found in {source}") + return tuple( + Migration(version, name, path, hashlib.sha256(path.read_bytes()).hexdigest()) + for _, version, name, path in sorted(parsed) + ) + + +def _engine(database_url: str) -> Engine: + url = make_url(database_url) + if not url.drivername.startswith("postgresql"): + raise ValueError("Session repository requires a direct PostgreSQL URL") + return create_engine(url) + + +def _applied(connection) -> dict[str, str]: + exists = connection.execute( + text("SELECT pg_catalog.to_regclass('public.tht_session_migrations')") + ).scalar() + if exists is None: + return {} + return dict( + connection.execute(text("SELECT version, checksum FROM public.tht_session_migrations")).all() + ) + + +def _reject_unknown_versions( + migrations: tuple[Migration, ...], applied_checksums: dict[str, str] +) -> None: + local_versions = {migration.version for migration in migrations} + unknown = sorted( + set(applied_checksums) - local_versions, + key=lambda value: (0, int(value)) if value.isdigit() else (1, value), + ) + if unknown: + raise MigrationError("Database migration versions absent from local manifest: " + ", ".join(unknown)) + + +def migration_status( + database_url: str, migrations_dir: Traversable | Path | str = MIGRATIONS_DIR +) -> MigrationStatus: + migrations = _discover(migrations_dir) + engine = _engine(database_url) + try: + with engine.connect() as connection: + connection.exec_driver_sql("SET LOCAL search_path = pg_catalog, pg_temp") + applied_checksums = _applied(connection) + finally: + engine.dispose() + _reject_unknown_versions(migrations, applied_checksums) + return MigrationStatus( + applied=tuple(item for item in migrations if applied_checksums.get(item.version) == item.checksum), + pending=tuple(item for item in migrations if item.version not in applied_checksums), + drifted=tuple( + item + for item in migrations + if item.version in applied_checksums and applied_checksums[item.version] != item.checksum + ), + ) + + +def migrate( + database_url: str, migrations_dir: Traversable | Path | str = MIGRATIONS_DIR +) -> MigrationStatus: + migrations = _discover(migrations_dir) + engine = _engine(database_url) + current: Migration | None = None + try: + with engine.begin() as connection: + connection.exec_driver_sql("SET LOCAL search_path = pg_catalog, pg_temp") + connection.execute(text("SELECT pg_catalog.pg_advisory_xact_lock(:key)"), {"key": _LOCK_KEY}) + connection.exec_driver_sql( + """CREATE TABLE IF NOT EXISTS public.tht_session_migrations ( + version text PRIMARY KEY, + name text NOT NULL, + checksum text NOT NULL, + applied_at timestamptz NOT NULL DEFAULT pg_catalog.now() + )""" + ) + connection.exec_driver_sql("REVOKE ALL ON public.tht_session_migrations FROM PUBLIC") + applied_checksums = _applied(connection) + _reject_unknown_versions(migrations, applied_checksums) + drifted = [ + item + for item in migrations + if item.version in applied_checksums and applied_checksums[item.version] != item.checksum + ] + if drifted: + raise MigrationError("Migration checksum drift: " + ", ".join(item.version for item in drifted)) + for current in migrations: + if current.version in applied_checksums: + continue + connection.exec_driver_sql(current.path.read_text()) + connection.execute( + text( + "INSERT INTO public.tht_session_migrations (version, name, checksum) " + "VALUES (:version, :name, :checksum)" + ), + {"version": current.version, "name": current.name, "checksum": current.checksum}, + ) + except MigrationError: + raise + except SQLAlchemyError as exc: + filename = current.path.name if current is not None else "migration setup" + raise MigrationError(f"Failed to apply {filename}: {type(exc).__name__}") from exc + finally: + engine.dispose() + return migration_status(database_url, migrations_dir) + + +class PostgresSessionRepository: + """Session data stored in private PostgreSQL tables under transaction-local RLS context.""" + + def __init__( + self, + database_url: str, + principal: PrincipalContext, + *, + engine: Engine | None = None, + runtime_role: str = _RUNTIME_ROLE, + ): + self.principal = principal + self._engine = engine or _engine(database_url) + self._owns_engine = engine is None + self._runtime_role = runtime_role + + @classmethod + def from_config(cls, config, principal: PrincipalContext) -> "PostgresSessionRepository": + query = {"sslmode": config.sslmode} + if config.sslrootcert is not None: + query["sslrootcert"] = str(config.sslrootcert) + url = URL.create( + "postgresql+psycopg2", + username=config.user, + password=config.password, + host=config.host, + port=config.port, + database=config.database, + query=query, + ) + return cls(url.render_as_string(hide_password=False), principal) + + def close(self) -> None: + if self._owns_engine: + self._engine.dispose() + self._owns_engine = False + + def create(self, manifest: SessionManifest) -> SessionSnapshot: + self._require_uuid4(manifest.id) + with self._transaction() as connection: + principal_id = self._upsert_principal(connection) + try: + connection.execute( + text( + "INSERT INTO thoth_sessions.sessions (id, principal_id, manifest) " + "VALUES (:id, :principal_id, CAST(:manifest AS jsonb))" + ), + { + "id": manifest.id, + "principal_id": principal_id, + "manifest": json.dumps(manifest.model_dump(by_alias=True, mode="json")), + }, + ) + except SQLAlchemyError as exc: + if "unique" in str(exc).lower(): + raise SessionError(f"Sessione esiste gia': {manifest.id}") from exc + raise + return self.get(manifest.id) + + def get(self, session_id: str) -> SessionSnapshot: + self._require_uuid4(session_id) + with self._transaction() as connection: + row = connection.execute( + text("SELECT manifest FROM thoth_sessions.sessions WHERE id = :id"), {"id": session_id} + ).mappings().first() + if row is None: + raise SessionError(f"Sessione non trovata: {session_id}") + artifacts = dict( + connection.execute( + text( + "SELECT artifact_key, content FROM thoth_sessions.session_artifacts " + "WHERE session_id = :id ORDER BY artifact_key" + ), + {"id": session_id}, + ).all() + ) + decisions = [ + DecisionRecord.model_validate(dict(item)) + for item in connection.execute( + text( + "SELECT seq, ts, phase, type, subject, detail, rationale, retracts " + "FROM thoth_sessions.review_decisions WHERE session_id = :id ORDER BY seq" + ), + {"id": session_id}, + ).mappings() + ] + return SessionSnapshot( + principal=self.principal, + manifest=SessionManifest.model_validate(row["manifest"]), + artifacts=artifacts, + decisions=decisions, + ) + + def save_manifest(self, manifest: SessionManifest) -> SessionSnapshot: + self._require_uuid4(manifest.id) + with self._transaction() as connection: + result = connection.execute( + text( + "UPDATE thoth_sessions.sessions " + "SET manifest = CAST(:manifest AS jsonb), updated_at = pg_catalog.now() " + "WHERE id = :id" + ), + {"id": manifest.id, "manifest": json.dumps(manifest.model_dump(by_alias=True, mode="json"))}, + ) + if result.rowcount != 1: + raise SessionError(f"Sessione non trovata: {manifest.id}") + return self.get(manifest.id) + + def read_artifact(self, session_id: str, key: str) -> str | None: + self._require_uuid4(session_id) + self._require_artifact_key(key) + with self._transaction() as connection: + self._require_session(connection, session_id) + return connection.execute( + text( + "SELECT content FROM thoth_sessions.session_artifacts " + "WHERE session_id = :id AND artifact_key = :key" + ), + {"id": session_id, "key": key}, + ).scalar() + + def write_artifact(self, session_id: str, key: str, content: str) -> None: + self._require_uuid4(session_id) + self._require_artifact_key(key) + with self._transaction() as connection: + self._lock_session(connection, session_id) + self._require_session(connection, session_id) + connection.execute( + text( + "INSERT INTO thoth_sessions.session_artifacts (session_id, artifact_key, content) " + "VALUES (:id, :key, :content) " + "ON CONFLICT (session_id, artifact_key) DO UPDATE " + "SET content = EXCLUDED.content, updated_at = pg_catalog.now()" + ), + {"id": session_id, "key": key, "content": content}, + ) + + def append_decisions( + self, session_id: str, decisions: Sequence[DecisionInput | dict] + ) -> list[DecisionRecord]: + self._require_uuid4(session_id) + inputs = [DecisionInput.model_validate(item) for item in decisions] + if not inputs: + return [] + with self._transaction() as connection: + self._lock_session(connection, session_id) + self._require_session(connection, session_id) + next_seq = connection.execute( + text( + "SELECT COALESCE(MAX(seq), 0) + 1 FROM thoth_sessions.review_decisions " + "WHERE session_id = :id" + ), + {"id": session_id}, + ).scalar_one() + now = datetime.now(UTC) + records = [ + DecisionRecord(seq=next_seq + offset, ts=now, phase=None, **item.model_dump()) + for offset, item in enumerate(inputs) + ] + connection.execute( + text( + "INSERT INTO thoth_sessions.review_decisions " + "(session_id, seq, ts, phase, type, subject, detail, rationale, retracts) " + "VALUES (:session_id, :seq, :ts, :phase, :type, :subject, :detail, :rationale, :retracts)" + ), + [ + {"session_id": session_id, **record.model_dump(mode="python")} + for record in records + ], + ) + return records + + def get_preferences(self) -> dict: + with self._transaction() as connection: + principal_id = self._upsert_principal(connection) + preferences = connection.execute( + text( + "SELECT preferences FROM thoth_sessions.principal_preferences " + "WHERE principal_id = :principal_id" + ), + {"principal_id": principal_id}, + ).scalar() + return preferences or {} + + def set_preferences(self, preferences: dict) -> None: + if not isinstance(preferences, dict): + raise TypeError("preferences must be a dictionary") + with self._transaction() as connection: + principal_id = self._upsert_principal(connection) + connection.execute( + text( + "INSERT INTO thoth_sessions.principal_preferences (principal_id, preferences) " + "VALUES (:principal_id, CAST(:preferences AS jsonb)) " + "ON CONFLICT (principal_id) DO UPDATE SET preferences = EXCLUDED.preferences, " + "updated_at = pg_catalog.now()" + ), + {"principal_id": principal_id, "preferences": json.dumps(preferences)}, + ) + + def delete(self, session_id: str) -> None: + self._require_uuid4(session_id) + with self._transaction() as connection: + self._lock_session(connection, session_id) + owner = connection.execute( + text( + "SELECT p.issuer, p.subject FROM thoth_sessions.sessions s " + "JOIN thoth_sessions.principals p ON p.id = s.principal_id WHERE s.id = :id" + ), + {"id": session_id}, + ).mappings().first() + if owner is None: + raise SessionError(f"Sessione non trovata: {session_id}") + connection.execute( + text( + "INSERT INTO thoth_sessions.audit_log " + "(action, session_id, actor_issuer, actor_subject, owner_issuer, owner_subject) " + "VALUES ('session_deleted', :id, :actor_issuer, :actor_subject, :owner_issuer, :owner_subject)" + ), + { + "id": session_id, + "actor_issuer": self.principal.issuer, + "actor_subject": self.principal.subject, + "owner_issuer": owner["issuer"], + "owner_subject": owner["subject"], + }, + ) + connection.execute(text("DELETE FROM thoth_sessions.sessions WHERE id = :id"), {"id": session_id}) + + @contextmanager + def _transaction(self) -> Iterator: + try: + with self._engine.begin() as connection: + connection.exec_driver_sql("SET LOCAL search_path = thoth_sessions, pg_catalog, pg_temp") + connection.exec_driver_sql(f"SET LOCAL ROLE {self._runtime_role}") + connection.execute( + text("SELECT pg_catalog.set_config('thoth_sessions.actor_issuer', :value, true)"), + {"value": self.principal.issuer}, + ) + connection.execute( + text("SELECT pg_catalog.set_config('thoth_sessions.actor_subject', :value, true)"), + {"value": self.principal.subject}, + ) + connection.execute( + text("SELECT pg_catalog.set_config('thoth_sessions.is_admin', :value, true)"), + {"value": "true" if self.principal.is_admin else "false"}, + ) + yield connection + except SessionError: + raise + except SQLAlchemyError as exc: + raise SessionError("Session storage unavailable") from exc + + def _upsert_principal(self, connection) -> int: + return connection.execute( + text( + "INSERT INTO thoth_sessions.principals (issuer, subject, display_name) " + "VALUES (:issuer, :subject, :display_name) " + "ON CONFLICT (issuer, subject) DO UPDATE SET display_name = EXCLUDED.display_name, " + "updated_at = pg_catalog.now() RETURNING id" + ), + self.principal.model_dump(include={"issuer", "subject", "display_name"}), + ).scalar_one() + + @staticmethod + def _lock_session(connection, session_id: str) -> None: + connection.execute( + text("SELECT pg_catalog.pg_advisory_xact_lock(pg_catalog.hashtextextended(:id, 0))"), + {"id": session_id}, + ) + + @staticmethod + def _require_session(connection, session_id: str) -> None: + if connection.execute( + text("SELECT 1 FROM thoth_sessions.sessions WHERE id = :id"), {"id": session_id} + ).scalar() is None: + raise SessionError(f"Sessione non trovata: {session_id}") + + @staticmethod + def _require_uuid4(session_id: str) -> None: + try: + parsed = uuid.UUID(session_id, version=4) + except ValueError as exc: + raise SessionError(f"ID sessione non UUIDv4: {session_id}") from exc + if str(parsed) != session_id or parsed.version != 4: + raise SessionError(f"ID sessione non UUIDv4: {session_id}") + + @staticmethod + def _require_artifact_key(key: str) -> None: + if key in _ARTIFACT_KEYS: + return + if key.startswith("cte_sql:") and _SAFE_CTE_NAME.fullmatch(key.removeprefix("cte_sql:")): + return + raise ValueError(f"Unsupported artifact key: {key}") + + +__all__ = ["MigrationError", "MigrationStatus", "PostgresSessionRepository", "migrate", "migration_status"] diff --git a/harness/tht/session/repository.py b/harness/tht/session/repository.py index df8a5925..8e588720 100644 --- a/harness/tht/session/repository.py +++ b/harness/tht/session/repository.py @@ -35,7 +35,13 @@ class SessionRepository(Protocol): def build_session_repository(config, principal: PrincipalContext, *, home=None) -> SessionRepository: - """Build the local adapter for Task 1; server selection arrives with Task 2.""" + """Build the configured private session persistence adapter.""" + session_storage = getattr(config, "session_storage", None) + if session_storage is not None: + from tht.session.postgres_repository import PostgresSessionRepository + + return PostgresSessionRepository.from_config(session_storage.connection, principal) + from tht.config import local_tht_home from tht.session.filesystem_repository import FilesystemSessionRepository