Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _post_send(self, target):
if len(self._post_test_case_methods) > 0:
try:
for f in self._post_test_case_methods:
self._fuzz_data_logger.open_test_step('Post- test case callback: "{0}"'.format(f.__name__))
f(target=target, fuzz_data_logger=self._fuzz_data_logger, session=self, sock=target)
except exception.BoofuzzTargetConnectionReset:
self._fuzz_data_logger.log_fail(constants.ERR_CONN_RESET_FAIL)
except exception.BoofuzzTargetConnectionAborted as e:
self._fuzz_data_logger.log_info(
constants.ERR_CONN_ABORTED.format(socket_errno=e.socket_errno, socket_errmsg=e.socket_errmsg)
)
except exception.BoofuzzTargetConnectionFailedError:
self._fuzz_data_logger.log_fail(constants.ERR_CONN_FAILED)
except Exception:
self._fuzz_data_logger.log_error(
constants.ERR_CALLBACK_FUNC.format(func_name="post_send") + traceback.format_exc()
)
finally:
self._fuzz_data_logger.open_test_step("Cleaning up connections from callbacks")
data = callback_data
else:
data = node.render()
try: # send
self.targets[0].send(data)
self.last_send = data
except exception.BoofuzzTargetConnectionReset:
# TODO: Switch _ignore_connection_reset for _ignore_transmission_error, or provide retry mechanism
if self._ignore_connection_reset:
self._fuzz_data_logger.log_info(constants.ERR_CONN_RESET)
else:
self._fuzz_data_logger.log_fail(constants.ERR_CONN_RESET)
except exception.BoofuzzTargetConnectionAborted as e:
# TODO: Switch _ignore_connection_aborted for _ignore_transmission_error, or provide retry mechanism
msg = constants.ERR_CONN_ABORTED.format(socket_errno=e.socket_errno, socket_errmsg=e.socket_errmsg)
if self._ignore_connection_aborted:
self._fuzz_data_logger.log_info(msg)
else:
self._fuzz_data_logger.log_fail(msg)
try: # recv
if self._receive_data_after_each_request:
self.last_recv = self.targets[0].recv()
if self._check_data_received_each_request:
self._fuzz_data_logger.log_check("Verify some data was received from the target.")
if not self.last_recv:
# Assume a crash?
self._fuzz_data_logger.log_fail("Nothing received from target.")
else:
self._fuzz_data_logger.log_pass("Some data received from target.")
except exception.BoofuzzTargetConnectionReset:
def _post_send(self, target):
if len(self._post_test_case_methods) > 0:
try:
for f in self._post_test_case_methods:
self._fuzz_data_logger.open_test_step('Post- test case callback: "{0}"'.format(f.__name__))
f(target=target, fuzz_data_logger=self._fuzz_data_logger, session=self, sock=target)
except exception.BoofuzzTargetConnectionReset:
self._fuzz_data_logger.log_fail(constants.ERR_CONN_RESET_FAIL)
except exception.BoofuzzTargetConnectionAborted as e:
self._fuzz_data_logger.log_info(
constants.ERR_CONN_ABORTED.format(socket_errno=e.socket_errno, socket_errmsg=e.socket_errmsg)
)
except exception.BoofuzzTargetConnectionFailedError:
self._fuzz_data_logger.log_fail(constants.ERR_CONN_FAILED)
except Exception:
self._fuzz_data_logger.log_error(
constants.ERR_CALLBACK_FUNC.format(func_name="post_send") + traceback.format_exc()
)
finally:
self._fuzz_data_logger.open_test_step("Cleaning up connections from callbacks")
msg = constants.ERR_CONN_ABORTED.format(socket_errno=e.socket_errno, socket_errmsg=e.socket_errmsg)
if self._ignore_connection_issues_when_sending_fuzz_data:
self._fuzz_data_logger.log_info(msg)
else:
self._fuzz_data_logger.log_fail(msg)
try: # recv
if self._receive_data_after_fuzz:
self.last_recv = self.targets[0].recv()
except exception.BoofuzzTargetConnectionReset:
if self._check_data_received_each_request:
self._fuzz_data_logger.log_fail(constants.ERR_CONN_RESET)
else:
self._fuzz_data_logger.log_info(constants.ERR_CONN_RESET)
except exception.BoofuzzTargetConnectionAborted as e:
msg = constants.ERR_CONN_ABORTED.format(socket_errno=e.socket_errno, socket_errmsg=e.socket_errmsg)
if self._check_data_received_each_request:
self._fuzz_data_logger.log_fail(msg)
else:
self._fuzz_data_logger.log_info(msg)
pass
self.last_recv = self.targets[0].recv()
if self._check_data_received_each_request:
self._fuzz_data_logger.log_check("Verify some data was received from the target.")
if not self.last_recv:
# Assume a crash?
self._fuzz_data_logger.log_fail("Nothing received from target.")
else:
self._fuzz_data_logger.log_pass("Some data received from target.")
except exception.BoofuzzTargetConnectionReset:
if self._check_data_received_each_request:
self._fuzz_data_logger.log_fail(constants.ERR_CONN_RESET)
else:
self._fuzz_data_logger.log_info(constants.ERR_CONN_RESET)
except exception.BoofuzzTargetConnectionAborted as e:
msg = constants.ERR_CONN_ABORTED.format(socket_errno=e.socket_errno, socket_errmsg=e.socket_errmsg)
if self._check_data_received_each_request:
self._fuzz_data_logger.log_fail(msg)
else:
self._fuzz_data_logger.log_info(msg)
data = callback_data
else:
data = node.render()
try: # send
self.targets[0].send(data)
self.last_send = data
except exception.BoofuzzTargetConnectionReset:
# TODO: Switch _ignore_connection_reset for _ignore_transmission_error, or provide retry mechanism
if self._ignore_connection_reset:
self._fuzz_data_logger.log_info(constants.ERR_CONN_RESET)
else:
self._fuzz_data_logger.log_fail(constants.ERR_CONN_RESET)
except exception.BoofuzzTargetConnectionAborted as e:
# TODO: Switch _ignore_connection_aborted for _ignore_transmission_error, or provide retry mechanism
msg = constants.ERR_CONN_ABORTED.format(socket_errno=e.socket_errno, socket_errmsg=e.socket_errmsg)
if self._ignore_connection_aborted:
self._fuzz_data_logger.log_info(msg)
else:
self._fuzz_data_logger.log_fail(msg)
try: # recv
if self._receive_data_after_each_request:
self.last_recv = self.targets[0].recv()
if self._check_data_received_each_request:
self._fuzz_data_logger.log_check("Verify some data was received from the target.")
if not self.last_recv:
# Assume a crash?
self._fuzz_data_logger.log_fail("Nothing received from target.")
else:
self._fuzz_data_logger.log_pass("Some data received from target.")
except exception.BoofuzzTargetConnectionReset:
msg = constants.ERR_CONN_ABORTED.format(socket_errno=e.socket_errno, socket_errmsg=e.socket_errmsg)
if self._ignore_connection_issues_when_sending_fuzz_data:
self._fuzz_data_logger.log_info(msg)
else:
self._fuzz_data_logger.log_fail(msg)
try: # recv
if self._receive_data_after_fuzz:
self.last_recv = self.targets[0].recv()
except exception.BoofuzzTargetConnectionReset:
if self._check_data_received_each_request:
self._fuzz_data_logger.log_fail(constants.ERR_CONN_RESET)
else:
self._fuzz_data_logger.log_info(constants.ERR_CONN_RESET)
except exception.BoofuzzTargetConnectionAborted as e:
msg = constants.ERR_CONN_ABORTED.format(socket_errno=e.socket_errno, socket_errmsg=e.socket_errmsg)
if self._check_data_received_each_request:
self._fuzz_data_logger.log_fail(msg)
else:
self._fuzz_data_logger.log_info(msg)
pass