dekun
da5eb4c18c
Strategy nodes gain fleet token APIs; control/ app for local ops; manage.sh offers strategy vs control one-click deploy. Co-authored-by: Cursor <cursoragent@cursor.com>
78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""中控登录 HMAC Token。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
from .config import ControlSettings, get_control_settings
|
|
|
|
_bearer = HTTPBearer(auto_error=False)
|
|
|
|
|
|
def _b64url(data: bytes) -> str:
|
|
return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=")
|
|
|
|
|
|
def _b64url_decode(s: str) -> bytes:
|
|
pad = "=" * (-len(s) % 4)
|
|
return base64.urlsafe_b64decode(s + pad)
|
|
|
|
|
|
def issue_token(username: str, settings: ControlSettings) -> tuple[str, int]:
|
|
exp = int(time.time()) + int(settings.control_token_ttl_sec)
|
|
payload = {
|
|
"u": username,
|
|
"exp": exp,
|
|
"v": int(settings.control_auth_token_version),
|
|
}
|
|
raw = _b64url(json.dumps(payload, separators=(",", ":")).encode("utf-8"))
|
|
sig = hmac.new(
|
|
settings.control_auth_secret.encode("utf-8"),
|
|
raw.encode("ascii"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
return f"{raw}.{sig}", settings.control_token_ttl_sec
|
|
|
|
|
|
def verify_token(token: str, settings: ControlSettings) -> str:
|
|
try:
|
|
raw, sig = token.rsplit(".", 1)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=401, detail="invalid token") from e
|
|
expect = hmac.new(
|
|
settings.control_auth_secret.encode("utf-8"),
|
|
raw.encode("ascii"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
if not hmac.compare_digest(expect, sig):
|
|
raise HTTPException(status_code=401, detail="invalid token")
|
|
try:
|
|
payload = json.loads(_b64url_decode(raw))
|
|
except Exception as e:
|
|
raise HTTPException(status_code=401, detail="invalid token") from e
|
|
if int(payload.get("exp") or 0) < int(time.time()):
|
|
raise HTTPException(status_code=401, detail="token expired")
|
|
if int(payload.get("v") or 0) != int(settings.control_auth_token_version):
|
|
raise HTTPException(status_code=401, detail="token revoked")
|
|
username = str(payload.get("u") or "")
|
|
if not username:
|
|
raise HTTPException(status_code=401, detail="invalid token")
|
|
return username
|
|
|
|
|
|
def require_control_user(
|
|
creds: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)],
|
|
settings: Annotated[ControlSettings, Depends(get_control_settings)],
|
|
) -> str:
|
|
if creds is None or not creds.credentials:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="login required")
|
|
return verify_token(creds.credentials, settings)
|