source: trunk/src/allmydata/test/test_tor_provider.py

Last change on this file was fec97256, checked in by Alexandre Detiste <alexandre.detiste@…>, at 2025-01-06T21:51:37Z

trim Python2 syntax
  • Property mode set to 100644
File size: 26.3 KB
Line 
1"""
2Ported to Python 3.
3"""
4
5import os
6from twisted.trial import unittest
7from twisted.internet import defer, error
8from io import StringIO
9from unittest import mock
10from ..util import tor_provider
11from ..scripts import create_node, runner
12from foolscap.eventual import flushEventualQueue
13
14def mock_txtorcon(txtorcon):
15    return mock.patch("allmydata.util.tor_provider._import_txtorcon",
16                      return_value=txtorcon)
17
18def mock_tor(tor):
19    return mock.patch("allmydata.util.tor_provider._import_tor",
20                      return_value=tor)
21
22def make_cli_config(basedir, *argv):
23    parent = runner.Options()
24    cli_config = create_node.CreateNodeOptions()
25    cli_config.parent = parent
26    cli_config.parseOptions(argv)
27    cli_config["basedir"] = basedir
28    cli_config.stdout = StringIO()
29    return cli_config
30
31class TryToConnect(unittest.TestCase):
32    def test_try(self):
33        reactor = object()
34        txtorcon = mock.Mock()
35        tor_state = object()
36        d = defer.succeed(tor_state)
37        txtorcon.build_tor_connection = mock.Mock(return_value=d)
38        ep = object()
39        stdout = StringIO()
40        with mock.patch("allmydata.util.tor_provider.clientFromString",
41                        return_value=ep) as cfs:
42            d = tor_provider._try_to_connect(reactor, "desc", stdout, txtorcon)
43        r = self.successResultOf(d)
44        self.assertIs(r, tor_state)
45        cfs.assert_called_with(reactor, "desc")
46        txtorcon.build_tor_connection.assert_called_with(ep)
47
48    def test_try_handled_error(self):
49        reactor = object()
50        txtorcon = mock.Mock()
51        d = defer.fail(error.ConnectError("oops"))
52        txtorcon.build_tor_connection = mock.Mock(return_value=d)
53        ep = object()
54        stdout = StringIO()
55        with mock.patch("allmydata.util.tor_provider.clientFromString",
56                        return_value=ep) as cfs:
57            d = tor_provider._try_to_connect(reactor, "desc", stdout, txtorcon)
58        r = self.successResultOf(d)
59        self.assertIs(r, None)
60        cfs.assert_called_with(reactor, "desc")
61        txtorcon.build_tor_connection.assert_called_with(ep)
62        self.assertEqual(stdout.getvalue(),
63                         "Unable to reach Tor at 'desc': "
64                         "An error occurred while connecting: oops.\n")
65
66    def test_try_unhandled_error(self):
67        reactor = object()
68        txtorcon = mock.Mock()
69        d = defer.fail(ValueError("oops"))
70        txtorcon.build_tor_connection = mock.Mock(return_value=d)
71        ep = object()
72        stdout = StringIO()
73        with mock.patch("allmydata.util.tor_provider.clientFromString",
74                        return_value=ep) as cfs:
75            d = tor_provider._try_to_connect(reactor, "desc", stdout, txtorcon)
76        f = self.failureResultOf(d)
77        self.assertIsInstance(f.value, ValueError)
78        self.assertEqual(str(f.value), "oops")
79        cfs.assert_called_with(reactor, "desc")
80        txtorcon.build_tor_connection.assert_called_with(ep)
81        self.assertEqual(stdout.getvalue(), "")
82
83class LaunchTor(unittest.TestCase):
84    def _do_test_launch(self, tor_executable):
85        reactor = object()
86        private_dir = "private"
87        txtorcon = mock.Mock()
88        tor = mock.Mock
89        txtorcon.launch = mock.Mock(return_value=tor)
90
91        with mock.patch("allmydata.util.tor_provider.allocate_tcp_port",
92                        return_value=999999):
93            d = tor_provider._launch_tor(reactor, tor_executable, private_dir,
94                                         txtorcon)
95        tor_control_endpoint, tor_result = self.successResultOf(d)
96        self.assertIs(tor_result, tor)
97
98    def test_launch(self):
99        return self._do_test_launch(None)
100    def test_launch_executable(self):
101        return self._do_test_launch("mytor")
102
103class ConnectToTor(unittest.TestCase):
104    def _do_test_connect(self, endpoint, reachable):
105        reactor = object()
106        txtorcon = object()
107        args = []
108        if endpoint:
109            args = ["--tor-control-port=%s" % endpoint]
110        cli_config = make_cli_config("basedir", "--listen=tor", *args)
111        stdout = cli_config.stdout
112        expected_port = "tcp:127.0.0.1:9151"
113        if endpoint:
114            expected_port = endpoint
115        tor_state = mock.Mock
116        tor_state.protocol = object()
117        tried = []
118        def _try_to_connect(reactor, port, stdout, txtorcon):
119            tried.append( (reactor, port, stdout, txtorcon) )
120            if not reachable:
121                return defer.succeed(None)
122            if port == expected_port: # second one on the list
123                return defer.succeed(tor_state)
124            return defer.succeed(None)
125
126        with mock.patch("allmydata.util.tor_provider._try_to_connect",
127                        _try_to_connect):
128            d = tor_provider._connect_to_tor(reactor, cli_config, txtorcon)
129        if not reachable:
130            f = self.failureResultOf(d)
131            self.assertIsInstance(f.value, ValueError)
132            self.assertEqual(str(f.value),
133                             "unable to reach any default Tor control port")
134            return
135        successful_port, tor_control_proto = self.successResultOf(d)
136        self.assertEqual(successful_port, expected_port)
137        self.assertIs(tor_control_proto, tor_state.protocol)
138        expected = [(reactor, "unix:/var/run/tor/control", stdout, txtorcon),
139                    (reactor, "tcp:127.0.0.1:9051", stdout, txtorcon),
140                    (reactor, "tcp:127.0.0.1:9151", stdout, txtorcon),
141                    ]
142        if endpoint:
143            expected = [(reactor, endpoint, stdout, txtorcon)]
144        self.assertEqual(tried, expected)
145
146    def test_connect(self):
147        return self._do_test_connect(None, True)
148    def test_connect_endpoint(self):
149        return self._do_test_connect("tcp:other:port", True)
150    def test_connect_unreachable(self):
151        return self._do_test_connect(None, False)
152
153
154class FakeTor:
155    """Pretends to be a ``txtorcon.Tor`` instance."""
156    def __init__(self):
157        self.protocol = object()
158
159
160class CreateOnion(unittest.TestCase):
161    def test_no_txtorcon(self):
162        with mock.patch("allmydata.util.tor_provider._import_txtorcon",
163                        return_value=None):
164            d = tor_provider.create_config("reactor", "cli_config")
165            f = self.failureResultOf(d)
166            self.assertIsInstance(f.value, ValueError)
167            self.assertEqual(str(f.value),
168                             "Cannot create onion without txtorcon. "
169                             "Please 'pip install tahoe-lafs[tor]' to fix this.")
170
171    def _do_test_launch(self, executable):
172        basedir = self.mktemp()
173        os.mkdir(basedir)
174        private_dir = os.path.join(basedir, "private")
175        os.mkdir(private_dir)
176        reactor = object()
177        args = ["--listen=tor", "--tor-launch"]
178        if executable:
179            args.append("--tor-executable=%s" % executable)
180        cli_config = make_cli_config(basedir, *args)
181        tor_instance = FakeTor()
182        launch_tor = mock.Mock(return_value=defer.succeed(("control_endpoint",
183                                                           tor_instance)))
184        txtorcon = mock.Mock()
185        ehs = mock.Mock()
186        # This appears to be a native string in the real txtorcon object...
187        ehs.private_key = "privkey"
188        ehs.hostname = "ONION.onion"
189        txtorcon.EphemeralHiddenService = mock.Mock(return_value=ehs)
190        ehs.add_to_tor = mock.Mock(return_value=defer.succeed(None))
191        ehs.remove_from_tor = mock.Mock(return_value=defer.succeed(None))
192
193        with mock_txtorcon(txtorcon):
194            with mock.patch("allmydata.util.tor_provider._launch_tor",
195                            launch_tor):
196                with mock.patch("allmydata.util.tor_provider.allocate_tcp_port",
197                                return_value=999999):
198                    d = tor_provider.create_config(reactor, cli_config)
199                    tor_config = self.successResultOf(d)
200
201        launch_tor.assert_called_with(reactor, executable,
202                                      os.path.abspath(private_dir), txtorcon)
203        txtorcon.EphemeralHiddenService.assert_called_with("3457 127.0.0.1:999999")
204        ehs.add_to_tor.assert_called_with(tor_instance.protocol)
205        ehs.remove_from_tor.assert_called_with(tor_instance.protocol)
206
207        expected = {"launch": "true",
208                    "onion": "true",
209                    "onion.local_port": "999999",
210                    "onion.external_port": "3457",
211                    "onion.private_key_file": os.path.join("private",
212                                                           "tor_onion.privkey"),
213                    }
214        if executable:
215            expected["tor.executable"] = executable
216        self.assertEqual(dict(tor_config.node_config["tor"]), expected)
217        self.assertEqual(tor_config.tub_ports, ["tcp:999999:interface=127.0.0.1"])
218        self.assertEqual(tor_config.tub_locations, ["tor:ONION.onion:3457"])
219        fn = os.path.join(basedir, dict(tor_config.node_config["tor"])["onion.private_key_file"])
220        with open(fn, "rb") as f:
221            privkey = f.read()
222        self.assertEqual(privkey, b"privkey")
223
224    def test_launch(self):
225        return self._do_test_launch(None)
226    def test_launch_executable(self):
227        return self._do_test_launch("mytor")
228
229    def test_control_endpoint(self):
230        basedir = self.mktemp()
231        os.mkdir(basedir)
232        private_dir = os.path.join(basedir, "private")
233        os.mkdir(private_dir)
234        reactor = object()
235        cli_config = make_cli_config(basedir, "--listen=tor")
236        protocol = object()
237        connect_to_tor = mock.Mock(return_value=defer.succeed(("goodport",
238                                                               protocol)))
239        txtorcon = mock.Mock()
240        ehs = mock.Mock()
241        ehs.private_key = b"privkey"
242        ehs.hostname = "ONION.onion"
243        txtorcon.EphemeralHiddenService = mock.Mock(return_value=ehs)
244        ehs.add_to_tor = mock.Mock(return_value=defer.succeed(None))
245        ehs.remove_from_tor = mock.Mock(return_value=defer.succeed(None))
246
247        with mock_txtorcon(txtorcon):
248            with mock.patch("allmydata.util.tor_provider._connect_to_tor",
249                            connect_to_tor):
250                with mock.patch("allmydata.util.tor_provider.allocate_tcp_port",
251                                return_value=999999):
252                    d = tor_provider.create_config(reactor, cli_config)
253                    tor_config = self.successResultOf(d)
254
255        connect_to_tor.assert_called_with(reactor, cli_config, txtorcon)
256        txtorcon.EphemeralHiddenService.assert_called_with("3457 127.0.0.1:999999")
257        ehs.add_to_tor.assert_called_with(protocol)
258        ehs.remove_from_tor.assert_called_with(protocol)
259
260        expected = {"control.port": "goodport",
261                    "onion": "true",
262                    "onion.local_port": "999999",
263                    "onion.external_port": "3457",
264                    "onion.private_key_file": os.path.join("private",
265                                                           "tor_onion.privkey"),
266                    }
267        self.assertEqual(dict(tor_config.node_config["tor"]), expected)
268        self.assertEqual(tor_config.tub_ports, ["tcp:999999:interface=127.0.0.1"])
269        self.assertEqual(tor_config.tub_locations, ["tor:ONION.onion:3457"])
270        fn = os.path.join(basedir, dict(tor_config.node_config["tor"])["onion.private_key_file"])
271        with open(fn, "rb") as f:
272            privkey = f.read()
273        self.assertEqual(privkey, b"privkey")
274
275
276_None = object()
277class FakeConfig(dict):
278    def get_config(self, section, option, default=_None, boolean=False):
279        if section != "tor":
280            raise ValueError(section)
281        value = self.get(option, default)
282        if value is _None:
283            raise KeyError
284        return value
285
286    def get_config_path(self, *args):
287        return os.path.join(self.get("basedir", "basedir"), *args)
288
289
290class EmptyContext:
291    def __init__(self):
292        pass
293    def __enter__(self):
294        pass
295    def __exit__(self, type, value, traceback):
296        pass
297
298class Provider(unittest.TestCase):
299    def test_build(self):
300        tor_provider.create("reactor", FakeConfig())
301
302    def test_handler_disabled(self):
303        p = tor_provider.create("reactor", FakeConfig(enabled=False))
304        self.assertEqual(p.get_tor_handler(), None)
305
306    def test_handler_no_tor(self):
307        with mock_tor(None):
308            p = tor_provider.create("reactor", FakeConfig())
309        self.assertEqual(p.get_tor_handler(), None)
310
311    def test_handler_launch_no_txtorcon(self):
312        with mock_txtorcon(None):
313            p = tor_provider.create("reactor", FakeConfig(launch=True))
314        self.assertEqual(p.get_tor_handler(), None)
315
316    @defer.inlineCallbacks
317    def test_handler_launch(self):
318        reactor = object()
319        tor = mock.Mock()
320        txtorcon = mock.Mock()
321        handler = object()
322        tor.control_endpoint_maker = mock.Mock(return_value=handler)
323        tor.add_context = mock.Mock(return_value=EmptyContext())
324        with mock_tor(tor):
325            with mock_txtorcon(txtorcon):
326                p = tor_provider.create(reactor, FakeConfig(launch=True))
327        h = p.get_tor_handler()
328        self.assertIs(h, handler)
329        tor.control_endpoint_maker.assert_called_with(p._make_control_endpoint,
330                                                      takes_status=True)
331
332        # make sure Tor is launched just once, the first time an endpoint is
333        # requested, and never again. The clientFromString() function is
334        # called once each time.
335
336        ep_desc = object()
337        launch_tor = mock.Mock(return_value=defer.succeed((ep_desc,None)))
338        ep = object()
339        cfs = mock.Mock(return_value=ep)
340        with mock.patch("allmydata.util.tor_provider._launch_tor", launch_tor):
341            with mock.patch("allmydata.util.tor_provider.clientFromString", cfs):
342                d = p._make_control_endpoint(reactor,
343                                             update_status=lambda status: None)
344                yield flushEventualQueue()
345                self.assertIs(self.successResultOf(d), ep)
346                launch_tor.assert_called_with(reactor, None,
347                                              os.path.join("basedir", "private"),
348                                              txtorcon)
349                cfs.assert_called_with(reactor, ep_desc)
350
351        launch_tor2 = mock.Mock(return_value=defer.succeed((ep_desc,None)))
352        cfs2 = mock.Mock(return_value=ep)
353        with mock.patch("allmydata.util.tor_provider._launch_tor", launch_tor2):
354            with mock.patch("allmydata.util.tor_provider.clientFromString", cfs2):
355                d2 = p._make_control_endpoint(reactor,
356                                              update_status=lambda status: None)
357                yield flushEventualQueue()
358                self.assertIs(self.successResultOf(d2), ep)
359                self.assertEqual(launch_tor2.mock_calls, [])
360                cfs2.assert_called_with(reactor, ep_desc)
361
362    def test_handler_socks_endpoint(self):
363        """
364        If not configured otherwise, the Tor provider returns a Socks-based
365        handler.
366        """
367        tor = mock.Mock()
368        handler = object()
369        tor.socks_endpoint = mock.Mock(return_value=handler)
370        ep = object()
371        cfs = mock.Mock(return_value=ep)
372        reactor = object()
373
374        with mock_tor(tor):
375            p = tor_provider.create(reactor,
376                                    FakeConfig(**{"socks.port": "ep_desc"}))
377            with mock.patch("allmydata.util.tor_provider.clientFromString", cfs):
378                h = p.get_tor_handler()
379        cfs.assert_called_with(reactor, "ep_desc")
380        tor.socks_endpoint.assert_called_with(ep)
381        self.assertIs(h, handler)
382
383    def test_handler_socks_unix_endpoint(self):
384        """
385        ``socks.port`` can be configured as a UNIX client endpoint.
386        """
387        tor = mock.Mock()
388        handler = object()
389        tor.socks_endpoint = mock.Mock(return_value=handler)
390        ep = object()
391        cfs = mock.Mock(return_value=ep)
392        reactor = object()
393
394        with mock_tor(tor):
395            p = tor_provider.create(reactor,
396                                    FakeConfig(**{"socks.port": "unix:path"}))
397            with mock.patch("allmydata.util.tor_provider.clientFromString", cfs):
398                h = p.get_tor_handler()
399        cfs.assert_called_with(reactor, "unix:path")
400        tor.socks_endpoint.assert_called_with(ep)
401        self.assertIs(h, handler)
402
403    def test_handler_socks_tcp_endpoint(self):
404        """
405        ``socks.port`` can be configured as a UNIX client endpoint.
406        """
407        tor = mock.Mock()
408        handler = object()
409        tor.socks_endpoint = mock.Mock(return_value=handler)
410        ep = object()
411        cfs = mock.Mock(return_value=ep)
412        reactor = object()
413
414        with mock_tor(tor):
415            p = tor_provider.create(reactor,
416                                    FakeConfig(**{"socks.port": "tcp:127.0.0.1:1234"}))
417            with mock.patch("allmydata.util.tor_provider.clientFromString", cfs):
418                h = p.get_tor_handler()
419        cfs.assert_called_with(reactor, "tcp:127.0.0.1:1234")
420        tor.socks_endpoint.assert_called_with(ep)
421        self.assertIs(h, handler)
422
423    def test_handler_control_endpoint(self):
424        tor = mock.Mock()
425        handler = object()
426        tor.control_endpoint = mock.Mock(return_value=handler)
427        ep = object()
428        cfs = mock.Mock(return_value=ep)
429        reactor = object()
430
431        with mock_tor(tor):
432            p = tor_provider.create(reactor,
433                                    FakeConfig(**{"control.port": "ep_desc"}))
434            with mock.patch("allmydata.util.tor_provider.clientFromString", cfs):
435                h = p.get_tor_handler()
436        self.assertIs(h, handler)
437        cfs.assert_called_with(reactor, "ep_desc")
438        tor.control_endpoint.assert_called_with(ep)
439
440    def test_handler_default(self):
441        tor = mock.Mock()
442        handler = object()
443        tor.default_socks = mock.Mock(return_value=handler)
444
445        with mock_tor(tor):
446            p = tor_provider.create("reactor", FakeConfig())
447            h = p.get_tor_handler()
448        self.assertIs(h, handler)
449        tor.default_socks.assert_called_with()
450
451class ProviderListener(unittest.TestCase):
452    def test_listener(self):
453        """Does the Tor Provider object's get_listener() method correctly
454        convert the [tor] section of tahoe.cfg into an
455        endpoint/descriptor?
456        """
457        tor = mock.Mock()
458        handler = object()
459        tor.socks_endpoint = mock.Mock(return_value=handler)
460        reactor = object()
461
462        with mock_tor(tor):
463            p = tor_provider.create(reactor,
464                                    FakeConfig(**{"onion.local_port": "321"}))
465        fake_ep = object()
466        with mock.patch("allmydata.util.tor_provider.TCP4ServerEndpoint",
467                        return_value=fake_ep) as e:
468            endpoint_or_description = p.get_listener()
469        self.assertIs(endpoint_or_description, fake_ep)
470        self.assertEqual(e.mock_calls, [mock.call(reactor, 321,
471                                                  interface="127.0.0.1")])
472
473class Provider_CheckOnionConfig(unittest.TestCase):
474    def test_default(self):
475        # default config doesn't start an onion service, so it should be
476        # happy both with and without txtorcon
477
478        p = tor_provider.create("reactor", FakeConfig())
479        p.check_onion_config()
480
481        with mock_txtorcon(None):
482            p = tor_provider.create("reactor", FakeConfig())
483            p.check_onion_config()
484
485    def test_no_txtorcon(self):
486        with mock_txtorcon(None):
487            with self.assertRaises(ValueError) as ctx:
488                tor_provider.create("reactor", FakeConfig(onion=True))
489            self.assertEqual(
490                str(ctx.exception),
491                "Cannot create onion without txtorcon. "
492                "Please 'pip install tahoe-lafs[tor]' to fix."
493            )
494
495    def test_no_launch_no_control(self):
496        """
497        onion=true but no way to launch/connect to tor
498        """
499        with self.assertRaises(ValueError) as ctx:
500            tor_provider.create("reactor", FakeConfig(onion=True))
501        self.assertEqual(
502            str(ctx.exception),
503            "[tor] onion = true, but we have neither "
504            "launch=true nor control.port="
505        )
506
507    def test_onion_no_local_port(self):
508        """
509        onion=true but no local_port configured is an error
510        """
511        with self.assertRaises(ValueError) as ctx:
512            tor_provider.create("reactor", FakeConfig(onion=True, launch=True))
513        self.assertEqual(
514            str(ctx.exception),
515            "[tor] onion = true, "
516            "but onion.local_port= is missing"
517        )
518
519    def test_onion_no_external_port(self):
520        """
521        onion=true but no external_port configured is an error
522        """
523        with self.assertRaises(ValueError) as ctx:
524            tor_provider.create("reactor",
525                                FakeConfig(onion=True, launch=True,
526                                           **{"onion.local_port": "x",
527                                           }))
528        self.assertEqual(
529            str(ctx.exception),
530            "[tor] onion = true, but onion.external_port= is missing"
531        )
532
533    def test_onion_no_private_key_file(self):
534        """
535        onion=true but no private_key_file configured is an error
536        """
537        with self.assertRaises(ValueError) as ctx:
538            tor_provider.create("reactor",
539                                FakeConfig(onion=True, launch=True,
540                                           **{"onion.local_port": "x",
541                                              "onion.external_port": "y",
542                                           }))
543        self.assertEqual(
544            str(ctx.exception),
545            "[tor] onion = true, but onion.private_key_file= is missing"
546        )
547
548    def test_ok(self):
549        p = tor_provider.create("reactor",
550                                FakeConfig(onion=True, launch=True,
551                                           **{"onion.local_port": "x",
552                                              "onion.external_port": "y",
553                                              "onion.private_key_file": "z",
554                                           }))
555        p.check_onion_config()
556
557
558class Provider_Service(unittest.TestCase):
559    def test_no_onion(self):
560        reactor = object()
561        p = tor_provider.create(reactor, FakeConfig(onion=False))
562        with mock.patch("allmydata.util.tor_provider._Provider._start_onion") as s:
563            p.startService()
564        self.assertEqual(s.mock_calls, [])
565        self.assertEqual(p.running, True)
566
567        p.stopService()
568        self.assertEqual(p.running, False)
569
570    @defer.inlineCallbacks
571    def test_launch(self):
572        basedir = self.mktemp()
573        os.mkdir(basedir)
574        fn = os.path.join(basedir, "keyfile")
575        with open(fn, "w") as f:
576            f.write("private key")
577        reactor = object()
578        cfg = FakeConfig(basedir=basedir, onion=True, launch=True,
579                         **{"onion.local_port": 123,
580                            "onion.external_port": 456,
581                            "onion.private_key_file": "keyfile",
582                            })
583
584        txtorcon = mock.Mock()
585        with mock_txtorcon(txtorcon):
586            p = tor_provider.create(reactor, cfg)
587        tor_instance = FakeTor()
588        tor_state = mock.Mock()
589        tor_state.protocol = tor_instance.protocol
590        ehs = mock.Mock()
591        ehs.add_to_tor = mock.Mock(return_value=defer.succeed(None))
592        ehs.remove_from_tor = mock.Mock(return_value=defer.succeed(None))
593        txtorcon.EphemeralHiddenService = mock.Mock(return_value=ehs)
594        launch_tor = mock.Mock(return_value=defer.succeed((None,tor_instance)))
595        with mock.patch("allmydata.util.tor_provider._launch_tor",
596                        launch_tor):
597            d = p.startService()
598            yield flushEventualQueue()
599        self.successResultOf(d)
600        self.assertIs(p._onion_ehs, ehs)
601        self.assertIs(p._onion_tor_control_proto, tor_state.protocol)
602        launch_tor.assert_called_with(reactor, None,
603                                      os.path.join(basedir, "private"), txtorcon)
604        txtorcon.EphemeralHiddenService.assert_called_with("456 127.0.0.1:123",
605                                                           b"private key")
606        ehs.add_to_tor.assert_called_with(tor_state.protocol)
607
608        yield p.stopService()
609        ehs.remove_from_tor.assert_called_with(tor_state.protocol)
610
611    @defer.inlineCallbacks
612    def test_control_endpoint(self):
613        basedir = self.mktemp()
614        os.mkdir(basedir)
615        fn = os.path.join(basedir, "keyfile")
616        with open(fn, "w") as f:
617            f.write("private key")
618        reactor = object()
619        cfg = FakeConfig(basedir=basedir, onion=True,
620                         **{"control.port": "ep_desc",
621                            "onion.local_port": 123,
622                            "onion.external_port": 456,
623                            "onion.private_key_file": "keyfile",
624                            })
625
626        txtorcon = mock.Mock()
627        with mock_txtorcon(txtorcon):
628            p = tor_provider.create(reactor, cfg)
629        tor_instance = FakeTor()
630        txtorcon.connect = mock.Mock(return_value=tor_instance)
631        ehs = mock.Mock()
632        ehs.add_to_tor = mock.Mock(return_value=defer.succeed(None))
633        ehs.remove_from_tor = mock.Mock(return_value=defer.succeed(None))
634        txtorcon.EphemeralHiddenService = mock.Mock(return_value=ehs)
635        tcep = object()
636        cfs = mock.Mock(return_value=tcep)
637        with mock.patch("allmydata.util.tor_provider.clientFromString", cfs):
638            d = p.startService()
639            yield flushEventualQueue()
640        self.successResultOf(d)
641        self.assertIs(p._onion_ehs, ehs)
642        self.assertIs(p._onion_tor_control_proto, tor_instance.protocol)
643        cfs.assert_called_with(reactor, "ep_desc")
644        txtorcon.connect.assert_called_with(reactor, tcep)
645        txtorcon.EphemeralHiddenService.assert_called_with("456 127.0.0.1:123",
646                                                           b"private key")
647        ehs.add_to_tor.assert_called_with(tor_instance.protocol)
648
649        yield p.stopService()
650        ehs.remove_from_tor.assert_called_with(tor_instance.protocol)
Note: See TracBrowser for help on using the repository browser.