mirrors/calibre-web
1
0
Fork 0
mirror of https://github.com/janeczku/calibre-web synced 2025-10-31 07:13:02 +00:00
Code Issues Releases Wiki Activity

further refactored user login

Browse Source
This commit is contained in:
Ozzie Isaacs
2023-02-04 14:51:41 +01:00
parent 98da7dd5b0
commit f8fbc807f1
11 changed files with 71 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
cps/MyLoginManager.py
View File

@@ -21,9 +21,10 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from flask_login import LoginManager
from flask import session
from flask_login import LoginManager, confirm_login
from flask import session, current_app
from flask_login.utils import decode_cookie
from flask_login.signals import user_loaded_from_cookie
class MyLoginManager(LoginManager):
def _session_protection_failed(self):
@@ -33,3 +34,18 @@ class MyLoginManager(LoginManager):
and _session.get('csrf_token', None))) and ident != _session.get('_id', None):
return super(). _session_protection_failed()
return False
def _load_user_from_remember_cookie(self, cookie):
user_id = decode_cookie(cookie)
if user_id is not None:
session["_user_id"] = user_id
session["_fresh"] = False
user = None
if self._user_callback:
user = self._user_callback(user_id)
if user is not None:
app = current_app._get_current_object()
user_loaded_from_cookie.send(app, user=user)
confirm_login()
return user
return None

2
cps/admin.py
View File

@@ -33,7 +33,7 @@ from datetime import time as datetime_time
from functools import wraps
from flask import Blueprint, flash, redirect, url_for, abort, request, make_response, send_from_directory, g, Response
from flask_login import login_required, current_user, logout_user, confirm_login
from flask_login import login_required, current_user, logout_user
from flask_babel import gettext as _
from flask_babel import get_locale, format_time, format_datetime, format_timedelta
from flask import session as flask_session

16
cps/opds.py
View File

@@ -23,10 +23,10 @@
import datetime
from urllib.parse import unquote_plus
from flask import Blueprint, request, render_template, g, make_response, abort
from flask import Blueprint, request, render_template, make_response, abort
from flask_login import current_user
from flask_babel import get_locale
from flask_babel import gettext as _
from sqlalchemy.sql.expression import func, text, or_, and_, true
from sqlalchemy.exc import InvalidRequestError, OperationalError
@@ -35,8 +35,7 @@ from .usermanagement import requires_basic_auth_if_no_ano
from .helper import get_download_link, get_book_cover
from .pagination import Pagination
from .web import render_read_books
from .usermanagement import load_user_from_request
from flask_babel import gettext as _
opds = Blueprint('opds', __name__)
@@ -342,7 +341,8 @@ def feed_languages(book_id):
@requires_basic_auth_if_no_ano
def feed_shelfindex():
off = request.args.get("offset") or 0
shelf = g.shelves_access
shelf = ub.session.query(ub.Shelf).filter(
or_(ub.Shelf.is_public == 1, ub.Shelf.user_id == current_user.id)).order_by(ub.Shelf.name).all()
number = len(shelf)
pagination = Pagination((int(off) / (int(config.config_books_per_page)) + 1), config.config_books_per_page,
number)
@@ -389,11 +389,7 @@ def feed_shelf(book_id):
@opds.route("/opds/download/<book_id>/<book_format>/")
@requires_basic_auth_if_no_ano
def opds_download_link(book_id, book_format):
# I gave up with this: With enabled ldap login, the user doesn't get logged in, therefore it's always guest
# workaround, loading the user from the request and checking its download rights here
# in case of anonymous browsing user is None
user = load_user_from_request(request) or current_user
if not user.role_download():
if not current_user.role_download():
return abort(403)
if "Kobo" in request.headers.get('User-Agent'):
client = "kobo"

7
cps/render_template.py
View File

