<filename>idiokit/tests/test_ssl.py<gh_stars>10-100
import os
import shutil
import inspect
import tempfile
import unittest
import contextlib
import socket as stdlib_socket

from .. import idiokit, socket, ssl


@contextlib.contextmanager
def tmpdir():
    path = tempfile.mkdtemp()
    try:
        yield path
    finally:
        shutil.rmtree(path)


class TestWrapSocket(unittest.TestCase):
    @idiokit.stream
    def _test_wrap_socket(self, client_conn, server_conn, require_cert=False):
        # Self-signed combined certfile/keyfile created using cfssl version
        # 1.2.0 with the following command:
        #   echo '{"key":{"algo":"rsa","size":2048}}' | cfssl selfsign idiokit.example -
        certdata = inspect.cleandoc("""
            -----BEGIN CERTIFICATE-----
            MIIDATCCAemgAwIBAgIICZaQBavdnb0wDQYJKoZIhvcNAQELBQAwADAeFw0xNjA0
            MDcxMzMzNDBaFw0xNjA3MDcxOTM4NDBaMAAwggEiMA0GCSqGSIb3DQEBAQUAA4IB
            DwAwggEKAoIBAQDLYuXhBLrkpgOE2QCtO5RzPo9hXu5c5oKxOF2ucHt/gEDDd8zv
            VkW680Q2J2p+yBDtrLojhQ83rLA5PXFP1neamafNY+mQ0kPAyVZaY+kDcjPVL0Ua
            mxJ2KX4bpcC6re58uoR4jmcrHzIfDaENHBnW2uxsO0nkZUBoAMYUxlk5kkaKxq3p
            OYoVxyNskd8emqZmzqeYq1+5FnbSXrjkbmmHRs7azpkYM558KdKH2teI3zghy19r
            jr2C9XIcHodDK+k1H8lW5ohMRNogl4g6kB6OiS/mCOPAO+JAjh1krA8yeDK1trh4
            wpZpPlJFVA<KEY>
            -----END CERTIFICATE-----
            -----BEGIN RSA PRIVATE KEY-----
            <KEY>vSiISHVR
            -----END RSA PRIVATE KEY-----
        """)

        @idiokit.stream
        def run_server(server_conn):
            with tempfile.NamedTemporaryFile() as certfile:
                certfile.write(certdata)
                certfile.flush()
                server = yield ssl.wrap_socket(
                    server_conn,
                    server_side=True,
                    certfile=certfile.name
                )

            yield server.sendall("Hello, World!")
            yield server.shutdown(socket.SHUT_RDWR)
            yield server.close()

        @idiokit.stream
        def run_client(client_conn):
            client = yield ssl.wrap_socket(
                client_conn,
                require_cert=require_cert
            )

            result = ""
            while True:
                data = yield client.recv(1024)
                if not data:
                    idiokit.stop(result)
                result += data
            idiokit.stop(result)

        result = yield run_server(server_conn) | run_client(client_conn)
        self.assertEqual(result, "Hello, World!")

    def test_wrapping_a_socket_should_work(self):
        @idiokit.stream
        def connect_client(socket_path):
            sock = socket.Socket(socket.AF_UNIX)
            yield sock.connect(socket_path)
            yield idiokit.send(sock)

        @idiokit.stream
        def connect_server(sock):
            conn, _ = yield sock.accept()
            client = yield idiokit.next()
            idiokit.stop(client, conn)

        @idiokit.stream
        def test():
            with tmpdir() as tmp:
                socket_path = os.path.join(tmp, "socket")

                server_sock = socket.Socket(socket.AF_UNIX)
                try:
                    yield server_sock.bind(socket_path)
                    yield server_sock.listen(1)
                    client_conn, server_conn = yield idiokit.pipe(
                        connect_client(socket_path),
                        connect_server(server_sock)
                    )
                finally:
                    yield server_sock.close()
            yield self._test_wrap_socket(client_conn, server_conn)
        idiokit.main_loop(test())

    def test_wrapping_a_socket_created_with_fromfd_should_work(self):
        left, right = socket.socketpair()
        idiokit.main_loop(self._test_wrap_socket(left, right))

    def test_wrapping_a_socket_created_with_socketpair_should_work(self):
        left, right = stdlib_socket.socketpair()
        idiokit.main_loop(self._test_wrap_socket(
            socket.fromfd(left.fileno(), left.family, left.type),
            socket.fromfd(right.fileno(), right.family, right.type)
        ))

    def test_should_check_certificates_when_require_cert_is_set(self):
        left, right = socket.socketpair()
        self.assertRaises(
            ssl.SSLError,
            idiokit.main_loop,
            self._test_wrap_socket(left, right, require_cert=True)
        )

<filename>idiokit/tests/test_ssl.py<gh_stars>10-100
import os
import shutil
import inspect
import tempfile
import unittest
import contextlib
import socket as stdlib_socket

from .. import idiokit, socket, ssl


@contextlib.contextmanager
def tmpdir():
    path = tempfile.mkdtemp()
    try:
        yield path
    finally:
        shutil.rmtree(path)


