mirror of
				https://github.com/jtesta/ssh-audit.git
				synced 2025-11-03 18:52:15 +01:00 
			
		
		
		
	Streamlined sending of KEXINIT messages.
This commit is contained in:
		@@ -21,14 +21,12 @@
 | 
			
		||||
   OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 | 
			
		||||
   THE SOFTWARE.
 | 
			
		||||
"""
 | 
			
		||||
import os
 | 
			
		||||
 | 
			
		||||
# pylint: disable=unused-import
 | 
			
		||||
from typing import Dict, List, Set, Sequence, Tuple, Iterable  # noqa: F401
 | 
			
		||||
from typing import Callable, Optional, Union, Any  # noqa: F401
 | 
			
		||||
 | 
			
		||||
from ssh_audit.kexdh import KexGroupExchange_SHA1, KexGroupExchange_SHA256
 | 
			
		||||
from ssh_audit.protocol import Protocol
 | 
			
		||||
from ssh_audit.ssh2_kexdb import SSH2_KexDB
 | 
			
		||||
from ssh_audit.ssh2_kex import SSH2_Kex
 | 
			
		||||
from ssh_audit.ssh_socket import SSH_Socket
 | 
			
		||||
@@ -55,10 +53,7 @@ class GEXTest:
 | 
			
		||||
 | 
			
		||||
        # Send our KEX using the specified group-exchange and most of the
 | 
			
		||||
        # server's own values.
 | 
			
		||||
        client_kex = SSH2_Kex(os.urandom(16), [gex_alg], kex.key_algorithms, kex.client, kex.server, False, 0)
 | 
			
		||||
        s.write_byte(Protocol.MSG_KEXINIT)
 | 
			
		||||
        client_kex.write(s)
 | 
			
		||||
        s.send_packet()
 | 
			
		||||
        s.send_kexinit(key_exchanges=[gex_alg], hostkeys=kex.key_algorithms, ciphers=kex.server.encryption, macs=kex.server.mac, compressions=kex.server.compression, languages=kex.server.languages)
 | 
			
		||||
 | 
			
		||||
        # Parse the server's KEX.
 | 
			
		||||
        _, payload = s.read_packet(2)
 | 
			
		||||
 
 | 
			
		||||
@@ -21,14 +21,12 @@
 | 
			
		||||
   OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 | 
			
		||||
   THE SOFTWARE.
 | 
			
		||||
"""
 | 
			
		||||
import os
 | 
			
		||||
 | 
			
		||||
# pylint: disable=unused-import
 | 
			
		||||
from typing import Dict, List, Set, Sequence, Tuple, Iterable  # noqa: F401
 | 
			
		||||
from typing import Callable, Optional, Union, Any  # noqa: F401
 | 
			
		||||
 | 
			
		||||
from ssh_audit.kexdh import KexDH, KexGroup1, KexGroup14_SHA1, KexGroup14_SHA256, KexCurve25519_SHA256, KexGroup16_SHA512, KexGroup18_SHA512, KexGroupExchange_SHA1, KexGroupExchange_SHA256, KexNISTP256, KexNISTP384, KexNISTP521
 | 
			
		||||
from ssh_audit.protocol import Protocol
 | 
			
		||||
from ssh_audit.ssh2_kex import SSH2_Kex
 | 
			
		||||
from ssh_audit.ssh2_kexdb import SSH2_KexDB
 | 
			
		||||
from ssh_audit.ssh_socket import SSH_Socket
 | 
			
		||||
@@ -118,10 +116,7 @@ class HostKeyTest:
 | 
			
		||||
                        return
 | 
			
		||||
 | 
			
		||||
                    # Send our KEX using the specified group-exchange and most of the server's own values.
 | 
			
		||||
                    client_kex = SSH2_Kex(os.urandom(16), [kex_str], [host_key_type], server_kex.client, server_kex.server, False, 0)
 | 
			
		||||
                    s.write_byte(Protocol.MSG_KEXINIT)
 | 
			
		||||
                    client_kex.write(s)
 | 
			
		||||
                    s.send_packet()
 | 
			
		||||
                    s.send_kexinit(key_exchanges=[kex_str], hostkeys=[host_key_type], ciphers=server_kex.server.encryption, macs=server_kex.server.mac, compressions=server_kex.server.compression, languages=server_kex.server.languages)
 | 
			
		||||
 | 
			
		||||
                    # Parse the server's KEX.
 | 
			
		||||
                    _, payload = s.read_packet()
 | 
			
		||||
 
 | 
			
		||||
