yangy
1564c8766d
- Modified load_dotenv() calls in multiple files to include override=True, ensuring that .env variables can replace existing environment variables. - Updated ota_storage.py to use default values for OTA upload directories, improving clarity and maintainability. - Enhanced get_ota_upload_dir() function to read environment variables dynamically, ensuring accurate directory paths based on the current environment.
65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
import os
|
|
import logging
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import jwt
|
|
from dotenv import load_dotenv
|
|
from fastapi import HTTPException, Security
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
load_dotenv(override=True)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
JWT_SECRET = os.getenv("JWT_SECRET", "dev-only-change-me-for-production")
|
|
JWT_ALGORITHM = "HS256"
|
|
TOKEN_TTL_HOURS = 12
|
|
|
|
ADMIN_USERNAME = os.getenv("DASHBOARD_ADMIN_USERNAME", "admin")
|
|
ADMIN_PASSWORD = os.getenv("DASHBOARD_ADMIN_PASSWORD", "Eafon123")
|
|
|
|
security_bearer = HTTPBearer(auto_error=False)
|
|
|
|
|
|
def create_access_token() -> str:
|
|
now = datetime.now(timezone.utc)
|
|
exp = now + timedelta(hours=TOKEN_TTL_HOURS)
|
|
payload = {
|
|
"sub": ADMIN_USERNAME,
|
|
"iat": int(now.timestamp()),
|
|
"exp": exp,
|
|
}
|
|
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
|
|
|
|
|
|
def decode_token(token: str) -> dict:
|
|
return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
|
|
|
|
|
|
def verify_credentials(username: str, password: str) -> bool:
|
|
if username != ADMIN_USERNAME:
|
|
return False
|
|
try:
|
|
return secrets.compare_digest(password, ADMIN_PASSWORD)
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def get_current_user(
|
|
credentials: HTTPAuthorizationCredentials | None = Security(security_bearer),
|
|
) -> str:
|
|
if credentials is None or (credentials.scheme or "").lower() != "bearer":
|
|
raise HTTPException(status_code=401, detail="未登录或缺少凭证")
|
|
token = credentials.credentials
|
|
try:
|
|
payload = decode_token(token)
|
|
sub = payload.get("sub")
|
|
if not sub:
|
|
raise HTTPException(status_code=401, detail="无效凭证")
|
|
return str(sub)
|
|
except jwt.ExpiredSignatureError:
|
|
raise HTTPException(status_code=401, detail="登录已过期，请重新登录")
|
|
except jwt.InvalidTokenError:
|
|
raise HTTPException(status_code=401, detail="无效凭证")
|