Files

61 lines
2.9 KiB
Python

"""Release 2: expand certificates table for CA/server-certificate PKI.
Revision ID: r2_0002
Revises: r1_0001
Create Date: 2026-09-19
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
revision: str = "r2_0002"
down_revision: Union[str, None] = "r1_0001"
branch_labels: Union[str, Sequence[str] | None] = None
depends_on: Union[str, Sequence[str] | None] = None
def upgrade() -> None:
# The certificates table exists since r1_0001 (forward-compatible
# placeholder). Release 2 adds CA semantics, hierarchy, SANs,
# serial/fingerprint identity, and a lifecycle status.
op.add_column("certificates", sa.Column("is_ca", sa.Boolean(), nullable=False, server_default=sa.false()))
op.add_column("certificates", sa.Column("parent_id", sa.Uuid(), nullable=True))
op.add_column("certificates", sa.Column("key_algorithm", sa.String(20), nullable=False, server_default="ecdsa-p256"))
# Serial numbers are unique per RFC 5280; existing rows have none, so
# the column is added nullable first, then made NOT NULL below (the
# table has never been written to, so no backfill is needed).
op.add_column("certificates", sa.Column("serial_number", sa.String(64), nullable=True))
op.add_column("certificates", sa.Column("dns_sans", sa.JSON(), nullable=False, server_default="[]"))
op.add_column("certificates", sa.Column("ip_sans", sa.JSON(), nullable=False, server_default="[]"))
op.add_column("certificates", sa.Column("fingerprint", sa.String(200), nullable=True))
op.add_column("certificates", sa.Column("status", sa.String(20), nullable=False, server_default="issued"))
op.create_foreign_key(
"fk_certificates_parent",
"certificates",
"certificates",
["parent_id"],
["id"],
ondelete="SET NULL",
)
# No rows exist yet in practice (table was never written to), so the
# unique index is safe; if a legacy row existed without a serial it
# must be removed manually first.
op.create_index("ix_certificates_serial_number", "certificates", ["serial_number"], unique=True)
op.create_index("ix_certificates_fingerprint", "certificates", ["fingerprint"])
op.alter_column("certificates", "serial_number", nullable=False)
def downgrade() -> None:
op.alter_column("certificates", "serial_number", nullable=True)
op.drop_index("ix_certificates_fingerprint", table_name="certificates")
op.drop_index("ix_certificates_serial_number", table_name="certificates")
op.drop_constraint("fk_certificates_parent", "certificates", type_="foreignkey")
op.drop_column("certificates", "status")
op.drop_column("certificates", "fingerprint")
op.drop_column("certificates", "ip_sans")
op.drop_column("certificates", "dns_sans")
op.drop_column("certificates", "serial_number")
op.drop_column("certificates", "key_algorithm")
op.drop_column("certificates", "parent_id")
op.drop_column("certificates", "is_ca")