Fixes #74 New RTBH records can be created in wrong state

CHANGELOG.md

@@ -5,6 +5,11 @@ All notable changes to ExaFS will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.1.9] - 2025-11-18
+
+### Fixed
+- bugfix for #74 new RTBH records can be created in wrong state
+
 ## [1.1.8] - 2025-10-20
 
 ### Fixed

flowapp/__about__.py

@@ -1,4 +1,4 @@
-__version__ = "1.1.8"
+__version__ = "1.1.9"
 __title__ = "ExaFS"
 __description__ = "Tool for creation, validation, and execution of ExaBGP messages."
 __author__ = "CESNET / Jiri Vrany, Petr Adamec, Josef Verich, Jakub Man"

flowapp/services/rule_service.py

@@ -294,6 +294,7 @@ def create_or_update_rtbh_rule(
     flashes = []
     if model:
         model.expires = round_to_ten_minutes(form_data["expires"])
+        model.rstate_id = get_state_by_time(form_data["expires"])
         flashes.append("Existing RTBH Rule found. Expiration time was updated to new value.")
     else:
         # Create new model

flowapp/tests/test_api_v3.py

@@ -523,7 +523,7 @@ def test_create_rtbh_lmit(client, db, app, jwt_token):
     """
     with app.app_context():
         org = db.session.query(Organization).filter_by(id=1).first()
-        org.limit_rtbh = 2
+        org.limit_rtbh = 1
         db.session.commit()
 
     sources = ["147.230.17.42", "147.230.17.43"]

flowapp/tests/test_zzz_api_rtbh_expired_bug.py

@@ -0,0 +1,240 @@
+import json
+from datetime import datetime, timedelta
+
+from flowapp.models import RTBH
+from flowapp.models.rules.whitelist import Whitelist
+
+
+def test_create_rtbh_after_expired_rule_exists(client, app, db, jwt_token):
+    """
+    Test that demonstrates the bug: creating a new RTBH rule with the same IP
+    as an expired rule results in the new rule having withdrawn state instead
+    of active state.
+
+    Test should be run in isolation or as the last in stack.
+
+    Steps:
+    1. Create an RTBH rule with expiration in the past (will be withdrawn, rstate_id=2)
+    2. Create another RTBH rule with the same IP but expiration in the future
+    3. Verify that the second rule should be active (rstate_id=1) but is actually withdrawn (rstate_id=2)
+    """
+    # cleanup any existing RTBH rules to avoid interference
+    cleanup_before_stack(app, db)
+
+    # Step 1: Create an expired RTBH rule
+    expired_payload = {
+        "community": 1,
+        "ipv4": "192.168.100.50",
+        "ipv4_mask": 32,
+        "expires": (datetime.now() - timedelta(days=1)).strftime("%m/%d/%Y %H:%M"),
+        "comment": "Expired rule that should be in withdrawn state",
+    }
+
+    response1 = client.post(
+        "/api/v3/rules/rtbh",
+        headers={"x-access-token": jwt_token},
+        json=expired_payload,
+    )
+
+    assert response1.status_code == 201
+    data1 = json.loads(response1.data)
+    rule_id_1 = data1["rule"]["id"]
+
+    # Verify the first rule is in withdrawn state
+    with app.app_context():
+        expired_rule = db.session.query(RTBH).filter_by(id=rule_id_1).first()
+        assert expired_rule is not None
+        assert expired_rule.rstate_id == 2, "Expired rule should be in withdrawn state (rstate_id=2)"
+        assert expired_rule.ipv4 == "192.168.100.50"
+        assert expired_rule.ipv4_mask == 32
+        print(f"✓ First rule created with ID {rule_id_1}, state: {expired_rule.rstate_id} (withdrawn)")
+
+    # Step 2: Create a new RTBH rule with the same IP but future expiration
+    future_payload = {
+        "community": 1,
+        "ipv4": "192.168.100.50",
+        "ipv4_mask": 32,
+        "expires": (datetime.now() + timedelta(days=7)).strftime("%m/%d/%Y %H:%M"),
+        "comment": "New rule that should be active but will be withdrawn due to bug",
+    }
+
+    response2 = client.post(
+        "/api/v3/rules/rtbh",
+        headers={"x-access-token": jwt_token},
+        json=future_payload,
+    )
+
+    assert response2.status_code == 201
+    data2 = json.loads(response2.data)
+    rule_id_2 = data2["rule"]["id"]
+
+    # Step 3: Verify the bug - the second rule should be active but is withdrawn
+    with app.app_context():
+        # The bug causes the expired rule to be updated instead of creating a new one
+        # OR if a new rule is created, it has the wrong state
+
+        # Check if it's the same rule (updated) or a new rule
+        total_rules = db.session.query(RTBH).filter_by(ipv4="192.168.100.50", ipv4_mask=32).count()
+
+        new_rule = db.session.query(RTBH).filter_by(id=rule_id_2).first()
+        assert new_rule is not None
+
+        print("\n--- Bug Verification ---")
+        print(f"Total rules with IP 192.168.100.50/32: {total_rules}")
+        print(f"First rule ID: {rule_id_1}")
+        print(f"Second rule ID: {rule_id_2}")
+        print(f"Same rule updated: {rule_id_1 == rule_id_2}")
+        print(f"Second rule state: {new_rule.rstate_id}")
+        print(f"Second rule expires: {new_rule.expires}")
+        print(f"Expiration is in future: {new_rule.expires > datetime.now()}")
+
+        # The bug: even though expiration is in the future, the rule is in withdrawn state
+        # EXPECTED: rstate_id should be 1 (active)
+        # ACTUAL: rstate_id is 2 (withdrawn) due to the bug
+
+        # This assertion will FAIL due to the bug, demonstrating the issue
+        assert new_rule.expires > datetime.now(), "Rule expiration should be in the future"
+
+        # THIS IS THE BUG: The rule has future expiration but is in withdrawn state
+        try:
+            assert new_rule.rstate_id == 1, (
+                f"BUG DETECTED: Rule with future expiration should be active (rstate_id=1), "
+                f"but is in state {new_rule.rstate_id}. "
+                f"This happens because the expired rule was found and updated without resetting the state."
+            )
+            print("✓ Test PASSED - bug is fixed!")
+        except AssertionError as e:
+            print(f"✗ Test FAILED - bug confirmed: {e}")
+            raise
+    cleanup_rtbh_rule(app, db, rule_id_1)
+    cleanup_rtbh_rule(app, db, rule_id_2)
+
+
+def test_create_rtbh_after_expired_rule_different_mask(client, app, db, jwt_token):
+    """
+    Test that verifies the bug only occurs when IP AND mask match.
+    When the mask is different, a new rule should be created successfully.
+    """
+
+    # Step 1: Create an expired RTBH rule with /32 mask
+    expired_payload = {
+        "community": 1,
+        "ipv4": "192.168.100.60",
+        "ipv4_mask": 32,
+        "expires": (datetime.now() - timedelta(days=1)).strftime("%m/%d/%Y %H:%M"),
+        "comment": "Expired /32 rule",
+    }
+
+    response1 = client.post(
+        "/api/v3/rules/rtbh",
+        headers={"x-access-token": jwt_token},
+        json=expired_payload,
+    )
+
+    assert response1.status_code == 201
+
+    # Step 2: Create a new rule with same IP but different mask (/24)
+    future_payload = {
+        "community": 1,
+        "ipv4": "192.168.100.0",
+        "ipv4_mask": 24,
+        "expires": (datetime.now() + timedelta(days=7)).strftime("%m/%d/%Y %H:%M"),
+        "comment": "New /24 rule - should be active",
+    }
+
+    response2 = client.post(
+        "/api/v3/rules/rtbh",
+        headers={"x-access-token": jwt_token},
+        json=future_payload,
+    )
+
+    assert response2.status_code == 201
+    data2 = json.loads(response2.data)
+
+    # Verify the new rule is active (this should work because IP+mask don't match)
+    with app.app_context():
+        new_rule = db.session.query(RTBH).filter_by(id=data2["rule"]["id"]).first()
+        assert new_rule is not None
+        assert new_rule.rstate_id == 1, "New rule with different mask should be active"
+        print("✓ Different mask creates new active rule correctly")
+
+    cleanup_rtbh_rule(app, db, new_rule.id)
+
+
+def test_create_rtbh_after_active_rule_exists(client, app, db, jwt_token):
+    """
+    Test that when an active rule exists, updating it with a new expiration
+    maintains the active state (this should work correctly).
+    """
+
+    # Step 1: Create an active RTBH rule
+    active_payload = {
+        "community": 1,
+        "ipv4": "192.168.100.70",
+        "ipv4_mask": 32,
+        "expires": (datetime.now() + timedelta(days=1)).strftime("%m/%d/%Y %H:%M"),
+        "comment": "Active rule",
+    }
+
+    response1 = client.post(
+        "/api/v3/rules/rtbh",
+        headers={"x-access-token": jwt_token},
+        json=active_payload,
+    )
+
+    assert response1.status_code == 201
+    data1 = json.loads(response1.data)
+    rule_id_1 = data1["rule"]["id"]
+
+    # Verify the first rule is active
+    with app.app_context():
+        first_rule = db.session.query(RTBH).filter_by(id=rule_id_1).first()
+        assert first_rule.rstate_id == 1, "First rule should be active"
+
+    # Step 2: Update the same rule with a new expiration
+    updated_payload = {
+        "community": 1,
+        "ipv4": "192.168.100.70",
+        "ipv4_mask": 32,
+        "expires": (datetime.now() + timedelta(days=7)).strftime("%m/%d/%Y %H:%M"),
+        "comment": "Updated active rule",
+    }
+
+    response2 = client.post(
+        "/api/v3/rules/rtbh",
+        headers={"x-access-token": jwt_token},
+        json=updated_payload,
+    )
+
+    assert response2.status_code == 201
+    data2 = json.loads(response2.data)
+
+    # Verify it maintains active state
+    with app.app_context():
+        updated_rule = db.session.query(RTBH).filter_by(id=data2["rule"]["id"]).first()
+        assert updated_rule is not None
+        assert updated_rule.rstate_id == 1, "Updated rule should remain active"
+        print("✓ Updating active rule maintains active state correctly")
+
+    cleanup_rtbh_rule(app, db, rule_id_1)
+
+
+def cleanup_before_stack(app, db):
+    """
+    Cleanup function to remove all RTBH rules created during tests.
+    """
+    with app.app_context():
+        db.session.query(RTBH).delete()
+        db.session.query(Whitelist).delete()
+        db.session.commit()
+
+
+def cleanup_rtbh_rule(app, db, rule_id):
+    """
+    Cleanup function to remove RTBH rule created during tests.
+    """
+    with app.app_context():
+        rule = db.session.get(RTBH, rule_id)
+        if rule:
+            db.session.delete(rule)
+            db.session.commit()

pyproject.toml

@@ -14,7 +14,7 @@ authors = [
 maintainers = [
     {name = "Jiri Vrany", email = "[email protected]"}
 ]
-license = {text = "MIT"}
+license = "MIT"
 description = "Tool for creation, validation, and execution of ExaBGP messages for network security."
 readme = "README.md"
 requires-python = ">=3.9"
@@ -24,7 +24,6 @@ classifiers = [
     "Intended Audience :: System Administrators",
     "Intended Audience :: Telecommunications Industry",
     "Operating System :: OS Independent",
-    "License :: OSI Approved :: MIT License",
     "Programming Language :: Python :: 3",
     "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",