ssh-audit/test/test_ssh2.py

import os
import struct
import pytest


# pylint: disable=line-too-long,attribute-defined-outside-init
class TestSSH2:
    @pytest.fixture(autouse=True)
    def init(self, ssh_audit):
        self.ssh = ssh_audit.SSH
        self.ssh2 = ssh_audit.SSH2
        self.rbuf = ssh_audit.ReadBuf
        self.wbuf = ssh_audit.WriteBuf
        self.audit = ssh_audit.audit
        self.AuditConf = ssh_audit.AuditConf

    def _conf(self):
        conf = self.AuditConf('localhost', 22)
        conf.colors = False
        conf.batch = True
        conf.verbose = True
        conf.ssh1 = False
        conf.ssh2 = True
        return conf

    @classmethod
    def _create_ssh2_packet(cls, payload):
        padding = -(len(payload) + 5) % 8
        if padding < 4:
            padding += 8
        plen = len(payload) + padding + 1
        pad_bytes = b'\x00' * padding
        data = struct.pack('>Ib', plen, padding) + payload + pad_bytes
        return data

    def _kex_payload(self):
        w = self.wbuf()
        w.write(b'\x00\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff')
        w.write_list(['bogus_kex1', 'bogus_kex2'])  # We use a bogus kex, otherwise the host key tests will kick off and fail.
        w.write_list(['ssh-rsa', 'rsa-sha2-512', 'rsa-sha2-256', 'ssh-ed25519'])
        w.write_list(['chacha20-poly1305@openssh.com', 'aes128-ctr', 'aes192-ctr', 'aes256-ctr', 'aes128-gcm@openssh.com', 'aes256-gcm@openssh.com', 'aes128-cbc', 'aes192-cbc', 'aes256-cbc'])
        w.write_list(['chacha20-poly1305@openssh.com', 'aes128-ctr', 'aes192-ctr', 'aes256-ctr', 'aes128-gcm@openssh.com', 'aes256-gcm@openssh.com', 'aes128-cbc', 'aes192-cbc', 'aes256-cbc'])
        w.write_list(['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'])
        w.write_list(['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'])
        w.write_list(['none', 'zlib@openssh.com'])
        w.write_list(['none', 'zlib@openssh.com'])
        w.write_list([''])
        w.write_list([''])
        w.write_byte(False)
        w.write_int(0)
        return w.write_flush()

    def test_kex_read(self):
        kex = self.ssh2.Kex.parse(self._kex_payload())
        assert kex is not None
        assert kex.cookie == b'\x00\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff'
        assert kex.kex_algorithms == ['bogus_kex1', 'bogus_kex2']
        assert kex.key_algorithms == ['ssh-rsa', 'rsa-sha2-512', 'rsa-sha2-256', 'ssh-ed25519']
        assert kex.client is not None
        assert kex.server is not None
        assert kex.client.encryption == ['chacha20-poly1305@openssh.com', 'aes128-ctr', 'aes192-ctr', 'aes256-ctr', 'aes128-gcm@openssh.com', 'aes256-gcm@openssh.com', 'aes128-cbc', 'aes192-cbc', 'aes256-cbc']
        assert kex.server.encryption == ['chacha20-poly1305@openssh.com', 'aes128-ctr', 'aes192-ctr', 'aes256-ctr', 'aes128-gcm@openssh.com', 'aes256-gcm@openssh.com', 'aes128-cbc', 'aes192-cbc', 'aes256-cbc']
        assert kex.client.mac == ['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']
        assert kex.server.mac == ['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']
        assert kex.client.compression == ['none', 'zlib@openssh.com']
        assert kex.server.compression == ['none', 'zlib@openssh.com']
        assert kex.client.languages == ['']
        assert kex.server.languages == ['']
        assert kex.follows is False
        assert kex.unused == 0

    def _get_empty_kex(self, cookie=None):
        kex_algs, key_algs = [], []
        enc, mac, compression, languages = [], [], ['none'], []
        cli = self.ssh2.KexParty(enc, mac, compression, languages)
        enc, mac, compression, languages = [], [], ['none'], []
        srv = self.ssh2.KexParty(enc, mac, compression, languages)
        if cookie is None:
            cookie = os.urandom(16)
        kex = self.ssh2.Kex(cookie, kex_algs, key_algs, cli, srv, 0)
        return kex

    def _get_kex_variat1(self):
        cookie = b'\x00\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff'
        kex = self._get_empty_kex(cookie)
        kex.kex_algorithms.append('bogus_kex1')
        kex.kex_algorithms.append('bogus_kex2')
        kex.key_algorithms.append('ssh-rsa')
        kex.key_algorithms.append('rsa-sha2-512')
        kex.key_algorithms.append('rsa-sha2-256')
        kex.key_algorithms.append('ssh-ed25519')
        kex.server.encryption.append('chacha20-poly1305@openssh.com')
        kex.server.encryption.append('aes128-ctr')
        kex.server.encryption.append('aes192-ctr')
        kex.server.encryption.append('aes256-ctr')
        kex.server.encryption.append('aes128-gcm@openssh.com')
        kex.server.encryption.append('aes256-gcm@openssh.com')
        kex.server.encryption.append('aes128-cbc')
        kex.server.encryption.append('aes192-cbc')
        kex.server.encryption.append('aes256-cbc')
        kex.server.mac.append('umac-64-etm@openssh.com')
        kex.server.mac.append('umac-128-etm@openssh.com')
        kex.server.mac.append('hmac-sha2-256-etm@openssh.com')
        kex.server.mac.append('hmac-sha2-512-etm@openssh.com')
        kex.server.mac.append('hmac-sha1-etm@openssh.com')
        kex.server.mac.append('umac-64@openssh.com')
        kex.server.mac.append('umac-128@openssh.com')
        kex.server.mac.append('hmac-sha2-256')
        kex.server.mac.append('hmac-sha2-512')
        kex.server.mac.append('hmac-sha1')
        kex.server.compression.append('zlib@openssh.com')
        for a in kex.server.encryption:
            kex.client.encryption.append(a)
        for a in kex.server.mac:
            kex.client.mac.append(a)
        for a in kex.server.compression:
            if a == 'none':
                continue
            kex.client.compression.append(a)
        return kex

    def test_key_payload(self):
        kex1 = self._get_kex_variat1()
        kex2 = self.ssh2.Kex.parse(self._kex_payload())
        assert kex1.payload == kex2.payload

    def test_ssh2_server_simple(self, output_spy, virtual_socket):
        vsocket = virtual_socket
        w = self.wbuf()
        w.write_byte(self.ssh.Protocol.MSG_KEXINIT)
        w.write(self._kex_payload())
        vsocket.rdata.append(b'SSH-2.0-OpenSSH_7.3 ssh-audit-test\r\n')
        vsocket.rdata.append(self._create_ssh2_packet(w.write_flush()))
        output_spy.begin()
        self.audit(self._conf())
        lines = output_spy.flush()
        assert len(lines) == 67

    def test_ssh2_server_invalid_first_packet(self, output_spy, virtual_socket):
        vsocket = virtual_socket
        w = self.wbuf()
        w.write_byte(self.ssh.Protocol.MSG_KEXINIT + 1)
        vsocket.rdata.append(b'SSH-2.0-OpenSSH_7.3 ssh-audit-test\r\n')
        vsocket.rdata.append(self._create_ssh2_packet(w.write_flush()))
        output_spy.begin()
        ret = self.audit(self._conf())
        assert ret != 0
        lines = output_spy.flush()
        assert len(lines) == 3
        assert 'unknown message' in lines[-1]