class TestWrapSocket(unittest.TestCase):
    @idiokit.stream
    def _test_wrap_socket(self, client_conn, server_conn, require_cert=False):
        # Self-signed combined certfile/keyfile created using cfssl version
        # 1.2.0 with the following command:
        #   echo '{"key":{"algo":"rsa","size":2048}}' | cfssl selfsign idiokit.example -
        certdata = inspect.cleandoc("""
            -----BEGIN CERTIFICATE-----
            MIIDATCCAemgAwIBAgIICZaQBavdnb0wDQYJKoZIhvcNAQELBQAwADAeFw0xNjA0
            MDcxMzMzNDBaFw0xNjA3MDcxOTM4NDBaMAAwggEiMA0GCSqGSIb3DQEBAQUAA4IB
            DwAwggEKAoIBAQDLYuXhBLrkpgOE2QCtO5RzPo9hXu5c5oKxOF2ucHt/gEDDd8zv
            VkW680Q2J2p+yBDtrLojhQ83rLA5PXFP1neamafNY+mQ0kPAyVZaY+kDcjPVL0Ua
            mxJ2KX4bpcC6re58uoR4jmcrHzIfDaENHBnW2uxsO0nkZUBoAMYUxlk5kkaKxq3p
            OYoVxyNskd8emqZmzqeYq1+5FnbSXrjkbmmHRs7azpkYM558KdKH2teI3zghy19r
            jr2C9XIcHodDK+k1H8lW5ohMRNogl4g6kB6OiS/mCOPAO+JAjh1krA8yeDK1trh4
            wpZpPlJFVA<KEY>
            -----END CERTIFICATE-----
            -----BEGIN RSA PRIVATE KEY-----
            <KEY>vSiISHVR
            -----END RSA PRIVATE KEY-----
        """)

        @idiokit.stream
        def run_server(server_conn):
            with tempfile.NamedTemporaryFile() as certfile:
                certfile.write(certdata)
                certfile.flush()
                server = yield ssl.wrap_socket(
                    server_conn,
                    server_side=True,
                    certfile=certfile.name
                )

            yield server.sendall("Hello, World!")
            yield server.shutdown(socket.SHUT_RDWR)
            yield server.close()

        @idiokit.stream
        def run_client(client_conn):
            client = yield ssl.wrap_socket(
                client_conn,
                require_cert=require_cert
            )

            result = ""
            while True:
                data = yield client.recv(1024)
                if not data:
                    idiokit.stop(result)
                result += data
            idiokit.stop(result)

        result = yield run_server(server_conn) | run_client(client_conn)
        self.assertEqual(result, "Hello, World!")

    def test_wrapping_a_socket_should_work(self):
        @idiokit.stream
        def connect_client(socket_path):
            sock = socket.Socket(socket.AF_UNIX)
            yield sock.connect(socket_path)
            yield idiokit.send(sock)

        @idiokit.stream
        def connect_server(sock):
            conn, _ = yield sock.accept()
            client = yield idiokit.next()
            idiokit.stop(client, conn)

        @idiokit.stream
        def test():
            with tmpdir() as tmp:
                socket_path = os.path.join(tmp, "socket")

                server_sock = socket.Socket(socket.AF_UNIX)
                try:
                    yield server_sock.bind(socket_path)
                    yield server_sock.listen(1)
                    client_conn, server_conn = yield idiokit.pipe(
                        connect_client(socket_path),
                        connect_server(server_sock)
                    )
                finally:
                    yield server_sock.close()
            yield self._test_wrap_socket(client_conn, server_conn)
        idiokit.main_loop(test())

    def test_wrapping_a_socket_created_with_fromfd_should_work(self):
        left, right = socket.socketpair()
        idiokit.main_loop(self._test_wrap_socket(left, right))

    def test_wrapping_a_socket_created_with_socketpair_should_work(self):
        left, right = stdlib_socket.socketpair()
        idiokit.main_loop(self._test_wrap_socket(
            socket.fromfd(left.fileno(), left.family, left.type),
            socket.fromfd(right.fileno(), right.family, right.type)
        ))

    def test_should_check_certificates_when_require_cert_is_set(self):
        left, right = socket.socketpair()
        self.assertRaises(
            ssl.SSLError,
            idiokit.main_loop,
            self._test_wrap_socket(left, right, require_cert=True)
        )

# Self-signed combined certfile/keyfile created using cfssl version # 1.2.0 with the following command: #   echo '{"key":{"algo":"rsa","size":2048}}' | cfssl selfsign idiokit.example - -----BEGIN CERTIFICATE-----             MIIDATCCAemgAwIBAgIICZaQBavdnb0wDQYJKoZIhvcNAQELBQAwADAeFw0xNjA0             MDcxMzMzNDBaFw0xNjA3MDcxOTM4NDBaMAAwggEiMA0GCSqGSIb3DQEBAQUAA4IB             DwAwggEKAoIBAQDLYuXhBLrkpgOE2QCtO5RzPo9hXu5c5oKxOF2ucHt/gEDDd8zv             VkW680Q2J2p+yBDtrLojhQ83rLA5PXFP1neamafNY+mQ0kPAyVZaY+kDcjPVL0Ua             mxJ2KX4bpcC6re58uoR4jmcrHzIfDaENHBnW2uxsO0nkZUBoAMYUxlk5kkaKxq3p             OYoVxyNskd8emqZmzqeYq1+5FnbSXrjkbmmHRs7azpkYM558KdKH2teI3zghy19r             jr2C9XIcHodDK+k1H8lW5ohMRNogl4g6kB6OiS/mCOPAO+JAjh1krA8yeDK1trh4             wpZpPlJFVA<KEY>             -----END CERTIFICATE-----             -----BEGIN RSA PRIVATE KEY-----             <KEY>vSiISHVR             -----END RSA PRIVATE KEY-----