aiohttp-security/demo/database_auth/handlers.py

78 lines
2.6 KiB
Python
Raw Permalink Normal View History

from textwrap import dedent
2015-11-29 13:03:35 +00:00
from aiohttp import web
from aiohttp_security import (
remember, forget, authorized_userid,
2018-09-06 10:06:55 +00:00
check_permission, check_authorized,
)
2015-11-29 13:03:35 +00:00
from .db_auth import check_credentials
2015-11-29 13:03:35 +00:00
class Web(object):
index_template = dedent("""
<!doctype html>
<head></head>
<body>
<p>{message}</p>
<form action="/login" method="post">
Login:
<input type="text" name="login">
Password:
<input type="password" name="password">
<input type="submit" value="Login">
</form>
<a href="/logout">Logout</a>
</body>
""")
2017-12-13 14:51:46 +00:00
async def index(self, request):
username = await authorized_userid(request)
if username:
template = self.index_template.format(
message='Hello, {username}!'.format(username=username))
else:
template = self.index_template.format(message='You need to login')
response = web.Response(body=template.encode())
return response
2015-11-29 13:03:35 +00:00
2017-12-13 14:51:46 +00:00
async def login(self, request):
response = web.HTTPFound('/')
2017-12-13 14:51:46 +00:00
form = await request.post()
login = form.get('login')
password = form.get('password')
db_engine = request.app.db_engine
2017-12-13 14:51:46 +00:00
if await check_credentials(db_engine, login, password):
await remember(request, response, login)
2018-09-06 10:06:55 +00:00
raise response
2015-11-29 13:03:35 +00:00
2018-09-06 10:06:55 +00:00
raise web.HTTPUnauthorized(
body=b'Invalid username/password combination')
2017-12-13 14:51:46 +00:00
async def logout(self, request):
2018-09-06 10:06:55 +00:00
await check_authorized(request)
response = web.Response(body=b'You have been logged out')
2017-12-13 14:51:46 +00:00
await forget(request, response)
return response
2015-11-29 13:03:35 +00:00
2017-12-13 14:51:46 +00:00
async def internal_page(self, request):
2018-09-06 10:06:55 +00:00
await check_permission(request, 'public')
response = web.Response(
body=b'This page is visible for all registered users')
return response
2015-11-29 13:03:35 +00:00
2017-12-13 14:51:46 +00:00
async def protected_page(self, request):
2018-09-06 10:06:55 +00:00
await check_permission(request, 'protected')
response = web.Response(body=b'You are on protected page')
return response
2015-11-30 12:58:49 +00:00
def configure(self, app):
router = app.router
router.add_route('GET', '/', self.index, name='index')
router.add_route('POST', '/login', self.login, name='login')
router.add_route('GET', '/logout', self.logout, name='logout')
router.add_route('GET', '/public', self.internal_page, name='public')
router.add_route('GET', '/protected', self.protected_page,
name='protected')