@@ -20,11 +20,13 @@ from flask import render_template, g, abort, request
from flask_babel import gettext as _
from werkzeug.local import LocalProxy
from flask_login import current_user
from sqlalchemy.sql.expression import or_
from . import config, constants, logger
from . import config, constants, logger, ub
from .ub import User
log = logger.create()
def get_sidebar_config(kwargs=None):
@@ -99,6 +101,9 @@ def get_sidebar_config(kwargs=None):
{"glyph": "glyphicon-th-list", "text": _('Books List'), "link": 'web.books_table', "id": "list",
"visibility": constants.SIDEBAR_LIST, 'public': (not current_user.is_anonymous), "page": "list",
"show_text": _('Show Books List'), "config_show": content})
g.shelves_access = ub.session.query(ub.Shelf).filter(
or_(ub.Shelf.is_public == 1, ub.Shelf.user_id == current_user.id)).order_by(ub.Shelf.name).all()
return sidebar, simple

1
cps/services/simpleldap.py
View File

@@ -89,6 +89,7 @@ def get_object_details(user=None,query_filter=None):
def bind():
print("bind")
return _ldap.bind()

2
cps/templates/book_edit.html
View File

@@ -219,7 +219,7 @@
{% endblock %}
{% block modal %}
{{ delete_book() }}
{{ delete_book(current_user.role_delete_books()) }}
{{ delete_confirm_modal() }}
<div class="modal fade" id="metaModal" tabindex="-1" role="dialog" aria-labelledby="metaModalLabel">

2
cps/templates/book_table.html
View File

@@ -104,7 +104,7 @@
</table>
{% endblock %}
{% block modal %}
{{ delete_book() }}
{{ delete_book(current_user.role_delete_books()) }}
{% if current_user.role_edit() %}
<div class="modal fade" id="mergeModal" role="dialog" aria-labelledby="metaMergeLabel">
<div class="modal-dialog">

4
cps/templates/modal_dialogs.html
View File

@@ -37,8 +37,8 @@
</div>
</div>
{% endmacro %}
{% macro delete_book() %}
{% if current_user.role_delete_books() %}
{% macro delete_book(allow) %}
{% if allow %}
<div class="modal fade" id="deleteModal" role="dialog" aria-labelledby="metaDeleteLabel">
<div class="modal-dialog">
<div class="modal-content">

2
cps/templates/tasks.html
View File

@@ -27,7 +27,7 @@
</div>
{% endblock %}
{% block modal %}
{{ delete_book() }}
{{ delete_book(current_user.role_delete_books()) }}
{% if current_user.role_admin() %}
<div class="modal fade" id="cancelTaskModal" role="dialog" aria-labelledby="metaCancelTaskLabel">
<div class="modal-dialog">

86
cps/usermanagement.py
View File

