| {{ rtbh_all_count }} / {{ rtbh_limit }} |
- {{ flowspec4_all_count }} / {{ flowspec_limit }} |
- {{ flowspec6_all_count }} / {{ flowspec_limit }} |
+ {{ flowspec4_all_count }} / {{ flowspec4_limit }} |
+ {{ flowspec6_all_count }} / {{ flowspec6_limit }} |
diff --git a/flowapp/utils/app_factory.py b/flowapp/utils/app_factory.py
index 234522fd..d8bd30e1 100644
--- a/flowapp/utils/app_factory.py
+++ b/flowapp/utils/app_factory.py
@@ -217,5 +217,4 @@ def _register_user_to_session(uuid, db):
roles = [i > 1 for i in session["user_role_ids"]]
session["can_edit"] = True if all(roles) and roles else []
# check if user has multiple organizations and return True if so
- print(f"DEBUG SESSION {session}")
return user, len(user.organization.all()) > 1
diff --git a/flowapp/views/admin.py b/flowapp/views/admin.py
index 7912e86a..5cc7309e 100644
--- a/flowapp/views/admin.py
+++ b/flowapp/views/admin.py
@@ -311,8 +311,6 @@ def organizations():
flowspec4_all_count = db.session.query(Flowspec4).filter(Flowspec4.rstate_id == 1).count()
flowspec6_all_count = db.session.query(Flowspec6).filter(Flowspec6.rstate_id == 1).count()
rtbh_all_count = db.session.query(RTBH).filter(RTBH.rstate_id == 1).count()
- flowspec_limit = current_app.config.get("FLOWSPEC_MAX_RULES", 9000)
- rtbh_limit = current_app.config.get("RTBH_MAX_RULES", 100000)
# Convert query result to a dictionary {org_id: count}
rtbh_counts = {org_id: count for org_id, count in rtbh_counts_query}
@@ -328,8 +326,9 @@ def organizations():
rtbh_all_count=rtbh_all_count,
flowspec4_all_count=flowspec4_all_count,
flowspec6_all_count=flowspec6_all_count,
- flowspec_limit=flowspec_limit,
- rtbh_limit=rtbh_limit,
+ flowspec4_limit=current_app.config.get("FLOWSPEC4_MAX_RULES", 9000),
+ flowspec6_limit=current_app.config.get("FLOWSPEC6_MAX_RULES", 9000),
+ rtbh_limit=current_app.config.get("RTBH_MAX_RULES", 100000),
)
diff --git a/flowapp/views/api_common.py b/flowapp/views/api_common.py
index 3951b093..2aaef731 100644
--- a/flowapp/views/api_common.py
+++ b/flowapp/views/api_common.py
@@ -1,9 +1,10 @@
import jwt
import ipaddress
+from typing import Dict, Any, List, Tuple, Optional, Union, Callable
+from datetime import datetime, timedelta
-from flask import request, jsonify, current_app
+from flask import request, jsonify, current_app, Response
from functools import wraps
-from datetime import datetime, timedelta
from flowapp.constants import RULE_NAMES_DICT, WITHDRAW, TIME_FORMAT_ARG, RuleTypes
from flowapp.models import (
@@ -33,10 +34,10 @@
from flowapp import db, validators, flowspec, messages
-def token_required(f):
+def token_required(f: Callable) -> Callable:
@wraps(f)
- def decorated(*args, **kwargs):
- token = None
+ def decorated(*args: Any, **kwargs: Any) -> Union[Response, Tuple[Response, int]]:
+ token: Optional[str] = None
if "x-access-token" in request.headers:
token = request.headers["x-access-token"]
@@ -45,8 +46,8 @@ def decorated(*args, **kwargs):
return jsonify({"message": "auth token is missing"}), 401
try:
- data = jwt.decode(token, current_app.config.get("JWT_SECRET"), algorithms=["HS256"])
- current_user = data["user"]
+ data: Dict[str, Any] = jwt.decode(token, current_app.config.get("JWT_SECRET"), algorithms=["HS256"])
+ current_user: Dict[str, Any] = data["user"]
except jwt.DecodeError:
return jsonify({"message": "auth token is invalid"}), 403
except jwt.ExpiredSignatureError:
@@ -57,14 +58,14 @@ def decorated(*args, **kwargs):
return decorated
-def authorize(user_key):
+def authorize(user_key: str) -> Tuple[Response, int]:
"""
- Generate API Key for the loged user using PyJWT
+ Generate API Key for the logged user using PyJWT
:return: page with token
"""
- jwt_key = current_app.config.get("JWT_SECRET")
+ jwt_key: Optional[str] = current_app.config.get("JWT_SECRET")
# try normal user key first
- model = db.session.query(ApiKey).filter_by(key=user_key).first()
+ model: Optional[Union[ApiKey, MachineApiKey]] = db.session.query(ApiKey).filter_by(key=user_key).first()
# if not found try machine key
if not model:
model = db.session.query(MachineApiKey).filter_by(key=user_key).first()
@@ -78,7 +79,7 @@ def authorize(user_key):
# check if the key is not used by different machine
if model and ipaddress.ip_address(model.machine) == ipaddress.ip_address(request.remote_addr):
- payload = {
+ payload: Dict[str, Any] = {
"user": {
"uuid": model.user.uuid,
"id": model.user.id,
@@ -91,24 +92,24 @@ def authorize(user_key):
"exp": datetime.now() + timedelta(minutes=30),
}
# encoded = jwt.encode(payload, jwt_key, algorithm="HS256").decode("utf-8")
- encoded = jwt.encode(payload, jwt_key, algorithm="HS256")
+ encoded: str = jwt.encode(payload, jwt_key, algorithm="HS256")
return jsonify({"token": encoded})
else:
return jsonify({"message": f"auth token is not valid from machine {request.remote_addr}"}), 403
-def check_readonly(func):
+def check_readonly(func: Callable) -> Callable:
"""
Check if the token is readonly
Used in api endpoints
"""
@wraps(func)
- def decorated_function(*args, **kwargs):
+ def decorated_function(*args: Any, **kwargs: Any) -> Union[Response, Tuple[Response, int]]:
# Access read only flag from first of the args
- current_user = kwargs.get("current_user", False)
- read_only = current_user.get("readonly", False)
+ current_user: Dict[str, Any] = kwargs.get("current_user", {})
+ read_only: bool = current_user.get("readonly", False)
if read_only:
return jsonify({"message": "read only token can't perform this action"}), 403
return func(*args, **kwargs)
@@ -116,20 +117,20 @@ def decorated_function(*args, **kwargs):
return decorated_function
-# endpints
+# endpoints
-def index(current_user, key_map):
- prefered_tf = request.args.get(TIME_FORMAT_ARG) if request.args.get(TIME_FORMAT_ARG) else ""
+def index(current_user: Dict[str, Any], key_map: Dict[str, str]) -> Response:
+ prefered_tf: str = request.args.get(TIME_FORMAT_ARG) if request.args.get(TIME_FORMAT_ARG) else ""
- net_ranges = get_user_nets(current_user["id"])
- rules4 = db.session.query(Flowspec4).order_by(Flowspec4.expires.desc()).all()
- rules6 = db.session.query(Flowspec6).order_by(Flowspec6.expires.desc()).all()
- rules_rtbh = db.session.query(RTBH).order_by(RTBH.expires.desc()).all()
+ net_ranges: List[str] = get_user_nets(current_user["id"])
+ rules4: List[Flowspec4] = db.session.query(Flowspec4).order_by(Flowspec4.expires.desc()).all()
+ rules6: List[Flowspec6] = db.session.query(Flowspec6).order_by(Flowspec6.expires.desc()).all()
+ rules_rtbh: List[RTBH] = db.session.query(RTBH).order_by(RTBH.expires.desc()).all()
# admin can see and edit any rules
if 3 in current_user["role_ids"]:
- payload = {
+ payload: Dict[str, List[Dict[str, Any]]] = {
key_map["ipv4_rules"]: [rule.to_dict(prefered_tf) for rule in rules4],
key_map["ipv6_rules"]: [rule.to_dict(prefered_tf) for rule in rules6],
key_map["rtbh_rules"]: [rule.to_dict(prefered_tf) for rule in rules_rtbh],
@@ -141,11 +142,16 @@ def index(current_user, key_map):
rules6 = validators.filter_rules_in_network(net_ranges, rules6)
rules_rtbh = validators.filter_rtbh_rules(net_ranges, rules_rtbh)
- user_actions = get_user_actions(current_user["role_ids"])
- user_actions = [act[0] for act in user_actions]
+ user_actions: List[Tuple[int, str]] = get_user_actions(current_user["role_ids"])
+ user_actions_ids: List[int] = [act[0] for act in user_actions]
- rules4_editable, rules4_visible = flowspec.filter_rules_action(user_actions, rules4)
- rules6_editable, rules6_visible = flowspec.filter_rules_action(user_actions, rules6)
+ rules4_editable: List[Flowspec4]
+ rules4_visible: List[Flowspec4]
+ rules4_editable, rules4_visible = flowspec.filter_rules_action(user_actions_ids, rules4)
+
+ rules6_editable: List[Flowspec6]
+ rules6_visible: List[Flowspec6]
+ rules6_editable, rules6_visible = flowspec.filter_rules_action(user_actions_ids, rules6)
payload = {
key_map["ipv4_rules"]: [rule.to_dict(prefered_tf) for rule in rules4_editable],
@@ -157,43 +163,43 @@ def index(current_user, key_map):
return jsonify(payload)
-def all_actions(current_user):
+def all_actions(current_user: Dict[str, Any]) -> Tuple[Response, int]:
"""
Returns Actions allowed for current user
:param current_user:
:return: json response
"""
- actions = get_user_actions(current_user["role_ids"])
+ actions: List[Tuple[int, str]] = get_user_actions(current_user["role_ids"])
if actions:
return jsonify(actions)
else:
return jsonify({"message": "no actions for this user?"}), 404
-def all_communities(current_user):
+def all_communities(current_user: Dict[str, Any]) -> Tuple[Response, int]:
"""
- Returns RTHB communites allowed for current user
+ Returns RTBH communities allowed for current user
:param current_user:
:return: json response
"""
- coms = get_user_communities(current_user["role_ids"])
+ coms: List[Tuple[int, str]] = get_user_communities(current_user["role_ids"])
if coms:
return jsonify(coms)
else:
return jsonify({"message": "no actions for this user?"}), 404
-def limit_reached(count, rule_type, org_id):
- rule_name = RULE_NAMES_DICT[rule_type.value]
- org = db.session.get(Organization, org_id)
+def limit_reached(count: int, rule_type: RuleTypes, org_id: int) -> Tuple[Response, int]:
+ rule_name: str = RULE_NAMES_DICT[rule_type.value]
+ org: Optional[Organization] = db.session.get(Organization, org_id)
if rule_type == RuleTypes.IPv4:
- limit = org.limit_flowspec4
+ limit: int = org.limit_flowspec4
elif rule_type == RuleTypes.IPv6:
limit = org.limit_flowspec6
elif rule_type == RuleTypes.RTBH:
- limit = org.rtbh
+ limit = org.limit_rtbh
return (
jsonify({"message": f"Rule limit {limit} reached for {rule_name}, currently you have {count} active rules."}),
@@ -201,12 +207,14 @@ def limit_reached(count, rule_type, org_id):
)
-def global_limit_reached(count, rule_type):
- rule_name = RULE_NAMES_DICT[rule_type.value]
- if rule_type == RuleTypes.IPv4 or rule_type == RuleTypes.IPv6:
- limit = current_app.config.get("FLOWSPEC_MAX_RULES")
+def global_limit_reached(count: int, rule_type: RuleTypes) -> Tuple[Response, int]:
+ rule_name: str = RULE_NAMES_DICT[rule_type.value]
+ if rule_type == RuleTypes.IPv4:
+ limit: int = current_app.config.get("FLOWSPEC4_MAX_RULES", 9000)
+ elif rule_type == RuleTypes.IPv6:
+ limit = current_app.config.get("FLOWSPEC6_MAX_RULES", 9000)
elif rule_type == RuleTypes.RTBH:
- limit = current_app.config.get("RTBH_MAX_RULES")
+ limit = current_app.config.get("RTBH_MAX_RULES", 100000)
return (
jsonify(
@@ -216,35 +224,36 @@ def global_limit_reached(count, rule_type):
)
-def create_ipv4(current_user):
+def create_ipv4(current_user: Dict[str, Any]) -> Tuple[Response, int]:
"""
Api method for new IPv4 rule
- :param data: parsed json request
:param current_user: data from jwt token
:return: json response
"""
if check_global_rule_limit(RuleTypes.IPv4):
- count = db.session.query(Flowspec4).filter_by(rstate_id=1).count()
+ count: int = db.session.query(Flowspec4).filter_by(rstate_id=1).count()
return global_limit_reached(count=count, rule_type=RuleTypes.IPv4)
if check_rule_limit(current_user["org_id"], RuleTypes.IPv4):
count = db.session.query(Flowspec4).filter_by(rstate_id=1, org_id=current_user["org_id"]).count()
return limit_reached(count=count, rule_type=RuleTypes.IPv4, org_id=current_user["org_id"])
- net_ranges = get_user_nets(current_user["id"])
- json_request_data = request.get_json()
- form = IPv4Form(data=json_request_data, meta={"csrf": False})
+ net_ranges: List[str] = get_user_nets(current_user["id"])
+ json_request_data: Optional[Dict[str, Any]] = request.get_json()
+ form: IPv4Form = IPv4Form(data=json_request_data, meta={"csrf": False})
# add values to form instance
form.action.choices = get_user_actions(current_user["role_ids"])
form.net_ranges = net_ranges
# if the form is not valid, we should return 404 with errors
if not form.validate():
- form_errors = get_form_errors(form)
+ form_errors: Union[Dict[str, Any], bool] = get_form_errors(form)
if form_errors:
return jsonify(form_errors), 400
# Use the service to create/update the rule
+ model: Flowspec4
+ flash_message: str
model, flash_message = rule_service.create_or_update_ipv4_rule(
form_data=form.data,
user_id=current_user["id"],
@@ -253,37 +262,38 @@ def create_ipv4(current_user):
org_name=current_user["org"],
)
- pref_format = output_date_format(json_request_data, form.expires.pref_format)
- response = {"message": flash_message, "rule": model.to_dict(pref_format)}
+ pref_format: str = output_date_format(json_request_data, form.expires.pref_format)
+ response: Dict[str, Any] = {"message": flash_message, "rule": model.to_dict(pref_format)}
return jsonify(response), 201
-def create_ipv6(current_user):
+def create_ipv6(current_user: Dict[str, Any]) -> Tuple[Response, int]:
"""
Create new IPv6 rule
- :param data: parsed json request
:param current_user: data from jwt token
:return:
"""
if check_global_rule_limit(RuleTypes.IPv6):
- count = db.session.query(Flowspec6).filter_by(rstate_id=1).count()
+ count: int = db.session.query(Flowspec6).filter_by(rstate_id=1).count()
return global_limit_reached(count=count, rule_type=RuleTypes.IPv6)
if check_rule_limit(current_user["org_id"], RuleTypes.IPv6):
count = db.session.query(Flowspec6).filter_by(rstate_id=1, org_id=current_user["org_id"]).count()
return limit_reached(count=count, rule_type=RuleTypes.IPv6, org_id=current_user["org_id"])
- net_ranges = get_user_nets(current_user["id"])
- json_request_data = request.get_json()
- form = IPv6Form(data=json_request_data, meta={"csrf": False})
+ net_ranges: List[str] = get_user_nets(current_user["id"])
+ json_request_data: Optional[Dict[str, Any]] = request.get_json()
+ form: IPv6Form = IPv6Form(data=json_request_data, meta={"csrf": False})
form.action.choices = get_user_actions(current_user["role_ids"])
form.net_ranges = net_ranges
if not form.validate():
- form_errors = get_form_errors(form)
+ form_errors: Union[Dict[str, Any], bool] = get_form_errors(form)
if form_errors:
return jsonify(form_errors), 400
+ model: Flowspec6
+ flash_message: str
model, flash_message = rule_service.create_or_update_ipv6_rule(
form_data=form.data,
user_id=current_user["id"],
@@ -292,39 +302,41 @@ def create_ipv6(current_user):
org_name=current_user["org"],
)
- pref_format = output_date_format(json_request_data, form.expires.pref_format)
+ pref_format: str = output_date_format(json_request_data, form.expires.pref_format)
return jsonify({"message": flash_message, "rule": model.to_dict(pref_format)}), 201
-def create_rtbh(current_user):
+def create_rtbh(current_user: Dict[str, Any]) -> Tuple[Response, int]:
"""
Create new RTBH rule
"""
if check_global_rule_limit(RuleTypes.RTBH):
- count = db.session.query(RTBH).filter_by(rstate_id=1).count()
+ count: int = db.session.query(RTBH).filter_by(rstate_id=1).count()
return global_limit_reached(count=count, rule_type=RuleTypes.RTBH)
if check_rule_limit(current_user["org_id"], RuleTypes.RTBH):
count = db.session.query(RTBH).filter_by(rstate_id=1, org_id=current_user["org_id"]).count()
return limit_reached(count=count, rule_type=RuleTypes.RTBH, org_id=current_user["org_id"])
- all_com = db.session.query(Community).all()
+ all_com: List[Community] = db.session.query(Community).all()
if not all_com:
insert_initial_communities()
- net_ranges = get_user_nets(current_user["id"])
+ net_ranges: List[str] = get_user_nets(current_user["id"])
- json_request_data = request.get_json()
- form = RTBHForm(data=json_request_data, meta={"csrf": False})
+ json_request_data: Optional[Dict[str, Any]] = request.get_json()
+ form: RTBHForm = RTBHForm(data=json_request_data, meta={"csrf": False})
form.community.choices = get_user_communities(current_user["role_ids"])
form.net_ranges = net_ranges
if not form.validate():
- form_errors = get_form_errors(form)
+ form_errors: Union[Dict[str, Any], bool] = get_form_errors(form)
if form_errors:
return jsonify(form_errors), 400
+ model: RTBH
+ flash_message: List[str]
model, flash_message = rule_service.create_or_update_rtbh_rule(
form_data=form.data,
user_id=current_user["id"],
@@ -333,50 +345,52 @@ def create_rtbh(current_user):
org_name=current_user["org"],
)
- pref_format = output_date_format(json_request_data, form.expires.pref_format)
+ pref_format: str = output_date_format(json_request_data, form.expires.pref_format)
return jsonify({"message": flash_message, "rule": model.to_dict(pref_format)}), 201
-def ipv4_rule_get(current_user, rule_id):
+def ipv4_rule_get(current_user: Dict[str, Any], rule_id: int) -> Tuple[Response, int]:
"""
Return IPv4 rule
:param current_user:
:param rule_id:
:return:
"""
- model = db.session.get(Flowspec4, rule_id)
+ model: Optional[Flowspec4] = db.session.get(Flowspec4, rule_id)
return get_rule(current_user, model, rule_id)
-def ipv6_rule_get(current_user, rule_id):
+def ipv6_rule_get(current_user: Dict[str, Any], rule_id: int) -> Tuple[Response, int]:
"""
Return IPv6 rule
:param current_user:
:param rule_id:
:return:
"""
- model = db.session.get(Flowspec6, rule_id)
+ model: Optional[Flowspec6] = db.session.get(Flowspec6, rule_id)
return get_rule(current_user, model, rule_id)
-def rtbh_rule_get(current_user, rule_id):
+def rtbh_rule_get(current_user: Dict[str, Any], rule_id: int) -> Tuple[Response, int]:
"""
Return RTBH rule
:param current_user:
:param rule_id:
:return:
"""
- model = db.session.query(RTBH).get(rule_id)
+ model: Optional[RTBH] = db.session.query(RTBH).get(rule_id)
return get_rule(current_user, model, rule_id)
-def get_rule(current_user, model, rule_id):
+def get_rule(
+ current_user: Dict[str, Any], model: Optional[Union[Flowspec4, Flowspec6, RTBH]], rule_id: int
+) -> Tuple[Response, int]:
"""
Common rule getter - return ipv4 or ipv6 model data
:param model: rule model
:return: json
"""
- prefered_tf = request.args.get(TIME_FORMAT_ARG) if request.args.get(TIME_FORMAT_ARG) else ""
+ prefered_tf: str = request.args.get(TIME_FORMAT_ARG) if request.args.get(TIME_FORMAT_ARG) else ""
if model:
if check_access_rights(current_user, model.user_id):
@@ -387,7 +401,7 @@ def get_rule(current_user, model, rule_id):
return jsonify({"message": "rule {} not found ".format(rule_id)}), 404
-def delete_v4_rule(current_user, rule_id):
+def delete_v4_rule(current_user: Dict[str, Any], rule_id: int) -> Tuple[Response, int]:
"""
Delete rule with given id and type
:param rule_id: integer - rule id
@@ -397,7 +411,7 @@ def delete_v4_rule(current_user, rule_id):
return delete_rule(current_user, rule_id, model_name, route_model, RuleTypes.IPv4)
-def delete_v6_rule(current_user, rule_id):
+def delete_v6_rule(current_user: Dict[str, Any], rule_id: int) -> Tuple[Response, int]:
"""
Delete rule with given id and type
:param rule_id: integer - rule id
@@ -407,7 +421,7 @@ def delete_v6_rule(current_user, rule_id):
return delete_rule(current_user, rule_id, model_name, route_model, RuleTypes.IPv6)
-def delete_rtbh_rule(current_user, rule_id):
+def delete_rtbh_rule(current_user: Dict[str, Any], rule_id: int) -> Tuple[Response, int]:
"""
Delete rule with given id and type
:param rule_id: integer - rule id
@@ -417,7 +431,9 @@ def delete_rtbh_rule(current_user, rule_id):
return delete_rule(current_user, rule_id, model_name, route_model, RuleTypes.RTBH)
-def delete_rule(current_user, rule_id, model_name, route_model, rule_type):
+def delete_rule(
+ current_user: Dict[str, Any], rule_id: int, model_name: type, route_model: Callable, rule_type: RuleTypes
+) -> Tuple[Response, int]:
"""
Common method for deleting ipv4 or ipv6 rules
:param current_user:
@@ -426,12 +442,12 @@ def delete_rule(current_user, rule_id, model_name, route_model, rule_type):
:param route_model:
:return:
"""
- model = db.session.get(model_name, rule_id)
+ model: Optional[Union[Flowspec4, Flowspec6, RTBH]] = db.session.get(model_name, rule_id)
if model:
if check_access_rights(current_user, model.user_id):
# withdraw route
- command = route_model(model, WITHDRAW)
- route = Route(
+ command: str = route_model(model, WITHDRAW)
+ route: Route = Route(
author=f"{current_user['uuid']} / {current_user['org']}",
source=RouteSources.API,
command=command,
@@ -455,14 +471,13 @@ def delete_rule(current_user, rule_id, model_name, route_model, rule_type):
return jsonify({"message": "rule {} not found ".format(rule_id)}), 404
-def token_test_get(current_user):
+def token_test_get(current_user: Dict[str, Any]) -> Tuple[Response, int]:
"""
Return IPv4 rule
:param current_user:
- :param rule_id:
:return:
"""
- my_response = {
+ my_response: Dict[str, Any] = {
"message": "token works as expected",
"uuid": current_user["uuid"],
"readonly": current_user["readonly"],
@@ -470,7 +485,7 @@ def token_test_get(current_user):
return jsonify(my_response), 200
-def get_form_errors(form):
+def get_form_errors(form: Union[IPv4Form, IPv6Form, RTBHForm]) -> Union[Dict[str, Any], bool]:
# if the only error is in CSRF then it is ok - csrf is exempt for this view
try:
del form.errors["csrf_token"]
diff --git a/flowapp/views/dashboard.py b/flowapp/views/dashboard.py
index c709f11f..44319272 100644
--- a/flowapp/views/dashboard.py
+++ b/flowapp/views/dashboard.py
@@ -59,10 +59,8 @@ def index(rtype=None, rstate="active"):
# params sanitization
if rtype not in current_app.config["DASHBOARD"].keys():
- print("DEBUG rtype not in dashboard keys config")
return abort(404)
if rstate not in COMP_FUNCS.keys():
- print("DEBUG rstate not in dashboard keys config")
return abort(404)
if sum(session["user_role_ids"]) == 1:
rstate = "active"
diff --git a/flowapp/views/rules.py b/flowapp/views/rules.py
index e3b82a1a..7a50dffb 100644
--- a/flowapp/views/rules.py
+++ b/flowapp/views/rules.py
@@ -502,7 +502,6 @@ def ipv4_rule():
for error in errors:
current_app.logger.debug("Error in the %s field - %s" % (getattr(form, field).label.text, error))
- print("NOW", datetime.now())
default_expires = datetime.now() + timedelta(hours=1)
form.expires.data = default_expires
@@ -598,8 +597,6 @@ def rtbh_rule():
default_expires = datetime.now() + timedelta(days=7)
form.expires.data = default_expires
- print(whitelistable)
-
return render_template(
"forms/rtbh_rule.html", form=form, action_url=url_for("rules.rtbh_rule"), whitelistable=whitelistable
)
diff --git a/flowapp/views/whitelist.py b/flowapp/views/whitelist.py
index 39a9352a..9cc85e4a 100644
--- a/flowapp/views/whitelist.py
+++ b/flowapp/views/whitelist.py
@@ -40,7 +40,6 @@ def add():
for error in errors:
current_app.logger.debug("Error in the %s field - %s" % (getattr(form, field).label.text, error))
- print("NOW", datetime.now())
default_expires = datetime.now() + timedelta(hours=1)
form.expires.data = default_expires
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 00000000..844b05c2
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,66 @@
+[build-system]
+requires = ["setuptools>=61.0", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "exafs"
+dynamic = ["version"]
+authors = [
+ {name = "CESNET / Jiri Vrany, Petr Adamec, Josef Verich, Jakub Man"}
+]
+description = "Tool for creation, validation, and execution of ExaBGP messages."
+readme = "README.md"
+license = {text = "MIT"}
+requires-python = ">=3.9"
+classifiers = [
+ "Development Status :: 4 - Beta",
+ "Intended Audience :: Developers",
+ "License :: OSI Approved :: MIT License",
+ "Programming Language :: Python :: 3",
+ "Programming Language :: Python :: 3.9",
+ "Programming Language :: Python :: 3.10",
+ "Programming Language :: Python :: 3.11",
+ "Programming Language :: Python :: 3.12",
+ "Programming Language :: Python :: 3.13",
+]
+dependencies = [
+ "Flask>=2.0.2",
+ "Flask-SQLAlchemy>=2.2",
+ "Flask-SSO>=0.4.0",
+ "Flask-WTF>=1.0.0",
+ "Flask-Migrate>=3.0.0",
+ "Flask-Script>=2.0.0",
+ "Flask-Session",
+ "PyJWT>=2.4.0",
+ "PyMySQL>=1.0.0",
+ "requests>=2.20.0",
+ "babel>=2.7.0",
+ "email_validator>=1.1",
+ "pika>=1.3.0",
+ "loguru",
+ "flasgger",
+ "python-dotenv",
+]
+
+[project.optional-dependencies]
+dev = [
+ "pytest>=7.0.0",
+ # Add other dev-only dependencies here if needed
+ # "black>=22.0",
+ # "mypy>=0.950",
+ # "flake8>=4.0",
+]
+
+[project.urls]
+Homepage = "https://github.com/CESNET/exafs"
+Repository = "https://github.com/CESNET/exafs"
+Issues = "https://github.com/CESNET/exafs/issues"
+
+[tool.setuptools.dynamic]
+version = {attr = "flowapp.__about__.__version__"}
+
+[tool.setuptools.packages.find]
+include = ["flowapp*"]
+
+[tool.setuptools.package-data]
+"*" = ["*"]
\ No newline at end of file
diff --git a/requirements-backup.txt b/requirements-backup.txt
deleted file mode 100644
index 33cf1c1e..00000000
--- a/requirements-backup.txt
+++ /dev/null
@@ -1,38 +0,0 @@
-appdirs>=1.4.2
-asn1crypto>=0.24.0
-attrs>=17.4.0
-blinker>=1.4
-certifi>=2018.8.24
-cffi>=1.11.5
-chardet>=3.0.4
-click>=6.7
-cryptography>=2.3
-cryptography-vectors>=2.3.1
-enum34>=1.1.6
-Flask>=1.0.2
-Flask-SQLAlchemy>=2.2
-Flask-SSO>=0.4.0
-Flask-WTF>=0.14.2
-funcsigs>=1.0.2
-honcho>=0.7.1
-idna>=2.7
-ipaddress>=1.0.22
-itsdangerous>=0.24
-Jinja2>=2.10
-MarkupSafe>=0.23
-MySQL-python>=1.2.5
-packaging>=16.8
-pluggy>=0.6.0
-py>=1.5.2
-pycparser>=2.18
-PyJWT>=1.6.4
-PyMySQL>=0.7.10
-pyparsing>=2.1.10
-pytest>=3.4.0
-requests>=2.20.0
-six>=1.11.0
-SQLAlchemy>=1.1.6
-urllib3>=1.21.1
-uWSGI>=2.0.14
-Werkzeug>=0.14.1
-WTForms>=2.1
diff --git a/requirements.txt b/requirements.txt
deleted file mode 100644
index fb33ad36..00000000
--- a/requirements.txt
+++ /dev/null
@@ -1,17 +0,0 @@
-Flask>=2.0.2
-Flask-SQLAlchemy>=2.2
-Flask-SSO>=0.4.0
-Flask-WTF>=1.0.0
-Flask-Migrate>=3.0.0
-Flask-Script>=2.0.0
-Flask-Session
-PyJWT>=2.4.0
-PyMySQL>=1.0.0
-pytest>=7.0.0
-requests>=2.20.0
-babel>=2.7.0
-email_validator>=1.1
-pika>=1.3.0
-loguru
-flasgger
-python-dotenv
\ No newline at end of file
diff --git a/setup.py b/setup.py
deleted file mode 100755
index c28a06d8..00000000
--- a/setup.py
+++ /dev/null
@@ -1,46 +0,0 @@
-"""
-Author(s):
-Jiri Vrany