File size: 1,852 Bytes
56cce61
2bdcb14
 
 
7fc9f8d
ac39a48
7fc9f8d
2bdcb14
 
 
168aee7
c14732f
2bdcb14
 
 
 
987a1bc
 
 
 
 
 
 
 
 
 
 
 
2bdcb14
 
 
 
 
 
987a1bc
2bdcb14
 
 
ac39a48
2bdcb14
7fc9f8d
987a1bc
7fc9f8d
02a86df
168aee7
 
 
 
 
 
 
 
7fc9f8d
2bdcb14
 
 
 
 
 
 
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
"""Router for Google OAuth2 login."""

from urllib.parse import urlparse, urlunparse

from authlib.integrations.starlette_client import OAuth, OAuthError
from fastapi import APIRouter, Request, Response
from fastapi.responses import HTMLResponse
from starlette.config import Config
from starlette.responses import RedirectResponse

from .auth import UserInfo
from .config import env
from .router_utils import RouteErrorHandler

router = APIRouter(route_class=RouteErrorHandler)

if env('LILAC_AUTH_ENABLED'):
  oauth = OAuth(
    Config(
      environ={
        'GOOGLE_CLIENT_ID': env('GOOGLE_CLIENT_ID'),
        'GOOGLE_CLIENT_SECRET': env('GOOGLE_CLIENT_SECRET')
      }))
  oauth.register(
    name='google',
    server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
    client_kwargs={'scope': 'openid email profile'},
  )


@router.get('/login')
async def login(request: Request, origin_url: str) -> RedirectResponse:
  """Redirects to Google OAuth login page."""
  auth_path = urlunparse(urlparse(origin_url)._replace(path='/google/auth'))
  return await oauth.google.authorize_redirect(request, auth_path)


@router.get('/auth')
async def auth(request: Request) -> Response:
  """Handles the Google OAuth callback."""
  try:
    token = await oauth.google.authorize_access_token(request)
  except OAuthError as error:
    return HTMLResponse(f'<h1>{error}</h1>')
  userinfo = token['userinfo']
  request.session['user'] = UserInfo(
    id=str(userinfo['sub']),
    email=userinfo['email'],
    name=userinfo['name'],
    given_name=userinfo['given_name'],
    family_name=userinfo['family_name']).dict()

  return RedirectResponse(url='/')


@router.get('/logout')
def logout(request: Request) -> RedirectResponse:
  """Logs the user out."""
  request.session.pop('user', None)
  return RedirectResponse(url='/')