From f28781dd1dd718f45bf1cef82d1ce6eb4792b4ca Mon Sep 17 00:00:00 2001 From: Tom van der Lee Date: Wed, 22 Jul 2026 17:37:10 +0200 Subject: Added management interface --- authentication/endpoints.py | 99 +++++++++++++++++++++++++++++----- authentication/templates/base.html | 65 ++-------------------- authentication/templates/register.html | 24 +++++++++ 3 files changed, 114 insertions(+), 74 deletions(-) create mode 100644 authentication/templates/register.html (limited to 'authentication') diff --git a/authentication/endpoints.py b/authentication/endpoints.py index f94d55b..18f3ba6 100644 --- a/authentication/endpoints.py +++ b/authentication/endpoints.py @@ -1,12 +1,11 @@ import json from base64 import b64encode from functools import partial -from typing import Annotated +from typing import Annotated, Self from uuid import uuid7 -from alembic.testing import requirements from fastapi import FastAPI, Request, Form, Query -from pydantic import BaseModel, ValidationError +from pydantic import BaseModel, ValidationError, model_validator, EmailStr from sqlmodel import select from starlette.authentication import requires from starlette.middleware.authentication import AuthenticationMiddleware @@ -31,24 +30,32 @@ oauth.add_middleware(SessionMiddleware, secret_key=conf.SECRET_KEY) auth_server = AuthorizationServer() code_grant = partial(AuthorizationCodeGrant, server=auth_server) -templates = Jinja2Templates(directory='authentication/templates') +templates = Jinja2Templates(directory=[ + 'templates', + 'authentication/templates' +]) + class LoginForm(BaseModel): username: str 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 get_login(request: Request, next: Annotated[str, Query()] = None): return await base_login(request) + @authentication.post('/login/', response_class=HTMLResponse) -async def post_login(request: Request, username: Annotated[str, Form()] = '', password: Annotated[str, Form()] = '', next: Annotated[str, Query()] = None): +async def post_login(request: Request, username: Annotated[str, Form()] = '', password: Annotated[str, Form()] = '', + next: Annotated[str, Query()] = None): form = None errors = {} @@ -70,35 +77,101 @@ async def post_login(request: Request, username: Annotated[str, Form()] = '', pa request.session.update({'id': str(session.id)}) if next is not None: - return RedirectResponse(next) + return RedirectResponse(next, status_code=303) except ValidationError as exc: - errors.update({ e['loc'][0]: e['msg'] for e in exc.errors() }) + 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, user: User | None = None): - if not conf.IS_LOCAL: + +@authentication.get('/logout/') +def logout(request: Request): + with get_session_context() as db_session: + db_session.query(Session).filter(Session.id == request.session.get('id')).delete() + + request.session.clear() + return RedirectResponse(request.headers.get('Referer', '/management/dashboard/'), status_code=303) + + +class RegisterForm(BaseModel): + email: EmailStr + password: str + verify_password: str + + @model_validator(mode='after') + def check_passwords_match(self) -> Self: + if self.password != self.verify_password: + raise ValueError('Passwords do not match') + return self + + +async def base_register(request: Request, form: RegisterForm | None = None, errors: dict[str, str] | None = None): + return templates.TemplateResponse(request, 'register.html', context={ + 'form': form, + 'errors': errors, + }) + + +@authentication.get('/register/', response_class=HTMLResponse) +async def get_register(request: Request, next: Annotated[str, Query()] = None): + return await base_register(request) + + +@authentication.post('/register/', response_class=HTMLResponse) +async def post_register(request: Request, email: Annotated[str, Form()] = '', password: Annotated[str, Form()] = '', + verify_password: Annotated[str, Form()] = '', next: Annotated[str, Query()] = None): + form = None + errors = {} + + try: + form = RegisterForm.model_validate({ + 'email': email, + 'password': password, + 'verify_password': verify_password + }) + with get_session_context() as db_session: - db_session.query(Session).filter(Session.id == request.session.get('id')).delete() + user = User(username=email, email=email) + user.set_password(password) + db_session.add(user) + + session = Session(id=uuid7(), user=user) + db_session.add(session) + + request.session.update({'id': str(session.id)}) + + if next is not None: + return RedirectResponse(next, status_code=303) + except ValidationError as exc: + errors.update({e['loc'][0]: e['msg'] for e in exc.errors()}) + + return await base_register(request, form, errors) + + +async def base_authorize(request: Request, user: User | None = None): return auth_server.create_authorization_response(request, request.user, grant=None) + @oauth.get('/authorize/') @requires(['authenticated', 'session'], redirect='get_login') async def get_authorize(request: Request): return await base_authorize(request, request.user) + @oauth.post('/authorize/') @requires(['authenticated', 'session'], redirect='get_login') async def post_authorize(request: Request): return await base_authorize(request, request.user) + @oauth.post('/token/') async def post_token(request: Request): await prepare_oauth_request(request) return auth_server.create_token_response(request) + @authentication.get('/connect/') async def connect(request: Request): origin = request.url.scheme + '://' + request.url.netloc @@ -107,16 +180,18 @@ async def connect(request: Request): app = session.exec(select(ClientApplication)).first() return RedirectResponse( - '/oauth' + get_url_path_with_query(oauth, 'get_authorize', { + '/oauth' + get_url_path_with_query(oauth, 'get_authorize', { 'client_id': app.client_id, 'redirect_uri': f'{origin}/oauth' + get_url_path_with_query(oauth, 'callback'), 'response_type': 'code', 'state': b64encode(json.dumps({ 'client_id': app.client_id, }).encode()).decode(), - }) + }), + status_code=303, ) + @oauth.get('/config/') async def get_config(request: Request): with get_session_context() as session: diff --git a/authentication/templates/base.html b/authentication/templates/base.html index 47d007e..9874313 100644 --- a/authentication/templates/base.html +++ b/authentication/templates/base.html @@ -1,63 +1,4 @@ - - - - - TTUN - - - -
- -
+{% extends 'global.html' %} +{% block template %} {% block content %}{% endblock %} - - +{% endblock %} diff --git a/authentication/templates/register.html b/authentication/templates/register.html new file mode 100644 index 0000000..576db78 --- /dev/null +++ b/authentication/templates/register.html @@ -0,0 +1,24 @@ +{% extends "./base.html" %} + +{% block content %} +
+ + + {% if errors and errors.email%} + {{ errors.email }} + {% endif %} + + + + {% if errors and errors.password%} + {{ errors.password }} + {% endif %} + + + {% if errors and errors.password%} + {{ errors.verify_password }} + {% endif %} + + +
+{% endblock %} -- cgit v1.2.3