156 lines
5.0 KiB
Python
156 lines
5.0 KiB
Python
"""Expiring client credentials for the protected gateway interfaces."""
|
|
|
|
import base64
|
|
import binascii
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import secrets
|
|
import time
|
|
from uuid import uuid4
|
|
|
|
from fastapi import APIRouter, Depends, Security
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from pydantic import BaseModel, Field
|
|
|
|
from gateway.config import get_config
|
|
|
|
ACCESS_TTL = 15 * 60
|
|
REFRESH_TTL = 7 * 24 * 60 * 60
|
|
_USERNAME = "yonghe"
|
|
_PASSWORD = "yonghe1999"
|
|
|
|
router = APIRouter(prefix="/api/v1/auth", tags=["客户端鉴权"])
|
|
_bearer = HTTPBearer(auto_error=False, scheme_name="ClientBearer")
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
username: str = Field(max_length=128)
|
|
password: str = Field(max_length=256)
|
|
|
|
|
|
class RefreshRequest(BaseModel):
|
|
refresh_token: str = Field(max_length=1024)
|
|
|
|
|
|
def _response(code, message, *, status=200, data=None, challenge=False):
|
|
headers = {"Cache-Control": "no-store"}
|
|
if challenge:
|
|
headers["WWW-Authenticate"] = "Bearer"
|
|
return JSONResponse(status_code=status, headers=headers, content={
|
|
"code": code, "message": message, "request_id": f"gw-{uuid4().hex[:8]}", "data": data,
|
|
})
|
|
|
|
|
|
def _secret():
|
|
value = os.getenv("HAIR_API_TOKEN_SECRET")
|
|
if value is None:
|
|
try:
|
|
value = get_config().get("api_token_secret")
|
|
except FileNotFoundError:
|
|
return None
|
|
if not isinstance(value, str) or len(value.encode("utf-8")) < 32:
|
|
return None
|
|
return value.encode("utf-8")
|
|
|
|
|
|
def _encode(raw):
|
|
return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
|
|
|
|
|
|
def _decode(value):
|
|
return base64.b64decode(value + "=" * (-len(value) % 4), altchars=b"-_", validate=True)
|
|
|
|
|
|
def _issue(secret, kind, ttl):
|
|
payload = _encode(json.dumps({
|
|
"sub": _USERNAME,
|
|
"kind": kind,
|
|
"exp": int(time.time()) + ttl,
|
|
"nonce": secrets.token_urlsafe(16),
|
|
}, separators=(",", ":")).encode())
|
|
signature = hmac.digest(secret, payload.encode("ascii"), hashlib.sha256)
|
|
return payload + "." + _encode(signature)
|
|
|
|
|
|
def _valid(token, secret, kind):
|
|
if not token or len(token) > 1024:
|
|
return False
|
|
try:
|
|
payload, signature = token.split(".")
|
|
expected = _encode(hmac.digest(secret, payload.encode("ascii"), hashlib.sha256))
|
|
if not hmac.compare_digest(signature.encode("ascii"), expected.encode("ascii")):
|
|
return False
|
|
claims = json.loads(_decode(payload))
|
|
return (
|
|
isinstance(claims, dict)
|
|
and claims.get("sub") == _USERNAME
|
|
and claims.get("kind") == kind
|
|
and type(claims.get("exp")) is int
|
|
and time.time() < claims["exp"]
|
|
)
|
|
except (ValueError, UnicodeError, binascii.Error):
|
|
return False
|
|
|
|
|
|
def _unconfigured():
|
|
return _response(1007, "客户端令牌服务未配置", status=503)
|
|
|
|
|
|
def _unauthorized(message):
|
|
return _response(1009, message, status=401, challenge=True)
|
|
|
|
|
|
def _access_data(secret):
|
|
return {
|
|
"access_token": _issue(secret, "access", ACCESS_TTL),
|
|
"token_type": "Bearer",
|
|
"expires_in": ACCESS_TTL,
|
|
}
|
|
|
|
|
|
@router.post("/login", summary="用户名和密码换取访问及刷新令牌")
|
|
async def login(body: LoginRequest):
|
|
username_ok = hmac.compare_digest(body.username.encode("utf-8"), _USERNAME.encode("utf-8"))
|
|
password_ok = hmac.compare_digest(body.password.encode("utf-8"), _PASSWORD.encode("utf-8"))
|
|
if not (username_ok and password_ok):
|
|
return _unauthorized("用户名或密码错误")
|
|
secret = _secret()
|
|
if secret is None:
|
|
return _unconfigured()
|
|
data = _access_data(secret)
|
|
data.update(refresh_token=_issue(secret, "refresh", REFRESH_TTL), refresh_expires_in=REFRESH_TTL)
|
|
return _response(0, "success", data=data)
|
|
|
|
|
|
@router.post("/refresh", summary="刷新访问令牌")
|
|
async def refresh(body: RefreshRequest):
|
|
secret = _secret()
|
|
if secret is None:
|
|
return _unconfigured()
|
|
if not _valid(body.refresh_token, secret, "refresh"):
|
|
return _unauthorized("刷新令牌无效或已过期，请重新登录")
|
|
return _response(0, "success", data=_access_data(secret))
|
|
|
|
|
|
async def client_auth_error(credentials: HTTPAuthorizationCredentials | None = Security(_bearer)):
|
|
"""Return an error response for protected handlers before local work or dispatch."""
|
|
if credentials is None:
|
|
return _unauthorized("请提供 Authorization: Bearer 访问令牌")
|
|
secret = _secret()
|
|
if secret is None:
|
|
return _unconfigured()
|
|
if not _valid(credentials.credentials, secret, "access"):
|
|
return _unauthorized("访问令牌无效或已过期")
|
|
return None
|
|
|
|
|
|
@router.get("/check", summary="验证当前访问令牌")
|
|
async def check(auth_error=Depends(client_auth_error)):
|
|
"""检查与 42、52、72 相同的 Bearer 令牌，无需提交图片。"""
|
|
if auth_error is not None:
|
|
return auth_error
|
|
return _response(0, "访问令牌有效")
|