ssh-audit/test/conftest.py

157 lines
4.3 KiB
Python
Raw Normal View History

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import os
import io
import sys
import socket
import pytest
2016-10-13 16:55:59 +02:00
if sys.version_info[0] == 2:
2020-06-13 03:01:10 +02:00
import StringIO # pylint: disable=import-error
StringIO = StringIO.StringIO
2016-10-13 16:55:59 +02:00
else:
2020-06-13 03:01:10 +02:00
StringIO = io.StringIO
2016-10-13 16:55:59 +02:00
@pytest.fixture(scope='module')
def ssh_audit():
2020-06-13 03:01:10 +02:00
__rdir = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..')
sys.path.append(os.path.abspath(__rdir))
return __import__('ssh-audit')
2016-10-13 16:55:59 +02:00
# pylint: disable=attribute-defined-outside-init
2016-10-13 16:55:59 +02:00
class _OutputSpy(list):
2020-06-13 03:01:10 +02:00
def begin(self):
self.__out = StringIO()
self.__old_stdout = sys.stdout
sys.stdout = self.__out
Flake8 fixes (#35) * Apply Flake8 also on `setup.py` modified: tox.ini * Fix W605 - invalid escape syntax modified: packages/setup.py modified: tox.ini * Update comment about Flake8: W504 W503 and W504 are mutual exclusive - so we have to keep one of them. modified: tox.ini * Fix F841 - variable assigned but never used modified: ssh-audit.py modified: tox.ini * Fix E741 - ambiguous variable name 'l' modified: ssh-audit.py modified: tox.ini * Fix E712 - comparison to False should be 'if cond is False' ... and not 'if conf == False'. modified: ssh-audit.py modified: tox.ini * Fix E711 - comparison to None should be 'if cond is not None' ... and not 'if cond != None'. modified: ssh-audit.py modified: tox.ini * Fix E305 - expected 2 blank lines ... after class or function definition, found 1. modified: ssh-audit.py modified: tox.ini * Fix E303 - too many blank lines modified: ssh-audit.py modified: tox.ini * Fix E303 - too many blank lines modified: ssh-audit.py modified: tox.ini * Fix E301 - expected 1 blank line, found 0 No code change necessary, probably fixed by another commit. modified: tox.ini * Fix E265 - block comment should start with '# ' There is lots of commented out code, which usually should be just deleted. I will keep it for now, as I am not yet very familiar with the code base. modified: ssh-audit.py modified: tox.ini * Fix E261 - at least two spaces before inline comment modified: ssh-audit.py modified: tox.ini * Fix E251 - unexpected spaces around keyword / parameter equals modified: packages/setup.py modified: tox.ini * Fix E231 - missing whitespace after ',' No code change necessary, probably fixed by previous commit. modified: tox.ini * Fix E226 - missing whitespace around arithmetic operator modified: ssh-audit.py modified: tox.ini * Fix W293 - blank line contains whitespace modified: ssh-audit.py modified: tox.ini * Fix E221 - multiple spaces before operator modified: ssh-audit.py modified: tox.ini * Update comment about Flake 8 E241 Lots of data is formatted as tables, so this warning is disabled for a good reason. modified: tox.ini * Fix E401 - multiple imports on one line modified: ssh-audit.py modified: tox.ini * Do not ignore Flake8 warning F401 ... as there were no errors in source code anyway. modified: tox.ini * Fix F821 - undefined name modified: ssh-audit.py modified: tox.ini * Reformat ignore section for Flake8 modified: tox.ini * Flake8 test suite modified: test/conftest.py modified: test/test_auditconf.py modified: test/test_banner.py modified: test/test_buffer.py modified: test/test_errors.py modified: test/test_output.py modified: test/test_resolve.py modified: test/test_socket.py modified: test/test_software.py modified: test/test_ssh1.py modified: test/test_ssh2.py modified: test/test_ssh_algorithm.py modified: test/test_utils.py modified: test/test_version_compare.py modified: tox.ini
2020-06-09 23:54:07 +02:00
2020-06-13 03:01:10 +02:00
def flush(self):
lines = self.__out.getvalue().splitlines()
sys.stdout = self.__old_stdout
self.__out = None
return lines
2016-10-13 16:55:59 +02:00
@pytest.fixture(scope='module')
def output_spy():
2020-06-13 03:01:10 +02:00
return _OutputSpy()
2016-11-02 18:28:16 +01:00
class _VirtualGlobalSocket(object):
2020-06-13 03:01:10 +02:00
def __init__(self, vsocket):
self.vsocket = vsocket
self.addrinfodata = {}
# pylint: disable=unused-argument
def create_connection(self, address, timeout=0, source_address=None):
# pylint: disable=protected-access
return self.vsocket._connect(address, True)
# pylint: disable=unused-argument
def socket(self,
family=socket.AF_INET,
socktype=socket.SOCK_STREAM,
proto=0,
fileno=None):
return self.vsocket
def getaddrinfo(self, host, port, family=0, socktype=0, proto=0, flags=0):
key = '{0}#{1}'.format(host, port)
if key in self.addrinfodata:
data = self.addrinfodata[key]
if isinstance(data, Exception):
raise data
return data
if host == 'localhost':
r = []
if family in (0, socket.AF_INET):
r.append((socket.AF_INET, 1, 6, '', ('127.0.0.1', port)))
if family in (0, socket.AF_INET6):
r.append((socket.AF_INET6, 1, 6, '', ('::1', port)))
return r
return []
2016-11-02 18:28:16 +01:00
class _VirtualSocket(object):
2020-06-13 03:01:10 +02:00
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 = {}
self.gsock = _VirtualGlobalSocket(self)
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):
2020-06-13 03:01:10 +02:00
vsocket = _VirtualSocket()
gsock = vsocket.gsock
monkeypatch.setattr(socket, 'create_connection', gsock.create_connection)
monkeypatch.setattr(socket, 'socket', gsock.socket)
monkeypatch.setattr(socket, 'getaddrinfo', gsock.getaddrinfo)
return vsocket