OXIESEC PANEL

Name	Size	Modified	Perms
📁 ..	-	03/31/2022 06:22:38 AM	rwxr-xr-x
📄 __init__.py	112 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📁 __pycache__	-	03/31/2022 06:22:39 AM	rwxr-xr-x
📄 _awaittests.py.3only	4.37 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 _posixifaces.py	4.58 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 _win32ifaces.py	3.79 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 _yieldfromtests.py.3only	4.17 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 connectionmixins.py	19.71 KB	09/08/2017 10:38:36 AM	rw-r--r--
📁 fake_CAs	-	03/31/2022 06:22:38 AM	rwxr-xr-x
📄 fakeendpoint.py	1.52 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 modulehelpers.py	1.69 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_cli.py	664 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_connectionlost.py	119 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_gireactornocompat.py	721 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_helper.py	1.06 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 reactormixins.py	13.72 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_abstract.py	1.98 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_address.py	11.45 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_base.py	10.98 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_baseprocess.py	2.55 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_core.py	11 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_coroutines.py	1.67 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 test_default.py	3.46 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_endpoints.py	143.15 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_epollreactor.py	7.16 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_fdset.py	13.18 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_filedescriptor.py	2.74 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_gireactor.py	8.83 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_glibbase.py	2.23 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_inlinecb.py	4.55 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_inotify.py	16.29 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_iocp.py	5.05 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_kqueuereactor.py	1.87 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_main.py	1.4 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_newtls.py	6.54 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_pollingfile.py	1.25 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_posixbase.py	9.31 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_posixprocess.py	10.9 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_process.py	31.29 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_protocol.py	18.29 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_resolver.py	19.58 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 test_serialport.py	1.9 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_sigchld.py	3.84 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_socket.py	9.11 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_stdio.py	6.3 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_tcp.py	83 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_threads.py	7.93 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_time.py	3.63 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_tls.py	12.32 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_udp.py	16.74 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_udp_internals.py	4.79 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_unix.py	33.94 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_win32events.py	6.35 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_win32serialport.py	5.16 KB	09/08/2017 10:38:35 AM	rw-r--r--

