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

@@ -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

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,

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")),
     )

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

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()