Git
/
ssh-audit
mirror of https://github.com/jtesta/ssh-audit.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Results from concurrent scans against multiple hosts are no longer improperly combined (#190).

This commit is contained in:
Joe Testa 2023-06-19 14:13:32 -04:00
parent 521a50a796
commit 639f11a5e5
8 changed files with 81 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@ -179,6 +179,7 @@ For convenience, a web front-end on top of the command-line tool is available at
## ChangeLog
### v3.0.0-dev (2023-??-??)
- Results from concurrent scans against multiple hosts are no longer improperly combined; bug discovered by [Adam Russell](https://github.com/thecliguy).
- Added 1 new key exchange: `curve448-sha512@libssh.org`.
### v2.9.0 (2023-04-29)

4
src/ssh_audit/algorithms.py
View File

@ -54,7 +54,7 @@ class Algorithms:
def ssh1(self) -> Optional['Algorithms.Item']:
if self.ssh1kex is None:
return None
item = Algorithms.Item(1, SSH1_KexDB.ALGORITHMS)
item = Algorithms.Item(1, SSH1_KexDB.get_db())
item.add('key', ['ssh-rsa1'])
item.add('enc', self.ssh1kex.supported_ciphers)
item.add('aut', self.ssh1kex.supported_authentications)
@ -64,7 +64,7 @@ class Algorithms:
def ssh2(self) -> Optional['Algorithms.Item']:
if self.ssh2kex is None:
return None
item = Algorithms.Item(2, SSH2_KexDB.ALGORITHMS)
item = Algorithms.Item(2, SSH2_KexDB.get_db())
item.add('kex', self.ssh2kex.kex_algorithms)
item.add('key', self.ssh2kex.key_algorithms)
item.add('enc', self.ssh2kex.server.encryption)

4
src/ssh_audit/gextest.py
View File

@ -208,7 +208,7 @@ class GEXTest:
# We flag moduli smaller than 2048 as a failure.
if smallest_modulus < 2048:
text = 'using small %d-bit modulus' % smallest_modulus
lst = SSH2_KexDB.ALGORITHMS['kex'][gex_alg]
lst = SSH2_KexDB.get_db()['kex'][gex_alg]
# For 'diffie-hellman-group-exchange-sha256', add
# a failure reason.
if len(lst) == 1:
@ -222,7 +222,7 @@ class GEXTest:
# Moduli smaller than 3072 get flagged as a warning.
elif smallest_modulus < 3072:
lst = SSH2_KexDB.ALGORITHMS['kex'][gex_alg]
lst = SSH2_KexDB.get_db()['kex'][gex_alg]
# Ensure that a warning list exists for us to append to, below.
while len(lst) < 3:

18
src/ssh_audit/hostkeytest.py
View File

@ -216,16 +216,18 @@ class HostKeyTest:
# If the current key is a member of the RSA family, then populate all RSA family members with the same
# failure and/or warning comments.
while len(SSH2_KexDB.ALGORITHMS['key'][rsa_type]) < 3:
SSH2_KexDB.ALGORITHMS['key'][rsa_type].append([])
db = SSH2_KexDB.get_db()
while len(db['key'][rsa_type]) < 3:
db['key'][rsa_type].append([])
SSH2_KexDB.ALGORITHMS['key'][rsa_type][1].extend(key_fail_comments)
SSH2_KexDB.ALGORITHMS['key'][rsa_type][2].extend(key_warn_comments)
db['key'][rsa_type][1].extend(key_fail_comments)
db['key'][rsa_type][2].extend(key_warn_comments)
else:
host_key_types[host_key_type]['parsed'] = True
while len(SSH2_KexDB.ALGORITHMS['key'][host_key_type]) < 3:
SSH2_KexDB.ALGORITHMS['key'][host_key_type].append([])
db = SSH2_KexDB.get_db()
while len(db['key'][host_key_type]) < 3:
db['key'][host_key_type].append([])
SSH2_KexDB.ALGORITHMS['key'][host_key_type][1].extend(key_fail_comments)
SSH2_KexDB.ALGORITHMS['key'][host_key_type][2].extend(key_warn_comments)
db['key'][host_key_type][1].extend(key_fail_comments)
db['key'][host_key_type][2].extend(key_warn_comments)

28
src/ssh_audit/ssh1_kexdb.py
View File

@ -22,6 +22,9 @@
THE SOFTWARE.
"""
# pylint: disable=unused-import
import copy
import threading
from typing import Dict, List, Set, Sequence, Tuple, Iterable # noqa: F401
from typing import Callable, Optional, Union, Any # noqa: F401
@ -34,7 +37,9 @@ class SSH1_KexDB: # pylint: disable=too-few-public-methods
FAIL_NA_UNSAFE = 'not implemented in OpenSSH (server), unsafe algorithm'
TEXT_CIPHER_IDEA = 'cipher used by commercial SSH'
ALGORITHMS: Dict[str, Dict[str, List[List[Optional[str]]]]] = {
DB_PER_THREAD: Dict[int, Dict[str, Dict[str, List[List[Optional[str]]]]]] = {}
MASTER_DB: Dict[str, Dict[str, List[List[Optional[str]]]]] = {
'key': {
'ssh-rsa1': [['1.2.2']],
},
@ -56,3 +61,24 @@ class SSH1_KexDB: # pylint: disable=too-few-public-methods
'kerberos': [['1.2.2', '3.6'], [FAIL_OPENSSH37_REMOVE]],
}
}
@staticmethod
def get_db() -> Dict[str, Dict[str, List[List[Optional[str]]]]]:
'''Returns a copy of the MASTER_DB that is private to the calling thread. This prevents multiple threads from polluting the results of other threads.'''
calling_thread_id = threading.get_ident()
if calling_thread_id not in SSH1_KexDB.DB_PER_THREAD:
SSH1_KexDB.DB_PER_THREAD[calling_thread_id] = copy.deepcopy(SSH1_KexDB.MASTER_DB)
return SSH1_KexDB.DB_PER_THREAD[calling_thread_id]
@staticmethod
def thread_exit() -> None:
'''Deletes the calling thread's copy of the MASTER_DB. This is needed because, in rare circumstances, a terminated thread's ID can be re-used by new threads.'''
calling_thread_id = threading.get_ident()
if calling_thread_id in SSH1_KexDB.DB_PER_THREAD:
del SSH1_KexDB.DB_PER_THREAD[calling_thread_id]

28
src/ssh_audit/ssh2_kexdb.py
View File

@ -23,6 +23,9 @@
THE SOFTWARE.
"""
# pylint: disable=unused-import
import copy
import threading
from typing import Dict, List, Set, Sequence, Tuple, Iterable # noqa: F401
from typing import Callable, Optional, Union, Any # noqa: F401
@ -69,8 +72,10 @@ class SSH2_KexDB: # pylint: disable=too-few-public-methods
INFO_REMOVED_IN_OPENSSH70 = 'removed in OpenSSH 7.0: https://www.openssh.com/txt/release-7.0'
INFO_WITHDRAWN_PQ_ALG = 'the sntrup4591761 algorithm was withdrawn, as it may not provide strong post-quantum security'
# Maintains a dictionary per calling thread that yields its own copy of MASTER_DB. This prevents results from one thread polluting the results of another thread.
DB_PER_THREAD: Dict[int, Dict[str, Dict[str, List[List[Optional[str]]]]]] = {}
ALGORITHMS: Dict[str, Dict[str, List[List[Optional[str]]]]] = {
MASTER_DB: Dict[str, Dict[str, List[List[Optional[str]]]]] = {
# Format: 'algorithm_name': [['version_first_appeared_in'], [reason_for_failure1, reason_for_failure2, ...], [warning1, warning2, ...], [info1, info2, ...]]
'kex': {
'Curve25519SHA256': [[]],
@ -390,3 +395,24 @@ class SSH2_KexDB: # pylint: disable=too-few-public-methods
'umac-96@openssh.com': [[], [], [WARN_ENCRYPT_AND_MAC], [INFO_NEVER_IMPLEMENTED_IN_OPENSSH]],
}
}
@staticmethod
def get_db() -> Dict[str, Dict[str, List[List[Optional[str]]]]]:
'''Returns a copy of the MASTER_DB that is private to the calling thread. This prevents multiple threads from polluting the results of other threads.'''
calling_thread_id = threading.get_ident()
if calling_thread_id not in SSH2_KexDB.DB_PER_THREAD:
SSH2_KexDB.DB_PER_THREAD[calling_thread_id] = copy.deepcopy(SSH2_KexDB.MASTER_DB)
return SSH2_KexDB.DB_PER_THREAD[calling_thread_id]
@staticmethod
def thread_exit() -> None:
'''Deletes the calling thread's copy of the MASTER_DB. This is needed because, in rare circumstances, a terminated thread's ID can be re-used by new threads.'''
calling_thread_id = threading.get_ident()
if calling_thread_id in SSH2_KexDB.DB_PER_THREAD:
del SSH2_KexDB.DB_PER_THREAD[calling_thread_id]

17
src/ssh_audit/ssh_audit.py
View File

@ -446,10 +446,11 @@ def post_process_findings(banner: Optional[Banner], algs: Algorithms) -> List[st
if (algs.ssh2kex is not None and 'diffie-hellman-group-exchange-sha256' in algs.ssh2kex.kex_algorithms and 'diffie-hellman-group-exchange-sha256' in algs.ssh2kex.dh_modulus_sizes() and algs.ssh2kex.dh_modulus_sizes()['diffie-hellman-group-exchange-sha256'] == 2048) and (banner is not None and banner.software is not None and banner.software.find('OpenSSH') != -1):
# Ensure a list for notes exists.
while len(SSH2_KexDB.ALGORITHMS['kex']['diffie-hellman-group-exchange-sha256']) < 4:
SSH2_KexDB.ALGORITHMS['kex']['diffie-hellman-group-exchange-sha256'].append([])
db = SSH2_KexDB.get_db()
while len(db['kex']['diffie-hellman-group-exchange-sha256']) < 4:
db['kex']['diffie-hellman-group-exchange-sha256'].append([])
SSH2_KexDB.ALGORITHMS['kex']['diffie-hellman-group-exchange-sha256'][3].append("A bug in OpenSSH causes it to fall back to a 2048-bit modulus regardless of server configuration (https://bugzilla.mindrot.org/show_bug.cgi?id=2793)")
db['kex']['diffie-hellman-group-exchange-sha256'][3].append("A bug in OpenSSH causes it to fall back to a 2048-bit modulus regardless of server configuration (https://bugzilla.mindrot.org/show_bug.cgi?id=2793)")
# Ensure that this algorithm doesn't appear in the recommendations section since the user cannot control this OpenSSH bug.
algorithm_recommendation_suppress_list.append('diffie-hellman-group-exchange-sha256')
@ -522,7 +523,7 @@ def output(out: OutputBuffer, aconf: AuditConf, banner: Optional[Banner], header
# SSHv1
if pkm is not None:
adb = SSH1_KexDB.ALGORITHMS
adb = SSH1_KexDB.get_db()
ciphers = pkm.supported_ciphers
auths = pkm.supported_authentications
title, atype = 'SSH1 host-key algorithms', 'key'
@ -534,7 +535,7 @@ def output(out: OutputBuffer, aconf: AuditConf, banner: Optional[Banner], header
# SSHv2
if kex is not None:
adb = SSH2_KexDB.ALGORITHMS
adb = SSH2_KexDB.get_db()
title, atype = 'key exchange algorithms', 'kex'
program_retval = output_algorithms(out, title, adb, atype, kex.kex_algorithms, unknown_algorithms, aconf.json, program_retval, maxlen, dh_modulus_sizes=kex.dh_modulus_sizes())
title, atype = 'host-key algorithms', 'key'
@ -1124,7 +1125,7 @@ def algorithm_lookup(out: OutputBuffer, alg_names: str) -> int:
}
algorithm_names = alg_names.split(",")
adb = SSH2_KexDB.ALGORITHMS
adb = SSH2_KexDB.get_db()
# Use nested dictionary comprehension to iterate an outer dictionary where
# each key is an alg type that consists of a value (which is itself a
@ -1376,6 +1377,10 @@ def main() -> int:
if aconf.json:
print(']')
# Send notification that this thread is exiting. This deletes the thread's local copy of the algorithm databases.
SSH1_KexDB.thread_exit()
SSH2_KexDB.thread_exit()
else: # Just a scan against a single target.
ret = audit(out, aconf)
out.write()

2
test/test_ssh2_kexdb.py
View File

@ -7,7 +7,7 @@ class Test_SSH2_KexDB:
@pytest.fixture(autouse=True)
def init(self):
self.db = SSH2_KexDB.ALGORITHMS
self.db = SSH2_KexDB.get_db()
def test_ssh2_kexdb(self):
'''Ensures that the SSH2_KexDB.ALGORITHMS dictionary is in the right format.'''