From 3672aaf2ec3f2901ef37dd40fda3dfc5000dc2fb Mon Sep 17 00:00:00 2001 From: "Mr. Walls" Date: Mon, 30 Sep 2024 00:51:28 -0700 Subject: [PATCH 1/5] [TESTING] implementing basic fuzzing (- WIP #13 -) Changes in file tests/requirements.txt: added hypothisis for fuzzing New file tests/test_fuzz.py: - Implements initial idea from #13 --- tests/requirements.txt | 2 + tests/test_fuzz.py | 110 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 112 insertions(+) create mode 100644 tests/test_fuzz.py diff --git a/tests/requirements.txt b/tests/requirements.txt index 18cb0da5..95654421 100644 --- a/tests/requirements.txt +++ b/tests/requirements.txt @@ -39,6 +39,8 @@ tox>=3.0.0, !=3.16.1 ### # flake8 - MIT license flake8>=3.9.2 +# hypothesis - MPL2 License (may not be bundled, take care with docker) +hypothesis>=6.112.2 # pyflakes - MIT license # pyflakes>=2.5.0 # pep8 - MIT license diff --git a/tests/test_fuzz.py b/tests/test_fuzz.py new file mode 100644 index 00000000..c3d3fcb1 --- /dev/null +++ b/tests/test_fuzz.py @@ -0,0 +1,110 @@ +#! /usr/bin/env python3 +# -*- coding: utf-8 -*- + +# Python Test Repo Template +# .................................. +# Copyright (c) 2017-2024, Mr. Walls +# .................................. +# Licensed under MIT (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# .......................................... +# http://www.github.com/reactive-firewall/python-repo/LICENSE.md +# .......................................... +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +__module__ = """tests""" + + +try: + import sys + if sys.__name__ is None: # pragma: no branch + raise ModuleNotFoundError("[CWE-440] OMG! we could not import sys! ABORT. ABORT.") from None +except Exception as err: # pragma: no branch + raise ImportError(err) from err + + +try: + try: + import context + except Exception as ImportErr: # pragma: no branch + ImportErr = None + del ImportErr # skipcq - cleanup any error leaks early + from . import context + if context.__name__ is None: + raise ModuleNotFoundError("[CWE-758] Failed to import context") from None + else: + from context import multicast # pylint: disable=cyclic-import - skipcq: PYL-R0401 + from context import unittest + from context import subprocess + from hypothesis import given, strategies as st + from context import Process + import random as _random +except Exception as err: + raise ImportError("[CWE-758] Failed to import test context") from err + +class HypothesisTestSuite(context.BasicUsageTestSuite): + + __module__ = """tests.test_fuzz""" + + __name__ = """tests.test_fuzz.HypothesisTestSuite""" + + @staticmethod + def _always_generate_random_port_WHEN_called(): + """Outputs a psuedo-random, RFC-6335 compliant, port number.""" + return _random.randint(49152, 65535) + + @given(st.binary(min_size=1, max_size=1472)) + def test_multicast_sender_with_random_data(self, data): + """Tests the basic send and recv test. Skips if fuzzing broke SAY fixture.""" + theResult = False + fail_fixture = str("""SAY --> HEAR == error""") + _fixture_port_num = self._always_generate_random_port_WHEN_called() + try: + self.assertIsNotNone(_fixture_port_num) + self.assertEqual(type(_fixture_port_num), type(int(0))) + _fixture_SAY_args = [ + """--port""", str(_fixture_port_num), + """--mcast-group""", """'224.0.0.1'""", + """--message""", str("""'{d}'""").format(d=data) + ] + _fixture_HEAR_args = [ + """--port""", str(_fixture_port_num), + """--join-mcast-groups""", """'224.0.0.1'""", + """--bind-group""", """'224.0.0.1'""" + ] + p = Process( + target=multicast.__main__.McastDispatch().doStep, + name="HEAR", args=("HEAR", _fixture_HEAR_args,) + ) + p.start() + try: + self.assertIsNotNone( + multicast.__main__.McastDispatch().doStep("SAY", _fixture_SAY_args) + ) + self.assertIsNotNone( + multicast.__main__.McastDispatch().doStep("SAY", _fixture_SAY_args) + ) + self.assertIsNotNone( + multicast.__main__.McastDispatch().doStep("SAY", _fixture_SAY_args) + ) + except Exception as _cause: + p.join() + raise unittest.SkipTest(fail_fixture) from _cause + p.join() + self.assertIsNotNone(p.exitcode) + self.assertEqual(int(p.exitcode), int(0)) + theResult = (int(p.exitcode) <= int(0)) + except Exception as err: + context.debugtestError(err) + self.skipTest(fail_fixture) + theResult = False + self.assertTrue(theResult, fail_fixture) + + +if __name__ == '__main__': + unittest.main() From 069fb6385d2ca8f63aebff97c830756175699669 Mon Sep 17 00:00:00 2001 From: "Mr. Walls" Date: Mon, 30 Sep 2024 02:17:28 -0700 Subject: [PATCH 2/5] [DOCUMENTATION] Added doc strings to new fuzz testing (- WIP #13 -) Changes in file tests/test_fuzz.py: - added docstrings --- tests/test_fuzz.py | 45 +++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 41 insertions(+), 4 deletions(-) diff --git a/tests/test_fuzz.py b/tests/test_fuzz.py index c3d3fcb1..113830c5 100644 --- a/tests/test_fuzz.py +++ b/tests/test_fuzz.py @@ -40,7 +40,6 @@ else: from context import multicast # pylint: disable=cyclic-import - skipcq: PYL-R0401 from context import unittest - from context import subprocess from hypothesis import given, strategies as st from context import Process import random as _random @@ -48,6 +47,22 @@ raise ImportError("[CWE-758] Failed to import test context") from err class HypothesisTestSuite(context.BasicUsageTestSuite): + """ + A test suite that uses Hypothesis to perform fuzz testing on the multicast sender and receiver. + + This class extends `context.BasicUsageTestSuite` and provides a test case + to verify the robustness of the multicast implementation when handling + random binary data of varying sizes. + + Methods: + - `test_multicast_sender_with_random_data(self, data)`: Tests sending and + receiving random binary data using the multicast sender and receiver. + + Note: + This test ensures that the multicast sender and receiver can handle binary + data up to 1472 bytes, which is the typical maximum size for UDP packets + without fragmentation. + """ __module__ = """tests.test_fuzz""" @@ -55,12 +70,35 @@ class HypothesisTestSuite(context.BasicUsageTestSuite): @staticmethod def _always_generate_random_port_WHEN_called(): - """Outputs a psuedo-random, RFC-6335 compliant, port number.""" + """ + Generates a pseudo-random port number within the dynamic/private port range. + + This method returns a random port number between 49152 and 65535, + compliant with RFC 6335, suitable for temporary testing purposes to + avoid port conflicts. + + Returns: + int: A random port number between 49152 and 65535. + """ return _random.randint(49152, 65535) @given(st.binary(min_size=1, max_size=1472)) def test_multicast_sender_with_random_data(self, data): - """Tests the basic send and recv test. Skips if fuzzing broke SAY fixture.""" + """ + Tests the multicast sender and receiver with random binary data. + + This test uses the `hypothesis` library to generate random binary data + between 1 and 1472 bytes, sends it using the multicast sender, and verifies + that the multicast receiver successfully receives the data. + + Args: + data (bytes): Random binary data generated by Hypothesis. + + Notes: + - A random port number is used for each test run to prevent port collisions. + - The test sets up a receiver process and sends the data multiple times. + - If the receiver process encounters an error, the test is skipped. + """ theResult = False fail_fixture = str("""SAY --> HEAR == error""") _fixture_port_num = self._always_generate_random_port_WHEN_called() @@ -102,7 +140,6 @@ def test_multicast_sender_with_random_data(self, data): except Exception as err: context.debugtestError(err) self.skipTest(fail_fixture) - theResult = False self.assertTrue(theResult, fail_fixture) From 8a58cf32e58274582e88b26631afc4bcb66f4afd Mon Sep 17 00:00:00 2001 From: "Mr. Walls" Date: Mon, 30 Sep 2024 02:38:33 -0700 Subject: [PATCH 3/5] [PATCH] apply as per review (- WIP #13 -) Changes in file tests/test_fuzz.py: - use self.assertIsInstance --- tests/test_fuzz.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_fuzz.py b/tests/test_fuzz.py index 113830c5..4133c667 100644 --- a/tests/test_fuzz.py +++ b/tests/test_fuzz.py @@ -104,7 +104,7 @@ def test_multicast_sender_with_random_data(self, data): _fixture_port_num = self._always_generate_random_port_WHEN_called() try: self.assertIsNotNone(_fixture_port_num) - self.assertEqual(type(_fixture_port_num), type(int(0))) + self.assertIsInstance(_fixture_port_num, int) _fixture_SAY_args = [ """--port""", str(_fixture_port_num), """--mcast-group""", """'224.0.0.1'""", From 8a43b452cd0e111c51d4ffcfdf01149db66f9f14 Mon Sep 17 00:00:00 2001 From: "Mr. Walls" Date: Mon, 30 Sep 2024 02:47:03 -0700 Subject: [PATCH 4/5] [PATCH] workaround for legacy testing ### ChangeLog: Changes in file tests/check_legacy_setup_coverage: printf "%s\n" "" >> ${_TEST_ROOT_DIR}/${LOG_FILE} ; wait ; --- tests/check_legacy_setup_coverage | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/tests/check_legacy_setup_coverage b/tests/check_legacy_setup_coverage index 1bd9f20c..f8a086dc 100755 --- a/tests/check_legacy_setup_coverage +++ b/tests/check_legacy_setup_coverage @@ -170,8 +170,14 @@ wait ; printf "%s\n" "End of Setup.py Test one-shot" >> ${_TEST_ROOT_DIR}/${LOG_FILE} ; wait ; printf "%s\n" "" >> ${_TEST_ROOT_DIR}/${LOG_FILE} ; wait ; -${COVERAGE_CMD} combine 2>${_TEST_ROOT_DIR}/${ERR_FILE} || EXIT_CODE=2 ; -${COVERAGE_CMD} xml --include=setup.py,multicast -o ${_TEST_ROOT_DIR}/test-reports/coverage_setup.xml || EXIT_CODE=2 ; +# sloppy workaround for now +if [[ ( -x $(command -v python3) ) ]] ; then + $(command -v python3) -m coverage combine 2>${_TEST_ROOT_DIR}/${ERR_FILE} || EXIT_CODE=2 ; + $(command -v python3) -m coverage xml --include=setup.py,multicast -o ${_TEST_ROOT_DIR}/test-reports/coverage_setup.xml || EXIT_CODE=2 ; +elif [[ ( -x $(command -v coverage) ) ]] ; then + $(command -v coverage) combine 2>${_TEST_ROOT_DIR}/${ERR_FILE} || EXIT_CODE=2 ; + $(command -v coverage) xml --include=setup.py,multicast -o ${_TEST_ROOT_DIR}/test-reports/coverage_setup.xml || EXIT_CODE=2 ; +fi wait ; cp -f ./${LOG_FILE} ${_TEST_ROOT_DIR}/test-reports/legacy_setup_version_log.log 2>./${ERR_FILE} ; wait ; From 0a41b0f3da962ef9209b26bccc65625b7f87269c Mon Sep 17 00:00:00 2001 From: "Mr. Walls" Date: Mon, 30 Sep 2024 03:19:06 -0700 Subject: [PATCH 5/5] [STYLE] fixed spaceing ### ChangeLog: Changes in file tests/test_fuzz.py: class HypothesisTestSuite(context.BasicUsageTestSuite): --- tests/test_fuzz.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_fuzz.py b/tests/test_fuzz.py index 4133c667..1feacc9f 100644 --- a/tests/test_fuzz.py +++ b/tests/test_fuzz.py @@ -56,7 +56,7 @@ class HypothesisTestSuite(context.BasicUsageTestSuite): Methods: - `test_multicast_sender_with_random_data(self, data)`: Tests sending and - receiving random binary data using the multicast sender and receiver. + receiving random binary data using the multicast sender and receiver. Note: This test ensures that the multicast sender and receiver can handle binary