1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
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)
|
