basement

auth/models.py

@@ -0,0 +1,173 @@
+from enum import Enum
+from backend.config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES
+from backend.config import pwd_context
+from datetime import datetime, timedelta, timezone
+from pydantic import BaseModel
+from fastapi import Depends, HTTPException
+from typing import Annotated
+from fastapi.security import OAuth2PasswordBearer
+import jwt
+
+class Token(BaseModel):
+    access_token : str
+    token_type : str
+
+class TokenData(BaseModel):
+    username : str | None = None
+    role : str | None = None
+    status : str | None = None
+
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
+
+#### ENUMS ####
+class Role(str, Enum):
+    user = "user"
+    admin = "admin"
+    guest = "guest"
+    mod = "mod"
+
+class Status(str, Enum):
+    active = "active"
+    banned = "banned"
+    suspended = "suspended"
+
+class User(BaseModel):
+    username : str | None = None
+    user_id : int | None = None
+    role : Role | None = None
+    status : Status | None = None
+
+
+class UserInDb(User):
+    hashed_password : str | None = None
+
+
+fake_db = {
+    "bedir": {
+        "username": "bedir",
+        "user_id": 1,
+        "hashed_password": "$2a$12$mYGWGo9c3Di3SJyYjYf3XOAsu5nP8jekf3KTItO9pbUBEm5BcapRO",  # Bcrypt örneği
+        "role": Role.user,
+        "status": Status.active,
+    },
+    "alice": {
+        "username": "alice",
+        "user_id": 2,
+        "hashed_password": "$2b$12$Alic3FakeHashedPasSw0rdxxxxxxxyyyyyyzzzzzz",
+        "role": Role.user,
+        "status": Status.suspended,
+    },
+    "adminuser": {
+        "username": "adminuser",
+        "user_id": 3,
+        "hashed_password": "$2b$12$AdminFakeHashedPasSw0rdxxxxxxxyyyyyyzzzzzz",
+        "role": Role.admin,
+        "status": Status.active,
+    }
+}
+
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    return pwd_context.verify(plain_password, hashed_password)
+
+def get_password_hash(password: str) -> str:
+    return pwd_context.hash(password)
+
+def authenticate_user(fake_db, username: str, password: str) -> UserInDb | bool:
+    print("username", username)
+    user = fake_db.get(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 : Annotated[timedelta, None] = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
+) -> str:
+    to_encode = data.copy()
+    expire = datetime.now(timezone.utc) + expires_delta
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+
+def get_user(db, username: str) -> UserInDb | None:
+    if username in db:
+        user_dict = db[username]
+        return UserInDb(**user_dict)
+    return None
+
+def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> UserInDb | None:
+    credentials_exception = HTTPException(
+        status_code=401,
+        detail="Burda bir hata var",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        token_data = TokenData(**payload)
+        username : str = token_data.username
+        if username is None:
+            raise credentials_exception
+        
+    except jwt.PyJWTError:
+        raise credentials_exception
+    
+    user = get_user(fake_db, username=username)
+    if user is None:
+        raise credentials_exception
+    
+    return user
+
+async def get_current_active_user(
+        current_user : Annotated[UserInDb, Depends(get_current_user)]
+) -> UserInDb | None:
+    
+    if current_user.status == Status.banned:
+        raise HTTPException(status_code=400, detail="Inactive user")
+    print("current_user", current_user)
+    return current_user
+
+"""
+class User(BaseModel):
+    username : str
+    name : str | None = None
+    surname : str | None = None
+    email : EmailStr | None = None
+    role : Role | None = None
+    status : Status | None = None
+    bio : str | None = None
+    created_date : datetime | None = None
+
+    collections : list[str] | None = None
+    items = list[str] | None = None
+
+class UserInDB(User):
+    hashed_password : str | None = None
+
+class UserSelfProfile(BaseModel):
+    username : str
+    name : str | None = None
+    surname : str | None = None
+    email : EmailStr | None = None
+    role : Role | None = None
+    status : Status | None = None
+    bio : str | None = None
+    created_date : datetime | None = None
+
+    collections : list[str] | None = None
+    items = list[str] | None = None
+
+class UserPublicProfile(BaseModel):
+    username : str
+    role : Role | None = None
+    bio : str | None = None
+    created_date : datetime | None = None
+    collections : list[str] | None = None
+    items = list[str] | None = None
+
+"""

auth/router.py

@@ -0,0 +1,50 @@
+from fastapi import APIRouter, Depends, HTTPException
+from .models import UserInDb, User, Role, Token
+from .models import get_current_active_user, authenticate_user, create_access_token , fake_db
+from datetime import timedelta, datetime, timezone
+from ..config import ACCESS_TOKEN_EXPIRE_MINUTES
+from typing import Annotated, Optional
+from fastapi.security import OAuth2PasswordRequestForm
+
+
+router = APIRouter(
+    prefix="/auth",
+    tags=["auth"],
+    responses={404: {"description": "Not found"}},
+    dependencies=[],
+)
+
+@router.get("/me")
+async def read_users_me(
+    current_user: Annotated[User, Depends(get_current_active_user)],
+):
+    return current_user
+
+
+def ADMIN(current_user: Annotated[UserInDb, Depends(get_current_active_user)]):
+    if current_user.role != Role.admin:
+        raise HTTPException(status_code=400, detail="You are not admin")
+    return current_user
+
+
+@router.get('/home')
+async def home(current_user : Annotated[User, Depends(ADMIN)]):
+    return {"message" : f"Welcome to home page {current_user.username}"}
+
+@router.post('/login')
+async def login_for_access_token(
+    form_data : Annotated[OAuth2PasswordRequestForm, Depends()],
+) -> Token:
+    
+    user = authenticate_user(fake_db, form_data.username, form_data.password)
+    if not user:
+        raise HTTPException(
+            status_code=400,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    access_token = create_access_token(
+        data={"sub": user['username'], "role": user['role'], 'status': user['status']}, expires_delta=access_token_expires
+    )
+    return {"access_token": access_token, "token_type": "bearer"}