val/irctest
1
0
Fork 0
mirror of https://github.com/progval/irctest.git synced 2025-04-04 22:39:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cherry-picking' ('Cherry pick commits from my old branch of irctest' GH-17)

Browse Source
This commit is contained in:
Valentin Lorentz
2021-02-22 18:44:17 +01:00
parent e9a2bdd008 f0141b0a93
commit 34ed62fd85
14 changed files with 221 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
irctest/basecontrollers.py
View File

@ -39,6 +39,11 @@ class DirectoryBasedController(_BaseController):
self.kill_proc()
if self.directory:
shutil.rmtree(self.directory)
def terminate(self):
"""Stops the process gracefully, and does not clean its config."""
self.proc.terminate()
self.proc.wait()
self.proc = None
def open_file(self, name, mode='a'):
"""Open a file in the configuration directory."""
assert self.directory
@ -49,7 +54,13 @@ class DirectoryBasedController(_BaseController):
assert os.path.isdir(dir_)
return open(os.path.join(self.directory, name), mode)
def create_config(self):
self.directory = tempfile.mkdtemp()
"""If there is no config dir, creates it and returns True.
Else returns False."""
if self.directory:
return False
else:
self.directory = tempfile.mkdtemp()
return True
def gen_ssl(self):
self.csr_path = os.path.join(self.directory, 'ssl.csr')

30
irctest/cases.py
View File

@ -118,8 +118,9 @@ class BaseClientTestCase(_IrcTestCase):
try:
self.conn.sendall(b'QUIT :end of test.')
except BrokenPipeError:
# client already disconnected
pass
pass # client already disconnected
except OSError:
pass # the conn was already closed by the test, or something
self.controller.kill()
if self.conn:
self.conn_file.close()
@ -131,9 +132,10 @@ class BaseClientTestCase(_IrcTestCase):
self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server.bind(('', 0)) # Bind any free port
self.server.listen(1)
def acceptClient(self, tls_cert=None, tls_key=None):
def acceptClient(self, tls_cert=None, tls_key=None, server=None):
"""Make the server accept a client connection. Blocking."""
(self.conn, addr) = self.server.accept()
server = server or self.server
(self.conn, addr) = server.accept()
if tls_cert is None and tls_key is None:
pass
else:
@ -171,11 +173,9 @@ class BaseClientTestCase(_IrcTestCase):
if not filter_pred or filter_pred(msg):
return msg
def sendLine(self, line):
ret = self.conn.sendall(line.encode())
assert ret is None
self.conn.sendall(line.encode())
if not line.endswith('\r\n'):
ret = self.conn.sendall(b'\r\n')
assert ret is None
self.conn.sendall(b'\r\n')
if self.show_io:
print('{:.3f} S: {}'.format(time.time(), line.strip()))
@ -455,6 +455,20 @@ class OptionalityHelper:
return f(self)
return newf
def checkCapabilitySupport(self, cap):
if cap in self.controller.supported_capabilities:
return
raise runner.CapabilityNotSupported(cap)
def skipUnlessSupportsCapability(cap):
def decorator(f):
@functools.wraps(f)
def newf(self):
self.checkCapabilitySupport(cap)
return f(self)
return newf
return decorator
class SpecificationSelector:
def requiredBySpecification(*specifications, strict=False):

3
irctest/client_mock.py
View File

@ -1,4 +1,5 @@
import ssl
import sys
import time
import socket
from .irc_utils import message_parser
@ -104,7 +105,7 @@ class ClientMock:
ret = self.conn.sendall(encoded_line)
except BrokenPipeError:
raise ConnectionClosed()
if self.ssl: # https://bugs.python.org/issue25951
if sys.version_info <= (3, 6) and self.ssl: # https://bugs.python.org/issue25951
assert ret == len(encoded_line), (ret, repr(encoded_line))
else:
assert ret is None, ret

59
irctest/client_tests/test_sasl.py
View File

