Twisted Producer was never unregistered

Question

I am writing a custom SSL proxy with Twisted. I keep running in to an issue that happens every so often and I cant figure out what the problem is.

When I try to connect the client transport to the server's transport through the registerProducer function exactly as twisted.protocols.portforward functions I keep getting this error.

      File "/opt/Memory/Mobile/Proxy/forwarder.py", line 40, in connectionMade
        self.peer.transport.registerProducer(self.transport, True)
      File "/usr/lib/python3.9/site-packages/twisted/protocols/tls.py", line 602, in registerProducer
        self.transport.registerProducer(producer, True)
      File "/usr/lib/python3.9/site-packages/twisted/internet/_newtls.py", line 233, in registerProducer
        FileDescriptor.registerProducer(self, producer, streaming)
      File "/usr/lib/python3.9/site-packages/twisted/internet/abstract.py", line 104, in registerProducer
        raise RuntimeError(
    builtins.RuntimeError: Cannot register producer <twisted.protocols.tls._ProducerMembrane object at 0x7fb5799f0910>, because producer <twisted.protocols.tls._ProducerMembrane object at 0x7fb579b474c0> was never unregistered.

Here are my Inherited Classes from twisted?

from twisted.internet import reactor
from twisted.internet import ssl
from twisted.protocols import portforward

from twisted.internet import protocol
from twisted.python import log
import sys

##SSLProxy base class that will be inherited
class SSLProxy(protocol.Protocol):
    noisy = True
    peer = None

    def setPeer(self, peer):
        #log.msg("SSLProxy.setPeer")
        self.peer = peer

    def connectionLost(self, reason):
        #log.msg("SSLProxy.connectionLost")
        if self.peer is not None:
            self.peer.transport.loseConnection()
            self.peer = None
        elif self.noisy:
            log.msg("Unable to connect to peer: {}".format(reason))

    def dataReceived(self, data):
        #log.msg("SSLProxy.dataReceived")
        if self.peer is not None:
            self.peer.transport.write(data)

##Foward data from Proxy to => Remote Server
class SSLProxyClient(SSLProxy):
    def connectionMade(self):
        #log.msg("SSLProxyClient.connectionMade")
        self.peer.setPeer(self)

        self.transport.registerProducer(self.peer.transport, True)
        self.peer.transport.registerProducer(self.transport, True)

        # We're connected, everybody can read to their hearts content.
        self.peer.transport.resumeProducing()



class SSLProxyClientFactory(protocol.ClientFactory):
    protocol = SSLProxyClient

    def setServer(self, server):
        #log.msg("SSLProxyClientFactory.setServer")
        self.server = server

    def buildProtocol(self, *args, **kw):
        #log.msg("SSLProxyClientFactory.buildProtocol")
        prot = protocol.ClientFactory.buildProtocol(self, *args, **kw)
        prot.setPeer(self.server)
        return prot

    def clientConnectionFailed(self, connector, reason):
        #log.msg("SSLProxyClientFactory.clientConnectionFailed")
        self.server.transport.loseConnection()


class SSLProxyServer(SSLProxy):
    clientProtocolFactory = SSLProxyClientFactory
    reactor = None

    def connectionMade(self):
        log.msg("SSLProxyServer.connectionMade")

        #Get Current SSL Context
        ssl_context = self.transport._tlsConnection.get_context()

        #Hack to get SNI to do two functions in diffrent classes
        ssl_context._finishSNI = self.SNICallback

    def SNICallback(self, connection):
        #log.msg("SSLProxyServer.SNICallback: {}".format(connection))
        #print(connection.get_context().new_host)

        self.transport.pauseProducing()
        #self.transport.transport.pauseProducing()
        #print(dir())

        self.dst_host, self.dst_port = connection.get_context().new_host

        #Setup Clients
        self.client = self.clientProtocolFactory()
        self.client.setServer(self)

        #Start stuff
        log.msg('Redirecting to {}:{}'.format(self.dst_host, self.dst_port))

        if self.reactor is None:
            self.reactor = reactor

        log.msg("Making Connection to Dest Server: {}:{}".format(self.dst_host, self.dst_port))
        self.reactor.connectSSL(self.dst_host, self.dst_port, self.client, ssl.ClientContextFactory())

        #self.transport.resumeProducing()

    #Client -> Proxy
    def dataReceived(self, data):
        log.msg("SSLProxyServer.dataReceived: {}".format(data))

        #Call Inherited Function
        super().dataReceived(data)

class SSLProxyFactory(protocol.Factory):
    """Factory for port forwarder."""
    protocol = SSLProxyServer

    def __init__(self):
        super().__init__()
        #log.msg("SSLProxyFactory.__init__")

def sslToSSL(localport, remotehost, remoteport, serverContextFactory):
    log.msg("SSL on localhost:{} forwarding to SSL {}:{}".format(localport, remotehost, remoteport))
    return reactor.listenSSL(localport, SSLProxyFactory(), serverContextFactory)

Any guidance would be appreciated.

Answer 1

I found that if you just unregister the current Producer you can register the new one.

##Foward data from Proxy to => Remote Server
class SSLProxyClient(SSLProxy):
    def connectionMade(self):
        #log.msg("SSLProxyClient.connectionMade")
        self.peer.setPeer(self)

        self.transport.registerProducer(self.peer.transport, True)
        self.peer.transport.unregisterProducer()
        self.peer.transport.registerProducer(self.transport, True)

        # We're connected, everybody can read to their hearts content.
        self.peer.transport.resumeProducing()