user01/calibre-web
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix for flask limiter 4.0

Browse Source 
Fix reseting flask_limiter keys
This commit is contained in:
Ozzie Isaacs
2026-02-17 18:30:47 +01:00
parent cabcace3f0
commit e3bf369ad6
7 changed files with 128 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
cps/kobo.py
View File

@@ -19,38 +19,30 @@
import base64
from datetime import datetime, timezone
from functools import wraps
import os
import uuid
import zipfile
from time import gmtime, strftime
import json
from urllib.parse import unquote
import requests
from flask import (
Blueprint,
request,
make_response,
jsonify,
current_app,
url_for,
redirect,
abort
)
from .cw_login import current_user
from flask import Blueprint, request, make_response, jsonify, current_app, url_for, redirect, abort, g
from werkzeug.datastructures import Headers
from sqlalchemy import func
from sqlalchemy.sql.expression import and_, or_
from sqlalchemy.exc import StatementError
import requests
from flask_limiter.util import get_remote_address
from .cw_login import current_user, login_user
from . import config, logger, kobo_auth, db, calibre_db, helper, shelf as shelf_lib, ub, csrf, kobo_sync_status
from . import isoLanguages
from .epub import get_epub_layout
from .constants import COVER_THUMBNAIL_SMALL, COVER_THUMBNAIL_MEDIUM, COVER_THUMBNAIL_LARGE, BASE_DIR
from .helper import get_download_link
from .services import SyncToken as SyncToken
from .web import download_required
from .kobo_auth import requires_kobo_auth, get_auth_token
from . import isoLanguages, limiter
KOBO_FORMATS = {"KEPUB": ["KEPUB"], "EPUB": ["EPUB3", "EPUB"]}
KOBO_STOREAPI_URL = "https://storeapi.kobo.com"
@@ -65,6 +57,33 @@ kobo_auth.register_url_value_preprocessor(kobo)
log = logger.create()
def get_auth_token():
if "auth_token" in g:
return g.get("auth_token")
else:
return None
def requires_kobo_auth(f):
@wraps(f)
def inner(*args, **kwargs):
auth_token = get_auth_token()
if auth_token is not None:
user = (
ub.session.query(ub.User)
.join(ub.RemoteAuthToken)
.filter(ub.RemoteAuthToken.auth_token == auth_token).filter(ub.RemoteAuthToken.token_type==1)
.first()
)
if user is not None:
login_user(user)
[limiter.limiter.clear(limit.limit, *limit.request_args) for limit in limiter.current_limits]
return f(*args, **kwargs)
log.debug("Received Kobo request without a recognizable auth token.")
return abort(401)
return inner
def get_store_url_for_current_request():
# Programmatically modify the current url to point to the official Kobo store
__, __, request_path_with_auth_token = request.full_path.rpartition("/kobo/")
@@ -140,7 +159,7 @@ def convert_to_kobo_timestamp_string(timestamp):
@kobo.route("/v1/library/sync")
@requires_kobo_auth
# @download_required
# @limiter.limit("3/minute", key_func=get_remote_address)
def HandleSyncRequest():
if not current_user.role_download():
log.info("Users need download permissions for syncing library to Kobo reader")
@@ -336,6 +355,7 @@ def generate_sync_response(sync_token, sync_results, set_cont=False):
@kobo.route("/v1/library/<book_uuid>/metadata")
@requires_kobo_auth
@download_required
# @limiter.limit("3/minute", key_func=get_remote_address)
def HandleMetadataRequest(book_uuid):
if not current_app.wsgi_app.is_proxied:
log.debug('Kobo: Received unproxied request, changed request port to external server port')
@@ -368,7 +388,7 @@ def get_download_url_for_book(book_id, book_format):
)
return url_for(
"kobo.download_book",
auth_token=kobo_auth.get_auth_token(),
auth_token=get_auth_token(),
book_id=book_id,
book_format=book_format.lower(),
_external=True,
@@ -503,9 +523,11 @@ def get_metadata(book):
return metadata
@csrf.exempt
@kobo.route("/v1/library/tags", methods=["POST", "DELETE"])
@requires_kobo_auth
# @limiter.limit("3/minute", key_func=get_remote_address)
# Creates a Shelf with the given items, and returns the shelf's uuid.
def HandleTagCreate():
# catch delete requests, otherwise they are handled in the book delete handler
@@ -541,6 +563,7 @@ def HandleTagCreate():
@csrf.exempt
@kobo.route("/v1/library/tags/<tag_id>", methods=["DELETE", "PUT"])
@requires_kobo_auth
# @limiter.limit("3/minute", key_func=get_remote_address)
def HandleTagUpdate(tag_id):
shelf = ub.session.query(ub.Shelf).filter(ub.Shelf.uuid == tag_id,
ub.Shelf.user_id == current_user.id).one_or_none()
@@ -595,6 +618,7 @@ def add_items_to_shelf(items, shelf):
@csrf.exempt
@kobo.route("/v1/library/tags/<tag_id>/items", methods=["POST"])
@requires_kobo_auth
# @limiter.limit("3/minute", key_func=get_remote_address)
def HandleTagAddItem(tag_id):
items = None
try:
@@ -625,6 +649,7 @@ def HandleTagAddItem(tag_id):
@csrf.exempt
@kobo.route("/v1/library/tags/<tag_id>/items/delete", methods=["POST"])
@requires_kobo_auth
# @limiter.limit("3/minute", key_func=get_remote_address)
def HandleTagRemoveItem(tag_id):
items = None
try:
@@ -758,6 +783,7 @@ def create_kobo_tag(shelf):
@csrf.exempt
@kobo.route("/v1/library/<book_uuid>/state", methods=["GET", "PUT"])
@requires_kobo_auth
# @limiter.limit("3/minute", key_func=get_remote_address)
def HandleStateRequest(book_uuid):
book = calibre_db.get_book_by_uuid(book_uuid)
if not book or not book.data:
@@ -907,6 +933,7 @@ def get_current_bookmark_response(current_bookmark):
@kobo.route("/<book_uuid>/<width>/<height>/<isGreyscale>/image.jpg", defaults={'Quality': ""})
@kobo.route("/<book_uuid>/<width>/<height>/<Quality>/<isGreyscale>/image.jpg")
@requires_kobo_auth
# @limiter.limit("3/minute", key_func=get_remote_address)
def HandleCoverImageRequest(book_uuid, width, height, Quality, isGreyscale):
try:
if int(height) > 1000:
@@ -943,6 +970,7 @@ def TopLevelEndpoint():
@csrf.exempt
@kobo.route("/v1/library/<book_uuid>", methods=["DELETE"])
@requires_kobo_auth
# @limiter.limit("3/minute", key_func=get_remote_address)
def HandleBookDeletionRequest(book_uuid):
log.info("Kobo book delete request received for book %s" % book_uuid)
book = calibre_db.get_book_by_uuid(book_uuid)
@@ -962,6 +990,7 @@ def HandleBookDeletionRequest(book_uuid):
@kobo.route("/v1/library/<dummy>", methods=["DELETE", "GET", "POST"])
@kobo.route("/v1/library/<dummy>/preview", methods=["POST"])
def HandleUnimplementedRequest(dummy=None):
[limiter.limiter.clear(limit.limit, *limit.request_args) for limit in limiter.current_limits]
log.debug("Unimplemented Library Request received: %s (request is forwarded to kobo if configured)",
request.base_url)
return redirect_or_proxy_request()
@@ -976,6 +1005,9 @@ def HandleUnimplementedRequest(dummy=None):
@kobo.route("/v1/analytics/<dummy>", methods=["GET", "POST"])
@kobo.route("/v1/assets", methods=["GET"])
def HandleUserRequest(dummy=None):
[limiter.limiter.clear(limit.limit, *limit.request_args) for limit in limiter.current_limits]
log.error("Key: {}".format(limiter.current_limit.key))
log.error("Remaining: {}".format(limiter.current_limit.remaining))
log.debug("Unimplemented User Request received: %s (request is forwarded to kobo if configured)", request.base_url)
return redirect_or_proxy_request()
@@ -983,6 +1015,7 @@ def HandleUserRequest(dummy=None):
@csrf.exempt
@kobo.route("/v1/user/loyalty/benefits", methods=["GET"])
def handle_benefits():
[limiter.limiter.clear(limit.limit, *limit.request_args) for limit in limiter.current_limits]
if config.config_kobo_proxy:
return redirect_or_proxy_request()
else:
@@ -992,6 +1025,7 @@ def handle_benefits():
@csrf.exempt
@kobo.route("/v1/analytics/gettests", methods=["GET", "POST"])
def handle_getests():
[limiter.limiter.clear(limit.limit, *limit.request_args) for limit in limiter.current_limits]
if config.config_kobo_proxy:
return redirect_or_proxy_request()
else:
@@ -1018,6 +1052,7 @@ def handle_getests():
@kobo.route("/v1/affiliate", methods=["GET", "POST"])
@kobo.route("/v1/deals", methods=["GET", "POST"])
def HandleProductsRequest(dummy=None):
[limiter.limiter.clear(limit.limit, *limit.request_args) for limit in limiter.current_limits]
log.debug("Unimplemented Products Request received: %s (request is forwarded to kobo if configured)",
request.base_url)
return redirect_or_proxy_request()
@@ -1046,7 +1081,10 @@ def make_calibre_web_auth_response():
@kobo.route("/v1/auth/refresh", methods=["POST"])
@kobo.route("/v1/auth/device", methods=["POST"])
@requires_kobo_auth
# @limiter.limit("3/minute", key_func=get_remote_address)
def HandleAuthRequest():
log.error(limiter.current_limit)
log.error(limiter.current_limit)
log.debug('Kobo Auth request')
if config.config_kobo_proxy:
try:
@@ -1058,6 +1096,7 @@ def HandleAuthRequest():
@kobo.route("/v1/initialization")
@requires_kobo_auth
# @limiter.limit("3/minute", key_func=get_remote_address)
def HandleInitRequest():
log.info('Init')
@@ -1088,7 +1127,7 @@ def HandleInitRequest():
kobo_resources["image_host"] = calibre_web_url
kobo_resources["image_url_quality_template"] = unquote(calibre_web_url +
url_for("kobo.HandleCoverImageRequest",
auth_token=kobo_auth.get_auth_token(),
auth_token=get_auth_token(),
book_uuid="{ImageId}",
width="{width}",
height="{height}",
@@ -1096,7 +1135,7 @@ def HandleInitRequest():
isGreyscale='isGreyscale'))
kobo_resources["image_url_template"] = unquote(calibre_web_url +
url_for("kobo.HandleCoverImageRequest",
auth_token=kobo_auth.get_auth_token(),
auth_token=get_auth_token(),
book_uuid="{ImageId}",
width="{width}",
height="{height}",
@@ -1104,7 +1143,7 @@ def HandleInitRequest():
else:
kobo_resources["image_host"] = url_for("web.index", _external=True).strip("/")
kobo_resources["image_url_quality_template"] = unquote(url_for("kobo.HandleCoverImageRequest",
auth_token=kobo_auth.get_auth_token(),
auth_token=get_auth_token(),
book_uuid="{ImageId}",
width="{width}",
height="{height}",
@@ -1112,7 +1151,7 @@ def HandleInitRequest():
isGreyscale='isGreyscale',
_external=True))
kobo_resources["image_url_template"] = unquote(url_for("kobo.HandleCoverImageRequest",
auth_token=kobo_auth.get_auth_token(),
auth_token=get_auth_token(),
book_uuid="{ImageId}",
width="{width}",
height="{height}",
@@ -1128,6 +1167,7 @@ def HandleInitRequest():
@kobo.route("/download/<book_id>/<book_format>")
@requires_kobo_auth
@download_required
# @limiter.limit("3/minute", key_func=get_remote_address)
def download_book(book_id, book_format):
return get_download_link(book_id, book_format, "kobo")
@@ -1299,3 +1339,5 @@ def NATIVE_KOBO_RESOURCES():
"userguide_host": "https://ereaderfiles.kobo.com",
"wishlist_page": "https://www.kobo.com/{region}/{language}/account/wishlist"
}
limiter.limit("3/minute", key_func=get_remote_address)(kobo)

40
cps/kobo_auth.py
View File

@@ -62,14 +62,12 @@ particular calls to non-Kobo specific endpoints such as the CalibreWeb book down
from binascii import hexlify
from datetime import datetime
from os import urandom
from functools import wraps
from flask import g, Blueprint, abort, request
from .cw_login import login_user, current_user
from flask import g, Blueprint, request
from .cw_login import current_user
from flask_babel import gettext as _
from flask_limiter import RateLimitExceeded
from . import logger, config, calibre_db, db, helper, ub, lm, limiter
from . import logger, config, calibre_db, db, helper, ub, lm
from .render_template import render_title_template
from .usermanagement import user_login_required
@@ -135,13 +133,6 @@ def disable_failed_auth_redirect_for_blueprint(bp):
lm.blueprint_login_views[bp.name] = None
def get_auth_token():
if "auth_token" in g:
return g.get("auth_token")
else:
return None
def register_url_value_preprocessor(kobo):
@kobo.url_value_preprocessor
# pylint: disable=unused-variable
@@ -149,28 +140,3 @@ def register_url_value_preprocessor(kobo):
g.auth_token = values.pop("auth_token")
def requires_kobo_auth(f):
@wraps(f)
def inner(*args, **kwargs):
auth_token = get_auth_token()
if auth_token is not None:
#try:
# limiter.check()
#except RateLimitExceeded:
# return abort(429)
#except (ConnectionError, Exception) as e:
# log.error("Connection error to limiter backend: %s", e)
# return abort(429)
user = (
ub.session.query(ub.User)
.join(ub.RemoteAuthToken)
.filter(ub.RemoteAuthToken.auth_token == auth_token).filter(ub.RemoteAuthToken.token_type==1)
.first()
)
if user is not None:
login_user(user)
[limiter.limiter.storage.clear(k.key) for k in limiter.current_limits]
return f(*args, **kwargs)
log.debug("Received Kobo request without a recognizable auth token.")
return abort(401)
return inner

4
cps/main.py
View File

@@ -46,11 +46,10 @@ def main():
try:
from .kobo import kobo, get_kobo_activated
from .kobo_auth import kobo_auth
from flask_limiter.util import get_remote_address
kobo_available = get_kobo_activated()
except (ImportError, AttributeError): # Catch also error for not installed flask-WTF (missing csrf decorator)
kobo_available = False
kobo = kobo_auth = get_remote_address = None
kobo = kobo_auth = None
try:
from .oauth_bb import oauth
@@ -77,7 +76,6 @@ def main():
app.register_blueprint(gdrive)
app.register_blueprint(editbook)
if kobo_available:
limiter.limit("3/minute", key_func=get_remote_address)(kobo)
app.register_blueprint(kobo)
app.register_blueprint(kobo_auth)
if oauth_available:

6
cps/usermanagement.py
View File

@@ -43,14 +43,14 @@ def verify_password(username, password):
if config.config_login_type == constants.LOGIN_LDAP and services.ldap:
login_result, error = services.ldap.bind_user(user.name, password)
if login_result:
[limiter.limiter.storage.clear(k.key) for k in limiter.current_limits]
[limiter.limiter.clear(limit.limit, *limit.request_args) for limit in limiter.current_limits]
return user
if error is not None:
log.error(error)
else:
# limiter.check()
if check_password_hash(str(user.password), password):
[limiter.limiter.storage.clear(k.key) for k in limiter.current_limits]
[limiter.limiter.clear(limit.limit, *limit.request_args) for limit in limiter.current_limits]
return user
ip_address = request.headers.get('X-Forwarded-For', request.remote_addr)
log.warning('OPDS Login failed for user "%s" IP-address: %s', username, ip_address)
@@ -120,7 +120,7 @@ def load_user_from_reverse_proxy_header(req):
if rp_header_username:
user = ub.session.query(ub.User).filter(func.lower(ub.User.name) == rp_header_username.lower()).first()
if user:
[limiter.limiter.storage.clear(k.key) for k in limiter.current_limits]
[limiter.limiter.clear(limit.limit, *limit.request_args) for limit in limiter.current_limits]
return user
return None

24
cps/web.py
View File

@@ -30,7 +30,6 @@ from flask import session as flask_session
from flask_babel import gettext as _
from flask_babel import get_locale
from .cw_login import login_user, logout_user, current_user
# from flask_limiter import RateLimitExceeded
from flask_limiter.util import get_remote_address
from sqlalchemy.exc import IntegrityError, InvalidRequestError, OperationalError
from sqlalchemy.sql.expression import text, func, false, not_, and_, or_
@@ -39,7 +38,7 @@ from sqlalchemy.sql.functions import coalesce
from werkzeug.datastructures import Headers
from werkzeug.security import generate_password_hash, check_password_hash
from . import constants, logger, isoLanguages, services
from . import constants, logger, isoLanguages, services, limiter
from . import db, ub, config, app
from . import calibre_db, kobo_sync_status
from .search import render_search_results, render_adv_search_results
@@ -55,7 +54,6 @@ 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 . import limiter
from .services.worker import WorkerThread
from .tasks_status import render_task_status
from .usermanagement import user_login_required
@@ -1285,15 +1283,6 @@ def register_post():
if not config.config_public_reg:
abort(404)
to_save = request.form.to_dict()
#try:
# limiter.check()
#except RateLimitExceeded:
# flash(_(u"Please wait one minute to register next user"), category="error")
# return render_title_template('register.html', config=config, title=_("Register"), page="register")
#except (ConnectionError, Exception) as e:
# log.error("Connection error to limiter backend: %s", e)
# flash(_("Connection error to limiter backend, please contact your administrator"), category="error")
# return render_title_template('register.html', config=config, title=_("Register"), page="register")
if current_user is not None and current_user.is_authenticated:
return redirect(url_for('web.index'))
if not config.get_mail_server_configured():
@@ -1358,7 +1347,7 @@ def register():
def handle_login_user(user, remember, message, category):
login_user(user, remember=remember)
flash(message, category=category)
[limiter.limiter.storage.clear(k.key) for k in limiter.current_limits]
[limiter.limiter.clear(limit.limit, *limit.request_args) for limit in limiter.current_limits]
return redirect(get_redirect_location(request.form.get('next', None), "web.index"))
@@ -1392,15 +1381,6 @@ def login():
def login_post():
form = request.form.to_dict()
username = strip_whitespaces(form.get('username', "")).lower().replace("\n","").replace("\r","")
#try:
# limiter.check()
#except RateLimitExceeded:
# flash(_("Please wait one minute before next login"), category="error")
# return render_login(username, form.get("password", ""))
#except (ConnectionError, Exception) as e:
# log.error("Connection error to limiter backend: %s", e)
# flash(_("Connection error to limiter backend, please contact your administrator"), category="error")
# return render_login(username, form.get("password", ""))
if current_user is not None and current_user.is_authenticated:
return redirect(url_for('web.index'))
if config.config_login_type == constants.LOGIN_LDAP and not services.ldap: