auth stuff with separate permissions

Author: ImShyMike

api/auth/__init__.py

@@ -4,31 +4,25 @@
     SessionClientRequest,
     OtpClientResponse,
     generate_session_id,
-    is_user_admin,
     is_user_authenticated,
-    is_user_reviewer,
     refresh_token,
-    require_admin,
     require_auth,
-    require_reviewer,
     router,
     send_otp,
     validate_otp,
+    permission_dependency,
 )
 
 __all__ = [
     "OtpClientRequest",
     "SessionClientRequest",
     "OtpClientResponse",
     "generate_session_id",
-    "is_user_admin",
     "is_user_authenticated",
-    "is_user_reviewer",
     "refresh_token",
-    "require_admin",
     "require_auth",
-    "require_reviewer",
     "router",
     "send_otp",
     "validate_otp",
+    "permission_dependency",
 ]

api/auth/main.py

@@ -4,6 +4,7 @@
 import secrets
 from datetime import datetime, timedelta, timezone
 from email.message import EmailMessage
+from enum import Enum
 from functools import wraps
 
 import aiosmtplib
@@ -18,6 +19,7 @@
 from fastapi.responses import RedirectResponse
 from pydantic import BaseModel, field_validator
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select
 
 from db import get_db
 from models.user import User
@@ -28,6 +30,12 @@
 r = redis.Redis(password=os.getenv("REDIS_PASSWORD", ""), host=HOST)
 
 
+class Permission(Enum):
+    """User permissions"""
+
+    ADMIN = 0
+
+
 class OtpClientRequest(BaseModel):
     """OTP send request from client"""
 
@@ -115,44 +123,51 @@ async def wrapper(request: Request, *args, **kwargs):
 #     return wrapper
 
 
-# @decorator
-def require_admin(func):
-    """Require admin status"""
-
-    async def wrapper(*args, request: Request, **kwargs):
-        if not await is_user_authenticated(request) or not await is_user_admin():
-            return RedirectResponse("/login", status_code=418)
-        elif await is_user_authenticated(request) and not await is_user_admin():
-            return RedirectResponse("/home", status_code=403)
-        return await func(request, *args, **kwargs)
+def permission_dependency(permission: Permission):
+    """Permission dependency"""
 
-    return wrapper
+    async def verifier(
+        request: Request,
+        session: AsyncSession = Depends(get_db),
+    ):
+        payload = await is_user_authenticated(request)
+        result = await session.execute(select(User).where(User.email == payload["sub"]))
+        user = result.scalar_one_or_none()
+        if not user:
+            raise HTTPException(status_code=401)
+        if permission.value not in (user.permissions or []):
+            raise HTTPException(status_code=403)
+        request.state.user = user
+
+    return verifier
 
 
 # @decorator
-def require_reviewer(func):
-    """Require reviewer status"""
-
-    async def wrapper(*args, request: Request, **kwargs):
-        if not await is_user_authenticated(request) or not await is_user_reviewer():
-            return RedirectResponse("/login", status_code=418)
-        elif await is_user_authenticated(request) and not await is_user_reviewer():
-            return RedirectResponse("/home", status_code=403)
-        return await func(request, *args, **kwargs)
+# def require_admin(func):
+#     """Require admin status"""
 
-    return wrapper
+#     async def wrapper(*args, request: Request, **kwargs):
+#         if not await is_user_authenticated(request) or not await is_user_admin():
+#             return RedirectResponse("/login", status_code=418)
+#         elif await is_user_authenticated(request) and not await is_user_admin():
+#             return RedirectResponse("/home", status_code=403)
+#         return await func(request, *args, **kwargs)
 
+#     return wrapper
 
-async def is_user_admin() -> bool:
-    """Check if user is an admin"""
-    # TODO: admin check
-    return False
 
+# @decorator
+# def require_reviewer(func):
+#     """Require reviewer status"""
 
