Pine
b6315f69b0
- park_screens 加 status/code/code_expires/bound_at、tenant_id 可空(alembic 0002)。
- auth 加 create/parse_device_token {tenant_id,device_id}。
- tenants 加 ensures_device/set_device_code/find_device_by_code/bind_device/unbind_device。
- 端点：/park/api/devices/register、/park/api/devices/{device_id}/code(8位码TTL10min)、
/park/tenants/{tid}/screens/bind-by-code(校验码→绑定→MQTT opc/display/bind/<dev> 下发 token)、
/park/api/devices/{device_id}/unbind。_resolve_tenant 支持设备 token。
- 验证(TestClient)：register→code→建园区→绑码(bound/命名)→设备token取数据返回绑定园区。
64 lines
2.4 KiB
Python
64 lines
2.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""园区租户认证 —— 大屏登录（账号密码 → 长效 tenant token，一次登录持久保持）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import datetime
|
||
import jwt
|
||
from fastapi import Header, HTTPException
|
||
|
||
from .config import settings
|
||
|
||
_SECRET = settings.ADMIN_SECRET
|
||
_ALG = "HS256"
|
||
_TTL_DAYS = 365
|
||
|
||
|
||
def create_token(tenant_id: str) -> str:
|
||
now = datetime.datetime.now(datetime.timezone.utc)
|
||
payload = {
|
||
"typ": "park",
|
||
"tid": tenant_id,
|
||
"iat": now,
|
||
"exp": now + datetime.timedelta(days=_TTL_DAYS),
|
||
}
|
||
return jwt.encode(payload, _SECRET, algorithm=_ALG)
|
||
|
||
|
||
def parse_token(token: str) -> str:
|
||
"""返回 tenant_id；非法/过期抛 401。"""
|
||
try:
|
||
payload = jwt.decode(token, _SECRET, algorithms=[_ALG])
|
||
except jwt.PyJWTError as e: # noqa: BLE001
|
||
raise HTTPException(status_code=401, detail=f"park token 无效: {e}") from e
|
||
if payload.get("typ") != "park" or not payload.get("tid"):
|
||
raise HTTPException(status_code=401, detail="park token 无效")
|
||
return payload["tid"]
|
||
|
||
|
||
def require_tenant(authorization: str | None = Header(default=None, description="Bearer <park-token>")) -> str:
|
||
"""大屏接口鉴权：从 Authorization 解析 tenant_id。"""
|
||
if not authorization or not authorization.startswith("Bearer "):
|
||
raise HTTPException(status_code=401, detail="缺少 park token")
|
||
return parse_token(authorization.split(" ", 1)[1])
|
||
|
||
|
||
# ---------------- 设备 token（大屏绑定后用，含 tenant_id + device_id） ----------------
|
||
|
||
def create_device_token(tenant_id: str, device_id: str) -> str:
|
||
now = datetime.datetime.now(datetime.timezone.utc)
|
||
payload = {"typ": "device", "tid": tenant_id, "dev": device_id,
|
||
"iat": now, "exp": now + datetime.timedelta(days=_TTL_DAYS)}
|
||
return jwt.encode(payload, _SECRET, algorithm=_ALG)
|
||
|
||
|
||
def parse_device_token(token: str) -> dict:
|
||
"""返回 {tenant_id, device_id}；非法抛 401。"""
|
||
try:
|
||
payload = jwt.decode(token, _SECRET, algorithms=[_ALG])
|
||
except jwt.PyJWTError as e: # noqa: BLE001
|
||
raise HTTPException(status_code=401, detail=f"设备 token 无效: {e}") from e
|
||
if payload.get("typ") != "device" or not payload.get("tid") or not payload.get("dev"):
|
||
raise HTTPException(status_code=401, detail="设备 token 无效")
|
||
return {"tenant_id": payload["tid"], "device_id": payload["dev"]}
|