elie/ssh-audit
1
0
Fork 0
mirror of https://github.com/jtesta/ssh-audit.git synced 2024-09-28 19:55:05 +02:00
Code Issues Packages Projects Releases Wiki Activity

Remove some more Python 2 leftovers (#37)

Browse Source 
* Remove mypy job for Python 2

modified:   tox.ini

* Remove Python 2 compatibility import

modified:   ssh-audit.py

* Remove compatibility import for BytesIO and StringIO

This is no longer necessary, as support for Python 2 was dropped.

modified:   ssh-audit.py

* Remove `text-type` compatibility layer

... as support for Python 2 was dropped already.

modified:   ssh-audit.py

* Remove `binary-type` compatibility layer

... as support for Python 2 was dropped already.

modified:   ssh-audit.py

* Remove try-except block for typing

... as since Python 3.5 it is included in the standard library.

modified:   ssh-audit.py

* Move typing import to top of module

modified:   ssh-audit.py

* Remove obsolete encoding declaration

modified:   ssh-audit.py

* Apply pyupgrade on ssh-audit.py

pyupgrade is a tool which updates Python code to modern syntax

modified:   ssh-audit.py

* Remove Python 2 compatibility from conftest.py

modified:   test/conftest.py

* Remove Python 2 compatibility from test_auditconf.py

modified:   test/test_auditconf.py

* Remove Python 2 compatibility from test_banner.py

modified:   test/test_banner.py

* Remove Python 2 compatibility from test_buffer.py

modified:   test/test_buffer.py

* Remove Python 2 compatibility from test_errors.py

modified:   test/test_errors.py

* Remove Python 2 compatibility from test_output.py

modified:   test/test_output.py

* Remove Python 2 compatibility from test_resolve.py

modified:   test/test_resolve.py

* Remove Python 2 compatibility from test_socket.py

modified:   test/test_socket.py

* Remove Python 2 compatibility from test_software.py

modified:   test/test_software.py

* Remove Python 2 compatibility from test_ssh_algorithm.py

modified:   test/test_ssh_algorithm.py

* Remove Python 2 compatibility from test_ssh1.py

modified:   test/test_ssh1.py

* Remove Python 2 compatibility from test_ssh2.py

modified:   test/test_ssh2.py

* Remove Python 2 compatibility and Py2 only tests

... from test_utils.py.

modified:   test/test_utils.py

* Remove Python 2 compatibility from test_version_compare.py

modified:   test/test_version_compare.py

* Remove Python 2 job from appveyor config

This was done blindly, as it is unclear whether appveyor runs at all.

modified:   .appveyor.yml
This commit is contained in:
Jürgen Gmach 2020-06-15 23:05:31 +02:00 committed by GitHub
parent 42fecf83e6
commit ec1dda8d7f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 231 additions and 433 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.appveyor.yml
View File

@ -8,8 +8,6 @@ branches:
environment: environment:
matrix: matrix:
- PYTHON: "C:\\Python27"
- PYTHON: "C:\\Python27-x64"
- PYTHON: "C:\\Python35" - PYTHON: "C:\\Python35"
- PYTHON: "C:\\Python35-x64" - PYTHON: "C:\\Python35-x64"
- PYTHON: "C:\\Python36" - PYTHON: "C:\\Python36"

305
ssh-audit.py
View File

@ -1,5 +1,4 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" """
The MIT License (MIT) The MIT License (MIT)
@ -24,7 +23,6 @@
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE. THE SOFTWARE.
""" """
from __future__ import print_function
import base64 import base64
import binascii import binascii
import errno import errno
@ -39,26 +37,13 @@ import select
import socket import socket
import struct import struct
import sys import sys
# pylint: disable=unused-import
from typing import Dict, List, Set, Sequence, Tuple, Iterable
from typing import Callable, Optional, Union, Any
VERSION = 'v2.2.1-dev' VERSION = 'v2.2.1-dev'
SSH_HEADER = 'SSH-{0}-OpenSSH_8.0' # SSH software to impersonate SSH_HEADER = 'SSH-{0}-OpenSSH_8.0' # SSH software to impersonate
if sys.version_info >= (3,): # pragma: nocover
StringIO, BytesIO = io.StringIO, io.BytesIO
text_type = str
binary_type = bytes
else: # pragma: nocover
import StringIO as _StringIO # pylint: disable=import-error
StringIO = BytesIO = _StringIO.StringIO
text_type = unicode # pylint: disable=undefined-variable # noqa: F821
binary_type = str
try: # pragma: nocover
# pylint: disable=unused-import
from typing import Dict, List, Set, Sequence, Tuple, Iterable
from typing import Callable, Optional, Union, Any
except ImportError: # pragma: nocover
pass
try: # pragma: nocover try: # pragma: nocover
from colorama import init as colorama_init from colorama import init as colorama_init
colorama_init(strip=False) # pragma: nocover colorama_init(strip=False) # pragma: nocover
@ -70,10 +55,10 @@ def usage(err=None):
# type: (Optional[str]) -> None # type: (Optional[str]) -> None
uout = Output() uout = Output()
p = os.path.basename(sys.argv[0]) p = os.path.basename(sys.argv[0])
uout.head('# {0} {1}, https://github.com/jtesta/ssh-audit\n'.format(p, VERSION)) uout.head('# {} {}, https://github.com/jtesta/ssh-audit\n'.format(p, VERSION))
if err is not None and len(err) > 0: if err is not None and len(err) > 0:
uout.fail('\n' + err) uout.fail('\n' + err)
uout.info('usage: {0} [-1246pbcnjvlt] <host>\n'.format(p)) uout.info('usage: {} [-1246pbcnjvlt] <host>\n'.format(p))
uout.info(' -h, --help print this help') uout.info(' -h, --help print this help')
uout.info(' -1, --ssh1 force ssh version 1 only') uout.info(' -1, --ssh1 force ssh version 1 only')
uout.info(' -2, --ssh2 force ssh version 2 only') uout.info(' -2, --ssh2 force ssh version 2 only')
@ -139,18 +124,18 @@ class AuditConf:
elif name == 'port': elif name == 'port':
valid, port = True, utils.parse_int(value) valid, port = True, utils.parse_int(value)
if port < 1 or port > 65535: if port < 1 or port > 65535:
raise ValueError('invalid port: {0}'.format(value)) raise ValueError('invalid port: {}'.format(value))
value = port value = port
elif name in ['level']: elif name in ['level']:
if value not in ('info', 'warn', 'fail'): if value not in ('info', 'warn', 'fail'):
raise ValueError('invalid level: {0}'.format(value)) raise ValueError('invalid level: {}'.format(value))
valid = True valid = True
elif name == 'host': elif name == 'host':
valid = True valid = True
elif name == 'timeout': elif name == 'timeout':
value = utils.parse_float(value) value = utils.parse_float(value)
if value == -1.0: if value == -1.0:
raise ValueError('invalid timeout: {0}'.format(value)) raise ValueError('invalid timeout: {}'.format(value))
valid = True valid = True
if valid: if valid:
object.__setattr__(self, name, value) object.__setattr__(self, name, value)
@ -195,7 +180,7 @@ class AuditConf:
aconf.verbose = True aconf.verbose = True
elif o in ('-l', '--level'): elif o in ('-l', '--level'):
if a not in ('info', 'warn', 'fail'): if a not in ('info', 'warn', 'fail'):
usage_cb('level {0} is not valid'.format(a)) usage_cb('level {} is not valid'.format(a))
aconf.level = a aconf.level = a
elif o in ('-t', '--timeout'): elif o in ('-t', '--timeout'):
aconf.timeout = float(a) aconf.timeout = float(a)
@ -223,7 +208,7 @@ class AuditConf:
oport = '2222' oport = '2222'
port = utils.parse_int(oport) port = utils.parse_int(oport)
if port <= 0 or port > 65535: if port <= 0 or port > 65535:
usage_cb('port {0} is not valid'.format(oport)) usage_cb('port {} is not valid'.format(oport))
aconf.host = host aconf.host = host
aconf.port = port aconf.port = port
if not (aconf.ssh1 or aconf.ssh2): if not (aconf.ssh1 or aconf.ssh2):
@ -275,27 +260,27 @@ class Output:
@staticmethod @staticmethod
def _colorized(color): def _colorized(color):
# type: (str) -> Callable[[text_type], None] # type: (str) -> Callable[[str], None]
return lambda x: print(u'{0}{1}\033[0m'.format(color, x)) return lambda x: print(u'{}{}\033[0m'.format(color, x))
def __getattr__(self, name): def __getattr__(self, name):
# type: (str) -> Callable[[text_type], None] # type: (str) -> Callable[[str], None]
if name == 'head' and self.batch: if name == 'head' and self.batch:
return lambda x: None return lambda x: None
if not self.get_level(name) >= self.__level: if not self.get_level(name) >= self.__level:
return lambda x: None return lambda x: None
if self.use_colors and self.colors_supported and name in self.COLORS: if self.use_colors and self.colors_supported and name in self.COLORS:
color = '\033[0;{0}m'.format(self.COLORS[name]) color = '\033[0;{}m'.format(self.COLORS[name])
return self._colorized(color) return self._colorized(color)
else: else:
return lambda x: print(u'{0}'.format(x)) return lambda x: print(u'{}'.format(x))
class OutputBuffer(list): class OutputBuffer(list):
def __enter__(self): def __enter__(self):
# type: () -> OutputBuffer # type: () -> OutputBuffer
# pylint: disable=attribute-defined-outside-init # pylint: disable=attribute-defined-outside-init
self.__buf = StringIO() self.__buf = io.StringIO()
self.__stdout = sys.stdout self.__stdout = sys.stdout
sys.stdout = self.__buf sys.stdout = self.__buf
return self return self
@ -536,7 +521,7 @@ class SSH2: # pylint: disable=too-few-public-methods
class KexParty: class KexParty:
def __init__(self, enc, mac, compression, languages): def __init__(self, enc, mac, compression, languages):
# type: (List[text_type], List[text_type], List[text_type], List[text_type]) -> None # type: (List[str], List[str], List[str], List[str]) -> None
self.__enc = enc self.__enc = enc
self.__mac = mac self.__mac = mac
self.__compression = compression self.__compression = compression
@ -544,27 +529,27 @@ class SSH2: # pylint: disable=too-few-public-methods
@property @property
def encryption(self): def encryption(self):
# type: () -> List[text_type] # type: () -> List[str]
return self.__enc return self.__enc
@property @property
def mac(self): def mac(self):
# type: () -> List[text_type] # type: () -> List[str]
return self.__mac return self.__mac
@property @property
def compression(self): def compression(self):
# type: () -> List[text_type] # type: () -> List[str]
return self.__compression return self.__compression
@property @property
def languages(self): def languages(self):
# type: () -> List[text_type] # type: () -> List[str]
return self.__languages return self.__languages
class Kex: class Kex:
def __init__(self, cookie, kex_algs, key_algs, cli, srv, follows, unused=0): def __init__(self, cookie, kex_algs, key_algs, cli, srv, follows, unused=0):
# type: (binary_type, List[text_type], List[text_type], SSH2.KexParty, SSH2.KexParty, bool, int) -> None # type: (bytes, List[str], List[str], SSH2.KexParty, SSH2.KexParty, bool, int) -> None
self.__cookie = cookie self.__cookie = cookie
self.__kex_algs = kex_algs self.__kex_algs = kex_algs
self.__key_algs = key_algs self.__key_algs = key_algs
@ -579,17 +564,17 @@ class SSH2: # pylint: disable=too-few-public-methods
@property @property
def cookie(self): def cookie(self):
# type: () -> binary_type # type: () -> bytes
return self.__cookie return self.__cookie
@property @property
def kex_algorithms(self): def kex_algorithms(self):
# type: () -> List[text_type] # type: () -> List[str]
return self.__kex_algs return self.__kex_algs
@property @property
def key_algorithms(self): def key_algorithms(self):
# type: () -> List[text_type] # type: () -> List[str]
return self.__key_algs return self.__key_algs
# client_to_server # client_to_server
@ -650,14 +635,14 @@ class SSH2: # pylint: disable=too-few-public-methods
@property @property
def payload(self): def payload(self):
# type: () -> binary_type # type: () -> bytes
wbuf = WriteBuf() wbuf = WriteBuf()
self.write(wbuf) self.write(wbuf)
return wbuf.write_flush() return wbuf.write_flush()
@classmethod @classmethod
def parse(cls, payload): def parse(cls, payload):
# type: (binary_type) -> SSH2.Kex # type: (bytes) -> SSH2.Kex
buf = ReadBuf(payload) buf = ReadBuf(payload)
cookie = buf.read(16) cookie = buf.read(16)
kex_algs = buf.read_list() kex_algs = buf.read_list()
@ -943,7 +928,7 @@ class SSH1:
self._table[i] = crc self._table[i] = crc
def calc(self, v): def calc(self, v):
# type: (binary_type) -> int # type: (bytes) -> int
crc, length = 0, len(v) crc, length = 0, len(v)
for i in range(length): for i in range(length):
n = ord(v[i:i + 1]) n = ord(v[i:i + 1])
@ -957,7 +942,7 @@ class SSH1:
@classmethod @classmethod
def crc32(cls, v): def crc32(cls, v):
# type: (binary_type) -> int # type: (bytes) -> int
if cls._crc32 is None: if cls._crc32 is None:
cls._crc32 = cls.CRC32() cls._crc32 = cls.CRC32()
return cls._crc32.calc(v) return cls._crc32.calc(v)
@ -995,11 +980,11 @@ class SSH1:
class PublicKeyMessage: class PublicKeyMessage:
def __init__(self, cookie, skey, hkey, pflags, cmask, amask): def __init__(self, cookie, skey, hkey, pflags, cmask, amask):
# type: (binary_type, Tuple[int, int, int], Tuple[int, int, int], int, int, int) -> None # type: (bytes, Tuple[int, int, int], Tuple[int, int, int], int, int, int) -> None
if len(skey) != 3: if len(skey) != 3:
raise ValueError('invalid server key pair: {0}'.format(skey)) raise ValueError('invalid server key pair: {}'.format(skey))
if len(hkey) != 3: if len(hkey) != 3:
raise ValueError('invalid host key pair: {0}'.format(hkey)) raise ValueError('invalid host key pair: {}'.format(hkey))
self.__cookie = cookie self.__cookie = cookie
self.__server_key = skey self.__server_key = skey
self.__host_key = hkey self.__host_key = hkey
@ -1009,7 +994,7 @@ class SSH1:
@property @property
def cookie(self): def cookie(self):
# type: () -> binary_type # type: () -> bytes
return self.__cookie return self.__cookie
@property @property
@ -1044,7 +1029,7 @@ class SSH1:
@property @property
def host_key_fingerprint_data(self): def host_key_fingerprint_data(self):
# type: () -> binary_type # type: () -> bytes
# pylint: disable=protected-access # pylint: disable=protected-access
mod = WriteBuf._create_mpint(self.host_key_public_modulus, False) mod = WriteBuf._create_mpint(self.host_key_public_modulus, False)
e = WriteBuf._create_mpint(self.host_key_public_exponent, False) e = WriteBuf._create_mpint(self.host_key_public_exponent, False)
@ -1062,7 +1047,7 @@ class SSH1:
@property @property
def supported_ciphers(self): def supported_ciphers(self):
# type: () -> List[text_type] # type: () -> List[str]
ciphers = [] ciphers = []
for i in range(len(SSH1.CIPHERS)): for i in range(len(SSH1.CIPHERS)):
if self.__supported_ciphers_mask & (1 << i) != 0: if self.__supported_ciphers_mask & (1 << i) != 0:
@ -1076,7 +1061,7 @@ class SSH1:
@property @property
def supported_authentications(self): def supported_authentications(self):
# type: () -> List[text_type] # type: () -> List[str]
auths = [] auths = []
for i in range(1, len(SSH1.AUTHS)): for i in range(1, len(SSH1.AUTHS)):
if self.__supported_authentications_mask & (1 << i) != 0: if self.__supported_authentications_mask & (1 << i) != 0:
@ -1098,14 +1083,14 @@ class SSH1:
@property @property
def payload(self): def payload(self):
# type: () -> binary_type # type: () -> bytes
wbuf = WriteBuf() wbuf = WriteBuf()
self.write(wbuf) self.write(wbuf)
return wbuf.write_flush() return wbuf.write_flush()
@classmethod @classmethod
def parse(cls, payload): def parse(cls, payload):
# type: (binary_type) -> SSH1.PublicKeyMessage # type: (bytes) -> SSH1.PublicKeyMessage
buf = ReadBuf(payload) buf = ReadBuf(payload)
cookie = buf.read(8) cookie = buf.read(8)
server_key_bits = buf.read_int() server_key_bits = buf.read_int()
@ -1125,9 +1110,9 @@ class SSH1:
class ReadBuf: class ReadBuf:
def __init__(self, data=None): def __init__(self, data=None):
# type: (Optional[binary_type]) -> None # type: (Optional[bytes]) -> None
super(ReadBuf, self).__init__() super(ReadBuf, self).__init__()
self._buf = BytesIO(data) if data is not None else BytesIO() self._buf = io.BytesIO(data) if data is not None else io.BytesIO()
self._len = len(data) if data is not None else 0 self._len = len(data) if data is not None else 0
@property @property
@ -1136,7 +1121,7 @@ class ReadBuf:
return self._len - self._buf.tell() return self._len - self._buf.tell()
def read(self, size): def read(self, size):
# type: (int) -> binary_type # type: (int) -> bytes
return self._buf.read(size) return self._buf.read(size)
def read_byte(self): def read_byte(self):
@ -1154,18 +1139,18 @@ class ReadBuf:
return v return v
def read_list(self): def read_list(self):
# type: () -> List[text_type] # type: () -> List[str]
list_size = self.read_int() list_size = self.read_int()
return self.read(list_size).decode('utf-8', 'replace').split(',') return self.read(list_size).decode('utf-8', 'replace').split(',')
def read_string(self): def read_string(self):
# type: () -> binary_type # type: () -> bytes
n = self.read_int() n = self.read_int()
return self.read(n) return self.read(n)
@classmethod @classmethod
def _parse_mpint(cls, v, pad, f): def _parse_mpint(cls, v, pad, f):
# type: (binary_type, binary_type, str) -> int # type: (bytes, bytes, str) -> int
r = 0 r = 0
if len(v) % 4 != 0: if len(v) % 4 != 0:
v = pad * (4 - (len(v) % 4)) + v v = pad * (4 - (len(v) % 4)) + v
@ -1190,22 +1175,22 @@ class ReadBuf:
return self._parse_mpint(v, pad, f) return self._parse_mpint(v, pad, f)
def read_line(self): def read_line(self):
# type: () -> text_type # type: () -> str
return self._buf.readline().rstrip().decode('utf-8', 'replace') return self._buf.readline().rstrip().decode('utf-8', 'replace')
def reset(self): def reset(self):
self._buf = BytesIO() self._buf = io.BytesIO()
self._len = 0 self._len = 0
class WriteBuf: class WriteBuf:
def __init__(self, data=None): def __init__(self, data=None):
# type: (Optional[binary_type]) -> None # type: (Optional[bytes]) -> None
super(WriteBuf, self).__init__() super(WriteBuf, self).__init__()
self._wbuf = BytesIO(data) if data is not None else BytesIO() self._wbuf = io.BytesIO(data) if data is not None else io.BytesIO()
def write(self, data): def write(self, data):
# type: (binary_type) -> WriteBuf # type: (bytes) -> WriteBuf
self._wbuf.write(data) self._wbuf.write(data)
return self return self
@ -1222,14 +1207,14 @@ class WriteBuf:
return self.write(struct.pack('>I', v)) return self.write(struct.pack('>I', v))
def write_string(self, v): def write_string(self, v):
# type: (Union[binary_type, text_type]) -> WriteBuf # type: (Union[bytes, str]) -> WriteBuf
if not isinstance(v, bytes): if not isinstance(v, bytes):
v = bytes(bytearray(v, 'utf-8')) v = bytes(bytearray(v, 'utf-8'))
self.write_int(len(v)) self.write_int(len(v))
return self.write(v) return self.write(v)
def write_list(self, v): def write_list(self, v):
# type: (List[text_type]) -> WriteBuf # type: (List[str]) -> WriteBuf
return self.write_string(u','.join(v)) return self.write_string(u','.join(v))
@classmethod @classmethod
@ -1242,12 +1227,12 @@ class WriteBuf:
@classmethod @classmethod
def _create_mpint(cls, n, signed=True, bits=None): def _create_mpint(cls, n, signed=True, bits=None):
# type: (int, bool, Optional[int]) -> binary_type # type: (int, bool, Optional[int]) -> bytes
if bits is None: if bits is None:
bits = cls._bitlength(n) bits = cls._bitlength(n)
length = bits // 8 + (1 if n != 0 else 0) length = bits // 8 + (1 if n != 0 else 0)
ql = (length + 7) // 8 ql = (length + 7) // 8
fmt, v2 = '>{0}Q'.format(ql), [0] * ql fmt, v2 = '>{}Q'.format(ql), [0] * ql
for i in range(ql): for i in range(ql):
v2[ql - i - 1] = n & 0xffffffffffffffff v2[ql - i - 1] = n & 0xffffffffffffffff
n >>= 64 n >>= 64
@ -1273,21 +1258,21 @@ class WriteBuf:
return self.write_string(data) return self.write_string(data)
def write_line(self, v): def write_line(self, v):
# type: (Union[binary_type, str]) -> WriteBuf # type: (Union[bytes, str]) -> WriteBuf
if not isinstance(v, bytes): if not isinstance(v, bytes):
v = bytes(bytearray(v, 'utf-8')) v = bytes(bytearray(v, 'utf-8'))
v += b'\r\n' v += b'\r\n'
return self.write(v) return self.write(v)
def write_flush(self): def write_flush(self):
# type: () -> binary_type # type: () -> bytes
payload = self._wbuf.getvalue() payload = self._wbuf.getvalue()
self._wbuf.truncate(0) self._wbuf.truncate(0)
self._wbuf.seek(0) self._wbuf.seek(0)
return payload return payload
def reset(self): def reset(self):
self._wbuf = BytesIO() self._wbuf = io.BytesIO()
class SSH: # pylint: disable=too-few-public-methods class SSH: # pylint: disable=too-few-public-methods
@ -1346,12 +1331,12 @@ class SSH: # pylint: disable=too-few-public-methods
return self.__os return self.__os
def compare_version(self, other): def compare_version(self, other):
# type: (Union[None, SSH.Software, text_type]) -> int # type: (Union[None, SSH.Software, str]) -> int
# pylint: disable=too-many-branches # pylint: disable=too-many-branches
if other is None: if other is None:
return 1 return 1
if isinstance(other, SSH.Software): if isinstance(other, SSH.Software):
other = '{0}{1}'.format(other.version, other.patch or '') other = '{}{}'.format(other.version, other.patch or '')
else: else:
other = str(other) other = str(other)
mx = re.match(r'^([\d\.]+\d+)(.*)$', other) mx = re.match(r'^([\d\.]+\d+)(.*)$', other)
@ -1366,9 +1351,9 @@ class SSH: # pylint: disable=too-few-public-methods
spatch = self.patch or '' spatch = self.patch or ''
if self.product == SSH.Product.DropbearSSH: if self.product == SSH.Product.DropbearSSH:
if not re.match(r'^test\d.*$', opatch): if not re.match(r'^test\d.*$', opatch):
opatch = 'z{0}'.format(opatch) opatch = 'z{}'.format(opatch)
if not re.match(r'^test\d.*$', spatch): if not re.match(r'^test\d.*$', spatch):
spatch = 'z{0}'.format(spatch) spatch = 'z{}'.format(spatch)
elif self.product == SSH.Product.OpenSSH: elif self.product == SSH.Product.OpenSSH:
mx1 = re.match(r'^p\d(.*)', opatch) mx1 = re.match(r'^p\d(.*)', opatch)
mx2 = re.match(r'^p\d(.*)', spatch) mx2 = re.match(r'^p\d(.*)', spatch)
@ -1393,10 +1378,10 @@ class SSH: # pylint: disable=too-few-public-methods
def display(self, full=True): def display(self, full=True):
# type: (bool) -> str # type: (bool) -> str
r = '{0} '.format(self.vendor) if bool(self.vendor) else '' r = '{} '.format(self.vendor) if bool(self.vendor) else ''
r += self.product r += self.product
if bool(self.version): if bool(self.version):
r += ' {0}'.format(self.version) r += ' {}'.format(self.version)
if full: if full:
patch = self.patch or '' patch = self.patch or ''
if self.product == SSH.Product.OpenSSH: if self.product == SSH.Product.OpenSSH:
@ -1405,9 +1390,9 @@ class SSH: # pylint: disable=too-few-public-methods
r += mx.group(1) r += mx.group(1)
patch = mx.group(2).strip() patch = mx.group(2).strip()
if bool(patch): if bool(patch):
r += ' ({0})'.format(patch) r += ' ({})'.format(patch)
if bool(self.os): if bool(self.os):
r += ' running on {0}'.format(self.os) r += ' running on {}'.format(self.os)
return r return r
def __str__(self): def __str__(self):
@ -1416,15 +1401,15 @@ class SSH: # pylint: disable=too-few-public-methods
def __repr__(self): def __repr__(self):
# type: () -> str # type: () -> str
r = 'vendor={0}, '.format(self.vendor) if bool(self.vendor) else '' r = 'vendor={}, '.format(self.vendor) if bool(self.vendor) else ''
r += 'product={0}'.format(self.product) r += 'product={}'.format(self.product)
if bool(self.version): if bool(self.version):
r += ', version={0}'.format(self.version) r += ', version={}'.format(self.version)
if bool(self.patch): if bool(self.patch):
r += ', patch={0}'.format(self.patch) r += ', patch={}'.format(self.patch)
if bool(self.os): if bool(self.os):
r += ', os={0}'.format(self.os) r += ', os={}'.format(self.os)
return '<{0}({1})>'.format(self.__class__.__name__, r) return '<{}({})>'.format(self.__class__.__name__, r)
@staticmethod @staticmethod
def _fix_patch(patch): def _fix_patch(patch):
@ -1435,7 +1420,7 @@ class SSH: # pylint: disable=too-few-public-methods
def _fix_date(d): def _fix_date(d):
# type: (str) -> Optional[str] # type: (str) -> Optional[str]
if d is not None and len(d) == 8: if d is not None and len(d) == 8:
return '{0}-{1}-{2}'.format(d[:4], d[4:6], d[6:8]) return '{}-{}-{}'.format(d[:4], d[4:6], d[6:8])
else: else:
return None return None
@ -1447,19 +1432,19 @@ class SSH: # pylint: disable=too-few-public-methods
mx = re.match(r'^NetBSD(?:_Secure_Shell)?(?:[\s-]+(\d{8})(.*))?$', c) mx = re.match(r'^NetBSD(?:_Secure_Shell)?(?:[\s-]+(\d{8})(.*))?$', c)
if bool(mx): if bool(mx):
d = cls._fix_date(mx.group(1)) d = cls._fix_date(mx.group(1))
return 'NetBSD' if d is None else 'NetBSD ({0})'.format(d) return 'NetBSD' if d is None else 'NetBSD ({})'.format(d)
mx = re.match(r'^FreeBSD(?:\slocalisations)?[\s-]+(\d{8})(.*)$', c) mx = re.match(r'^FreeBSD(?:\slocalisations)?[\s-]+(\d{8})(.*)$', c)
if not bool(mx): if not bool(mx):
mx = re.match(r'^[^@]+@FreeBSD\.org[\s-]+(\d{8})(.*)$', c) mx = re.match(r'^[^@]+@FreeBSD\.org[\s-]+(\d{8})(.*)$', c)
if bool(mx): if bool(mx):
d = cls._fix_date(mx.group(1)) d = cls._fix_date(mx.group(1))
return 'FreeBSD' if d is None else 'FreeBSD ({0})'.format(d) return 'FreeBSD' if d is None else 'FreeBSD ({})'.format(d)
w = ['RemotelyAnywhere', 'DesktopAuthority', 'RemoteSupportManager'] w = ['RemotelyAnywhere', 'DesktopAuthority', 'RemoteSupportManager']
for win_soft in w: for win_soft in w:
mx = re.match(r'^in ' + win_soft + r' ([\d\.]+\d)$', c) mx = re.match(r'^in ' + win_soft + r' ([\d\.]+\d)$', c)
if bool(mx): if bool(mx):
ver = mx.group(1) ver = mx.group(1)
return 'Microsoft Windows ({0} {1})'.format(win_soft, ver) return 'Microsoft Windows ({} {})'.format(win_soft, ver)
generic = ['NetBSD', 'FreeBSD'] generic = ['NetBSD', 'FreeBSD']
for g in generic: for g in generic:
if c.startswith(g) or c.endswith(g): if c.startswith(g) or c.endswith(g):
@ -1552,26 +1537,26 @@ class SSH: # pylint: disable=too-few-public-methods
def __str__(self): def __str__(self):
# type: () -> str # type: () -> str
r = 'SSH-{0}.{1}'.format(self.protocol[0], self.protocol[1]) r = 'SSH-{}.{}'.format(self.protocol[0], self.protocol[1])
if self.software is not None: if self.software is not None:
r += '-{0}'.format(self.software) r += '-{}'.format(self.software)
if bool(self.comments): if bool(self.comments):
r += ' {0}'.format(self.comments) r += ' {}'.format(self.comments)
return r return r
def __repr__(self): def __repr__(self):
# type: () -> str # type: () -> str
p = '{0}.{1}'.format(self.protocol[0], self.protocol[1]) p = '{}.{}'.format(self.protocol[0], self.protocol[1])
r = 'protocol={0}'.format(p) r = 'protocol={}'.format(p)
if self.software is not None: if self.software is not None:
r += ', software={0}'.format(self.software) r += ', software={}'.format(self.software)
if bool(self.comments): if bool(self.comments):
r += ', comments={0}'.format(self.comments) r += ', comments={}'.format(self.comments)
return '<{0}({1})>'.format(self.__class__.__name__, r) return '<{}({})>'.format(self.__class__.__name__, r)
@classmethod @classmethod
def parse(cls, banner): def parse(cls, banner):
# type: (text_type) -> Optional[SSH.Banner] # type: (str) -> Optional[SSH.Banner]
valid_ascii = utils.is_print_ascii(banner) valid_ascii = utils.is_print_ascii(banner)
ascii_banner = utils.to_print_ascii(banner) ascii_banner = utils.to_print_ascii(banner)
mx = cls.RX_BANNER.match(ascii_banner) mx = cls.RX_BANNER.match(ascii_banner)
@ -1589,22 +1574,22 @@ class SSH: # pylint: disable=too-few-public-methods
class Fingerprint: class Fingerprint:
def __init__(self, fpd): def __init__(self, fpd):
# type: (binary_type) -> None # type: (bytes) -> None
self.__fpd = fpd self.__fpd = fpd
@property @property
def md5(self): def md5(self):
# type: () -> text_type # type: () -> str
h = hashlib.md5(self.__fpd).hexdigest() h = hashlib.md5(self.__fpd).hexdigest()
r = u':'.join(h[i:i + 2] for i in range(0, len(h), 2)) r = u':'.join(h[i:i + 2] for i in range(0, len(h), 2))
return u'MD5:{0}'.format(r) return u'MD5:{}'.format(r)
@property @property
def sha256(self): def sha256(self):
# type: () -> text_type # type: () -> str
h = base64.b64encode(hashlib.sha256(self.__fpd).digest()) h = base64.b64encode(hashlib.sha256(self.__fpd).digest())
r = h.decode('ascii').rstrip('=') r = h.decode('ascii').rstrip('=')
return u'SHA256:{0}'.format(r) return u'SHA256:{}'.format(r)
class Algorithm: class Algorithm:
class Timeframe: class Timeframe:
@ -1679,7 +1664,7 @@ class SSH: # pylint: disable=too-few-public-methods
@classmethod @classmethod
def get_since_text(cls, versions): def get_since_text(cls, versions):
# type: (List[Optional[str]]) -> Optional[text_type] # type: (List[Optional[str]]) -> Optional[str]
tv = [] tv = []
if len(versions) == 0 or versions[0] is None: if len(versions) == 0 or versions[0] is None:
return None return None
@ -1690,8 +1675,8 @@ class SSH: # pylint: disable=too-few-public-methods
if ssh_prod in [SSH.Product.LibSSH]: if ssh_prod in [SSH.Product.LibSSH]:
continue continue
if is_cli: if is_cli:
ssh_ver = '{0} (client only)'.format(ssh_ver) ssh_ver = '{} (client only)'.format(ssh_ver)
tv.append('{0} {1}'.format(ssh_prod, ssh_ver)) tv.append('{} {}'.format(ssh_prod, ssh_ver))
if len(tv) == 0: if len(tv) == 0:
return None return None
return 'available since ' + ', '.join(tv).rstrip(', ') return 'available since ' + ', '.join(tv).rstrip(', ')
@ -1746,7 +1731,7 @@ class SSH: # pylint: disable=too-few-public-methods
def maxlen(self): def maxlen(self):
# type: () -> int # type: () -> int
def _ml(items): def _ml(items):
# type: (Sequence[text_type]) -> int # type: (Sequence[str]) -> int
return max(len(i) for i in items) return max(len(i) for i in items)
maxlen = 0 maxlen = 0
if self.ssh1kex is not None: if self.ssh1kex is not None:
@ -1888,7 +1873,7 @@ class SSH: # pylint: disable=too-few-public-methods
# type: (int, Dict[str, Dict[str, List[List[Optional[str]]]]]) -> None # type: (int, Dict[str, Dict[str, List[List[Optional[str]]]]]) -> None
self.__sshv = sshv self.__sshv = sshv
self.__db = db self.__db = db
self.__storage = {} # type: Dict[str, List[text_type]] self.__storage = {} # type: Dict[str, List[str]]
@property @property
def sshv(self): def sshv(self):
@ -1901,11 +1886,11 @@ class SSH: # pylint: disable=too-few-public-methods
return self.__db return self.__db
def add(self, key, value): def add(self, key, value):
# type: (str, List[text_type]) -> None # type: (str, List[str]) -> None
self.__storage[key] = value self.__storage[key] = value
def items(self): def items(self):
# type: () -> Iterable[Tuple[str, List[text_type]]] # type: () -> Iterable[Tuple[str, List[str]]]
return self.__storage.items() return self.__storage.items()
class Security: # pylint: disable=too-few-public-methods class Security: # pylint: disable=too-few-public-methods
@ -2043,13 +2028,13 @@ class SSH: # pylint: disable=too-few-public-methods
self.__sock_map = {} self.__sock_map = {}
self.__block_size = 8 self.__block_size = 8
self.__state = 0 self.__state = 0
self.__header = [] # type: List[text_type] self.__header = [] # type: List[str]
self.__banner = None # type: Optional[SSH.Banner] self.__banner = None # type: Optional[SSH.Banner]
if host is None: if host is None:
raise ValueError('undefined host') raise ValueError('undefined host')
nport = utils.parse_int(port) nport = utils.parse_int(port)
if nport < 1 or nport > 65535: if nport < 1 or nport > 65535:
raise ValueError('invalid port: {0}'.format(port)) raise ValueError('invalid port: {}'.format(port))
self.__host = host self.__host = host
self.__port = nport self.__port = nport
if ipvo is not None: if ipvo is not None:
@ -2081,7 +2066,7 @@ class SSH: # pylint: disable=too-few-public-methods
if not check or socktype == socket.SOCK_STREAM: if not check or socktype == socket.SOCK_STREAM:
yield af, addr yield af, addr
except socket.error as e: except socket.error as e:
out.fail('[exception] {0}'.format(e)) out.fail('[exception] {}'.format(e))
sys.exit(1) sys.exit(1)
# Listens on a server socket and accepts one connection (used for # Listens on a server socket and accepts one connection (used for
@ -2154,15 +2139,15 @@ class SSH: # pylint: disable=too-few-public-methods
err = e err = e
self._close_socket(s) self._close_socket(s)
if err is None: if err is None:
errm = 'host {0} has no DNS records'.format(self.__host) errm = 'host {} has no DNS records'.format(self.__host)
else: else:
errt = (self.__host, self.__port, err) errt = (self.__host, self.__port, err)
errm = 'cannot connect to {0} port {1}: {2}'.format(*errt) errm = 'cannot connect to {} port {}: {}'.format(*errt)
out.fail('[exception] {0}'.format(errm)) out.fail('[exception] {}'.format(errm))
sys.exit(1) sys.exit(1)
def get_banner(self, sshv=2): def get_banner(self, sshv=2):
# type: (int) -> Tuple[Optional[SSH.Banner], List[text_type], Optional[str]] # type: (int) -> Tuple[Optional[SSH.Banner], List[str], Optional[str]]
if self.__sock is None: if self.__sock is None:
return self.__banner, self.__header, 'not connected' return self.__banner, self.__header, 'not connected'
banner = SSH_HEADER.format('1.5' if sshv == 1 else '2.0') banner = SSH_HEADER.format('1.5' if sshv == 1 else '2.0')
@ -2214,7 +2199,7 @@ class SSH: # pylint: disable=too-few-public-methods
return len(data), None return len(data), None
def send(self, data): def send(self, data):
# type: (binary_type) -> Tuple[int, Optional[str]] # type: (bytes) -> Tuple[int, Optional[str]]
if self.__sock is None: if self.__sock is None:
return -1, 'not connected' return -1, 'not connected'
try: try:
@ -2238,7 +2223,7 @@ class SSH: # pylint: disable=too-few-public-methods
raise SSH.Socket.InsufficientReadException(e) raise SSH.Socket.InsufficientReadException(e)
def read_packet(self, sshv=2): def read_packet(self, sshv=2):
# type: (int) -> Tuple[int, binary_type] # type: (int) -> Tuple[int, bytes]
try: try:
header = WriteBuf() header = WriteBuf()
self.ensure_read(4) self.ensure_read(4)
@ -2669,7 +2654,7 @@ class KexGroupExchange_SHA256(KexGroupExchange):
def output_algorithms(title, alg_db, alg_type, algorithms, unknown_algs, maxlen=0, alg_sizes=None): def output_algorithms(title, alg_db, alg_type, algorithms, unknown_algs, maxlen=0, alg_sizes=None):
# type: (str, Dict[str, Dict[str, List[List[Optional[str]]]]], str, List[text_type], int) -> None # type: (str, Dict[str, Dict[str, List[List[Optional[str]]]]], str, List[str], int) -> None
with OutputBuffer() as obuf: with OutputBuffer() as obuf:
for algorithm in algorithms: for algorithm in algorithms:
output_algorithm(alg_db, alg_type, algorithm, unknown_algs, maxlen, alg_sizes) output_algorithm(alg_db, alg_type, algorithm, unknown_algs, maxlen, alg_sizes)
@ -2680,7 +2665,7 @@ def output_algorithms(title, alg_db, alg_type, algorithms, unknown_algs, maxlen=
def output_algorithm(alg_db, alg_type, alg_name, unknown_algs, alg_max_len=0, alg_sizes=None): def output_algorithm(alg_db, alg_type, alg_name, unknown_algs, alg_max_len=0, alg_sizes=None):
# type: (Dict[str, Dict[str, List[List[Optional[str]]]]], str, text_type, int) -> None # type: (Dict[str, Dict[str, List[List[Optional[str]]]]], str, str, int) -> None
prefix = '(' + alg_type + ') ' prefix = '(' + alg_type + ') '
if alg_max_len == 0: if alg_max_len == 0:
alg_max_len = len(alg_name) alg_max_len = len(alg_name)
@ -2758,9 +2743,9 @@ def output_compatibility(algs, client_audit, for_server=True):
if v_from is None: if v_from is None:
continue continue
if v_till is None: if v_till is None:
comp_text.append('{0} {1}+'.format(ssh_prod, v_from)) comp_text.append('{} {}+'.format(ssh_prod, v_from))
elif v_from == v_till: elif v_from == v_till:
comp_text.append('{0} {1}'.format(ssh_prod, v_from)) comp_text.append('{} {}'.format(ssh_prod, v_from))
else: else:
software = SSH.Software(None, ssh_prod, v_from, None, None) software = SSH.Software(None, ssh_prod, v_from, None, None)
if software.compare_version(v_till) > 0: if software.compare_version(v_till) > 0:
@ -2797,10 +2782,10 @@ def output_security_sub(sub, software, client_audit, padlen):
out_func = out.warn out_func = out.warn
if cvss >= 8.0: if cvss >= 8.0:
out_func = out.fail out_func = out.fail
out_func('(cve) {0}{1} -- (CVSSv2: {2}) {3}'.format(name, p, cvss, descr)) out_func('(cve) {}{} -- (CVSSv2: {}) {}'.format(name, p, cvss, descr))
else: else:
descr = line[4] descr = line[4]
out.fail('(sec) {0}{1} -- {2}'.format(name, p, descr)) out.fail('(sec) {}{} -- {}'.format(name, p, descr))
def output_security(banner, client_audit, padlen): def output_security(banner, client_audit, padlen):
@ -2847,7 +2832,7 @@ def output_fingerprints(algs, sha256=True):
fpo = fp.sha256 if sha256 else fp.md5 fpo = fp.sha256 if sha256 else fp.md5
# p = '' if out.batch else ' ' * (padlen - len(name)) # p = '' if out.batch else ' ' * (padlen - len(name))
# out.good('(fin) {0}{1} -- {2} {3}'.format(name, p, bits, fpo)) # out.good('(fin) {0}{1} -- {2} {3}'.format(name, p, bits, fpo))
out.good('(fin) {0}: {1}'.format(name, fpo)) out.good('(fin) {}: {}'.format(name, fpo))
if len(obuf) > 0: if len(obuf) > 0:
out.head('# fingerprints') out.head('# fingerprints')
obuf.flush() obuf.flush()
@ -2913,15 +2898,15 @@ def output_recommendations(algs, software, padlen=0):
an, sg, fn = 'change', '!', out.fail an, sg, fn = 'change', '!', out.fail
ret = False ret = False
chg_additional_info = ' (increase modulus size to 2048 bits or larger)' chg_additional_info = ' (increase modulus size to 2048 bits or larger)'
b = '(SSH{0})'.format(sshv) if sshv == 1 else '' b = '(SSH{})'.format(sshv) if sshv == 1 else ''
fm = '(rec) {0}{1}{2}-- {3} algorithm to {4}{5} {6}' fm = '(rec) {0}{1}{2}-- {3} algorithm to {4}{5} {6}'
fn(fm.format(sg, name, p, alg_type, an, chg_additional_info, b)) fn(fm.format(sg, name, p, alg_type, an, chg_additional_info, b))
if len(obuf) > 0: if len(obuf) > 0:
if software is not None: if software is not None:
title = '(for {0})'.format(software.display(False)) title = '(for {})'.format(software.display(False))
else: else:
title = '' title = ''
out.head('# algorithm recommendations {0}'.format(title)) out.head('# algorithm recommendations {}'.format(title))
obuf.flush(True) # Sort the output so that it is always stable (needed for repeatable testing). obuf.flush(True) # Sort the output so that it is always stable (needed for repeatable testing).
out.sep() out.sep()
return ret return ret
@ -2945,17 +2930,17 @@ def output_info(software, client_audit, any_problems):
def output(banner, header, client_host=None, kex=None, pkm=None): def output(banner, header, client_host=None, kex=None, pkm=None):
# type: (Optional[SSH.Banner], List[text_type], Optional[SSH2.Kex], Optional[SSH1.PublicKeyMessage]) -> None # type: (Optional[SSH.Banner], List[str], Optional[SSH2.Kex], Optional[SSH1.PublicKeyMessage]) -> None
client_audit = client_host is not None # If set, this is a client audit. client_audit = client_host is not None # If set, this is a client audit.
sshv = 1 if pkm is not None else 2 sshv = 1 if pkm is not None else 2
algs = SSH.Algorithms(pkm, kex) algs = SSH.Algorithms(pkm, kex)
with OutputBuffer() as obuf: with OutputBuffer() as obuf:
if client_audit: if client_audit:
out.good('(gen) client IP: {0}'.format(client_host)) out.good('(gen) client IP: {}'.format(client_host))
if len(header) > 0: if len(header) > 0:
out.info('(gen) header: ' + '\n'.join(header)) out.info('(gen) header: ' + '\n'.join(header))
if banner is not None: if banner is not None:
out.good('(gen) banner: {0}'.format(banner)) out.good('(gen) banner: {}'.format(banner))
if not banner.valid_ascii: if not banner.valid_ascii:
# NOTE: RFC 4253, Section 4.2 # NOTE: RFC 4253, Section 4.2
out.warn('(gen) banner contains non-printable ASCII') out.warn('(gen) banner contains non-printable ASCII')
@ -2963,17 +2948,17 @@ def output(banner, header, client_host=None, kex=None, pkm=None):
out.fail('(gen) protocol SSH1 enabled') out.fail('(gen) protocol SSH1 enabled')
software = SSH.Software.parse(banner) software = SSH.Software.parse(banner)
if software is not None: if software is not None:
out.good('(gen) software: {0}'.format(software)) out.good('(gen) software: {}'.format(software))
else: else:
software = None software = None
output_compatibility(algs, client_audit) output_compatibility(algs, client_audit)
if kex is not None: if kex is not None:
compressions = [x for x in kex.server.compression if x != 'none'] compressions = [x for x in kex.server.compression if x != 'none']
if len(compressions) > 0: if len(compressions) > 0:
cmptxt = 'enabled ({0})'.format(', '.join(compressions)) cmptxt = 'enabled ({})'.format(', '.join(compressions))
else: else:
cmptxt = 'disabled' cmptxt = 'disabled'
out.good('(gen) compression: {0}'.format(cmptxt)) out.good('(gen) compression: {}'.format(cmptxt))
if len(obuf) > 0: if len(obuf) > 0:
out.head('# general') out.head('# general')
obuf.flush() obuf.flush()
@ -3013,43 +2998,41 @@ def output(banner, header, client_host=None, kex=None, pkm=None):
class Utils: class Utils:
@classmethod @classmethod
def _type_err(cls, v, target): def _type_err(cls, v, target):
# type: (Any, text_type) -> TypeError # type: (Any, str) -> TypeError
return TypeError('cannot convert {0} to {1}'.format(type(v), target)) return TypeError('cannot convert {} to {}'.format(type(v), target))
@classmethod @classmethod
def to_bytes(cls, v, enc='utf-8'): def to_bytes(cls, v, enc='utf-8'):
# type: (Union[binary_type, text_type], str) -> binary_type # type: (Union[bytes, str], str) -> bytes
if isinstance(v, binary_type): if isinstance(v, bytes):
return v return v
elif isinstance(v, text_type): elif isinstance(v, str):
return v.encode(enc) return v.encode(enc)
raise cls._type_err(v, 'bytes') raise cls._type_err(v, 'bytes')
@classmethod @classmethod
def to_utext(cls, v, enc='utf-8'): def to_utext(cls, v, enc='utf-8'):
# type: (Union[text_type, binary_type], str) -> text_type # type: (Union[str, bytes], str) -> str
if isinstance(v, text_type): if isinstance(v, str):
return v return v
elif isinstance(v, binary_type): elif isinstance(v, bytes):
return v.decode(enc) return v.decode(enc)
raise cls._type_err(v, 'unicode text') raise cls._type_err(v, 'unicode text')
@classmethod @classmethod
def to_ntext(cls, v, enc='utf-8'): def to_ntext(cls, v, enc='utf-8'):
# type: (Union[text_type, binary_type], str) -> str # type: (Union[str, bytes], str) -> str
if isinstance(v, str): if isinstance(v, str):
return v return v
elif isinstance(v, text_type): elif isinstance(v, bytes):
return v.encode(enc) # PY2 only return v.decode(enc)
elif isinstance(v, binary_type):
return v.decode(enc) # PY3 only
raise cls._type_err(v, 'native text') raise cls._type_err(v, 'native text')
@classmethod @classmethod
def _is_ascii(cls, v, char_filter=lambda x: x <= 127): def _is_ascii(cls, v, char_filter=lambda x: x <= 127):
# type: (Union[text_type, str], Callable[[int], bool]) -> bool # type: (str, Callable[[int], bool]) -> bool
r = False r = False
if isinstance(v, (text_type, str)): if isinstance(v, str):
for c in v: for c in v:
i = cls.ctoi(c) i = cls.ctoi(c)
if not char_filter(i): if not char_filter(i):
@ -3059,8 +3042,8 @@ class Utils:
@classmethod @classmethod
def _to_ascii(cls, v, char_filter=lambda x: x <= 127, errors='replace'): def _to_ascii(cls, v, char_filter=lambda x: x <= 127, errors='replace'):
# type: (Union[text_type, str], Callable[[int], bool], str) -> str # type: (str, Callable[[int], bool], str) -> str
if isinstance(v, (text_type, str)): if isinstance(v, str):
r = bytearray() r = bytearray()
for c in v: for c in v:
i = cls.ctoi(c) i = cls.ctoi(c)
@ -3075,22 +3058,22 @@ class Utils:
@classmethod @classmethod
def is_ascii(cls, v): def is_ascii(cls, v):
# type: (Union[text_type, str]) -> bool # type: (str) -> bool
return cls._is_ascii(v) return cls._is_ascii(v)
@classmethod @classmethod
def to_ascii(cls, v, errors='replace'): def to_ascii(cls, v, errors='replace'):
# type: (Union[text_type, str], str) -> str # type: (str, str) -> str
return cls._to_ascii(v, errors=errors) return cls._to_ascii(v, errors=errors)
@classmethod @classmethod
def is_print_ascii(cls, v): def is_print_ascii(cls, v):
# type: (Union[text_type, str]) -> bool # type: (str) -> bool
return cls._is_ascii(v, lambda x: 126 >= x >= 32) return cls._is_ascii(v, lambda x: 126 >= x >= 32)
@classmethod @classmethod
def to_print_ascii(cls, v, errors='replace'): def to_print_ascii(cls, v, errors='replace'):
# type: (Union[text_type, str], str) -> str # type: (str, str) -> str
return cls._to_ascii(v, lambda x: 126 >= x >= 32, errors) return cls._to_ascii(v, lambda x: 126 >= x >= 32, errors)
@classmethod @classmethod
@ -3110,8 +3093,8 @@ class Utils:
@classmethod @classmethod
def ctoi(cls, c): def ctoi(cls, c):
# type: (Union[text_type, str, int]) -> int # type: (Union[str, int]) -> int
if isinstance(c, (text_type, str)): if isinstance(c, str):
return ord(c[0]) return ord(c[0])
else: else:
return c return c
@ -3230,7 +3213,7 @@ def audit(aconf, sshv=None):
if err is None: if err is None:
err = '[exception] did not receive banner.' err = '[exception] did not receive banner.'
else: else:
err = '[exception] did not receive banner: {0}'.format(err) err = '[exception] did not receive banner: {}'.format(err)
if err is None: if err is None:
packet_type, payload = s.read_packet(sshv) packet_type, payload = s.read_packet(sshv)
if packet_type < 0: if packet_type < 0:
@ -3240,12 +3223,12 @@ def audit(aconf, sshv=None):
else: else:
payload_txt = u'empty' payload_txt = u'empty'
except UnicodeDecodeError: except UnicodeDecodeError:
payload_txt = u'"{0}"'.format(repr(payload).lstrip('b')[1:-1]) payload_txt = u'"{}"'.format(repr(payload).lstrip('b')[1:-1])
if payload_txt == u'Protocol major versions differ.': if payload_txt == u'Protocol major versions differ.':
if sshv == 2 and aconf.ssh1: if sshv == 2 and aconf.ssh1:
audit(aconf, 1) audit(aconf, 1)
return return
err = '[exception] error reading packet ({0})'.format(payload_txt) err = '[exception] error reading packet ({})'.format(payload_txt)
else: else:
err_pair = None err_pair = None
if sshv == 1 and packet_type != SSH.Protocol.SMSG_PUBLIC_KEY: if sshv == 1 and packet_type != SSH.Protocol.SMSG_PUBLIC_KEY:

23
test/conftest.py
View File

@ -1,5 +1,3 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import os import os
import io import io
import sys import sys
@ -7,13 +5,6 @@ import socket
import pytest 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') @pytest.fixture(scope='module')
def ssh_audit(): def ssh_audit():
__rdir = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..') __rdir = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..')
@ -24,7 +15,7 @@ def ssh_audit():
# pylint: disable=attribute-defined-outside-init # pylint: disable=attribute-defined-outside-init
class _OutputSpy(list): class _OutputSpy(list):
def begin(self): def begin(self):
self.__out = StringIO() self.__out = io.StringIO()
self.__old_stdout = sys.stdout self.__old_stdout = sys.stdout
sys.stdout = self.__out sys.stdout = self.__out
@ -40,7 +31,7 @@ def output_spy():
return _OutputSpy() return _OutputSpy()
class _VirtualGlobalSocket(object): class _VirtualGlobalSocket:
def __init__(self, vsocket): def __init__(self, vsocket):
self.vsocket = vsocket self.vsocket = vsocket
self.addrinfodata = {} self.addrinfodata = {}
@ -59,7 +50,7 @@ class _VirtualGlobalSocket(object):
return self.vsocket return self.vsocket
def getaddrinfo(self, host, port, family=0, socktype=0, proto=0, flags=0): def getaddrinfo(self, host, port, family=0, socktype=0, proto=0, flags=0):
key = '{0}#{1}'.format(host, port) key = '{}#{}'.format(host, port)
if key in self.addrinfodata: if key in self.addrinfodata:
data = self.addrinfodata[key] data = self.addrinfodata[key]
if isinstance(data, Exception): if isinstance(data, Exception):
@ -75,7 +66,7 @@ class _VirtualGlobalSocket(object):
return [] return []
class _VirtualSocket(object): class _VirtualSocket:
def __init__(self): def __init__(self):
self.sock_address = ('127.0.0.1', 0) self.sock_address = ('127.0.0.1', 0)
self.peer_address = None self.peer_address = None
@ -108,7 +99,7 @@ class _VirtualSocket(object):
def getpeername(self): def getpeername(self):
if self.peer_address is None or not self._connected: if self.peer_address is None or not self._connected:
raise socket.error(57, 'Socket is not connected') raise OSError(57, 'Socket is not connected')
return self.peer_address return self.peer_address
def getsockname(self): def getsockname(self):
@ -131,7 +122,7 @@ class _VirtualSocket(object):
def recv(self, bufsize, flags=0): def recv(self, bufsize, flags=0):
# pylint: disable=unused-argument # pylint: disable=unused-argument
if not self._connected: if not self._connected:
raise socket.error(54, 'Connection reset by peer') raise OSError(54, 'Connection reset by peer')
if not len(self.rdata) > 0: if not len(self.rdata) > 0:
return b'' return b''
data = self.rdata.pop(0) data = self.rdata.pop(0)
@ -141,7 +132,7 @@ class _VirtualSocket(object):
def send(self, data): def send(self, data):
if self.peer_address is None or not self._connected: if self.peer_address is None or not self._connected:
raise socket.error(32, 'Broken pipe') raise OSError(32, 'Broken pipe')
self._check_err('send') self._check_err('send')
self.sdata.append(data) self.sdata.append(data)

4
test/test_auditconf.py
View File

@ -1,10 +1,8 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pytest import pytest
# pylint: disable=attribute-defined-outside-init # pylint: disable=attribute-defined-outside-init
class TestAuditConf(object): class TestAuditConf:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def init(self, ssh_audit): def init(self, ssh_audit):
self.AuditConf = ssh_audit.AuditConf self.AuditConf = ssh_audit.AuditConf

4
test/test_banner.py
View File

@ -1,10 +1,8 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pytest import pytest
# pylint: disable=line-too-long,attribute-defined-outside-init # pylint: disable=line-too-long,attribute-defined-outside-init
class TestBanner(object): class TestBanner:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def init(self, ssh_audit): def init(self, ssh_audit):
self.ssh = ssh_audit.SSH self.ssh = ssh_audit.SSH

8
test/test_buffer.py
View File

@ -1,11 +1,9 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import re import re
import pytest import pytest
# pylint: disable=attribute-defined-outside-init,bad-whitespace # pylint: disable=attribute-defined-outside-init,bad-whitespace
class TestBuffer(object): class TestBuffer:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def init(self, ssh_audit): def init(self, ssh_audit):
self.rbuf = ssh_audit.ReadBuf self.rbuf = ssh_audit.ReadBuf
@ -61,7 +59,7 @@ class TestBuffer(object):
def test_string(self): def test_string(self):
w = lambda x: self.wbuf().write_string(x).write_flush() # noqa w = lambda x: self.wbuf().write_string(x).write_flush() # noqa
r = lambda x: self.rbuf(x).read_string() # noqa r = lambda x: self.rbuf(x).read_string() # noqa
tc = [(u'abc1', '00 00 00 04 61 62 63 31'), tc = [('abc1', '00 00 00 04 61 62 63 31'),
(b'abc2', '00 00 00 04 61 62 63 32')] (b'abc2', '00 00 00 04 61 62 63 32')]
for p in tc: for p in tc:
v = p[0] v = p[0]
@ -87,7 +85,7 @@ class TestBuffer(object):
def test_line(self): def test_line(self):
w = lambda x: self.wbuf().write_line(x).write_flush() # noqa w = lambda x: self.wbuf().write_line(x).write_flush() # noqa
r = lambda x: self.rbuf(x).read_line() # 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')] tc = [('example line', '65 78 61 6d 70 6c 65 20 6c 69 6e 65 0d 0a')]
for p in tc: for p in tc:
assert w(p[0]) == self._b(p[1]) assert w(p[0]) == self._b(p[1])
assert r(self._b(p[1])) == p[0] assert r(self._b(p[1])) == p[0]

4
test/test_errors.py
View File

@ -1,12 +1,10 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import socket import socket
import errno import errno
import pytest import pytest
# pylint: disable=attribute-defined-outside-init # pylint: disable=attribute-defined-outside-init
class TestErrors(object): class TestErrors:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def init(self, ssh_audit): def init(self, ssh_audit):
self.AuditConf = ssh_audit.AuditConf self.AuditConf = ssh_audit.AuditConf

35
test/test_output.py
View File

@ -1,11 +1,8 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import print_function
import pytest import pytest
# pylint: disable=attribute-defined-outside-init # pylint: disable=attribute-defined-outside-init
class TestOutput(object): class TestOutput:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def init(self, ssh_audit): def init(self, ssh_audit):
self.Output = ssh_audit.Output self.Output = ssh_audit.Output
@ -25,17 +22,17 @@ class TestOutput(object):
def test_output_buffer_no_flush(self, output_spy): def test_output_buffer_no_flush(self, output_spy):
output_spy.begin() output_spy.begin()
with self.OutputBuffer(): with self.OutputBuffer():
print(u'abc') print('abc')
assert output_spy.flush() == [] assert output_spy.flush() == []
def test_output_buffer_flush(self, output_spy): def test_output_buffer_flush(self, output_spy):
output_spy.begin() output_spy.begin()
with self.OutputBuffer() as obuf: with self.OutputBuffer() as obuf:
print(u'abc') print('abc')
print() print()
print(u'def') print('def')
obuf.flush() obuf.flush()
assert output_spy.flush() == [u'abc', u'', u'def'] assert output_spy.flush() == ['abc', '', 'def']
def test_output_defaults(self): def test_output_defaults(self):
out = self.Output() out = self.Output()
@ -50,38 +47,38 @@ class TestOutput(object):
out.use_colors = False out.use_colors = False
output_spy.begin() output_spy.begin()
out.info('info color') out.info('info color')
assert output_spy.flush() == [u'info color'] assert output_spy.flush() == ['info color']
output_spy.begin() output_spy.begin()
out.head('head color') out.head('head color')
assert output_spy.flush() == [u'head color'] assert output_spy.flush() == ['head color']
output_spy.begin() output_spy.begin()
out.good('good color') out.good('good color')
assert output_spy.flush() == [u'good color'] assert output_spy.flush() == ['good color']
output_spy.begin() output_spy.begin()
out.warn('warn color') out.warn('warn color')
assert output_spy.flush() == [u'warn color'] assert output_spy.flush() == ['warn color']
output_spy.begin() output_spy.begin()
out.fail('fail color') out.fail('fail color')
assert output_spy.flush() == [u'fail color'] assert output_spy.flush() == ['fail color']
if not out.colors_supported: if not out.colors_supported:
return return
# test with colors # test with colors
out.use_colors = True out.use_colors = True
output_spy.begin() output_spy.begin()
out.info('info color') out.info('info color')
assert output_spy.flush() == [u'info color'] assert output_spy.flush() == ['info color']
output_spy.begin() output_spy.begin()
out.head('head color') out.head('head color')
assert output_spy.flush() == [u'\x1b[0;36mhead color\x1b[0m'] assert output_spy.flush() == ['\x1b[0;36mhead color\x1b[0m']
output_spy.begin() output_spy.begin()
out.good('good color') out.good('good color')
assert output_spy.flush() == [u'\x1b[0;32mgood color\x1b[0m'] assert output_spy.flush() == ['\x1b[0;32mgood color\x1b[0m']
output_spy.begin() output_spy.begin()
out.warn('warn color') out.warn('warn color')
assert output_spy.flush() == [u'\x1b[0;33mwarn color\x1b[0m'] assert output_spy.flush() == ['\x1b[0;33mwarn color\x1b[0m']
output_spy.begin() output_spy.begin()
out.fail('fail color') out.fail('fail color')
assert output_spy.flush() == [u'\x1b[0;31mfail color\x1b[0m'] assert output_spy.flush() == ['\x1b[0;31mfail color\x1b[0m']
def test_output_sep(self, output_spy): def test_output_sep(self, output_spy):
out = self.Output() out = self.Output()
@ -89,7 +86,7 @@ class TestOutput(object):
out.sep() out.sep()
out.sep() out.sep()
out.sep() out.sep()
assert output_spy.flush() == [u'', u'', u''] assert output_spy.flush() == ['', '', '']
def test_output_levels(self): def test_output_levels(self):
out = self.Output() out = self.Output()

4
test/test_resolve.py
View File

@ -1,11 +1,9 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import socket import socket
import pytest import pytest
# pylint: disable=attribute-defined-outside-init,protected-access # pylint: disable=attribute-defined-outside-init,protected-access
class TestResolve(object): class TestResolve:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def init(self, ssh_audit): def init(self, ssh_audit):
self.AuditConf = ssh_audit.AuditConf self.AuditConf = ssh_audit.AuditConf

4
test/test_socket.py
View File

@ -1,10 +1,8 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pytest import pytest
# pylint: disable=attribute-defined-outside-init # pylint: disable=attribute-defined-outside-init
class TestSocket(object): class TestSocket:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def init(self, ssh_audit): def init(self, ssh_audit):
self.ssh = ssh_audit.SSH self.ssh = ssh_audit.SSH

4
test/test_software.py
View File

@ -1,10 +1,8 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pytest import pytest
# pylint: disable=line-too-long,attribute-defined-outside-init # pylint: disable=line-too-long,attribute-defined-outside-init
class TestSoftware(object): class TestSoftware:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def init(self, ssh_audit): def init(self, ssh_audit):
self.ssh = ssh_audit.SSH self.ssh = ssh_audit.SSH

4
test/test_ssh1.py
View File

@ -1,11 +1,9 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import struct import struct
import pytest import pytest
# pylint: disable=line-too-long,attribute-defined-outside-init # pylint: disable=line-too-long,attribute-defined-outside-init
class TestSSH1(object): class TestSSH1:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def init(self, ssh_audit): def init(self, ssh_audit):
self.ssh = ssh_audit.SSH self.ssh = ssh_audit.SSH

44
test/test_ssh2.py
View File

@ -1,12 +1,10 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import os import os
import struct import struct
import pytest import pytest
# pylint: disable=line-too-long,attribute-defined-outside-init # pylint: disable=line-too-long,attribute-defined-outside-init
class TestSSH2(object): class TestSSH2:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def init(self, ssh_audit): def init(self, ssh_audit):
self.ssh = ssh_audit.SSH self.ssh = ssh_audit.SSH
@ -38,16 +36,16 @@ class TestSSH2(object):
def _kex_payload(self): def _kex_payload(self):
w = self.wbuf() w = self.wbuf()
w.write(b'\x00\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff') w.write(b'\x00\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff')
w.write_list([u'bogus_kex1', u'bogus_kex2']) # We use a bogus kex, otherwise the host key tests will kick off and fail. 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([u'ssh-rsa', u'rsa-sha2-512', u'rsa-sha2-256', u'ssh-ed25519']) w.write_list(['ssh-rsa', 'rsa-sha2-512', 'rsa-sha2-256', '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(['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([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(['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([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(['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([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(['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([u'none', u'zlib@openssh.com']) w.write_list(['none', 'zlib@openssh.com'])
w.write_list([u'none', u'zlib@openssh.com']) w.write_list(['none', 'zlib@openssh.com'])
w.write_list([u'']) w.write_list([''])
w.write_list([u'']) w.write_list([''])
w.write_byte(False) w.write_byte(False)
w.write_int(0) w.write_int(0)
return w.write_flush() return w.write_flush()
@ -56,18 +54,18 @@ class TestSSH2(object):
kex = self.ssh2.Kex.parse(self._kex_payload()) kex = self.ssh2.Kex.parse(self._kex_payload())
assert kex is not None 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.cookie == b'\x00\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff'
assert kex.kex_algorithms == [u'bogus_kex1', u'bogus_kex2'] assert kex.kex_algorithms == ['bogus_kex1', 'bogus_kex2']
assert kex.key_algorithms == [u'ssh-rsa', u'rsa-sha2-512', u'rsa-sha2-256', u'ssh-ed25519'] assert kex.key_algorithms == ['ssh-rsa', 'rsa-sha2-512', 'rsa-sha2-256', 'ssh-ed25519']
assert kex.client is not None assert kex.client is not None
assert kex.server 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.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 == [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 == ['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 == [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.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 == [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 == ['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 == [u'none', u'zlib@openssh.com'] assert kex.client.compression == ['none', 'zlib@openssh.com']
assert kex.server.compression == [u'none', u'zlib@openssh.com'] assert kex.server.compression == ['none', 'zlib@openssh.com']
assert kex.client.languages == [u''] assert kex.client.languages == ['']
assert kex.server.languages == [u''] assert kex.server.languages == ['']
assert kex.follows is False assert kex.follows is False
assert kex.unused == 0 assert kex.unused == 0

4
test/test_ssh_algorithm.py
View File

@ -1,10 +1,8 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pytest import pytest
# pylint: disable=attribute-defined-outside-init # pylint: disable=attribute-defined-outside-init
class TestSSHAlgorithm(object): class TestSSHAlgorithm:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def init(self, ssh_audit): def init(self, ssh_audit):
self.ssh = ssh_audit.SSH self.ssh = ssh_audit.SSH

160
test/test_utils.py
View File

@ -1,202 +1,60 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys
import pytest import pytest
# pylint: disable=attribute-defined-outside-init # pylint: disable=attribute-defined-outside-init
class TestUtils(object): class TestUtils:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def init(self, ssh_audit): def init(self, ssh_audit):
self.utils = ssh_audit.Utils self.utils = ssh_audit.Utils
self.PY3 = sys.version_info >= (3,)
def test_to_bytes_py2(self): def test_to_bytes(self):
if self.PY3:
return
# binary_type (native str, bytes as str)
assert self.utils.to_bytes('fran\xc3\xa7ais') == 'fran\xc3\xa7ais'
assert self.utils.to_bytes(b'fran\xc3\xa7ais') == 'fran\xc3\xa7ais'
# text_type (unicode)
assert self.utils.to_bytes(u'fran\xe7ais') == 'fran\xc3\xa7ais'
# other
with pytest.raises(TypeError):
self.utils.to_bytes(123)
def test_to_bytes_py3(self):
if not self.PY3:
return
# binary_type (bytes)
assert self.utils.to_bytes(b'fran\xc3\xa7ais') == b'fran\xc3\xa7ais' assert self.utils.to_bytes(b'fran\xc3\xa7ais') == b'fran\xc3\xa7ais'
# text_type (native str as unicode, unicode)
assert self.utils.to_bytes('fran\xe7ais') == b'fran\xc3\xa7ais' assert self.utils.to_bytes('fran\xe7ais') == b'fran\xc3\xa7ais'
assert self.utils.to_bytes(u'fran\xe7ais') == b'fran\xc3\xa7ais'
# other # other
with pytest.raises(TypeError): with pytest.raises(TypeError):
self.utils.to_bytes(123) self.utils.to_bytes(123)
def test_to_utext_py2(self): def test_to_utext(self):
if self.PY3: assert self.utils.to_utext(b'fran\xc3\xa7ais') == 'fran\xe7ais'
return
# binary_type (native str, bytes as str)
assert self.utils.to_utext('fran\xc3\xa7ais') == u'fran\xe7ais'
assert self.utils.to_utext(b'fran\xc3\xa7ais') == u'fran\xe7ais'
# text_type (unicode)
assert self.utils.to_utext(u'fran\xe7ais') == u'fran\xe7ais'
# other
with pytest.raises(TypeError):
self.utils.to_utext(123)
def test_to_utext_py3(self):
if not self.PY3:
return
# binary_type (bytes)
assert self.utils.to_utext(b'fran\xc3\xa7ais') == u'fran\xe7ais'
# text_type (native str as unicode, unicode)
assert self.utils.to_utext('fran\xe7ais') == 'fran\xe7ais' assert self.utils.to_utext('fran\xe7ais') == 'fran\xe7ais'
assert self.utils.to_utext(u'fran\xe7ais') == u'fran\xe7ais'
# other # other
with pytest.raises(TypeError): with pytest.raises(TypeError):
self.utils.to_utext(123) self.utils.to_utext(123)
def test_to_ntext_py2(self): def test_to_ntext(self):
if self.PY3:
return
# str (native str, bytes as str)
assert self.utils.to_ntext('fran\xc3\xa7ais') == 'fran\xc3\xa7ais' assert self.utils.to_ntext('fran\xc3\xa7ais') == 'fran\xc3\xa7ais'
assert self.utils.to_ntext(b'fran\xc3\xa7ais') == 'fran\xc3\xa7ais'
# text_type (unicode)
assert self.utils.to_ntext(u'fran\xe7ais') == 'fran\xc3\xa7ais'
# other
with pytest.raises(TypeError):
self.utils.to_ntext(123)
def test_to_ntext_py3(self):
if not self.PY3:
return
# str (native str)
assert self.utils.to_ntext('fran\xc3\xa7ais') == 'fran\xc3\xa7ais'
assert self.utils.to_ntext(u'fran\xe7ais') == 'fran\xe7ais'
# binary_type (bytes)
assert self.utils.to_ntext(b'fran\xc3\xa7ais') == 'fran\xe7ais' assert self.utils.to_ntext(b'fran\xc3\xa7ais') == 'fran\xe7ais'
# other # other
with pytest.raises(TypeError): with pytest.raises(TypeError):
self.utils.to_ntext(123) self.utils.to_ntext(123)
def test_is_ascii_py2(self): def test_is_ascii(self):
if self.PY3:
return
# text_type (unicode)
assert self.utils.is_ascii(u'francais') is True
assert self.utils.is_ascii(u'fran\xe7ais') is False
# str
assert self.utils.is_ascii('francais') is True assert self.utils.is_ascii('francais') is True
assert self.utils.is_ascii('fran\xc3\xa7ais') is False
# other
assert self.utils.is_ascii(123) is False
def test_is_ascii_py3(self):
if not self.PY3:
return
# text_type (str)
assert self.utils.is_ascii('francais') is True
assert self.utils.is_ascii(u'francais') is True
assert self.utils.is_ascii('fran\xe7ais') is False assert self.utils.is_ascii('fran\xe7ais') is False
assert self.utils.is_ascii(u'fran\xe7ais') is False
# other # other
assert self.utils.is_ascii(123) is False assert self.utils.is_ascii(123) is False
def test_to_ascii_py2(self): def test_to_ascii(self):
if self.PY3:
return
# text_type (unicode)
assert self.utils.to_ascii(u'francais') == 'francais'
assert self.utils.to_ascii(u'fran\xe7ais') == 'fran?ais'
assert self.utils.to_ascii(u'fran\xe7ais', 'ignore') == 'franais'
# str
assert self.utils.to_ascii('francais') == 'francais' assert self.utils.to_ascii('francais') == 'francais'
assert self.utils.to_ascii('fran\xc3\xa7ais') == 'fran??ais'
assert self.utils.to_ascii('fran\xc3\xa7ais', 'ignore') == 'franais'
with pytest.raises(TypeError):
self.utils.to_ascii(123)
def test_to_ascii_py3(self):
if not self.PY3:
return
# text_type (str)
assert self.utils.to_ascii('francais') == 'francais'
assert self.utils.to_ascii(u'francais') == 'francais'
assert self.utils.to_ascii('fran\xe7ais') == 'fran?ais' assert self.utils.to_ascii('fran\xe7ais') == 'fran?ais'
assert self.utils.to_ascii('fran\xe7ais', 'ignore') == 'franais' assert self.utils.to_ascii('fran\xe7ais', 'ignore') == 'franais'
assert self.utils.to_ascii(u'fran\xe7ais') == 'fran?ais'
assert self.utils.to_ascii(u'fran\xe7ais', 'ignore') == 'franais'
with pytest.raises(TypeError): with pytest.raises(TypeError):
self.utils.to_ascii(123) self.utils.to_ascii(123)
def test_is_print_ascii_py2(self): def test_is_print_ascii(self):
if self.PY3:
return
# text_type (unicode)
assert self.utils.is_print_ascii(u'francais') is True
assert self.utils.is_print_ascii(u'francais\n') is False
assert self.utils.is_print_ascii(u'fran\xe7ais') is False
assert self.utils.is_print_ascii(u'fran\xe7ais\n') is False
# str
assert self.utils.is_print_ascii('francais') is True assert self.utils.is_print_ascii('francais') is True
assert self.utils.is_print_ascii('francais\n') is False assert self.utils.is_print_ascii('francais\n') is False
assert self.utils.is_print_ascii('fran\xc3\xa7ais') is False
# other
assert self.utils.is_print_ascii(123) is False
def test_is_print_ascii_py3(self):
if not self.PY3:
return
# text_type (str)
assert self.utils.is_print_ascii('francais') is True
assert self.utils.is_print_ascii('francais\n') is False
assert self.utils.is_print_ascii(u'francais') is True
assert self.utils.is_print_ascii(u'francais\n') is False
assert self.utils.is_print_ascii('fran\xe7ais') is False assert self.utils.is_print_ascii('fran\xe7ais') is False
assert self.utils.is_print_ascii(u'fran\xe7ais') is False
# other # other
assert self.utils.is_print_ascii(123) is False assert self.utils.is_print_ascii(123) is False
def test_to_print_ascii_py2(self): def test_to_print_ascii(self):
if self.PY3:
return
# text_type (unicode)
assert self.utils.to_print_ascii(u'francais') == 'francais'
assert self.utils.to_print_ascii(u'francais\n') == 'francais?'
assert self.utils.to_print_ascii(u'fran\xe7ais') == 'fran?ais'
assert self.utils.to_print_ascii(u'fran\xe7ais\n') == 'fran?ais?'
assert self.utils.to_print_ascii(u'fran\xe7ais', 'ignore') == 'franais'
assert self.utils.to_print_ascii(u'fran\xe7ais\n', 'ignore') == 'franais'
# str
assert self.utils.to_print_ascii('francais') == 'francais' assert self.utils.to_print_ascii('francais') == 'francais'
assert self.utils.to_print_ascii('francais\n') == 'francais?' assert self.utils.to_print_ascii('francais\n') == 'francais?'
assert self.utils.to_print_ascii('fran\xc3\xa7ais') == 'fran??ais'
assert self.utils.to_print_ascii('fran\xc3\xa7ais\n') == 'fran??ais?'
assert self.utils.to_print_ascii('fran\xc3\xa7ais', 'ignore') == 'franais'
assert self.utils.to_print_ascii('fran\xc3\xa7ais\n', 'ignore') == 'franais'
with pytest.raises(TypeError):
self.utils.to_print_ascii(123)
def test_to_print_ascii_py3(self):
if not self.PY3:
return
# text_type (str)
assert self.utils.to_print_ascii('francais') == 'francais'
assert self.utils.to_print_ascii('francais\n') == 'francais?'
assert self.utils.to_print_ascii(u'francais') == 'francais'
assert self.utils.to_print_ascii(u'francais\n') == 'francais?'
assert self.utils.to_print_ascii('fran\xe7ais') == 'fran?ais' assert self.utils.to_print_ascii('fran\xe7ais') == 'fran?ais'
assert self.utils.to_print_ascii('fran\xe7ais\n') == 'fran?ais?' assert self.utils.to_print_ascii('fran\xe7ais\n') == 'fran?ais?'
assert self.utils.to_print_ascii('fran\xe7ais', 'ignore') == 'franais' assert self.utils.to_print_ascii('fran\xe7ais', 'ignore') == 'franais'
assert self.utils.to_print_ascii('fran\xe7ais\n', 'ignore') == 'franais' assert self.utils.to_print_ascii('fran\xe7ais\n', 'ignore') == 'franais'
assert self.utils.to_print_ascii(u'fran\xe7ais') == 'fran?ais'
assert self.utils.to_print_ascii(u'fran\xe7ais\n') == 'fran?ais?'
assert self.utils.to_print_ascii(u'fran\xe7ais', 'ignore') == 'franais'
assert self.utils.to_print_ascii(u'fran\xe7ais\n', 'ignore') == 'franais'
with pytest.raises(TypeError): with pytest.raises(TypeError):
self.utils.to_print_ascii(123) self.utils.to_print_ascii(123)

48
test/test_version_compare.py
View File

@ -1,24 +1,22 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pytest import pytest
# pylint: disable=attribute-defined-outside-init # pylint: disable=attribute-defined-outside-init
class TestVersionCompare(object): class TestVersionCompare:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def init(self, ssh_audit): def init(self, ssh_audit):
self.ssh = ssh_audit.SSH self.ssh = ssh_audit.SSH
def get_dropbear_software(self, v): def get_dropbear_software(self, v):
b = self.ssh.Banner.parse('SSH-2.0-dropbear_{0}'.format(v)) b = self.ssh.Banner.parse('SSH-2.0-dropbear_{}'.format(v))
return self.ssh.Software.parse(b) return self.ssh.Software.parse(b)
def get_openssh_software(self, v): def get_openssh_software(self, v):
b = self.ssh.Banner.parse('SSH-2.0-OpenSSH_{0}'.format(v)) b = self.ssh.Banner.parse('SSH-2.0-OpenSSH_{}'.format(v))
return self.ssh.Software.parse(b) return self.ssh.Software.parse(b)
def get_libssh_software(self, v): def get_libssh_software(self, v):
b = self.ssh.Banner.parse('SSH-2.0-libssh-{0}'.format(v)) b = self.ssh.Banner.parse('SSH-2.0-libssh-{}'.format(v))
return self.ssh.Software.parse(b) return self.ssh.Software.parse(b)
def test_dropbear_compare_version_pre_years(self): def test_dropbear_compare_version_pre_years(self):
@ -84,27 +82,27 @@ class TestVersionCompare(object):
def test_dropbear_compare_version_sequential(self): def test_dropbear_compare_version_sequential(self):
versions = [] versions = []
for i in range(28, 44): for i in range(28, 44):
versions.append('0.{0}'.format(i)) versions.append('0.{}'.format(i))
for i in range(1, 5): for i in range(1, 5):
versions.append('0.44test{0}'.format(i)) versions.append('0.44test{}'.format(i))
for i in range(44, 49): for i in range(44, 49):
versions.append('0.{0}'.format(i)) versions.append('0.{}'.format(i))
versions.append('0.48.1') versions.append('0.48.1')
for i in range(49, 54): for i in range(49, 54):
versions.append('0.{0}'.format(i)) versions.append('0.{}'.format(i))
versions.append('0.53.1') versions.append('0.53.1')
for v in ['2011.54', '2012.55']: for v in ['2011.54', '2012.55']:
versions.append(v) versions.append(v)
for i in range(56, 61): for i in range(56, 61):
versions.append('2013.{0}'.format(i)) versions.append('2013.{}'.format(i))
for v in ['2013.61test', '2013.62']: for v in ['2013.61test', '2013.62']:
versions.append(v) versions.append(v)
for i in range(63, 67): for i in range(63, 67):
versions.append('2014.{0}'.format(i)) versions.append('2014.{}'.format(i))
for i in range(67, 72): for i in range(67, 72):
versions.append('2015.{0}'.format(i)) versions.append('2015.{}'.format(i))
for i in range(72, 75): for i in range(72, 75):
versions.append('2016.{0}'.format(i)) versions.append('2016.{}'.format(i))
length = len(versions) length = len(versions)
for i in range(length): for i in range(length):
v = versions[i] v = versions[i]
@ -151,19 +149,19 @@ class TestVersionCompare(object):
for v in ['3.0', '3.0.1', '3.0.2', '3.1', '3.2.2', '3.2.3']: for v in ['3.0', '3.0.1', '3.0.2', '3.1', '3.2.2', '3.2.3']:
versions.append(v) versions.append(v)
for i in range(3, 7): for i in range(3, 7):
versions.append('3.{0}'.format(i)) versions.append('3.{}'.format(i))
for v in ['3.6.1', '3.7.0', '3.7.1']: for v in ['3.6.1', '3.7.0', '3.7.1']:
versions.append(v) versions.append(v)
for i in range(8, 10): for i in range(8, 10):
versions.append('3.{0}'.format(i)) versions.append('3.{}'.format(i))
for i in range(0, 10): for i in range(0, 10):
versions.append('4.{0}'.format(i)) versions.append('4.{}'.format(i))
for i in range(0, 10): for i in range(0, 10):
versions.append('5.{0}'.format(i)) versions.append('5.{}'.format(i))
for i in range(0, 10): for i in range(0, 10):
versions.append('6.{0}'.format(i)) versions.append('6.{}'.format(i))
for i in range(0, 4): for i in range(0, 4):
versions.append('7.{0}'.format(i)) versions.append('7.{}'.format(i))
length = len(versions) length = len(versions)
for i in range(length): for i in range(length):
v = versions[i] v = versions[i]
@ -193,15 +191,15 @@ class TestVersionCompare(object):
for v in ['0.2', '0.3']: for v in ['0.2', '0.3']:
versions.append(v) versions.append(v)
for i in range(1, 5): for i in range(1, 5):
versions.append('0.3.{0}'.format(i)) versions.append('0.3.{}'.format(i))
for i in range(0, 9): for i in range(0, 9):
versions.append('0.4.{0}'.format(i)) versions.append('0.4.{}'.format(i))
for i in range(0, 6): for i in range(0, 6):
versions.append('0.5.{0}'.format(i)) versions.append('0.5.{}'.format(i))
for i in range(0, 6): for i in range(0, 6):
versions.append('0.6.{0}'.format(i)) versions.append('0.6.{}'.format(i))
for i in range(0, 5): for i in range(0, 5):
versions.append('0.7.{0}'.format(i)) versions.append('0.7.{}'.format(i))
length = len(versions) length = len(versions)
for i in range(length): for i in range(length):
v = versions[i] v = versions[i]

7
tox.ini
View File

@ -57,13 +57,6 @@ commands =
--config-file {toxinidir}/tox.ini \ --config-file {toxinidir}/tox.ini \
--html-report {env:MYPYHTML}.py3.{envname} \ --html-report {env:MYPYHTML}.py3.{envname} \
{posargs:{env:SSHAUDIT}} {posargs:{env:SSHAUDIT}}
-mypy \
-2 \
--no-warn-incomplete-stub \
--show-error-context \
--config-file {toxinidir}/tox.ini \
--html-report {env:MYPYHTML}.py2.{envname} \
{posargs:{env:SSHAUDIT}}
[testenv:pylint] [testenv:pylint]
deps = deps =