Add coverage tests and Python 3.12 SSL compatibility fixes

nettacker/core/lib/socket.py

@@ -26,7 +26,8 @@ def create_tcp_socket(host, port, timeout):
         return None
 
     try:
-        socket_connection = ssl.wrap_socket(socket_connection)
+        context = ssl.create_default_context()
+        socket_connection = context.wrap_socket(socket_connection, server_hostname=host)
         ssl_flag = True
     except Exception:
         socket_connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

nettacker/core/lib/ssl.py

@@ -117,7 +117,8 @@ def create_tcp_socket(host, port, timeout):
         return None
 
     try:
-        socket_connection = ssl.wrap_socket(socket_connection)
+        context = ssl.create_default_context()
+        socket_connection = context.wrap_socket(socket_connection, server_hostname=host)
         ssl_flag = True
     except Exception:
         socket_connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

report.html

@@ -0,0 +1 @@
+<table><graph_html>/*css*/</table>datetargetmodule_nameportlogsjson_event<tr>nowx</tr></table><div id="json_length">1</div><p class="footer">Software Details: OWASP Nettacker version 1.0 [beta] in now ScanID: scan-id</p><script>/*js*/</script>

tests/core/lib/test_base.py

@@ -0,0 +1,75 @@
+from unittest.mock import MagicMock, patch
+
+from nettacker.core.lib.base import BaseEngine
+
+
+def test_filter_large_content_truncates():
+    engine = BaseEngine()
+    content = "abcdefghij klm"
+    result = engine.filter_large_content(content, filter_rate=10)
+    assert result != content
+
+
+@patch("nettacker.core.lib.base.submit_logs_to_db")
+@patch("nettacker.core.lib.base.merge_logs_to_list", return_value=["logA"])
+@patch("nettacker.core.lib.base.remove_sensitive_header_keys")
+def test_process_conditions_success(mock_remove, mock_merge, mock_submit):
+    engine = BaseEngine()
+    event = {
+        "headers": {"Authorization": "secret"},
+        "response": {
+            "conditions_results": {"log": "entry"},
+            "conditions": {"dummy": {"reverse": False, "regex": ""}},
+            "condition_type": "and",
+        },
+        "ports": 80,
+    }
+    options = {"retries": 1}
+    mock_remove.return_value = event
+
+    result = engine.process_conditions(
+        event,
+        "module",
+        "target",
+        "scan",
+        options,
+        {"resp": True},
+        1,
+        1,
+        1,
+        1,
+        1,
+    )
+    assert result is True
+    mock_submit.assert_called_once()
+    mock_merge.assert_called_once()
+    mock_remove.assert_called_once()
+
+
+@patch("nettacker.core.lib.base.submit_temp_logs_to_db")
+def test_process_conditions_save_temp(mock_submit_temp):
+    engine = BaseEngine()
+    event = {
+        "response": {
+            "conditions_results": [],
+            "conditions": {},
+            "condition_type": "and",
+            "save_to_temp_events_only": "temp_evt",
+        }
+    }
+    options = {"retries": 1}
+    result = engine.process_conditions(
+        event,
+        "module",
+        "target",
+        "scan",
+        options,
+        {},
+        1,
+        1,
+        1,
+        1,
+        1,
+    )
+    assert result is True
+    mock_submit_temp.assert_called_once()

tests/core/lib/test_base_extended.py

