Format everything with black

This commit is contained in:
Shreya 2021-03-31 14:50:48 +05:30
parent b0f85f6857
commit 0b65a07ec4
2 changed files with 191 additions and 137 deletions

View File

@ -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 "
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 "
"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"
"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
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,
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
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

View File

@ -5,82 +5,91 @@ 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",'
_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'
"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']
_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"
],
# 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']
"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"],
],
}
@ -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])