@@ -798,7 +798,7 @@ def audit(aconf: AuditConf, sshv: Optional[int] = None, print_target: bool = Fal
 | 
			
		||||
        else:
 | 
			
		||||
            err = '[exception] did not receive banner: {}'.format(err)
 | 
			
		||||
    if err is None:
 | 
			
		||||
        s.send_algorithms()  # Send the algorithms we support (except we don't since this isn't a real SSH connection).
 | 
			
		||||
        s.send_kexinit()  # Send the algorithms we support (except we don't since this isn't a real SSH connection).
 | 
			
		||||
 | 
			
		||||
        packet_type, payload = s.read_packet(sshv)
 | 
			
		||||
        if packet_type < 0:
 | 
			
		||||
 
 | 
			
		||||
@@ -230,16 +230,10 @@ class SSH_Socket(ReadBuf, WriteBuf):
 | 
			
		||||
        except socket.error as e:
 | 
			
		||||
            return -1, str(e.args[-1])
 | 
			
		||||
 | 
			
		||||
    def send_algorithms(self) -> None:
 | 
			
		||||
    # Send a KEXINIT with the lists of key exchanges, hostkeys, ciphers, MACs, compressions, and languages that we "support".
 | 
			
		||||
    def send_kexinit(self, key_exchanges: List[str] = ['curve25519-sha256', 'curve25519-sha256@libssh.org', 'ecdh-sha2-nistp256', 'ecdh-sha2-nistp384', 'ecdh-sha2-nistp521', 'diffie-hellman-group-exchange-sha256', 'diffie-hellman-group16-sha512', 'diffie-hellman-group18-sha512', 'diffie-hellman-group14-sha256'], hostkeys: List[str] = ['rsa-sha2-512', 'rsa-sha2-256', 'ssh-rsa', 'ecdsa-sha2-nistp256', 'ssh-ed25519'], ciphers: List[str] = ['chacha20-poly1305@openssh.com', 'aes128-ctr', 'aes192-ctr', 'aes256-ctr', 'aes128-gcm@openssh.com', 'aes256-gcm@openssh.com'], macs: List[str] = ['umac-64-etm@openssh.com', 'umac-128-etm@openssh.com', 'hmac-sha2-256-etm@openssh.com', 'hmac-sha2-512-etm@openssh.com', 'hmac-sha1-etm@openssh.com', 'umac-64@openssh.com', 'umac-128@openssh.com', 'hmac-sha2-256', 'hmac-sha2-512', 'hmac-sha1'], compressions: List[str] = ['none', 'zlib@openssh.com'], languages: List[str] = ['']) -> None:  # pylint: disable=dangerous-default-value
 | 
			
		||||
        '''Sends the list of supported host keys, key exchanges, ciphers, and MACs.  Emulates OpenSSH v8.2.'''
 | 
			
		||||
 | 
			
		||||
        key_exchanges = ['curve25519-sha256', 'curve25519-sha256@libssh.org', 'ecdh-sha2-nistp256', 'ecdh-sha2-nistp384', 'ecdh-sha2-nistp521', 'diffie-hellman-group-exchange-sha256', 'diffie-hellman-group16-sha512', 'diffie-hellman-group18-sha512', 'diffie-hellman-group14-sha256']
 | 
			
		||||
        hostkeys = ['rsa-sha2-512', 'rsa-sha2-256', 'ssh-rsa', 'ecdsa-sha2-nistp256', 'ssh-ed25519']
 | 
			
		||||
        ciphers = ['chacha20-poly1305@openssh.com', 'aes128-ctr', 'aes192-ctr', 'aes256-ctr', 'aes128-gcm@openssh.com', 'aes256-gcm@openssh.com']
 | 
			
		||||
        macs = ['umac-64-etm@openssh.com', 'umac-128-etm@openssh.com', 'hmac-sha2-256-etm@openssh.com', 'hmac-sha2-512-etm@openssh.com', 'hmac-sha1-etm@openssh.com', 'umac-64@openssh.com', 'umac-128@openssh.com', 'hmac-sha2-256', 'hmac-sha2-512', 'hmac-sha1']
 | 
			
		||||
        compressions = ['none', 'zlib@openssh.com']
 | 
			
		||||
        languages = ['']
 | 
			
		||||
 | 
			
		||||
        kexparty = SSH2_KexParty(ciphers, macs, compressions, languages)
 | 
			
		||||
        kex = SSH2_Kex(os.urandom(16), key_exchanges, hostkeys, kexparty, kexparty, False, 0)
 | 
			
		||||
 | 
			
		||||
 
 | 
			
		||||
		Reference in New Issue
	
	Block a user