feat(api_keys): implement API key management with CRUD operations and permissions

app/db/crud/admin.py

@@ -12,6 +12,7 @@
     get_complete_period_start_for_filter,
     to_utc_for_filter,
 )
+from app.db.crud.api_key import update_api_keys_role
 from app.db.models import Admin, AdminNotificationReminder, AdminRole, AdminUsageLogs, NodeUserUsage, ReminderType, User
 from app.models.admin import (
     AdminCreate,
@@ -218,6 +219,9 @@ async def update_admin(db: AsyncSession, db_admin: Admin, modified_admin: AdminM
     if modified_admin.notification_enable is not None:
         db_admin.notification_enable = modified_admin.notification_enable.model_dump()
 
+    if modified_admin.role_id is not None:
+        await update_api_keys_role(db, db_admin.id, modified_admin.role_id)
+
     await db.commit()
     await db.refresh(db_admin)
     await load_admin_attrs(db_admin)

app/db/crud/api_key.py

@@ -0,0 +1,127 @@
+import uuid
+from datetime import datetime as dt, timezone as tz
+
+from sqlalchemy import func, or_, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.db.models import Admin, AdminStatus, APIKey, APIKeyStatus
+from app.models.api_key import APIKeyCreate
+from app.utils.crypto import api_key_lookup_id, hash_api_key, verify_api_key
+from app.utils.jwt import get_secret_key
+
+
+async def create_api_key(
+    db: AsyncSession,
+    admin_id: int,
+    model: APIKeyCreate,
+) -> tuple[str, APIKey]:
+    raw_key = str(uuid.uuid4())
+    pepper = await get_secret_key()
+    db_key = APIKey(
+        admin_id=admin_id,
+        role_id=model.role_id,
+        name=model.name,
+        note=model.note,
+        key_hash=hash_api_key(raw_key, pepper=pepper),
+        expire_date=model.expire_date,
+    )
+    db.add(db_key)
+    await db.flush()
+    await db.refresh(db_key)
+    return raw_key, db_key
+
+
+async def get_api_key_by_raw_key(db: AsyncSession, raw_api_key: str) -> APIKey | None:
+    pepper = await get_secret_key()
+    lookup_id = api_key_lookup_id(raw_api_key)
+
+    stmt = (
+        select(APIKey)
+        .where(
+            or_(
+                APIKey.key_hash.startswith(f"v2${lookup_id}$"),
+                APIKey.key_hash.startswith(f"v1${lookup_id}$"),
+                # Handle cases where version prefix might be missing in some older implementations
+                APIKey.key_hash.startswith(f"{lookup_id}$"),
+            ),
+            APIKey.status != APIKeyStatus.disabled,
+        )
+        .options(selectinload(APIKey.admin).selectinload(Admin.role), selectinload(APIKey.role))
+        .limit(1)
+    )
+    db_key = (await db.execute(stmt)).scalar_one_or_none()
+
+    if db_key is None or not verify_api_key(raw_api_key, db_key.key_hash, pepper=pepper):
+        return None
+    # Reject if the owning admin is disabled
+    if db_key.admin is not None and db_key.admin.status == AdminStatus.disabled:
+        return None
+    return db_key
+
+
+async def get_api_key_by_id(db: AsyncSession, key_id: int) -> APIKey | None:
+    stmt = select(APIKey).where(APIKey.id == key_id).options(selectinload(APIKey.admin), selectinload(APIKey.role))
+    return (await db.execute(stmt)).scalar_one_or_none()
+
+
+async def get_api_keys(
+    db: AsyncSession,
+    *,
+    admin_id: int | None,
+    offset: int,
+    limit: int,
+    key_id: int | None = None,
+    name: str | None = None,
+    status: APIKeyStatus | None = None,
+) -> tuple[list[APIKey], int]:
+    stmt = select(APIKey).options(selectinload(APIKey.role))
+    if admin_id is not None:
+        stmt = stmt.where(APIKey.admin_id == admin_id)
+    if key_id is not None:
+        stmt = stmt.where(APIKey.id == key_id)
+    if name is not None:
+        stmt = stmt.where(APIKey.name == name)
+    if status is not None:
+        if status == APIKeyStatus.active:
+            # active = stored status is active AND not past expire_date
+            stmt = stmt.where(APIKey.status == APIKeyStatus.active, ~APIKey.is_expired)
+        else:
+            # disabled = stored status is disabled (expire_date irrelevant)
+            stmt = stmt.where(APIKey.status == APIKeyStatus.disabled)
+
+    total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar() or 0
+
+    stmt = stmt.order_by(APIKey.created_at.desc()).offset(offset).limit(limit)
+    rows = list((await db.execute(stmt)).scalars().all())
+    return rows, total
+
+
+async def delete_api_key(db: AsyncSession, db_key: APIKey) -> None:
+    await db.delete(db_key)
+    await db.flush()
+
+
+async def revoke_api_key(db: AsyncSession, db_key: APIKey) -> tuple[str, APIKey]:
+    raw_key = str(uuid.uuid4())
+    pepper = await get_secret_key()
+    db_key.key_hash = hash_api_key(raw_key, pepper=pepper)
+    db_key.revoked_at = dt.now(tz.utc)
+    db_key.status = APIKeyStatus.active
+    await db.flush()
+    await db.refresh(db_key)
+    return raw_key, db_key
+
+
+async def update_api_key(db: AsyncSession, db_key: APIKey, update_data: dict) -> APIKey:
+    for key, value in update_data.items():
+        setattr(db_key, key, value)
+    await db.flush()
+    await db.refresh(db_key)
+    return db_key
+
+
+async def update_api_keys_role(db: AsyncSession, admin_id: int, new_role_id: int) -> int:
+    """Update role_id on all API keys belonging to admin_id. Returns affected row count."""
+    result = await db.execute(update(APIKey).where(APIKey.admin_id == admin_id).values(role_id=new_role_id))
+    return result.rowcount

app/db/migrations/versions/c9b48df42f10_add_api_keys_table.py

@@ -0,0 +1,130 @@
+"""add api keys table
+
+Revision ID: c9b48df42f10
+Revises: f9c69a49f544
+Create Date: 2026-05-25 00:00:00.000000
+
+"""
+
+import json
+
+from alembic import op
+import sqlalchemy as sa
+import app.db.compiles_types
+
+
+# revision identifiers, used by Alembic.
+revision = "c9b48df42f10"
+down_revision = "f9c69a49f544"
+branch_labels = None
+depends_on = None
+
+
+OWNER_ADMIN_API_KEY_PERMS = {
+    "create": True,
+    "read": True,
+    "read_simple": True,
+    "modify": True,
+    "delete": True,
+}
+
+OPERATOR_API_KEY_PERMS = {
+    "read": {"scope": 1},
+    "read_simple": {"scope": 1},
+    "modify": {"scope": 1},
+    "delete": {"scope": 1},
+}
+
+
+def _normalize_permissions(value):
+    if value is None:
+        return {}
+    if isinstance(value, str):
+        try:
+            return json.loads(value)
+        except json.JSONDecodeError:
+            return {}
+    if isinstance(value, dict):
+        return value
+    return {}
+
+
+def upgrade() -> None:
+    op.create_table(
+        "api_keys",
+        sa.Column("id", app.db.compiles_types.SqliteCompatibleBigInteger(), autoincrement=True, nullable=False),
+        sa.Column("admin_id", app.db.compiles_types.SqliteCompatibleBigInteger(), nullable=False),
+        sa.Column("name", sa.String(length=128), nullable=False),
+        sa.Column("note", sa.String(length=512), nullable=True),
+        sa.Column("key_hash", sa.String(length=128), nullable=False),
+        sa.Column("role_id", app.db.compiles_types.SqliteCompatibleBigInteger(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("expire_date", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "status",
+            sa.Enum("active", "disabled", name="apikeystatus"),
+            nullable=False,
+            server_default="active",
+        ),
+        sa.ForeignKeyConstraint(
+            ["admin_id"], ["admins.id"], name=op.f("fk_api_keys_admin_id_admins"), ondelete="CASCADE"
+        ),
+        sa.ForeignKeyConstraint(["role_id"], ["admin_roles.id"], name=op.f("fk_api_keys_role_id_admin_roles")),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_api_keys")),
+        sa.UniqueConstraint("key_hash", name=op.f("uq_api_keys_key_hash")),
+        sa.UniqueConstraint("admin_id", "name", name=op.f("uq_api_keys_admin_id")),
+    )
+    with op.batch_alter_table("api_keys", schema=None) as batch_op:
+        batch_op.create_index(batch_op.f("ix_api_keys_admin_id"), ["admin_id"], unique=False)
+        batch_op.create_index(batch_op.f("ix_api_keys_created_at"), ["created_at"], unique=False)
+        batch_op.create_index(batch_op.f("ix_api_keys_expire_date"), ["expire_date"], unique=False)
+
+    conn = op.get_bind()
+    admin_roles = sa.table(
+        "admin_roles",
+        sa.column("id", sa.Integer),
+        sa.column("name", sa.String),
+        sa.column("permissions", sa.JSON),
+    )
+
+    rows = conn.execute(sa.select(admin_roles.c.id, admin_roles.c.name, admin_roles.c.permissions)).fetchall()
+    for role_id, role_name, role_permissions in rows:
+        permissions = _normalize_permissions(role_permissions)
+        if "api_keys" in permissions:
+            continue
+
+        if role_name in {"owner", "administrator"}:
+            permissions["api_keys"] = OWNER_ADMIN_API_KEY_PERMS
+        else:
+            permissions["api_keys"] = OPERATOR_API_KEY_PERMS
+
+        conn.execute(admin_roles.update().where(admin_roles.c.id == role_id).values(permissions=permissions))
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    admin_roles = sa.table(
+        "admin_roles",
+        sa.column("id", sa.Integer),
+        sa.column("permissions", sa.JSON),
+    )
+
+    rows = conn.execute(sa.select(admin_roles.c.id, admin_roles.c.permissions)).fetchall()
+    for role_id, role_permissions in rows:
+        permissions = _normalize_permissions(role_permissions)
+        if "api_keys" not in permissions:
+            continue
+        permissions.pop("api_keys", None)
+        conn.execute(admin_roles.update().where(admin_roles.c.id == role_id).values(permissions=permissions))
+
+    with op.batch_alter_table("api_keys", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_api_keys_expire_date"))
+        batch_op.drop_index(batch_op.f("ix_api_keys_created_at"))
+        batch_op.drop_index(batch_op.f("ix_api_keys_admin_id"))
+
+    op.drop_table("api_keys")
+
+    # Drop the enum type for PostgreSQL
+    if conn.dialect.name == "postgresql":
+        op.execute("DROP TYPE IF EXISTS apikeystatus")

app/db/models.py

@@ -84,6 +84,9 @@ class Admin(Base, CreatedAtUTCMixin):
     notification_reminders: Mapped[List["AdminNotificationReminder"]] = relationship(
         back_populates="admin", init=False, default_factory=list, cascade="all, delete-orphan"
     )
+    api_keys: Mapped[List["APIKey"]] = relationship(
+        back_populates="admin", init=False, default_factory=list, cascade="all, delete-orphan"
+    )
 
     password_reset_at: Mapped[Optional[dt]] = mapped_column(DateTime(timezone=True), default=None)
     telegram_id: Mapped[Optional[int]] = mapped_column(BigInteger, default=None)
@@ -151,6 +154,11 @@ def users_sync_blocked(self) -> bool:
     def total_users(self) -> int:
         return len(self.users)
 
+    @property
+    def has_api_keys(self) -> bool:
+        """True when the admin owns at least one API key."""
+        return len(self.api_keys) > 0
+
 
 class AdminUsageLogs(Base, IdMixin):
     __tablename__ = "admin_usage_logs"
@@ -877,6 +885,7 @@ class AdminRole(Base, CreatedAtUTCMixin):
     disconnect_users_when_limited: Mapped[bool] = mapped_column(default=True, server_default="1")
     disconnect_users_when_disabled: Mapped[bool] = mapped_column(default=True, server_default="1")
     admins: Mapped[List["Admin"]] = relationship(back_populates="role", init=False, viewonly=True, lazy="noload")
+    api_keys: Mapped[List["APIKey"]] = relationship(back_populates="role", init=False, lazy="noload")
 
     @hybrid_property
     def is_builtin(self) -> bool:
@@ -888,6 +897,58 @@ def is_builtin(cls):
         return cls.id <= 3
 
 
+class APIKeyStatus(str, Enum):
+    active = "active"
+    disabled = "disabled"
+
+
+class APIKey(Base, CreatedAtUTCMixin):
+    __tablename__ = "api_keys"
+    __table_args__ = (
+        UniqueConstraint("key_hash"),
+        UniqueConstraint("admin_id", "name"),
+        Index("ix_api_keys_admin_id", "admin_id"),
+        Index("ix_api_keys_created_at", "created_at"),
+        Index("ix_api_keys_expire_date", "expire_date"),
+    )
+
+    admin_id: Mapped[int] = fk_id_column("admins.id", ondelete="CASCADE")
+    admin: Mapped["Admin"] = relationship(back_populates="api_keys", init=False)
+    name: Mapped[str] = mapped_column(String(128), nullable=False)
+    key_hash: Mapped[str] = mapped_column(String(128), nullable=False)
+    role_id: Mapped[int] = fk_id_column("admin_roles.id")
+    role: Mapped["AdminRole"] = relationship(back_populates="api_keys", init=False, lazy="selectin")
+    note: Mapped[Optional[str]] = mapped_column(String(512), default=None)
+    expire_date: Mapped[Optional[dt]] = mapped_column(DateTime(timezone=True), default=None)
+    revoked_at: Mapped[Optional[dt]] = mapped_column(DateTime(timezone=True), default=None)
+    status: Mapped[APIKeyStatus] = mapped_column(
+        SQLEnum(APIKeyStatus, name="apikeystatus", create_constraint=True),
+        default=APIKeyStatus.active,
+        server_default="active",
+    )
+
+    @hybrid_property
+    def is_expired(self) -> bool:
+        """True when expire_date is set and is in the past."""
+        if self.expire_date is None:
+            return False
+        expire_at = self.expire_date if self.expire_date.tzinfo else self.expire_date.replace(tzinfo=tz.utc)
+        return expire_at <= dt.now(tz.utc)
+
+    @is_expired.expression
+    def is_expired(cls):
+        return and_(cls.expire_date.isnot(None), cls.expire_date <= func.current_timestamp())
+
+    @property
+    def is_usable(self) -> bool:
+        """False if the key is disabled, its admin is missing/disabled, or it has expired."""
+        if self.status == APIKeyStatus.disabled:
+            return False
+        if self.admin is None or self.admin.status == AdminStatus.disabled:
+            return False
+        return not self.is_expired
+
+
 class TempKey(Base):
     __tablename__ = "temp_keys"
 

app/models/admin_role.py

@@ -81,6 +81,14 @@ class HwidsPermissions(_ResourcePermissions):
     delete: RoleActionValue | None = None
 
 
+class APIKeysPermissions(_ResourcePermissions):
+    create: RoleActionValue | None = None
+    read: RoleActionValue | None = None
+    read_simple: RoleActionValue | None = None
+    modify: RoleActionValue | None = None
+    delete: RoleActionValue | None = None
+
+
 class RoleLimits(BaseModel):
     max_users: int | None = None
     data_limit_min: int | None = None
@@ -126,6 +134,7 @@ class RolePermissions(BaseModel):
     system: SystemPermissions | None = None
     hwids: HwidsPermissions | None = None
     admin_roles: CRUDPermissions | None = None
+    api_keys: APIKeysPermissions | None = None
 
     model_config = ConfigDict(from_attributes=True)