Saltar a contenido

SQL SIGFA 2.0

trx_refresh_tokens

Especificación Técnica: `trx_refresh_tokens`¶

Esquema: core.identidad
Base de Datos: core
Servicio: svc-identidad
Propósito: Tokenización de sesiones: almacén de refresh tokens con rotación y revocación.

1. Justificación y Mejoras de Arquitectura¶

token_hash (SHA-256) evita almacenar tokens en claro.
Ausencia de FK textual por desempeño; unicidad del hash y revocación lógica.
RLS por empresa actual aislada la rotación entre empresas.
Solo created_at como auditoría transaccional ligera.

2. Definiciones de Implementación¶

DBMLSQLSQLAlchemyAlembic

Table core.identidad.trx_refresh_tokens {
    id                         uuid          [pk, default: `gen_random_uuid()`]
    empresa_id                 bigint        [not null]
    token_hash                 varchar(64)   [not null]
    usuario_id                 uuid          [not null, ref: > mae_usuarios.id]
    expiracion                 timestamptz   [not null]
    revocado                   boolean       [default: `false`]
    revocado_en                timestamptz  
    user_agent                 varchar(300) 
    ip                         varchar(45)  

    // Auditoría General
    created_at              timestamptz   [not null, default: `now()`, note: 'Fecha de creación']

    Indexes {
        (token_hash) [unique, name: 'uq_trx_refresh_tokens_token_hash']
    }

    // RLS por empresa actual
}

CREATE TABLE IF NOT EXISTS core.identidad.trx_refresh_tokens (
    id UUID NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
    empresa_id BIGINT NOT NULL,
    token_hash VARCHAR(64) NOT NULL,
    usuario_id UUID NOT NULL,
    expiracion TIMESTAMPTZ NOT NULL,
    revocado BOOLEAN DEFAULT false,
    revocado_en TIMESTAMPTZ,
    user_agent VARCHAR(300),
    ip VARCHAR(45),
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    CONSTRAINT uq_trx_refresh_tokens_token_hash UNIQUE (token_hash),
    CONSTRAINT fk_trx_refresh_tokens_usuario FOREIGN KEY (usuario_id) REFERENCES core.identidad.mae_usuarios (id)
);

-- Row-Level Security
ALTER TABLE core.identidad.trx_refresh_tokens ENABLE ROW LEVEL SECURITY;
CREATE POLICY aislamiento_empresa ON core.identidad.trx_refresh_tokens
    FOR ALL
    USING (empresa_id = NULLIF(current_setting('app.current_empresa_id', true), '')::BIGINT);

-- Comentarios
COMMENT ON COLUMN core.identidad.trx_refresh_tokens.empresa_id IS 'Empresa de la sesión';
COMMENT ON COLUMN core.identidad.trx_refresh_tokens.token_hash IS 'Hash SHA-256 del token';
COMMENT ON COLUMN core.identidad.trx_refresh_tokens.usuario_id IS 'Usuario propietario del token';
COMMENT ON COLUMN core.identidad.trx_refresh_tokens.expiracion IS 'Fecha de expiración';
COMMENT ON COLUMN core.identidad.trx_refresh_tokens.revocado IS 'Token revocado';
COMMENT ON COLUMN core.identidad.trx_refresh_tokens.revocado_en IS 'Fecha de revocación';
COMMENT ON COLUMN core.identidad.trx_refresh_tokens.user_agent IS 'User-Agent del cliente';
COMMENT ON COLUMN core.identidad.trx_refresh_tokens.ip IS 'IP origen (IPv4/IPv6)';

from datetime import datetime, date
from typing import Optional
import uuid

from sqlalchemy import (
    BigInteger,
    Boolean,
    CheckConstraint,
    Date,
    DateTime,
    ForeignKey,
    Identity,
    Index,
    Integer,
    SmallInteger,
    String,
    Text,
    UniqueConstraint,
    text,
)
from sqlalchemy.dialects import postgresql
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass

class TrxRefreshToken(Base):
    __tablename__ = 'trx_refresh_tokens'
    __table_args__ = (
        UniqueConstraint('token_hash', name='uq_trx_refresh_tokens_token_hash'),
        {"schema": 'core.identidad'},
    )

    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        primary_key=True,
        server_default=text("gen_random_uuid()"),
    )
    empresa_id: Mapped[int] = mapped_column(
        BigInteger,
        nullable=False,
        comment='Empresa de la sesión',
    )
    token_hash: Mapped[str] = mapped_column(
        String(64),
        nullable=False,
        comment='Hash SHA-256 del token',
    )
    usuario_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        nullable=False,
        ForeignKey("mae_usuarios.id"),
        comment='Usuario propietario del token',
    )
    expiracion: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        comment='Fecha de expiración',
    )
    revocado: Mapped[bool | None] = mapped_column(
        Boolean,
        server_default=text("false"),
        comment='Token revocado',
    )
    revocado_en: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
        comment='Fecha de revocación',
    )
    user_agent: Mapped[str | None] = mapped_column(
        String(300),
        comment='User-Agent del cliente',
    )
    ip: Mapped[str | None] = mapped_column(
        String(45),
        comment='IP origen (IPv4/IPv6)',
    )

    # Auditoría General
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        server_default=text("now()"),
        comment='Fecha de creación',
    )

"""revision: core_id_0011
create table core.identidad.trx_refresh_tokens
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql


revision: str = 'core_id_0011'
down_revision: str | None = 'core_id_0010'
branch_labels: str | None = None
depends_on: str | None = None


def upgrade() -> None:
    op.create_table(
        'trx_refresh_tokens',
        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
        sa.Column("empresa_id", sa.BigInteger(), nullable=False),
        sa.Column("token_hash", sa.String(length=64), nullable=False),
        sa.Column("usuario_id", postgresql.UUID(as_uuid=True), nullable=False),
        sa.Column("expiracion", sa.DateTime(timezone=True), nullable=False),
        sa.Column("revocado", sa.Boolean(), server_default=sa.text("false")),
        sa.Column("revocado_en", sa.DateTime(timezone=True)),
        sa.Column("user_agent", sa.String(length=300)),
        sa.Column("ip", sa.String(length=45)),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
        sa.UniqueConstraint('token_hash', name='uq_trx_refresh_tokens_token_hash'),
        sa.ForeignKeyConstraint(['usuario_id'], ['core.identidad.mae_usuarios.id'], name='fk_trx_refresh_tokens_usuario'),
        schema='core.identidad',
    )

    op.execute(
        """
        ALTER TABLE core.identidad.trx_refresh_tokens ENABLE ROW LEVEL SECURITY;
        CREATE POLICY aislamiento_empresa
            ON core.identidad.trx_refresh_tokens
            FOR ALL
            USING (empresa_id = NULLIF(current_setting('app.current_empresa_id', true), '')::BIGINT);
        """
    )


def downgrade() -> None:
    op.drop_table('trx_refresh_tokens', schema='core.identidad')