Add support for login

alembic/versions/5e923e0b74d9_add_users.py

@@ -0,0 +1,36 @@
+"""Add users
+
+Revision ID: 5e923e0b74d9
+Revises: d76300ba3765
+Create Date: 2024-05-01 18:01:40.835538
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '5e923e0b74d9'
+down_revision: Union[str, None] = 'd76300ba3765'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('users',
+    sa.Column('username', sa.String(), nullable=False),
+    sa.Column('password_hash', sa.String(), nullable=False),
+    sa.Column('id', sa.UUID(), nullable=False),
+    sa.PrimaryKeyConstraint('id'),
+    sa.UniqueConstraint('username')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('users')
+    # ### end Alembic commands ###

expenses_server/db_models/__init__.py

@@ -2,5 +2,6 @@
 from .category import Category
 from .expense import Expense
 from .core import Base
+from .user import User
 
-__all__ = ["Account", "Category", "Base", "Expense"]
+__all__ = ["Account", "Category", "Base", "Expense", "User"]

expenses_server/db_models/user.py

@@ -0,0 +1,16 @@
+from uuid import UUID, uuid4
+from sqlalchemy import String, types
+from sqlalchemy.orm import MappedColumn as Mapped, mapped_column
+from .core import Base
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    username: Mapped[str] = mapped_column(String, nullable=False, unique=True)
+    password_hash: Mapped[str] = mapped_column(String, nullable=False)
+    id: Mapped[UUID] = mapped_column(
+        types.UUID,
+        default_factory=uuid4,
+        primary_key=True,
+    )

expenses_server/dtos/users.py

@@ -0,0 +1,11 @@
+from pydantic import UUID4, BaseModel
+
+
+class User(BaseModel):
+    id: UUID4
+    username: str
+
+
+class UserToken(BaseModel):
+    access_token: str
+    token_type: str

expenses_server/main.py

@@ -1,11 +1,16 @@
 from fastapi import APIRouter, FastAPI
-from fastapi.responses import RedirectResponse, FileResponse
+from fastapi.responses import FileResponse
 from fastapi.staticfiles import StaticFiles
+from passlib.context import CryptContext
+
 
 from .settings import settings
 
 from .routes.accounts import router as account_router
 from .routes.expenses import router as expense_router
+from .routes.users import router as user_router
+
+password_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
 app = FastAPI()
 
@@ -21,6 +26,7 @@ async def root() -> FileResponse:
 
 base_router.include_router(account_router)
 base_router.include_router(expense_router)
+base_router.include_router(user_router)
 
 app.include_router(base_router)
 

expenses_server/routes/expenses.py

@@ -1,5 +1,5 @@
 from http import HTTPStatus
-from typing import cast
+from typing import Annotated, cast
 from fastapi import APIRouter, Depends, HTTPException
 from pydantic import UUID4
 
@@ -10,7 +10,7 @@
 from ..dtos.expenses import ExpenseDTO, ExpenseCreate, ExpenseUpdate
 from ..db_models.expense import Expense as DBExpense
 from sqlalchemy.orm import Session
-
+from ..security import oauth2_scheme
 
 router = APIRouter(prefix="/expenses")
 
@@ -38,7 +38,9 @@ async def create_expense(
 
 
 @router.get("", response_model=list[ExpenseDTO])
-async def get_all_expenses(session: Session = Depends(get_db)) -> list[ExpenseDTO]:
+async def get_all_expenses(
+    token: Annotated[str, Depends(oauth2_scheme)], session: Session = Depends(get_db)
+) -> list[ExpenseDTO]:
     db_expenses = session.query(DBExpense).order_by(DBExpense.timestamp.desc()).all()
     expenses = [ExpenseDTO.model_validate(expense) for expense in db_expenses]
     return expenses

expenses_server/routes/users.py

@@ -0,0 +1,81 @@
+from http import HTTPStatus
+from typing import Annotated
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.security import OAuth2PasswordRequestForm
+import jwt
+from sqlalchemy.orm import Session
+from sqlalchemy.exc import NoResultFound
+
+from expenses_server.dtos.users import UserToken
+from expenses_server.settings import settings
+from expenses_server.db import get_db
+from expenses_server.db_models.user import User
+from expenses_server.security import create_access_token
+from expenses_server.utils import verify_password
+from expenses_server.security import oauth2_scheme
+
+router = APIRouter(prefix="/users")
+
+
+async def get_current_user(
+    token: Annotated[str, Depends(oauth2_scheme)], db_session: Session = Depends(get_db)
+) -> User:
+    credentials_exception = HTTPException(
+        status_code=HTTPStatus.UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(
+            token, settings.jwt_encode_key, algorithms=[settings.jwt_algorithm]
+        )
+        username: str | None = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+    except jwt.InvalidTokenError:
+        raise credentials_exception
+    try:
+        user = (
+            db_session.query(User)
+            .where(
+                User.username == username,
+            )
+            .one()
+        )
+    except NoResultFound:
+        raise credentials_exception
+    return user
+
+
+@router.get("/me", response_model=User)  # TODO: No need to send password_hash
+async def read_user_current(
+    current_user: Annotated[User, Depends(get_current_user)],
+) -> User:
+    # TODO: Only accounts for user should be sent
+    return current_user
+
+
+@router.post("/token")
+async def login(
+    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
+    db_session: Session = Depends(get_db),
+) -> UserToken:
+    try:
+        user = (
+            db_session.query(User)
+            .where(
+                User.username == form_data.username,
+            )
+            .one()
+        )
+        if not verify_password(form_data.password, user.password_hash):
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST,
+                detail="Incorrect username or password",
+            )
+        access_token = create_access_token(data={"sub": user.username})
+        return UserToken(access_token=access_token, token_type="bearer")
+    except NoResultFound:
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST, detail="Incorrect username or password"
+        )

expenses_server/security.py

@@ -0,0 +1,22 @@
+from typing import Any
+from fastapi.security import OAuth2PasswordBearer
+from datetime import datetime, timedelta, timezone
+import jwt
+
+from expenses_server.settings import settings
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/users/token")
+
+
+def create_access_token(
+    data: dict[str, Any],  # TODO: Make data's type def more strict
+) -> str:
+    to_encode = data.copy()
+    expire = datetime.now(timezone.utc) + timedelta(
+        minutes=settings.jwt_token_expire_minutes
+    )
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(
+        to_encode, settings.jwt_encode_key, algorithm=settings.jwt_algorithm
+    )
+    return encoded_jwt

expenses_server/settings.py

@@ -9,5 +9,13 @@ class Settings(BaseSettings):
     )
     frontend_path: str | None = None  # "../expenses-react/build"
 
+    # Password
+    hash_algorithm: str = "sha256"
+    password_seed: str = "dummy_seed"
+
+    jwt_encode_key: str = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"  # TODO: Generate correctly
+    jwt_algorithm: str = "HS256"
+    jwt_token_expire_minutes: int = 1  # TODO: Longer expiration time
+
 
 settings = Settings()

expenses_server/utils.py

@@ -0,0 +1,22 @@
+from getpass import getpass
+
+from expenses_server.db import get_db
+from expenses_server.db_models.user import User
+from expenses_server import main
+
+
+def hash_password(password: str) -> str:
+    return main.password_context.hash(password)
+
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    return main.password_context.verify(plain_password, hashed_password)
+
+
+def create_user(username: str) -> None:
+    password = getpass()
+    password_hash = main.password_context.hash(password)
+    user = User(username=username, password_hash=password_hash)
+    session = next(get_db())
+    session.add(user)
+    session.commit()