diff --git a/codebeaver.yml b/codebeaver.yml new file mode 100644 index 00000000..ac19b7a3 --- /dev/null +++ b/codebeaver.yml @@ -0,0 +1,2 @@ +from: pytest +# This file was generated automatically by CodeBeaver based on your repository. Learn how to customize it here: https://docs.codebeaver.ai/open-source/codebeaver-yml/ \ No newline at end of file diff --git a/tests/test_log.py b/tests/test_log.py new file mode 100644 index 00000000..67b5fa93 --- /dev/null +++ b/tests/test_log.py @@ -0,0 +1,210 @@ +import logging +import sys +import json +import io +import pytest + +from core.log import setup_logger, console_log_level, file_log_level, log_file, log_config + +class TestLogger: + """Test suite for custom logging from xsstrike's core.log""" + + def setup_logger_instance(self, name="testlogger"): + """Helper method to create a logger instance using setup_logger""" + return setup_logger(name) + + def test_setup_logger_attributes(self): + """Test that the logger has the expected custom methods and handlers.""" + logger = self.setup_logger_instance() + # Check that at least one StreamHandler exists + assert any(isinstance(h, logging.StreamHandler) for h in logger.handlers) + # Check custom methods are attached to the logger + assert hasattr(logger, "red_line") + assert hasattr(logger, "yellow_summary_line") + assert hasattr(logger, "no_format") + assert hasattr(logger, "debug_json") + + def test_basic_logging(self, capsys): + """Test basic logging at INFO and ERROR levels and capture their output.""" + logger = self.setup_logger_instance() + logger.info("info message") + logger.error("error message") + captured = capsys.readouterr().out + assert "info message" in captured + assert "error message" in captured + + def test_custom_levels(self, capsys): + """Test custom logging levels VULN, RUN, and GOOD produce output.""" + logger = self.setup_logger_instance() + logger.vuln("vuln message") + logger.run("run message") + logger.good("good message") + captured = capsys.readouterr().out + # Check that messages corresponding to custom levels are present + assert "vuln message" in captured + assert "run message" in captured + assert "good message" in captured + + def test_no_format_logging(self, capsys): + """Test logging without formatting using the no_format method.""" + logger = self.setup_logger_instance() + logger.no_format("plain message", level='INFO') + captured = capsys.readouterr().out + # The output should contain the plain message without extra formatting prefixes + assert "plain message" in captured + # Check that typical formatting characters (such as '[') are absent in the output + assert ('[' not in captured) or (']' not in captured) + + def test_debug_json(self): + """Test the debug_json method using a StringIO stream to capture logger output.""" + logger = self.setup_logger_instance() + logger.setLevel(logging.DEBUG) + import io + buf = io.StringIO() + test_handler = logging.StreamHandler(buf) + test_handler.setLevel(logging.DEBUG) + test_handler.setFormatter(logging.Formatter("%(message)s")) + logger.addHandler(test_handler) + # Test debug_json with dictionary data + logger.debug_json("debug dict", {"a": 1}) + test_handler.flush() + output = buf.getvalue() + assert ('"a": 1' in output) or ("'a': 1" in output) + # Clear buffer for next test + buf.truncate(0) + buf.seek(0) + # Test debug_json with non-dictionary data + logger.debug_json("debug non-dict", "simple string") + test_handler.flush() + output = buf.getvalue() + assert "simple string" in output + # Clean up: remove test handler + logger.removeHandler(test_handler) + + def test_red_line_and_yellow_summary_line(self, capsys): + """Test the red_line and yellow_summary_line methods produce expected patterns.""" + logger = self.setup_logger_instance() + # Test red_line: should log a line with a repeated '-' pattern of specified length + logger.red_line(amount=10, level='ERROR') + output_red = capsys.readouterr().out + assert '-' * 10 in output_red + + # Test yellow_summary_line: should log a line with a repeated '=' pattern of specified length + logger.yellow_summary_line(amount=15, level='INFO') + output_yellow = capsys.readouterr().out + assert '=' * 15 in output_yellow + + def test_file_logging(self, tmp_path, monkeypatch): + """Test that file logging is set up correctly when file_log_level is enabled.""" + # Temporarily enable file logging by setting file_log_level to DEBUG + from core import log as core_log + monkeypatch.setattr(core_log, 'file_log_level', 'DEBUG') + test_log_file = tmp_path / "test_xsstrike.log" + monkeypatch.setattr(core_log, 'log_file', str(test_log_file)) + logger = setup_logger("filelogger") + # Log a debug message that should be written to the file + logger.debug("debug file message") + # Ensure that the file_handler attribute is present (indicating file logging is active) + assert hasattr(logger, "file_handler") + # Flush handlers so that content is written to the file + for handler in logger.handlers: + if hasattr(handler, "flush"): + handler.flush() + # Read the file and check that it contains the debug message + log_contents = test_log_file.read_text() + assert "debug file message" in log_contents + def test_debug_json_non_serializable(self): + """Test the debug_json method with non-JSON serializable data and ensure fallback logging.""" + logger = self.setup_logger_instance() + logger.setLevel(logging.DEBUG) + import io + buf = io.StringIO() + test_handler = logging.StreamHandler(buf) + test_handler.setLevel(logging.DEBUG) + test_handler.setFormatter(logging.Formatter("%(message)s")) + logger.addHandler(test_handler) + # Provide non-serializable data (a set) to force a TypeError in json.dumps + logger.debug_json("test nonserializable", {"a": set([1])}) + test_handler.flush() + output = buf.getvalue() + # Check that the fallback logging logged the message and a string representation of the data + assert "test nonserializable" in output + assert "set(" in output or "{" in output + logger.removeHandler(test_handler) + + def test_non_recognized_level_in_no_format(self, capsys): + """Test that no_format logs using INFO level when an unrecognized level is provided.""" + logger = self.setup_logger_instance() + # Calling no_format with an undefined logging level should fallback to info level logging + logger.no_format("fallback message", level="NONEXISTENT") + captured = capsys.readouterr().out + assert "fallback message" in captured + + def test_custom_handler_terminator(self): + """Test that CustomStreamHandler uses a custom terminator for messages ending with '\\r'.""" + from core.log import CustomStreamHandler + import io + stream = io.StringIO() + handler = CustomStreamHandler(stream) + handler.setFormatter(logging.Formatter("%(message)s")) + # Create a LogRecord with a message ending in a carriage return + record = logging.LogRecord("test", logging.INFO, "", 0, "line ending with carriage return\r", None, None) + handler.emit(record) + output = stream.getvalue() + # Verify that the output ends with '\r' + assert output.endswith("\r") + def test_custom_formatter_prefix(self): + """Test that CustomFormatter adds the correct prefix for levels present in log_config.""" + from core.log import CustomFormatter + from logging import LogRecord + # Temporarily override the INFO prefix to a known value for testing + original_prefix = log_config['INFO']['prefix'] + log_config['INFO']['prefix'] = "INFO_PREFIX" + record = LogRecord("test", logging.INFO, "", 0, "dummy", None, None) + formatter = CustomFormatter("%(message)s") + formatted_message = formatter.format(record) + # Restore original prefix + log_config['INFO']['prefix'] = original_prefix + assert formatted_message.startswith("INFO_PREFIX") + assert "dummy" in formatted_message + + def test_setup_logger_idempotency(self): + """Test that calling setup_logger multiple times with the same name does not duplicate handlers or remove custom methods.""" + # Create a logger instance using our helper. + logger1 = self.setup_logger_instance("dup") + initial_handlers = logger1.handlers.copy() + + # Call setup_logger again for the same logger name. + logger2 = setup_logger("dup") + # They should be the same instance. + assert logger1 is logger2 + # Custom methods should remain attached. + assert hasattr(logger2, "red_line") + assert hasattr(logger2, "yellow_summary_line") + # All initially attached handlers should still be present. + for handler in initial_handlers: + assert handler in logger2.handlers + + # Also check that repeated configuration does not cause duplicate logging output. + import io + buf = io.StringIO() + temp_handler = logging.StreamHandler(buf) + temp_handler.setLevel(logging.DEBUG) + temp_handler.setFormatter(logging.Formatter("%(message)s")) + logger1.addHandler(temp_handler) + logger1.info("dup test") + temp_handler.flush() + output = buf.getvalue() + # We expect the message to appear only once. + assert output.count("dup test") == 1 + logger1.removeHandler(temp_handler) + + def test_handler_switching(self, capsys): + """Test that the logger correctly switches between no format and default format handlers.""" + logger = self.setup_logger_instance() + # Save a reference to the default console handler. + default_handler = logger.console_handler + # Call no_format to log a message using a blank formatter. + logger.no_format("switch test", level="INFO") + # After the no_format method has been called the default handler should be re-attached. + assert default_handler in logger.handlers \ No newline at end of file diff --git a/tests/test_requester.py b/tests/test_requester.py new file mode 100644 index 00000000..da28d9e9 --- /dev/null +++ b/tests/test_requester.py @@ -0,0 +1,167 @@ +import time +import requests +import random +import pytest + +from urllib3.exceptions import ProtocolError +from core.requester import requester + +# Import the module so we can patch its attributes (getVar, converter, logger) +import core.requester as requester_mod + +class DummyResponse: + def __init__(self, text="dummy response", status_code=200): + self.text = text + self.status_code = status_code + +def dummy_converter(data, url=None): + """A dummy converter that appends '_converted' to its input.""" + if url: + return url + "_converted" + return str(data) + "_converted" + +@pytest.fixture(autouse=True) +def patch_getVar(monkeypatch): + """Patch getVar to return False by default for any key.""" + monkeypatch.setattr(requester_mod, 'getVar', lambda key: False) + +@pytest.fixture(autouse=True) +def patch_converter(monkeypatch): + """Patch converter to use our dummy_converter.""" + monkeypatch.setattr(requester_mod, 'converter', dummy_converter) + +class DummyLogger: + def debug(self, msg): pass + def debug_json(self, msg, obj): pass + def warning(self, msg): pass + +@pytest.fixture(autouse=True) +def patch_logger(monkeypatch): + """Patch logger with a dummy logger to avoid real logging during tests.""" + monkeypatch.setattr(requester_mod, 'logger', DummyLogger()) + +@pytest.fixture(autouse=True) +def patch_sleep(monkeypatch): + """Patch time.sleep to avoid delay during tests.""" + monkeypatch.setattr(time, 'sleep', lambda x: None) + +def test_requester_get(monkeypatch): + """Test that a GET request is made correctly when GET flag is True.""" + test_headers = {} # no User-Agent given + test_data = {'param': 'value'} + test_url = "http://example.com" + + def dummy_get(url, params, headers, timeout, verify, proxies): + assert url == test_url + assert params == test_data + # Ensure headers gets a valid User-Agent (injected randomly) + assert 'User-Agent' in headers and headers['User-Agent'] != '' + return DummyResponse(text="GET success") + + monkeypatch.setattr(requests, 'get', dummy_get) + + response = requester(test_url, test_data, test_headers, GET=True, delay=0, timeout=5) + assert response.text == "GET success" + +def test_requester_post_json(monkeypatch): + """Test that a POST request with json data is executed when getVar('jsonData') returns True.""" + # Patch getVar: return True for 'jsonData', False otherwise. + custom_getvar = lambda key: True if key == 'jsonData' else False + monkeypatch.setattr(requester_mod, 'getVar', custom_getvar) + + test_headers = {'User-Agent': '$'} # This should be replaced with a random user-agent. + test_data = {'key': 'value'} + test_url = "http://example.com" + + def dummy_post(url, json, headers, timeout, verify, proxies): + assert url == test_url + # dummy_converter converts dict to string with '_converted' appended + expected = str(test_data) + "_converted" + assert json == expected + # Ensure user-agent is replaced and is not '$' + assert headers['User-Agent'] != '$' + return DummyResponse(text="POST json success") + + monkeypatch.setattr(requests, 'post', dummy_post) + + response = requester(test_url, test_data, test_headers, GET=False, delay=0, timeout=5) + assert response.text == "POST json success" + +def test_requester_post_regular(monkeypatch): + """Test that a regular POST request (with form data) executes correctly.""" + test_headers = {'User-Agent': 'TestAgent'} + test_data = {'key': 'value'} + test_url = "http://example.com" + + def dummy_post(url, data, headers, timeout, verify, proxies): + assert url == test_url + # When not converting, the data stays the same. + assert data == test_data + # Ensure the given User-Agent remains unchanged. + assert headers['User-Agent'] == 'TestAgent' + return DummyResponse(text="POST regular success") + + monkeypatch.setattr(requests, 'post', dummy_post) + + response = requester(test_url, test_data, test_headers, GET=False, delay=0, timeout=5) + assert response.text == "POST regular success" + +def test_requester_path(monkeypatch): + """Test that when getVar('path') returns True, the URL is converted and data is cleared.""" + # Patch getVar: return True for 'path', False otherwise. + custom_getvar = lambda key: True if key == 'path' else False + monkeypatch.setattr(requester_mod, 'getVar', custom_getvar) + + test_headers = {} + test_data = "data" # will be passed to converter + test_url = "http://example.com" + + def dummy_get(url, params, headers, timeout, verify, proxies): + # After conversion, the url should have been appended with '_converted' + assert url == test_url + "_converted" + # data should become [] per the implementation + assert params == [] + return DummyResponse(text="GET path success") + + monkeypatch.setattr(requests, 'get', dummy_get) + + response = requester(test_url, test_data, test_headers, GET=False, delay=0, timeout=5) + assert response.text == "GET path success" + +def test_protocol_exception(monkeypatch): + """Test that a ProtocolError is handled by logging a warning and sleeping for 10 minutes.""" + test_headers = {} + test_data = {'param': 'value'} + test_url = "http://example.com" + + def raising_get(*args, **kwargs): + raise ProtocolError("protocol error") + + # Use a mutable container to flag that sleep was called. + sleep_called = [False] + def dummy_sleep(duration): + sleep_called[0] = True + + monkeypatch.setattr(requests, 'get', raising_get) + monkeypatch.setattr(time, 'sleep', dummy_sleep) + + response = requester(test_url, test_data, test_headers, GET=True, delay=0, timeout=5) + # The ProtocolError except block should have been executed. + assert sleep_called[0] is True + # Function does not return a response in ProtocolError case. + assert response is None + +def test_general_exception(monkeypatch): + """Test that a general Exception is handled by returning an empty requests.Response object.""" + test_headers = {} + test_data = {'param': 'value'} + test_url = "http://example.com" + + def raising_get(*args, **kwargs): + raise Exception("general error") + + monkeypatch.setattr(requests, 'get', raising_get) + + response = requester(test_url, test_data, test_headers, GET=True, delay=0, timeout=5) + # When a generic Exception is raised, an instance of requests.Response is returned. + assert isinstance(response, requests.Response) \ No newline at end of file diff --git a/tests/test_scan.py b/tests/test_scan.py new file mode 100644 index 00000000..9212da1e --- /dev/null +++ b/tests/test_scan.py @@ -0,0 +1,320 @@ +import pytest +import core.config +from concurrent.futures import Future +from modes.scan import scan, checky + +# Dummy response class for simulating HTTP responses +class DummyResponse: + def __init__(self, text): + self.text = text + +# Define dummy functions to override dependencies in scan +def dummy_requester(url, params, headers, GET, delay, timeout): + # Return a dummy response object + return DummyResponse("dummy response") + +def dummy_dom(response_text): + # Simulate that there is no DOM vulnerability + return False + +def dummy_getUrl(target, GET): + return target + "/dummy" + +def dummy_getParams(target, paramData, GET): + # If paramData is provided, return a non-empty dict, else empty. + if paramData: + return {"a": "test"} + return {} + +def dummy_wafDetector(url, params, headers, GET, delay, timeout): + return None + +def dummy_htmlParser(response, encoding): + # If response text contains 'xsschecker', simulate finding reflections. + if "xsschecker" in response.text: + return {0: "found"} + return {} + +def dummy_filterChecker(url, params, headers, GET, delay, occurences, timeout, encoding): + # Dummy efficiencies for simulation. + return [50] + +def dummy_generator(occurences, response_text): + # Return a payload that simulates a high-efficiency test. + return {100: ["payload1"]} + +def dummy_checker(url, params, headers, GET, delay, vect, positions, timeout, encoding): + # If vect equals "payload1", simulate 100 efficiency; otherwise 0. + if vect == "payload1": + return [100] + return [0] + +class TestScan: + """Test suite for scan module.""" + + @pytest.fixture(autouse=True) + def patch_dependencies(self, monkeypatch): + # Patch all external dependencies in modes.scan to use our dummy implementations. + from modes import scan as s + monkeypatch.setattr(s, "requester", dummy_requester) + monkeypatch.setattr(s, "dom", dummy_dom) + monkeypatch.setattr(s, "getUrl", dummy_getUrl) + monkeypatch.setattr(s, "getParams", dummy_getParams) + monkeypatch.setattr(s, "wafDetector", dummy_wafDetector) + monkeypatch.setattr(s, "htmlParser", dummy_htmlParser) + monkeypatch.setattr(s, "filterChecker", dummy_filterChecker) + monkeypatch.setattr(s, "generator", dummy_generator) + monkeypatch.setattr(s, "checker", dummy_checker) + monkeypatch.setitem(core.config.globalVariables, 'path', False) + + def test_scan_no_params(self, monkeypatch): + """Test that scan exits when no parameters are provided.""" + # Override getParams to always return an empty dict. + monkeypatch.setattr("modes.scan.getParams", lambda target, paramData, GET: {}) + with pytest.raises(SystemExit): + scan("http://example.com", None, None, {}, 0, 10, True, True, 2) + + def test_scan_no_reflections(self, monkeypatch): + """Test that scan handles no reflection found in the response.""" + # Override htmlParser to always return an empty dict. + monkeypatch.setattr("modes.scan.htmlParser", lambda response, encoding: {}) + # Provide non-empty paramData so getParams returns a parameter. + result = scan("http://example.com", {"dummy": "data"}, None, {}, 0, 10, True, True, 2) + # Since no reflection is found, scan continues and returns None. + assert result is None + + def test_checky_efficiency(self): + """Test checky's calculation of best efficiency using the dummy checker.""" + paramsCopy = {"a": "test"} + positions = [0] + occurences = {0: "found"} + # Call checky with dummy vector "payload1" that must lead to 100 efficiency. + bestEfficiency, target, loggerVector, confidence = checky( + "http://example.com", + "http://example.com/dummy", + paramsCopy, + {}, + True, + 0, + "payload1", + positions, + 10, + None, + occurences, + 100, + {"lap": 0}, + 1) + assert bestEfficiency == 100 + assert target == "http://example.com" + assert loggerVector == "payload1" + assert confidence == 100 + + def test_scan_payload_found(self, monkeypatch): + """Test scan when a high-efficiency payload is found, simulating user input 'n' to stop further scanning.""" + # Simulate user input of 'n' so that scanning stops after a payload is found. + monkeypatch.setattr("builtins.input", lambda prompt: "n") + # Override htmlParser to simulate a reflection by checking for 'xsschecker'. + monkeypatch.setattr("modes.scan.htmlParser", lambda response, encoding: {0: "xsschecker"}) + # Override requester to return a response that contains 'xsschecker'. + def custom_requester(url, params, headers, GET, delay, timeout): + return DummyResponse("contains xsschecker") + monkeypatch.setattr("modes.scan.requester", custom_requester) + result = scan("http://example.com", {"dummy": "data"}, None, {}, 0, 10, True, False, 2) + # Check that scan returns a tuple with the target and the payload. + assert isinstance(result, tuple) + assert result[0] == "http://example.com" + assert result[1] == "payload1" + + def test_scan_with_https_prefix(self, monkeypatch): + monkeypatch.setattr("builtins.input", lambda prompt: "n") + """Test scan handling a target without an http(s) prefix by forcing an https attempt that fails then falls back to http.""" + call_count = {"count": 0} + def custom_requester(url, params, headers, GET, delay, timeout): + if call_count["count"] == 0: + call_count["count"] += 1 + raise Exception("HTTPS failed") + return DummyResponse("dummy response xsschecker") + monkeypatch.setattr("modes.scan.requester", custom_requester) + result = scan("example.com", {"dummy": "data"}, None, {}, 0, 10, True, False, 2) + # Since HTTPS fails, scan should fall back to http. Either a payload is found or not, + # so we check that the result is either a tuple or None. + if result is not None: + assert isinstance(result, tuple) + else: + assert result is None + def test_scan_continue_scanning(self, monkeypatch): + """Test scan continues scanning when user inputs 'y' after a high-efficiency payload is found.""" + monkeypatch.setattr("builtins.input", lambda prompt: "y") + monkeypatch.setattr("modes.scan.htmlParser", lambda response, encoding: {0: "xsschecker"}) + monkeypatch.setattr("modes.scan.requester", lambda url, params, headers, GET, delay, timeout: DummyResponse("contains xsschecker")) + # With user response "y", the scan does not break early and returns None. + result = scan("http://example.com", {"dummy": "data"}, None, {}, 0, 10, True, False, 2) + assert result is None + + def test_scan_no_vectors_generated(self, monkeypatch): + """Test scan handling when the payload generator produces no vectors.""" + monkeypatch.setattr("modes.scan.generator", lambda occurences, response_text: {}) + monkeypatch.setattr("modes.scan.htmlParser", lambda response, encoding: {0: "xsschecker"}) + result = scan("http://example.com", {"dummy": "data"}, None, {}, 0, 10, True, True, 2) + # With no payloads generated, scan should complete without an early return. + assert result is None + + def test_scan_with_encoding(self, monkeypatch): + """Test scan behavior when an encoding function is provided.""" + monkeypatch.setattr("builtins.input", lambda prompt: "n") + monkeypatch.setattr("modes.scan.htmlParser", lambda response, encoding: {0: "xsschecker"}) + encoding_func = lambda s: "encoded_" + s + monkeypatch.setattr("modes.scan.requester", lambda url, params, headers, GET, delay, timeout: DummyResponse("contains xsschecker")) + result = scan("http://example.com", {"dummy": "data"}, encoding_func, {}, 0, 10, True, False, 2) + # Expect a tuple result with the target and the payload since a high-efficiency payload is found. + assert isinstance(result, tuple) + assert result[0] == "http://example.com" + assert result[1] == "payload1" + def test_scan_vector_with_slash(self, monkeypatch): + """Test scan branch when vector starts with '\' and efficiency >=95 triggers early exit.""" + monkeypatch.setattr("builtins.input", lambda prompt: "n") + monkeypatch.setattr("modes.scan.htmlParser", lambda response, encoding: {0: "xsschecker"}) + # Override generator to produce a vector starting with '\' and dummy checker to return an efficiency of 95. + monkeypatch.setattr("modes.scan.generator", lambda occurences, response_text: {90: ["\\payload_special"]}) + monkeypatch.setattr("modes.scan.checker", lambda url, params, headers, GET, delay, vect, positions, timeout, encoding: [95]) + result = scan("http://example.com", {"dummy": "data"}, None, {}, 0, 10, True, False, 2) + # Assert that a tuple is returned with the expected payload and target. + assert isinstance(result, tuple) + assert result[0] == "http://example.com" + assert result[1] == "\\payload_special" + + def test_checky_with_post(self): + """Test checky function behavior when POST method is used (GET is False) so that unquote is applied.""" + paramsCopy = {"a": "test"} + positions = [0] + occurences = {0: "found"} + # Provide a percent-encoded payload that, when unquoted, becomes 'payload1' + vect = "%70ayload1" + # Define a dummy checker that returns 100 efficiency when vect is 'payload1' + def dummy_post_checker(url, params, headers, GET, delay, vect, positions, timeout, encoding): + if vect == "payload1": + return [100] + return [0] + from modes import scan as s + s.checker = dummy_post_checker + bestEfficiency, target, loggerVector, confidence = checky( + "http://example.com", + "http://example.com/dummy", + paramsCopy, + {}, + False, + 0, + vect, + positions, + 10, + None, + occurences, + 100, + {"lap": 0}, + 1) + assert bestEfficiency == 100 + assert target == "http://example.com" + assert loggerVector == "%70ayload1" + assert confidence == 100 + + def test_scan_multiple_params(self, monkeypatch): + """Test scan behavior when multiple parameters are provided, ensuring all are processed.""" + # Override getParams to return multiple parameters. + monkeypatch.setattr("modes.scan.getParams", lambda target, paramData, GET: {"a": "test", "b": "test2"}) + # Override htmlParser to simulate reflections. + monkeypatch.setattr("modes.scan.htmlParser", lambda response, encoding: {0: "xsschecker"}) + # Override generator to produce payloads for both parameters with efficiencies below early exit threshold. + monkeypatch.setattr("modes.scan.generator", lambda occurences, response_text: {50: ["payloadA"], 60: ["payloadB"]}) + # Override checker to always return an efficiency of 60. + monkeypatch.setattr("modes.scan.checker", lambda url, params, headers, GET, delay, vect, positions, timeout, encoding: [60]) + # Override filterChecker to return dummy efficiencies. + monkeypatch.setattr("modes.scan.filterChecker", lambda url, params, headers, GET, delay, occurences, timeout, encoding: [60]) + # Simulate user input 'n' (even though no early break should be triggered). + monkeypatch.setattr("builtins.input", lambda prompt: "n") + result = scan("http://example.com", {"dummy": "data"}, None, {}, 0, 10, True, True, 2) + # Expect the scan to complete processing all parameters and return None. + assert result is None + def test_checky_empty_efficiencies(self, monkeypatch): + """Test checky behavior when checker returns an empty list. In this case, the function will add zero efficiencies for each occurrence and the best efficiency should be 0.""" + paramsCopy = {"a": "test"} + positions = [0] + occurences = {0: "found"} + vect = "empty_test_payload" + progress = {"lap": 0} + # Override checker to return an empty list regardless of input. + monkeypatch.setattr("modes.scan.checker", lambda url, params, headers, GET, delay, vect, positions, timeout, encoding: []) + bestEfficiency, target, loggerVector, confidence = checky( + "http://example.com", + "http://example.com/dummy", + paramsCopy, + {}, + True, + 0, + vect, + positions, + 10, + None, + occurences, + 50, + progress, + 1) + assert bestEfficiency == 0 + assert target == "http://example.com" + assert loggerVector == "empty_test_payload" + assert confidence == 50 + assert progress["lap"] == 1 + + def test_checky_with_global_path_true(self, monkeypatch): + """Test checky when core.config.globalVariables['path'] is True. The payload vector should have "/" replaced with "%2F" before being used.""" + from core import config + # Set the global path flag to True so that vect replacement takes place. + config.globalVariables['path'] = True + paramsCopy = {"a": "test"} + positions = [0] + occurences = {0: "found"} + vect = "a/b" + progress = {"lap": 0} + # Override checker to return 100 efficiency. + monkeypatch.setattr("modes.scan.checker", lambda url, params, headers, GET, delay, vect, positions, timeout, encoding: [100]) + bestEfficiency, target, loggerVector, confidence = checky( + "http://example.com", + "http://example.com/dummy", + paramsCopy, + {}, + True, + 0, + vect, + positions, + 10, + None, + occurences, + 80, + progress, + 1) + # Since globalVariables['path'] is True, the "/" in vect should have been replaced with "%2F". + assert loggerVector == "a%2Fb" + assert bestEfficiency == 100 + assert target == "http://example.com" + assert confidence == 80 + assert progress["lap"] == 1 + # Reset the flag to avoid side effects on other tests. + config.globalVariables['path'] = False + def test_scan_dom_vulnerabilities(self, monkeypatch): + """Test scan when DOM vulnerabilities are found and payload is triggered.""" + monkeypatch.setattr("builtins.input", lambda prompt: "n") + monkeypatch.setattr("modes.scan.dom", lambda response: ["Vuln found line 1", "Vuln found line 2"]) + monkeypatch.setattr("modes.scan.htmlParser", lambda response, encoding: {0: "xsschecker"}) + payloads = {100: ["payload_dom"]} + monkeypatch.setattr("modes.scan.generator", lambda occurences, response_text: payloads) + monkeypatch.setattr("modes.scan.filterChecker", lambda url, params, headers, GET, delay, occurences, timeout, encoding: [100]) + monkeypatch.setattr("modes.scan.checker", lambda url, params, headers, GET, delay, vect, positions, timeout, encoding: [100]) + result = scan("http://example.com", {"dummy": "data"}, None, {}, 0, 10, False, False, 2) + assert isinstance(result, tuple) + assert result[1] == "payload_dom" + + def test_scan_waf_detected(self, monkeypatch): + """Test scan behavior when WAF is detected, ensuring proper logging but continuing scan.""" + monkeypatch.setattr("modes.scan.wafDetector", lambda url, params, headers, GET, delay, timeout: "WAF_Flag") + monkeypatch.setattr("modes.scan.htmlParser", lambda response, encoding: {}) + result = scan("http://example.com", {"dummy": "data"}, None, {}, 0, 10, True, True, 2) + assert result is None \ No newline at end of file diff --git a/tests/test_wafDetector.py b/tests/test_wafDetector.py new file mode 100644 index 00000000..01a6f606 --- /dev/null +++ b/tests/test_wafDetector.py @@ -0,0 +1,222 @@ +import io +import json +import pytest +import builtins # Import builtins to patch the built-in open function + +from core.wafDetector import wafDetector + +# FakeResponse class to simulate the response from the requester function +class FakeResponse: + def __init__(self, text, status_code, headers): + self.text = text + self.status_code = status_code + self.headers = headers + +# A fake open() function to simulate reading the wafSignatures.json file. +def fake_open(*args, **kwargs): + json_content = json.dumps({ + "FakeWAF": { + "page": "trigger", + "code": "^403$", + "headers": "header_trigger" + }, + "NoMatchWAF": { + "page": "nomatch", + "code": "^599$", + "headers": "notfound" + } + }) + return io.StringIO(json_content) + +# A fake requester() that returns a FakeResponse. +def fake_requester(url, params, headers, GET, delay, timeout): + # Use attributes attached to the function to simulate different responses. + return FakeResponse(fake_requester.text, fake_requester.status_code, fake_requester.headers) + + +def test_code_below_400(monkeypatch): + """Test wafDetector returns None when HTTP status code is below 400.""" + # Monkey-patch the open and requester in the wafDetector module. + monkeypatch.setattr(builtins, "open", fake_open) + monkeypatch.setattr("core.wafDetector.requester", fake_requester) + + # Set the fake response attributes to simulate a safe page. + fake_requester.text = "Safe page content" + fake_requester.status_code = 200 + fake_requester.headers = {"Server": "Apache"} + + params = {} + result = wafDetector("http://example.com", params, {"User-Agent": "test"}, True, 0, 5) + # Ensure the noise payload was added. + assert params.get("xss") == '' + assert result is None + + +def test_no_match(monkeypatch): + """Test wafDetector returns None when the fingerprints do not match the response data.""" + monkeypatch.setattr(builtins, "open", fake_open) + monkeypatch.setattr("core.wafDetector.requester", fake_requester) + + fake_requester.text = "Page with safe content" + fake_requester.status_code = 500 + fake_requester.headers = {"Server": "nginx"} + + params = {} + result = wafDetector("http://example.com", params, {"User-Agent": "test"}, True, 0, 5) + assert result is None + + +def test_match(monkeypatch): + """Test wafDetector returns the matching WAF name when the response is consistent with a known fingerprint.""" + monkeypatch.setattr(builtins, "open", fake_open) + monkeypatch.setattr("core.wafDetector.requester", fake_requester) + + fake_requester.text = "This page includes trigger message" + fake_requester.status_code = 403 + fake_requester.headers = {"X-Powered-By": "header_trigger"} + + params = {} + result = wafDetector("http://example.com", params, {"User-Agent": "test"}, True, 0, 5) + assert result == "FakeWAF" + + +def test_no_code(monkeypatch): + """Test that wafDetector returns None when the response has no valid status code.""" + monkeypatch.setattr(builtins, "open", fake_open) + monkeypatch.setattr("core.wafDetector.requester", fake_requester) + + fake_requester.text = "Some content with trigger" + fake_requester.status_code = None + fake_requester.headers = {"X": "header_trigger"} + + params = {} + result = wafDetector("http://example.com", params, {"User-Agent": "test"}, True, 0, 5) + assert result is None +def fake_open_custom(*args, **kwargs): + json_content = json.dumps({ + "WAF1": {"page": "alpha", "code": "^500$", "headers": "hdr1"}, + "WAF2": {"page": "alpha", "code": "^500$", "headers": "hdr2"} + }) + return io.StringIO(json_content) + +def fake_open_empty(*args, **kwargs): + json_content = "{}" + return io.StringIO(json_content) + +def fake_open_empty_signatures(*args, **kwargs): + json_content = json.dumps({ + "EmptyWAF": {"page": "", "code": "", "headers": ""} + }) + return io.StringIO(json_content) + +def test_multiple_matches(monkeypatch): + """Test wafDetector returns the WAF with the highest score when multiple signatures match.""" + monkeypatch.setattr(builtins, "open", fake_open_custom) + monkeypatch.setattr("core.wafDetector.requester", fake_requester) + # Response that matches both signatures on page and code. + # However, only WAF1 will have its header pattern matched. + fake_requester.text = "This alpha content string" + fake_requester.status_code = 500 + fake_requester.headers = {"Custom": "hdr1"} + params = {} + result = wafDetector("http://example.com", params, {"User-Agent": "test"}, True, 0, 5) + assert result == "WAF1" + +def test_empty_waf_signatures(monkeypatch): + """Test that wafDetector returns None when the wafSignatures JSON file is empty.""" + monkeypatch.setattr(builtins, "open", fake_open_empty) + monkeypatch.setattr("core.wafDetector.requester", fake_requester) + fake_requester.text = "Content with trigger" + fake_requester.status_code = 403 + fake_requester.headers = {"X": "header_trigger"} + params = {} + result = wafDetector("http://example.com", params, {"User-Agent": "test"}, True, 0, 5) + assert result is None + +def test_empty_signatures_fields(monkeypatch): + """Test that wafDetector returns None when wafSignatures have empty fields.""" + monkeypatch.setattr(builtins, "open", fake_open_empty_signatures) + monkeypatch.setattr("core.wafDetector.requester", fake_requester) + fake_requester.text = "Any content that might match trigger" + fake_requester.status_code = 404 + fake_requester.headers = {"Server": "Apache"} + params = {} + result = wafDetector("http://example.com", params, {"User-Agent": "test"}, True, 0, 5) + assert result is None + +def fake_open_header_only(*args, **kwargs): + json_content = json.dumps({ + "HeaderOnly": {"page": "", "code": "", "headers": "header_only"} + }) + return io.StringIO(json_content) + +def test_header_only_match(monkeypatch): + """Test that wafDetector detects a WAF based solely on header matching.""" + monkeypatch.setattr(builtins, "open", fake_open_header_only) + monkeypatch.setattr("core.wafDetector.requester", fake_requester) + fake_requester.text = "Non triggering text" + fake_requester.status_code = 404 + fake_requester.headers = {"X": "header_only_value"} + params = {} + result = wafDetector("http://example.com", params, {"User-Agent": "test"}, True, 0, 5) + assert result == "HeaderOnly" +def test_code_zero(monkeypatch): + """Test that wafDetector returns None when HTTP status code is 0 (falsy).""" + monkeypatch.setattr(builtins, "open", fake_open) + monkeypatch.setattr("core.wafDetector.requester", fake_requester) + fake_requester.text = "Content with no triggering" + fake_requester.status_code = 0 + fake_requester.headers = {"Server": "Apache"} + params = {} + result = wafDetector("http://example.com", params, {"User-Agent": "test"}, True, 0, 5) + # xss payload should be injected regardless + assert params.get("xss") == '' + assert result is None + +def fake_open_equal(*args, **kwargs): + json_content = json.dumps({ + "WAF1": {"page": "trigger", "code": "", "headers": ""}, + "WAF2": {"page": "trigger", "code": "", "headers": ""} + }) + return io.StringIO(json_content) + +def test_equal_score(monkeypatch): + """Test that wafDetector returns the first matching WAF when two have equal scores.""" + monkeypatch.setattr(builtins, "open", fake_open_equal) + monkeypatch.setattr("core.wafDetector.requester", fake_requester) + # Both signatures will match the page text, giving them an equal score. + fake_requester.text = "This text contains trigger somewhere." + fake_requester.status_code = 403 + fake_requester.headers = {"Test": "none"} + params = {} + result = wafDetector("http://example.com", params, {"User-Agent": "test"}, True, 0, 5) + # The first WAF in the JSON ("WAF1") is expected to win in the case of a tie. + assert result == "WAF1" +def fake_open_code_only(*args, **kwargs): + json_content = json.dumps({ + "CodeOnly": {"page": "", "code": "^403$", "headers": ""} + }) + return io.StringIO(json_content) + +def test_code_only_match(monkeypatch): + """Test that wafDetector detects a WAF based solely on code signature matching.""" + monkeypatch.setattr(builtins, "open", fake_open_code_only) + monkeypatch.setattr("core.wafDetector.requester", fake_requester) + fake_requester.text = "Normal content that does not trigger page or header match" + fake_requester.status_code = 403 + fake_requester.headers = {"X": "none"} + params = {} + result = wafDetector("http://example.com", params, {"User-Agent": "test"}, True, 0, 5) + assert result == "CodeOnly" + +def test_requester_exception(monkeypatch): + """Test that wafDetector propagates an exception raised by the requester.""" + def fake_requester_exception(url, params, headers, GET, delay, timeout): + raise Exception("Requester error") + monkeypatch.setattr(builtins, "open", fake_open) + monkeypatch.setattr("core.wafDetector.requester", fake_requester_exception) + params = {} + with pytest.raises(Exception, match="Requester error"): + wafDetector("http://example.com", params, {"User-Agent": "test"}, True, 0, 5) + # Ensure that the payload was injected despite the exception. + assert params.get("xss") == '' \ No newline at end of file