Added basic oauth flow

1 files changed, 87 insertions, 0 deletions

diff --git a/authentication/endpoints.py b/authentication/endpoints.py
new file mode 100644
index 0000000..ebdc7a1
--- /dev/null
+++ b/authentication/endpoints.py
@@ -0,0 +1,87 @@
+from typing import Annotated
+from uuid import uuid7
+
+from fastapi import FastAPI, Request, Form, Query, HTTPException
+from pydantic import BaseModel, EmailStr, ValidationError
+from starlette.middleware.authentication import AuthenticationMiddleware
+from starlette.middleware.sessions import SessionMiddleware
+from starlette.responses import HTMLResponse, RedirectResponse
+from starlette.templating import Jinja2Templates
+
+import conf
+from authentication.backend import AuthBackend
+from authentication.utils import bcrypt_sha256_verify
+from db.models import User, Session, ClientApplication
+from db.session import get_session_context
+from proxy.utils import get_path_with_query_string
+
+authentication = FastAPI()
+authentication.add_middleware(SessionMiddleware, secret_key=conf.SECRET_KEY)
+
+oauth = FastAPI()
+oauth.add_middleware(AuthenticationMiddleware, backend=AuthBackend())
+oauth.add_middleware(SessionMiddleware, secret_key=conf.SECRET_KEY)
+
+templates = Jinja2Templates(directory='authentication/templates')
+
+class LoginForm(BaseModel):
+    username: EmailStr
+    password: str
+
+async def base_login(request: Request, form: LoginForm | None = None, errors: dict[str, str] | None = None):
+    return templates.TemplateResponse(request, 'login.html', context={
+        'form': form,
+        'errors': errors,
+    })
+
+@authentication.get('/login/', response_class=HTMLResponse)
+async def login(request: Request):
+    return await base_login(request)
+
+@authentication.post('/login/', response_class=HTMLResponse)
+async def login(request: Request, username: Annotated[str, Form()] = '', password: Annotated[str, Form()] = '', redirect: Annotated[str, Query()] = None):
+    form = None
+    errors = {}
+
+    try:
+        form = LoginForm.model_validate({
+            'username': username,
+            'password': password
+        })
+        with get_session_context() as db_session:
+            user = db_session.query(User).filter(User.username == form.username).first()
+
+        if not bcrypt_sha256_verify(password, user.password if user is not None else '') or user is None:
+            errors.update({'password': 'Invalid username or password'})
+
+        with get_session_context() as db_session:
+            session = Session(id=uuid7(), user=user)
+            db_session.add(session)
+
+        if redirect is not None:
+            return RedirectResponse(redirect)
+
+    except ValidationError as exc:
+        errors.update({ e['loc'][0]: e['msg'] for e in exc.errors() })
+
+    return await base_login(request, form, errors)
+
+
+async def base_authorize(request: Request, client_id: str, redirect_uri: str):
+    with get_session_context() as session:
+        app = session.query(ClientApplication).where(ClientApplication.client_id == client_id).first()
+
+    if app is None:
+        raise HTTPException(status_code=404, detail='Invalid client id')
+
+    if not request.user.is_authenticated:
+        return RedirectResponse(f'/auth/login/?redirect={get_path_with_query_string(request)}')
+
+
+@oauth.get('/authorize/')
+async def authorize(request: Request, client_id: Annotated[str, Query()], redirect_uri: Annotated[str, Query()]):
+    return await base_authorize(request, client_id, redirect_uri)
+
+@oauth.post('/authorize/')
+async def authorize(request: Request, client_id: Annotated[str, Query()], redirect_uri: Annotated[str, Query()]):
+    return await base_authorize(request, client_id, redirect_uri)