"""SQLAlchemy models."""
import uuid
from datetime import UTC, datetime
from sqlalchemy import (
JSON,
Boolean,
DateTime,
ForeignKey,
Index,
Integer,
LargeBinary,
String,
Text,
UniqueConstraint,
Uuid,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from gnexus_creds.db import Base
def utcnow() -> datetime:
return datetime.now(UTC)
class User(Base):
__tablename__ = "users"
id: Mapped[uuid.UUID] = mapped_column(Uuid(as_uuid=True), primary_key=True, default=uuid.uuid4)
auth_subject: Mapped[str] = mapped_column(String(255), unique=True, index=True)
email: Mapped[str] = mapped_column(String(320), index=True)
display_name: Mapped[str | None] = mapped_column(String(255))
locale: Mapped[str | None] = mapped_column(String(32))
status: Mapped[str] = mapped_column(String(32), default="enabled", index=True)
system_role: Mapped[str] = mapped_column(String(32), default="user", index=True)
profile: Mapped[dict] = mapped_column(JSON, default=dict)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=utcnow, onupdate=utcnow
)
last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
encryption_key: Mapped["UserEncryptionKey"] = relationship(back_populates="user", uselist=False)
class UserEncryptionKey(Base):
__tablename__ = "user_encryption_keys"
id: Mapped[uuid.UUID] = mapped_column(Uuid(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), unique=True
)
key_id: Mapped[str] = mapped_column(String(64), unique=True)
encrypted_key: Mapped[bytes] = mapped_column(LargeBinary)
nonce: Mapped[bytes] = mapped_column(LargeBinary)
algorithm: Mapped[str] = mapped_column(String(64), default="AESGCM")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
user: Mapped[User] = relationship(back_populates="encryption_key")
class Secret(Base):
__tablename__ = "secrets"
id: Mapped[uuid.UUID] = mapped_column(Uuid(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), index=True
)
title: Mapped[str] = mapped_column(String(255))
purpose: Mapped[str | None] = mapped_column(String(255))
category: Mapped[str | None] = mapped_column(String(120), index=True)
source: Mapped[str | None] = mapped_column(String(255))
notes: Mapped[str | None] = mapped_column(String(140))
status: Mapped[str] = mapped_column(String(24), default="actual", index=True)
archived: Mapped[bool] = mapped_column(Boolean, default=False, index=True)
allow_ui: Mapped[bool] = mapped_column(Boolean, default=True)
allow_rest_api: Mapped[bool] = mapped_column(Boolean, default=True)
allow_mcp: Mapped[bool] = mapped_column(Boolean, default=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=utcnow, onupdate=utcnow
)
versions: Mapped[list["SecretVersion"]] = relationship(
back_populates="secret",
cascade="all, delete-orphan",
order_by="SecretVersion.version_number",
)
tags: Mapped[list["SecretTag"]] = relationship(
back_populates="secret",
cascade="all, delete-orphan",
order_by="SecretTag.name",
)
__table_args__ = (
Index("ix_secrets_user_title", "user_id", "title"),
Index("ix_secrets_user_category", "user_id", "category"),
)
class SecretVersion(Base):
__tablename__ = "secret_versions"
__table_args__ = (UniqueConstraint("secret_id", "version_number"),)
id: Mapped[uuid.UUID] = mapped_column(Uuid(as_uuid=True), primary_key=True, default=uuid.uuid4)
secret_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("secrets.id", ondelete="CASCADE"), index=True
)
version_number: Mapped[int] = mapped_column(Integer)
fields: Mapped[list[dict]] = mapped_column(JSON)
search_text: Mapped[str] = mapped_column(Text, default="")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
secret: Mapped[Secret] = relationship(back_populates="versions")
class SecretTag(Base):
__tablename__ = "secret_tags"
__table_args__ = (UniqueConstraint("secret_id", "name"),)
id: Mapped[uuid.UUID] = mapped_column(Uuid(as_uuid=True), primary_key=True, default=uuid.uuid4)
secret_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("secrets.id", ondelete="CASCADE"), index=True
)
user_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), index=True
)
name: Mapped[str] = mapped_column(String(80), index=True)
secret: Mapped[Secret] = relationship(back_populates="tags")
class ApiToken(Base):
__tablename__ = "api_tokens"
id: Mapped[uuid.UUID] = mapped_column(Uuid(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), index=True
)
public_id: Mapped[str] = mapped_column(String(32), unique=True, index=True)
name: Mapped[str] = mapped_column(String(120))
token_hash: Mapped[str] = mapped_column(String(128), unique=True)
scopes: Mapped[list[str]] = mapped_column(JSON, default=list)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
class AuditEvent(Base):
__tablename__ = "audit_events"
id: Mapped[uuid.UUID] = mapped_column(Uuid(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id: Mapped[uuid.UUID | None] = mapped_column(
ForeignKey("users.id", ondelete="SET NULL"), index=True
)
actor_user_id: Mapped[uuid.UUID | None] = mapped_column(
ForeignKey("users.id", ondelete="SET NULL")
)
api_token_id: Mapped[uuid.UUID | None] = mapped_column(
ForeignKey("api_tokens.id", ondelete="SET NULL")
)
secret_id: Mapped[uuid.UUID | None] = mapped_column(Uuid(as_uuid=True), index=True)
channel: Mapped[str] = mapped_column(String(24), index=True)
action: Mapped[str] = mapped_column(String(80), index=True)
ip_address: Mapped[str | None] = mapped_column(String(80))
user_agent: Mapped[str | None] = mapped_column(String(500))
audit_metadata: Mapped[dict] = mapped_column("metadata", JSON, default=dict)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=utcnow, index=True
)
class SessionRecord(Base):
__tablename__ = "sessions"
id: Mapped[str] = mapped_column(String(128), primary_key=True)
user_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), index=True
)
data: Mapped[dict] = mapped_column(JSON, default=dict)
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
class OAuthState(Base):
__tablename__ = "oauth_states"
state: Mapped[str] = mapped_column(String(255), primary_key=True)
pkce_verifier: Mapped[str] = mapped_column(String(255))
return_to: Mapped[str | None] = mapped_column(Text)
scopes: Mapped[list[str]] = mapped_column(JSON, default=list)
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
class RateLimitBucket(Base):
__tablename__ = "rate_limits"
key: Mapped[str] = mapped_column(String(255), primary_key=True)
count: Mapped[int] = mapped_column(Integer, default=0)
window_start: Mapped[datetime] = mapped_column(DateTime(timezone=True))
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=utcnow, onupdate=utcnow
)