Stickup/stickup.py

#!/usr/bin/env python3
"""
An email management frontend.
"""
import asyncio

from aiohttp import web
import jinja2
import aiohttp_jinja2
from aiohttp_jinja2 import render_template
import uvloop
import asyncpg
from passlib.hash import argon2

import config
import buckler_aiohttp

uvloop.install()
routes = web.RouteTableDef()

@routes.get('/', name='index')
@routes.post('/', name='index')
async def index(request):
	"""The index page."""
	user_id = int(request.cookies.get('userid'))

	async with request.app['pool'].acquire() as conn:
		user_data = await conn.fetch(
			"SELECT * FROM virtual_users WHERE buckler_id = $1",
			user_id)

	result = {}
	if request.method == 'POST':
		data = await request.post()
		current_password = data.get('current_password', '')
		new_password = data.get('new_password', '')
		verify_new_password = data.get('verify_new_password', '')

		if not argon2.verify(current_password, user_data[0]['password']):
			result = {
				'ok': False,
				'message': "Current password does not match."
			}
			return render_template('index.html', request, locals())

		if new_password != verify_new_password:
			result = {
				'ok': False,
				'message': "New passwords do not match."
			}
			return render_template('index.html', request, locals())

		if len(new_password) > config.max_password:
			result = {
				'ok': False,
				'message': "Maximum password length is 1024 characters."
			}
			return render_template('index.html', request, locals())

		if len(new_password) < config.min_password:
			result = {
				'ok': False,
				'message': "Minimum password length is 8 characters."
			}
			return render_template('index.html', request, locals())

		pw_hash = argon2.hash(new_password)
		async with request.app['pool'].acquire() as conn:
			await conn.fetch(
				"UPDATE virtual_users SET password = $1 WHERE buckler_id = $2",
				pw_hash, user_id)
		result = {
			'ok': True,
			'message': "Password has been changed."
		}

	return render_template('index.html', request, locals())


async def init_app():
	"""Initializes the application."""
	app = web.Application(middlewares=[buckler_aiohttp.buckler_session])
	aiohttp_jinja2.setup(
		app,
		trim_blocks=True,
		lstrip_blocks=True,
		undefined=jinja2.StrictUndefined,
		loader=jinja2.FileSystemLoader('templates'),
	)
	app['pool'] = await asyncpg.create_pool(**config.mailserver_db)
	app.router.add_routes(routes)
	app_wrap = web.Application()
	app_wrap.add_subapp(config.url_prefix, app)
	return app_wrap

if __name__ == "__main__":
	app = asyncio.run(init_app())
	aiohttp.web.run_app(app, host='0.0.0.0', port=5050)