<s> """<STR_LIT>""" <EOL> import struct , socket <EOL> from twisted . trial import unittest <EOL> from twisted . test import proto_helpers <EOL> from twisted . internet import defer , address , reactor <EOL> from twisted . internet . error import DNSLookupError <EOL> from twisted . protocols import socks <EOL> class StringTCPTransport ( proto_helpers . StringTransport ) : <EOL> stringTCPTransport_closing = False <EOL> peer = None <EOL> def getPeer ( self ) : <EOL> return self . peer <EOL> def getHost ( self ) : <EOL> return address . IPv4Address ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> ) <EOL> def loseConnection ( self ) : <EOL> self . stringTCPTransport_closing = True <EOL> class FakeResolverReactor : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , names ) : <EOL> """<STR_LIT>""" <EOL> self . names = names <EOL> def resolve ( self , hostname ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return defer . succeed ( self . names [ hostname ] ) <EOL> except KeyError : <EOL> return defer . fail ( <EOL> DNSLookupError ( "<STR_LIT>" + hostname ) ) <EOL> class SOCKSv4Driver ( socks . SOCKSv4 ) : <EOL> driver_outgoing = None <EOL> driver_listen = None <EOL> def connectClass ( self , host , port , klass , * args ) : <EOL> proto = klass ( * args ) <EOL> proto . transport = StringTCPTransport ( ) <EOL> proto . transport . peer = address . IPv4Address ( '<STR_LIT>' , host , port ) <EOL> proto . connectionMade ( ) <EOL> self . driver_outgoing = proto <EOL> return defer . succeed ( proto ) <EOL> def listenClass ( self , port , klass , * args ) : <EOL> factory = klass ( * args ) <EOL> self . driver_listen = factory <EOL> if port == <NUM_LIT:0> : <EOL> port = <NUM_LIT> <EOL> return defer . succeed ( ( '<STR_LIT>' , port ) ) <EOL> class Connect ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . sock = SOCKSv4Driver ( ) <EOL> self . sock . transport = StringTCPTransport ( ) <EOL> self . sock . connectionMade ( ) <EOL> self . sock . reactor = FakeResolverReactor ( { "<STR_LIT:localhost>" : "<STR_LIT:127.0.0.1>" } ) <EOL> def tearDown ( self ) : <EOL> outgoing = self . sock . driver_outgoing <EOL> if outgoing is not None : <EOL> self . assert_ ( outgoing . transport . stringTCPTransport_closing , <EOL> "<STR_LIT>" ) <EOL> def test_simple ( self ) : <EOL> self . sock . dataReceived ( <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:4> , <NUM_LIT:1> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) <EOL> + '<STR_LIT>' <EOL> + '<STR_LIT>' ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . sock . transport . clear ( ) <EOL> self . assertEqual ( sent , <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) ) <EOL> self . assert_ ( not self . sock . transport . stringTCPTransport_closing ) <EOL> self . assert_ ( self . sock . driver_outgoing is not None ) <EOL> self . sock . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( self . sock . driver_outgoing . transport . value ( ) , <EOL> '<STR_LIT>' ) <EOL> self . sock . driver_outgoing . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( self . sock . transport . value ( ) , '<STR_LIT>' ) <EOL> self . sock . connectionLost ( '<STR_LIT>' ) <EOL> def test_socks4aSuccessfulResolution ( self ) : <EOL> """<STR_LIT>""" <EOL> clientRequest = ( <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:4> , <NUM_LIT:1> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) <EOL> + '<STR_LIT>' <EOL> + '<STR_LIT>' ) <EOL> for byte in clientRequest : <EOL> self . sock . dataReceived ( byte ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . sock . transport . clear ( ) <EOL> self . assertEqual ( <EOL> sent , <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT> ) + socket . inet_aton ( '<STR_LIT:127.0.0.1>' ) ) <EOL> self . assertFalse ( self . sock . transport . stringTCPTransport_closing ) <EOL> self . assertNotIdentical ( self . sock . driver_outgoing , None ) <EOL> self . sock . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . sock . driver_outgoing . transport . value ( ) , '<STR_LIT>' ) <EOL> self . sock . driver_outgoing . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( self . sock . transport . value ( ) , '<STR_LIT>' ) <EOL> self . sock . connectionLost ( '<STR_LIT>' ) <EOL> def test_socks4aFailedResolution ( self ) : <EOL> """<STR_LIT>""" <EOL> clientRequest = ( <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:4> , <NUM_LIT:1> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) <EOL> + '<STR_LIT>' <EOL> + '<STR_LIT>' ) <EOL> for byte in clientRequest : <EOL> self . sock . dataReceived ( byte ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . assertEqual ( <EOL> sent , <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:0> ) + socket . inet_aton ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( self . sock . transport . stringTCPTransport_closing ) <EOL> self . assertIdentical ( self . sock . driver_outgoing , None ) <EOL> def test_accessDenied ( self ) : <EOL> self . sock . authorize = lambda code , server , port , user : <NUM_LIT:0> <EOL> self . sock . dataReceived ( <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:4> , <NUM_LIT:1> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) <EOL> + '<STR_LIT>' <EOL> + '<STR_LIT>' ) <EOL> self . assertEqual ( self . sock . transport . value ( ) , <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:0> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) ) <EOL> self . assert_ ( self . sock . transport . stringTCPTransport_closing ) <EOL> self . assertIdentical ( self . sock . driver_outgoing , None ) <EOL> def test_eofRemote ( self ) : <EOL> self . sock . dataReceived ( <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:4> , <NUM_LIT:1> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) <EOL> + '<STR_LIT>' <EOL> + '<STR_LIT>' ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . sock . transport . clear ( ) <EOL> self . sock . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( self . sock . driver_outgoing . transport . value ( ) , <EOL> '<STR_LIT>' ) <EOL> self . sock . driver_outgoing . transport . loseConnection ( ) <EOL> self . sock . driver_outgoing . connectionLost ( '<STR_LIT>' ) <EOL> def test_eofLocal ( self ) : <EOL> self . sock . dataReceived ( <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:4> , <NUM_LIT:1> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) <EOL> + '<STR_LIT>' <EOL> + '<STR_LIT>' ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . sock . transport . clear ( ) <EOL> self . sock . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( self . sock . driver_outgoing . transport . value ( ) , <EOL> '<STR_LIT>' ) <EOL> self . sock . connectionLost ( '<STR_LIT>' ) <EOL> class Bind ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . sock = SOCKSv4Driver ( ) <EOL> self . sock . transport = StringTCPTransport ( ) <EOL> self . sock . connectionMade ( ) <EOL> self . sock . reactor = FakeResolverReactor ( { "<STR_LIT:localhost>" : "<STR_LIT:127.0.0.1>" } ) <EOL> def test_simple ( self ) : <EOL> self . sock . dataReceived ( <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:4> , <NUM_LIT:2> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) <EOL> + '<STR_LIT>' <EOL> + '<STR_LIT>' ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . sock . transport . clear ( ) <EOL> self . assertEqual ( sent , <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) ) <EOL> self . assert_ ( not self . sock . transport . stringTCPTransport_closing ) <EOL> self . assert_ ( self . sock . driver_listen is not None ) <EOL> incoming = self . sock . driver_listen . buildProtocol ( ( '<STR_LIT>' , <NUM_LIT> ) ) <EOL> self . assertNotIdentical ( incoming , None ) <EOL> incoming . transport = StringTCPTransport ( ) <EOL> incoming . connectionMade ( ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . sock . transport . clear ( ) <EOL> self . assertEqual ( sent , <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:0> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) ) <EOL> self . assert_ ( not self . sock . transport . stringTCPTransport_closing ) <EOL> self . sock . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( incoming . transport . value ( ) , <EOL> '<STR_LIT>' ) <EOL> incoming . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( self . sock . transport . value ( ) , '<STR_LIT>' ) <EOL> self . sock . connectionLost ( '<STR_LIT>' ) <EOL> def test_socks4a ( self ) : <EOL> """<STR_LIT>""" <EOL> clientRequest = ( <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:4> , <NUM_LIT:2> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) <EOL> + '<STR_LIT>' <EOL> + '<STR_LIT>' ) <EOL> for byte in clientRequest : <EOL> self . sock . dataReceived ( byte ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . sock . transport . clear ( ) <EOL> self . assertEqual ( <EOL> sent , <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT> ) + socket . inet_aton ( '<STR_LIT>' ) ) <EOL> self . assertFalse ( self . sock . transport . stringTCPTransport_closing ) <EOL> self . assertNotIdentical ( self . sock . driver_listen , None ) <EOL> incoming = self . sock . driver_listen . buildProtocol ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) ) <EOL> self . assertNotIdentical ( incoming , None ) <EOL> incoming . transport = StringTCPTransport ( ) <EOL> incoming . connectionMade ( ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . sock . transport . clear ( ) <EOL> self . assertEqual ( sent , <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:0> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) ) <EOL> self . assertNotIdentical ( <EOL> self . sock . transport . stringTCPTransport_closing , None ) <EOL> self . sock . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( incoming . transport . value ( ) , '<STR_LIT>' ) <EOL> incoming . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( self . sock . transport . value ( ) , '<STR_LIT>' ) <EOL> self . sock . connectionLost ( '<STR_LIT>' ) <EOL> def test_socks4aFailedResolution ( self ) : <EOL> """<STR_LIT>""" <EOL> clientRequest = ( <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:4> , <NUM_LIT:2> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) <EOL> + '<STR_LIT>' <EOL> + '<STR_LIT>' ) <EOL> for byte in clientRequest : <EOL> self . sock . dataReceived ( byte ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . assertEqual ( <EOL> sent , <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:0> ) + socket . inet_aton ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( self . sock . transport . stringTCPTransport_closing ) <EOL> self . assertIdentical ( self . sock . driver_outgoing , None ) <EOL> def test_accessDenied ( self ) : <EOL> self . sock . authorize = lambda code , server , port , user : <NUM_LIT:0> <EOL> self . sock . dataReceived ( <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:4> , <NUM_LIT:2> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) <EOL> + '<STR_LIT>' <EOL> + '<STR_LIT>' ) <EOL> self . assertEqual ( self . sock . transport . value ( ) , <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:0> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) ) <EOL> self . assert_ ( self . sock . transport . stringTCPTransport_closing ) <EOL> self . assertIdentical ( self . sock . driver_listen , None ) <EOL> def test_eofRemote ( self ) : <EOL> self . sock . dataReceived ( <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:4> , <NUM_LIT:2> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) <EOL> + '<STR_LIT>' <EOL> + '<STR_LIT>' ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . sock . transport . clear ( ) <EOL> incoming = self . sock . driver_listen . buildProtocol ( ( '<STR_LIT>' , <NUM_LIT> ) ) <EOL> self . assertNotIdentical ( incoming , None ) <EOL> incoming . transport = StringTCPTransport ( ) <EOL> incoming . connectionMade ( ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . sock . transport . clear ( ) <EOL> self . assertEqual ( sent , <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:0> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) ) <EOL> self . assert_ ( not self . sock . transport . stringTCPTransport_closing ) <EOL> self . sock . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( incoming . transport . value ( ) , <EOL> '<STR_LIT>' ) <EOL> incoming . transport . loseConnection ( ) <EOL> incoming . connectionLost ( '<STR_LIT>' ) <EOL> def test_eofLocal ( self ) : <EOL> self . sock . dataReceived ( <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:4> , <NUM_LIT:2> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) <EOL> + '<STR_LIT>' <EOL> + '<STR_LIT>' ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . sock . transport . clear ( ) <EOL> incoming = self . sock . driver_listen . buildProtocol ( ( '<STR_LIT>' , <NUM_LIT> ) ) <EOL> self . assertNotIdentical ( incoming , None ) <EOL> incoming . transport = StringTCPTransport ( ) <EOL> incoming . connectionMade ( ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . sock . transport . clear ( ) <EOL> self . assertEqual ( sent , <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:0> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) ) <EOL> self . assert_ ( not self . sock . transport . stringTCPTransport_closing ) <EOL> self . sock . dataReceived ( '<STR_LIT>' ) <EOL> self . assertEqual ( incoming . transport . value ( ) , <EOL> '<STR_LIT>' ) <EOL> self . sock . connectionLost ( '<STR_LIT>' ) <EOL> def test_badSource ( self ) : <EOL> self . sock . dataReceived ( <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:4> , <NUM_LIT:2> , <NUM_LIT> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) <EOL> + '<STR_LIT>' <EOL> + '<STR_LIT>' ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . sock . transport . clear ( ) <EOL> incoming = self . sock . driver_listen . buildProtocol ( ( '<STR_LIT>' , <NUM_LIT> ) ) <EOL> self . assertIdentical ( incoming , None ) <EOL> sent = self . sock . transport . value ( ) <EOL> self . sock . transport . clear ( ) <EOL> self . assertEqual ( sent , <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:0> ) <EOL> + socket . inet_aton ( '<STR_LIT>' ) ) <EOL> self . assert_ ( self . sock . transport . stringTCPTransport_closing ) </s>
