FlaskFloodgate

FlaskFloodgate#

Module contents#

class FlaskFloodgate.RateLimiter(db: DBHandler, amount: int, time_window: timedelta, block_duration: timedelta, block_limit: int | None = 5, block_exceed_duration: timedelta | Literal['FOREVER'] = datetime.timedelta(days=1), accumulate_requests: bool = False, relative_block: bool = True, max_window_duration: timedelta | Literal['FOREVER'] = datetime.timedelta(days=2), dl_data_wb: bool = True, db_error_retries: int = 3, logger: Logger = None)#

Bases: object

Represents a IP Rate Limit Handler. It helps prevent spam requests and blocks them according to their IPs. If an IP passes the specified request limit per specified time duration, the IP is blocked for the specified block duration. If the IP gets blocked more than the specified block limit, it is blacklisted. Most of the work is done by the specified DB handler.

Parameters:

  • db (DBHandler) – The IP handler to use.

  • amount (datetime.timedelta) – The maximum amount of requests allowed for an IP in time_window time.

  • time_window – The time window in which the specified amount requests are allowed.

  • block_duration (timedelta) – The time for which an IP is blocked (if it is still under the block_limit).

  • block_limit (Union[int, None], optional) – The maximum number of times an IP can be blocked, defaults to 5.

  • block_exceed_duration (Union[datetime.timedelta, Literal[‘FOREVER’]], optional) – The time duration for which an IP is blocked if it exceeds the specified block_limit. If set to ‘FOREVER’, it will be blacklisted forever until specifically removed from the blacklist, defaults to datetime.timedelta(days=1).

  • relative_block (bool, optional) – If set to True, the block_exceed_duration timer (if set to a timedelta object) will reset and start again everytime the IP sends a request during the ongoing block_exceed_duration timer. If set to False, the block_exceed_duration timer (if set to a timedelta object) will not reset and start from the first ever blocked request for the window, defaults to True.

  • max_window_duration (Union[datetime.timedelta, Literal[‘FOREVER’]], optional) – The time duration in which a request window data is removed from the DB. (Does not include the Blacklist and Whitelist DB.) Defaults to datetime.timedelta(days=2).

  • accumulate_requests (bool, optional) – If set to True, it allows an IP to use the left-over amount of requests from the previous time-window along with the new one, defaults to True.

  • dl_data_wb (bool, optional) – Indicates whether to delete the IP data when it is blacklisted or whitelisted, defaults to True.

  • db_error_retries (int, optional) – The number of times to retry in case of of a DB failure/exception, defaults to 2.

  • logger (logging.Logger, optional) – The logger to use, defaults to None.

attempt_func(*, func: Callable, attempts: int, fail_msg: str, args: tuple = (), kwargs: dict = {}, success_msg: str = None, backoff: Literal['Linear', 'Exponential'] = 'Linear')#

Attempts the specified func specified attempts num of times. Only keyword arguments.

Parameters:

  • func (Callable) – The function to call.

  • attempts (int) – The number of attempts to make.

  • fail_msg (str) – The message to log when it fails.

  • args (tuple, optional) – The arguments for the func, defaults to ().

  • kwargs (dict, optional) – The keyword arguments for the func, defaults to {}.

  • success_msg (str, optional) – The message to log when it succeeds, defaults to None.

  • backoff (Literal['Linear', 'Exponential'], optional) – Either ‘Linear’ or ‘Exponential’, defaults to Linear.

export_params(export_fp: str = None)#

Used to export the RateLimiter params. Note that the DBHandler, rule and logger are not exported.

Parameters:

export_fp (str, optional) – The json file, opened with w, where the params are to be exported, defaults to None and the params are returned in a dict.

static load_params(db: DBHandler, export_fp: str = None, rule: Callable[[Request], bool] = None, logger=None)#

Used to load the previously exported parameters.

Parameters:

  • db (DBHandler) – The DBHandler previously used. It needs to be specified while loading the parameters as it is not exported.

  • export_fp (str, optional) – The path of the JSON file where the exported data was stored. If not specified, looks for Rate-Limit-Params.json in the current working dir.

  • rule (Callable[[Flask.Request], bool], optional) – The rule function. It needs to specified (if used earlier) while loading the parameters as it is not exported, defaults to None.

  • logger (logging.Logger) – The logger to use needs to specified (if used earlier) while loading the parameters as the logger is not exported, defaults to None.

Raises:

  • ValueError – Indicates that there is something wrong with the export_fp.

  • json.JSONDecodeError – Indicates that the export_fp does not contain valid JSON data.

  • KeyError – Indicates that the data stored in the file was invalid.

log(msg: str, level: int = 20)#

Used to log messages using the logger. Helps prevent excess lines of checking if the logger is set or not.

Parameters:

  • msg (str) – The message to log.

  • level (int, optional) – The level to log the message, defaults to 20.

rate_limited_route()#

It wraps a Flask route and rate-limits the IPs.

Usage#

from datetime import timedelta
from FlaskFloodgate import RateLimiter
from FlaskFloodgate.handlers import MemoryHandler

rlhandler = RateLimitHandler(
    db=MemoryHandler(),
    amount=30,
    time_window=timedelta(minutes=1),
    block_duration=timedelta(minutes=5)
)

# Initialization of the `Flask` app and other essentials.

@app.route("/rate-limited")
@rlhandler.rate_limited_route()
def rate_limited():
    return "Hello World!", 200

if __name__ == "__main__":
    app.run(host="localhost", port=5000)
set_rule(rule: Callable[[Request], bool], override: bool = False)#

Used to add a function to check for a specific flask.Request object data. You can only add one rule.

The function should return a bool where True indicates that the request is to be exempt from rate-limiting and vice-versa.

Parameters:

  • func (Callable[[flask.Request], bool]) – A function which takes in a flask.Request and returns a bool.

  • override (bool, optional) – If set to True, the previous set rule (if exists) will be replaced with the new specified rule, defaults to False.

Raises:

ValueError – Indicates that either the specified rule is not callable or a rule already exists.

terminal_op()#

Can be used to execute commands during runtime. Is run in a thread.

Currently supported commands:#

  1. whitelist: To whitelist an IP.

  2. de-whitelist: To de-whitelist an IP.

  3. blacklist: To blacklist an IP.

  4. de-blacklist: To de-blacklist an IP.

  5. help: For help.

  6. exit: To exit the FlaskFloodgate terminal.

Usage#

from FlaskFloodgate import RateLimiter
from FlaskFloodgate.handlers import MemoryHandler

rlhandler = RateLimitHandler(
    db=MemoryHandler(),
    amount=30,
    time_window=timedelta(minutes=1),
    block_duration=timedelta(minutes=30) # Configure other params if required.
)

# Initialization of the `Flask` app and other essentials.

@app.route("/rate-limited")
@rate_limited_route()
def rate_limited():
    return "Hello World!", 200

if __name__ == "__main__":
    rlhandler.terminal_op()
    app.run(host="localhost", port=5000)
Contents