Codex
140 lines
4.0 KiB
Python
140 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
|
|
from fastapi import Depends, Header, HTTPException, Request, status
|
|
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
|
|
|
from .config import (
|
|
ACCOUNTS_AUTH_PASSWORD,
|
|
ACCOUNTS_SESSION_COOKIE,
|
|
ADMIN_AUTH_PASSWORD,
|
|
ADMIN_SESSION_COOKIE,
|
|
BASIC_AUTH_PASSWORD,
|
|
INGEST_AUTH_TOKEN,
|
|
RECORDS_SESSION_COOKIE,
|
|
)
|
|
|
|
|
|
security = HTTPBasic(auto_error=False)
|
|
|
|
|
|
def configured_password(label: str, password: str) -> str:
|
|
if not password:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=f"服务未配置{label}访问密码",
|
|
)
|
|
return password
|
|
|
|
|
|
def session_token(password: str, purpose: bytes) -> str:
|
|
return hmac.new(
|
|
password.encode("utf-8"),
|
|
purpose,
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
|
|
def has_valid_session(request: Request, cookie_name: str, password: str, purpose: bytes) -> bool:
|
|
token = request.cookies.get(cookie_name, "")
|
|
return bool(password and token) and hmac.compare_digest(token, session_token(password, purpose))
|
|
|
|
|
|
def has_valid_basic_auth(credentials: HTTPBasicCredentials | None, password: str) -> bool:
|
|
if credentials is None:
|
|
return False
|
|
return bool(password) and hmac.compare_digest(credentials.password, password)
|
|
|
|
|
|
def is_records_authenticated(
|
|
request: Request,
|
|
credentials: HTTPBasicCredentials | None = None,
|
|
) -> bool:
|
|
return has_valid_session(
|
|
request,
|
|
RECORDS_SESSION_COOKIE,
|
|
BASIC_AUTH_PASSWORD,
|
|
b"xsk-education-management-records-session-v1",
|
|
) or has_valid_basic_auth(credentials, BASIC_AUTH_PASSWORD)
|
|
|
|
|
|
def is_admin_authenticated(
|
|
request: Request,
|
|
credentials: HTTPBasicCredentials | None = None,
|
|
) -> bool:
|
|
return has_valid_session(
|
|
request,
|
|
ADMIN_SESSION_COOKIE,
|
|
ADMIN_AUTH_PASSWORD,
|
|
b"xsk-admin-web-session-v1",
|
|
) or has_valid_session(
|
|
request,
|
|
ACCOUNTS_SESSION_COOKIE,
|
|
ADMIN_AUTH_PASSWORD,
|
|
b"xsk-accounts-web-session-v1",
|
|
) or has_valid_basic_auth(credentials, ADMIN_AUTH_PASSWORD)
|
|
|
|
|
|
def is_accounts_authenticated(
|
|
request: Request,
|
|
credentials: HTTPBasicCredentials | None = None,
|
|
) -> bool:
|
|
return is_admin_authenticated(request, credentials)
|
|
|
|
|
|
def verify_records_auth(
|
|
request: Request,
|
|
credentials: HTTPBasicCredentials | None = Depends(security),
|
|
) -> str:
|
|
configured_password("课程记录", BASIC_AUTH_PASSWORD)
|
|
if not is_records_authenticated(request, credentials):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="请先登录",
|
|
)
|
|
return "records"
|
|
|
|
|
|
def verify_admin_auth(
|
|
request: Request,
|
|
credentials: HTTPBasicCredentials | None = Depends(security),
|
|
) -> str:
|
|
configured_password("管理后台", ADMIN_AUTH_PASSWORD)
|
|
if not is_admin_authenticated(request, credentials):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="请先登录管理后台",
|
|
)
|
|
return "admin"
|
|
|
|
|
|
def verify_accounts_auth(
|
|
request: Request,
|
|
credentials: HTTPBasicCredentials | None = Depends(security),
|
|
) -> str:
|
|
return verify_admin_auth(request, credentials)
|
|
|
|
|
|
def verify_any_auth(
|
|
request: Request,
|
|
credentials: HTTPBasicCredentials | None = Depends(security),
|
|
) -> str:
|
|
if is_records_authenticated(request, credentials) or is_admin_authenticated(request, credentials):
|
|
return "authenticated"
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="请先登录",
|
|
)
|
|
|
|
|
|
def verify_ingest_token(x_ingest_token: str = Header(default="")) -> str:
|
|
configured_password("课程小结推送", INGEST_AUTH_TOKEN)
|
|
if not hmac.compare_digest(x_ingest_token, INGEST_AUTH_TOKEN):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="课程小结推送 token 不正确",
|
|
)
|
|
return "ingest"
|