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):
reactor = Mock()
proto = Mock()
directlyProvides(proto, ITorControlProtocol)
self.cfg = TorConfig()
self.cfg.HidServAuth = ["existing.onion some_token"]
self.tor = Tor(reactor, proto, _tor_config=self.cfg)
def test_ephemeral_ports_bad0(self):
protocol = FakeControlProtocol([])
config = TorConfig(protocol)
with self.assertRaises(ValueError) as ctx:
yield EphemeralAuthenticatedOnionService.create(
Mock(),
config,
ports="80 127.0.0.1:80",
auth=AuthBasic(["xavier"]),
)
self.assertIn(
"'ports' must be a list of strings",
str(ctx.exception),
)
def test_hidden_service_parse_error(self):
conf = TorConfig(FakeControlProtocol(['config/names=']))
try:
conf._setup_hidden_services('''FakeHiddenServiceKey=foo''')
self.fail()
except RuntimeError as e:
self.assertTrue('parse' in str(e))
def test_float_parser_error(self):
self.protocol.answers.append('config/names=\nfoo Float')
self.protocol.answers.append({'foo': '1.23fff'})
cfg = TorConfig(self.protocol)
return self.assertFailure(cfg.post_bootstrap, ValueError)
def test_ephemeral_key_whitespace0(self):
protocol = FakeControlProtocol([])
config = TorConfig(protocol)
d = EphemeralOnionService.create(
Mock(),
config,
ports=["80 127.0.0.1:80"],
private_key=_test_private_key_blob + '\r',
detach=True,
)
return self.assertFailure(d, ValueError)
def test_log_double_save(self):
self.protocol.answers.append(
'config/names=\nLog LineList\nFoo String'''
)
self.protocol.answers.append(
{'Log': 'notice file /var/log/tor/notices.log'}
)
self.protocol.answers.append({'Foo': 'foo'})
conf = TorConfig(self.protocol)
conf.log.append('info file /tmp/foo.log')
conf.foo = 'bar'
self.assertTrue(conf.needs_save())
conf.save()
conf.save() # just for the code coverage...
self.assertTrue(not conf.needs_save())
self.protocol.sets = []
conf.save()
self.assertEqual(self.protocol.sets, [])
def really_setup(state):
print 'Connected to a Tor version %s' % state.protocol.version
probe = CircuitProber(reactor, state)
state.add_circuit_listener(probe)
task.LoopingCall(probe.print_update).start(60.0)
def setup_failed(arg):
print "SETUP FAILED",arg
print arg
reactor.stop()
def update(percent, tag, summary):
print " %d%% %s" % (int(percent), summary)
print "Launching new Tor instance:"
config = txtorcon.TorConfig()
config.ControlPort = 9876
config.ORPort = 9002
config.SocksPort = 0
d = txtorcon.launch_tor(config, reactor, progress_updates=update)
d.addCallback(setup).addErrback(setup_failed)
reactor.run()
@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():
'before', 'shutdown',
functools.partial(
txtorcon.util.delete_file_or_tree,
hs_temp
)
)
# configure the hidden service we want.
# obviously, we'd want a more-persistent place to keep the hidden
# service directory for a "real" setup. If the directory is empty at
# startup as here, Tor creates new keys etcetera (which IS the .onion
# address). That is, every time you run this script you get a new
# hidden service URI, which is probably not what you want.
# The launch_tor method adds other needed config directives to give
# us a minimal config.
config = txtorcon.TorConfig()
config.SOCKSPort = 0
config.ORPort = 9089
config.HiddenServices = [
txtorcon.HiddenService(
config,
hs_temp,
["%d 127.0.0.1:%d" % (hs_public_port, hs_port)]
)
]
config.save()
# next we set up our service to listen on hs_port which is forwarded
# (via the HiddenService options) from the hidden service address on
# port hs_public_port
site = server.Site(Simple())
hs_endpoint = TCP4ServerEndpoint(reactor, hs_port, interface='127.0.0.1')
def main(reactor):
config = txtorcon.TorConfig()
config.ORPort = 0
config.SocksPort = 9998
try:
os.mkdir('tor-data')
except OSError:
pass
config.DataDirectory = './tor-data'
try:
process = yield txtorcon.launch_tor(
config, reactor, progress_updates=progress
)
except Exception as e:
print("Error launching tor:", e)
return