From 0b65a07ec48adcd9e719e8d6ad961a003598926a Mon Sep 17 00:00:00 2001 From: Shreya Date: Wed, 31 Mar 2021 14:50:48 +0530 Subject: [PATCH] Format everything with black --- .../network/postgresql_finger.py | 110 +++++---- .../network/test_postgresql_finger.py | 218 ++++++++++-------- 2 files changed, 191 insertions(+), 137 deletions(-) diff --git a/monkey/infection_monkey/network/postgresql_finger.py b/monkey/infection_monkey/network/postgresql_finger.py index 2bc402615..477342806 100644 --- a/monkey/infection_monkey/network/postgresql_finger.py +++ b/monkey/infection_monkey/network/postgresql_finger.py @@ -13,45 +13,48 @@ class PostgreSQLFinger(HostFinger): """ Fingerprints PostgreSQL databases, only on port 5432 """ + # Class related consts - _SCANNED_SERVICE = 'PostgreSQL' + _SCANNED_SERVICE = "PostgreSQL" POSTGRESQL_DEFAULT_PORT = 5432 - CREDS = {'username': ID_STRING, - 'password': ID_STRING} - CONNECTION_DETAILS =\ - { - 'ssl_conf': "SSL is configured on the PostgreSQL server.\n", - 'ssl_not_conf': "SSL is NOT configured on the PostgreSQL server.\n", - 'all_ssl': "SSL connections can be made by all.\n", - 'all_non_ssl': "Non-SSL connections can be made by all.\n", - 'selected_ssl': "SSL connections can be made by selected hosts only OR " - "non-SSL usage is forced.\n", - 'selected_non_ssl': "Non-SSL connections can be made by selected hosts only OR " - "SSL usage is forced.\n", - 'only_selected': "Only selected hosts can make connections (SSL or non-SSL).\n" - } - RELEVANT_EX_SUBSTRINGS =\ - { - 'no_auth': "password authentication failed", - 'no_entry': "entry for host" # "no pg_hba.conf entry for host" but filename may be diff - } + CREDS = {"username": ID_STRING, "password": ID_STRING} + CONNECTION_DETAILS = { + "ssl_conf": "SSL is configured on the PostgreSQL server.\n", + "ssl_not_conf": "SSL is NOT configured on the PostgreSQL server.\n", + "all_ssl": "SSL connections can be made by all.\n", + "all_non_ssl": "Non-SSL connections can be made by all.\n", + "selected_ssl": "SSL connections can be made by selected hosts only OR " + "non-SSL usage is forced.\n", + "selected_non_ssl": "Non-SSL connections can be made by selected hosts only OR " + "SSL usage is forced.\n", + "only_selected": "Only selected hosts can make connections (SSL or non-SSL).\n", + } + RELEVANT_EX_SUBSTRINGS = { + "no_auth": "password authentication failed", + "no_entry": "entry for host", # "no pg_hba.conf entry for host" but filename may be diff + } def get_host_fingerprint(self, host): try: - psycopg2.connect(host=host.ip_addr, - port=self.POSTGRESQL_DEFAULT_PORT, - user=self.CREDS['username'], - password=self.CREDS['password'], - sslmode='prefer', - connect_timeout=MEDIUM_REQUEST_TIMEOUT) # don't need to worry about DB name; creds are wrong, won't check + psycopg2.connect( + host=host.ip_addr, + port=self.POSTGRESQL_DEFAULT_PORT, + user=self.CREDS["username"], + password=self.CREDS["password"], + sslmode="prefer", + connect_timeout=MEDIUM_REQUEST_TIMEOUT, + ) # don't need to worry about DB name; creds are wrong, won't check # if it comes here, the creds worked # this shouldn't happen since capital letters are not supported in postgres usernames # perhaps the service is a honeypot - self.init_service(host.services, self._SCANNED_SERVICE, self.POSTGRESQL_DEFAULT_PORT) - host.services[self._SCANNED_SERVICE]['communication_encryption_details'] =\ - f'The PostgreSQL server was unexpectedly accessible with the credentials - ' +\ - f"user: \'{self.CREDS['username']}\' and password: \'{self.CREDS['password']}\'. Is this a honeypot?" + self.init_service( + host.services, self._SCANNED_SERVICE, self.POSTGRESQL_DEFAULT_PORT + ) + host.services[self._SCANNED_SERVICE]["communication_encryption_details"] = ( + f"The PostgreSQL server was unexpectedly accessible with the credentials - " + + f"user: '{self.CREDS['username']}' and password: '{self.CREDS['password']}'. Is this a honeypot?" + ) return True except psycopg2.OperationalError as ex: @@ -72,13 +75,18 @@ class PostgreSQLFinger(HostFinger): return False def _is_relevant_exception(self, exception_string): - if not any(substr in exception_string for substr in self.RELEVANT_EX_SUBSTRINGS.values()): + if not any( + substr in exception_string + for substr in self.RELEVANT_EX_SUBSTRINGS.values() + ): # OperationalError due to some other reason - irrelevant exception return False return True def analyze_operational_error(self, host, exception_string): - self.init_service(host.services, self._SCANNED_SERVICE, self.POSTGRESQL_DEFAULT_PORT) + self.init_service( + host.services, self._SCANNED_SERVICE, self.POSTGRESQL_DEFAULT_PORT + ) exceptions = exception_string.split("\n") @@ -90,46 +98,58 @@ class PostgreSQLFinger(HostFinger): else: # SSL not configured self.get_connection_details_ssl_not_configured(exceptions) - host.services[self._SCANNED_SERVICE]['communication_encryption_details'] = ''.join(self.ssl_connection_details) + host.services[self._SCANNED_SERVICE][ + "communication_encryption_details" + ] = "".join(self.ssl_connection_details) @staticmethod def is_ssl_configured(exceptions): # when trying to authenticate, it checks pg_hba.conf file: # first, for a record where it can connect with SSL and second, without SSL - if len(exceptions) == 1: # SSL not configured on server so only checks for non-SSL record + if ( + len(exceptions) == 1 + ): # SSL not configured on server so only checks for non-SSL record return False elif len(exceptions) == 2: # SSL configured so checks for both return True def get_connection_details_ssl_configured(self, exceptions): - self.ssl_connection_details.append(self.CONNECTION_DETAILS['ssl_conf']) + self.ssl_connection_details.append(self.CONNECTION_DETAILS["ssl_conf"]) ssl_selected_comms_only = False # check exception message for SSL connection if self.found_entry_for_host_but_pwd_auth_failed(exceptions[0]): - self.ssl_connection_details.append(self.CONNECTION_DETAILS['all_ssl']) + self.ssl_connection_details.append(self.CONNECTION_DETAILS["all_ssl"]) else: - self.ssl_connection_details.append(self.CONNECTION_DETAILS['selected_ssl']) + self.ssl_connection_details.append(self.CONNECTION_DETAILS["selected_ssl"]) ssl_selected_comms_only = True # check exception message for non-SSL connection if self.found_entry_for_host_but_pwd_auth_failed(exceptions[1]): - self.ssl_connection_details.append(self.CONNECTION_DETAILS['all_non_ssl']) + self.ssl_connection_details.append(self.CONNECTION_DETAILS["all_non_ssl"]) else: - if ssl_selected_comms_only: # if only selected SSL allowed and only selected non-SSL allowed - self.ssl_connection_details[-1] = self.CONNECTION_DETAILS['only_selected'] + if ( + ssl_selected_comms_only + ): # if only selected SSL allowed and only selected non-SSL allowed + self.ssl_connection_details[-1] = self.CONNECTION_DETAILS[ + "only_selected" + ] else: - self.ssl_connection_details.append(self.CONNECTION_DETAILS['selected_non_ssl']) + self.ssl_connection_details.append( + self.CONNECTION_DETAILS["selected_non_ssl"] + ) def get_connection_details_ssl_not_configured(self, exceptions): - self.ssl_connection_details.append(self.CONNECTION_DETAILS['ssl_not_conf']) + self.ssl_connection_details.append(self.CONNECTION_DETAILS["ssl_not_conf"]) if self.found_entry_for_host_but_pwd_auth_failed(exceptions[0]): - self.ssl_connection_details.append(self.CONNECTION_DETAILS['all_non_ssl']) + self.ssl_connection_details.append(self.CONNECTION_DETAILS["all_non_ssl"]) else: - self.ssl_connection_details.append(self.CONNECTION_DETAILS['selected_non_ssl']) + self.ssl_connection_details.append( + self.CONNECTION_DETAILS["selected_non_ssl"] + ) @staticmethod def found_entry_for_host_but_pwd_auth_failed(exception): - if PostgreSQLFinger.RELEVANT_EX_SUBSTRINGS['no_auth'] in exception: + if PostgreSQLFinger.RELEVANT_EX_SUBSTRINGS["no_auth"] in exception: return True # entry found in pg_hba.conf file but password authentication failed return False # entry not found in pg_hba.conf file diff --git a/monkey/infection_monkey/network/test_postgresql_finger.py b/monkey/infection_monkey/network/test_postgresql_finger.py index f0bf5998f..4cbf7d94f 100644 --- a/monkey/infection_monkey/network/test_postgresql_finger.py +++ b/monkey/infection_monkey/network/test_postgresql_finger.py @@ -5,83 +5,92 @@ from infection_monkey.network.postgresql_finger import PostgreSQLFinger IRRELEVANT_EXCEPTION_STRING = "This is an irrelevant exception string." -_RELEVANT_EXCEPTION_STRING_PARTS =\ - { - 'pwd_auth_failed': 'FATAL: password authentication failed for user "root"', - 'ssl_on_entry_not_found': 'FATAL: no pg_hba.conf entry for host "127.0.0.1",' - 'user "random", database "postgres", SSL on', - 'ssl_off_entry_not_found': 'FATAL: no pg_hba.conf entry for host "127.0.0.1",' - 'user "random", database "postgres", SSL off' - } +_RELEVANT_EXCEPTION_STRING_PARTS = { + "pwd_auth_failed": 'FATAL: password authentication failed for user "root"', + "ssl_on_entry_not_found": 'FATAL: no pg_hba.conf entry for host "127.0.0.1",' + 'user "random", database "postgres", SSL on', + "ssl_off_entry_not_found": 'FATAL: no pg_hba.conf entry for host "127.0.0.1",' + 'user "random", database "postgres", SSL off', +} -_RELEVANT_EXCEPTION_STRINGS =\ - { - 'pwd_auth_failed': _RELEVANT_EXCEPTION_STRING_PARTS['pwd_auth_failed'], - 'ssl_off_entry_not_found': _RELEVANT_EXCEPTION_STRING_PARTS['ssl_off_entry_not_found'], - 'pwd_auth_failed_pwd_auth_failed': '\n'.join([_RELEVANT_EXCEPTION_STRING_PARTS['pwd_auth_failed'], - _RELEVANT_EXCEPTION_STRING_PARTS['pwd_auth_failed']]), - 'pwd_auth_failed_ssl_off_entry_not_found': '\n'.join([_RELEVANT_EXCEPTION_STRING_PARTS['pwd_auth_failed'], - _RELEVANT_EXCEPTION_STRING_PARTS['ssl_off_entry_not_found']]), - 'ssl_on_entry_not_found_pwd_auth_failed': '\n'.join([_RELEVANT_EXCEPTION_STRING_PARTS['ssl_on_entry_not_found'], - _RELEVANT_EXCEPTION_STRING_PARTS['pwd_auth_failed']]), - 'ssl_on_entry_not_found_ssl_off_entry_not_found': '\n'.join([_RELEVANT_EXCEPTION_STRING_PARTS['ssl_on_entry_not_found'], - _RELEVANT_EXCEPTION_STRING_PARTS['ssl_off_entry_not_found']]) - } - -_RESULT_STRINGS =\ - { - 'ssl_conf': "SSL is configured on the PostgreSQL server.\n", - 'ssl_not_conf': "SSL is NOT configured on the PostgreSQL server.\n", - 'all_ssl': "SSL connections can be made by all.\n", - 'all_non_ssl': "Non-SSL connections can be made by all.\n", - 'selected_ssl': "SSL connections can be made by selected hosts only OR " - "non-SSL usage is forced.\n", - 'selected_non_ssl': "Non-SSL connections can be made by selected hosts only OR " - "SSL usage is forced.\n", - 'only_selected': "Only selected hosts can make connections (SSL or non-SSL).\n" - } - -RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS =\ - { - # SSL not configured, all non-SSL allowed - _RELEVANT_EXCEPTION_STRINGS['pwd_auth_failed']: [ - _RESULT_STRINGS['ssl_not_conf'], - _RESULT_STRINGS['all_non_ssl'] - ], - - # SSL not configured, selected non-SSL allowed - _RELEVANT_EXCEPTION_STRINGS['ssl_off_entry_not_found']: [ - _RESULT_STRINGS['ssl_not_conf'], - _RESULT_STRINGS['selected_non_ssl'] - ], - - # all SSL allowed, all non-SSL allowed - _RELEVANT_EXCEPTION_STRINGS['pwd_auth_failed_pwd_auth_failed']: [ - _RESULT_STRINGS['ssl_conf'], - _RESULT_STRINGS['all_ssl'], - _RESULT_STRINGS['all_non_ssl'] - ], - - # all SSL allowed, selected non-SSL allowed - _RELEVANT_EXCEPTION_STRINGS['pwd_auth_failed_ssl_off_entry_not_found']: [ - _RESULT_STRINGS['ssl_conf'], - _RESULT_STRINGS['all_ssl'], - _RESULT_STRINGS['selected_non_ssl'] - ], - - # selected SSL allowed, all non-SSL allowed - _RELEVANT_EXCEPTION_STRINGS['ssl_on_entry_not_found_pwd_auth_failed']: [ - _RESULT_STRINGS['ssl_conf'], - _RESULT_STRINGS['selected_ssl'], - _RESULT_STRINGS['all_non_ssl'] - ], - - # selected SSL allowed, selected non-SSL allowed - _RELEVANT_EXCEPTION_STRINGS['ssl_on_entry_not_found_ssl_off_entry_not_found']: [ - _RESULT_STRINGS['ssl_conf'], - _RESULT_STRINGS['only_selected'] +_RELEVANT_EXCEPTION_STRINGS = { + "pwd_auth_failed": _RELEVANT_EXCEPTION_STRING_PARTS["pwd_auth_failed"], + "ssl_off_entry_not_found": _RELEVANT_EXCEPTION_STRING_PARTS[ + "ssl_off_entry_not_found" + ], + "pwd_auth_failed_pwd_auth_failed": "\n".join( + [ + _RELEVANT_EXCEPTION_STRING_PARTS["pwd_auth_failed"], + _RELEVANT_EXCEPTION_STRING_PARTS["pwd_auth_failed"], ] - } + ), + "pwd_auth_failed_ssl_off_entry_not_found": "\n".join( + [ + _RELEVANT_EXCEPTION_STRING_PARTS["pwd_auth_failed"], + _RELEVANT_EXCEPTION_STRING_PARTS["ssl_off_entry_not_found"], + ] + ), + "ssl_on_entry_not_found_pwd_auth_failed": "\n".join( + [ + _RELEVANT_EXCEPTION_STRING_PARTS["ssl_on_entry_not_found"], + _RELEVANT_EXCEPTION_STRING_PARTS["pwd_auth_failed"], + ] + ), + "ssl_on_entry_not_found_ssl_off_entry_not_found": "\n".join( + [ + _RELEVANT_EXCEPTION_STRING_PARTS["ssl_on_entry_not_found"], + _RELEVANT_EXCEPTION_STRING_PARTS["ssl_off_entry_not_found"], + ] + ), +} + +_RESULT_STRINGS = { + "ssl_conf": "SSL is configured on the PostgreSQL server.\n", + "ssl_not_conf": "SSL is NOT configured on the PostgreSQL server.\n", + "all_ssl": "SSL connections can be made by all.\n", + "all_non_ssl": "Non-SSL connections can be made by all.\n", + "selected_ssl": "SSL connections can be made by selected hosts only OR " + "non-SSL usage is forced.\n", + "selected_non_ssl": "Non-SSL connections can be made by selected hosts only OR " + "SSL usage is forced.\n", + "only_selected": "Only selected hosts can make connections (SSL or non-SSL).\n", +} + +RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS = { + # SSL not configured, all non-SSL allowed + _RELEVANT_EXCEPTION_STRINGS["pwd_auth_failed"]: [ + _RESULT_STRINGS["ssl_not_conf"], + _RESULT_STRINGS["all_non_ssl"], + ], + # SSL not configured, selected non-SSL allowed + _RELEVANT_EXCEPTION_STRINGS["ssl_off_entry_not_found"]: [ + _RESULT_STRINGS["ssl_not_conf"], + _RESULT_STRINGS["selected_non_ssl"], + ], + # all SSL allowed, all non-SSL allowed + _RELEVANT_EXCEPTION_STRINGS["pwd_auth_failed_pwd_auth_failed"]: [ + _RESULT_STRINGS["ssl_conf"], + _RESULT_STRINGS["all_ssl"], + _RESULT_STRINGS["all_non_ssl"], + ], + # all SSL allowed, selected non-SSL allowed + _RELEVANT_EXCEPTION_STRINGS["pwd_auth_failed_ssl_off_entry_not_found"]: [ + _RESULT_STRINGS["ssl_conf"], + _RESULT_STRINGS["all_ssl"], + _RESULT_STRINGS["selected_non_ssl"], + ], + # selected SSL allowed, all non-SSL allowed + _RELEVANT_EXCEPTION_STRINGS["ssl_on_entry_not_found_pwd_auth_failed"]: [ + _RESULT_STRINGS["ssl_conf"], + _RESULT_STRINGS["selected_ssl"], + _RESULT_STRINGS["all_non_ssl"], + ], + # selected SSL allowed, selected non-SSL allowed + _RELEVANT_EXCEPTION_STRINGS["ssl_on_entry_not_found_ssl_off_entry_not_found"]: [ + _RESULT_STRINGS["ssl_conf"], + _RESULT_STRINGS["only_selected"], + ], +} @pytest.fixture @@ -100,52 +109,77 @@ def host(): def test_irrelevant_exception(mock_PostgreSQLFinger): - assert mock_PostgreSQLFinger._is_relevant_exception(IRRELEVANT_EXCEPTION_STRING) is False + assert ( + mock_PostgreSQLFinger._is_relevant_exception(IRRELEVANT_EXCEPTION_STRING) + is False + ) def test_exception_ssl_not_configured_all_non_ssl_allowed(mock_PostgreSQLFinger, host): - exception = _RELEVANT_EXCEPTION_STRINGS['pwd_auth_failed'] + exception = _RELEVANT_EXCEPTION_STRINGS["pwd_auth_failed"] assert mock_PostgreSQLFinger._is_relevant_exception(exception) is True result = mock_PostgreSQLFinger.analyze_operational_error(host, exception) - assert host.services[mock_PostgreSQLFinger._SCANNED_SERVICE]['communication_encryption_details'] == ''.join(RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS[exception]) + assert host.services[mock_PostgreSQLFinger._SCANNED_SERVICE][ + "communication_encryption_details" + ] == "".join(RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS[exception]) -def test_exception_ssl_not_configured_selected_non_ssl_allowed(mock_PostgreSQLFinger, host): - exception = _RELEVANT_EXCEPTION_STRINGS['ssl_off_entry_not_found'] +def test_exception_ssl_not_configured_selected_non_ssl_allowed( + mock_PostgreSQLFinger, host +): + exception = _RELEVANT_EXCEPTION_STRINGS["ssl_off_entry_not_found"] assert mock_PostgreSQLFinger._is_relevant_exception(exception) is True result = mock_PostgreSQLFinger.analyze_operational_error(host, exception) - assert host.services[mock_PostgreSQLFinger._SCANNED_SERVICE]['communication_encryption_details'] == ''.join(RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS[exception]) + assert host.services[mock_PostgreSQLFinger._SCANNED_SERVICE][ + "communication_encryption_details" + ] == "".join(RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS[exception]) def test_exception_all_ssl_allowed_all_non_ssl_allowed(mock_PostgreSQLFinger, host): - exception = _RELEVANT_EXCEPTION_STRINGS['pwd_auth_failed_pwd_auth_failed'] + exception = _RELEVANT_EXCEPTION_STRINGS["pwd_auth_failed_pwd_auth_failed"] assert mock_PostgreSQLFinger._is_relevant_exception(exception) is True result = mock_PostgreSQLFinger.analyze_operational_error(host, exception) - assert host.services[mock_PostgreSQLFinger._SCANNED_SERVICE]['communication_encryption_details'] == ''.join(RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS[exception]) + assert host.services[mock_PostgreSQLFinger._SCANNED_SERVICE][ + "communication_encryption_details" + ] == "".join(RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS[exception]) -def test_exception_all_ssl_allowed_selected_non_ssl_allowed(mock_PostgreSQLFinger, host): - exception = _RELEVANT_EXCEPTION_STRINGS['pwd_auth_failed_ssl_off_entry_not_found'] +def test_exception_all_ssl_allowed_selected_non_ssl_allowed( + mock_PostgreSQLFinger, host +): + exception = _RELEVANT_EXCEPTION_STRINGS["pwd_auth_failed_ssl_off_entry_not_found"] assert mock_PostgreSQLFinger._is_relevant_exception(exception) is True result = mock_PostgreSQLFinger.analyze_operational_error(host, exception) - assert host.services[mock_PostgreSQLFinger._SCANNED_SERVICE]['communication_encryption_details'] == ''.join(RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS[exception]) + assert host.services[mock_PostgreSQLFinger._SCANNED_SERVICE][ + "communication_encryption_details" + ] == "".join(RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS[exception]) -def test_exception_selected_ssl_allowed_all_non_ssl_allowed(mock_PostgreSQLFinger, host): - exception = _RELEVANT_EXCEPTION_STRINGS['ssl_on_entry_not_found_pwd_auth_failed'] +def test_exception_selected_ssl_allowed_all_non_ssl_allowed( + mock_PostgreSQLFinger, host +): + exception = _RELEVANT_EXCEPTION_STRINGS["ssl_on_entry_not_found_pwd_auth_failed"] assert mock_PostgreSQLFinger._is_relevant_exception(exception) is True result = mock_PostgreSQLFinger.analyze_operational_error(host, exception) - assert host.services[mock_PostgreSQLFinger._SCANNED_SERVICE]['communication_encryption_details'] == ''.join(RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS[exception]) + assert host.services[mock_PostgreSQLFinger._SCANNED_SERVICE][ + "communication_encryption_details" + ] == "".join(RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS[exception]) -def test_exception_selected_ssl_allowed_selected_non_ssl_allowed(mock_PostgreSQLFinger, host): - exception = _RELEVANT_EXCEPTION_STRINGS['ssl_on_entry_not_found_ssl_off_entry_not_found'] +def test_exception_selected_ssl_allowed_selected_non_ssl_allowed( + mock_PostgreSQLFinger, host +): + exception = _RELEVANT_EXCEPTION_STRINGS[ + "ssl_on_entry_not_found_ssl_off_entry_not_found" + ] assert mock_PostgreSQLFinger._is_relevant_exception(exception) is True result = mock_PostgreSQLFinger.analyze_operational_error(host, exception) - assert host.services[mock_PostgreSQLFinger._SCANNED_SERVICE]['communication_encryption_details'] == ''.join(RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS[exception]) + assert host.services[mock_PostgreSQLFinger._SCANNED_SERVICE][ + "communication_encryption_details" + ] == "".join(RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS[exception])