64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
from . import auth
|
|
from flask import request, render_template, redirect, url_for, flash, jsonify
|
|
from flask_login import login_required, login_user, current_user, logout_user
|
|
from authlib.integrations.flask_oauth2 import current_token
|
|
from authlib.oauth2 import OAuth2Error
|
|
from ..models import User, db
|
|
from .oauth2 import authorization, require_oauth
|
|
from .forms import LoginForm, ConfirmAccessForm
|
|
|
|
|
|
@auth.route('/login/', methods=['post', 'get'])
|
|
def login():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('home.index'))
|
|
form = LoginForm()
|
|
if form.validate_on_submit():
|
|
user = db.session.query(User).filter(User.username == form.username.data).first()
|
|
if user and user.check_password(form.password.data):
|
|
login_user(user, remember=form.remember.data)
|
|
nextpage = request.args.get('next', url_for('home.index'))
|
|
return redirect(nextpage)
|
|
else:
|
|
flash("Invalid username/password", 'error')
|
|
return render_template('login.html', form=form)
|
|
|
|
|
|
@auth.route('/logout/')
|
|
@login_required
|
|
def logout():
|
|
logout_user()
|
|
flash("You have been logged out.")
|
|
return redirect(url_for('home.index'))
|
|
|
|
|
|
@auth.route('/oauth/token', methods=['POST'])
|
|
def issue_token():
|
|
return authorization.create_token_response()
|
|
|
|
|
|
@auth.route('/oauth/revoke', methods=['POST'])
|
|
def revoke_token():
|
|
return authorization.create_endpoint_response('revocation')
|
|
|
|
|
|
@auth.route('/oauth/authorize', methods=['GET', 'POST'])
|
|
@login_required
|
|
def authorize():
|
|
user = current_user
|
|
grant_user = None
|
|
form = ConfirmAccessForm()
|
|
|
|
if request.method == 'GET':
|
|
try:
|
|
grant = authorization.validate_consent_request(end_user=user)
|
|
except OAuth2Error as error:
|
|
return error.error
|
|
return render_template('authorize.html', user=user, grant=grant, form=form)
|
|
|
|
if form.validate_on_submit():
|
|
if form.confirm.data:
|
|
grant_user = user
|
|
|
|
return authorization.create_authorization_response(grant_user=grant_user)
|