Editing: test_newtls.py

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for L{twisted.internet._newtls}.
"""

from __future__ import division, absolute_import

from twisted.trial import unittest
from twisted.internet import interfaces
from twisted.internet.test.reactormixins import ReactorBuilder
from twisted.internet.test.connectionmixins import (
    ConnectableProtocol, runProtocolsWithReactor)
from twisted.internet.test.test_tls import SSLCreator, TLSMixin
from twisted.internet.test.test_tls import StartTLSClientCreator
from twisted.internet.test.test_tls import ContextGeneratingMixin
from twisted.internet.test.test_tcp import TCPCreator
try:
    from twisted.protocols import tls
    from twisted.internet import _newtls
except ImportError:
    _newtls = None
from zope.interface import implementer

class BypassTLSTests(unittest.TestCase):
    """
    Tests for the L{_newtls._BypassTLS} class.
    """

if not _newtls:
        skip = "Couldn't import _newtls, perhaps pyOpenSSL is old or missing"

def test_loseConnectionPassThrough(self):
        """
        C{_BypassTLS.loseConnection} calls C{loseConnection} on the base
        class, while preserving any default argument in the base class'
        C{loseConnection} implementation.
        """
        default = object()
        result = []

class FakeTransport(object):
            def loseConnection(self, _connDone=default):
                result.append(_connDone)

bypass = _newtls._BypassTLS(FakeTransport, FakeTransport())

# The default from FakeTransport is used:
        bypass.loseConnection()
        self.assertEqual(result, [default])

# And we can pass our own:
        notDefault = object()
        bypass.loseConnection(notDefault)
        self.assertEqual(result, [default, notDefault])

class FakeProducer(object):
    """
    A producer that does nothing.
    """

def pauseProducing(self):
        pass

def resumeProducing(self):
        pass

def stopProducing(self):
        pass

@implementer(interfaces.IHandshakeListener)
class ProducerProtocol(ConnectableProtocol):
    """
    Register a producer, unregister it, and verify the producer hooks up to
    innards of C{TLSMemoryBIOProtocol}.
    """

def __init__(self, producer, result):
        self.producer = producer
        self.result = result

def handshakeCompleted(self):
        if not isinstance(self.transport.protocol,
                          tls.TLSMemoryBIOProtocol):
            # Either the test or the code have a bug...
            raise RuntimeError("TLSMemoryBIOProtocol not hooked up.")

self.transport.registerProducer(self.producer, True)
        # The producer was registered with the TLSMemoryBIOProtocol:
        self.result.append(self.transport.protocol._producer._producer)

self.transport.unregisterProducer()
        # The producer was unregistered from the TLSMemoryBIOProtocol:
        self.result.append(self.transport.protocol._producer)
        self.transport.loseConnection()

class ProducerTestsMixin(ReactorBuilder, TLSMixin, ContextGeneratingMixin):
    """
    Test the new TLS code integrates C{TLSMemoryBIOProtocol} correctly.
    """

if not _newtls:
        skip = "Could not import twisted.internet._newtls"

def test_producerSSLFromStart(self):
        """
        C{registerProducer} and C{unregisterProducer} on TLS transports
        created as SSL from the get go are passed to the
        C{TLSMemoryBIOProtocol}, not the underlying transport directly.
        """
        result = []
        producer = FakeProducer()

runProtocolsWithReactor(self, ConnectableProtocol(),
                                ProducerProtocol(producer, result),
                                SSLCreator())
        self.assertEqual(result, [producer, None])

def test_producerAfterStartTLS(self):
        """
        C{registerProducer} and C{unregisterProducer} on TLS transports
        created by C{startTLS} are passed to the C{TLSMemoryBIOProtocol}, not
        the underlying transport directly.
        """
        result = []
        producer = FakeProducer()

runProtocolsWithReactor(self, ConnectableProtocol(),
                                ProducerProtocol(producer, result),
                                StartTLSClientCreator())
        self.assertEqual(result, [producer, None])

def startTLSAfterRegisterProducer(self, streaming):
        """
        When a producer is registered, and then startTLS is called,
        the producer is re-registered with the C{TLSMemoryBIOProtocol}.
        """
        clientContext = self.getClientContext()
        serverContext = self.getServerContext()
        result = []
        producer = FakeProducer()

class RegisterTLSProtocol(ConnectableProtocol):
            def connectionMade(self):
                self.transport.registerProducer(producer, streaming)
                self.transport.startTLS(serverContext)
                # Store TLSMemoryBIOProtocol and underlying transport producer
                # status:
                if streaming:
                    # _ProducerMembrane -> producer:
                    result.append(self.transport.protocol._producer._producer)
                    result.append(self.transport.producer._producer)
                else:
                    # _ProducerMembrane -> _PullToPush -> producer:
                    result.append(
                        self.transport.protocol._producer._producer._producer)
                    result.append(self.transport.producer._producer._producer)
                self.transport.unregisterProducer()
                self.transport.loseConnection()

class StartTLSProtocol(ConnectableProtocol):
            def connectionMade(self):
                self.transport.startTLS(clientContext)

runProtocolsWithReactor(self, RegisterTLSProtocol(),
                                StartTLSProtocol(), TCPCreator())
        self.assertEqual(result, [producer, producer])

def test_startTLSAfterRegisterProducerStreaming(self):
        """
        When a streaming producer is registered, and then startTLS is called,
        the producer is re-registered with the C{TLSMemoryBIOProtocol}.
        """
        self.startTLSAfterRegisterProducer(True)

def test_startTLSAfterRegisterProducerNonStreaming(self):
        """
        When a non-streaming producer is registered, and then startTLS is
        called, the producer is re-registered with the
        C{TLSMemoryBIOProtocol}.
        """
        self.startTLSAfterRegisterProducer(False)

globals().update(ProducerTestsMixin.makeTestCaseClasses())