powerdns-admin/app/oauth.py
2018-08-19 15:29:50 +07:00

80 lines
2.7 KiB
Python

from ast import literal_eval
from flask import request, session, redirect, url_for
from flask_oauthlib.client import OAuth
from app import app, oauth
from app.models import Setting
# TODO:
# - Replace Flask-OAuthlib by authlib
# - Fix flask-migrate issue with calling Setting() class during downgrade / upgrade
# - Fix github/google enabling (Currently need to reload the flask app)
def github_oauth():
if not Setting().get('github_oauth_enabled'):
return None
github = oauth.remote_app(
'github',
consumer_key = Setting().get('github_oauth_key'),
consumer_secret = Setting().get('github_oauth_secret'),
request_token_params = {'scope': Setting().get('github_oauth_scope')},
base_url = Setting().get('github_oauth_api_url'),
request_token_url = None,
access_token_method = 'POST',
access_token_url = Setting().get('github_oauth_token_url'),
authorize_url = Setting().get('github_oauth_authorize_url')
)
@app.route('/github/authorized')
def github_authorized():
session['github_oauthredir'] = url_for('.github_authorized', _external=True)
resp = github.authorized_response()
if resp is None:
return 'Access denied: reason=%s error=%s' % (
request.args['error'],
request.args['error_description']
)
session['github_token'] = (resp['access_token'], '')
return redirect(url_for('.login'))
@github.tokengetter
def get_github_oauth_token():
return session.get('github_token')
return github
def google_oauth():
if not Setting().get('google_oauth_enabled'):
return None
google = oauth.remote_app(
'google',
consumer_key=Setting().get('google_oauth_client_id'),
consumer_secret=Setting().get('google_oauth_client_secret'),
request_token_params=literal_eval(Setting().get('google_token_params')),
base_url=Setting().get('google_base_url'),
request_token_url=None,
access_token_method='POST',
access_token_url=Setting().get('google_token_url'),
authorize_url=Setting().get('google_authorize_url'),
)
@app.route('/google/authorized')
def google_authorized():
resp = google.authorized_response()
if resp is None:
return 'Access denied: reason=%s error=%s' % (
request.args['error_reason'],
request.args['error_description']
)
session['google_token'] = (resp['access_token'], '')
return redirect(url_for('.login'))
@google.tokengetter
def get_google_oauth_token():
return session.get('google_token')
return google