Enhance backend functionality and frontend UI
- Updated main.py to include authentication for brand, model, and OTA routers. - Added new OTA schemas in schemas.py for version management. - Enhanced model retrieval with sorting options in models.py. - Improved model update functionality to support multipart/form-data uploads. - Updated frontend layout and styles for a more modern look, including new font integration. - Implemented login route and authentication checks in router/index.js. - Added sorting capabilities in model table and improved file handling in model view. - Updated requirements.txt to include PyJWT for token management.
This commit is contained in:
@@ -0,0 +1,64 @@
|
||||
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()
|
||||
|
||||
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="无效凭证")
|
||||
Reference in New Issue
Block a user