Remove PostgreSQL fingerprinter

Issue #1077
This commit is contained in:
Mike Salvatore 2021-04-22 10:10:09 -04:00
parent 6a54c1e85a
commit 3b32ef2f4a
9 changed files with 13 additions and 412 deletions

View File

@ -32,7 +32,6 @@ STATUS_FAILED = "Failed"
# Don't change order! The statuses are ordered by importance/severity. # Don't change order! The statuses are ordered by importance/severity.
ORDERED_TEST_STATUSES = [STATUS_FAILED, STATUS_VERIFY, STATUS_PASSED, STATUS_UNEXECUTED] ORDERED_TEST_STATUSES = [STATUS_FAILED, STATUS_VERIFY, STATUS_PASSED, STATUS_UNEXECUTED]
TEST_DATA_ENDPOINT_POSTGRESQL = "unencrypted_data_endpoint_postgresql"
TEST_DATA_ENDPOINT_ELASTIC = "unencrypted_data_endpoint_elastic" TEST_DATA_ENDPOINT_ELASTIC = "unencrypted_data_endpoint_elastic"
TEST_DATA_ENDPOINT_HTTP = "unencrypted_data_endpoint_http" TEST_DATA_ENDPOINT_HTTP = "unencrypted_data_endpoint_http"
TEST_MACHINE_EXPLOITED = "machine_exploited" TEST_MACHINE_EXPLOITED = "machine_exploited"
@ -58,7 +57,6 @@ TESTS = (
TEST_MACHINE_EXPLOITED, TEST_MACHINE_EXPLOITED,
TEST_DATA_ENDPOINT_HTTP, TEST_DATA_ENDPOINT_HTTP,
TEST_DATA_ENDPOINT_ELASTIC, TEST_DATA_ENDPOINT_ELASTIC,
TEST_DATA_ENDPOINT_POSTGRESQL,
TEST_TUNNELING, TEST_TUNNELING,
TEST_COMMUNICATE_AS_NEW_USER, TEST_COMMUNICATE_AS_NEW_USER,
TEST_SCOUTSUITE_PERMISSIVE_FIREWALL_RULES, TEST_SCOUTSUITE_PERMISSIVE_FIREWALL_RULES,
@ -197,19 +195,6 @@ TESTS_MAP = {
PILLARS_KEY: [DATA], PILLARS_KEY: [DATA],
POSSIBLE_STATUSES_KEY: [STATUS_UNEXECUTED, STATUS_FAILED, STATUS_PASSED], POSSIBLE_STATUSES_KEY: [STATUS_UNEXECUTED, STATUS_FAILED, STATUS_PASSED],
}, },
TEST_DATA_ENDPOINT_POSTGRESQL: {
TEST_EXPLANATION_KEY: "The Monkey scanned for unencrypted access to " "PostgreSQL servers.",
FINDING_EXPLANATION_BY_STATUS_KEY: {
STATUS_FAILED: "Monkey accessed PostgreSQL servers. Limit access to data by encrypting"
" it in in-transit.",
STATUS_PASSED: "Monkey didn't find open PostgreSQL servers. If you have such servers, "
"look for alerts that "
"indicate attempts to access them. ",
},
PRINCIPLE_KEY: PRINCIPLE_DATA_CONFIDENTIALITY,
PILLARS_KEY: [DATA],
POSSIBLE_STATUSES_KEY: [STATUS_UNEXECUTED, STATUS_FAILED, STATUS_PASSED],
},
TEST_TUNNELING: { TEST_TUNNELING: {
TEST_EXPLANATION_KEY: "The Monkey tried to tunnel traffic using other monkeys.", TEST_EXPLANATION_KEY: "The Monkey tried to tunnel traffic using other monkeys.",
FINDING_EXPLANATION_BY_STATUS_KEY: { FINDING_EXPLANATION_BY_STATUS_KEY: {

View File

@ -14,7 +14,6 @@ netifaces = ">=0.10.9"
odict = "==1.7.0" odict = "==1.7.0"
paramiko = ">=2.7.1" paramiko = ">=2.7.1"
psutil = ">=5.7.0" psutil = ">=5.7.0"
psycopg2-binary = "==2.8.6"
pycryptodome = "==3.9.8" pycryptodome = "==3.9.8"
pyftpdlib = "==1.5.6" pyftpdlib = "==1.5.6"
pymssql = "==2.1.5" pymssql = "==2.1.5"

View File

@ -1,7 +1,7 @@
{ {
"_meta": { "_meta": {
"hash": { "hash": {
"sha256": "e1f1176012139ef9cca4ca5e51e9b150729ae69c73f381e678fb9d0181318c64" "sha256": "78392106c26ec104e3734ec0b69975330baa9ff7761d63e1091e2687bb149265"
}, },
"pipfile-spec": 6, "pipfile-spec": 6,
"requires": { "requires": {
@ -61,19 +61,19 @@
}, },
"boto3": { "boto3": {
"hashes": [ "hashes": [
"sha256:1e55df93aa47a84e2a12a639c7f145e16e6e9ef959542d69d5526d50d2e92692", "sha256:5910c868c2cf0d30b6c9caed1d38a2b2c2c83e9713eadae0f43de4f42bfe863f",
"sha256:eab42daaaf68cdad5b112d31dcb0684162098f6558ba7b64156be44f993525fa" "sha256:d0d1e8ca76a8e1b74f87a8324f97001d60bd8bbe6cca35a8e9e7b9abe5aa9ddb"
], ],
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4, 3.5'", "markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4, 3.5'",
"version": "==1.17.54" "version": "==1.17.55"
}, },
"botocore": { "botocore": {
"hashes": [ "hashes": [
"sha256:20a864fc6570ba11d52532c72c3ccabab5c71a9b4a9418601a313d56f1d2ce5b", "sha256:5632c129e6c1c1a15e273fd3ec6f4431490e99ec61b6cff833538f456202e833",
"sha256:37ec76ea2df8609540ba6cb0fe360ae1c589d2e1ee91eb642fd767823f3fcedd" "sha256:94a62f7f848b37757c3419193727e183bccdf5cb74167df30bafee5d8d649b7a"
], ],
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4, 3.5'", "markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4, 3.5'",
"version": "==1.20.54" "version": "==1.20.55"
}, },
"certifi": { "certifi": {
"hashes": [ "hashes": [
@ -544,47 +544,6 @@
"index": "pypi", "index": "pypi",
"version": "==5.8.0" "version": "==5.8.0"
}, },
"psycopg2-binary": {
"hashes": [
"sha256:0deac2af1a587ae12836aa07970f5cb91964f05a7c6cdb69d8425ff4c15d4e2c",
"sha256:0e4dc3d5996760104746e6cfcdb519d9d2cd27c738296525d5867ea695774e67",
"sha256:11b9c0ebce097180129e422379b824ae21c8f2a6596b159c7659e2e5a00e1aa0",
"sha256:15978a1fbd225583dd8cdaf37e67ccc278b5abecb4caf6b2d6b8e2b948e953f6",
"sha256:1fabed9ea2acc4efe4671b92c669a213db744d2af8a9fc5d69a8e9bc14b7a9db",
"sha256:2dac98e85565d5688e8ab7bdea5446674a83a3945a8f416ad0110018d1501b94",
"sha256:42ec1035841b389e8cc3692277a0bd81cdfe0b65d575a2c8862cec7a80e62e52",
"sha256:6422f2ff0919fd720195f64ffd8f924c1395d30f9a495f31e2392c2efafb5056",
"sha256:6a32f3a4cb2f6e1a0b15215f448e8ce2da192fd4ff35084d80d5e39da683e79b",
"sha256:7312e931b90fe14f925729cde58022f5d034241918a5c4f9797cac62f6b3a9dd",
"sha256:7d92a09b788cbb1aec325af5fcba9fed7203897bbd9269d5691bb1e3bce29550",
"sha256:833709a5c66ca52f1d21d41865a637223b368c0ee76ea54ca5bad6f2526c7679",
"sha256:89705f45ce07b2dfa806ee84439ec67c5d9a0ef20154e0e475e2b2ed392a5b83",
"sha256:8cd0fb36c7412996859cb4606a35969dd01f4ea34d9812a141cd920c3b18be77",
"sha256:950bc22bb56ee6ff142a2cb9ee980b571dd0912b0334aa3fe0fe3788d860bea2",
"sha256:a0c50db33c32594305b0ef9abc0cb7db13de7621d2cadf8392a1d9b3c437ef77",
"sha256:a0eb43a07386c3f1f1ebb4dc7aafb13f67188eab896e7397aa1ee95a9c884eb2",
"sha256:aaa4213c862f0ef00022751161df35804127b78adf4a2755b9f991a507e425fd",
"sha256:ac0c682111fbf404525dfc0f18a8b5f11be52657d4f96e9fcb75daf4f3984859",
"sha256:ad20d2eb875aaa1ea6d0f2916949f5c08a19c74d05b16ce6ebf6d24f2c9f75d1",
"sha256:b4afc542c0ac0db720cf516dd20c0846f71c248d2b3d21013aa0d4ef9c71ca25",
"sha256:b8a3715b3c4e604bcc94c90a825cd7f5635417453b253499664f784fc4da0152",
"sha256:ba28584e6bca48c59eecbf7efb1576ca214b47f05194646b081717fa628dfddf",
"sha256:ba381aec3a5dc29634f20692349d73f2d21f17653bda1decf0b52b11d694541f",
"sha256:bd1be66dde2b82f80afb9459fc618216753f67109b859a361cf7def5c7968729",
"sha256:c2507d796fca339c8fb03216364cca68d87e037c1f774977c8fc377627d01c71",
"sha256:cec7e622ebc545dbb4564e483dd20e4e404da17ae07e06f3e780b2dacd5cee66",
"sha256:d14b140a4439d816e3b1229a4a525df917d6ea22a0771a2a78332273fd9528a4",
"sha256:d1b4ab59e02d9008efe10ceabd0b31e79519da6fb67f7d8e8977118832d0f449",
"sha256:d5227b229005a696cc67676e24c214740efd90b148de5733419ac9aaba3773da",
"sha256:e1f57aa70d3f7cc6947fd88636a481638263ba04a742b4a37dd25c373e41491a",
"sha256:e74a55f6bad0e7d3968399deb50f61f4db1926acf4a6d83beaaa7df986f48b1c",
"sha256:e82aba2188b9ba309fd8e271702bd0d0fc9148ae3150532bbb474f4590039ffb",
"sha256:ee69dad2c7155756ad114c02db06002f4cded41132cc51378e57aad79cc8e4f4",
"sha256:f5ab93a2cb2d8338b1674be43b442a7f544a0971da062a5da774ed40587f18f5"
],
"index": "pypi",
"version": "==2.8.6"
},
"pyasn1": { "pyasn1": {
"hashes": [ "hashes": [
"sha256:014c0e9976956a08139dc0712ae195324a75e142284d5f87f1a87ee1b068a359", "sha256:014c0e9976956a08139dc0712ae195324a75e142284d5f87f1a87ee1b068a359",
@ -831,10 +790,10 @@
}, },
"s3transfer": { "s3transfer": {
"hashes": [ "hashes": [
"sha256:af1af6384bd7fb8208b06480f9be73d0295d965c4c073a5c95ea5b6661dccc18", "sha256:81b7b3516739b0cfbecaa9077a1baf783e7a790c0e49261fcc6ceda468765efa",
"sha256:f3dfd791cad2799403e3c8051810a7ca6ee1d2e630e5d2a8f9649d892bdb3db6" "sha256:b5130849df909773254099d085790456665f8d7e0032bbef6e3407f28adb1ad9"
], ],
"version": "==0.4.0" "version": "==0.4.1"
}, },
"scoutsuite": { "scoutsuite": {
"git": "git://github.com/guardicode/ScoutSuite", "git": "git://github.com/guardicode/ScoutSuite",
@ -957,7 +916,7 @@
"sha256:a2ad9c0f6d70f6e0e0d1f54b8582054c62d8a09f346b5ccaf55da68628ca10e1", "sha256:a2ad9c0f6d70f6e0e0d1f54b8582054c62d8a09f346b5ccaf55da68628ca10e1",
"sha256:a64624a25fc2d3663a2c5376c5291f3c7531e9c8051571de9ca9db8bf25746c2" "sha256:a64624a25fc2d3663a2c5376c5291f3c7531e9c8051571de9ca9db8bf25746c2"
], ],
"markers": "python_version >= '3.6'", "markers": "platform_system == 'Windows'",
"version": "==0.0.9" "version": "==0.0.9"
}, },
"winsys-3.x": { "winsys-3.x": {
@ -972,6 +931,7 @@
"sha256:1d6b085e5c445141c475476000b661f60fff1aaa19f76bf82b7abb92e0ff4942", "sha256:1d6b085e5c445141c475476000b661f60fff1aaa19f76bf82b7abb92e0ff4942",
"sha256:b6a6be5711b1b6c8d55bda7a8befd75c48c12b770b9d227d31c1737dbf0d40a6" "sha256:b6a6be5711b1b6c8d55bda7a8befd75c48c12b770b9d227d31c1737dbf0d40a6"
], ],
"index": "pypi",
"markers": "sys_platform == 'win32'", "markers": "sys_platform == 'win32'",
"version": "==1.5.1" "version": "==1.5.1"
}, },

View File

@ -175,7 +175,7 @@ class Configuration(object):
8008, # HTTP alternate 8008, # HTTP alternate
7001, # Oracle Weblogic default server port 7001, # Oracle Weblogic default server port
] ]
tcp_target_ports = [22, 2222, 445, 135, 3389, 80, 8080, 443, 8008, 3306, 9200, 5432] tcp_target_ports = [22, 2222, 445, 135, 3389, 80, 8080, 443, 8008, 3306, 9200]
tcp_target_ports.extend(HTTP_PORTS) tcp_target_ports.extend(HTTP_PORTS)
tcp_scan_timeout = 3000 # 3000 Milliseconds tcp_scan_timeout = 3000 # 3000 Milliseconds
tcp_scan_interval = 0 # in milliseconds tcp_scan_interval = 0 # in milliseconds

View File

@ -1,142 +0,0 @@
import logging
import psycopg2
from common.common_consts.timeouts import MEDIUM_REQUEST_TIMEOUT
from infection_monkey.model import ID_STRING
from infection_monkey.network.HostFinger import HostFinger
LOG = logging.getLogger(__name__)
class PostgreSQLFinger(HostFinger):
"""
Fingerprints PostgreSQL databases, only on port 5432
"""
# Class related consts
_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
}
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
# 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"] = (
"The PostgreSQL server was unexpectedly accessible with the credentials - "
+ f"user: '{self.CREDS['username']}' and password: '"
f"{self.CREDS['password']}'. Is this a honeypot?"
)
return True
except psycopg2.OperationalError as ex:
# try block will throw an OperationalError since the credentials are wrong, which we
# then analyze
try:
exception_string = str(ex)
if not self._is_relevant_exception(exception_string):
return False
# all's well; start analyzing errors
self.analyze_operational_error(host, exception_string)
return True
except Exception as err:
LOG.debug("Error getting PostgreSQL fingerprint: %s", err)
return False
def _is_relevant_exception(self, exception_string):
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)
exceptions = exception_string.split("\n")
self.ssl_connection_details = []
ssl_conf_on_server = self.is_ssl_configured(exceptions)
if ssl_conf_on_server: # SSL configured
self.get_connection_details_ssl_configured(exceptions)
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
)
@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
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"])
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"])
else:
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"])
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"]
else:
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"])
if self.found_entry_for_host_but_pwd_auth_failed(exceptions[0]):
self.ssl_connection_details.append(self.CONNECTION_DETAILS["all_non_ssl"])
else:
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:
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

@ -59,14 +59,5 @@ FINGER_CLASSES = {
"info": "Checks if ElasticSearch is running and attempts to find it's " "version.", "info": "Checks if ElasticSearch is running and attempts to find it's " "version.",
"attack_techniques": ["T1210"], "attack_techniques": ["T1210"],
}, },
{
"type": "string",
"enum": ["PostgreSQLFinger"],
"title": "PostgreSQLFinger",
"safe": True,
"info": "Checks if PostgreSQL service is running and if "
"its communication is encrypted.",
"attack_techniques": ["T1210"],
},
], ],
} }

