Scan project for malware and security issues

src/logparseiqx/cli.py

@@ -28,6 +28,7 @@
     filter_errors,
     filter_slow_requests,
     filter_security_events,
+    filter_by_status_class,
     aggregate_by_status,
     aggregate_by_country,
     aggregate_by_ip,
@@ -308,7 +309,7 @@ def cf_errors(ctx, logfile, tail, status):
     console.print("[orange3][CF] Scanning Cloudflare logs for errors...[/orange3]")
 
     if status:
-        filter_func = lambda x: str(x.get('EdgeResponseStatus', '')).startswith(status[:1]) or str(x.get('EdgeResponseStatus', '')) == status
+        filter_func = filter_by_status_class(status)
     else:
         filter_func = filter_errors
 

src/logparseiqx/parsers/cloudflare.py

@@ -153,6 +153,28 @@ def _filter(log: Dict[str, Any]) -> bool:
     return _filter
 
 
+def filter_by_status_class(status_code: str) -> Callable[[Dict], bool]:
+    """
+    Create a filter that matches an entire HTTP status class.
+
+    Uses the first digit to match all statuses in that class:
+    - '502' or '5' -> matches all 5xx (500, 502, 503, etc.)
+    - '404' or '4' -> matches all 4xx (400, 401, 404, etc.)
+
+    Args:
+        status_code: A status code like '502' or class like '5'
+
+    Returns:
+        Filter function that matches the status class
+    """
+    status_class = status_code[:1]  # First digit determines the class
+
+    def _filter(log: Dict[str, Any]) -> bool:
+        status = str(log.get('EdgeResponseStatus', ''))
+        return status.startswith(status_class) or status == status_code
+    return _filter
+
+
 def filter_slow_requests(threshold_ms: int) -> Callable[[Dict], bool]:
     """Create a filter for slow requests"""
     def _filter(log: Dict[str, Any]) -> bool:

tests/test_logparseiqx.py

@@ -35,6 +35,7 @@ def safe_unlink(filepath):
     filter_server_errors,
     filter_client_errors,
     filter_by_status,
+    filter_by_status_class,
     filter_slow_requests,
     filter_security_events,
     filter_by_country,
@@ -410,6 +411,21 @@ def test_filter_by_status(self):
         assert filter_5xx({'EdgeResponseStatus': 502}) is True
         assert filter_5xx({'EdgeResponseStatus': 404}) is False
 
+    def test_filter_by_status_class(self):
+        """Test that filter_by_status_class matches entire status class by first digit"""
+        # Passing '502' should match all 5xx (uses first digit)
+        filter_from_502 = filter_by_status_class("502")
+        assert filter_from_502({'EdgeResponseStatus': 502}) is True  # Exact match
+        assert filter_from_502({'EdgeResponseStatus': 500}) is True  # Same class (5xx)
+        assert filter_from_502({'EdgeResponseStatus': 503}) is True  # Same class (5xx)
+        assert filter_from_502({'EdgeResponseStatus': 404}) is False  # Different class
+
+        # Passing '4' should match all 4xx
+        filter_4xx = filter_by_status_class("4")
+        assert filter_4xx({'EdgeResponseStatus': 400}) is True
+        assert filter_4xx({'EdgeResponseStatus': 404}) is True
+        assert filter_4xx({'EdgeResponseStatus': 500}) is False
+
     def test_filter_slow_requests(self):
         filter_func = filter_slow_requests(1000)
         assert filter_func({'OriginResponseTime': 2000}) is True