SimpleSyndicate/buckler_flask.py

#!/usr/bin/env python3
"""
Session interface middlewares to integrate the flask app with Buckler.
"""
import json
import urllib.parse
import urllib.request
from datetime import datetime

from flask.sessions import SessionInterface, SessionMixin
from flask import session, redirect, request

import config

class BucklerSessionInterface(SessionInterface):
	"""
	Queries the Buckler server for session data to the current user and
	application.
	"""

	def __init__(self):
		self.url = config.buckler['url']
		self.app_id = config.buckler['app_id']
		self.app_key = config.buckler['app_key']
	
	def open_session(self, app, request):
		"""Called when a request is initiated."""
		user_id = request.cookies.get('userid')
		user_sid = request.cookies.get('session')

		params = {
			'app_id': self.app_id,
			'app_key': self.app_key,
			'userid': user_id,
			'session': user_sid
		}
		params = urllib.parse.urlencode(params)
		req = urllib.request.Request(self.url + f"/get_session?{params}")
		res = urllib.request.urlopen(req)
		data = json.loads(res.read())
		if data.get('error'):
			return None
		session = BucklerSession()
		session.update(data['session_data'])
		session.meta = data['meta']
		session.cookies = request.cookies
		return session

	def save_session(self, app, session, response):
		"""Called at the end of a request."""
		if not session.modified:
			return
		user_id = session.meta.get('user_id')
		user_sid = session.meta.get('user_sid')

		params = {
			'app_id': self.app_id,
			'app_key': self.app_key,
			'userid': user_id,
			'session': user_sid
		}
		params = urllib.parse.urlencode(params)
		data = json.dumps(session)
		req = urllib.request.Request(
			self.url + f"/set_session?{params}",
			data=data.encode('utf8'),
			method='POST')
		res = urllib.request.urlopen(req)

		last_used = datetime.fromisoformat(session.meta['last_used'])
		now = datetime.now(last_used.tzinfo)
		delta = now - last_used
		if delta.seconds > 600:
			response.set_cookie(
				'userid',
				session.cookies['userid'],
				max_age=30*24*60*60,
				secure=True,
				httponly=True,
				samesite='strict')
			response.set_cookie(
				'session',
				session.cookies['session'],
				max_age=30*24*60*60,
				secure=True,
				httponly=True,
				samesite='strict')


class BucklerSession(dict, SessionMixin):
	"""A server side session class based on the Buckler security shield."""
	def __init__(self):
		super().__init__()
		self.modified = False
	
	def __setitem__(self, key, value):
		super().__setitem__(key, value)
		self.modified = True


def require_auth():
	"""
	Requires the user to be properly authenticated with Buckler before
	accessing any views on the application.
	"""
	if not hasattr(session, 'meta'):
		resp = redirect(config.buckler['login_url'])
		resp.set_cookie(
			'redirect',
			request.url,
			secure=True,
			httponly=True,
			samesite='strict')
		return resp