Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_two_hidden_services_before_save(self):
conf = TorConfig()
conf.HiddenServices = [HiddenService(conf, '/fake/path', ['80 127.0.0.1:1234'])]
conf.HiddenServices.append(HiddenService(conf, '/fake/path/two', ['1234 127.0.0.1:1234']))
conf.save()
self.assertEqual(2, len(conf.HiddenServices))
def test_multiple_append(self):
conf = TorConfig()
h0 = HiddenService(conf, '/fake/path', ['80 127.0.0.1:1234'], '', 3)
h1 = HiddenService(conf, '/fake/path', ['90 127.0.0.1:4321'], '', 3)
h2 = HiddenService(conf, '/fake/path', ['90 127.0.0.1:5432'], '', 3, True)
conf.hiddenservices = [h0]
conf.hiddenservices.append(h1)
conf.hiddenservices.append(h2)
self.assertEqual(len(conf.hiddenservices), 3)
self.assertEqual(h0, conf.hiddenservices[0])
self.assertEqual(h1, conf.hiddenservices[1])
self.assertEqual(h2, conf.hiddenservices[2])
self.assertTrue(conf.needs_save())
def test_hidden_service_same_directory(self):
conf = TorConfig(FakeControlProtocol(['config/names=']))
servicelines = '''HiddenServiceDir=/fake/path
HiddenServiceDir=/fake/path'''
self.assertRaises(RuntimeError, conf._setup_hidden_services, servicelines)
conf = TorConfig()
conf.HiddenServices = [HiddenService(conf, '/fake/path', ['80 127.0.0.1:1234'])]
conf.HiddenServices.append(HiddenService(conf, '/fake/path', ['80 127.0.0.1:1234']))
self.assertTrue(conf.needs_save())
self.assertRaises(RuntimeError, conf.save)
conf = TorConfig()
conf.HiddenServices = [HiddenService(conf, '/fake/path', ['80 127.0.0.1:1234'])]
conf.HiddenServices.append(HiddenService(conf, '/fake/path/two', ['80 127.0.0.1:1234']))
self.assertTrue(conf.needs_save())
conf.save()
conf.hiddenservices[1].dir = '/fake/path'
self.assertTrue(conf.needs_save())
self.assertRaises(RuntimeError, conf.save)
def test_options_hidden(self):
self.protocol.answers.append(
'HiddenServiceDir=/fake/path\nHiddenServicePort=80 '
'127.0.0.1:1234\nHiddenServiceDirGroupReadable=1\n'
)
conf = TorConfig(self.protocol)
yield conf.post_bootstrap
self.assertTrue(conf.post_bootstrap.called)
self.assertTrue('HiddenServiceOptions' not in conf.config)
self.assertTrue('HiddenServices' in conf.config)
self.assertEqual(len(conf.HiddenServices), 1)
self.assertTrue(not conf.needs_save())
conf.hiddenservices.append(
HiddenService(conf, '/some/dir', '80 127.0.0.1:2345', 'auth', 2, True)
)
conf.hiddenservices[0].ports.append('443 127.0.0.1:443')
self.assertTrue(conf.needs_save())
conf.save()
self.assertEqual(len(self.protocol.sets), 9)
self.assertEqual(self.protocol.sets[0], ('HiddenServiceDir', '/fake/path'))
self.assertEqual(self.protocol.sets[1], ('HiddenServiceDirGroupReadable', '1'))
self.assertEqual(self.protocol.sets[2], ('HiddenServicePort', '80 127.0.0.1:1234'))
self.assertEqual(self.protocol.sets[3], ('HiddenServicePort', '443 127.0.0.1:443'))
self.assertEqual(self.protocol.sets[4], ('HiddenServiceDir', '/some/dir'))
self.assertEqual(self.protocol.sets[5], ('HiddenServiceDirGroupReadable', '1'))
self.assertEqual(self.protocol.sets[6], ('HiddenServicePort', '80 127.0.0.1:2345'))
self.assertEqual(self.protocol.sets[7], ('HiddenServiceVersion', '2'))
self.assertEqual(self.protocol.sets[8], ('HiddenServiceAuthorizeClient', 'auth'))
def test_save_no_protocol(self):
conf = TorConfig()
conf.HiddenServices = [HiddenService(conf, '/fake/path', ['80 127.0.0.1:1234'])]
conf.save()
def test_stealth_clients(self):
# FIXME test without crapping on filesystem
self.protocol.answers.append('HiddenServiceDir=/fake/path\n')
d = tempfile.mkdtemp()
try:
with open(os.path.join(d, 'hostname'), 'w') as f:
f.write('oniona cookiea\n')
f.write('onionb cookieb\n')
conf = TorConfig(self.protocol)
hs = HiddenService(conf, d, [])
self.assertEqual(2, len(hs.clients))
self.assertEqual('oniona', hs.clients[0][0])
self.assertEqual('cookiea', hs.clients[0][1])
self.assertEqual('onionb', hs.clients[1][0])
self.assertEqual('cookieb', hs.clients[1][1])
self.assertRaises(RuntimeError, getattr, hs, 'hostname')
finally:
shutil.rmtree(d, ignore_errors=True)
def test_add_hidden_service_to_empty_config(self):
conf = TorConfig()
h = HiddenService(conf, '/fake/path', ['80 127.0.0.1:1234'], '', 3)
conf.HiddenServices.append(h)
self.assertEqual(len(conf.hiddenservices), 1)
self.assertEqual(h, conf.hiddenservices[0])
self.assertTrue(conf.needs_save())
def test_single_client(self):
# FIXME test without crapping on filesystem
self.protocol.answers.append('HiddenServiceDir=/fake/path\n')
d = tempfile.mkdtemp()
try:
with open(os.path.join(d, 'hostname'), 'w') as f:
f.write('gobledegook\n')
conf = TorConfig(self.protocol)
hs = HiddenService(conf, d, [])
self.assertEqual(1, len(hs.clients))
self.assertEqual('default', hs.clients[0][0])
self.assertEqual('gobledegook', hs.clients[0][1])
finally:
shutil.rmtree(d, ignore_errors=True)
)
# Create a directory to hold our hidden service. Twisted will unlink it
# when we exit.
hs_temp = tempfile.mkdtemp(prefix='torhiddenservice')
reactor.addSystemEventTrigger(
'before', 'shutdown',
functools.partial(txtorcon.util.delete_file_or_tree, hs_temp)
)
# Add the hidden service to a blank configuration
config = txtorcon.TorConfig()
config.SOCKSPort = 0
config.ORPort = 9089
config.HiddenServices = [
txtorcon.HiddenService(
config, hs_temp,
ports=['%i %s:%i' % (hs_public_port, web_host, web_port)]
)
]
config.save()
# Now launch tor
# Notice that we use a partial function as a callback so we have a
# reference to the config object when tor is fully running.
tordeferred = txtorcon.launch_tor(config, reactor,
progress_updates=print_tor_updates)
tordeferred.addCallback(functools.partial(setup_complete, config,
hs_public_port))
tordeferred.addErrback(setup_failed)
reactor.run()
)
)
# 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')
hs_endpoint.listen(site)
# we've got our Twisted service listening locally and our options
# ready to go, so we now launch Tor. Once it's done (see above