@ -1,4 +1,8 @@
import base64
import hashlib
import ecdsa
from ecdsa.util import sigencode_der, sigdecode_der
try:
import pyxmpp2_scram as scram
@ -9,6 +13,27 @@ from irctest import cases
from irctest import authentication
from irctest.irc_utils.message_parser import Message
ECDSA_KEY = """
-----BEGIN EC PARAMETERS-----
BggqhkjOPQMBBw==
-----END EC PARAMETERS-----
-----BEGIN EC PRIVATE KEY-----
MHcCAQEEIIJueQ3W2IrGbe9wKdOI75yGS7PYZSj6W4tg854hlsvmoAoGCCqGSM49
AwEHoUQDQgAEAZmaVhNSMmV5r8FXPvKuMnqDKyIA9pDHN5TNMfiF3mMeikGgK10W
IRX9cyi2wdYg9mUUYyh9GKdBCYHGUJAiCA==
-----END EC PRIVATE KEY-----
"""
CHALLENGE = bytes(range(32))
assert len(CHALLENGE) == 32
class IdentityHash:
def __init__(self, data):
self._data = data
def digest(self):
return self._data
class SaslTestCase(cases.BaseClientTestCase, cases.ClientNegociationHelper,
cases.OptionalityHelper):
@cases.OptionalityHelper.skipUnlessHasMechanism('PLAIN')
@ -122,6 +147,36 @@ class SaslTestCase(cases.BaseClientTestCase, cases.ClientNegociationHelper,
m = self.negotiateCapabilities(['sasl'], False)
self.assertEqual(m, Message({}, None, 'CAP', ['END']))
@cases.OptionalityHelper.skipUnlessHasMechanism('ECDSA-NIST256P-CHALLENGE')
def testEcdsa(self):
"""Test ECDSA authentication.
"""
auth = authentication.Authentication(
mechanisms=[authentication.Mechanisms.ecdsa_nist256p_challenge],
username='jilles',
ecdsa_key=ECDSA_KEY,
)
m = self.negotiateCapabilities(['sasl'], auth=auth)
self.assertEqual(m, Message({}, None, 'AUTHENTICATE', ['ECDSA-NIST256P-CHALLENGE']))
self.sendLine('AUTHENTICATE +')
m = self.getMessage()
self.assertEqual(m, Message({}, None, 'AUTHENTICATE',
['amlsbGVz'])) # jilles
self.sendLine('AUTHENTICATE {}'.format(base64.b64encode(CHALLENGE).decode('ascii')))
m = self.getMessage()
self.assertMessageEqual(m, command='AUTHENTICATE')
sk = ecdsa.SigningKey.from_pem(ECDSA_KEY)
vk = sk.get_verifying_key()
signature = base64.b64decode(m.params[0])
try:
vk.verify(signature, CHALLENGE, hashfunc=IdentityHash, sigdecode=sigdecode_der)
except ecdsa.BadSignatureError:
raise AssertionError('Bad signature')
self.sendLine('900 * * foo :You are now logged in.')
self.sendLine('903 * :SASL authentication successful')
m = self.negotiateCapabilities(['sasl'], False)
self.assertEqual(m, Message({}, None, 'CAP', ['END']))
@cases.OptionalityHelper.skipUnlessHasMechanism('SCRAM-SHA-256')
def testScram(self):
"""Test SCRAM-SHA-256 authentication.
@ -158,6 +213,10 @@ class SaslTestCase(cases.BaseClientTestCase, cases.ClientNegociationHelper,
self.sendLine('AUTHENTICATE :' + base64.b64encode(response).decode())
self.assertEqual(properties, {'authzid': None, 'username': 'jilles'})
m = self.getMessage()
self.assertEqual(m.command, 'AUTHENTICATE', m)
self.assertEqual(m.params, ['+'], m)
@cases.OptionalityHelper.skipUnlessHasMechanism('SCRAM-SHA-256')
def testScramBadPassword(self):
"""Test SCRAM-SHA-256 authentication with a bad password.

92
irctest/client_tests/test_tls.py
View File

@ -1,3 +1,6 @@
import socket
import ssl
from irctest import tls
from irctest import cases
from irctest.exceptions import ConnectionClosed
@ -141,3 +144,92 @@ class TlsTestCase(cases.BaseClientTestCase):
self.acceptClient(tls_cert=BAD_CERT, tls_key=BAD_KEY)
with self.assertRaises((ConnectionClosed, ConnectionResetError)):
m = self.getMessage()
class StsTestCase(cases.BaseClientTestCase, cases.OptionalityHelper):
def setUp(self):
super().setUp()
self.insecure_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.insecure_server.bind(('', 0)) # Bind any free port
self.insecure_server.listen(1)
def tearDown(self):
self.insecure_server.close()
super().tearDown()
@cases.OptionalityHelper.skipUnlessSupportsCapability('sts')
def testSts(self):
tls_config = tls.TlsConfig(
enable=False,
trusted_fingerprints=[GOOD_FINGERPRINT])
# Connect client to insecure server
(hostname, port) = self.insecure_server.getsockname()
self.controller.run(
hostname=hostname,
port=port,
auth=None,
tls_config=tls_config,
)
self.acceptClient(server=self.insecure_server)
# Send STS policy to client
m = self.getMessage()
self.assertEqual(m.command, 'CAP',
'First message is not CAP LS.')
self.assertEqual(m.params[0], 'LS',
'First message is not CAP LS.')
self.sendLine('CAP * LS :sts=port={}'.format(self.server.getsockname()[1]))
# "If the client is not already connected securely to the server
# at the requested hostname, it MUST close the insecure connection
# and reconnect securely on the stated port."
self.acceptClient(tls_cert=GOOD_CERT, tls_key=GOOD_KEY)
# Send the STS policy, over secure connection this time
self.sendLine('CAP * LS :sts=duration=10,port={}'.format(
self.server.getsockname()[1]))
# Make the client reconnect. It should reconnect to the secure server.
self.sendLine('ERROR :closing link')
self.acceptClient()
# Kill the client
self.controller.terminate()
# Run the client, still configured to connect to the insecure server
self.controller.run(
hostname=hostname,
port=port,
auth=None,
tls_config=tls_config,
)
# The client should remember the STS policy and connect to the secure
# server
self.acceptClient()
@cases.OptionalityHelper.skipUnlessSupportsCapability('sts')
def testStsInvalidCertificate(self):
# Connect client to insecure server
(hostname, port) = self.insecure_server.getsockname()
self.controller.run(
hostname=hostname,
port=port,
auth=None,
)
self.acceptClient(server=self.insecure_server)
# Send STS policy to client
m = self.getMessage()
self.assertEqual(m.command, 'CAP',
'First message is not CAP LS.')
self.assertEqual(m.params[0], 'LS',
'First message is not CAP LS.')
self.sendLine('CAP * LS :sts=port={}'.format(self.server.getsockname()[1]))
# The client will reconnect to the TLS port. Unfortunately, it does
# not trust its fingerprint.
with self.assertRaises((ssl.SSLError, socket.error)):
self.acceptClient(tls_cert=GOOD_CERT, tls_key=GOOD_KEY)

1
irctest/controllers/charybdis.py
View File

@ -45,6 +45,7 @@ TEMPLATE_SSL_CONFIG = """
class CharybdisController(BaseServerController, DirectoryBasedController):
software_name = 'Charybdis'
supported_sasl_mechanisms = set()
supported_capabilities = set() # Not exhaustive
def create_config(self):
super().create_config()
with self.open_file('server.conf'):

3
irctest/controllers/girc.py
View File

@ -4,12 +4,13 @@ from irctest.basecontrollers import BaseClientController, NotImplementedByContro
class GircController(BaseClientController):
software_name = 'gIRC'
supported_sasl_mechanisms = ['PLAIN']
supported_capabilities = set() # Not exhaustive
def __init__(self):
super().__init__()
self.directory = None
self.proc = None
self.supported_sasl_mechanisms = ['PLAIN']
def kill(self):
if self.proc:

2
irctest/controllers/hybrid.py
View File

@ -43,6 +43,8 @@ TEMPLATE_SSL_CONFIG = """
class HybridController(BaseServerController, DirectoryBasedController):
software_name = 'Hybrid'
supported_sasl_mechanisms = set()
supported_capabilities = set() # Not exhaustive
def create_config(self):
super().create_config()
with self.open_file('server.conf'):

2
irctest/controllers/inspircd.py
View File

@ -30,6 +30,8 @@ TEMPLATE_SSL_CONFIG = """
class InspircdController(BaseServerController, DirectoryBasedController):
software_name = 'InspIRCd'
supported_sasl_mechanisms = set()
supported_capabilities = set() # Not exhaustive
def create_config(self):
super().create_config()
with self.open_file('server.conf'):

20
irctest/controllers/limnoria.py
View File

@ -2,6 +2,7 @@ import os
import subprocess
from irctest import authentication
from irctest import tls
from irctest.basecontrollers import NotImplementedByController
from irctest.basecontrollers import BaseClientController, DirectoryBasedController
@ -30,14 +31,19 @@ class LimnoriaController(BaseClientController, DirectoryBasedController):
supported_sasl_mechanisms = {
'PLAIN', 'ECDSA-NIST256P-CHALLENGE', 'SCRAM-SHA-256', 'EXTERNAL',
}
def create_config(self):
super().create_config()
with self.open_file('bot.conf'):
pass
with self.open_file('conf/users.conf'):
pass
supported_capabilities = set(['sts']) # Not exhaustive
def run(self, hostname, port, auth, tls_config):
def create_config(self):
create_config = super().create_config()
if create_config:
with self.open_file('bot.conf'):
pass
with self.open_file('conf/users.conf'):
pass
def run(self, hostname, port, auth, tls_config=None):
if tls_config is None:
tls_config = tls.TlsConfig(enable=False, trusted_fingerprints=[])
# Runs a client with the config given as arguments
assert self.proc is None
self.create_config()

2
irctest/controllers/mammon.py
View File

@ -66,6 +66,8 @@ class MammonController(BaseServerController, DirectoryBasedController):
supported_sasl_mechanisms = {
'PLAIN', 'ECDSA-NIST256P-CHALLENGE',
}
supported_capabilities = set() # Not exhaustive
def create_config(self):
super().create_config()
with self.open_file('server.conf'):

6
irctest/controllers/oragono.py
View File

@ -153,6 +153,12 @@ class OragonoController(BaseServerController, DirectoryBasedController):
'PLAIN',
}
_port_wait_interval = .01
supported_capabilities = set() # Not exhaustive
def create_config(self):
super().create_config()
with self.open_file('ircd.yaml'):
pass
def kill_proc(self):
self.proc.kill()

2
irctest/controllers/sopel.py
View File

@ -24,6 +24,8 @@ class SopelController(BaseClientController):
supported_sasl_mechanisms = {
'PLAIN',
}
supported_capabilities = set() # Not exhaustive
def __init__(self, test_config):
super().__init__(test_config)
self.filename = next(tempfile._get_candidate_names()) + '.cfg'

4
irctest/runner.py
View File

@ -19,6 +19,10 @@ class OptionalSaslMechanismNotSupported(unittest.SkipTest):
def __str__(self):
return 'Unsupported SASL mechanism: {}'.format(self.args[0])
class CapabilityNotSupported(unittest.SkipTest):
def __str__(self):
return 'Unsupported capability: {}'.format(self.args[0])
class NotRequiredBySpecifications(unittest.SkipTest):
def __str__(self):
return 'Tests not required by the set of tested specification(s).'