@@ -0,0 +1,289 @@
+"""
+Comprehensive tests for TemplateLoader including parse, format, and load methods.
+"""
+
+import pytest
+from pathlib import Path
+from unittest.mock import MagicMock, patch, mock_open
+import yaml
+
+from nettacker.core.template import TemplateLoader
+
+
+class TestTemplateLoaderInit:
+    """Test TemplateLoader initialization."""
+
+    def test_initialization_with_name_only(self):
+        """Test initialization with just a name."""
+        loader = TemplateLoader("http_scan_scan")
+        assert loader.name == "http_scan_scan"
+        assert loader.inputs == {}
+
+    def test_initialization_with_name_and_inputs(self):
+        """Test initialization with name and inputs."""
+        inputs = {"port": "80", "timeout": "10"}
+        loader = TemplateLoader("http_scan_scan", inputs=inputs)
+        assert loader.name == "http_scan_scan"
+        assert loader.inputs == inputs
+
+    def test_initialization_with_none_inputs(self):
+        """Test that None inputs are converted to empty dict."""
+        loader = TemplateLoader("http_scan_scan", inputs=None)
+        assert loader.inputs == {}
+
+
+class TestTemplateLoaderParse:
+    """Test static parse method."""
+
+    def test_parse_dict_with_matching_input(self):
+        """Test parse replaces dict values with matching inputs."""
+        content = {"host": "{host}", "port": "80"}
+        inputs = {"host": "example.com"}
+        result = TemplateLoader.parse(content, inputs)
+        assert result["host"] == "example.com"
+        assert result["port"] == "80"
+
+    def test_parse_dict_with_nonmatching_input(self):
+        """Test parse ignores non-matching dict keys."""
+        content = {"host": "default.com", "port": "80"}
+        inputs = {"timeout": "10"}
+        result = TemplateLoader.parse(content, inputs)
+        assert result["host"] == "default.com"
+        assert result["port"] == "80"
+
+    def test_parse_nested_dict(self):
+        """Test parse handles nested dictionaries."""
+        content = {"level1": {"host": "{host}", "port": 80}}
+        inputs = {"host": "example.com"}
+        result = TemplateLoader.parse(content, inputs)
+        assert result["level1"]["host"] == "example.com"
+
+    def test_parse_list_elements(self):
+        """Test parse handles list elements."""
+        content = [{"host": "{host}"}, {"port": 80}]
+        inputs = {"host": "example.com"}
+        result = TemplateLoader.parse(content, inputs)
+        assert result[0]["host"] == "example.com"
+        assert result[1]["port"] == 80
+
+    def test_parse_nested_list_in_dict(self):
+        """Test parse handles lists within dicts."""
+        content = {"servers": [{"host": "primary.com"}, {"host": "backup.com"}]}
+        inputs = {}
+        result = TemplateLoader.parse(content, inputs)
+        assert result["servers"][0]["host"] == "primary.com"
+        assert result["servers"][1]["host"] == "backup.com"
+
+    def test_parse_with_truthy_input_value(self):
+        """Test parse uses input value when present and truthy."""
+        content = {"enabled": False}
+        inputs = {"enabled": True}
+        result = TemplateLoader.parse(content, inputs)
+        assert result["enabled"] is True
+
+    def test_parse_with_falsy_input_value(self):
+        """Test parse skips empty/falsy input values."""
+        content = {"enabled": True}
+        inputs = {"enabled": ""}
+        result = TemplateLoader.parse(content, inputs)
+        # Empty string is falsy, so original value is kept
+        assert result["enabled"] is True
+
+    def test_parse_preserves_types_in_nested_dict(self):
+        """Test parse handles nested structures with various types."""
+        content = {
+            "timeout": 30,
+            "nested": {
+                "delay": 0.5,
+                "data": b"binary"
+            }
+        }
+        inputs = {}
+        result = TemplateLoader.parse(content, inputs)
+        assert result["timeout"] == 30
+        assert result["nested"]["delay"] == 0.5
+        assert result["nested"]["data"] == b"binary"
+
+    def test_parse_empty_dict(self):
+        """Test parse with empty dict."""
+        content = {}
+        inputs = {"host": "example.com"}
+        result = TemplateLoader.parse(content, inputs)
+        assert result == {}
+
+    def test_parse_empty_list(self):
+        """Test parse with empty list."""
+        content = []
+        inputs = {"host": "example.com"}
+        result = TemplateLoader.parse(content, inputs)
+        assert result == []
+
+
+class TestTemplateLoaderOpen:
+    """Test open method for reading YAML files."""
+
+    def test_open_valid_template(self):
+        """Test opening a valid template file."""
+        # Mock the Config.path.modules_dir
+        mock_yaml_content = "target: '{target}'\nport: 80\n"
+
+        with patch("nettacker.core.template.Config.path.modules_dir") as mock_modules_dir:
+            # Create a mock Path object
+            mock_path = MagicMock()
+            mock_modules_dir.__truediv__ = MagicMock(return_value=mock_path)
+            mock_path.__truediv__ = MagicMock(return_value=MagicMock(__truediv__=MagicMock(return_value=MagicMock())))
+
+            with patch("builtins.open", mock_open(read_data=mock_yaml_content)):
+                loader = TemplateLoader("http_scan_scan")
+                result = loader.open()
+                assert isinstance(result, str)
+
+    def test_open_extracts_module_name_correctly(self):
+        """Test that open correctly parses module name."""
+        mock_yaml_content = "test: data\n"
+
+        with patch("nettacker.core.template.Config.path.modules_dir") as mock_modules_dir:
+            mock_dir = MagicMock()
+            mock_modules_dir.__truediv__ = MagicMock(return_value=mock_dir)
+
+            with patch("builtins.open", mock_open(read_data=mock_yaml_content)):
+                loader = TemplateLoader("port_scan_scan")
+                # This should split "port_scan_scan" into action="scan", library="port_scan"
+                loader.open()
+                # Verify the correct path construction was attempted
+                assert loader.name == "port_scan_scan"
+
+
+class TestTemplateLoaderFormat:
+    """Test format method."""
+
+    def test_format_with_inputs(self):
+        """Test format substitutes inputs into YAML string."""
+        mock_yaml = "target: '{target}'\nport: {port}\n"
+
+        with patch.object(TemplateLoader, "open", return_value=mock_yaml):
+            loader = TemplateLoader("http_scan_scan", inputs={"target": "example.com", "port": "80"})
+            result = loader.format()
+            assert "example.com" in result
+            assert "80" in result
+
+    def test_format_without_inputs(self):
+        """Test format on YAML without placeholders."""
+        mock_yaml = "target: localhost\nport: 80\n"
+
+        with patch.object(TemplateLoader, "open", return_value=mock_yaml):
+            loader = TemplateLoader("http_scan_scan")
+            result = loader.format()
+            assert result == mock_yaml
+
+    def test_format_with_all_inputs_provided(self):
+        """Test format with all required inputs provided."""
+        mock_yaml = "target: '{target}'\nport: '{port}'\n"
+
+        with patch.object(TemplateLoader, "open", return_value=mock_yaml):
+            loader = TemplateLoader("http_scan_scan", inputs={"target": "example.com", "port": "80"})
+            result = loader.format()
+            assert "example.com" in result
+            assert "80" in result
+
+
+class TestTemplateLoaderLoad:
+    """Test load method which combines format and parse."""
+
+    def test_load_yaml_with_inputs(self):
+        """Test load properly parses YAML and applies inputs."""
+        mock_yaml = "requests:\n  - host: '{host}'\n    port: 80\n"
+        formatted_yaml = "requests:\n  - host: 'example.com'\n    port: 80\n"
+
+        with patch.object(TemplateLoader, "open", return_value=mock_yaml):
+            loader = TemplateLoader("http_scan_scan", inputs={"host": "example.com"})
+            result = loader.load()
+
+            # Result should be a parsed YAML dict
+            assert isinstance(result, dict)
+            assert "requests" in result
+
+    def test_load_returns_parsed_dict(self):
+        """Test load returns parsed YAML as dict."""
+        mock_yaml = "key: value\nnumber: 42\n"
+
+        with patch.object(TemplateLoader, "open", return_value=mock_yaml):
+            loader = TemplateLoader("scan_test")
+            result = loader.load()
+
+            assert isinstance(result, dict)
+            assert result.get("key") == "value"
+            assert result.get("number") == 42
+
+    def test_load_with_nested_yaml(self):
+        """Test load with complex nested YAML structure."""
+        mock_yaml = """
+steps:
+  - name: scan
+    params:
+      host: '{target}'
+      port: 80
+"""
+
+        with patch.object(TemplateLoader, "open", return_value=mock_yaml):
+            loader = TemplateLoader("port_scan_scan", inputs={"target": "example.com"})
+            result = loader.load()
+
+            assert isinstance(result, dict)
+            assert "steps" in result
+            assert isinstance(result["steps"], list)
+
+    def test_load_with_list_yaml(self):
+        """Test load when YAML root is a list."""
+        mock_yaml = "- host: localhost\n  port: 80\n- host: example.com\n  port: 443\n"
+
+        with patch.object(TemplateLoader, "open", return_value=mock_yaml):
+            loader = TemplateLoader("scan_test")
+            result = loader.load()
+
+            assert isinstance(result, list)
+            assert len(result) == 2
+
+
+class TestTemplateLoaderIntegration:
+    """Integration tests combining multiple methods."""
+
+    def test_full_workflow(self):
+        """Test complete template loading workflow."""
+        mock_yaml = """
+module: scan
+target: '{host}'
+ports:
+  - 80
+  - 443
+"""
+
+        with patch.object(TemplateLoader, "open", return_value=mock_yaml):
+            loader = TemplateLoader("http_scan_scan", inputs={"host": "target.com"})
+
+            # Test each method in sequence
+            formatted = loader.format()
+            assert "target.com" in formatted
+
+            loaded = loader.load()
+            assert isinstance(loaded, dict)
+            assert loaded["module"] == "scan"
+            assert loaded["target"] == "target.com"
+            assert 80 in loaded["ports"]
+
+    def test_loader_with_multiple_templates(self):
+        """Test creating multiple loaders with different templates."""
+        mock_yaml1 = "type: scan\ntarget: '{host}'\n"
+        mock_yaml2 = "type: brute\nuser: '{user}'\n"
+
+        with patch.object(TemplateLoader, "open") as mock_open_method:
+            mock_open_method.side_effect = [mock_yaml1, mock_yaml2]
+
+            loader1 = TemplateLoader("port_scan_scan", {"host": "example.com"})
+            loader2 = TemplateLoader("ssh_brute_brute", {"user": "admin"})
+
+            result1 = loader1.load()
+            result2 = loader2.load()
+
+            assert result1["target"] == "example.com"
+            assert result2["user"] == "admin"