From 4143f7c004c0f064cc77569ad951cb42d59ffdbf Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 22 Aug 2026 18:15:23 +0200 Subject: [PATCH] fix: migrate as durable database owner Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a028f0-a42f-7582-89a8-ebaad7343834 --- docs/operator-guide.md | 5 +++++ migrations/env.py | 7 ++++++- migrations/versions/0001_initial_schema.py | 17 +++++++++++++---- src/sbom_nexus/config.py | 13 +++++++++++++ tests/test_config.py | 16 +++++++++++++++- 5 files changed, 52 insertions(+), 6 deletions(-) diff --git a/docs/operator-guide.md b/docs/operator-guide.md index 66becc4..bce6007 100644 --- a/docs/operator-guide.md +++ b/docs/operator-guide.md @@ -18,6 +18,11 @@ make migrate make run ``` +When migrations use short-lived login roles, set +`SBOM_NEXUS_MIGRATION_ROLE` to the reviewed durable owner role. Alembic issues +`SET ROLE` before creating or changing any object, so ownership does not become +tied to an expiring lease. The production package uses `sbom_nexus_owner`. + Production deployments should mount a Secret and set `SBOM_NEXUS_DATABASE_URL_FILE` to its `url` file rather than exposing the DSN in a manifest or command argument. The same file setting is consumed by both diff --git a/migrations/env.py b/migrations/env.py index a35b63a..40e4f85 100644 --- a/migrations/env.py +++ b/migrations/env.py @@ -5,7 +5,7 @@ from logging.config import fileConfig from alembic import context from sqlalchemy import engine_from_config, pool -from sbom_nexus.config import database_target +from sbom_nexus.config import database_target, migration_role from sbom_nexus.database import database_url, metadata config = context.config @@ -41,6 +41,11 @@ def run_migrations_online() -> None: poolclass=pool.NullPool, ) with connectable.connect() as connection: + role = migration_role() + if role and connection.dialect.name != "postgresql": + raise RuntimeError("SBOM_NEXUS_MIGRATION_ROLE requires PostgreSQL") + if role: + connection.exec_driver_sql(f'SET ROLE "{role}"') context.configure( connection=connection, target_metadata=target_metadata, diff --git a/migrations/versions/0001_initial_schema.py b/migrations/versions/0001_initial_schema.py index ebb7887..06dc056 100644 --- a/migrations/versions/0001_initial_schema.py +++ b/migrations/versions/0001_initial_schema.py @@ -7,8 +7,8 @@ Create Date: 2026-08-22 from __future__ import annotations -from alembic import op import sqlalchemy as sa +from alembic import op revision = "0001" down_revision = None @@ -46,7 +46,10 @@ def upgrade() -> None: sa.Column("legacy_id", sa.String(length=100), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), sa.ForeignKeyConstraint( - ["repo_id"], ["repositories.id"], name=op.f("fk_snapshots_repo_id_repositories"), ondelete="RESTRICT" + ["repo_id"], + ["repositories.id"], + name=op.f("fk_snapshots_repo_id_repositories"), + ondelete="RESTRICT", ), sa.PrimaryKeyConstraint("id", name=op.f("pk_snapshots")), sa.UniqueConstraint("legacy_id", name=op.f("uq_snapshots_legacy_id")), @@ -66,10 +69,16 @@ def upgrade() -> None: sa.Column("source_path", sa.Text(), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), sa.ForeignKeyConstraint( - ["repo_id"], ["repositories.id"], name=op.f("fk_entries_repo_id_repositories"), ondelete="RESTRICT" + ["repo_id"], + ["repositories.id"], + name=op.f("fk_entries_repo_id_repositories"), + ondelete="RESTRICT", ), sa.ForeignKeyConstraint( - ["snapshot_id"], ["snapshots.id"], name=op.f("fk_entries_snapshot_id_snapshots"), ondelete="RESTRICT" + ["snapshot_id"], + ["snapshots.id"], + name=op.f("fk_entries_snapshot_id_snapshots"), + ondelete="RESTRICT", ), sa.PrimaryKeyConstraint("id", name=op.f("pk_entries")), ) diff --git a/src/sbom_nexus/config.py b/src/sbom_nexus/config.py index e1e84b4..23fdb76 100644 --- a/src/sbom_nexus/config.py +++ b/src/sbom_nexus/config.py @@ -3,8 +3,11 @@ from __future__ import annotations import os +import re from pathlib import Path +POSTGRES_IDENTIFIER = re.compile(r"^[a-z_][a-z0-9_]*$") + def database_target(default: str | Path | None = None) -> str | Path: """Return the configured database target, preferring a mounted secret file.""" @@ -33,3 +36,13 @@ def database_target(default: str | Path | None = None) -> str | Path: "or SBOM_NEXUS_DATABASE_PATH" ) return default + + +def migration_role() -> str | None: + """Return the durable PostgreSQL role that must own migrated objects.""" + role = os.getenv("SBOM_NEXUS_MIGRATION_ROLE") + if not role: + return None + if not POSTGRES_IDENTIFIER.fullmatch(role): + raise RuntimeError("SBOM_NEXUS_MIGRATION_ROLE must be a PostgreSQL identifier") + return role diff --git a/tests/test_config.py b/tests/test_config.py index e926ca3..9f1bffd 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -4,7 +4,7 @@ from pathlib import Path import pytest -from sbom_nexus.config import database_target +from sbom_nexus.config import database_target, migration_role def test_database_target_prefers_secret_file(monkeypatch, tmp_path: Path) -> None: @@ -23,3 +23,17 @@ def test_database_target_rejects_empty_secret_file(monkeypatch, tmp_path: Path) with pytest.raises(RuntimeError, match="is empty"): database_target() + + +def test_migration_role_accepts_a_postgres_identifier(monkeypatch) -> None: + monkeypatch.setenv("SBOM_NEXUS_MIGRATION_ROLE", "sbom_nexus_owner") + + assert migration_role() == "sbom_nexus_owner" + + +@pytest.mark.parametrize("role", ["owner; DROP DATABASE postgres", "UpperCase", 'bad"role']) +def test_migration_role_rejects_unsafe_identifiers(monkeypatch, role: str) -> None: + monkeypatch.setenv("SBOM_NEXUS_MIGRATION_ROLE", role) + + with pytest.raises(RuntimeError, match="PostgreSQL identifier"): + migration_role()