diff --git a/searx/plugins/limiter.py b/searx/plugins/limiter.py index b66a0805..31ac1e28 100644 --- a/searx/plugins/limiter.py +++ b/searx/plugins/limiter.py @@ -13,10 +13,12 @@ Enable the plugin in ``settings.yml``: - ``redis.url: ...`` check the value, see :ref:`settings redis` """ +import ipaddress import re + from flask import request -from searx import redisdb +from searx import get_setting, redisdb from searx.redislib import incr_sliding_window name = "Request limiter" @@ -36,22 +38,43 @@ re_bot = re.compile( ) +WHITELISTED_IPS = get_setting('server.limiter_whitelist_ip', default=[]) +WHITELISTED_SUBNET = get_setting('server.limiter_whitelist_subnet', default=[]) + + +def is_whitelist_ip(ip): + ''' + Check if the given IP address belongs to the whitelisted list + of IP addresses or subnets. + ''' + return ip in WHITELISTED_IPS or any(ipaddress.ip_address(ip) in + ipaddress.ip_network(subnet) + for subnet in WHITELISTED_SUBNET) + + def is_accepted_request() -> bool: # pylint: disable=too-many-return-statements redis_client = redisdb.client() user_agent = request.headers.get('User-Agent', '') x_forwarded_for = request.headers.get('X-Forwarded-For', '') + # if the request source ip belongs to the whitelisted list of ip addresses or subnets + if is_whitelist_ip(x_forwarded_for): + return True + if request.path == '/image_proxy': if re_bot.match(user_agent): return False return True if request.path == '/search': - c_burst = incr_sliding_window(redis_client, 'IP limit, burst' + x_forwarded_for, 20) - c_10min = incr_sliding_window(redis_client, 'IP limit, 10 minutes' + x_forwarded_for, 600) + c_burst = incr_sliding_window( + redis_client, 'IP limit, burst' + x_forwarded_for, 20) + c_10min = incr_sliding_window( + redis_client, 'IP limit, 10 minutes' + x_forwarded_for, 600) if c_burst > 15 or c_10min > 150: - logger.debug("to many request") # pylint: disable=undefined-variable + logger.debug( + "to many request") # pylint: disable=undefined-variable return False if re_bot.match(user_agent): @@ -59,26 +82,33 @@ def is_accepted_request() -> bool: return False if len(request.headers.get('Accept-Language', '').strip()) == '': - logger.debug("missing Accept-Language") # pylint: disable=undefined-variable + logger.debug( + "missing Accept-Language") # pylint: disable=undefined-variable return False if request.headers.get('Connection') == 'close': - logger.debug("got Connection=close") # pylint: disable=undefined-variable + logger.debug( + "got Connection=close") # pylint: disable=undefined-variable return False - accept_encoding_list = [l.strip() for l in request.headers.get('Accept-Encoding', '').split(',')] + accept_encoding_list = [l.strip() for l in request.headers.get( + 'Accept-Encoding', '').split(',')] if 'gzip' not in accept_encoding_list and 'deflate' not in accept_encoding_list: - logger.debug("suspicious Accept-Encoding") # pylint: disable=undefined-variable + logger.debug( + "suspicious Accept-Encoding") # pylint: disable=undefined-variable return False if 'text/html' not in request.accept_mimetypes: - logger.debug("Accept-Encoding misses text/html") # pylint: disable=undefined-variable + logger.debug( + "Accept-Encoding misses text/html") # pylint: disable=undefined-variable return False if request.args.get('format', 'html') != 'html': - c = incr_sliding_window(redis_client, 'API limit' + x_forwarded_for, 3600) + c = incr_sliding_window( + redis_client, 'API limit' + x_forwarded_for, 3600) if c > 4: - logger.debug("API limit exceeded") # pylint: disable=undefined-variable + logger.debug( + "API limit exceeded") # pylint: disable=undefined-variable return False return True @@ -94,7 +124,8 @@ def init(app, settings): return False if not redisdb.client(): - logger.error("The limiter requires Redis") # pylint: disable=undefined-variable + logger.error( + "The limiter requires Redis") # pylint: disable=undefined-variable return False app.before_request(pre_request)