95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
from flask import Blueprint, request, url_for, session, redirect, abort, flash
|
|
from flask_oauthlib.client import OAuth, OAuthException
|
|
from flask_login import LoginManager, login_user, logout_user, current_user, login_required, UserMixin
|
|
from spaceauth.caps import cap_required
|
|
|
|
|
|
class SpaceAuth(object):
|
|
def __init__(self, app=None, *args, **kwargs):
|
|
self.oauth = OAuth()
|
|
self.remote = self.oauth.remote_app(
|
|
'spaceauth',
|
|
base_url='https://sso.hackerspace.pl/api/',
|
|
access_token_url='https://sso.hackerspace.pl/oauth/token',
|
|
authorize_url='https://sso.hackerspace.pl/oauth/authorize',
|
|
request_token_params={'scope': 'profile:read'},
|
|
app_key='SPACEAUTH')
|
|
self.remote.tokengetter(self.tokengetter)
|
|
|
|
bp = Blueprint('spaceauth', __name__)
|
|
bp.add_url_rule('/login', 'login', self.login_view)
|
|
bp.add_url_rule('/logout', 'logout', self.logout_view)
|
|
bp.add_url_rule('/callback', 'callback', self.callback_view)
|
|
self.blueprint = bp
|
|
|
|
self.login_manager = LoginManager()
|
|
self.login_manager.refresh_view = 'spaceauth.login'
|
|
self.login_manager.login_view = 'spaceauth.login'
|
|
self.login_manager.user_loader(self.user_loader_handler)
|
|
|
|
if app:
|
|
self.init_app(app, *args, **kwargs)
|
|
|
|
def init_app(self, app, url_prefix='/oauth'):
|
|
self.oauth.init_app(app)
|
|
self.login_manager.init_app(app)
|
|
app.register_blueprint(self.blueprint, url_prefix=url_prefix)
|
|
|
|
@app.errorhandler(OAuthException)
|
|
def errorhandler(err):
|
|
flash('OAuth error occured', 'error')
|
|
return redirect('/')
|
|
|
|
def login_view(self):
|
|
session['spaceauth_next'] = request.args.get('next') or request.referrer
|
|
return self.remote.authorize(
|
|
callback=url_for('spaceauth.callback', _external=True)
|
|
)
|
|
|
|
def logout_view(self):
|
|
# TODO revoke token
|
|
session.pop('spaceauth_token', None)
|
|
session.pop('spaceauth_next', None)
|
|
logout_user()
|
|
return redirect('/')
|
|
|
|
def callback_view(self):
|
|
resp = self.remote.authorized_response()
|
|
if resp is None:
|
|
raise OAuthException(
|
|
'Access denied', type=request.args.get('error'))
|
|
|
|
# TODO encrypt token...?
|
|
session['spaceauth_token'] = resp['access_token']
|
|
profile = self.remote.get('profile').data
|
|
|
|
login_user(self.user_loader_handler(profile['username'], profile))
|
|
return redirect(session.pop('spaceauth_next', None) or '/')
|
|
|
|
def tokengetter(self):
|
|
return (session.get('spaceauth_token'), '')
|
|
|
|
def user_loader_handler(self, uid, profile=None):
|
|
"""
|
|
Default user loader just to differentiate authenticated user from
|
|
anonymous.
|
|
"""
|
|
|
|
user = UserMixin()
|
|
user.id = uid
|
|
return user
|
|
|
|
def user_loader(self, func):
|
|
"""
|
|
Define flask_login-like user loader. Application is supposed to create
|
|
its user model when missing. Additional `profile` argument is passed
|
|
with user profile information right after login.
|
|
"""
|
|
|
|
self.user_loader_handler = func
|
|
self.login_manager.user_loader(self.user_loader_handler)
|
|
return func
|
|
|
|
def user_profile(self):
|
|
return self.remote.get('profile').data
|