From 3f96c9c6502243b06efd4bb34a189e72e3a5557c Mon Sep 17 00:00:00 2001
From: yearning <10538594+wangweifeng1999@user.noreply.gitee.com>
Date: 星期一, 28 九月 2026 19:24:55 +0800
Subject: [PATCH] 复制项目
---
backend/app/auth.py | 54 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 54 insertions(+), 0 deletions(-)
diff --git a/backend/app/auth.py b/backend/app/auth.py
new file mode 100644
index 0000000..c3f02be
--- /dev/null
+++ b/backend/app/auth.py
@@ -0,0 +1,54 @@
+from datetime import datetime, timedelta
+from typing import Annotated
+
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+from sqlalchemy.orm import Session
+
+from app.config import settings
+from app.database import get_db
+from app.models import User
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+bearer = HTTPBearer(auto_error=False)
+
+
+def hash_password(password: str) -> str:
+ return pwd_context.hash(password)
+
+
+def verify_password(plain: str, hashed: str) -> bool:
+ return pwd_context.verify(plain, hashed)
+
+
+def create_access_token(user_id: int, role: str) -> str:
+ expire = datetime.utcnow() + timedelta(hours=settings.jwt_expire_hours)
+ payload = {"sub": str(user_id), "role": role, "exp": expire}
+ return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
+
+
+def get_current_user(
+ creds: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)],
+ db: Annotated[Session, Depends(get_db)],
+) -> User:
+ if not creds or not creds.credentials:
+ raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="鏈櫥褰�")
+ try:
+ payload = jwt.decode(
+ creds.credentials, settings.jwt_secret, algorithms=[settings.jwt_algorithm]
+ )
+ user_id = int(payload.get("sub", 0))
+ except (JWTError, ValueError):
+ raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="鏃犳晥浠ょ墝")
+ user = db.get(User, user_id)
+ if not user or not user.is_active:
+ raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="鐢ㄦ埛涓嶅彲鐢�")
+ return user
+
+
+def require_admin(user: Annotated[User, Depends(get_current_user)]) -> User:
+ if user.role != "admin":
+ raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="闇�瑕佺鐞嗗憳鏉冮檺")
+ return user
--
Gitblit v1.8.0