diff --git a/.github/workflows/shared-state-postgres.yml b/.github/workflows/shared-state-postgres.yml new file mode 100644 index 000000000..027b4f8fa --- /dev/null +++ b/.github/workflows/shared-state-postgres.yml @@ -0,0 +1,40 @@ +name: Shared-state PostgreSQL acceptance + +on: + pull_request: + paths: + - "src/youtube_extension/services/shared_sql_state.py" + - "apps/web/supabase/migrations/**" + - "scripts/testing/verify_shared_state_postgres.py" + - "tests/unit/test_shared_sql_state.py" + - ".github/workflows/shared-state-postgres.yml" + +permissions: + contents: read + +concurrency: + group: shared-state-postgres-${{ github.event.pull_request.number }} + cancel-in-progress: true + +jobs: + disposable-postgres: + runs-on: ubuntu-latest + timeout-minutes: 5 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.12" + - name: Install bounded acceptance dependencies + run: >- + python -m pip install + pgserver==0.1.4 SQLAlchemy==2.1.8 'psycopg[binary]==3.3.9' + pydantic==2.14.3 aiohttp==3.14.5 requests==2.34.2 redis==8.1.0 + - name: Verify temporary PostgreSQL hardening and persistence + env: + PYTHONPATH: src + DATABASE_URL: "" + EVENTRELAY_SHARED_STATE_DATABASE_URL: "" + run: python scripts/testing/verify_shared_state_postgres.py diff --git a/apps/web/supabase/migrations/20261109062000_create_service_role_shared_state_tables.sql b/apps/web/supabase/migrations/20261109062000_create_service_role_shared_state_tables.sql new file mode 100644 index 000000000..01e0bb0e1 --- /dev/null +++ b/apps/web/supabase/migrations/20261109062000_create_service_role_shared_state_tables.sql @@ -0,0 +1,62 @@ +-- Shared SQL state for session timelines and agent-to-agent (A2A) messages. +-- Source models: src/youtube_extension/services/shared_sql_state.py (SharedSQLStateStore). +-- Columns, types, nullability, primary keys and indexes match the SQLAlchemy Table +-- definitions exactly (SQLAlchemy emits SERIAL for the integer autoincrement keys and +-- ix__ names for index=True). There are no foreign keys in the models. +-- Idempotent: these tables may already exist in a database where an older build ran +-- metadata.create_all(), so tables, indexes, RLS and grants are all safe to re-apply. +-- Access is service-role only, mirroring public.studio_chats: RLS enabled and forced, +-- no policies, nothing granted to public, anon or authenticated. + +create table if not exists public.shared_sessions ( + session_id varchar(255) not null, + payload text not null, + version integer not null, + created_at timestamptz not null, + updated_at timestamptz not null, + constraint shared_sessions_pkey primary key (session_id) +); + +create table if not exists public.shared_timeline_events ( + id serial not null, + session_id varchar(255) not null, + "timestamp" varchar(64) not null, + summary text not null, + content text not null, + constraint shared_timeline_events_pkey primary key (id) +); + +create index if not exists ix_shared_timeline_events_session_id on public.shared_timeline_events (session_id); + +create table if not exists public.shared_a2a_messages ( + id serial not null, + sender varchar(255) not null, + recipient varchar(255) not null, + conversation_id varchar(255) not null, + "timestamp" varchar(64) not null, + content text not null, + constraint shared_a2a_messages_pkey primary key (id) +); + +create index if not exists ix_shared_a2a_messages_conversation_id on public.shared_a2a_messages (conversation_id); + +alter table public.shared_sessions enable row level security; +alter table public.shared_sessions force row level security; +revoke all on table public.shared_sessions from public, anon, authenticated; +grant select, insert, update, delete on table public.shared_sessions to service_role; + +alter table public.shared_timeline_events enable row level security; +alter table public.shared_timeline_events force row level security; +revoke all on table public.shared_timeline_events from public, anon, authenticated; +grant select, insert, update, delete on table public.shared_timeline_events to service_role; + +alter table public.shared_a2a_messages enable row level security; +alter table public.shared_a2a_messages force row level security; +revoke all on table public.shared_a2a_messages from public, anon, authenticated; +grant select, insert, update, delete on table public.shared_a2a_messages to service_role; + +-- The serial keys own sequences; inserts need them. Same role boundary as the tables. +revoke all on sequence public.shared_timeline_events_id_seq from public, anon, authenticated; +grant usage, select on sequence public.shared_timeline_events_id_seq to service_role; +revoke all on sequence public.shared_a2a_messages_id_seq from public, anon, authenticated; +grant usage, select on sequence public.shared_a2a_messages_id_seq to service_role; diff --git a/apps/web/vercel.json b/apps/web/vercel.json new file mode 100644 index 000000000..6502f26e7 --- /dev/null +++ b/apps/web/vercel.json @@ -0,0 +1,7 @@ +{ + "git": { + "deploymentEnabled": { + "fix/2169-reconcile-models-migrations": false + } + } +} diff --git a/docs/MIGRATION_TRUTH.md b/docs/MIGRATION_TRUTH.md index 9c37739d2..8da5401cb 100644 --- a/docs/MIGRATION_TRUTH.md +++ b/docs/MIGRATION_TRUTH.md @@ -1,15 +1,120 @@ # SQL model and migration truth (AUDIT-007 / #2169) -## Current repository evidence — 2026-10-08 +## Current state — 2026-11-09 -The previous inventory in this file named backend model modules and three Alembic revisions. That inventory is obsolete: the current recursive repository tree does not contain `src/youtube_extension/backend/models/`, `src/youtube_extension/backend/migrations/`, or `tests/unit/test_migration_truth.py`. `alembic.ini` still points to the missing migration directory. Do not treat the old count guard or directory list as verified current coverage. +Machine-readable inventory: [`docs/schema_inventory.json`](schema_inventory.json). It is +guarded by `tests/unit/test_schema_inventory.py`, which fails if a listed source or +migration file is missing, if the shared-state migration drifts from the SQLAlchemy +models (columns, types, nullability, primary keys, indexes, foreign keys), or if a +migrated table stops forcing row-level security. -One SQL migration is present at `apps/web/supabase/migrations/20260912214530_create_private_studio_chat_ownership.sql`. Its presence alone does not prove that all live tables, tenant policies, deployed revisions, or Python persistence models are reconciled. +Migrations live in `apps/web/supabase/migrations/` and are named +`_.sql`. -## Acceptance and blocking requirements +| Table | Model / owner | Migration | RLS | Access | +| --- | --- | --- | --- | --- | +| `studio_chats` | `apps/web/src/lib/studio/chat-store.ts` | `20260912214530_create_private_studio_chat_ownership.sql` | enabled + forced | `service_role` only | +| `shared_sessions` | `src/youtube_extension/services/shared_sql_state.py` | `20261109062000_create_service_role_shared_state_tables.sql` | enabled + forced | `service_role` only | +| `shared_timeline_events` | same | same | enabled + forced | `service_role` only | +| `shared_a2a_messages` | same | same | enabled + forced | `service_role` only | -Issue #2169 remains open. Before creating migrations, identify the actual active database and owning service, inventory current persistence models and deployed revision/RLS state using approved read-only access, and map each model to its authoritative migration history. The missing legacy directory needs an explicit retirement or restoration decision; do not silently manufacture a replacement migration chain. +### Schema ownership for the shared-state tables -No live schema comparison, database access, migration generation, or migration application was performed in this correction. A destructive migration requires separate approval. New tables/columns must accompany the matching migration and fixture validation in the same PR. +- **PostgreSQL / Supabase:** the migration is the only schema source. + `SharedSQLStateStore` no longer runs `metadata.create_all()` there; it inspects the + database and raises `SharedStateSchemaError` (naming the migration) if a table or + column is missing, or PostgreSQL tables lack forced RLS, deny-all client ACLs, + required service-role grants or their secured serial sequences. Existing tables + from `create_all()` therefore cannot pass by having the right columns alone. + `AgentOrchestrator` already catches this and falls back to its + in-memory A2A log with a warning; `scripts/maintenance/session_orchestration_manager.py` + fails loudly. +- **SQLite (local default, tests):** `create_all()` still creates the tables, so local + runs need no setup. +- The migration is idempotent (`create table if not exists`, `create index if not + exists`), so it can be applied to a database where an older build already created + these tables with `create_all()`. That path was checked against PostgreSQL 17: the + resulting schema is identical to a fresh apply. -Historical claim retained for audit: the previous file asserted three committed revisions (`001`–`003`) and a `test_migration_truth.py` guard. That claim is not valid for the inspected current tree. +### Tenancy and RLS strategy + +Tables written only by trusted server code are service-role only: RLS enabled and +forced, no policies, all privileges revoked from `public`, `anon` and `authenticated`, +and `select, insert, update, delete` granted to `service_role` (sequences: `usage, +select`). The connecting role must be `service_role` or another role with `BYPASSRLS` +(on Supabase, `postgres` has it). A role without `BYPASSRLS` — including a table owner, +because RLS is forced — reads zero rows and cannot write. + +## Not yet reconciled + +- `supabase/supabase_setup.sql` (projects, chats, files, media, mcp_logs, nods_page, + nods_page_section): a Docker init script, not a migration; RLS enabled but not forced, + with owner/authenticated policies. Needs its own migrate-or-retire decision. +- `infrastructure/database/query_optimization.sql` (video_processing_results, + api_usage_optimized, performance_metrics_optimized): unmanaged script, no application + model, no migration. `index_analysis.py` inspects video_processing_results and + performance_metrics_optimized. +- `supabase/setup_mcp_db.js` attempts to create `mcp_contexts`; bootstrap script, + no checked-in migration or RLS. +- `src/integration/supabase_db.py` `generate_schema_sql()` emits template tables + (profiles, subscriptions) for generated apps; out of scope for EventRelay's own schema. +- The Supabase CLI config is `supabase/config.toml` (project `nsfrhirwsjqwhagtuaxx`), + while migrations are in `apps/web/supabase/migrations/`. Applying them is a manual + step until the two are aligned. + +## Deployment + +Applying migrations to the live Supabase project is a human step; nothing in this +repository applies them automatically. Before applying +`20261109062000_create_service_role_shared_state_tables.sql`, check whether any +deployed service's `DATABASE_URL` / `EVENTRELAY_SHARED_STATE_DATABASE_URL` already +created these tables via `create_all()` — if so they currently have no RLS and default +grants, and the migration will lock them down in place. + +### Plain PostgreSQL prerequisites + +Supabase supplies `anon`, `authenticated`, and `service_role`. A plain PostgreSQL +administrator must provision these roles **before applying the migration**. Run the +following only for roles confirmed absent (existing roles must be inspected, not replaced): + +```sql +CREATE ROLE anon NOLOGIN; +CREATE ROLE authenticated NOLOGIN; +CREATE ROLE service_role NOLOGIN BYPASSRLS; +GRANT USAGE ON SCHEMA public TO service_role; +``` + +Then apply the checked-in migration as the database owner or administrator. Connect +the server with a dedicated login role that has `BYPASSRLS` and inherits the +`service_role` object grants, or use an appropriately configured `service_role` +login. `BYPASSRLS` itself is not inherited through role membership. Provision login +authentication through the database's secret-management process; never commit it. +Client-facing roles must have neither `BYPASSRLS` nor membership in privileged +server roles. No application start-up path creates roles or applies migrations. + +## History + +The October 2026 version of this file noted that the backend model modules and the three +Alembic revisions (`001`–`003`) named in the original audit, and +`tests/unit/test_migration_truth.py`, are absent from the tree. That remains true; the +legacy Alembic chain is treated as retired, not restored. + +## Disposable PostgreSQL acceptance + +`scripts/testing/verify_shared_state_postgres.py` starts its own temporary PostgreSQL +cluster (optional `pgserver` and `psycopg` packages), refuses IP listeners, and tests +legacy tables, effective client privileges, role attributes, sequence access, +server-role writes, fresh-store readback and a hostile `search_path`. Run from the +repository root with `PYTHONPATH=src python scripts/testing/verify_shared_state_postgres.py`. +It never reads deployment database environment variables. + +The November 12, 2026 local attempt was blocked before server startup by +`PermissionError: [Errno 1] Operation not permitted` when creating the private +Unix socket. The same script subsequently passed all **13 cases on PostgreSQL +16.2** in the isolated GitHub runner at source head +`190fc17f7b38a19978094e08aaf3deac5f745c59`: +[acceptance run](https://github.com/groupthinking/EventRelay/actions/runs/43347589645/job/131630572786). +The runner verified no IP listener and stopped and deleted the temporary cluster. +This establishes disposable PostgreSQL acceptance for that source, not application +of the migration or acceptance against any production database. The separate +environment-gated real-PostgreSQL unit test remains skipped in local unit runs. diff --git a/docs/schema_inventory.json b/docs/schema_inventory.json new file mode 100644 index 000000000..09ba48024 --- /dev/null +++ b/docs/schema_inventory.json @@ -0,0 +1,97 @@ +{ + "issue": "https://github.com/groupthinking/EventRelay/issues/2169", + "updated": "2026-11-09", + "migrations_dir": "apps/web/supabase/migrations", + "supabase_project_ref": "nsfrhirwsjqwhagtuaxx", + "notes": "Repository inventory only. No live database was read or changed. 'migrated' means a checked-in migration defines the table; whether it is applied to a given database must be confirmed separately.", + "tables": [ + { + "name": "studio_chats", + "schema": "public", + "model_source": "apps/web/src/lib/studio/chat-store.ts", + "migration": "apps/web/supabase/migrations/20260912214530_create_private_studio_chat_ownership.sql", + "status": "migrated", + "rls": "forced", + "access": "service_role only" + }, + { + "name": "shared_sessions", + "schema": "public", + "model_source": "src/youtube_extension/services/shared_sql_state.py", + "migration": "apps/web/supabase/migrations/20261109062000_create_service_role_shared_state_tables.sql", + "status": "migrated", + "rls": "forced", + "access": "service_role only" + }, + { + "name": "shared_timeline_events", + "schema": "public", + "model_source": "src/youtube_extension/services/shared_sql_state.py", + "migration": "apps/web/supabase/migrations/20261109062000_create_service_role_shared_state_tables.sql", + "status": "migrated", + "rls": "forced", + "access": "service_role only" + }, + { + "name": "shared_a2a_messages", + "schema": "public", + "model_source": "src/youtube_extension/services/shared_sql_state.py", + "migration": "apps/web/supabase/migrations/20261109062000_create_service_role_shared_state_tables.sql", + "status": "migrated", + "rls": "forced", + "access": "service_role only" + } + ], + "unmigrated_sql_sources": [ + { + "source": "supabase/supabase_setup.sql", + "tables": [ + "projects", + "chats", + "files", + "media", + "mcp_logs", + "nods_page", + "nods_page_section" + ], + "status": "setup_script_not_migration", + "rls": "enabled_not_forced", + "note": "Docker init script with owner/authenticated policies; not part of the migrations directory. Needs its own migrate-or-retire decision." + }, + { + "source": "infrastructure/database/query_optimization.sql", + "tables": [ + "video_processing_results", + "api_usage_optimized", + "performance_metrics_optimized" + ], + "status": "unmanaged_sql_script", + "rls": "none", + "note": "infrastructure/database/index_analysis.py inspects video_processing_results and performance_metrics_optimized; api_usage_optimized is also defined by this SQL script. No application model and no migration." + }, + { + "source": "supabase/setup_mcp_db.js", + "tables": ["mcp_contexts"], + "status": "setup_script_not_migration", + "rls": "none", + "note": "JavaScript bootstrap attempts to create mcp_contexts; not part of the migrations directory. Needs its own migrate-or-retire decision." + }, + { + "source": "src/integration/supabase_db.py (generate_schema_sql)", + "tables": [ + "profiles", + "subscriptions" + ], + "status": "out_of_scope_template", + "rls": "enabled_not_forced", + "note": "SQL template emitted for generated apps, not EventRelay's own schema." + } + ], + "retired": [ + { + "source": "src/youtube_extension/backend/models/ and src/youtube_extension/backend/migrations/versions/001-003", + "status": "absent_from_tree", + "note": "Named in the original audit; not present in the current repository." + } + ] +} diff --git a/scripts/testing/verify_shared_state_postgres.py b/scripts/testing/verify_shared_state_postgres.py new file mode 100644 index 000000000..786a2cbf0 --- /dev/null +++ b/scripts/testing/verify_shared_state_postgres.py @@ -0,0 +1,129 @@ +"""Disposable local PostgreSQL acceptance; uses only a private Unix socket.""" +import json +import logging +import signal +import tempfile +from pathlib import Path +from unittest.mock import patch + +import pgserver +from sqlalchemy import create_engine, text +from sqlalchemy.engine import make_url + +from youtube_extension.services.shared_sql_state import ( + SHARED_STATE_MIGRATION, + SharedSQLStateStore, + SharedStateSchemaError, +) + +logging.disable(logging.CRITICAL) +signal.signal(signal.SIGTERM, lambda *_: (_ for _ in ()).throw(SystemExit(124))) +receipt = [] +engines = [] +server = None + + +def expect_hold(label, uri): + try: + store = SharedSQLStateStore(uri) + except SharedStateSchemaError as exc: + assert "security" in str(exc), str(exc) + receipt.append({"case": label, "result": "PASS: rejected"}) + else: + store.engine.dispose() + raise AssertionError(label + " unexpectedly passed") + + +try: + server = pgserver.get_server(tempfile.mkdtemp(prefix="eventrelay-pg-"), cleanup_mode="delete") + base_url = make_url(server.get_uri()).set(drivername="postgresql+psycopg") + assert base_url.host is None or base_url.host.startswith("/"), "Unix-socket-only server required" + admin = create_engine(base_url) + engines.append(admin) + with admin.begin() as conn: + version = conn.scalar(text("SELECT version()")) + assert conn.scalar(text("SHOW listen_addresses")) == "", "IP listeners must be disabled" + conn.exec_driver_sql("CREATE ROLE anon NOLOGIN") + conn.exec_driver_sql("CREATE ROLE authenticated NOLOGIN") + conn.exec_driver_sql("CREATE ROLE service_role NOLOGIN BYPASSRLS") + conn.exec_driver_sql("CREATE ROLE review_server LOGIN BYPASSRLS") + conn.exec_driver_sql("CREATE ROLE review_no_bypass LOGIN") + conn.exec_driver_sql("GRANT service_role TO review_server, review_no_bypass") + conn.exec_driver_sql("GRANT USAGE ON SCHEMA public TO service_role") + admin_uri = base_url.render_as_string(hide_password=False) + with patch.object(SharedSQLStateStore, "_ensure_schema", return_value=None): + legacy = SharedSQLStateStore(admin_uri) + legacy.metadata.create_all(legacy.engine) + legacy.engine.dispose() + expect_hold("legacy create_all without security", admin_uri) + + migration = Path(SHARED_STATE_MIGRATION).read_text() + with admin.begin() as conn: + conn.exec_driver_sql(migration) + conn.exec_driver_sql(migration) + receipt.append({"case": "migration applied twice", "result": "PASS"}) + + service_url = base_url.set(username="review_server") + service_uri = service_url.render_as_string(hide_password=False) + store = SharedSQLStateStore(service_uri) + engines.append(store.engine) + assert store.metadata.schema == "public" + store.save_session({"id": "pg-acceptance", "title": "original"}) + store.append_timeline_event("pg-acceptance", {"timestamp": "00:04", "summary": "source event", "content": "evidence"}) + store.append_a2a_message({"sender": "a", "recipient": "b", "conversation_id": "local-review", "timestamp": "00:04", "content": {"verified": True}}) + reopened = SharedSQLStateStore(service_uri) + engines.append(reopened.engine) + assert reopened.get_session("pg-acceptance")["title"] == "original" + assert len(reopened.get_session("pg-acceptance")["timeline"]) == 1 + assert reopened.get_a2a_messages(conversation_id="local-review")[0]["content"] == {"verified": True} + receipt.append({"case": "dedicated server-role table/sequence writes and fresh-store readback", "result": "PASS"}) + + mutations = [ + ("RLS disabled", "ALTER TABLE public.shared_sessions DISABLE ROW LEVEL SECURITY", "ALTER TABLE public.shared_sessions ENABLE ROW LEVEL SECURITY"), + ("RLS not forced", "ALTER TABLE public.shared_sessions NO FORCE ROW LEVEL SECURITY", "ALTER TABLE public.shared_sessions FORCE ROW LEVEL SECURITY"), + ("direct client table grant", "GRANT SELECT ON public.shared_sessions TO anon", "REVOKE ALL ON public.shared_sessions FROM anon"), + ("client BYPASSRLS", "ALTER ROLE anon BYPASSRLS", "ALTER ROLE anon NOBYPASSRLS"), + ("client superuser", "ALTER ROLE anon SUPERUSER", "ALTER ROLE anon NOSUPERUSER"), + ("missing service grant", "REVOKE UPDATE ON public.shared_sessions FROM service_role", "GRANT UPDATE ON public.shared_sessions TO service_role"), + ("client sequence grant", "GRANT USAGE ON SEQUENCE public.shared_timeline_events_id_seq TO authenticated", "REVOKE ALL ON SEQUENCE public.shared_timeline_events_id_seq FROM authenticated"), + ] + for label, weaken, restore in mutations: + with admin.begin() as conn: + conn.exec_driver_sql(weaken) + try: + expect_hold(label, service_uri) + finally: + with admin.begin() as conn: + conn.exec_driver_sql(restore) + + with admin.begin() as conn: + conn.exec_driver_sql("CREATE ROLE review_client_group NOLOGIN") + conn.exec_driver_sql("GRANT SELECT ON public.shared_sessions TO review_client_group") + conn.exec_driver_sql("GRANT review_client_group TO authenticated") + expect_hold("inherited client table privilege", service_uri) + with admin.begin() as conn: + conn.exec_driver_sql("REVOKE review_client_group FROM authenticated") + conn.exec_driver_sql("REVOKE ALL ON public.shared_sessions FROM review_client_group") + + expect_hold("server role without BYPASSRLS", base_url.set(username="review_no_bypass").render_as_string(hide_password=False)) + + with admin.begin() as conn: + conn.exec_driver_sql("CREATE SCHEMA review_shadow") + conn.exec_driver_sql("GRANT USAGE ON SCHEMA review_shadow TO service_role") + conn.exec_driver_sql("CREATE TABLE review_shadow.shared_sessions (LIKE public.shared_sessions INCLUDING ALL)") + conn.exec_driver_sql("GRANT ALL ON review_shadow.shared_sessions TO service_role") + shadow_uri = service_url.update_query_dict({"options": "-csearch_path=review_shadow,public"}).render_as_string(hide_password=False) + shadow_store = SharedSQLStateStore(shadow_uri) + engines.append(shadow_store.engine) + shadow_store.save_session({"id": "shadow-test", "title": "must stay public"}) + with admin.connect() as conn: + assert conn.scalar(text("SELECT count(*) FROM public.shared_sessions WHERE session_id='shadow-test'")) == 1 + assert conn.scalar(text("SELECT count(*) FROM review_shadow.shared_sessions")) == 0 + receipt.append({"case": "hostile search_path still writes to validated public schema", "result": "PASS"}) + print(json.dumps({"postgres": version, "cases": receipt, "passed": len(receipt), "network": "private Unix socket; no IP listener"}, indent=2)) +finally: + for engine in engines: + engine.dispose() + if server is not None: + server.cleanup() + print("DISPOSABLE_POSTGRES_STOPPED_AND_DELETED") diff --git a/src/youtube_extension/services/shared_sql_state.py b/src/youtube_extension/services/shared_sql_state.py index 635c91edf..a57bd2823 100644 --- a/src/youtube_extension/services/shared_sql_state.py +++ b/src/youtube_extension/services/shared_sql_state.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import logging import os from datetime import datetime, timezone from pathlib import Path @@ -14,18 +15,33 @@ String, Table, Text, + bindparam, create_engine, insert, select, + text, update, ) +from sqlalchemy import inspect as sa_inspect from sqlalchemy.engine import Engine, make_url -from sqlalchemy.exc import IntegrityError +from sqlalchemy.exc import IntegrityError, SQLAlchemyError from sqlalchemy.pool import StaticPool _PROJECT_ROOT = Path(__file__).resolve().parents[3] DEFAULT_SHARED_STATE_PATH = _PROJECT_ROOT / ".runtime" / "shared_state.db" +# The Supabase migration is the schema source of truth for non-SQLite databases. +SHARED_STATE_MIGRATION = ( + "apps/web/supabase/migrations/" + "20261109062000_create_service_role_shared_state_tables.sql" +) + +logger = logging.getLogger(__name__) + + +class SharedStateSchemaError(SQLAlchemyError): + """Raised when a non-SQLite shared-state database lacks the migrated schema.""" + def normalize_shared_storage_url(database_url: str) -> str: """Normalize supported PostgreSQL URLs onto the Psycopg sync driver.""" @@ -77,7 +93,8 @@ def __init__( database_url, sqlite_path=sqlite_path ) self.engine = self._create_engine(self.database_url) - self.metadata = MetaData() + # Validate and use the same schema even with a non-public search_path. + self.metadata = MetaData(schema="public" if self.engine.dialect.name == "postgresql" else None) self.sessions = Table( "shared_sessions", self.metadata, @@ -106,7 +123,108 @@ def __init__( Column("timestamp", String(64), nullable=False), Column("content", Text, nullable=False), ) - self.metadata.create_all(self.engine) + self._ensure_schema() + + def _ensure_schema(self) -> None: + """Create tables for local SQLite; verify migrated tables everywhere else. + + SQLite is the local/test default, so the tables are created on demand. + For PostgreSQL/Supabase the checked-in migration owns the schema (and its + row-level security), so this never issues DDL there: it only confirms the + tables, columns, forced RLS and role privileges match the migration. + """ + + if self.engine.dialect.name == "sqlite": + self.metadata.create_all(self.engine) + return + + inspector = sa_inspect(self.engine) + problems: list[str] = [] + for table in self.metadata.sorted_tables: + if not inspector.has_table(table.name, schema="public"): + problems.append(f"missing table {table.name}") + continue + present = {column["name"] for column in inspector.get_columns(table.name, schema="public")} + missing = [column.name for column in table.columns if column.name not in present] + if missing: + problems.append(f"{table.name} missing columns {', '.join(missing)}") + if not problems and self.engine.dialect.name == "postgresql": + problems.extend(self._postgres_security_problems()) + if problems: + message = ( + "Shared state schema is not migrated on " + f"{self.engine.dialect.name} ({'; '.join(problems)}). " + f"Apply {SHARED_STATE_MIGRATION}; the service does not create " + "these tables outside SQLite." + ) + logger.error(message) + raise SharedStateSchemaError(message) + + def _postgres_security_problems(self) -> list[str]: + """Reject legacy create_all tables that have columns but no hardening.""" + tables = {table.name for table in self.metadata.sorted_tables} + sequences = {"shared_timeline_events_id_seq", "shared_a2a_messages_id_seq"} + query = text(""" + SELECT c.relname, c.relkind, c.relrowsecurity, c.relforcerowsecurity, + (EXISTS ( + SELECT 1 FROM aclexplode(coalesce(c.relacl, acldefault( + CASE WHEN c.relkind = 'S' THEN 'S'::\"char\" ELSE 'r'::\"char\" END, + c.relowner))) a + WHERE a.grantee = 0 OR a.grantee IN ( + SELECT oid FROM pg_roles WHERE rolname IN ('anon', 'authenticated') + ) + ) OR EXISTS ( + SELECT 1 FROM pg_roles r + WHERE r.rolname IN ('anon', 'authenticated') AND ( + r.rolsuper OR r.rolbypassrls OR + CASE WHEN c.relkind = 'S' + THEN has_sequence_privilege(r.oid, c.oid, 'SELECT,USAGE,UPDATE') + ELSE has_table_privilege(r.oid, c.oid, 'SELECT,INSERT,UPDATE,DELETE,TRUNCATE,REFERENCES,TRIGGER') + END + ) + )) AS untrusted_acl, + (SELECT rolsuper OR rolbypassrls FROM pg_roles WHERE rolname = current_user) + AS server_bypasses_rls, + CASE WHEN c.relkind = 'S' + THEN has_sequence_privilege(current_user, c.oid, 'USAGE') + AND has_sequence_privilege(current_user, c.oid, 'SELECT') + ELSE has_table_privilege(current_user, c.oid, 'SELECT') + AND has_table_privilege(current_user, c.oid, 'INSERT') + AND has_table_privilege(current_user, c.oid, 'UPDATE') + AND has_table_privilege(current_user, c.oid, 'DELETE') + END AS server_privileges, + ARRAY( + SELECT privilege_type FROM aclexplode(coalesce(c.relacl, acldefault( + CASE WHEN c.relkind = 'S' THEN 'S'::\"char\" ELSE 'r'::\"char\" END, + c.relowner))) a + WHERE a.grantee = (SELECT oid FROM pg_roles WHERE rolname = 'service_role') + ) AS service_privileges + FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'public' AND c.relname IN :relations + """).bindparams(bindparam("relations", expanding=True)) + with self.engine.connect() as connection: + rows = connection.execute(query, {"relations": sorted(tables | sequences)}).mappings().all() + states = {row["relname"]: row for row in rows} + problems = [] + for name in sorted(tables | sequences): + state = states.get(name) + if state is None: + problems.append(f"security state missing for {name}") + continue + if name in tables and not (state["relrowsecurity"] and state["relforcerowsecurity"]): + problems.append(f"security: {name} must enable and force RLS") + if state["untrusted_acl"]: + problems.append(f"security: {name} grants privileges to public or client roles") + if state["relkind"] != ("r" if name in tables else "S"): + problems.append(f"security: unexpected relation type for {name}") + if not state["server_bypasses_rls"]: + problems.append(f"security: server role cannot bypass forced RLS for {name}") + if not state["server_privileges"]: + problems.append(f"security: server role lacks required privileges for {name}") + required = {"SELECT", "INSERT", "UPDATE", "DELETE"} if name in tables else {"USAGE", "SELECT"} + if not required <= set(state["service_privileges"]): + problems.append(f"security: {name} lacks required service_role privileges") + return problems @staticmethod def _create_engine(database_url: str) -> Engine: diff --git a/tests/unit/test_schema_inventory.py b/tests/unit/test_schema_inventory.py new file mode 100644 index 000000000..bf1b8f1fd --- /dev/null +++ b/tests/unit/test_schema_inventory.py @@ -0,0 +1,232 @@ +"""Schema drift guard for AUDIT-007 (#2169). + +Checks that docs/schema_inventory.json points at real files, that the shared-state +Supabase migration matches the SQLAlchemy models in shared_sql_state.py column for +column, and that every migrated table forces row-level security with no anon or +authenticated access. +""" + +from __future__ import annotations + +import json +import re +from pathlib import Path + +import pytest +from sqlalchemy.dialects import postgresql + +from youtube_extension.services.shared_sql_state import ( + SHARED_STATE_MIGRATION, + SharedSQLStateStore, +) + +REPO_ROOT = Path(__file__).resolve().parents[2] +INVENTORY_PATH = REPO_ROOT / "docs" / "schema_inventory.json" +MIGRATIONS_DIR = REPO_ROOT / "apps" / "web" / "supabase" / "migrations" +SHARED_TABLES = ("shared_sessions", "shared_timeline_events", "shared_a2a_messages") + +_TYPE_ALIASES = { + "TIMESTAMP WITH TIME ZONE": "TIMESTAMPTZ", + "CHARACTER VARYING": "VARCHAR", +} + + +def _strip_comments(sql: str) -> str: + return re.sub(r"--[^\n]*", "", sql) + + +def _load_inventory() -> dict: + return json.loads(INVENTORY_PATH.read_text(encoding="utf-8")) + + +def _normalize_type(raw: str) -> str: + value = re.sub(r"\s+", " ", raw.strip().upper()) + for alias, canonical in _TYPE_ALIASES.items(): + value = value.replace(alias, canonical) + return value.replace(" (", "(") + + +def _parse_create_tables(sql: str) -> dict[str, dict]: + """Parse `create table` blocks into columns, primary keys and foreign keys.""" + + tables: dict[str, dict] = {} + pattern = re.compile( + r"create\s+table\s+(?:if\s+not\s+exists\s+)?(?:public\.)?(\w+)\s*\((.*?)\);", + re.IGNORECASE | re.DOTALL, + ) + for name, body in pattern.findall(sql): + columns: dict[str, dict] = {} + primary_key: list[str] = [] + foreign_keys: list[str] = [] + for raw_line in body.split(",\n"): + line = raw_line.strip().rstrip(",") + if not line: + continue + lowered = line.lower() + pk_match = re.search(r"primary\s+key\s*\(([^)]*)\)", lowered) + if lowered.startswith("constraint") or lowered.startswith("primary key"): + if pk_match: + primary_key.extend(c.strip().strip('"') for c in pk_match.group(1).split(",")) + if "foreign key" in lowered or "references" in lowered: + foreign_keys.append(line) + continue + col_match = re.match(r'"?(\w+)"?\s+(.*)$', line, re.DOTALL) + assert col_match, f"unparsed column line in {name}: {line!r}" + col_name, rest = col_match.groups() + type_match = re.match( + r"((?:timestamp with time zone|timestamptz|character varying|varchar|" + r"serial|integer|text)(?:\s*\(\d+\))?)", + rest, + re.IGNORECASE, + ) + assert type_match, f"unknown type for {name}.{col_name}: {rest!r}" + remainder = rest[type_match.end():].lower() + if "primary key" in remainder: + primary_key.append(col_name) + if "references" in remainder: + foreign_keys.append(line) + columns[col_name] = { + "type": _normalize_type(type_match.group(1)), + "not_null": "not null" in remainder or "primary key" in remainder, + "default": "default" in remainder, + } + tables[name.lower()] = { + "columns": columns, + "primary_key": primary_key, + "foreign_keys": foreign_keys, + } + return tables + + +def _parse_indexes(sql: str) -> dict[str, tuple[str, tuple[str, ...], bool]]: + pattern = re.compile( + r"create\s+(unique\s+)?index\s+(?:if\s+not\s+exists\s+)?(\w+)\s+on\s+" + r"(?:public\.)?(\w+)\s*\(([^)]*)\)", + re.IGNORECASE, + ) + return { + name: (table, tuple(c.strip().strip('"') for c in cols.split(",")), bool(unique)) + for unique, name, table, cols in pattern.findall(sql) + } + + +def _model_tables() -> dict: + store = SharedSQLStateStore("sqlite://") + dialect = postgresql.dialect() + expected: dict[str, dict] = {} + for table in store.metadata.sorted_tables: + columns = {} + for column in table.columns: + if column.primary_key and column.autoincrement is True: + col_type = "SERIAL" + else: + col_type = _normalize_type(column.type.compile(dialect=dialect)) + columns[column.name] = { + "type": col_type, + "not_null": not column.nullable, + "default": column.server_default is not None, + } + indexes = { + index.name: (table.name, tuple(c.name for c in index.columns), bool(index.unique)) + for index in table.indexes + } + expected[table.name] = { + "columns": columns, + "primary_key": [c.name for c in table.primary_key.columns], + "foreign_keys": [fk.target_fullname for fk in table.foreign_keys], + "indexes": indexes, + } + store.engine.dispose() + return expected + + +@pytest.fixture(scope="module") +def shared_migration_sql() -> str: + path = REPO_ROOT / SHARED_STATE_MIGRATION + assert path.is_file(), f"missing migration {SHARED_STATE_MIGRATION}" + return _strip_comments(path.read_text(encoding="utf-8")) + + +def test_inventory_paths_exist_and_migrations_are_checked_in() -> None: + inventory = _load_inventory() + assert (REPO_ROOT / inventory["migrations_dir"]).is_dir() + for entry in inventory["tables"]: + assert (REPO_ROOT / entry["model_source"]).is_file(), entry["name"] + if entry["status"] == "migrated": + migration = REPO_ROOT / entry["migration"] + assert migration.is_file(), entry["name"] + assert migration.parent == MIGRATIONS_DIR + assert re.fullmatch(r"\d{14}_[a-z0-9_]+\.sql", migration.name), migration.name + + +def test_shared_state_tables_are_inventoried_as_migrated_with_forced_rls() -> None: + entries = {entry["name"]: entry for entry in _load_inventory()["tables"]} + for table in SHARED_TABLES: + entry = entries[table] + assert entry["status"] == "migrated" + assert entry["rls"] == "forced" + assert entry["migration"] == SHARED_STATE_MIGRATION + assert entry["model_source"] == "src/youtube_extension/services/shared_sql_state.py" + + +def test_shared_state_migration_matches_sqlalchemy_models(shared_migration_sql: str) -> None: + parsed = _parse_create_tables(shared_migration_sql) + parsed_indexes = _parse_indexes(shared_migration_sql) + expected = _model_tables() + + assert set(parsed) == set(expected) == set(SHARED_TABLES) + for name, model in expected.items(): + table = parsed[name] + assert table["columns"] == model["columns"], name + assert list(table["columns"]) == list(model["columns"]), f"{name} column order" + assert table["primary_key"] == model["primary_key"], name + assert table["foreign_keys"] == [] and model["foreign_keys"] == [], name + migration_indexes = {k: v for k, v in parsed_indexes.items() if v[0] in expected} + model_indexes = {k: v for m in expected.values() for k, v in m["indexes"].items()} + assert migration_indexes == model_indexes + + +def test_shared_state_migration_is_idempotent(shared_migration_sql: str) -> None: + lowered = shared_migration_sql.lower() + assert len(re.findall(r"create\s+table\s+(?!if\s+not\s+exists)", lowered)) == 0 + assert len(re.findall(r"create\s+(?:unique\s+)?index\s+(?!if\s+not\s+exists)", lowered)) == 0 + assert "drop " not in lowered + + +@pytest.mark.parametrize( + "migration_name", + sorted( + { + entry["migration"] + for entry in json.loads(INVENTORY_PATH.read_text(encoding="utf-8"))["tables"] + if entry["rls"] == "forced" + } + ), +) +def test_migrated_tables_force_rls_without_client_access(migration_name: str) -> None: + sql = _strip_comments((REPO_ROOT / migration_name).read_text(encoding="utf-8")).lower() + tables = [ + entry["name"] + for entry in _load_inventory()["tables"] + if entry["migration"] == migration_name and entry["rls"] == "forced" + ] + assert tables + assert "create policy" not in sql + assert not re.search(r"grant[^;]*\bto\s+[^;]*\b(anon|authenticated|public)\b", sql) + for table in tables: + qualified = rf"public\.{table}" + assert re.search(rf"alter table {qualified} enable row level security;", sql), table + assert re.search(rf"alter table {qualified} force row level security;", sql), table + assert re.search( + rf"revoke all on table {qualified} from public, anon, authenticated;", sql + ), table + assert re.search( + rf"grant select, insert, update, delete on table {qualified} to service_role;", sql + ), table + + +def test_shared_state_sequences_are_service_role_only(shared_migration_sql: str) -> None: + sql = shared_migration_sql.lower() + for sequence in ("shared_timeline_events_id_seq", "shared_a2a_messages_id_seq"): + assert f"revoke all on sequence public.{sequence} from public, anon, authenticated;" in sql + assert f"grant usage, select on sequence public.{sequence} to service_role;" in sql diff --git a/tests/unit/test_shared_sql_state.py b/tests/unit/test_shared_sql_state.py new file mode 100644 index 000000000..a2b66d622 --- /dev/null +++ b/tests/unit/test_shared_sql_state.py @@ -0,0 +1,187 @@ +"""Unit tests for youtube_extension/services/shared_sql_state.py schema ownership.""" + +from __future__ import annotations + +import os +from unittest.mock import MagicMock, patch + +import pytest +from sqlalchemy import create_engine, inspect + +from youtube_extension.services import shared_sql_state +from youtube_extension.services.shared_sql_state import ( + SHARED_STATE_MIGRATION, + SharedSQLStateStore, + SharedStateSchemaError, +) + +SHARED_TABLES = {"shared_sessions", "shared_timeline_events", "shared_a2a_messages"} + + +def _secured_relations() -> list[dict]: + return [ + {"relname": name, "relkind": "r", "relrowsecurity": True, "relforcerowsecurity": True, + "untrusted_acl": False, "server_bypasses_rls": True, "server_privileges": True, "service_privileges": ["SELECT", "INSERT", "UPDATE", "DELETE"]} + for name in SHARED_TABLES + ] + [ + {"relname": name, "relkind": "S", "relrowsecurity": False, "relforcerowsecurity": False, + "untrusted_acl": False, "server_bypasses_rls": True, "server_privileges": True, "service_privileges": ["USAGE", "SELECT"]} + for name in ("shared_timeline_events_id_seq", "shared_a2a_messages_id_seq") + ] + + +def _fake_postgres_engine(relations=None) -> MagicMock: + engine = MagicMock() + engine.dialect.name = "postgresql" + engine.connect.return_value.__enter__.return_value.execute.return_value.mappings.return_value.all.return_value = ( + _secured_relations() if relations is None else relations + ) + return engine + + +def _fake_inspector(columns_by_table: dict[str, list[str]]) -> MagicMock: + inspector = MagicMock() + inspector.has_table.side_effect = lambda name, **kwargs: name in columns_by_table + inspector.get_columns.side_effect = lambda name, **kwargs: [ + {"name": column} for column in columns_by_table[name] + ] + return inspector + + +def _full_columns() -> dict[str, list[str]]: + store = SharedSQLStateStore("sqlite://") + columns = {table.name: [c.name for c in table.columns] for table in store.metadata.sorted_tables} + store.engine.dispose() + return columns + + +def test_sqlite_default_creates_tables_and_round_trips(tmp_path) -> None: + db_path = tmp_path / "shared.db" + store = SharedSQLStateStore(f"sqlite:///{db_path}") + + assert SHARED_TABLES <= set(inspect(store.engine).get_table_names()) + store.save_session({"id": "s1", "title": "first"}) + store.append_timeline_event("s1", {"timestamp": "t1", "summary": "sum", "content": "c"}) + store.append_a2a_message( + { + "sender": "a", + "recipient": "b", + "conversation_id": "conv", + "timestamp": "t1", + "content": {"k": "v"}, + } + ) + + reopened = SharedSQLStateStore(f"sqlite:///{db_path}") + session = reopened.get_session("s1") + assert session is not None and session["title"] == "first" + assert session["timeline"] == [{"timestamp": "t1", "summary": "sum", "content": "c"}] + assert reopened.get_a2a_messages(conversation_id="conv")[0]["content"] == {"k": "v"} + + +def test_sqlite_path_default_used_without_database_url(tmp_path) -> None: + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("EVENTRELAY_SHARED_STATE_DATABASE_URL", None) + os.environ.pop("DATABASE_URL", None) + store = SharedSQLStateStore(sqlite_path=tmp_path / "local.db") + assert store.engine.dialect.name == "sqlite" + assert (tmp_path / "local.db").exists() + assert SHARED_TABLES <= set(inspect(store.engine).get_table_names()) + + +def test_postgres_never_runs_create_all_when_schema_is_migrated() -> None: + columns = _full_columns() + with ( + patch.object(SharedSQLStateStore, "_create_engine", return_value=_fake_postgres_engine()), + patch.object(shared_sql_state, "sa_inspect", return_value=_fake_inspector(columns)), + patch("sqlalchemy.MetaData.create_all") as create_all, + ): + store = SharedSQLStateStore("postgresql+psycopg://user:pw@db.example.com/app") + create_all.assert_not_called() + assert all(table.schema == "public" for table in store.metadata.sorted_tables) + + +def test_postgres_missing_table_raises_and_names_migration(caplog) -> None: + columns = _full_columns() + columns.pop("shared_a2a_messages") + with ( + patch.object(SharedSQLStateStore, "_create_engine", return_value=_fake_postgres_engine()), + patch.object(shared_sql_state, "sa_inspect", return_value=_fake_inspector(columns)), + patch("sqlalchemy.MetaData.create_all") as create_all, + caplog.at_level("ERROR"), + pytest.raises(SharedStateSchemaError) as excinfo, + ): + SharedSQLStateStore("postgresql+psycopg://user:pw@db.example.com/app") + create_all.assert_not_called() + assert "missing table shared_a2a_messages" in str(excinfo.value) + assert SHARED_STATE_MIGRATION in str(excinfo.value) + assert SHARED_STATE_MIGRATION in caplog.text + + +def test_postgres_missing_column_raises() -> None: + columns = _full_columns() + columns["shared_sessions"].remove("version") + with ( + patch.object(SharedSQLStateStore, "_create_engine", return_value=_fake_postgres_engine()), + patch.object(shared_sql_state, "sa_inspect", return_value=_fake_inspector(columns)), + pytest.raises(SharedStateSchemaError, match="shared_sessions missing columns version"), + ): + SharedSQLStateStore("postgresql+psycopg://user:pw@db.example.com/app") + + +@pytest.mark.parametrize("weakness", ["legacy-create-all", "unforced-rls", "client-grant", "missing-service-grant", "missing-sequence", "server-without-bypass", "server-without-grants", "wrong-relation-type"]) +def test_postgres_existing_columns_do_not_bypass_required_security(weakness) -> None: + columns = _full_columns() + relations = _secured_relations() + if weakness == "legacy-create-all": + relations[0]["relrowsecurity"] = False + relations[0]["relforcerowsecurity"] = False + elif weakness == "unforced-rls": + relations[0]["relforcerowsecurity"] = False + elif weakness == "client-grant": + relations[0]["untrusted_acl"] = True + elif weakness == "missing-service-grant": + relations[0]["service_privileges"] = ["SELECT"] + elif weakness == "server-without-bypass": + relations[0]["server_bypasses_rls"] = False + elif weakness == "server-without-grants": + relations[0]["server_privileges"] = False + elif weakness == "wrong-relation-type": + relations[0]["relkind"] = "v" + else: + relations.pop() + with ( + patch.object(SharedSQLStateStore, "_create_engine", return_value=_fake_postgres_engine(relations)), + patch.object(shared_sql_state, "sa_inspect", return_value=_fake_inspector(columns)), + pytest.raises(SharedStateSchemaError, match="security") as error, + ): + SharedSQLStateStore("postgresql+psycopg://user:pw@db.example.com/app") + assert SHARED_STATE_MIGRATION in str(error.value) + + +@pytest.mark.skipif( + not os.getenv("EVENTRELAY_TEST_POSTGRES_URL"), + reason=( + "set EVENTRELAY_TEST_POSTGRES_URL to a throwaway Postgres (superuser, with " + "anon, authenticated and service_role roles) to run" + ), +) +def test_real_postgres_requires_migration_then_works() -> None: + from pathlib import Path + + url = shared_sql_state.normalize_shared_storage_url(os.environ["EVENTRELAY_TEST_POSTGRES_URL"]) + engine = create_engine(url) + with engine.begin() as conn: + for table in SHARED_TABLES: + conn.exec_driver_sql(f"drop table if exists public.{table} cascade") + with pytest.raises(SharedStateSchemaError): + SharedSQLStateStore(url) + + migration = Path(__file__).resolve().parents[2] / SHARED_STATE_MIGRATION + with engine.begin() as conn: + conn.exec_driver_sql(migration.read_text(encoding="utf-8")) + store = SharedSQLStateStore(url) + store.save_session({"id": "pg1"}) + assert store.get_session("pg1") == {"id": "pg1", "timeline": []} + engine.dispose() + store.engine.dispose() diff --git a/vercel.json b/vercel.json index d77369556..d41b5ab69 100644 --- a/vercel.json +++ b/vercel.json @@ -7,7 +7,10 @@ { "source": "/(.*)", "has": [ - { "type": "host", "value": "event-relay-web.vercel.app" } + { + "type": "host", + "value": "event-relay-web.vercel.app" + } ], "destination": "https://uvai.io/$1", "permanent": true @@ -15,10 +18,18 @@ { "source": "/(.*)", "has": [ - { "type": "host", "value": "v0-uvai.vercel.app" } + { + "type": "host", + "value": "v0-uvai.vercel.app" + } ], "destination": "https://uvai.io/$1", "permanent": true } - ] + ], + "git": { + "deploymentEnabled": { + "fix/2169-reconcile-models-migrations": false + } + } }