feature: Add check on waf ip blocklist to white list gc owned ip addresses (#726)

Co-authored-by: Pat Heard <[email protected]>

Author: bryan-robitaille

waf_ip_blocklist/lambda/blocklist.py

@@ -7,6 +7,9 @@
 import os
 import time
 import boto3
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
 
 # Required
 ATHENA_OUTPUT_BUCKET = os.environ["ATHENA_OUTPUT_BUCKET"]
@@ -129,6 +132,9 @@ def update_waf_ip_set(ip_addresses, waf_ip_set_name, waf_ip_set_id, waf_scope):
         print(f"Reducing {len(ip_addresses)} address to 10,000 addresses.")
         ip_addresses = ip_addresses[:10000]
 
+    # Remove any ip addresses known to be owned by the Government of Canada
+    ip_addresses = [ip for ip in ip_addresses if not gc_ip(ip)]
+
     # Check if new addresses have been added to the list of existing addresses.
     # This is useful to monitor the number of new IPs added to the blocklist
     # and to set up alarms if the number of new IPs added is too high
@@ -155,3 +161,61 @@ def update_waf_ip_set(ip_addresses, waf_ip_set_name, waf_ip_set_id, waf_scope):
     print(
         f"Updated WAF IP set with {new_ips} new IPs for a total of {len(ip_addresses)} blocked IPs."
     )
+
+
+def recursive_entity_search(data):
+    """Recursively search through a list of entities to see if one matches GoC"""
+    if isinstance(data, list):
+        registrants = [d for d in data if "registrant" in d.get("roles", [])]
+        top_level_result = any(
+            entity.get("handle") == "SSC-299" for entity in registrants
+        )
+        if not top_level_result:
+            # Go deeper and see if there are any other entities associated with this record
+            for entity in registrants:
+                top_level_result = top_level_result or recursive_entity_search(entity)
+                # short circuit once we hit a positive identification
+                if top_level_result:
+                    return top_level_result
+        return top_level_result
+    if isinstance(data, dict) and "entities" in data:
+        return recursive_entity_search(data["entities"])
+    return False
+
+
+def create_retrying_request(
+    total_retries=3,
+    backoff_factor=0.5,
+    status_forcelist=(500, 502, 503, 504),
+    allowed_methods=("GET"),
+):
+    """
+    Creates a requests session with retry logic.
+    """
+    retry_strategy = Retry(
+        total=total_retries,
+        backoff_factor=backoff_factor,
+        status_forcelist=status_forcelist,
+        allowed_methods=allowed_methods,
+    )
+    adapter = HTTPAdapter(max_retries=retry_strategy)
+    session = requests.Session()
+    session.mount("https://", adapter)
+    return session
+
+
+def gc_ip(ip):
+    """Check if IP is owned by Government of Canada"""
+    session = create_retrying_request(total_retries=5, backoff_factor=1)
+    try:
+        api_url = f"https://rdap.arin.net/registry/ip/{ip}"
+        is_gc_ip = False
+        response = session.get(api_url, timeout=5)
+        if response.ok:
+            record = response.json().get("entities")
+            if record is not None:
+                is_gc_ip = recursive_entity_search(record)
+        return is_gc_ip
+    except requests.exceptions.RequestException:
+        print(f"Could not successfully retrieve information about IP {ip}")
+        return False

waf_ip_blocklist/lambda/blocklist_test.py

@@ -1,7 +1,8 @@
 import tempfile
 import os
+import json
 
-from unittest.mock import call, patch
+from unittest.mock import call, patch, Mock, MagicMock
 
 os.environ["AWS_DEFAULT_REGION"] = "ca-central-1"
 os.environ["ATHENA_OUTPUT_BUCKET"] = "test_bucket"
@@ -31,6 +32,7 @@ def test_handler_with_ips_to_block(mock_waf_client, mock_athena_client, capsys):
                     {"Data": [{"VarCharValue": "header"}]},
                     {"Data": [{"VarCharValue": "192.168.1.1"}]},
                     {"Data": [{"VarCharValue": "192.168.1.2"}]},
+                    {"Data": [{"VarCharValue": "198.103.1.2"}]},
                 ]
             }
         },
