219 lines
8.1 KiB
Python
219 lines
8.1 KiB
Python
# ============================================================
|
||
# 超级管理员授权码管理 API（moduleId: gateway-license-admin-api, 可重生 ✅）
|
||
# codes 模式（APS_LICENSE_PROVIDER=codes）下，超管在设置面板「用户与授权」
|
||
# 发放/停用/删除授权码；每个授权码即一个用户身份，数据按码隔离。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import os
|
||
import secrets
|
||
import time
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
|
||
from fastapi import APIRouter, HTTPException
|
||
from pydantic import BaseModel, Field
|
||
from sqlalchemy import select
|
||
|
||
from server.auth.context import IdentityContext, get_identity
|
||
from server.auth.licenses import ADMIN_LICENSE_CODE, LICENSE_DURATIONS, LICENSE_LABELS
|
||
from server.db.database import get_session
|
||
from server.db.models import LicenseActivation, LicenseCode
|
||
|
||
router = APIRouter(prefix="/api/admin/license", tags=["license-admin"])
|
||
|
||
# 去掉 0/O/1/I/L 等易混淆字符，便于人工抄录下发
|
||
_CODE_ALPHABET = "ABCDEFGHJKMNPQRSTUVWXYZ23456789"
|
||
|
||
|
||
def _raise(status_code: int, code: str, message: str) -> None:
|
||
raise HTTPException(status_code=status_code, detail={"code": code, "message": message})
|
||
|
||
|
||
def _require_admin() -> IdentityContext:
|
||
identity = get_identity(required=True)
|
||
if "admin" not in identity.roles:
|
||
_raise(403, "LICENSE_ADMIN_REQUIRED", "仅超级管理员可管理用户与授权码")
|
||
return identity
|
||
|
||
|
||
def _codes_enabled() -> None:
|
||
if (os.environ.get("APS_LICENSE_PROVIDER") or "").strip().lower() != "codes":
|
||
_raise(409, "LICENSE_MODE_DISABLED", "当前部署未启用授权码模式（APS_LICENSE_PROVIDER=codes）")
|
||
|
||
|
||
def _tenant() -> str:
|
||
tenant = (os.environ.get("APS_LICENSE_TENANT_UUID") or "platform").strip()
|
||
return tenant or "platform"
|
||
|
||
|
||
def _code_hash(tenant: str, code: str) -> str:
|
||
# 与 CodeLicenseProvider 的派生口径一致：codes:{tenant}:{code}
|
||
return hashlib.sha256(f"codes:{tenant}:{code}".encode()).hexdigest()
|
||
|
||
|
||
def _generate_code() -> str:
|
||
return "-".join("".join(secrets.choice(_CODE_ALPHABET) for _ in range(4)) for _ in range(4))
|
||
|
||
|
||
def _utc_text(epoch: int) -> str:
|
||
return datetime.fromtimestamp(epoch, tz=timezone.utc).isoformat()
|
||
|
||
|
||
class LicenseCodeCreateRequest(BaseModel):
|
||
name: str = Field(min_length=1, max_length=64)
|
||
mobile: str = Field(default="", max_length=32)
|
||
company: str = Field(default="", max_length=128)
|
||
licenseType: str = Field(min_length=1, max_length=16)
|
||
code: str = Field(default="", max_length=128)
|
||
expiresAt: int | None = Field(default=None)
|
||
|
||
|
||
class LicenseCodeStatusRequest(BaseModel):
|
||
status: str = Field(min_length=1, max_length=16)
|
||
|
||
|
||
def _serialize(row: LicenseCode, activation: LicenseActivation | None) -> dict:
|
||
return {
|
||
"id": row.id,
|
||
"code": row.code,
|
||
"name": row.name,
|
||
"mobile": row.mobile,
|
||
"company": row.company,
|
||
"licenseType": row.license_type,
|
||
"licenseLabel": LICENSE_LABELS.get(row.license_type, row.license_type),
|
||
"expiresAt": row.expires_at,
|
||
"status": row.status,
|
||
"createdAt": row.created_at,
|
||
"activation": None if activation is None else {
|
||
"status": activation.status,
|
||
"activatedAt": activation.activated_at,
|
||
"lastSeenAt": activation.last_seen_at,
|
||
"expiresAt": activation.expires_at,
|
||
},
|
||
}
|
||
|
||
|
||
@router.get("/codes")
|
||
async def list_license_codes() -> dict:
|
||
_codes_enabled()
|
||
_require_admin()
|
||
tenant = _tenant()
|
||
with get_session() as session:
|
||
rows = list(session.scalars(select(LicenseCode).where(
|
||
LicenseCode.tenant_uuid == tenant,
|
||
LicenseCode.deleted == 0,
|
||
).order_by(LicenseCode.created_at.desc())).all())
|
||
activations: dict[str, LicenseActivation] = {}
|
||
hashes = [row.code_hash for row in rows]
|
||
if hashes:
|
||
for act in session.scalars(select(LicenseActivation).where(
|
||
LicenseActivation.code_hash.in_(hashes),
|
||
LicenseActivation.deleted == 0,
|
||
)).all():
|
||
activations[act.code_hash] = act
|
||
return {"codes": [_serialize(row, activations.get(row.code_hash)) for row in rows]}
|
||
|
||
|
||
@router.post("/codes")
|
||
async def create_license_code(req: LicenseCodeCreateRequest) -> dict:
|
||
_codes_enabled()
|
||
identity = _require_admin()
|
||
license_type = req.licenseType.strip().lower()
|
||
now = int(time.time())
|
||
expires_at = req.expiresAt
|
||
if license_type == "custom":
|
||
if not expires_at:
|
||
_raise(400, "LICENSE_EXPIRY_REQUIRED", "自定义授权必须选择到期时间")
|
||
elif license_type not in LICENSE_DURATIONS:
|
||
_raise(400, "LICENSE_TYPE_INVALID", "授权时长类型无效")
|
||
if expires_at is not None and expires_at <= now:
|
||
_raise(400, "LICENSE_EXPIRY_PAST", "到期时间必须晚于当前时间")
|
||
tenant = _tenant()
|
||
code = req.code.strip()
|
||
with get_session() as session:
|
||
if code:
|
||
if code == ADMIN_LICENSE_CODE:
|
||
_raise(409, "LICENSE_CODE_DUPLICATED", "该授权码为内置超级管理员码")
|
||
code_hash = _code_hash(tenant, code)
|
||
clash = session.scalar(select(LicenseCode).where(
|
||
LicenseCode.code_hash == code_hash,
|
||
LicenseCode.deleted == 0,
|
||
))
|
||
if clash is not None:
|
||
_raise(409, "LICENSE_CODE_DUPLICATED", "授权码已存在")
|
||
else:
|
||
code_hash = ""
|
||
for _ in range(20):
|
||
candidate = _generate_code()
|
||
if candidate == ADMIN_LICENSE_CODE:
|
||
continue
|
||
candidate_hash = _code_hash(tenant, candidate)
|
||
clash = session.scalar(select(LicenseCode).where(
|
||
LicenseCode.code_hash == candidate_hash,
|
||
LicenseCode.deleted == 0,
|
||
))
|
||
if clash is None:
|
||
code, code_hash = candidate, candidate_hash
|
||
break
|
||
if not code:
|
||
_raise(500, "LICENSE_CODE_GEN_FAILED", "授权码生成失败，请重试")
|
||
now = int(time.time())
|
||
row = LicenseCode(
|
||
id=uuid.uuid4().hex,
|
||
code=code,
|
||
code_hash=code_hash,
|
||
name=req.name.strip(),
|
||
mobile=req.mobile.strip(),
|
||
company=req.company.strip(),
|
||
license_type=license_type,
|
||
status="active",
|
||
expires_at=expires_at,
|
||
tenant_uuid=tenant,
|
||
creator_id=identity.user_id or None,
|
||
updater_id=identity.user_id or None,
|
||
updated_at=_utc_text(now),
|
||
deleted=0,
|
||
created_at=_utc_text(now),
|
||
)
|
||
session.add(row)
|
||
session.commit()
|
||
session.refresh(row)
|
||
return {"code": _serialize(row, None)}
|
||
|
||
|
||
@router.post("/codes/{code_id}/status")
|
||
async def set_license_code_status(code_id: str, req: LicenseCodeStatusRequest) -> dict:
|
||
_codes_enabled()
|
||
identity = _require_admin()
|
||
status = req.status.strip().lower()
|
||
if status not in {"active", "disabled"}:
|
||
_raise(400, "LICENSE_STATUS_INVALID", "状态无效（active/disabled）")
|
||
now = int(time.time())
|
||
with get_session() as session:
|
||
row = session.get(LicenseCode, code_id)
|
||
if row is None or row.deleted != 0:
|
||
_raise(404, "LICENSE_CODE_NOT_FOUND", "授权码不存在")
|
||
row.status = status
|
||
row.updater_id = identity.user_id or None
|
||
row.updated_at = _utc_text(now)
|
||
session.commit()
|
||
return {"ok": True, "status": status}
|
||
|
||
|
||
@router.delete("/codes/{code_id}")
|
||
async def delete_license_code(code_id: str) -> dict:
|
||
_codes_enabled()
|
||
identity = _require_admin()
|
||
now = int(time.time())
|
||
with get_session() as session:
|
||
row = session.get(LicenseCode, code_id)
|
||
if row is None or row.deleted != 0:
|
||
_raise(404, "LICENSE_CODE_NOT_FOUND", "授权码不存在")
|
||
row.deleted = 1
|
||
row.updater_id = identity.user_id or None
|
||
row.updated_at = _utc_text(now)
|
||
session.commit()
|
||
return {"ok": True}
|