@@ -16,8 +16,6 @@
# 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 base64
import binascii
from functools import wraps
from sqlalchemy.sql.expression import func
@@ -42,44 +40,46 @@ def requires_basic_auth_if_no_ano(f):
@wraps(f)
def decorated(*args, **kwargs):
auth = request.authorization
if config.config_anonbrowse != 1:
if not auth or auth.type != 'basic' or not check_auth(auth.username, auth.password):
return authenticate()
print("opds_requires_basic_auth")
user = load_user_from_auth_header(auth.username, auth.password)
if not user:
return None
login_user(user)
print("opds_requires_basic_auth")
if (not auth or auth.type != 'basic'):
if config.config_anonbrowse != 1:
return _authenticate()
else:
return f(*args, **kwargs)
if config.config_login_type == constants.LOGIN_LDAP and services.ldap:
result, error = services.ldap.bind_user(auth.username, auth.password)
if result:
user = _fetch_user_by_name(auth.username)
login_user(user)
else:
log.error(error)
user = None
else:
user = _load_user_from_auth_header(auth.username, auth.password)
if not user:
return _authenticate()
return f(*args, **kwargs)
if config.config_login_type == constants.LOGIN_LDAP and services.ldap and config.config_anonbrowse != 1:
return services.ldap.basic_auth_required(f)
return decorated
def check_auth(username, password):
try:
username = username.encode('windows-1252')
except UnicodeEncodeError:
username = username.encode('utf-8')
user = ub.session.query(ub.User).filter(func.lower(ub.User.name) ==
username.decode('utf-8').lower()).first()
def _load_user_from_auth_header(username, password):
user = _fetch_user_by_name(username)
if bool(user and check_password_hash(str(user.password), password)):
return True
login_user(user)
return user
else:
ip_address = request.headers.get('X-Forwarded-For', request.remote_addr)
log.warning('OPDS Login failed for user "%s" IP-address: %s', username.decode('utf-8'), ip_address)
return False
log.warning('OPDS Login failed for user "%s" IP-address: %s', username, ip_address)
return None
def authenticate():
def _authenticate():
return Response(
'Could not verify your access level for that URL.\n'
'You have to login with proper credentials', 401,
{'WWW-Authenticate': 'Basic realm="Login Required"'})
def _fetch_user_by_name(username):
return ub.session.query(ub.User).filter(func.lower(ub.User.name) == username.lower()).first()
@@ -87,49 +87,21 @@ def _fetch_user_by_name(username):
@lm.user_loader
def load_user(user_id):
print("load_user: {}".format(user_id))
return ub.session.query(ub.User).filter(ub.User.id == int(user_id)).first()
user = ub.session.query(ub.User).filter(ub.User.id == int(user_id)).first()
return user
@lm.request_loader
def load_user_from_request(request):
def load_user_from_request(req):
print("load_from_request")
if config.config_allow_reverse_proxy_header_login:
rp_header_name = config.config_reverse_proxy_login_header_name
if rp_header_name:
rp_header_username = request.headers.get(rp_header_name)
rp_header_username = req.headers.get(rp_header_name)
if rp_header_username:
user = _fetch_user_by_name(rp_header_username)
if user:
login_user(user)
return user
#auth_header = request.headers.get("Authorization")
#if auth_header:
# user = load_user_from_auth_header(auth_header)
# if user:
# login_user(user)
# return user
return None
def load_user_from_auth_header(basic_username, basic_password):
#if header_val.startswith('Basic '):
# header_val = header_val.replace('Basic ', '', 1)
#basic_username = basic_password = '' # nosec
#try:
# header_val = base64.b64decode(header_val).decode('utf-8')
# # Users with colon are invalid: rfc7617 page 4
# basic_username = header_val.split(':', 1)[0]
# basic_password = header_val.split(':', 1)[1]
#except (TypeError, UnicodeDecodeError, binascii.Error):
# pass
user = _fetch_user_by_name(basic_username)
if user and config.config_login_type == constants.LOGIN_LDAP and services.ldap:
if services.ldap.bind_user(str(user.password), basic_password):
login_user(user)
return user
if user and check_password_hash(str(user.password), basic_password):
login_user(user)
return user
return None

9
cps/web.py
View File

@@ -24,7 +24,7 @@ import mimetypes
import chardet # dependency of requests
import copy
from flask import Blueprint, jsonify, g
from flask import Blueprint, jsonify
from flask import request, redirect, send_from_directory, make_response, flash, abort, url_for
from flask import session as flask_session
from flask_babel import gettext as _
@@ -54,6 +54,8 @@ from .usermanagement import login_required_if_no_ano
from .kobo_sync_status import remove_synced_book
from .render_template import render_title_template
from .kobo_sync_status import change_archived_books
from .services.worker import WorkerThread
from .tasks_status import render_task_status
feature_support = {
'ldap': bool(services.ldap),
@@ -79,7 +81,7 @@ except ImportError:
@app.after_request
def add_security_headers_and_shelves(resp):
def add_security_headers(resp):
csp = "default-src 'self'"
csp += ''.join([' ' + host for host in config.config_trustedhosts.strip().split(',')])
csp += " 'unsafe-inline' 'unsafe-eval'; font-src 'self' data:; img-src 'self'"
@@ -98,9 +100,6 @@ def add_security_headers_and_shelves(resp):
resp.headers['X-Frame-Options'] = 'SAMEORIGIN'
resp.headers['X-XSS-Protection'] = '1; mode=block'
resp.headers['Strict-Transport-Security'] = 'max-age=31536000;'
g.shelves_access = ub.session.query(ub.Shelf).filter(
or_(ub.Shelf.is_public == 1, ub.Shelf.user_id == current_user.id)).order_by(ub.Shelf.name).all()
return resp