Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self):
self.controller = TorState(TorControlProtocol())
self.controller.connectionMade = do_nothing
def test_attacher_error_handler(self):
# make sure error-handling "does something" that isn't blowing up
with patch('sys.stdout'):
TorState(self.protocol)._attacher_error(Failure(RuntimeError("quote")))
def setUp(self):
self.controller = TorState(TorControlProtocol())
self.controller.connectionMade = lambda _: None
'''
from .test_torconfig import FakeControlProtocol
protocol = FakeControlProtocol(
[
"ns/all=", # ns/all
"", # circuit-status
"", # stream-status
"", # address-mappings/all
"entry-guards=\r\n$0000000000000000000000000000000000000000=name up\r\n$1111111111111111111111111111111111111111=foo up\r\n$9999999999999999999999999999999999999999=eman unusable 2012-01-01 22:00:00\r\n", # entry-guards
"99999", # process/pid
"??", # ip-to-country/0.0.0.0
]
)
state = yield TorState.from_protocol(protocol)
self.assertEqual(len(state.entry_guards), 2)
self.assertTrue('$0000000000000000000000000000000000000000' in state.entry_guards)
self.assertTrue('$1111111111111111111111111111111111111111' in state.entry_guards)
self.assertEqual(len(state.unusable_entry_guards), 1)
self.assertTrue('$9999999999999999999999999999999999999999' in state.unusable_entry_guards[0])
d2.addCallback(lambda tpp: txtorcon.TorState(tpp.tor_protocol).post_bootstrap)
d2.addCallback(lambda state: state.protocol)
d2.addCallback(lambda tpp: txtorcon.TorState(tpp.tor_protocol).post_bootstrap)
d2.addCallback(lambda state: state.protocol)
def main(reactor):
# change the port to 9151 for Tor Browser Bundle
tor_ep = TCP4ClientEndpoint(reactor, "localhost", 9051)
connection = yield txtorcon.build_tor_connection(tor_ep, build_state=False)
version = yield connection.get_info('version', 'events/names')
print("Connected to Tor {version}".format(**version))
print("Events:", version['events/names'])
print("Building state.")
state = yield txtorcon.TorState.from_protocol(connection)
print("listening for circuit events")
state.add_circuit_listener(MyCircuitListener())
print("Issuing NEWNYM.")
yield connection.signal('NEWNYM')
print("OK.")
print("Existing circuits:")
for c in state.circuits.values():
print(' ', c)
print("listening for INFO events")
def print_info(i):
print("INFO:", i)
connection.add_event_listener('INFO', print_info)
def setup(processprotocol, options):
proto = processprotocol.tor_protocol
state = txtorcon.TorState(proto)
state.post_bootstrap.addCallback(really_setup, options).addErrback(setup_failed)
@inlineCallbacks
def main(reactor):
config = txtorcon.TorConfig()
config.OrPort = 1234
config.SocksPort = 9999
try:
yield txtorcon.launch_tor(config, reactor, stdout=stdout)
except RuntimeError as e:
print "Error:", e
return
proto = config.protocol
print "Connected to Tor version", proto.version
state = yield txtorcon.TorState.from_protocol(proto)
print "This Tor has PID", state.tor_pid
print "This Tor has the following %d Circuits:" % len(state.circuits)
for c in state.circuits.values():
print c
print "Changing our config (SOCKSPort=9876)"
config.SOCKSPort = 9876
yield config.save()
print "Querying to see it changed:"
socksport = yield proto.get_conf("SOCKSPort")
print "SOCKSPort", socksport