access tokens are being generated

pyproject.toml

@@ -16,6 +16,7 @@ python-multipart = "^0.0.9"
 python-dotenv = "^1.0.1"
 pydantic-settings = "^2.4.0"
 hypercorn = "^0.17.3"
+pyjwt = {extras = ["crypto"], version = "^2.9.0"}
 bcrypt = "^4.2.0"
 
 [tool.poetry.group.dev.dependencies]
@@ -28,6 +29,8 @@ pytest-asyncio = "^0.23.7"
 moto = {extras = ["all"], version = "^5.0.12"}
 invoke = "^2.2.0"
 rich = "^13.7.1"
+types-pyjwt = "^1.7.1"
+types-passlib = "^1.7.7.20240327"
 
 [tool.ruff]
 line-length = 120

src/smolvault/auth/decoder.py

@@ -1,23 +1,72 @@
-from typing import Annotated
+from datetime import datetime, timedelta
+from typing import Annotated, Any
+from zoneinfo import ZoneInfo
 
+import bcrypt
+import jwt
 from fastapi import Depends, HTTPException
 from fastapi.security import OAuth2PasswordBearer
+from jwt import InvalidTokenError
 
-from smolvault.auth.models import User
+from smolvault.auth.models import Token, TokenData, User
 from smolvault.clients.database import DatabaseClient, UserInfo
+from smolvault.config import get_settings
 
+settings = get_settings()
+SECRET_KEY = settings.auth_secret_key
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
 
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    return bcrypt.checkpw(plain_password.encode(), hashed_password.encode())
+
+
 def decode_token(token: str, db_client: DatabaseClient) -> UserInfo | None:
     user = db_client.get_user(token)
     return user
 
 
+def create_access_token(data: dict[str, Any], expires_delta: timedelta | None = None) -> Token:
+    to_encode = data.copy()
+    if expires_delta:  # noqa: SIM108
+        expire = datetime.now(ZoneInfo("UTC"))
+    else:
+        expire = datetime.now(ZoneInfo("UTC")) + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt: str = jwt.encode(payload=to_encode, key=SECRET_KEY, algorithm=ALGORITHM)  # type: ignore
+    return Token(access_token=encoded_jwt, token_type="bearer")  # noqa: S106
+
+
+def authenticate_user(db_client: DatabaseClient, username: str, password: str) -> UserInfo | None:
+    user = db_client.get_user(username)
+    if not user:
+        return None
+    if not bcrypt.checkpw(password.encode(), user.hashed_password.encode()):
+        return None
+    return user
+
+
 async def get_current_user(
     token: Annotated[str, Depends(oauth2_scheme)], db_client: Annotated[DatabaseClient, Depends(DatabaseClient)]
 ) -> User:
-    user = decode_token(token, db_client)
+    credentials_exception = HTTPException(
+        status_code=401,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+        token_data = TokenData(username=username)
+    except InvalidTokenError as e:
+        raise credentials_exception from e
+    if token_data.username is None:
+        raise credentials_exception
+    user = db_client.get_user(token_data.username)
     if not user:
         raise HTTPException(
             status_code=401,

src/smolvault/auth/models.py

@@ -20,3 +20,12 @@ class NewUserDTO(BaseModel):
     @cached_property
     def hashed_password(self) -> str:
         return bcrypt.hashpw(self.password.get_secret_value().encode(), bcrypt.gensalt()).decode()
+
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+
+class TokenData(BaseModel):
+    username: str | None = None

src/smolvault/config.py

@@ -9,6 +9,7 @@ class Settings(BaseSettings):
     smolvault_bucket: str
     smolvault_db: str
     smolvault_cache: str
+    auth_secret_key: str
 
     model_config = SettingsConfigDict(env_file=".env")
 

src/smolvault/main.py

@@ -6,15 +6,14 @@
 from logging.handlers import RotatingFileHandler
 from typing import Annotated
 
-import bcrypt
 from fastapi import BackgroundTasks, Depends, FastAPI, File, Form, HTTPException, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.middleware.gzip import GZipMiddleware
 from fastapi.responses import FileResponse, Response
 from fastapi.security import OAuth2PasswordRequestForm
 
-from smolvault.auth.decoder import get_current_user
-from smolvault.auth.models import NewUserDTO, User
+from smolvault.auth.decoder import authenticate_user, create_access_token, get_current_user
+from smolvault.auth.models import NewUserDTO, Token, User
 from smolvault.cache.cache_manager import CacheManager
 from smolvault.clients.aws import S3Client
 from smolvault.clients.database import DatabaseClient, FileMetadataRecord
@@ -59,20 +58,19 @@ async def create_user(
 
 
 @app.post("/token")
-async def login(
+async def login_for_access_token(
     form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
     db_client: Annotated[DatabaseClient, Depends(DatabaseClient)],
-) -> dict[str, str]:
-    user = db_client.get_user(form_data.username)
+) -> Token:
+    user = authenticate_user(db_client, form_data.username, form_data.password)
     if not user:
         raise HTTPException(
             status_code=400,
             detail="Incorrect username or password",
             headers={"WWW-Authenticate": "Bearer"},
         )
-    if bcrypt.checkpw(form_data.password.encode(), user.hashed_password.encode()) is False:
-        raise HTTPException(status_code=400, detail="Incorrect username or password")
-    return {"access_token": user.username, "token_type": "bearer"}
+    access_token = create_access_token(data={"sub": user.username})
+    return access_token
 
 
 @app.post("/file/upload")