@@ -40,6 +42,7 @@ def test_handler_with_ips_to_block(mock_waf_client, mock_athena_client, capsys):
                     {"Data": [{"VarCharValue": "header"}]},
                     {"Data": [{"VarCharValue": "192.168.1.1"}]},
                     {"Data": [{"VarCharValue": "192.168.1.3"}]},
+                    {"Data": [{"VarCharValue": "205.193.1.2"}]},
                 ]
             }
         },
@@ -282,3 +285,213 @@ def test_get_query_from_file_with_single_rule_id():
 
     # Cleanup
     os.remove(temp_file_path)
+
+
+def test_recursive_entity_search_with_list_and_goc_handle():
+    """Test recursive_entity_search with a list containing GoC handle"""
+    test_data = [
+        {"roles": ["registrant"], "handle": "SSC-299"},
+        {"roles": ["admin"], "handle": "OTHER-123"},
+    ]
+
+    result = blocklist.recursive_entity_search(test_data)
+    assert result is True
+
+
+def test_recursive_entity_search_with_list_without_goc_handle():
+    """Test recursive_entity_search with a list not containing GoC handle"""
+    test_data = [
+        {"roles": ["registrant"], "handle": "OTHER-123"},
+        {"roles": ["admin"], "handle": "ANOTHER-456"},
+    ]
+
+    result = blocklist.recursive_entity_search(test_data)
+    assert result is False
+
+
+def test_recursive_entity_search_with_nested_entities():
+    """Test recursive_entity_search with nested entities containing GoC handle"""
+    test_data = [
+        {
+            "roles": ["registrant"],
+            "handle": "OTHER-123",
+            "entities": [{"roles": ["registrant"], "handle": "SSC-299"}],
+        }
+    ]
+
+    result = blocklist.recursive_entity_search(test_data)
+    assert result is True
+
+
+def test_recursive_entity_search_with_dict_containing_entities():
+    """Test recursive_entity_search with dict containing entities key"""
+    test_data = {"entities": [{"roles": ["registrant"], "handle": "SSC-299"}]}
+
+    result = blocklist.recursive_entity_search(test_data)
+    assert result is True
+
+
+def test_recursive_entity_search_with_empty_list():
+    """Test recursive_entity_search with empty list"""
+    test_data = []
+
+    result = blocklist.recursive_entity_search(test_data)
+    assert result is False
+
+
+def test_recursive_entity_search_with_none():
+    """Test recursive_entity_search with None input"""
+    result = blocklist.recursive_entity_search(None)
+    assert result is False
+
+
+def test_recursive_entity_search_with_no_registrants():
+    """Test recursive_entity_search with list containing no registrants"""
+    test_data = [
+        {"roles": ["admin"], "handle": "SSC-299"},
+        {"roles": ["tech"], "handle": "OTHER-123"},
+    ]
+
+    result = blocklist.recursive_entity_search(test_data)
+    assert result is False
+
+
+def test_recursive_entity_search_with_missing_roles():
+    """Test recursive_entity_search with entities missing roles field"""
+    test_data = [
+        {"handle": "SSC-299"},  # Missing roles field
+        {"roles": ["admin"], "handle": "OTHER-123"},
+    ]
+
+    result = blocklist.recursive_entity_search(test_data)
+    assert result is False
+
+
+@patch("blocklist.create_retrying_request")
+def test_gc_ip_success_with_goc_ip(mock_create_session):
+    """Test gc_ip with successful response indicating GoC IP"""
+    # Setup mock response
+    mock_response = Mock()
+    mock_response.ok = True
+    mock_response.json.return_value = {
+        "entities": [{"roles": ["registrant"], "handle": "SSC-299"}]
+    }
+
+    # Setup mock session
+    mock_session = Mock()
+    mock_session.get.return_value = mock_response
+    mock_create_session.return_value = mock_session
+
+    # Test
+    result = blocklist.gc_ip("192.168.1.1")
+
+    # Verify
+    assert result is True
+    mock_create_session.assert_called_once_with(total_retries=5, backoff_factor=1)
+    mock_session.get.assert_called_once_with(
+        "https://rdap.arin.net/registry/ip/192.168.1.1", timeout=5
+    )
+
+
+@patch("blocklist.create_retrying_request")
+def test_gc_ip_success_with_non_goc_ip(mock_create_session):
+    """Test gc_ip with successful response indicating non-GoC IP"""
+    # Setup mock response
+    mock_response = Mock()
+    mock_response.ok = True
+    mock_response.json.return_value = {
+        "entities": [{"roles": ["registrant"], "handle": "OTHER-123"}]
+    }
+
+    # Setup mock session
+    mock_session = Mock()
+    mock_session.get.return_value = mock_response
+    mock_create_session.return_value = mock_session
+
+    # Test
+    result = blocklist.gc_ip("8.8.8.8")
+
+    # Verify
+    assert result is False
+
+
+@patch("blocklist.create_retrying_request")
+def test_gc_ip_http_error(mock_create_session):
+    """Test gc_ip with HTTP error response"""
+    # Setup mock response
+    mock_response = Mock()
+    mock_response.ok = False
+    mock_response.status_code = 404
+
+    # Setup mock session
+    mock_session = Mock()
+    mock_session.get.return_value = mock_response
+    mock_create_session.return_value = mock_session
+
+    # Test
+    result = blocklist.gc_ip("192.168.1.1")
+
+    # Verify
+    assert result is False
+
+
+@patch("blocklist.create_retrying_request")
+def test_gc_ip_request_exception(mock_create_session, capsys):
+    """Test gc_ip with request exception"""
+    # Setup mock session to raise exception
+    mock_session = Mock()
+    mock_session.get.side_effect = blocklist.requests.exceptions.RequestException(
+        "Connection error"
+    )
+    mock_create_session.return_value = mock_session
+
+    # Test
+    result = blocklist.gc_ip("192.168.1.1")
+
+    # Verify
+    assert result is False
+    captured = capsys.readouterr()
+    assert (
+        "Could not successfully retrieve information about IP 192.168.1.1"
+        in captured.out
+    )
+
+
+@patch("blocklist.create_retrying_request")
+def test_gc_ip_no_entities_in_response(mock_create_session):
+    """Test gc_ip when response has no entities"""
+    # Setup mock response
+    mock_response = Mock()
+    mock_response.ok = True
+    mock_response.json.return_value = {"other_field": "value"}
+
+    # Setup mock session
+    mock_session = Mock()
+    mock_session.get.return_value = mock_response
+    mock_create_session.return_value = mock_session
+
+    # Test
+    result = blocklist.gc_ip("192.168.1.1")
+
+    # Verify
+    assert result is False
+
+
+@patch("blocklist.create_retrying_request")
+def test_gc_ip_entities_is_none(mock_create_session):
+    """Test gc_ip when entities field is None"""
+    # Setup mock response
+    mock_response = Mock()
+    mock_response.ok = True
+    mock_response.json.return_value = {"entities": None}
+
+    # Setup mock session
+    mock_session = Mock()
+    mock_session.get.return_value = mock_response
+    mock_create_session.return_value = mock_session
+
+    # Test
+    result = blocklist.gc_ip("192.168.1.1")
+
+    # Verify
+    assert result is False

waf_ip_blocklist/lambda/requirements.txt

@@ -1,4 +1,5 @@
 black==24.10.0 
 boto3==1.40.11
 pylint==3.3.8
-pytest==8.4.1
+pytest==8.4.1
+requests==2.32.5