Tatta
4e2775c8c3
- server: FastAPI 后端（多屏管理、素材上传、节目编排、客户端注册/配置/心跳/崩溃上报、管理台托管） - android: Java 客户端（minSdk 23，全屏图片/视频轮播、远程配置、开机自启、崩溃上报） - web: React + Vite + antd 管理台（屏幕/素材/节目管理） - 屏幕设备 ID 关联机制、gunicorn 生产部署脚本
45 lines
1.4 KiB
Python
45 lines
1.4 KiB
Python
"""管理台认证：JWT 签发与校验（内网管理场景的轻量方案）。"""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import jwt
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
from .config import get_admin_config
|
|
|
|
ALGORITHM = "HS256"
|
|
TOKEN_TTL_HOURS = 24
|
|
|
|
_bearer = HTTPBearer(auto_error=False)
|
|
|
|
|
|
def create_token(username: str) -> str:
|
|
cfg = get_admin_config()
|
|
payload = {
|
|
"sub": username,
|
|
"exp": datetime.now(timezone.utc) + timedelta(hours=TOKEN_TTL_HOURS),
|
|
"iat": datetime.now(timezone.utc),
|
|
}
|
|
return jwt.encode(payload, cfg["token_secret"], algorithm=ALGORITHM)
|
|
|
|
|
|
def verify_credentials(username: str, password: str) -> bool:
|
|
cfg = get_admin_config()
|
|
return username == cfg.get("username") and password == cfg.get("password")
|
|
|
|
|
|
def get_current_user(
|
|
credentials: HTTPAuthorizationCredentials | None = Depends(_bearer),
|
|
) -> str:
|
|
if credentials is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="未登录")
|
|
try:
|
|
payload = jwt.decode(
|
|
credentials.credentials,
|
|
get_admin_config()["token_secret"],
|
|
algorithms=[ALGORITHM],
|
|
)
|
|
return payload["sub"]
|
|
except jwt.PyJWTError:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="登录已过期，请重新登录")
|