Spaces:
Sleeping
Sleeping
| from datetime import datetime, timedelta | |
| from jose import JWTError, jwt | |
| from passlib.context import CryptContext | |
| from fastapi import Depends, FastAPI, HTTPException, status | |
| from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm | |
| from typing import Optional | |
| from pydantic import BaseModel | |
| # to get a string like this run: | |
| # openssl rand -hex 32 | |
| SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" | |
| ALGORITHM = "HS256" | |
| ACCESS_TOKEN_EXPIRE_MINUTES = 60 | |
| fake_users_db = { | |
| "abhisheky127": { | |
| "username": "abhisheky127", | |
| "full_name": "Abhishek Yadav", | |
| "email": "abhishek.yadav@quara.ai", | |
| "hashed_password": "$2b$12$MaEi1YGqfFyp.lf0IR6mJulnX.HaT/ZhyT4vvy8LAzRPo.tHYmOoS", | |
| "disabled": False, | |
| } | |
| } | |
| class TokenData(BaseModel): | |
| username: Optional[str] = None | |
| class Token(BaseModel): | |
| access_token: str | |
| token_type: 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, username: str): | |
| if username in db: | |
| user_dict = db[username] | |
| return user_dict | |
| def authenticate_user(fake_db, username: str, password: str): | |
| user = get_user(fake_db, username) | |
| 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: Optional[timedelta] = 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: 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]) | |
| username: str = payload.get("sub") | |
| if username is None: | |
| raise credentials_exception | |
| token_data = TokenData(username=username) | |
| except JWTError: | |
| raise credentials_exception | |
| user = get_user(fake_users_db, username=token_data.username) | |
| if user is None: | |
| raise credentials_exception | |
| return user | |