ipv6.py   [plain text]

import socket
from twisted.internet import tcp
from twisted.internet import default
from twisted.internet import protocol
from twisted.internet import reactor

class IPv6Address(object):
    def __init__(self, type, host, port, flowInfo, scope):
        self.type = type
        self.host = host
        self.port = port
        self.flowInfo = flowInfo
        self.scope = scope

    def __eq__(self, other):
        if isinstance(other, IPv6Address):
            a = (self.type, self.host, self.port, self.flowInfo, self.scope)
            b = (other.type, other.host, other.port, other.flowInfo, other.scope)
            return a == b
        return False
    
    def __str__(self):
        return 'IPv6Address(%s, %r, %d, %d, %d)' % (
            self.type, self.host, self.port, self.flowInfo, self.scope)

def isIPv6Address(ip):
    try:
        socket.inet_pton(socket.AF_INET6, ip)
    except:
        return 0
    return 1

class Client(tcp.Client):
    addressFamily = socket.AF_INET6

    def resolveAddress(self):
        if isIPv6Address(self.addr[0]):
            self._setRealAddress(self.addr[0])
        else:
            reactor.resolve(self.addr[0]).addCallbacks(
                self._setRealAddress, self.failIfNotConnected
            )

    def getHost(self):
        return IPv6Address('TCP', *self.socket.getsockname())

    def getPeer(self):
        return IPv6Address('TCP', *self.socket.getpeername())


class Connector(tcp.Connector):
    def _makeTransport(self):
        return Client(self.host, self.port, self.bindAddress, self, self.reactor)
    
    def getDestination(self):
        return IPv6Address('TCP', self.host, self.port)

class Server(tcp.Server):
    def getHost(self):
        return IPv6Address('TCP', *self.socket.getsockname())

    def getPeer(self):
        return IPv6Address('TCP', *self.client)

class Port(tcp.Port):
    addressFamily = socket.AF_INET6

    transport = Server

    def _buildAddr(self, address):
        return IPv6Address('TCP', *address)

    def getHost(self):
        return IPv6Address('TCP', *self.socket.getsockname())

    def getPeer(self):
        return IPv6Address('TCP', *self.socket.getpeername())

def connectTCP6(host, port, factory, timeout=30, bindAddress=None, reactor=None):
    if reactor is None:
        from twisted.internet import reactor
    return reactor.connectWith(
        Connector, host, port, factory, timeout, bindAddress
    )


def listenTCP6(port, factory, backlog=5, interface='::', reactor=None):
    if reactor is None:
        from twisted.internet import reactor
    return reactor.listenWith(Port, port, factory, backlog, interface)

def main():
    from twisted.internet import reactor

    class TrivialProtocol(protocol.Protocol):
        def connectionMade(self):
            print 'I (', self.transport.getHost(), ') am connected! (to ', self.transport.getPeer(), ')'
            self.transport.write('Hello, world!\n')
        
        def dataReceived(self, data):
            print 'Received: ' + repr(data)

    class TrivialServerFactory(protocol.ServerFactory):
        protocol = TrivialProtocol
    class TrivialClientFactory(protocol.ClientFactory):
        protocol = TrivialProtocol
    
    p = listenTCP6(6666, TrivialServerFactory())
    c = connectTCP6('::1', 6666, TrivialClientFactory())
    
    reactor.run()

if __name__ == '__main__':
    main()