View File

@ -219,7 +219,6 @@ INTERNAL = {
"MySQLFinger", "MySQLFinger",
"MSSQLFinger", "MSSQLFinger",
"ElasticFinger", "ElasticFinger",
"PostgreSQLFinger",
], ],
} }
}, },

View File

@ -9,7 +9,6 @@ from monkey_island.cc.services.zero_trust.monkey_findings.monkey_zt_finding_serv
) )
HTTP_SERVERS_SERVICES_NAMES = ["tcp-80"] HTTP_SERVERS_SERVICES_NAMES = ["tcp-80"]
POSTGRESQL_SERVER_SERVICE_NAME = "PostgreSQL"
def check_open_data_endpoints(telemetry_json): def check_open_data_endpoints(telemetry_json):
@ -17,7 +16,6 @@ def check_open_data_endpoints(telemetry_json):
current_monkey = Monkey.get_single_monkey_by_guid(telemetry_json["monkey_guid"]) current_monkey = Monkey.get_single_monkey_by_guid(telemetry_json["monkey_guid"])
found_http_server_status = zero_trust_consts.STATUS_PASSED found_http_server_status = zero_trust_consts.STATUS_PASSED
found_elastic_search_server = zero_trust_consts.STATUS_PASSED found_elastic_search_server = zero_trust_consts.STATUS_PASSED
found_postgresql_server = zero_trust_consts.STATUS_PASSED
events = [ events = [
Event.create_event( Event.create_event(
@ -66,20 +64,6 @@ def check_open_data_endpoints(telemetry_json):
event_type=zero_trust_consts.EVENT_TYPE_MONKEY_NETWORK, event_type=zero_trust_consts.EVENT_TYPE_MONKEY_NETWORK,
) )
) )
if service_name == POSTGRESQL_SERVER_SERVICE_NAME:
found_postgresql_server = zero_trust_consts.STATUS_FAILED
events.append(
Event.create_event(
title="Scan telemetry analysis",
message="Service {} on {} recognized as an open data endpoint! "
"Service details: {}".format(
service_data["display_name"],
telemetry_json["data"]["machine"]["ip_addr"],
json.dumps(service_data),
),
event_type=zero_trust_consts.EVENT_TYPE_MONKEY_NETWORK,
)
)
MonkeyZTFindingService.create_or_add_to_existing( MonkeyZTFindingService.create_or_add_to_existing(
test=zero_trust_consts.TEST_DATA_ENDPOINT_HTTP, test=zero_trust_consts.TEST_DATA_ENDPOINT_HTTP,
@ -93,10 +77,4 @@ def check_open_data_endpoints(telemetry_json):
events=events, events=events,
) )
MonkeyZTFindingService.create_or_add_to_existing(
test=zero_trust_consts.TEST_DATA_ENDPOINT_POSTGRESQL,
status=found_postgresql_server,
events=events,
)
MonkeyZTFindingService.add_malicious_activity_to_timeline(events) MonkeyZTFindingService.add_malicious_activity_to_timeline(events)

