Cookie auth + /users/me endpoint

.env.test

@@ -1,2 +1,4 @@
-JWT_SECRET=test-secret
+ENV=test
+COOKIE_SECRET=test-secret
+COOKIE_MAX_AGE_MINUTES=10
 SQLALCHEMY_DATABASE_URL=sqlite:///:memory:

Pipfile

@@ -7,8 +7,8 @@ name = "pypi"
 fastapi = "~=0.70"
 uvicorn = "~=0.16"
 alembic = "~=1.7"
-passlib = "~=1.7"
-python-jose = "~=3.3"
+pyjwt = "*"
+passlib = "*"
 
 [dev-packages]
 Faker = "*"

src/auth.py

@@ -1,31 +1,71 @@
-from datetime import datetime, timedelta
-from typing import Optional
+import logging
+from datetime import datetime, timedelta, timezone
 
-from jose import jwt
+import jwt
+from fastapi import HTTPException, Request, Response
 from passlib.context import CryptContext
+from starlette.status import HTTP_403_FORBIDDEN
 
 from src import settings
 
+logger = logging.getLogger("bakery")
+
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
 
 def verify_password(plain_password: str, hashed_password: str) -> bool:
     return pwd_context.verify(plain_password, hashed_password)
 
 
-def create_access_token(username: str) -> str:
-    to_encode = {
+def set_cookie(response: Response, username: str):
+    created_at = datetime.now(timezone.utc)
+    expires_at = created_at + timedelta(minutes=settings.COOKIE_MAX_AGE_MINUTES)
+    token = create_token(username, created_at, expires_at)
+    response.set_cookie(
+        "token",
+        token,
+        max_age=settings.COOKIE_MAX_AGE_MINUTES * 60,
+        httponly=True,
+        samesite="strict",
+    )
+    return response
+
+
+def remove_cookie(response: Response):
+    response.set_cookie("token", "", max_age=0)
+    return response
+
+
+def verify_cookie(request: Request) -> dict:
+    token = request.cookies.get("token")
+    if token is None:
+        raise HTTPException(HTTP_403_FORBIDDEN, "Not authenticated")
+    try:
+        return decode_token(token)
+    except jwt.PyJWTError:
+        raise HTTPException(HTTP_403_FORBIDDEN, "Credentials invalid or expired")
+
+
+def create_token(username: str, created_at: datetime, expires_at: datetime) -> str:
+    logger.debug("create_token")
+    payload = {
         "sub": username,
-        "exp": datetime.utcnow() + timedelta(minutes=settings.JWT_TIMEOUT_MINUTES),
+        "iat": created_at,
+        "exp": expires_at,
     }
-    return jwt.encode(to_encode, settings.JWT_SECRET, algorithm=settings.JWT_ALGO)
+    return jwt.encode(payload, settings.COOKIE_SECRET, algorithm="HS256")
 
 
-def decode_username(token: str) -> Optional[str]:
+def decode_token(token: str) -> dict:
+    logger.debug("decode_token")
     payload = jwt.decode(
         token,
-        settings.JWT_SECRET,
-        settings.JWT_ALGO,
-        options={"require_exp": True},
+        settings.COOKIE_SECRET,
+        algorithms=["HS256"],
+        options={"require": ["sub", "iat", "exp"]},
     )
-    return payload["sub"]
+    return {
+        "username": payload["sub"],
+        "created_at": datetime.fromtimestamp(payload["iat"], timezone.utc),
+        "expires_at": datetime.fromtimestamp(payload["exp"], timezone.utc),
+    }

src/dependencies.py

@@ -1,7 +1,6 @@
-from fastapi import Depends, HTTPException
-from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
-from jose import JWTError
+from fastapi import Depends, HTTPException, Request
 from sqlalchemy.orm import Session
+from starlette.status import HTTP_403_FORBIDDEN
 
 from src import auth, crud, database, models
 
@@ -14,21 +13,12 @@ def get_db():
         db.close()
 
 
-_http_bearer = HTTPBearer()
-
-
 def get_authorized_user(
-    creds: HTTPAuthorizationCredentials = Depends(_http_bearer),
+    request: Request,
     db: Session = Depends(get_db),
 ) -> models.User:
-    try:
-        username = auth.decode_username(creds.credentials)
-    except JWTError:
-        raise HTTPException(403, "Credentials invalid or expired")
-
-    if username is not None:
-        user = crud.read_user(db, username)
-        if user is not None:
-            return user
-
-    raise HTTPException(403, "Could not validate credentials")
+    session = auth.verify_cookie(request)
+    user = crud.read_user(db, session["username"])
+    if user is None:
+        raise HTTPException(HTTP_403_FORBIDDEN, "Could not validate credentials")
+    return user

src/main.py

@@ -2,7 +2,7 @@
 import sys
 import traceback
 
-from fastapi import FastAPI, HTTPException, Request
+from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from fastapi_pagination import add_pagination
@@ -18,6 +18,8 @@
 @app.on_event("startup")
 def startup():
     logging.basicConfig(level=logging.INFO)
+    if settings.ENV == "dev":
+        logger.setLevel(logging.DEBUG)
     if logger.isEnabledFor(logging.INFO):
         pretty_settings = (f"{k.lower():<25} {v}" for k, v in sorted(settings.dump().items()))
         logger.info("Settings\n%s", "\n".join(pretty_settings))
@@ -52,9 +54,8 @@ async def catch_exceptions_middleware(request: Request, call_next):
 
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["http://localhost:3000"],
+    allow_origins=["http://local.bakery.com:3000"],
     allow_methods=["*"],
-    allow_headers=["Authorization"],
     allow_credentials=True,
 )
 app.include_router(routes.auth.router, prefix="/api")

src/routes/auth.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Response
 from pydantic import BaseModel
 from sqlalchemy.orm import Session
 
@@ -16,11 +16,19 @@ class LoginSchema(BaseModel):
     password: str
 
 
-@router.post("")
-def login(schema: LoginSchema, db: Session = Depends(get_db)):
+@router.post("/login")
+def login(schema: LoginSchema, response: Response, db: Session = Depends(get_db)):
     db_user = crud.read_user(db, schema.username)
     if db_user is not None:
         if auth.verify_password(schema.password, db_user.hashed_password):
-            token = auth.create_access_token(schema.username)
-            return {"user": {"name": db_user.name}, "token": token}
+            auth.set_cookie(response, db_user.name)
+            response.status_code = 200
+            return response
     raise HTTPException(400, "Username or password is incorrect")
+
+
+@router.get("/logout")
+def logout(response: Response):
+    auth.remove_cookie(response)
+    response.status_code = 200
+    return response