File size: 1,849 Bytes
bfc0ec6
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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 .env 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='/')