Draft: Authentication Service Model and Views

requirements.txt

@@ -7,3 +7,6 @@ jinja2==3.1.3
 pyyaml==6.0.1
 pytest
 types-requests
+passlib
+python-jose
+authlib

src/constants.py

@@ -15,3 +15,9 @@
 
 # API
 URL_PATH = "/rest"
+
+# AUTHENTICATION :TODO - Move to a separate file
+
+SECRET_KEY = "mysecretkey"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30

src/data/database.py

@@ -1,8 +1,11 @@
 from typing import Any, Generator
 from sqlmodel import SQLModel, create_engine, Session
 from sqlalchemy_utils import database_exists, create_database
+from sqlalchemy.ext.declarative import declarative_base
 import constants
 
+Base = declarative_base()
+
 
 def create_tables(db_engine) -> None:
     from models.turnilo_dashboard import TurniloDashboard

src/exceptions/http.py

@@ -0,0 +1,24 @@
+from fastapi import HTTPException, status
+
+
+def new_credentials_exception() -> HTTPException:
+    return HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+
+def new_conflict_exception(msg: str = "Conflict with existing data") -> HTTPException:
+    return HTTPException(
+        status_code=status.HTTP_409_CONFLICT,
+        detail=msg,
+    )
+
+
+def new_user_not_found_exception() -> HTTPException:
+    return HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Incorrect email or password",
+        headers={"WWW-Authenticate": "Bearer"},
+    )

src/models/authentication.py

@@ -0,0 +1,17 @@
+from sqlalchemy import Column, Integer, String, Boolean
+
+from data.database import Base
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id = Column(Integer, primary_key=True, index=True)
+    tenant_id = Column(Integer, index=True)  # Not sure about this one
+    email = Column(String, unique=True, index=True)
+    hashed_password = Column(String)
+    is_active = Column(Boolean, default=True)
+    oauth_provider = Column(String, default=None)  # Provider used to Register/Login this one
+
+    def get_hashed_password(self) -> str:
+        return self.hashed_password.to_string()

src/routes/authentication.py

@@ -0,0 +1,57 @@
+import http
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+from sqlalchemy.orm import Session
+from pydantic import BaseModel
+from passlib.context import CryptContext
+from datetime import datetime, timedelta
+from authlib.integrations.starlette_client import OAuth
+from starlette.config import Config as StarletteConfig
+from starlette.middleware.sessions import SessionMiddleware
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+
+from data import database as db
+from exceptions import http as http_exceptions
+from src.views import authentication as auth_view
+from src.models import authentication as auth_model
+from src.services import authentication as auth_service
+
+auth_router = APIRouter()
+
+
+SECRET_KEY = "mysecretkey"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+
+config = StarletteConfig()
+oauth = OAuth(config)
+
+
+@auth_router.post("/register")
+async def register_user(user: auth_view.UserRequest, db: Session = Depends(db.get_session)):
+    db_user = auth_service.get_user_by_email(db, user.email)
+    if db_user:
+        raise http_exceptions.new_conflict_exception("User already exists")
+    await auth_service.create_user(db, user.to_model(tenant_id=1, oauth_provider="local"))
+    # TODO: Find response model
+    return {"message": "User registered successfully"}
+
+
+@auth_router.post("/token", response_model=auth_view.Token)
+async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(),
+                                 db: Session = Depends(db.get_session)):
+    user = await auth_service.get_user_by_email(db, form_data.username)
+    if not user or not auth_service.verify_password(form_data.password, user.get_hashed_password()):
+        raise http_exceptions.new_user_not_found_exception()
+    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    access_token = auth_service.create_access_token(
+        data={"sub": user.email}, expires_delta=access_token_expires
+    )
+    return {"access_token": access_token, "token_type": "bearer"}
+
+
+@auth_router.get("/protected")
+async def read_protected_route(current_user: auth_model.User = Depends(auth_service.get_current_user)):
+    return {"message": "You are authenticated!", "user": current_user.email}

src/services/authentication.py

@@ -0,0 +1,73 @@
+"""
+Authentication service definition for the DataHangar backend.
+"""
+
+from datetime import datetime, timedelta
+from jose import JWTError, jwt
+from sqlalchemy.orm import Session
+from passlib.context import CryptContext
+
+from src.exceptions import http as http_exceptions
+from src import constants
+from src.views import authentication as auth_view
+from src.models import authentication as auth_model
+import constants
+
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+def get_password_hash(password) -> str:
+    return pwd_context.hash(password)
+
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+def new_user(tenant_id: int, email: str, password: str, oauth_provider: str) -> auth_model.User:
+    return auth_model.User(
+        tenant_id=tenant_id,
+        email=email,
+        hashed_password=get_password_hash(password),
+        oauth_provider=oauth_provider
+    )
+
+
+def create_access_token(data: dict, expires_delta: timedelta) -> str:
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.now() + expires_delta
+    else:
+        expire = datetime.now() + timedelta(minutes=constants.ACCESS_TOKEN_EXPIRE_MINUTES)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, constants.SECRET_KEY, algorithm=constants.ALGORITHM)
+    return encoded_jwt
+
+
+def parse_token(token: str) -> auth_view.Token:
+    try:
+        payload = jwt.decode(token, constants.SECRET_KEY, algorithms=[constants.ALGORITHM])
+        return auth_view.Token(**payload)
+    except JWTError:
+        raise http_exceptions.new_credentials_exception()
+
+
+async def get_current_user(token: str, db: Session):
+    t = parse_token(token)
+    user = get_user_by_email(db, email=t.email)
+    if user is None:
+        raise http_exceptions.new_credentials_exception()
+    return user
+
+
+async def get_user_by_email(db: Session, email: str) -> auth_model.User:
+    return db.query(auth_model.User).filter(auth_model.User.email == email).one()
+
+
+async def create_user(db: Session, user: auth_model.User) -> auth_model.User:
+    # TODO: Maybe we could create the transaction for the session before calling this function?
+    db.add(user)
+    db.commit()
+    db.refresh(user)
+    return user

src/views/authentication.py

@@ -0,0 +1,22 @@
+from pydantic import BaseModel
+
+
+from models import authentication as auth_model
+from services import authentication as auth_service
+
+
+class UserRequest(BaseModel):
+    email: str
+    password: str
+
+    def to_model(self, tenant_id: int, oauth_provider: str) -> auth_model.User:
+        return auth_service.new_user(
+            tenant_id=tenant_id,
+            email=self.email,
+            password=self.password,
+            oauth_provider=oauth_provider,
+        )
+
+
+class Token(BaseModel):
+    email: str