View File

@ -1,169 +0,0 @@
import pytest
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_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
def mock_PostgreSQLFinger():
return PostgreSQLFinger()
class DummyHost:
def __init__(self):
self.services = {}
@pytest.fixture
def host():
return DummyHost()
def test_irrelevant_exception(mock_PostgreSQLFinger):
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"]
assert mock_PostgreSQLFinger._is_relevant_exception(exception) is True
mock_PostgreSQLFinger.analyze_operational_error(host, 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"]
assert mock_PostgreSQLFinger._is_relevant_exception(exception) is True
mock_PostgreSQLFinger.analyze_operational_error(host, 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"]
assert mock_PostgreSQLFinger._is_relevant_exception(exception) is True
mock_PostgreSQLFinger.analyze_operational_error(host, 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"]
assert mock_PostgreSQLFinger._is_relevant_exception(exception) is True
mock_PostgreSQLFinger.analyze_operational_error(host, 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"]
assert mock_PostgreSQLFinger._is_relevant_exception(exception) is True
mock_PostgreSQLFinger.analyze_operational_error(host, 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"]
assert mock_PostgreSQLFinger._is_relevant_exception(exception) is True
mock_PostgreSQLFinger.analyze_operational_error(host, exception)
assert host.services[mock_PostgreSQLFinger._SCANNED_SERVICE][
"communication_encryption_details"
] == "".join(RELEVANT_EXCEPTIONS_WITH_EXPECTED_RESULTS[exception])