-async def is_user_reviewer() -> bool:
-    """Check if user is a reviewer"""
-    # TODO: reviewer check
-    return False
+#     async def wrapper(*args, request: Request, **kwargs):
+#         if not await is_user_authenticated(request) or not await is_user_reviewer():
+#             return RedirectResponse("/login", status_code=418)
+#         elif await is_user_authenticated(request) and not await is_user_reviewer():
+#             return RedirectResponse("/home", status_code=403)
+#         return await func(request, *args, **kwargs)
+
+#     return wrapper
 
 
 async def is_user_authenticated(request: Request) -> dict:

api/projects/main.py

@@ -86,6 +86,7 @@ async def create_project(
 
     return {"success": True}
 
+
 # async def run():
 #     conn = await asyncpg.connect(user='user', password='password',
 #                                  database='database', host='127.0.0.1')

api/users/main.py

@@ -6,7 +6,7 @@
 # import asyncpg
 # import orjson
 # import sqlalchemy
-from fastapi import APIRouter # , Depends, HTTPException, Request
+from fastapi import APIRouter  # , Depends, HTTPException, Request
 # from pydantic import BaseModel
 # from sqlalchemy.ext.asyncio import AsyncSession
 # from sqlalchemy.orm import selectinload
@@ -36,6 +36,7 @@ async def get_user():
     """Get user details"""
     # TODO: implement get user functionality
 
+
 # @protect
 async def delete_user():  # can only delete their own user!!! don't let them delete other users!!!
     """Delete a user account"""

main.py

@@ -8,7 +8,7 @@
 # import orjson
 # import os
 import dotenv
-from fastapi import FastAPI, HTTPException, Request  # , Form, Depends
+from fastapi import Depends, FastAPI, HTTPException, Request  # , Form
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import FileResponse, HTMLResponse  # , RedirectResponse
 from fastapi.staticfiles import StaticFiles
@@ -25,8 +25,9 @@
 # from sqlalchemy.ext.asyncio import async_sessionmaker
 from api.auth import require_auth  # , is_user_authenticated
 from api.auth import router as auth_router
-from api.users import router as users_router
+from api.auth.main import Permission, permission_dependency
 from api.projects import router as projects_router
+from api.users import router as users_router
 from db import engine  # , get_db
 from models.user import Base
 
@@ -77,6 +78,8 @@ async def validation_exception_handler(request: Request, exc: RequestValidationE
         status_code=400,
         detail={"errors": exc.errors(), "body": exc.body},
     )
+
+
 app.include_router(auth_router)
 app.include_router(users_router)
 app.include_router(projects_router)
@@ -103,7 +106,7 @@ async def protected_route(request: Request):
     """Protected route example"""
     user_email = request.state.user["sub"]
     return HTMLResponse(
-        f"<h1>Hello World! This is authenticated! Your email is {user_email}! <br>" \
+        f"<h1>Hello World! This is authenticated! Your email is {user_email}! <br>"
         f"Your full string should be {request.state.user}</h1>"
     )
 
@@ -120,6 +123,14 @@ async def serve_projects_test(_request: Request):
     return FileResponse("static/projectstest.html")
 
 
+@app.get("/admin")
+async def serve_admin(
+    _request: Request, _permission=Depends(permission_dependency(Permission.ADMIN))
+):
+    """Admin page"""
+    return "test"
+
+
 # @app.post("/login")
 # async def handle_login(email: Annotated[str, Form()], otp: Annotated[int, Form()]):
 #     pass

models/user.py

@@ -1,7 +1,18 @@
 """User database models"""
+
 from datetime import datetime, timezone
 
-from sqlalchemy import JSON, Column, DateTime, Float, ForeignKey, Integer, String
+from sqlalchemy import (
+    ARRAY,
+    JSON,
+    Column,
+    DateTime,
+    Float,
+    ForeignKey,
+    Integer,
+    SmallInteger,
+    String,
+)
 from sqlalchemy.orm import declarative_base, relationship
 
 Base = declarative_base()
@@ -12,7 +23,9 @@ class User(Base):
 
     __tablename__ = "users"
 
-    email = Column(String, primary_key=True)
+    id = Column(Integer, autoincrement=True, primary_key=True, unique=True)
+    email = Column(String, nullable=False, unique=True)
+    permissions = Column(ARRAY(SmallInteger), nullable=False, server_default="{}")
     projects = relationship(
         "UserProject", back_populates="user", cascade="all, delete-orphan"
     )