| from flask import abort, session, current_app |
| from flask_login import current_user |
| from flask_login.signals import user_logged_out |
| import requests |
| import functools |
| import time |
| |
| |
| def cap_check(capability, user=None): |
| '''Checks if specified user (or current user) has desired capacifier |
| capability''' |
| if not user and not current_user.is_authenticated: |
| return False |
| |
| user = user or current_user.get_id() |
| |
| cache_key = '{}-{}'.format(user, capability) |
| cached_cap = session.get('_caps', {}).get(cache_key, (False, 0)) |
| |
| if cached_cap[1] > time.time(): |
| return cached_cap[0] |
| |
| allowed = requests.get( |
| 'https://capacifier.hackerspace.pl/%s/%s' % (capability, user) |
| ).status_code == 200 |
| |
| if '_caps' not in session: |
| session['_caps'] = {} |
| |
| session['_caps'][cache_key] = \ |
| (allowed, time.time() + current_app.config.get('CAP_TTL', 3600)) |
| |
| return allowed |
| |
| |
| @user_logged_out.connect |
| def caps_cleanup(app, user): |
| # Cleanup caps cache |
| session.pop('_caps', None) |
| |
| |
| def cap_required(capability): |
| '''Decorator to check if user has desired capacifier capability, returns |
| 403 otherwise''' |
| |
| def inner(func): |
| @functools.wraps(func) |
| def wrapped(*args, **kwargs): |
| if not cap_check(capability): |
| abort(403) |
| |
| return func(*args, **kwargs) |
| |
| return wrapped |
| return inner |