mirror of
https://github.com/janeczku/calibre-web
synced 2024-11-08 07:10:39 +00:00
139 lines
4.9 KiB
Python
139 lines
4.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
# This file is part of the Calibre-Web (https://github.com/janeczku/calibre-web)
|
|
# Copyright (C) 2018-2019 OzzieIsaacs, cervinko, jkrehm, bodybybuddha, ok11,
|
|
# andy29485, idalin, Kyosfonica, wuqi, Kennyl, lemmsh,
|
|
# falgh1, grunjol, csitko, ytils, xybydy, trasba, vrabe,
|
|
# ruben-herold, marblepebble, JackED42, SiphonSquirrel,
|
|
# apetresc, nanu-c, mutschler
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import json
|
|
from datetime import datetime
|
|
|
|
from flask import Blueprint, request, make_response, abort, url_for, flash, redirect
|
|
from flask_login import login_required, current_user, login_user
|
|
from flask_babel import gettext as _
|
|
from sqlalchemy.sql.expression import true
|
|
|
|
from . import config, logger, ub
|
|
from .render_template import render_title_template
|
|
|
|
try:
|
|
from functools import wraps
|
|
except ImportError:
|
|
pass # We're not using Python 3
|
|
|
|
remotelogin = Blueprint('remotelogin', __name__)
|
|
log = logger.create()
|
|
|
|
|
|
def remote_login_required(f):
|
|
@wraps(f)
|
|
def inner(*args, **kwargs):
|
|
if config.config_remote_login:
|
|
return f(*args, **kwargs)
|
|
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
|
|
data = {'status': 'error', 'message': 'Forbidden'}
|
|
response = make_response(json.dumps(data, ensure_ascii=False))
|
|
response.headers["Content-Type"] = "application/json; charset=utf-8"
|
|
return response, 403
|
|
abort(403)
|
|
|
|
return inner
|
|
|
|
@remotelogin.route('/remote/login')
|
|
@remote_login_required
|
|
def remote_login():
|
|
auth_token = ub.RemoteAuthToken()
|
|
ub.session.add(auth_token)
|
|
ub.session_commit()
|
|
verify_url = url_for('remotelogin.verify_token', token=auth_token.auth_token, _external=true)
|
|
log.debug(u"Remot Login request with token: %s", auth_token.auth_token)
|
|
return render_title_template('remote_login.html', title=_(u"login"), token=auth_token.auth_token,
|
|
verify_url=verify_url, page="remotelogin")
|
|
|
|
|
|
@remotelogin.route('/verify/<token>')
|
|
@remote_login_required
|
|
@login_required
|
|
def verify_token(token):
|
|
auth_token = ub.session.query(ub.RemoteAuthToken).filter(ub.RemoteAuthToken.auth_token == token).first()
|
|
|
|
# Token not found
|
|
if auth_token is None:
|
|
flash(_(u"Token not found"), category="error")
|
|
log.error(u"Remote Login token not found")
|
|
return redirect(url_for('web.index'))
|
|
|
|
# Token expired
|
|
elif datetime.now() > auth_token.expiration:
|
|
ub.session.delete(auth_token)
|
|
ub.session_commit()
|
|
|
|
flash(_(u"Token has expired"), category="error")
|
|
log.error(u"Remote Login token expired")
|
|
return redirect(url_for('web.index'))
|
|
|
|
# Update token with user information
|
|
auth_token.user_id = current_user.id
|
|
auth_token.verified = True
|
|
ub.session_commit()
|
|
|
|
flash(_(u"Success! Please return to your device"), category="success")
|
|
log.debug(u"Remote Login token for userid %s verified", auth_token.user_id)
|
|
return redirect(url_for('web.index'))
|
|
|
|
|
|
@remotelogin.route('/ajax/verify_token', methods=['POST'])
|
|
@remote_login_required
|
|
def token_verified():
|
|
token = request.form['token']
|
|
auth_token = ub.session.query(ub.RemoteAuthToken).filter(ub.RemoteAuthToken.auth_token == token).first()
|
|
|
|
data = {}
|
|
|
|
# Token not found
|
|
if auth_token is None:
|
|
data['status'] = 'error'
|
|
data['message'] = _(u"Token not found")
|
|
|
|
# Token expired
|
|
elif datetime.now() > auth_token.expiration:
|
|
ub.session.delete(auth_token)
|
|
ub.session_commit()
|
|
|
|
data['status'] = 'error'
|
|
data['message'] = _(u"Token has expired")
|
|
|
|
elif not auth_token.verified:
|
|
data['status'] = 'not_verified'
|
|
|
|
else:
|
|
user = ub.session.query(ub.User).filter(ub.User.id == auth_token.user_id).first()
|
|
login_user(user)
|
|
|
|
ub.session.delete(auth_token)
|
|
ub.session_commit("User {} logged in via remotelogin, token deleted".format(user.name))
|
|
|
|
data['status'] = 'success'
|
|
log.debug(u"Remote Login for userid %s succeded", user.id)
|
|
flash(_(u"you are now logged in as: '%(nickname)s'", nickname=user.name), category="success")
|
|
|
|
response = make_response(json.dumps(data, ensure_ascii=False))
|
|
response.headers["Content-Type"] = "application/json; charset=utf-8"
|
|
|
|
return response
|