Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
trans = FakeProcessTransportNoProtocol()
trans.protocol = self.protocol
def creator(*args, **kw):
print("Bad: connection creator called")
self.fail()
def on_protocol(proto):
self.process_proto = proto
proto.outReceived(b'Bootstrapped 90%\n')
proto.outReceived(b'Bootstrapped 100%\n')
reactor = FakeReactor(self, trans, on_protocol, [9052, 9999])
tor = yield launch(
reactor=reactor,
connection_creator=creator,
tor_binary='/bin/echo',
socks_port=1234,
control_port=0,
)
self.assertEqual(tor._process_protocol, self.process_proto)
d = tor.quit()
reactor.advance(0)
yield d
errs = self.flushLoggedErrors()
self.assertEqual(1, len(errs))
self.assertTrue("Tor was killed" in str(errs[0]))
self.tor_state.set_attacher(None, reactor)
class DummyResource(Resource):
isLeaf = True
def render_GET(self, request):
size = request.uri.split('/')[-1]
if 'k' in size:
size = int(size[:-1])*(2**10)
elif 'M' in size:
size = int(size[:-1])*(2**20)
# FIXME: to simplify test, return always same data no matter what the size is
return 'a'
return 'a'*size
self.port = yield available_tcp_port(reactor)
self.test_service = yield reactor.listenTCP(
self.port, Site(DummyResource()))
def test_listener_attach_no_remap(self):
"Attachment is via SENTCONNECT on .onion addresses (for example)"
self.circuits[186] = FakeCircuit(186)
listener = Listener([('new', {'target_host': 'www.yahoo.com',
'target_port': 80}),
('attach', {})])
stream = Stream(self)
stream.listen(listener)
stream.update("316 NEW 0 www.yahoo.com:80 SOURCE_ADDR=127.0.0.1:55877 PURPOSE=USER".split())
stream.update("316 SENTCONNECT 186 1.2.3.4:80 SOURCE=EXIT".split())
self.assertEqual(self.circuits[186].streams[0], stream)
def test_circuit_already_valid_in_new(self):
stream = Stream(self)
stream.circuit = FakeCircuit(1)
stream.update("1 NEW 0 94.23.164.42.$43ED8310EB968746970896E8835C2F1991E50B69.exit:9001 SOURCE_ADDR=(Tor_internal):0 PURPOSE=DIR_FETCH".split())
errs = self.flushLoggedErrors()
self.assertEqual(len(errs), 1)
self.assertTrue('Weird' in errs[0].getErrorMessage())
def test_listen_unlisten(self):
self.circuits[186] = FakeCircuit(186)
listener = Listener([])
stream = Stream(self)
stream.listen(listener)
stream.listen(listener)
self.assertEqual(len(stream.listeners), 1)
stream.unlisten(listener)
self.assertEqual(len(stream.listeners), 0)
def test_listener_new(self):
listener = Listener([('new', {'target_port': 9001})])
stream = Stream(self)
stream.listen(listener)
stream.update("1 NEW 0 94.23.164.42.$43ED8310EB968746970896E8835C2F1991E50B69.exit:9001 SOURCE_ADDR=(Tor_internal):0 PURPOSE=DIR_FETCH".split())
def test_magic_circuit_detach(self):
stream = Stream(self)
stream.circuit = FakeCircuit(1)
stream.circuit.streams = [stream]
stream.update("1 SENTCONNECT 0 94.23.164.42.$43ED8310EB968746970896E8835C2F1991E50B69.exit:9001 SOURCE_ADDR=(Tor_internal):0 PURPOSE=DIR_FETCH".split())
self.assertTrue(stream.circuit is None)
def test_stream_changed_with_detach(self):
"Change a stream-id mid-stream, but with a DETACHED message"
self.circuits[123] = FakeCircuit(123)
self.circuits[456] = FakeCircuit(456)
listener = Listener(
[
('new', {'target_host': 'www.yahoo.com', 'target_port': 80}),
('attach', {}),
('detach', {'kwargs': dict(reason='END', remote_reason='MISC')}),
('attach', {})
]
)
stream = Stream(self)
stream.listen(listener)
stream.update("999 NEW 0 www.yahoo.com:80 SOURCE_ADDR=127.0.0.1:55877 PURPOSE=USER".split())
stream.update("999 SENTCONNECT 123 1.2.3.4:80".split())
self.assertEqual(len(self.circuits[123].streams), 1)
self.assertEqual(self.circuits[123].streams[0], stream)
stream.update("999 DETACHED 123 1.2.3.4:80 REASON=END REMOTE_REASON=MISC".split())
self.assertEqual(len(self.circuits[123].streams), 0)
stream.update("999 SENTCONNECT 456 1.2.3.4:80 SOURCE=EXIT".split())
self.assertEqual(len(self.circuits[456].streams), 1)
self.assertEqual(self.circuits[456].streams[0], stream)
def test_listener_close(self):
self.circuits[186] = FakeCircuit(186)
listener = Listener(
[
('new', {'target_host': 'www.yahoo.com', 'target_port': 80}),
('attach', {'target_addr': maybe_ip_addr('1.2.3.4')}),
('closed', {'kwargs': dict(REASON='END', REMOTE_REASON='DONE')})
]
)
stream = Stream(self)
stream.listen(listener)
stream.update("316 NEW 0 www.yahoo.com:80 SOURCE_ADDR=127.0.0.1:55877 PURPOSE=USER".split())
stream.update("316 REMAP 186 1.2.3.4:80 SOURCE=EXIT".split())
stream.update("316 CLOSED 186 1.2.3.4:80 REASON=END REMOTE_REASON=DONE".split())
self.assertEqual(len(self.circuits[186].streams), 0)
def test_str(self):
stream = Stream(self)
stream.update("316 NEW 0 www.yahoo.com:80 SOURCE_ADDR=127.0.0.1:55877 PURPOSE=USER".split())
stream.circuit = FakeCircuit(1)
str(stream)