+from contextlib import contextmanager
import random
import socket
return True
-@pytest.fixture
-def kresd(tmpdir):
+@contextmanager
+def make_kresd(workdir, certname=None):
ip = '127.0.0.1'
ip6 = '::1'
port = make_port()
tls_port = make_port()
- with Kresd(tmpdir, port, tls_port, ip, ip6) as kresd:
+ with Kresd(workdir, port, tls_port, ip, ip6, certname) as kresd:
yield kresd
# TODO: add verbose option?
# with open(kresd.logfile_path) as log:
# print(log.read()) # display log for debugging purposes
+@pytest.fixture
+def kresd(tmpdir):
+ with make_kresd(tmpdir) as kresd:
+ yield kresd
+
+
+@pytest.fixture
+def kresd_tt(tmpdir):
+ with make_kresd(tmpdir, 'tt') as kresd:
+ yield kresd
+
+
+@pytest.fixture
+def kresd_tt_expired(tmpdir):
+ with make_kresd(tmpdir, 'tt-expired') as kresd:
+ yield kresd
+
+
@pytest.fixture(params=[
'ip_tcp_socket',
'ip6_tcp_socket',
import utils
-TEMPLATES_DIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'templates')
+PYTESTS_DIR = os.path.dirname(os.path.realpath(__file__))
+CERTS_DIR = os.path.join(PYTESTS_DIR, 'certs')
+TEMPLATES_DIR = os.path.join(PYTESTS_DIR, 'templates')
KRESD_CONF_TEMPLATE = 'kresd.conf.j2'
class Kresd(ContextDecorator):
- def __init__(self, workdir, port, tls_port, ip=None, ip6=None):
+ def __init__(self, workdir, port, tls_port, ip=None, ip6=None, certname=None):
if ip is None and ip6 is None:
raise ValueError("IPv4 or IPv6 must be specified!")
self.workdir = str(workdir)
self.sockets = []
self.logfile = None
+ if certname:
+ self.tls_cert_path = os.path.join(CERTS_DIR, certname + '.cert.pem')
+ self.tls_key_path = os.path.join(CERTS_DIR, certname + '.key.pem')
+ else:
+ self.tls_cert_path = None
+ self.tls_key_path = None
+
@property
def config_path(self):
return str(os.path.join(self.workdir, 'kresd.conf'))
net.ipv4=true
net.ipv6=false
net.tcp_pipeline(65535)
---net.tls("./certs/tt.cert.pem", "./certs/tt.key.pem")
+{% if kresd.tls_key_path and kresd.tls_cert_path %}
+net.tls("{{ kresd.tls_cert_path }}", "{{ kresd.tls_key_path }}")
+{% endif %}
modules.unload("ta_signal_query")
modules.unload("priming")
modules.unload("detect_time_skew")
ssock.connect(dest)
utils.ping_alive(ssock)
+
+
+def test_tls_selfsigned_cert(kresd_tt, sock_family):
+ sock, dest = kresd_tt.stream_socket(sock_family, tls=True)
+ ctx = utils.make_ssl_context(verify_location=kresd_tt.tls_cert_path)
+ ssock = ctx.wrap_socket(sock, server_hostname='transport-test-server.com')
+ ssock.connect(dest)
+
+ utils.ping_alive(ssock)