Release 1.7.0.

By default both IPv4 and IPv6 is supported and order of precedence depends on OS.
By using -46, IPv4 is prefered, but by using -64, IPv6 is preferd.
For now the old way how to specify port (host:port) has been kept intact.

.gitignore

@@ -1,2 +1,5 @@
 *~
 *.pyc
+html/
+venv/
+.cache/

.travis.yml

@@ -0,0 +1,18 @@
+language: python
+python:
+  - 2.6
+  - 2.7
+  - 3.3
+  - 3.4
+  - 3.5
+  - pypy
+  - pypy3
+install:
+  - pip install --upgrade pytest
+  - pip install --upgrade pytest-cov
+  - pip install --upgrade coveralls
+script:
+  - py.test --cov-report= --cov=ssh-audit -v test
+after_success:
+  - coveralls
+

README.md

@@ -1,4 +1,6 @@
 # ssh-audit
+[![build status](https://api.travis-ci.org/arthepsy/ssh-audit.svg)](https://travis-ci.org/arthepsy/ssh-audit)
+[![coverage status](https://coveralls.io/repos/github/arthepsy/ssh-audit/badge.svg)](https://coveralls.io/github/arthepsy/ssh-audit)  
 **ssh-audit** is a tool for ssh server auditing.  
 
 ## Features
@@ -6,30 +8,50 @@
 - grab banner, recognize device or software and operating system, detect compression;
 - gather key-exchange, host-key, encryption and message authentication code algorithms;
 - output algorithm information (available since, removed/disabled, unsafe/weak/legacy, etc);
+- output algorithm recommendations (append or remove based on recognized software version);
 - output security information (related issues, assigned CVE list, etc);
 - analyze SSH version compatibility based on algorithm information;
-- historical information from OpenSSH and Dropbear SSH;
-- no dependencies, compatible with Python2 and Python3;
+- historical information from OpenSSH, Dropbear SSH and libssh;
+- no dependencies, compatible with Python 2.6+, Python 3.x and PyPy;
 
 ## Usage
 ```
-usage: ssh-audit.py [-bnv] [-l <level>] <host[:port]>
+usage: ssh-audit.py [-1246pbnvl] <host>
 
    -1,  --ssh1             force ssh version 1 only
-   -2,  --ssh2             force ssh version 1 only
+   -2,  --ssh2             force ssh version 2 only
+   -4,  --ipv4             enable IPv4 (order of precedence)
+   -6,  --ipv6             enable IPv6 (order of precedence)
+   -p,  --port=<port>      port to connect
    -b,  --batch            batch output
    -n,  --no-colors        disable colors
    -v,  --verbose          verbose output
    -l,  --level=<level>    minimum output level (info|warn|fail)
    
 ```
+* if both IPv4 and IPv6 are used, order of precedence can be set by using either `-46` or `-64`.  
 * batch flag `-b` will output sections without header and without empty lines (implies verbose flag).  
 * verbose flag `-v` will prefix each line with section type and algorithm name.  
 
 ### example
-![screenshot](https://cloud.githubusercontent.com/assets/7356025/17623665/da5281c8-60a9-11e6-9582-13f9971c22e0.png)  
+![screenshot](https://cloud.githubusercontent.com/assets/7356025/19233757/3e09b168-8ef0-11e6-91b4-e880bacd0b8a.png)
 
 ## ChangeLog
+### v1.7.0 (2016-10-26)
+ - implement options to allow specify IPv4/IPv6 usage and order of precedence
+ - implement option to specify remote port (old behavior kept for compatibility)
+ - add colors support for Microsoft Windows via optional colorama dependency
+ - fix encoding and decoding issues, add tests, do not crash on encoding errors
+ - use mypy-lang for static type checking and verify all code
+
+### v1.6.0 (2016-10-14)
+ - implement algorithm recommendations section (based on recognized software)
+ - implement full libssh support (version history, algorithms, security, etc)
+ - fix SSH-1.99 banner recognition and version comparison functionality
+ - do not output empty algorithms (happens for misconfigured servers)
+ - make consistent output for Python 3.x versions
+ - add a lot more tests (conf, banner, software, SSH1/SSH2, output, etc)
+ - use Travis CI to test for multiple Python versions (2.6-3.5, pypy, pypy3)
 
 ### v1.5.0 (2016-09-20)
  - create security section for related security information
@@ -37,7 +59,7 @@ usage: ssh-audit.py [-bnv] [-l <level>] <host[:port]>
  - implement full SSH1 support with fingerprint information
  - automatically fallback to SSH1 on protocol mismatch
  - add new options to force SSH1 or SSH2 (both allowed by default)
- - parse banner information and convert it to specific sofware and OS version
+ - parse banner information and convert it to specific software and OS version
  - do not use padding in batch mode
  - several fixes (Cisco sshd, rare hangs, error handling, etc)
 

test/conftest.py

@@ -1,10 +1,130 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
-import os, sys
+import os
+import io
+import sys
+import socket
 import pytest
 
+
+if sys.version_info[0] == 2:
+	import StringIO  # pylint: disable=import-error
+	StringIO = StringIO.StringIO
+else:
+	StringIO = io.StringIO
+
+
 @pytest.fixture(scope='module')
 def ssh_audit():
 	__rdir = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..')
 	sys.path.append(os.path.abspath(__rdir))
 	return __import__('ssh-audit')
+
+
+# pylint: disable=attribute-defined-outside-init
+class _OutputSpy(list):
+	def begin(self):
+		self.__out = StringIO()
+		self.__old_stdout = sys.stdout
+		sys.stdout = self.__out
+		
+	def flush(self):
+		lines = self.__out.getvalue().splitlines()
+		sys.stdout = self.__old_stdout
+		self.__out = None
+		return lines
+
+
+@pytest.fixture(scope='module')
+def output_spy():
+	return _OutputSpy()
+
+
+class _VirtualSocket(object):
+	def __init__(self):
+		self.sock_address = ('127.0.0.1', 0)
+		self.peer_address = None
+		self._connected = False
+		self.timeout = -1.0
+		self.rdata = []
+		self.sdata = []
+		self.errors = {}
+	
+	def _check_err(self, method):
+		method_error = self.errors.get(method)
+		if method_error:
+			raise method_error
+	
+	def connect(self, address):
+		return self._connect(address, False)
+	
+	def _connect(self, address, ret=True):
+		self.peer_address = address
+		self._connected = True
+		self._check_err('connect')
+		return self if ret else None
+	
+	def settimeout(self, timeout):
+		self.timeout = timeout
+	
+	def gettimeout(self):
+		return self.timeout
+	
+	def getpeername(self):
+		if self.peer_address is None or not self._connected:
+			raise socket.error(57, 'Socket is not connected')
+		return self.peer_address
+	
+	def getsockname(self):
+		return self.sock_address
+	
+	def bind(self, address):
+		self.sock_address = address
+	
+	def listen(self, backlog):
+		pass
+	
+	def accept(self):
+		# pylint: disable=protected-access
+		conn = _VirtualSocket()
+		conn.sock_address = self.sock_address
+		conn.peer_address = ('127.0.0.1', 0)
+		conn._connected = True
+		return conn, conn.peer_address
+	
+	def recv(self, bufsize, flags=0):
+		# pylint: disable=unused-argument
+		if not self._connected:
+			raise socket.error(54, 'Connection reset by peer')
+		if not len(self.rdata) > 0:
+			return b''
+		data = self.rdata.pop(0)
+		if isinstance(data, Exception):
+			raise data
+		return data
+	
+	def send(self, data):
+		if self.peer_address is None or not self._connected:
+			raise socket.error(32, 'Broken pipe')
+		self._check_err('send')
+		self.sdata.append(data)
+
+
+@pytest.fixture()
+def virtual_socket(monkeypatch):
+	vsocket = _VirtualSocket()
+	
+	# pylint: disable=unused-argument
+	def _socket(family=socket.AF_INET,
+	            socktype=socket.SOCK_STREAM,
+	            proto=0,
+	            fileno=None):
+		return vsocket
+	
+	def _cc(address, timeout=0, source_address=None):
+		# pylint: disable=protected-access
+		return vsocket._connect(address, True)
+	
+	monkeypatch.setattr(socket, 'create_connection', _cc)
+	monkeypatch.setattr(socket, 'socket', _socket)
+	return vsocket

test/coverage.sh

@@ -0,0 +1,10 @@
+#!/bin/sh
+_cdir=$(cd -- "$(dirname "$0")" && pwd)
+type py.test > /dev/null 2>&1
+if [ $? -ne 0 ]; then
+	echo "err: py.test (Python testing framework) not found."
+	exit 1
+fi
+cd -- "${_cdir}/.."
+mkdir -p html
+py.test -v --cov-report=html:html/coverage --cov=ssh-audit test

test/mypy-py2.sh

@@ -0,0 +1,10 @@
+#!/bin/sh
+_cdir=$(cd -- "$(dirname "$0")" && pwd)
+type mypy > /dev/null 2>&1
+if [ $? -ne 0 ]; then
+	echo "err: mypy (Optional Static Typing for Python) not found."
+	exit 1
+fi
+_htmldir="${_cdir}/../html/mypy-py2"
+mkdir -p "${_htmldir}"
+mypy --python-version 2.7 --config-file "${_cdir}/mypy.ini" --html-report "${_htmldir}" "${_cdir}/../ssh-audit.py"

test/mypy-py3.sh

@@ -0,0 +1,10 @@
+#!/bin/sh
+_cdir=$(cd -- "$(dirname "$0")" && pwd)
+type mypy > /dev/null 2>&1
+if [ $? -ne 0 ]; then
+	echo "err: mypy (Optional Static Typing for Python) not found."
+	exit 1
+fi
+_htmldir="${_cdir}/../html/mypy-py3"
+mkdir -p "${_htmldir}"
+mypy --python-version 3.5 --config-file "${_cdir}/mypy.ini" --html-report "${_htmldir}" "${_cdir}/../ssh-audit.py"

test/mypy.ini

@@ -0,0 +1,9 @@
+[mypy]
+silent_imports = True
+disallow_untyped_calls = True
+disallow_untyped_defs = True
+check_untyped_defs = True
+disallow-subclassing-any = True
+warn-incomplete-stub = True
+warn-redundant-casts = True
+

test/prospector.sh

@@ -5,4 +5,9 @@ if [ $? -ne 0 ]; then
 	echo "err: prospector (Python Static Analysis) not found."
 	exit 1
 fi
-prospector --profile-path "${_cdir}" -P prospector "${_cdir}/../ssh-audit.py"
+if [ X"$1" == X"" ]; then
+	_file="${_cdir}/../ssh-audit.py"
+else
+	_file="$1"
+fi
+prospector -E --profile-path "${_cdir}" -P prospector "${_file}"

test/prospector.yml

@@ -1,9 +1,42 @@
-inherits:
-    - strictness_veryhigh
+strictness: veryhigh
+doc-warnings: false
+
+pylint:
+    disable:
+        - multiple-imports
+        - invalid-name
+        - trailing-whitespace
+
+    options:
+        max-args: 8 # default: 5
+        max-locals: 20 # default: 15
+        max-returns: 6
+        max-branches: 15 # default: 12
+        max-statements: 60 # default: 50
+        max-parents: 7
+        max-attributes: 8 # default: 7
+        min-public-methods: 1 # default: 2
+        max-public-methods: 20
+        max-bool-expr: 5
+        max-nested-blocks: 6 # default: 5
+        max-line-length: 80 # default: 100
+        ignore-long-lines: ^\s*(#\s+type:\s+.*|[A-Z0-9_]+\s+=\s+.*|('.*':\s+)?\[.*\],?)$
+        max-module-lines: 2500 # default: 10000
 
 pep8:
     disable:
-        - W191
-        - W293
-        - E501
-        - E221
+        - W191 # indentation contains tabs
+        - W293 # blank line contains whitespace
+        - E101 # indentation contains mixed spaces and tabs
+        - E401 # multiple imports on one line
+        - E501 # line too long
+        - E221 # multiple spaces before operator
+
+pyflakes:
+    disable:
+        - F401 # module imported but unused
+        - F821 # undefined name
+
+mccabe:
+    options:
+        max-complexity: 15

test/test_auditconf.py

@@ -0,0 +1,192 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import pytest
+
+
+# pylint: disable=attribute-defined-outside-init
+class TestAuditConf(object):
+	@pytest.fixture(autouse=True)
+	def init(self, ssh_audit):
+		self.AuditConf = ssh_audit.AuditConf
+		self.usage = ssh_audit.usage
+	
+	@classmethod
+	def _test_conf(cls, conf, **kwargs):
+		options = {
+			'host': None,
+			'port': 22,
+			'ssh1': True,
+			'ssh2': True,
+			'batch': False,
+			'colors': True,
+			'verbose': False,
+			'minlevel': 'info',
+			'ipv4': True,
+			'ipv6': True,
+			'ipvo': ()
+		}
+		for k, v in kwargs.items():
+			options[k] = v
+		assert conf.host == options['host']
+		assert conf.port == options['port']
+		assert conf.ssh1 is options['ssh1']
+		assert conf.ssh2 is options['ssh2']
+		assert conf.batch is options['batch']
+		assert conf.colors is options['colors']
+		assert conf.verbose is options['verbose']
+		assert conf.minlevel == options['minlevel']
+		assert conf.ipv4 == options['ipv4']
+		assert conf.ipv6 == options['ipv6']
+		assert conf.ipvo == options['ipvo']
+	
+	def test_audit_conf_defaults(self):
+		conf = self.AuditConf()
+		self._test_conf(conf)
+	
+	def test_audit_conf_booleans(self):
+		conf = self.AuditConf()
+		for p in ['ssh1', 'ssh2', 'batch', 'colors', 'verbose']:
+			for v in [True, 1]:
+				setattr(conf, p, v)
+				assert getattr(conf, p) is True
+			for v in [False, 0]:
+				setattr(conf, p, v)
+				assert getattr(conf, p) is False
+	
+	def test_audit_conf_port(self):
+		conf = self.AuditConf()
+		for port in [22, 2222]:
+			conf.port = port
+			assert conf.port == port
+		for port in [-1, 0, 65536, 99999]:
+			with pytest.raises(ValueError) as excinfo:
+				conf.port = port
+			excinfo.match(r'.*invalid port.*')
+	
+	def test_audit_conf_ipvo(self):
+		# ipv4-only
+		conf = self.AuditConf()
+		conf.ipv4 = True
+		assert conf.ipv4 is True
+		assert conf.ipv6 is False
+		assert conf.ipvo == (4,)
+		# ipv6-only
+		conf = self.AuditConf()
+		conf.ipv6 = True
+		assert conf.ipv4 is False
+		assert conf.ipv6 is True
+		assert conf.ipvo == (6,)
+		# ipv4-only (by removing ipv6)
+		conf = self.AuditConf()
+		conf.ipv6 = False
+		assert conf.ipv4 is True
+		assert conf.ipv6 is False
+		assert conf.ipvo == (4, )
+		# ipv6-only (by removing ipv4)
+		conf = self.AuditConf()
+		conf.ipv4 = False
+		assert conf.ipv4 is False
+		assert conf.ipv6 is True
+		assert conf.ipvo == (6, )
+		# ipv4-preferred
+		conf = self.AuditConf()
+		conf.ipv4 = True
+		conf.ipv6 = True
+		assert conf.ipv4 is True
+		assert conf.ipv6 is True
+		assert conf.ipvo == (4, 6)
+		# ipv6-preferred
+		conf = self.AuditConf()
+		conf.ipv6 = True
+		conf.ipv4 = True
+		assert conf.ipv4 is True
+		assert conf.ipv6 is True
+		assert conf.ipvo == (6, 4)
+		# ipvo empty
+		conf = self.AuditConf()
+		conf.ipvo = ()
+		assert conf.ipv4 is True
+		assert conf.ipv6 is True
+		assert conf.ipvo == ()
+		# ipvo validation
+		conf = self.AuditConf()
+		conf.ipvo = (1, 2, 3, 4, 5, 6)
+		assert conf.ipvo == (4, 6)
+		conf.ipvo = (4, 4, 4, 6, 6)
+		assert conf.ipvo == (4, 6)
+	
+	def test_audit_conf_minlevel(self):
+		conf = self.AuditConf()
+		for level in ['info', 'warn', 'fail']:
+			conf.minlevel = level
+			assert conf.minlevel == level
+		for level in ['head', 'good', 'unknown', None]:
+			with pytest.raises(ValueError) as excinfo:
+				conf.minlevel = level
+			excinfo.match(r'.*invalid level.*')
+	
+	def test_audit_conf_cmdline(self):
+		# pylint: disable=too-many-statements
+		c = lambda x: self.AuditConf.from_cmdline(x.split(), self.usage)  # noqa
+		with pytest.raises(SystemExit):
+			conf = c('')
+		with pytest.raises(SystemExit):
+			conf = c('-x')
+		with pytest.raises(SystemExit):
+			conf = c('-h')
+		with pytest.raises(SystemExit):
+			conf = c('--help')
+		with pytest.raises(SystemExit):
+			conf = c(':')
+		with pytest.raises(SystemExit):
+			conf = c(':22')
+		conf = c('localhost')
+		self._test_conf(conf, host='localhost')
+		conf = c('github.com')
+		self._test_conf(conf, host='github.com')
+		conf = c('localhost:2222')
+		self._test_conf(conf, host='localhost', port=2222)
+		conf = c('-p 2222 localhost')
+		self._test_conf(conf, host='localhost', port=2222)
+		with pytest.raises(SystemExit):
+			conf = c('localhost:')
+		with pytest.raises(SystemExit):
+			conf = c('localhost:abc')
+		with pytest.raises(SystemExit):
+			conf = c('-p abc localhost')
+		with pytest.raises(SystemExit):
+			conf = c('localhost:-22')
+		with pytest.raises(SystemExit):
+			conf = c('-p -22 localhost')
+		with pytest.raises(SystemExit):
+			conf = c('localhost:99999')
+		with pytest.raises(SystemExit):
+			conf = c('-p 99999 localhost')
+		conf = c('-1 localhost')
+		self._test_conf(conf, host='localhost', ssh1=True, ssh2=False)
+		conf = c('-2 localhost')
+		self._test_conf(conf, host='localhost', ssh1=False, ssh2=True)
+		conf = c('-12 localhost')
+		self._test_conf(conf, host='localhost', ssh1=True, ssh2=True)
+		conf = c('-4 localhost')
+		self._test_conf(conf, host='localhost', ipv4=True, ipv6=False, ipvo=(4,))
+		conf = c('-6 localhost')
+		self._test_conf(conf, host='localhost', ipv4=False, ipv6=True, ipvo=(6,))
+		conf = c('-46 localhost')
+		self._test_conf(conf, host='localhost', ipv4=True, ipv6=True, ipvo=(4, 6))
+		conf = c('-64 localhost')
+		self._test_conf(conf, host='localhost', ipv4=True, ipv6=True, ipvo=(6, 4))
+		conf = c('-b localhost')
+		self._test_conf(conf, host='localhost', batch=True, verbose=True)
+		conf = c('-n localhost')
+		self._test_conf(conf, host='localhost', colors=False)
+		conf = c('-v localhost')
+		self._test_conf(conf, host='localhost', verbose=True)
+		conf = c('-l info localhost')
+		self._test_conf(conf, host='localhost', minlevel='info')
+		conf = c('-l warn localhost')
+		self._test_conf(conf, host='localhost', minlevel='warn')
+		conf = c('-l fail localhost')
+		self._test_conf(conf, host='localhost', minlevel='fail')
+		with pytest.raises(SystemExit):
+			conf = c('-l something localhost')

test/test_banner.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import pytest
+
+
+# pylint: disable=line-too-long,attribute-defined-outside-init
+class TestBanner(object):
+	@pytest.fixture(autouse=True)
+	def init(self, ssh_audit):
+		self.ssh = ssh_audit.SSH
+	
+	def test_simple_banners(self):
+		banner = lambda x: self.ssh.Banner.parse(x)  # noqa
+		b = banner('SSH-2.0-OpenSSH_7.3')
+		assert b.protocol == (2, 0)
+		assert b.software == 'OpenSSH_7.3'
+		assert b.comments is None
+		assert str(b) == 'SSH-2.0-OpenSSH_7.3'
+		b = banner('SSH-1.99-Sun_SSH_1.1.3')
+		assert b.protocol == (1, 99)
+		assert b.software == 'Sun_SSH_1.1.3'
+		assert b.comments is None
+		assert str(b) == 'SSH-1.99-Sun_SSH_1.1.3'
+		b = banner('SSH-1.5-Cisco-1.25')
+		assert b.protocol == (1, 5)
+		assert b.software == 'Cisco-1.25'
+		assert b.comments is None
+		assert str(b) == 'SSH-1.5-Cisco-1.25'
+	
+	def test_invalid_banners(self):
+		b = lambda x: self.ssh.Banner.parse(x)  # noqa
+		assert b('Something') is None
+		assert b('SSH-XXX-OpenSSH_7.3') is None
+	
+	def test_banners_with_spaces(self):
+		b = lambda x: self.ssh.Banner.parse(x)  # noqa
+		s = 'SSH-2.0-OpenSSH_4.3p2'
+		assert str(b('SSH-2.0-OpenSSH_4.3p2    ')) == s
+		assert str(b('SSH-2.0-    OpenSSH_4.3p2')) == s
+		assert str(b('SSH-2.0-  OpenSSH_4.3p2  ')) == s
+		s = 'SSH-2.0-OpenSSH_4.3p2 Debian-9etch3 on i686-pc-linux-gnu'
+		assert str(b('SSH-2.0-  OpenSSH_4.3p2 Debian-9etch3   on i686-pc-linux-gnu')) == s
+		assert str(b('SSH-2.0-OpenSSH_4.3p2 Debian-9etch3 on i686-pc-linux-gnu  ')) == s
+		assert str(b('SSH-2.0-  OpenSSH_4.3p2 Debian-9etch3   on   i686-pc-linux-gnu  ')) == s
+	
+	def test_banners_without_software(self):
+		b = lambda x: self.ssh.Banner.parse(x)  # noqa
+		assert b('SSH-2.0').protocol == (2, 0)
+		assert b('SSH-2.0').software is None
+		assert b('SSH-2.0').comments is None
+		assert str(b('SSH-2.0')) == 'SSH-2.0'
+		assert b('SSH-2.0-').protocol == (2, 0)
+		assert b('SSH-2.0-').software == ''
+		assert b('SSH-2.0-').comments is None
+		assert str(b('SSH-2.0-')) == 'SSH-2.0-'
+	
+	def test_banners_with_comments(self):
+		b = lambda x: self.ssh.Banner.parse(x)  # noqa
+		assert repr(b('SSH-2.0-OpenSSH_7.2p2 Ubuntu-1')) == '<Banner(protocol=2.0, software=OpenSSH_7.2p2, comments=Ubuntu-1)>'
+		assert repr(b('SSH-1.99-OpenSSH_3.4p1 Debian 1:3.4p1-1.woody.3')) == '<Banner(protocol=1.99, software=OpenSSH_3.4p1, comments=Debian 1:3.4p1-1.woody.3)>'
+		assert repr(b('SSH-1.5-1.3.7 F-SECURE SSH')) == '<Banner(protocol=1.5, software=1.3.7, comments=F-SECURE SSH)>'
+	
+	def test_banners_with_multiple_protocols(self):
+		b = lambda x: self.ssh.Banner.parse(x)  # noqa
+		assert str(b('SSH-1.99-SSH-1.99-OpenSSH_3.6.1p2')) == 'SSH-1.99-OpenSSH_3.6.1p2'
+		assert str(b('SSH-2.0-SSH-2.0-OpenSSH_4.3p2 Debian-9')) == 'SSH-2.0-OpenSSH_4.3p2 Debian-9'
+		assert str(b('SSH-1.99-SSH-2.0-dropbear_0.5')) == 'SSH-1.99-dropbear_0.5'
+		assert str(b('SSH-2.0-SSH-1.99-OpenSSH_4.2p1 SSH Secure Shell (non-commercial)')) == 'SSH-1.99-OpenSSH_4.2p1 SSH Secure Shell (non-commercial)'
+		assert str(b('SSH-1.99-SSH-1.99-SSH-1.99-OpenSSH_3.9p1')) == 'SSH-1.99-OpenSSH_3.9p1'

test/test_buffer.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import re
+import pytest
+
+
+# pylint: disable=attribute-defined-outside-init,bad-whitespace
+class TestBuffer(object):
+	@pytest.fixture(autouse=True)
+	def init(self, ssh_audit):
+		self.rbuf = ssh_audit.ReadBuf
+		self.wbuf = ssh_audit.WriteBuf
+		self.utf8rchar = b'\xef\xbf\xbd'
+	
+	@classmethod
+	def _b(cls, v):
+		v = re.sub(r'\s', '', v)
+		data = [int(v[i * 2:i * 2 + 2], 16) for i in range(len(v) // 2)]
+		return bytes(bytearray(data))
+	
+	def test_unread(self):
+		w = self.wbuf().write_byte(1).write_int(2).write_flush()
+		r = self.rbuf(w)
+		assert r.unread_len == 5
+		r.read_byte()
+		assert r.unread_len == 4
+		r.read_int()
+		assert r.unread_len == 0
+	
+	def test_byte(self):
+		w = lambda x: self.wbuf().write_byte(x).write_flush()  # noqa
+		r = lambda x: self.rbuf(x).read_byte()  # noqa
+		tc = [(0x00, '00'),
+		      (0x01, '01'),
+		      (0x10, '10'),
+		      (0xff, 'ff')]
+		for p in tc:
+			assert w(p[0]) == self._b(p[1])
+			assert r(self._b(p[1])) == p[0]
+	
+	def test_bool(self):
+		w = lambda x: self.wbuf().write_bool(x).write_flush()  # noqa
+		r = lambda x: self.rbuf(x).read_bool()  # noqa
+		tc = [(True,  '01'),
+		      (False, '00')]
+		for p in tc:
+			assert w(p[0]) == self._b(p[1])
+			assert r(self._b(p[1])) == p[0]
+	
+	def test_int(self):
+		w = lambda x: self.wbuf().write_int(x).write_flush()  # noqa
+		r = lambda x: self.rbuf(x).read_int()  # noqa
+		tc = [(0x00,       '00 00 00 00'),
+		      (0x01,       '00 00 00 01'),
+		      (0xabcd,     '00 00 ab cd'),
+		      (0xffffffff, 'ff ff ff ff')]
+		for p in tc:
+			assert w(p[0]) == self._b(p[1])
+			assert r(self._b(p[1])) == p[0]
+	
+	def test_string(self):
+		w = lambda x: self.wbuf().write_string(x).write_flush()  # noqa
+		r = lambda x: self.rbuf(x).read_string()  # noqa
+		tc = [(u'abc1',  '00 00 00 04 61 62 63 31'),
+		      (b'abc2',  '00 00 00 04 61 62 63 32')]
+		for p in tc:
+			v = p[0]
+			assert w(v) == self._b(p[1])
+			if not isinstance(v, bytes):
+				v = bytes(bytearray(v, 'utf-8'))
+			assert r(self._b(p[1])) == v
+	
+	def test_list(self):
+		w = lambda x: self.wbuf().write_list(x).write_flush()  # noqa
+		r = lambda x: self.rbuf(x).read_list()  # noqa
+		tc = [(['d', 'ef', 'ault'], '00 00 00 09 64 2c 65 66 2c 61 75 6c 74')]
+		for p in tc:
+			assert w(p[0]) == self._b(p[1])
+			assert r(self._b(p[1])) == p[0]
+	
+	def test_list_nonutf8(self):
+		r = lambda x: self.rbuf(x).read_list()  # noqa
+		src = self._b('00 00 00 04 de ad be ef')
+		dst = [(b'\xde\xad' + self.utf8rchar + self.utf8rchar).decode('utf-8')]
+		assert r(src) == dst
+	
+	def test_line(self):
+		w = lambda x: self.wbuf().write_line(x).write_flush()  # noqa
+		r = lambda x: self.rbuf(x).read_line()  # noqa
+		tc = [(u'example line', '65 78 61 6d 70 6c 65 20 6c 69 6e 65 0d 0a')]
+		for p in tc:
+			assert w(p[0]) == self._b(p[1])
+			assert r(self._b(p[1])) == p[0]
+	
+	def test_line_nonutf8(self):
+		r = lambda x: self.rbuf(x).read_line()  # noqa
+		src = self._b('de ad be af')
+		dst = (b'\xde\xad' + self.utf8rchar + self.utf8rchar).decode('utf-8')
+		assert r(src) == dst
+	
+	def test_bitlen(self):
+		# pylint: disable=protected-access
+		class Py26Int(int):
+			def bit_length(self):
+				raise AttributeError
+		assert self.wbuf._bitlength(42) == 6
+		assert self.wbuf._bitlength(Py26Int(42)) == 6
+	
+	def test_mpint1(self):
+		mpint1w = lambda x: self.wbuf().write_mpint1(x).write_flush()  # noqa
+		mpint1r = lambda x: self.rbuf(x).read_mpint1()  # noqa
+		tc = [(0x0,     '00 00'),
+		      (0x1234,  '00 0d 12 34'),
+		      (0x12345, '00 11 01 23 45'),
+		      (0xdeadbeef, '00 20 de ad be ef')]
+		for p in tc:
+			assert mpint1w(p[0]) == self._b(p[1])
+			assert mpint1r(self._b(p[1])) == p[0]
+	
+	def test_mpint2(self):
+		mpint2w = lambda x: self.wbuf().write_mpint2(x).write_flush()  # noqa
+		mpint2r = lambda x: self.rbuf(x).read_mpint2()  # noqa
+		tc = [(0x0,               '00 00 00 00'),
+		      (0x80,              '00 00 00 02 00 80'),
+		      (0x9a378f9b2e332a7, '00 00 00 08 09 a3 78 f9 b2 e3 32 a7'),
+		      (-0x1234,           '00 00 00 02 ed cc'),
+		      (-0xdeadbeef,       '00 00 00 05 ff 21 52 41 11'),
+		      (-0x8000,           '00 00 00 02 80 00'),
+		      (-0x80,             '00 00 00 01 80')]
+		for p in tc:
+			assert mpint2w(p[0]) == self._b(p[1])
+			assert mpint2r(self._b(p[1])) == p[0]
+		assert mpint2r(self._b('00 00 00 02 ff 80')) == -0x80

test/test_errors.py

@@ -0,0 +1,123 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import socket
+import pytest
+
+
+# pylint: disable=attribute-defined-outside-init
+class TestErrors(object):
+	@pytest.fixture(autouse=True)
+	def init(self, ssh_audit):
+		self.AuditConf = ssh_audit.AuditConf
+		self.audit = ssh_audit.audit
+	
+	def _conf(self):
+		conf = self.AuditConf('localhost', 22)
+		conf.colors = False
+		conf.batch = True
+		return conf
+	
+	def test_connection_refused(self, output_spy, virtual_socket):
+		vsocket = virtual_socket
+		vsocket.errors['connect'] = socket.error(61, 'Connection refused')
+		output_spy.begin()
+		with pytest.raises(SystemExit):
+			self.audit(self._conf())
+		lines = output_spy.flush()
+		assert len(lines) == 1
+		assert 'Connection refused' in lines[-1]
+	
+	def test_connection_closed_before_banner(self, output_spy, virtual_socket):
+		vsocket = virtual_socket
+		vsocket.rdata.append(socket.error(54, 'Connection reset by peer'))
+		output_spy.begin()
+		with pytest.raises(SystemExit):
+			self.audit(self._conf())
+		lines = output_spy.flush()
+		assert len(lines) == 1
+		assert 'did not receive banner' in lines[-1]
+	
+	def test_connection_closed_after_header(self, output_spy, virtual_socket):
+		vsocket = virtual_socket
+		vsocket.rdata.append(b'header line 1\n')
+		vsocket.rdata.append(b'header line 2\n')
+		vsocket.rdata.append(socket.error(54, 'Connection reset by peer'))
+		output_spy.begin()
+		with pytest.raises(SystemExit):
+			self.audit(self._conf())
+		lines = output_spy.flush()
+		assert len(lines) == 3
+		assert 'did not receive banner' in lines[-1]
+	
+	def test_connection_closed_after_banner(self, output_spy, virtual_socket):
+		vsocket = virtual_socket
+		vsocket.rdata.append(b'SSH-2.0-ssh-audit-test\r\n')
+		vsocket.rdata.append(socket.error(54, 'Connection reset by peer'))
+		output_spy.begin()
+		with pytest.raises(SystemExit):
+			self.audit(self._conf())
+		lines = output_spy.flush()
+		assert len(lines) == 2
+		assert 'error reading packet' in lines[-1]
+		assert 'reset by peer' in lines[-1]
+	
+	def test_empty_data_after_banner(self, output_spy, virtual_socket):
+		vsocket = virtual_socket
+		vsocket.rdata.append(b'SSH-2.0-ssh-audit-test\r\n')
+		output_spy.begin()
+		with pytest.raises(SystemExit):
+			self.audit(self._conf())
+		lines = output_spy.flush()
+		assert len(lines) == 2
+		assert 'error reading packet' in lines[-1]
+		assert 'empty' in lines[-1]
+	
+	def test_wrong_data_after_banner(self, output_spy, virtual_socket):
+		vsocket = virtual_socket
+		vsocket.rdata.append(b'SSH-2.0-ssh-audit-test\r\n')
+		vsocket.rdata.append(b'xxx\n')
+		output_spy.begin()
+		with pytest.raises(SystemExit):
+			self.audit(self._conf())
+		lines = output_spy.flush()
+		assert len(lines) == 2
+		assert 'error reading packet' in lines[-1]
+		assert 'xxx' in lines[-1]
+	
+	def test_non_ascii_banner(self, output_spy, virtual_socket):
+		vsocket = virtual_socket
+		vsocket.rdata.append(b'SSH-2.0-ssh-audit-test\xc3\xbc\r\n')
+		output_spy.begin()
+		with pytest.raises(SystemExit):
+			self.audit(self._conf())
+		lines = output_spy.flush()
+		assert len(lines) == 3
+		assert 'error reading packet' in lines[-1]
+		assert 'ASCII' in lines[-2]
+		assert lines[-3].endswith('SSH-2.0-ssh-audit-test?')
+	
+	def test_nonutf8_data_after_banner(self, output_spy, virtual_socket):
+		vsocket = virtual_socket
+		vsocket.rdata.append(b'SSH-2.0-ssh-audit-test\r\n')
+		vsocket.rdata.append(b'\x81\xff\n')
+		output_spy.begin()
+		with pytest.raises(SystemExit):
+			self.audit(self._conf())
+		lines = output_spy.flush()
+		assert len(lines) == 2
+		assert 'error reading packet' in lines[-1]
+		assert '\\x81\\xff' in lines[-1]
+	
+	def test_protocol_mismatch_by_conf(self, output_spy, virtual_socket):
+		vsocket = virtual_socket
+		vsocket.rdata.append(b'SSH-1.3-ssh-audit-test\r\n')
+		vsocket.rdata.append(b'Protocol major versions differ.\n')
+		output_spy.begin()
+		with pytest.raises(SystemExit):
+			conf = self._conf()
+			conf.ssh1, conf.ssh2 = True, False
+			self.audit(conf)
+		lines = output_spy.flush()
+		assert len(lines) == 3
+		assert 'error reading packet' in lines[-1]
+		assert 'major versions differ' in lines[-1]

test/test_output.py

@@ -0,0 +1,175 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+from __future__ import print_function
+import pytest
+
+
+# pylint: disable=attribute-defined-outside-init
+class TestOutput(object):
+	@pytest.fixture(autouse=True)
+	def init(self, ssh_audit):
+		self.Output = ssh_audit.Output
+		self.OutputBuffer = ssh_audit.OutputBuffer
+	
+	def test_output_buffer_no_lines(self, output_spy):
+		output_spy.begin()
+		with self.OutputBuffer() as obuf:
+			pass
+		assert output_spy.flush() == []
+		output_spy.begin()
+		with self.OutputBuffer() as obuf:
+			pass
+		obuf.flush()
+		assert output_spy.flush() == []
+	
+	def test_output_buffer_no_flush(self, output_spy):
+		output_spy.begin()
+		with self.OutputBuffer():
+			print(u'abc')
+		assert output_spy.flush() == []
+	
+	def test_output_buffer_flush(self, output_spy):
+		output_spy.begin()
+		with self.OutputBuffer() as obuf:
+			print(u'abc')
+			print()
+			print(u'def')
+		obuf.flush()
+		assert output_spy.flush() == [u'abc', u'', u'def']
+	
+	def test_output_defaults(self):
+		out = self.Output()
+		# default: on
+		assert out.batch is False
+		assert out.colors is True
+		assert out.minlevel == 'info'
+	
+	def test_output_colors(self, output_spy):
+		out = self.Output()
+		# test without colors
+		out.colors = False
+		output_spy.begin()
+		out.info('info color')
+		assert output_spy.flush() == [u'info color']
+		output_spy.begin()
+		out.head('head color')
+		assert output_spy.flush() == [u'head color']
+		output_spy.begin()
+		out.good('good color')
+		assert output_spy.flush() == [u'good color']
+		output_spy.begin()
+		out.warn('warn color')
+		assert output_spy.flush() == [u'warn color']
+		output_spy.begin()
+		out.fail('fail color')
+		assert output_spy.flush() == [u'fail color']
+		if not out.colors_supported:
+			return
+		# test with colors
+		out.colors = True
+		output_spy.begin()
+		out.info('info color')
+		assert output_spy.flush() == [u'info color']
+		output_spy.begin()
+		out.head('head color')
+		assert output_spy.flush() == [u'\x1b[0;36mhead color\x1b[0m']
+		output_spy.begin()
+		out.good('good color')
+		assert output_spy.flush() == [u'\x1b[0;32mgood color\x1b[0m']
+		output_spy.begin()
+		out.warn('warn color')
+		assert output_spy.flush() == [u'\x1b[0;33mwarn color\x1b[0m']
+		output_spy.begin()
+		out.fail('fail color')
+		assert output_spy.flush() == [u'\x1b[0;31mfail color\x1b[0m']
+	
+	def test_output_sep(self, output_spy):
+		out = self.Output()
+		output_spy.begin()
+		out.sep()
+		out.sep()
+		out.sep()
+		assert output_spy.flush() == [u'', u'', u'']
+	
+	def test_output_levels(self):
+		out = self.Output()
+		assert out.getlevel('info') == 0
+		assert out.getlevel('good') == 0
+		assert out.getlevel('warn') == 1
+		assert out.getlevel('fail') == 2
+		assert out.getlevel('unknown') > 2
+	
+	def test_output_minlevel_property(self):
+		out = self.Output()
+		out.minlevel = 'info'
+		assert out.minlevel == 'info'
+		out.minlevel = 'good'
+		assert out.minlevel == 'info'
+		out.minlevel = 'warn'
+		assert out.minlevel == 'warn'
+		out.minlevel = 'fail'
+		assert out.minlevel == 'fail'
+		out.minlevel = 'invalid level'
+		assert out.minlevel == 'unknown'
+	
+	def test_output_minlevel(self, output_spy):
+		out = self.Output()
+		# visible: all
+		out.minlevel = 'info'
+		output_spy.begin()
+		out.info('info color')
+		out.head('head color')
+		out.good('good color')
+		out.warn('warn color')
+		out.fail('fail color')
+		assert len(output_spy.flush()) == 5
+		# visible: head, warn, fail
+		out.minlevel = 'warn'
+		output_spy.begin()
+		out.info('info color')
+		out.head('head color')
+		out.good('good color')
+		out.warn('warn color')
+		out.fail('fail color')
+		assert len(output_spy.flush()) == 3
+		# visible: head, fail
+		out.minlevel = 'fail'
+		output_spy.begin()
+		out.info('info color')
+		out.head('head color')
+		out.good('good color')
+		out.warn('warn color')
+		out.fail('fail color')
+		assert len(output_spy.flush()) == 2
+		# visible: head
+		out.minlevel = 'invalid level'
+		output_spy.begin()
+		out.info('info color')
+		out.head('head color')
+		out.good('good color')
+		out.warn('warn color')
+		out.fail('fail color')
+		assert len(output_spy.flush()) == 1
+	
+	def test_output_batch(self, output_spy):
+		out = self.Output()
+		# visible: all
+		output_spy.begin()
+		out.minlevel = 'info'
+		out.batch = False
+		out.info('info color')
+		out.head('head color')
+		out.good('good color')
+		out.warn('warn color')
+		out.fail('fail color')
+		assert len(output_spy.flush()) == 5
+		# visible: all except head
+		output_spy.begin()
+		out.minlevel = 'info'
+		out.batch = True
+		out.info('info color')
+		out.head('head color')
+		out.good('good color')
+		out.warn('warn color')
+		out.fail('fail color')
+		assert len(output_spy.flush()) == 4

test/test_protocol.py

@@ -1,42 +0,0 @@
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-import pytest
-import re
-
-
-class TestProtocol(object):
-	@pytest.fixture(autouse=True)
-	def init(self, ssh_audit):
-		self.rbuf = ssh_audit.ReadBuf
-		self.wbuf = ssh_audit.WriteBuf
-	
-	def _b(self, v):
-		v = re.sub(r'\s', '', v)
-		data = [int(v[i * 2:i * 2 + 2], 16) for i in range(len(v) // 2)]
-		return bytes(bytearray(data))
-	
-	def test_mpint1(self):
-		mpint1w = lambda x: self.wbuf().write_mpint1(x).write_flush()
-		mpint1r = lambda x: self.rbuf(x).read_mpint1()
-		tc = [(0x0,     '00 00'),
-		      (0x1234,  '00 0d 12 34'),
-		      (0x12345, '00 11 01 23 45'),
-		      (0xdeadbeef, '00 20 de ad be ef')]
-		for p in tc:
-			assert mpint1w(p[0]) == self._b(p[1])
-			assert mpint1r(self._b(p[1])) == p[0]
-	
-	def test_mpint2(self):
-		mpint2w = lambda x: self.wbuf().write_mpint2(x).write_flush()
-		mpint2r = lambda x: self.rbuf(x).read_mpint2()
-		tc = [(0x0,               '00 00 00 00'),
-		      (0x80,              '00 00 00 02 00 80'),
-		      (0x9a378f9b2e332a7, '00 00 00 08 09 a3 78 f9 b2 e3 32 a7'),
-		      (-0x1234,           '00 00 00 02 ed cc'),
-		      (-0xdeadbeef,       '00 00 00 05 ff 21 52 41 11'),
-		      (-0x8000,           '00 00 00 02 80 00'),
-		      (-0x80,             '00 00 00 01 80')]
-		for p in tc:
-			assert mpint2w(p[0]) == self._b(p[1])
-			assert mpint2r(self._b(p[1])) == p[0]
-		assert mpint2r(self._b('00 00 00 02 ff 80')) == -0x80

test/test_software.py

@@ -0,0 +1,287 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import pytest
+
+
+# pylint: disable=line-too-long,attribute-defined-outside-init
+class TestSoftware(object):
+	@pytest.fixture(autouse=True)
+	def init(self, ssh_audit):
+		self.ssh = ssh_audit.SSH
+	
+	def test_unknown_software(self):
+		ps = lambda x: self.ssh.Software.parse(self.ssh.Banner.parse(x))  # noqa
+		assert ps('SSH-1.5') is None
+		assert ps('SSH-1.99-AlfaMegaServer') is None
+		assert ps('SSH-2.0-BetaMegaServer 0.0.1') is None
+	
+	def test_openssh_software(self):
+		# pylint: disable=too-many-statements
+		ps = lambda x: self.ssh.Software.parse(self.ssh.Banner.parse(x))  # noqa
+		# common
+		s = ps('SSH-2.0-OpenSSH_7.3')
+		assert s.vendor is None
+		assert s.product == 'OpenSSH'
+		assert s.version == '7.3'
+		assert s.patch is None
+		assert s.os is None
+		assert str(s) == 'OpenSSH 7.3'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == str(s)
+		assert repr(s) == '<Software(product=OpenSSH, version=7.3)>'
+		# common, portable
+		s = ps('SSH-2.0-OpenSSH_7.2p1')
+		assert s.vendor is None
+		assert s.product == 'OpenSSH'
+		assert s.version == '7.2'
+		assert s.patch == 'p1'
+		assert s.os is None
+		assert str(s) == 'OpenSSH 7.2p1'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == 'OpenSSH 7.2'
+		assert repr(s) == '<Software(product=OpenSSH, version=7.2, patch=p1)>'
+		# dot instead of underline
+		s = ps('SSH-2.0-OpenSSH.6.6')
+		assert s.vendor is None
+		assert s.product == 'OpenSSH'
+		assert s.version == '6.6'
+		assert s.patch is None
+		assert s.os is None
+		assert str(s) == 'OpenSSH 6.6'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == str(s)
+		assert repr(s) == '<Software(product=OpenSSH, version=6.6)>'
+		# dash instead of underline
+		s = ps('SSH-2.0-OpenSSH-3.9p1')
+		assert s.vendor is None
+		assert s.product == 'OpenSSH'
+		assert s.version == '3.9'
+		assert s.patch == 'p1'
+		assert s.os is None
+		assert str(s) == 'OpenSSH 3.9p1'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == 'OpenSSH 3.9'
+		assert repr(s) == '<Software(product=OpenSSH, version=3.9, patch=p1)>'
+		# patch prefix with dash
+		s = ps('SSH-2.0-OpenSSH_7.2-hpn14v5')
+		assert s.vendor is None
+		assert s.product == 'OpenSSH'
+		assert s.version == '7.2'
+		assert s.patch == 'hpn14v5'
+		assert s.os is None
+		assert str(s) == 'OpenSSH 7.2 (hpn14v5)'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == 'OpenSSH 7.2'
+		assert repr(s) == '<Software(product=OpenSSH, version=7.2, patch=hpn14v5)>'
+		# patch prefix with underline
+		s = ps('SSH-1.5-OpenSSH_6.6.1_hpn13v11')
+		assert s.vendor is None
+		assert s.product == 'OpenSSH'
+		assert s.version == '6.6.1'
+		assert s.patch == 'hpn13v11'
+		assert s.os is None
+		assert str(s) == 'OpenSSH 6.6.1 (hpn13v11)'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == 'OpenSSH 6.6.1'
+		assert repr(s) == '<Software(product=OpenSSH, version=6.6.1, patch=hpn13v11)>'
+		# patch prefix with dot
+		s = ps('SSH-2.0-OpenSSH_5.9.CASPUR')
+		assert s.vendor is None
+		assert s.product == 'OpenSSH'
+		assert s.version == '5.9'
+		assert s.patch == 'CASPUR'
+		assert s.os is None
+		assert str(s) == 'OpenSSH 5.9 (CASPUR)'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == 'OpenSSH 5.9'
+		assert repr(s) == '<Software(product=OpenSSH, version=5.9, patch=CASPUR)>'
+	
+	def test_dropbear_software(self):
+		ps = lambda x: self.ssh.Software.parse(self.ssh.Banner.parse(x))  # noqa
+		# common
+		s = ps('SSH-2.0-dropbear_2016.74')
+		assert s.vendor is None
+		assert s.product == 'Dropbear SSH'
+		assert s.version == '2016.74'
+		assert s.patch is None
+		assert s.os is None
+		assert str(s) == 'Dropbear SSH 2016.74'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == str(s)
+		assert repr(s) == '<Software(product=Dropbear SSH, version=2016.74)>'
+		# common, patch
+		s = ps('SSH-2.0-dropbear_0.44test4')
+		assert s.vendor is None
+		assert s.product == 'Dropbear SSH'
+		assert s.version == '0.44'
+		assert s.patch == 'test4'
+		assert s.os is None
+		assert str(s) == 'Dropbear SSH 0.44 (test4)'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == 'Dropbear SSH 0.44'
+		assert repr(s) == '<Software(product=Dropbear SSH, version=0.44, patch=test4)>'
+		# patch prefix with dash
+		s = ps('SSH-2.0-dropbear_0.44-Freesco-p49')
+		assert s.vendor is None
+		assert s.product == 'Dropbear SSH'
+		assert s.version == '0.44'
+		assert s.patch == 'Freesco-p49'
+		assert s.os is None
+		assert str(s) == 'Dropbear SSH 0.44 (Freesco-p49)'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == 'Dropbear SSH 0.44'
+		assert repr(s) == '<Software(product=Dropbear SSH, version=0.44, patch=Freesco-p49)>'
+		# patch prefix with underline
+		s = ps('SSH-2.0-dropbear_2014.66_agbn_1')
+		assert s.vendor is None
+		assert s.product == 'Dropbear SSH'
+		assert s.version == '2014.66'
+		assert s.patch == 'agbn_1'
+		assert s.os is None
+		assert str(s) == 'Dropbear SSH 2014.66 (agbn_1)'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == 'Dropbear SSH 2014.66'
+		assert repr(s) == '<Software(product=Dropbear SSH, version=2014.66, patch=agbn_1)>'
+	
+	def test_libssh_software(self):
+		ps = lambda x: self.ssh.Software.parse(self.ssh.Banner.parse(x))  # noqa
+		# common
+		s = ps('SSH-2.0-libssh-0.2')
+		assert s.vendor is None
+		assert s.product == 'libssh'
+		assert s.version == '0.2'
+		assert s.patch is None
+		assert s.os is None
+		assert str(s) == 'libssh 0.2'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == str(s)
+		assert repr(s) == '<Software(product=libssh, version=0.2)>'
+		s = ps('SSH-2.0-libssh-0.7.3')
+		assert s.vendor is None
+		assert s.product == 'libssh'
+		assert s.version == '0.7.3'
+		assert s.patch is None
+		assert s.os is None
+		assert str(s) == 'libssh 0.7.3'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == str(s)
+		assert repr(s) == '<Software(product=libssh, version=0.7.3)>'
+	
+	def test_romsshell_software(self):
+		ps = lambda x: self.ssh.Software.parse(self.ssh.Banner.parse(x))  # noqa
+		# common
+		s = ps('SSH-2.0-RomSShell_5.40')
+		assert s.vendor == 'Allegro Software'
+		assert s.product == 'RomSShell'
+		assert s.version == '5.40'
+		assert s.patch is None
+		assert s.os is None
+		assert str(s) == 'Allegro Software RomSShell 5.40'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == str(s)
+		assert repr(s) == '<Software(vendor=Allegro Software, product=RomSShell, version=5.40)>'
+	
+	def test_hp_ilo_software(self):
+		ps = lambda x: self.ssh.Software.parse(self.ssh.Banner.parse(x))  # noqa
+		# common
+		s = ps('SSH-2.0-mpSSH_0.2.1')
+		assert s.vendor == 'HP'
+		assert s.product == 'iLO (Integrated Lights-Out) sshd'
+		assert s.version == '0.2.1'
+		assert s.patch is None
+		assert s.os is None
+		assert str(s) == 'HP iLO (Integrated Lights-Out) sshd 0.2.1'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == str(s)
+		assert repr(s) == '<Software(vendor=HP, product=iLO (Integrated Lights-Out) sshd, version=0.2.1)>'
+	
+	def test_cisco_software(self):
+		ps = lambda x: self.ssh.Software.parse(self.ssh.Banner.parse(x))  # noqa
+		# common
+		s = ps('SSH-1.5-Cisco-1.25')
+		assert s.vendor == 'Cisco'
+		assert s.product == 'IOS/PIX sshd'
+		assert s.version == '1.25'
+		assert s.patch is None
+		assert s.os is None
+		assert str(s) == 'Cisco IOS/PIX sshd 1.25'
+		assert str(s) == s.display()
+		assert s.display(True) == str(s)
+		assert s.display(False) == str(s)
+		assert repr(s) == '<Software(vendor=Cisco, product=IOS/PIX sshd, version=1.25)>'
+	
+	def test_software_os(self):
+		ps = lambda x: self.ssh.Software.parse(self.ssh.Banner.parse(x))  # noqa
+		# unknown
+		s = ps('SSH-2.0-OpenSSH_3.7.1 MegaOperatingSystem 123')
+		assert s.os is None
+		# NetBSD
+		s = ps('SSH-1.99-OpenSSH_2.5.1 NetBSD_Secure_Shell-20010614')
+		assert s.os == 'NetBSD (2001-06-14)'
+		assert str(s) == 'OpenSSH 2.5.1 running on NetBSD (2001-06-14)'
+		assert repr(s) == '<Software(product=OpenSSH, version=2.5.1, os=NetBSD (2001-06-14))>'
+		s = ps('SSH-1.99-OpenSSH_5.0 NetBSD_Secure_Shell-20080403+-hpn13v1')
+		assert s.os == 'NetBSD (2008-04-03)'
+		assert str(s) == 'OpenSSH 5.0 running on NetBSD (2008-04-03)'
+		assert repr(s) == '<Software(product=OpenSSH, version=5.0, os=NetBSD (2008-04-03))>'
+		s = ps('SSH-2.0-OpenSSH_6.6.1_hpn13v11 NetBSD-20100308')
+		assert s.os == 'NetBSD (2010-03-08)'
+		assert str(s) == 'OpenSSH 6.6.1 (hpn13v11) running on NetBSD (2010-03-08)'
+		assert repr(s) == '<Software(product=OpenSSH, version=6.6.1, patch=hpn13v11, os=NetBSD (2010-03-08))>'
+		s = ps('SSH-2.0-OpenSSH_4.4 NetBSD')
+		assert s.os == 'NetBSD'
+		assert str(s) == 'OpenSSH 4.4 running on NetBSD'
+		assert repr(s) == '<Software(product=OpenSSH, version=4.4, os=NetBSD)>'
+		s = ps('SSH-2.0-OpenSSH_3.0.2 NetBSD Secure Shell')
+		assert s.os == 'NetBSD'
+		assert str(s) == 'OpenSSH 3.0.2 running on NetBSD'
+		assert repr(s) == '<Software(product=OpenSSH, version=3.0.2, os=NetBSD)>'
+		# FreeBSD
+		s = ps('SSH-2.0-OpenSSH_7.2 FreeBSD-20160310')
+		assert s.os == 'FreeBSD (2016-03-10)'
+		assert str(s) == 'OpenSSH 7.2 running on FreeBSD (2016-03-10)'
+		assert repr(s) == '<Software(product=OpenSSH, version=7.2, os=FreeBSD (2016-03-10))>'
+		s = ps('SSH-1.99-OpenSSH_2.9 FreeBSD localisations 20020307')
+		assert s.os == 'FreeBSD (2002-03-07)'
+		assert str(s) == 'OpenSSH 2.9 running on FreeBSD (2002-03-07)'
+		assert repr(s) == '<Software(product=OpenSSH, version=2.9, os=FreeBSD (2002-03-07))>'
+		s = ps('SSH-2.0-OpenSSH_2.3.0 green@FreeBSD.org 20010321')
+		assert s.os == 'FreeBSD (2001-03-21)'
+		assert str(s) == 'OpenSSH 2.3.0 running on FreeBSD (2001-03-21)'
+		assert repr(s) == '<Software(product=OpenSSH, version=2.3.0, os=FreeBSD (2001-03-21))>'
+		s = ps('SSH-1.99-OpenSSH_4.4p1 FreeBSD-openssh-portable-overwrite-base-4.4.p1_1,1')
+		assert s.os == 'FreeBSD'
+		assert str(s) == 'OpenSSH 4.4p1 running on FreeBSD'
+		assert repr(s) == '<Software(product=OpenSSH, version=4.4, patch=p1, os=FreeBSD)>'
+		s = ps('SSH-2.0-OpenSSH_7.2-OVH-rescue FreeBSD')
+		assert s.os == 'FreeBSD'
+		assert str(s) == 'OpenSSH 7.2 (OVH-rescue) running on FreeBSD'
+		assert repr(s) == '<Software(product=OpenSSH, version=7.2, patch=OVH-rescue, os=FreeBSD)>'
+		# Windows
+		s = ps('SSH-2.0-OpenSSH_3.7.1 in RemotelyAnywhere 5.21.422')
+		assert s.os == 'Microsoft Windows (RemotelyAnywhere 5.21.422)'
+		assert str(s) == 'OpenSSH 3.7.1 running on Microsoft Windows (RemotelyAnywhere 5.21.422)'
+		assert repr(s) == '<Software(product=OpenSSH, version=3.7.1, os=Microsoft Windows (RemotelyAnywhere 5.21.422))>'
+		s = ps('SSH-2.0-OpenSSH_3.8 in DesktopAuthority 7.1.091')
+		assert s.os == 'Microsoft Windows (DesktopAuthority 7.1.091)'
+		assert str(s) == 'OpenSSH 3.8 running on Microsoft Windows (DesktopAuthority 7.1.091)'
+		assert repr(s) == '<Software(product=OpenSSH, version=3.8, os=Microsoft Windows (DesktopAuthority 7.1.091))>'
+		s = ps('SSH-2.0-OpenSSH_3.8 in RemoteSupportManager 1.0.023')
+		assert s.os == 'Microsoft Windows (RemoteSupportManager 1.0.023)'
+		assert str(s) == 'OpenSSH 3.8 running on Microsoft Windows (RemoteSupportManager 1.0.023)'
+		assert repr(s) == '<Software(product=OpenSSH, version=3.8, os=Microsoft Windows (RemoteSupportManager 1.0.023))>'

test/test_ssh1.py

@@ -0,0 +1,139 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import struct
+import pytest
+
+
+# pylint: disable=line-too-long,attribute-defined-outside-init
+class TestSSH1(object):
+	@pytest.fixture(autouse=True)
+	def init(self, ssh_audit):
+		self.ssh = ssh_audit.SSH
+		self.ssh1 = ssh_audit.SSH1
+		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 = True
+		conf.ssh2 = False
+		return conf
+	
+	def _create_ssh1_packet(self, payload, valid_crc=True):
+		padding = -(len(payload) + 4) % 8
+		plen = len(payload) + 4
+		pad_bytes = b'\x00' * padding
+		cksum = self.ssh1.crc32(pad_bytes + payload) if valid_crc else 0
+		data = struct.pack('>I', plen) + pad_bytes + payload + struct.pack('>I', cksum)
+		return data
+	
+	@classmethod
+	def _server_key(cls):
+		return (1024, 0x10001, 0xee6552da432e0ac2c422df1a51287507748bfe3b5e3e4fa989a8f49fdc163a17754939ef18ef8a667ea3b71036a151fcd7f5e01ceef1e4439864baf3ac569047582c69d6c128212e0980dcb3168f00d371004039983f6033cd785b8b8f85096c7d9405cbfdc664e27c966356a6b4eb6ee20ad43414b50de18b22829c1880b551)
+	
+	@classmethod
+	def _host_key(cls):
+		return (2048, 0x10001, 0xdfa20cd2a530ccc8c870aa60d9feb3b35deeab81c3215a96557abbd683d21f4600f38e475d87100da9a4404220eeb3bb5584e5a2b5b48ffda58530ea19104a32577d7459d91e76aa711b241050f4cc6d5327ccce254f371acad3be56d46eb5919b73f20dbdb1177b700f00891c5bf4ed128bb90ed541b778288285bcfa28432ab5cbcb8321b6e24760e998e0daa519f093a631e44276d7dd252ce0c08c75e2ab28a7349ead779f97d0f20a6d413bf3623cd216dc35375f6366690bcc41e3b2d5465840ec7ee0dc7e3f1c101d674a0c7dbccbc3942788b111396add2f8153b46a0e4b50d66e57ee92958f1c860dd97cc0e40e32febff915343ed53573142bdf4b)
+	
+	def _pkm_payload(self):
+		w = self.wbuf()
+		w.write(b'\x88\x99\xaa\xbb\xcc\xdd\xee\xff')
+		b, e, m = self._server_key()
+		w.write_int(b).write_mpint1(e).write_mpint1(m)
+		b, e, m = self._host_key()
+		w.write_int(b).write_mpint1(e).write_mpint1(m)
+		w.write_int(2)
+		w.write_int(72)
+		w.write_int(36)
+		return w.write_flush()
+	
+	def test_crc32(self):
+		assert self.ssh1.crc32(b'') == 0x00
+		assert self.ssh1.crc32(b'The quick brown fox jumps over the lazy dog') == 0xb9c60808
+	
+	def test_fingerprint(self):
+		# pylint: disable=protected-access
+		b, e, m = self._host_key()
+		fpd = self.wbuf._create_mpint(m, False)
+		fpd += self.wbuf._create_mpint(e, False)
+		fp = self.ssh.Fingerprint(fpd)
+		assert b == 2048
+		assert fp.md5 == 'MD5:9d:26:f8:39:fc:20:9d:9b:ca:cc:4a:0f:e1:93:f5:96'
+		assert fp.sha256 == 'SHA256:vZdx3mhzbvVJmn08t/ruv8WDhJ9jfKYsCTuSzot+QIs'
+	
+	def test_pkm_read(self):
+		pkm = self.ssh1.PublicKeyMessage.parse(self._pkm_payload())
+		assert pkm is not None
+		assert pkm.cookie == b'\x88\x99\xaa\xbb\xcc\xdd\xee\xff'
+		b, e, m = self._server_key()
+		assert pkm.server_key_bits == b
+		assert pkm.server_key_public_exponent == e
+		assert pkm.server_key_public_modulus == m
+		b, e, m = self._host_key()
+		assert pkm.host_key_bits == b
+		assert pkm.host_key_public_exponent == e
+		assert pkm.host_key_public_modulus == m
+		fp = self.ssh.Fingerprint(pkm.host_key_fingerprint_data)
+		assert pkm.protocol_flags == 2
+		assert pkm.supported_ciphers_mask == 72
+		assert pkm.supported_ciphers == ['3des', 'blowfish']
+		assert pkm.supported_authentications_mask == 36
+		assert pkm.supported_authentications == ['rsa', 'tis']
+		assert fp.md5 == 'MD5:9d:26:f8:39:fc:20:9d:9b:ca:cc:4a:0f:e1:93:f5:96'
+		assert fp.sha256 == 'SHA256:vZdx3mhzbvVJmn08t/ruv8WDhJ9jfKYsCTuSzot+QIs'
+	
+	def test_pkm_payload(self):
+		cookie = b'\x88\x99\xaa\xbb\xcc\xdd\xee\xff'
+		skey = self._server_key()
+		hkey = self._host_key()
+		pflags = 2
+		cmask = 72
+		amask = 36
+		pkm1 = self.ssh1.PublicKeyMessage(cookie, skey, hkey, pflags, cmask, amask)
+		pkm2 = self.ssh1.PublicKeyMessage.parse(self._pkm_payload())
+		assert pkm1.payload == pkm2.payload
+	
+	def test_ssh1_server_simple(self, output_spy, virtual_socket):
+		vsocket = virtual_socket
+		w = self.wbuf()
+		w.write_byte(self.ssh.Protocol.SMSG_PUBLIC_KEY)
+		w.write(self._pkm_payload())
+		vsocket.rdata.append(b'SSH-1.5-OpenSSH_7.2 ssh-audit-test\r\n')
+		vsocket.rdata.append(self._create_ssh1_packet(w.write_flush()))
+		output_spy.begin()
+		self.audit(self._conf())
+		lines = output_spy.flush()
+		assert len(lines) == 10
+	
+	def test_ssh1_server_invalid_first_packet(self, output_spy, virtual_socket):
+		vsocket = virtual_socket
+		w = self.wbuf()
+		w.write_byte(self.ssh.Protocol.SMSG_PUBLIC_KEY + 1)
+		w.write(self._pkm_payload())
+		vsocket.rdata.append(b'SSH-1.5-OpenSSH_7.2 ssh-audit-test\r\n')
+		vsocket.rdata.append(self._create_ssh1_packet(w.write_flush()))
+		output_spy.begin()
+		with pytest.raises(SystemExit):
+			self.audit(self._conf())
+		lines = output_spy.flush()
+		assert len(lines) == 4
+		assert 'unknown message' in lines[-1]
+
+	def test_ssh1_server_invalid_checksum(self, output_spy, virtual_socket):
+		vsocket = virtual_socket
+		w = self.wbuf()
+		w.write_byte(self.ssh.Protocol.SMSG_PUBLIC_KEY + 1)
+		w.write(self._pkm_payload())
+		vsocket.rdata.append(b'SSH-1.5-OpenSSH_7.2 ssh-audit-test\r\n')
+		vsocket.rdata.append(self._create_ssh1_packet(w.write_flush(), False))
+		output_spy.begin()
+		with pytest.raises(SystemExit):
+			self.audit(self._conf())
+		lines = output_spy.flush()
+		assert len(lines) == 1
+		assert 'checksum' in lines[-1]

test/test_ssh2.py

@@ -0,0 +1,155 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import struct, os
+import pytest
+
+
+# pylint: disable=line-too-long,attribute-defined-outside-init
+class TestSSH2(object):
+	@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([u'curve25519-sha256@libssh.org', u'ecdh-sha2-nistp256', u'ecdh-sha2-nistp384', u'ecdh-sha2-nistp521', u'diffie-hellman-group-exchange-sha256', u'diffie-hellman-group14-sha1'])
+		w.write_list([u'ssh-rsa', u'rsa-sha2-512', u'rsa-sha2-256', u'ssh-ed25519'])
+		w.write_list([u'chacha20-poly1305@openssh.com', u'aes128-ctr', u'aes192-ctr', u'aes256-ctr', u'aes128-gcm@openssh.com', u'aes256-gcm@openssh.com', u'aes128-cbc', u'aes192-cbc', u'aes256-cbc'])
+		w.write_list([u'chacha20-poly1305@openssh.com', u'aes128-ctr', u'aes192-ctr', u'aes256-ctr', u'aes128-gcm@openssh.com', u'aes256-gcm@openssh.com', u'aes128-cbc', u'aes192-cbc', u'aes256-cbc'])
+		w.write_list([u'umac-64-etm@openssh.com', u'umac-128-etm@openssh.com', u'hmac-sha2-256-etm@openssh.com', u'hmac-sha2-512-etm@openssh.com', u'hmac-sha1-etm@openssh.com', u'umac-64@openssh.com', u'umac-128@openssh.com', u'hmac-sha2-256', u'hmac-sha2-512', u'hmac-sha1'])
+		w.write_list([u'umac-64-etm@openssh.com', u'umac-128-etm@openssh.com', u'hmac-sha2-256-etm@openssh.com', u'hmac-sha2-512-etm@openssh.com', u'hmac-sha1-etm@openssh.com', u'umac-64@openssh.com', u'umac-128@openssh.com', u'hmac-sha2-256', u'hmac-sha2-512', u'hmac-sha1'])
+		w.write_list([u'none', u'zlib@openssh.com'])
+		w.write_list([u'none', u'zlib@openssh.com'])
+		w.write_list([u''])
+		w.write_list([u''])
+		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 == [u'curve25519-sha256@libssh.org', u'ecdh-sha2-nistp256', u'ecdh-sha2-nistp384', u'ecdh-sha2-nistp521', u'diffie-hellman-group-exchange-sha256', u'diffie-hellman-group14-sha1']
+		assert kex.key_algorithms == [u'ssh-rsa', u'rsa-sha2-512', u'rsa-sha2-256', u'ssh-ed25519']
+		assert kex.client is not None
+		assert kex.server is not None
+		assert kex.client.encryption == [u'chacha20-poly1305@openssh.com', u'aes128-ctr', u'aes192-ctr', u'aes256-ctr', u'aes128-gcm@openssh.com', u'aes256-gcm@openssh.com', u'aes128-cbc', u'aes192-cbc', u'aes256-cbc']
+		assert kex.server.encryption == [u'chacha20-poly1305@openssh.com', u'aes128-ctr', u'aes192-ctr', u'aes256-ctr', u'aes128-gcm@openssh.com', u'aes256-gcm@openssh.com', u'aes128-cbc', u'aes192-cbc', u'aes256-cbc']
+		assert kex.client.mac == [u'umac-64-etm@openssh.com', u'umac-128-etm@openssh.com', u'hmac-sha2-256-etm@openssh.com', u'hmac-sha2-512-etm@openssh.com', u'hmac-sha1-etm@openssh.com', u'umac-64@openssh.com', u'umac-128@openssh.com', u'hmac-sha2-256', u'hmac-sha2-512', u'hmac-sha1']
+		assert kex.server.mac == [u'umac-64-etm@openssh.com', u'umac-128-etm@openssh.com', u'hmac-sha2-256-etm@openssh.com', u'hmac-sha2-512-etm@openssh.com', u'hmac-sha1-etm@openssh.com', u'umac-64@openssh.com', u'umac-128@openssh.com', u'hmac-sha2-256', u'hmac-sha2-512', u'hmac-sha1']
+		assert kex.client.compression == [u'none', u'zlib@openssh.com']
+		assert kex.server.compression == [u'none', u'zlib@openssh.com']
+		assert kex.client.languages == [u'']
+		assert kex.server.languages == [u'']
+		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('curve25519-sha256@libssh.org')
+		kex.kex_algorithms.append('ecdh-sha2-nistp256')
+		kex.kex_algorithms.append('ecdh-sha2-nistp384')
+		kex.kex_algorithms.append('ecdh-sha2-nistp521')
+		kex.kex_algorithms.append('diffie-hellman-group-exchange-sha256')
+		kex.kex_algorithms.append('diffie-hellman-group14-sha1')
+		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) == 72
+
+	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()
+		with pytest.raises(SystemExit):
+			self.audit(self._conf())
+		lines = output_spy.flush()
+		assert len(lines) == 3
+		assert 'unknown message' in lines[-1]

test/test_version_compare.py

@@ -2,6 +2,8 @@
 # -*- coding: utf-8 -*-
 import pytest
 
+
+# pylint: disable=attribute-defined-outside-init
 class TestVersionCompare(object):
 	@pytest.fixture(autouse=True)
 	def init(self, ssh_audit):
@@ -15,34 +17,69 @@ class TestVersionCompare(object):
 		b = self.ssh.Banner.parse('SSH-2.0-OpenSSH_{0}'.format(v))
 		return self.ssh.Software.parse(b)
 	
+	def get_libssh_software(self, v):
+		b = self.ssh.Banner.parse('SSH-2.0-libssh-{0}'.format(v))
+		return self.ssh.Software.parse(b)
+	
 	def test_dropbear_compare_version_pre_years(self):
 		s = self.get_dropbear_software('0.44')
+		assert s.compare_version(None) == 1
+		assert s.compare_version('') == 1
 		assert s.compare_version('0.43') > 0
 		assert s.compare_version('0.44') == 0
+		assert s.compare_version(s) == 0
 		assert s.compare_version('0.45') < 0
-		assert s.between_versions('0.43', '0.45') == True
+		assert s.between_versions('0.43', '0.45')
+		assert s.between_versions('0.43', '0.43') is False
+		assert s.between_versions('0.45', '0.43') is False
 	
 	def test_dropbear_compare_version_with_years(self):
 		s = self.get_dropbear_software('2015.71')
-		assert s.compare_version('2014.67') > 0
+		assert s.compare_version(None) == 1
+		assert s.compare_version('') == 1
+		assert s.compare_version('2014.66') > 0
 		assert s.compare_version('2015.71') == 0
+		assert s.compare_version(s) == 0
 		assert s.compare_version('2016.74') < 0
-		assert s.between_versions('2014.67', '2016.74') == True
+		assert s.between_versions('2014.66', '2016.74')
+		assert s.between_versions('2014.66', '2015.69') is False
+		assert s.between_versions('2016.74', '2014.66') is False
 	
 	def test_dropbear_compare_version_mixed(self):
 		s = self.get_dropbear_software('0.53.1')
+		assert s.compare_version(None) == 1
+		assert s.compare_version('') == 1
 		assert s.compare_version('0.53') > 0
 		assert s.compare_version('0.53.1') == 0
+		assert s.compare_version(s) == 0
 		assert s.compare_version('2011.54') < 0
-		assert s.between_versions('0.53', '2011.54') == True
+		assert s.between_versions('0.53', '2011.54')
+		assert s.between_versions('0.53', '0.53') is False
+		assert s.between_versions('2011.54', '0.53') is False
 	
 	def test_dropbear_compare_version_patchlevel(self):
 		s1 = self.get_dropbear_software('0.44')
 		s2 = self.get_dropbear_software('0.44test3')
+		assert s1.compare_version(None) == 1
+		assert s1.compare_version('') == 1
+		assert s1.compare_version('0.44') == 0
+		assert s1.compare_version(s1) == 0
 		assert s1.compare_version('0.43') > 0
 		assert s1.compare_version('0.44test4') > 0
+		assert s1.between_versions('0.44test4', '0.45')
+		assert s1.between_versions('0.43', '0.44test4') is False
+		assert s1.between_versions('0.45', '0.44test4') is False
+		assert s2.compare_version(None) == 1
+		assert s2.compare_version('') == 1
+		assert s2.compare_version('0.44test3') == 0
+		assert s2.compare_version(s2) == 0
 		assert s2.compare_version('0.44') < 0
 		assert s2.compare_version('0.44test4') < 0
+		assert s2.between_versions('0.43', '0.44')
+		assert s2.between_versions('0.43', '0.44test2') is False
+		assert s2.between_versions('0.44', '0.43') is False
+		assert s1.compare_version(s2) > 0
+		assert s2.compare_version(s1) < 0
 	
 	def test_dropbear_compare_version_sequential(self):
 		versions = []
@@ -82,20 +119,28 @@ class TestVersionCompare(object):
 	
 	def test_openssh_compare_version_simple(self):
 		s = self.get_openssh_software('3.7.1')
+		assert s.compare_version(None) == 1
+		assert s.compare_version('') == 1
 		assert s.compare_version('3.7') > 0
 		assert s.compare_version('3.7.1') == 0
+		assert s.compare_version(s) == 0
 		assert s.compare_version('3.8') < 0
-		assert s.between_versions('3.7', '3.8') == True
-		
+		assert s.between_versions('3.7', '3.8')
+		assert s.between_versions('3.6', '3.7') is False
+		assert s.between_versions('3.8', '3.7') is False
 	
 	def test_openssh_compare_version_patchlevel(self):
 		s1 = self.get_openssh_software('2.1.1')
 		s2 = self.get_openssh_software('2.1.1p2')
+		assert s1.compare_version(s1) == 0
+		assert s2.compare_version(s2) == 0
 		assert s1.compare_version('2.1.1p1') == 0
 		assert s1.compare_version('2.1.1p2') == 0
 		assert s2.compare_version('2.1.1') == 0
 		assert s2.compare_version('2.1.1p1') > 0
 		assert s2.compare_version('2.1.1p3') < 0
+		assert s1.compare_version(s2) == 0
+		assert s2.compare_version(s1) == 0
 	
 	def test_openbsd_compare_version_sequential(self):
 		versions = []
@@ -130,3 +175,41 @@ class TestVersionCompare(object):
 			if i + 1 < l:
 				vnext = versions[i + 1]
 				assert s.compare_version(vnext) < 0
+	
+	def test_libssh_compare_version_simple(self):
+		s = self.get_libssh_software('0.3')
+		assert s.compare_version(None) == 1
+		assert s.compare_version('') == 1
+		assert s.compare_version('0.2') > 0
+		assert s.compare_version('0.3') == 0
+		assert s.compare_version(s) == 0
+		assert s.compare_version('0.3.1') < 0
+		assert s.between_versions('0.2', '0.3.1')
+		assert s.between_versions('0.1', '0.2') is False
+		assert s.between_versions('0.3.1', '0.2') is False
+	
+	def test_libssh_compare_version_sequential(self):
+		versions = []
+		for v in ['0.2', '0.3']:
+			versions.append(v)
+		for i in range(1, 5):
+			versions.append('0.3.{0}'.format(i))
+		for i in range(0, 9):
+			versions.append('0.4.{0}'.format(i))
+		for i in range(0, 6):
+			versions.append('0.5.{0}'.format(i))
+		for i in range(0, 6):
+			versions.append('0.6.{0}'.format(i))
+		for i in range(0, 4):
+			versions.append('0.7.{0}'.format(i))
+		l = len(versions)
+		for i in range(l):
+			v = versions[i]
+			s = self.get_libssh_software(v)
+			assert s.compare_version(v) == 0
+			if i - 1 >= 0:
+				vbefore = versions[i - 1]
+				assert s.compare_version(vbefore) > 0
+			if i + 1 < l:
+				vnext = versions[i + 1]
+				assert s.compare_version(vnext) < 0