summaryrefslogtreecommitdiff
path: root/src/leap/bitmask/provider/tests/test_providerbootstrapper.py
blob: fe5b52bd1f6c5915e86474fa8cd31415b6cefbee (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
# -*- coding: utf-8 -*-
# test_providerbootstrapper.py
# Copyright (C) 2013 LEAP
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
Tests for the Provider Boostrapper checks

These will be whitebox tests since we want to make sure the private
implementation is checking what we expect.
"""
import os
import mock
import socket
import stat
import tempfile
import time
import requests
try:
    import unittest2 as unittest
except ImportError:
    import unittest

from nose.twistedtools import deferred, reactor
from twisted.internet import threads
from requests.models import Response

from leap.bitmask.config.providerconfig import ProviderConfig
from leap.bitmask.crypto.tests import fake_provider
from leap.bitmask.provider.providerbootstrapper import ProviderBootstrapper
from leap.bitmask.provider.providerbootstrapper import UnsupportedProviderAPI
from leap.bitmask.provider.providerbootstrapper import WrongFingerprint
from leap.bitmask.provider.supportedapis import SupportedAPIs
from leap.bitmask import util
from leap.bitmask.util import get_path_prefix
from leap.common.files import mkdir_p
from leap.common.testing.https_server import where
from leap.common.testing.basetest import BaseLeapTest


class ProviderBootstrapperTest(BaseLeapTest):
    def setUp(self):
        self.pb = ProviderBootstrapper()

    def tearDown(self):
        pass

    def test_name_resolution_check(self):
        # Something highly likely to success
        self.pb._domain = "google.com"
        self.pb._check_name_resolution()
        # Something highly likely to fail
        self.pb._domain = "uquhqweuihowquie.abc.def"

        # In python 2.7.4 raises socket.error
        # In python 2.7.5 raises socket.gaierror
        with self.assertRaises((socket.gaierror, socket.error)):
            self.pb._check_name_resolution()

    @deferred()
    def test_run_provider_select_checks(self):
        self.pb._check_name_resolution = mock.MagicMock()
        self.pb._check_https = mock.MagicMock()
        self.pb._download_provider_info = mock.MagicMock()

        d = self.pb.run_provider_select_checks("somedomain")

        def check(*args):
            self.pb._check_name_resolution.assert_called_once_with()
            self.pb._check_https.assert_called_once_with(None)
            self.pb._download_provider_info.assert_called_once_with(None)
        d.addCallback(check)
        return d

    @deferred()
    def test_run_provider_setup_checks(self):
        self.pb._download_ca_cert = mock.MagicMock()
        self.pb._check_ca_fingerprint = mock.MagicMock()
        self.pb._check_api_certificate = mock.MagicMock()

        d = self.pb.run_provider_setup_checks(ProviderConfig())

        def check(*args):
            self.pb._download_ca_cert.assert_called_once_with()
            self.pb._check_ca_fingerprint.assert_called_once_with(None)
            self.pb._check_api_certificate.assert_called_once_with(None)
        d.addCallback(check)
        return d

    def test_should_proceed_cert(self):
        self.pb._provider_config = mock.Mock()
        self.pb._provider_config.get_ca_cert_path = mock.MagicMock(
            return_value=where("cacert.pem"))

        self.pb._download_if_needed = False
        self.assertTrue(self.pb._should_proceed_cert())

        self.pb._download_if_needed = True
        self.assertFalse(self.pb._should_proceed_cert())

        self.pb._provider_config.get_ca_cert_path = mock.MagicMock(
            return_value=where("somefilethatdoesntexist.pem"))
        self.assertTrue(self.pb._should_proceed_cert())

    def _check_download_ca_cert(self, should_proceed):
        """
        Helper to check different paths easily for the download ca
        cert check

        :param should_proceed: sets the _should_proceed_cert in the
                               provider bootstrapper being tested
        :type should_proceed: bool

        :returns: The contents of the certificate, the expected
                  content depending on should_proceed, and the mode of
                  the file to be checked by the caller
        :rtype: tuple of str, str, int
        """
        old_content = "NOT THE NEW CERT"
        new_content = "NEW CERT"
        new_cert_path = os.path.join(tempfile.mkdtemp(),
                                     "mynewcert.pem")

        with open(new_cert_path, "w") as c:
            c.write(old_content)

        self.pb._provider_config = mock.Mock()
        self.pb._provider_config.get_ca_cert_path = mock.MagicMock(
            return_value=new_cert_path)
        self.pb._domain = "somedomain"

        self.pb._should_proceed_cert = mock.MagicMock(
            return_value=should_proceed)

        read = None
        content_to_check = None
        mode = None

        with mock.patch('requests.models.Response.content',
                        new_callable=mock.PropertyMock) as \
                content:
            content.return_value = new_content
            response_obj = Response()
            response_obj.raise_for_status = mock.MagicMock()

            self.pb._session.get = mock.MagicMock(return_value=response_obj)
            self.pb._download_ca_cert()
            with open(new_cert_path, "r") as nc:
                read = nc.read()
                if should_proceed:
                    content_to_check = new_content
                else:
                    content_to_check = old_content
            mode = stat.S_IMODE(os.stat(new_cert_path).st_mode)

        os.unlink(new_cert_path)
        return read, content_to_check, mode

    def test_download_ca_cert_no_saving(self):
        read, expected_read, mode = self._check_download_ca_cert(False)
        self.assertEqual(read, expected_read)
        self.assertEqual(mode, int("600", 8))

    def test_download_ca_cert_saving(self):
        read, expected_read, mode = self._check_download_ca_cert(True)
        self.assertEqual(read, expected_read)
        self.assertEqual(mode, int("600", 8))

    def test_check_ca_fingerprint_skips(self):
        self.pb._provider_config = mock.Mock()
        self.pb._provider_config.get_ca_cert_fingerprint = mock.MagicMock(
            return_value="")
        self.pb._domain = "somedomain"

        self.pb._should_proceed_cert = mock.MagicMock(return_value=False)

        self.pb._check_ca_fingerprint()
        self.assertFalse(self.pb._provider_config.
                         get_ca_cert_fingerprint.called)

    def test_check_ca_cert_fingerprint_raises_bad_format(self):
        self.pb._provider_config = mock.Mock()
        self.pb._provider_config.get_ca_cert_fingerprint = mock.MagicMock(
            return_value="wrongfprformat!!")
        self.pb._domain = "somedomain"

        self.pb._should_proceed_cert = mock.MagicMock(return_value=True)

        with self.assertRaises(WrongFingerprint):
            self.pb._check_ca_fingerprint()

    # This two hashes different in the last byte, but that's good enough
    # for the tests
    KNOWN_BAD_HASH = "SHA256: 0f17c033115f6b76ff67871872303ff65034efe" \
                     "7dd1b910062ca323eb4da5c7f"
    KNOWN_GOOD_HASH = "SHA256: 0f17c033115f6b76ff67871872303ff65034ef" \
                      "e7dd1b910062ca323eb4da5c7e"
    KNOWN_GOOD_CERT = """
-----BEGIN CERTIFICATE-----
MIIFbzCCA1egAwIBAgIBATANBgkqhkiG9w0BAQ0FADBKMRgwFgYDVQQDDA9CaXRt
YXNrIFJvb3QgQ0ExEDAOBgNVBAoMB0JpdG1hc2sxHDAaBgNVBAsME2h0dHBzOi8v
Yml0bWFzay5uZXQwHhcNMTIxMTA2MDAwMDAwWhcNMjIxMTA2MDAwMDAwWjBKMRgw
FgYDVQQDDA9CaXRtYXNrIFJvb3QgQ0ExEDAOBgNVBAoMB0JpdG1hc2sxHDAaBgNV
BAsME2h0dHBzOi8vYml0bWFzay5uZXQwggIiMA0GCSqGSIb3DQEBAQUAA4ICDwAw
ggIKAoICAQC1eV4YvayaU+maJbWrD4OHo3d7S1BtDlcvkIRS1Fw3iYDjsyDkZxai
dHp4EUasfNQ+EVtXUvtk6170EmLco6Elg8SJBQ27trE6nielPRPCfX3fQzETRfvB
7tNvGw4Jn2YKiYoMD79kkjgyZjkJ2r/bEHUSevmR09BRp86syHZerdNGpXYhcQ84
CA1+V+603GFIHnrP+uQDdssW93rgDNYu+exT+Wj6STfnUkugyjmPRPjL7wh0tzy+
znCeLl4xiV3g9sjPnc7r2EQKd5uaTe3j71sDPF92KRk0SSUndREz+B1+Dbe/RGk4
MEqGFuOzrtsgEhPIX0hplhb0Tgz/rtug+yTT7oJjBa3u20AAOQ38/M99EfdeJvc4
lPFF1XBBLh6X9UKF72an2NuANiX6XPySnJgZ7nZ09RiYZqVwu/qt3DfvLfhboq+0
bQvLUPXrVDr70onv5UDjpmEA/cLmaIqqrduuTkFZOym65/PfAPvpGnt7crQj/Ibl
DEDYZQmP7AS+6zBjoOzNjUGE5r40zWAR1RSi7zliXTu+yfsjXUIhUAWmYR6J3KxB
lfsiHBQ+8dn9kC3YrUexWoOqBiqJOAJzZh5Y1tqgzfh+2nmHSB2dsQRs7rDRRlyy
YMbkpzL9ZsOUO2eTP1mmar6YjCN+rggYjRrX71K2SpBG6b1zZxOG+wIDAQABo2Aw
XjAdBgNVHQ4EFgQUuYGDLL2sswnYpHHvProt1JU+D48wDgYDVR0PAQH/BAQDAgIE
MAwGA1UdEwQFMAMBAf8wHwYDVR0jBBgwFoAUuYGDLL2sswnYpHHvProt1JU+D48w
DQYJKoZIhvcNAQENBQADggIBADeG67vaFcbITGpi51264kHPYPEWaXUa5XYbtmBl
cXYyB6hY5hv/YNuVGJ1gWsDmdeXEyj0j2icGQjYdHRfwhrbEri+h1EZOm1cSBDuY
k/P5+ctHyOXx8IE79DBsZ6IL61UKIaKhqZBfLGYcWu17DVV6+LT+AKtHhOrv3TSj
RnAcKnCbKqXLhUPXpK0eTjPYS2zQGQGIhIy9sQXVXJJJsGrPgMxna1Xw2JikBOCG
htD/JKwt6xBmNwktH0GI/LVtVgSp82Clbn9C4eZN9E5YbVYjLkIEDhpByeC71QhX
EIQ0ZR56bFuJA/CwValBqV/G9gscTPQqd+iETp8yrFpAVHOW+YzSFbxjTEkBte1J
aF0vmbqdMAWLk+LEFPQRptZh0B88igtx6tV5oVd+p5IVRM49poLhuPNJGPvMj99l
mlZ4+AeRUnbOOeAEuvpLJbel4rhwFzmUiGoeTVoPZyMevWcVFq6BMkS+jRR2w0jK
G6b0v5XDHlcFYPOgUrtsOBFJVwbutLvxdk6q37kIFnWCd8L3kmES5q4wjyFK47Co
Ja8zlx64jmMZPg/t3wWqkZgXZ14qnbyG5/lGsj5CwVtfDljrhN0oCWK1FZaUmW3d
69db12/g4f6phldhxiWuGC/W6fCW5kre7nmhshcltqAJJuU47iX+DarBFiIj816e
yV8e
-----END CERTIFICATE-----
"""

    def _prepare_provider_config_with(self, cert_path, cert_hash):
        """
        Mocks the provider config to give the cert_path and cert_hash
        specified

        :param cert_path: path for the certificate
        :type cert_path: str
        :param cert_hash: hash for the certificate as it would appear
                          in the provider config json
        :type cert_hash: str
        """
        self.pb._provider_config = mock.Mock()
        self.pb._provider_config.get_ca_cert_fingerprint = mock.MagicMock(
            return_value=cert_hash)
        self.pb._provider_config.get_ca_cert_path = mock.MagicMock(
            return_value=cert_path)
        self.pb._domain = "somedomain"

    def test_check_ca_fingerprint_checksout(self):
        cert_path = os.path.join(tempfile.mkdtemp(),
                                 "mynewcert.pem")

        with open(cert_path, "w") as c:
            c.write(self.KNOWN_GOOD_CERT)

        self._prepare_provider_config_with(cert_path, self.KNOWN_GOOD_HASH)

        self.pb._should_proceed_cert = mock.MagicMock(return_value=True)

        self.pb._check_ca_fingerprint()

        os.unlink(cert_path)

    def test_check_ca_fingerprint_fails(self):
        cert_path = os.path.join(tempfile.mkdtemp(),
                                 "mynewcert.pem")

        with open(cert_path, "w") as c:
            c.write(self.KNOWN_GOOD_CERT)

        self._prepare_provider_config_with(cert_path, self.KNOWN_BAD_HASH)

        self.pb._should_proceed_cert = mock.MagicMock(return_value=True)

        with self.assertRaises(WrongFingerprint):
            self.pb._check_ca_fingerprint()

        os.unlink(cert_path)


###############################################################################
# Tests with a fake provider                                                  #
###############################################################################

class ProviderBootstrapperActiveTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        factory = fake_provider.get_provider_factory()
        http = reactor.listenTCP(8002, factory)
        https = reactor.listenSSL(
            0, factory,
            fake_provider.OpenSSLServerContextFactory())
        get_port = lambda p: p.getHost().port
        cls.http_port = get_port(http)
        cls.https_port = get_port(https)

    def setUp(self):
        self.pb = ProviderBootstrapper()

        # At certain points we are going to be replacing these methods
        # directly in ProviderConfig to be able to catch calls from
        # new ProviderConfig objects inside the methods tested. We
        # need to save the old implementation and restore it in
        # tearDown so we are sure everything is as expected for each
        # test. If we do it inside each specific test, a failure in
        # the test will leave the implementation with the mock.
        self.old_gpp = util.get_path_prefix

        self.old_load = ProviderConfig.load
        self.old_save = ProviderConfig.save
        self.old_api_version = ProviderConfig.get_api_version
        self.old_api_uri = ProviderConfig.get_api_uri

    def tearDown(self):
        util.get_path_prefix = self.old_gpp
        ProviderConfig.load = self.old_load
        ProviderConfig.save = self.old_save
        ProviderConfig.get_api_version = self.old_api_version
        ProviderConfig.get_api_uri = self.old_api_uri

    def test_check_https_succeeds(self):
        # XXX: Need a proper CA signed cert to test this
        pass

    @deferred()
    def test_check_https_fails(self):
        self.pb._domain = "localhost:%s" % (self.https_port,)

        def check(*args):
            with self.assertRaises(requests.exceptions.SSLError):
                self.pb._check_https()
        return threads.deferToThread(check)

    @deferred()
    def test_second_check_https_fails(self):
        self.pb._domain = "localhost:1234"

        def check(*args):
            with self.assertRaises(Exception):
                self.pb._check_https()
        return threads.deferToThread(check)

    @deferred()
    def test_check_https_succeeds_if_danger(self):
        self.pb._domain = "localhost:%s" % (self.https_port,)
        self.pb._bypass_checks = True

        def check(*args):
            self.pb._check_https()

        return threads.deferToThread(check)

    def _setup_provider_config_with(self, api, path_prefix):
        """
        Sets up the ProviderConfig with mocks for the path prefix, the
        api returned and load/save methods.
        It modifies ProviderConfig directly instead of an object
        because the object used is created in the method itself and we
        cannot control that.

        :param api: API to return
        :type api: str
        :param path_prefix: path prefix to be used when calculating
                            paths
        :type path_prefix: str
        """
        util.get_path_prefix = mock.MagicMock(return_value=path_prefix)
        ProviderConfig.get_api_version = mock.MagicMock(
            return_value=api)
        ProviderConfig.get_api_uri = mock.MagicMock(
            return_value="https://localhost:%s" % (self.https_port,))
        ProviderConfig.load = mock.MagicMock()
        ProviderConfig.save = mock.MagicMock()

    def _setup_providerbootstrapper(self, ifneeded):
        """
        Sets the provider bootstrapper's domain to
        localhost:https_port, sets it to bypass https checks and sets
        the download if needed based on the ifneeded value.

        :param ifneeded: Value for _download_if_needed
        :type ifneeded: bool
        """
        self.pb._domain = "localhost:%s" % (self.https_port,)
        self.pb._bypass_checks = True
        self.pb._download_if_needed = ifneeded

    def _produce_dummy_provider_json(self):
        """
        Creates a dummy provider json on disk in order to test
        behaviour around it (download if newer online, etc)

        :returns: the provider.json path used
        :rtype: str
        """
        provider_dir = os.path.join(get_path_prefix(),
                                    "leap", "providers",
                                    self.pb._domain)
        mkdir_p(provider_dir)
        provider_path = os.path.join(provider_dir,
                                     "provider.json")

        with open(provider_path, "w") as p:
            p.write("A")
        return provider_path

    @mock.patch(
        'leap.bitmask.config.providerconfig.ProviderConfig.get_domain',
        lambda x: where('testdomain.com'))
    def test_download_provider_info_new_provider(self):
        self._setup_provider_config_with("1", tempfile.mkdtemp())
        self._setup_providerbootstrapper(True)

        self.pb._download_provider_info()
        self.assertTrue(ProviderConfig.save.called)

    @mock.patch(
        'leap.bitmask.config.providerconfig.ProviderConfig.get_ca_cert_path',
        lambda x: where('cacert.pem'))
    def test_download_provider_info_not_modified(self):
        self._setup_provider_config_with("1", tempfile.mkdtemp())
        self._setup_providerbootstrapper(True)
        provider_path = self._produce_dummy_provider_json()

        # set mtime to something really new
        os.utime(provider_path, (-1, time.time()))

        self.pb._download_provider_info()
        # we check that it doesn't save the provider
        # config, because it's new enough
        self.assertFalse(ProviderConfig.save.called)

    @mock.patch(
        'leap.bitmask.config.providerconfig.ProviderConfig.get_domain',
        lambda x: where('testdomain.com'))
    def test_download_provider_info_not_modified_and_no_cacert(self):
        self._setup_provider_config_with("1", tempfile.mkdtemp())
        self._setup_providerbootstrapper(True)
        provider_path = self._produce_dummy_provider_json()

        # set mtime to something really new
        os.utime(provider_path, (-1, time.time()))

        self.pb._download_provider_info()
        # we check that it doesn't save the provider
        # config, because it's new enough
        self.assertFalse(ProviderConfig.save.called)

    @mock.patch(
        'leap.bitmask.config.providerconfig.ProviderConfig.get_ca_cert_path',
        lambda x: where('cacert.pem'))
    def test_download_provider_info_modified(self):
        self._setup_provider_config_with("1", tempfile.mkdtemp())
        self._setup_providerbootstrapper(True)
        provider_path = self._produce_dummy_provider_json()

        # set mtime to something really old
        os.utime(provider_path, (-1, 100))

        self.pb._download_provider_info()
        self.assertTrue(ProviderConfig.load.called)
        self.assertTrue(ProviderConfig.save.called)

    @mock.patch(
        'leap.bitmask.config.providerconfig.ProviderConfig.get_ca_cert_path',
        lambda x: where('cacert.pem'))
    def test_download_provider_info_unsupported_api_raises(self):
        self._setup_provider_config_with("9999999", tempfile.mkdtemp())
        self._setup_providerbootstrapper(False)
        self._produce_dummy_provider_json()

        with self.assertRaises(UnsupportedProviderAPI):
            self.pb._download_provider_info()

    @mock.patch(
        'leap.bitmask.config.providerconfig.ProviderConfig.get_ca_cert_path',
        lambda x: where('cacert.pem'))
    def test_download_provider_info_unsupported_api(self):
        self._setup_provider_config_with(SupportedAPIs.SUPPORTED_APIS[0],
                                         tempfile.mkdtemp())
        self._setup_providerbootstrapper(False)
        self._produce_dummy_provider_json()

        self.pb._download_provider_info()

    @mock.patch(
        'leap.bitmask.config.providerconfig.ProviderConfig.get_api_uri',
        lambda x: 'api.uri')
    @mock.patch(
        'leap.bitmask.config.providerconfig.ProviderConfig.get_ca_cert_path',
        lambda x: '/cert/path')
    def test_check_api_certificate_skips(self):
        self.pb._provider_config = ProviderConfig()
        self.pb._session.get = mock.MagicMock(return_value=Response())

        self.pb._should_proceed_cert = mock.MagicMock(return_value=False)
        self.pb._check_api_certificate()
        self.assertFalse(self.pb._session.get.called)

    @deferred()
    def test_check_api_certificate_fails(self):
        self.pb._provider_config = ProviderConfig()
        self.pb._provider_config.get_api_uri = mock.MagicMock(
            return_value="https://localhost:%s" % (self.https_port,))
        self.pb._provider_config.get_ca_cert_path = mock.MagicMock(
            return_value=os.path.join(
                os.path.split(__file__)[0],
                "wrongcert.pem"))
        self.pb._provider_config.get_api_version = mock.MagicMock(
            return_value="1")

        self.pb._should_proceed_cert = mock.MagicMock(return_value=True)

        def check(*args):
            with self.assertRaises(requests.exceptions.SSLError):
                self.pb._check_api_certificate()
        d = threads.deferToThread(check)
        return d

    @deferred()
    def test_check_api_certificate_succeeds(self):
        self.pb._provider_config = ProviderConfig()
        self.pb._provider_config.get_api_uri = mock.MagicMock(
            return_value="https://localhost:%s" % (self.https_port,))
        self.pb._provider_config.get_ca_cert_path = mock.MagicMock(
            return_value=where('cacert.pem'))
        self.pb._provider_config.get_api_version = mock.MagicMock(
            return_value="1")

        self.pb._should_proceed_cert = mock.MagicMock(return_value=True)

        def check(*args):
            self.pb._check_api_certificate()
        d = threads.deferToThread(check)
        return d