# from passlib.context import CryptContext # import os # from datetime import datetime, timedelta # from typing import Union, Any # from jose import jwt # ACCESS_TOKEN_EXPIRE_MINUTES = 30 # 30 minutes # REFRESH_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 7 days # ALGORITHM = "HS256" # # В предположении, что попыток взлома не будет, возьмем простейший ключ # JWT_SECRET_KEY = "secret key" # может также быть os.environ["JWT_SECRET_KEY"] # JWT_REFRESH_SECRET_KEY = "refresh secret key" # может также быть os.environ["JWT_REFRESH_SECRET_KEY"] # def get_hashed_password(password: str) -> str: # return password_context.hash(password) # def verify_password(password: str, hashed_pass: str) -> bool: # return password_context.verify(password, hashed_pass) # def create_access_token(subject: Union[str, Any], expires_delta: int = None) -> str: # if expires_delta is not None: # expires_delta = datetime.utcnow() + expires_delta # else: # expires_delta = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) # to_encode = {"exp": expires_delta, "sub": str(subject)} # encoded_jwt = jwt.encode(to_encode, JWT_SECRET_KEY, ALGORITHM) # return encoded_jwt # def create_refresh_token(subject: Union[str, Any], expires_delta: int = None) -> str: # if expires_delta is not None: # expires_delta = datetime.utcnow() + expires_delta # else: # expires_delta = datetime.utcnow() + timedelta(minutes=REFRESH_TOKEN_EXPIRE_MINUTES) # to_encode = {"exp": expires_delta, "sub": str(subject)} # encoded_jwt = jwt.encode(to_encode, JWT_REFRESH_SECRET_KEY, ALGORITHM) # return encoded_jwt from datetime import datetime, timedelta from typing import Annotated, Union from fastapi import Depends, FastAPI, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from jose import JWTError, jwt from passlib.context import CryptContext from pydantic import BaseModel # to get a string like this run: # openssl rand -hex 32 SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 # fake_users_db = { # "johndoe": { # "email": "johndoe", # "full_name": "John Doe", # "email": "johndoe@example.com", # "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW", # "disabled": False, # } # } class Token(BaseModel): access_token: str token_type: str class TokenData(BaseModel): email: Union[str, None] = None class User(BaseModel): email: str email: Union[str, None] = None # password: str # password: Union[str, None] = None full_name: Union[str, None] = None disabled: Union[bool, None] = None class UserInDB(User): hashed_password: str pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password) def get_user(db, email: str): user = None for person_with_correct_email in db: if person_with_correct_email.email == email: user = person_with_correct_email break return user #UserInDB(user_email) def authenticate_user(db, email: str, password: str): user = get_user(db, email) if not user: return False if not verify_password(password, user.hashed_password): return False return user def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) email: str = payload.get("sub") if email is None: raise credentials_exception token_data = TokenData(email=email) except JWTError: raise credentials_exception user = get_user(fake_users_db, email=token_data.email) if user is None: raise credentials_exception return user async def get_current_active_user( current_user: Annotated[User, Depends(get_current_user)] ): if current_user.disabled: raise HTTPException(status_code=400, detail="Inactive user") return current_user