shopdb-flask/plugins/endpoints/migrations/versions/0023_endpoint_credentials.py

"""Per-PC endpoint credentials; remove the client certificate leftovers.

Revision ID: endpoints0023credentials
Revises: endpoints0022certrevocation
Create Date: 2026-09-15

Three things, each guarded against the live schema so the revision is safe on
a database built by create_all, on one that ran the earlier client certificate
builds of 0021/0022, and on a re-run:

1. Drop the client certificate columns (and their index) from the core assets
   table where an earlier build of 0021/0022 created them.
2. Rename endpointapplications.app_fingerprint to appfingerprint where an
   earlier build of 0020 used the underscore spelling.
3. Create endpointcredentials (with the reenrollblocked flag and its reason),
   endpointenrollrequests and endpointmigrationwaves.

MySQL 5.6 safe: no JSON columns, and the enroll request lookup index uses
prefix lengths so the key stays under 767 bytes on utf8mb4.

Downgrade drops the three tables. The certificate columns are not restored:
the code that read them is gone.
"""
from alembic import op

import sqlalchemy as sa


revision = 'endpoints0023credentials'
down_revision = 'endpoints0022certrevocation'
branch_labels = None
depends_on = None
plugin_name = 'endpoints'

CREDENTIALS = 'endpointcredentials'
REQUESTS = 'endpointenrollrequests'
WAVES = 'endpointmigrationwaves'

_CERT_COLUMNS = ('client_cert_fingerprint', 'cert_issued_at', 'cert_expires_at',
                 'cert_revoked_at', 'cert_revocation_reason')
_APPS = 'endpointapplications'
_FINGERPRINT_INDEX = 'idx_endpointapplications_fingerprint'


def _inspector():
    from sqlalchemy import inspect

    return inspect(op.get_bind())


def _has_table(name):
    return _inspector().has_table(name)


def _columns(table):
    return {c['name'] for c in _inspector().get_columns(table)}


def _indexes(table):
    return _inspector().get_indexes(table)


def _drop_certificate_columns():
    if not _has_table('assets'):
        return
    present = _columns('assets') & set(_CERT_COLUMNS)
    if not present:
        return
    # Any index over a certificate column goes first: 0021 named its own,
    # create_all names by convention, and SQLite refuses to drop an indexed column.
    for index in _indexes('assets'):
        if set(index.get('column_names') or ()) & present:
            op.drop_index(index['name'], table_name='assets')
    for column in _CERT_COLUMNS:
        if column in present:
            op.drop_column('assets', column)


def _rename_fingerprint():
    if not _has_table(_APPS):
        return
    columns = _columns(_APPS)
    if 'app_fingerprint' not in columns:
        return
    if 'appfingerprint' in columns:
        return
    for index in _indexes(_APPS):
        if 'app_fingerprint' in (index.get('column_names') or ()):
            op.drop_index(index['name'], table_name=_APPS)
    op.alter_column(_APPS, 'app_fingerprint', new_column_name='appfingerprint',
                    existing_type=sa.String(64), existing_nullable=True)
    op.create_index(_FINGERPRINT_INDEX, _APPS, ['assetid', 'appfingerprint'],
                    unique=True)


def upgrade():
    _drop_certificate_columns()
    _rename_fingerprint()

    if not _has_table(CREDENTIALS):
        op.create_table(
            CREDENTIALS,
            sa.Column('credentialid', sa.Integer(), nullable=False),
            sa.Column('assetid', sa.Integer(), nullable=False),
            sa.Column('credentialhash', sa.CHAR(length=64), nullable=False),
            sa.Column('credentialprefix', sa.String(length=12), nullable=False),
            sa.Column('enrolledvia', sa.String(length=16), nullable=False),
            sa.Column('issuedat', sa.DateTime(), nullable=False),
            sa.Column('lastusedat', sa.DateTime(), nullable=True),
            sa.Column('revokedat', sa.DateTime(), nullable=True),
            sa.Column('revokedby', sa.String(length=64), nullable=True),
            sa.Column('reenrollblocked', sa.Boolean(), nullable=False,
                      server_default=sa.false()),
            sa.Column('reenrollblockedreason', sa.String(length=64),
                      nullable=True),
            sa.ForeignKeyConstraint(['assetid'], ['assets.assetid'],
                                    ondelete='CASCADE'),
            sa.PrimaryKeyConstraint('credentialid'),
            sa.UniqueConstraint('assetid'),
            sa.UniqueConstraint('credentialhash'),
        )
    else:
        # A test database that ran an earlier build of this revision.
        columns = _columns(CREDENTIALS)
        if 'reenrollblocked' not in columns:
            op.add_column(CREDENTIALS, sa.Column(
                'reenrollblocked', sa.Boolean(), nullable=False,
                server_default=sa.false()))
        if 'reenrollblockedreason' not in columns:
            op.add_column(CREDENTIALS, sa.Column(
                'reenrollblockedreason', sa.String(length=64), nullable=True))

    if not _has_table(REQUESTS):
        op.create_table(
            REQUESTS,
            sa.Column('requestid', sa.Integer(), nullable=False),
            sa.Column('hostname', sa.String(length=255), nullable=False),
            sa.Column('serialnumber', sa.String(length=128), nullable=False),
            sa.Column('deviceid', sa.String(length=64), nullable=True),
            sa.Column('enrollnonce', sa.String(length=128), nullable=False),
            sa.Column('agentversion', sa.String(length=32), nullable=True),
            sa.Column('status', sa.String(length=16), nullable=False),
            sa.Column('reason', sa.String(length=64), nullable=True),
            sa.Column('matchedassetid', sa.Integer(), nullable=True),
            sa.Column('sourceip', sa.String(length=45), nullable=True),
            sa.Column('requestedat', sa.DateTime(), nullable=False),
            sa.Column('lastseenat', sa.DateTime(), nullable=False),
            sa.Column('decidedat', sa.DateTime(), nullable=True),
            sa.Column('decidedby', sa.String(length=64), nullable=True),
            sa.ForeignKeyConstraint(['matchedassetid'], ['assets.assetid'],
                                    ondelete='SET NULL'),
            sa.PrimaryKeyConstraint('requestid'),
        )
    existing = {tuple(ix.get('column_names') or ()) for ix in _indexes(REQUESTS)}
    if ('hostname', 'serialnumber') not in existing:
        op.create_index('idx_endpointenrollrequests_hostserial', REQUESTS,
                        ['hostname', 'serialnumber'],
                        mysql_length={'hostname': 128, 'serialnumber': 63})

    if not _has_table(WAVES):
        op.create_table(
            WAVES,
            sa.Column('waveid', sa.Integer(), nullable=False),
            sa.Column('createdat', sa.DateTime(), nullable=False),
            sa.Column('createdby', sa.String(length=64), nullable=True),
            sa.Column('assetids', sa.Text(), nullable=False),
            sa.Column('notes', sa.String(length=255), nullable=True),
            sa.PrimaryKeyConstraint('waveid'),
        )


def downgrade():
    if _has_table(WAVES):
        op.drop_table(WAVES)
    if _has_table(REQUESTS):
        for index in _indexes(REQUESTS):
            if index['name'] == 'idx_endpointenrollrequests_hostserial':
                op.drop_index(index['name'], table_name=REQUESTS)
        op.drop_table(REQUESTS)
    if _has_table(CREDENTIALS):
        op.drop_table(CREDENTIALS)