Commit existing codebase

app/auth/dto.py

@@ -0,0 +1,10 @@
+from pydantic import BaseModel
+
+
+class Credentials(BaseModel):
+    username: str
+    password: str
+
+
+class TokenModel(BaseModel):
+    token: str

app/auth/handlers.py

@@ -0,0 +1,17 @@
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+
+from app.auth.dto import Credentials, TokenModel
+from app.auth.service import authenticate
+from app.db import get_db
+
+router = APIRouter()
+
+
+@router.post("/auth", status_code=200, response_model=TokenModel)
+def issue_access_token(credentials: Credentials, db: Session = Depends(get_db)) -> TokenModel:
+    token = authenticate(credentials, db)
+    if token is None:
+        raise HTTPException(status_code=401)
+    else:
+        return TokenModel(token=token)

app/auth/middleware.py

@@ -0,0 +1,17 @@
+from fastapi import Depends, HTTPException
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from jose import JWTError
+
+from app.auth.service import verify_token
+from app.user.model import User
+from app.user.service import get_user_by_id
+from app.db import get_db
+
+
+def get_auth_user(auth: HTTPAuthorizationCredentials = Depends(HTTPBearer()), db=Depends(get_db)) -> User:
+    try:
+        decoded_token = verify_token(auth.credentials)
+        auth_user = get_user_by_id(db, int(decoded_token["sub"]))
+        return auth_user
+    except JWTError:
+        raise HTTPException(status_code=403, detail="Invalid token")

app/auth/service.py

@@ -0,0 +1,39 @@
+import os
+from datetime import datetime, timedelta
+from typing import Optional
+
+from jose import jwt
+from jose.constants import ALGORITHMS
+from sqlalchemy.orm import Session
+
+from app.auth.dto import Credentials
+from app.config import config
+from app.user.service import get_user_by_username, passwords_match
+
+JWT_SECRET = config["CGNO_ID_JWT_SECRET"]
+JWT_ISSUER = "Energia"
+
+
+def authenticate(credentials: Credentials, db: Session) -> Optional[str]:
+    user = get_user_by_username(db, credentials.username)
+    if passwords_match(user.password, credentials.password):
+        token = issue_token(user.id)
+        return token
+    else:
+        return None
+
+
+def issue_token(user_id: int) -> str:
+    now = datetime.utcnow()
+    claims = {
+        "sub": str(user_id),
+        "iss": JWT_ISSUER,
+        "iat": now,
+        "nbf": now,
+        "exp": now + timedelta(weeks=1)
+    }
+    return jwt.encode(claims, JWT_SECRET, algorithm=ALGORITHMS.HS256)
+
+
+def verify_token(token: str) -> dict:
+    return jwt.decode(token, JWT_SECRET, algorithms=ALGORITHMS.HS256, issuer=JWT_ISSUER)