From 83ebaef86cb90aa01b1216a007b171b8a92ced52 Mon Sep 17 00:00:00 2001 From: Jean-Paul Calderone Date: Wed, 16 Dec 2020 15:24:33 -0500 Subject: [PATCH 01/10] Stop mocking safe_load The comment implies this will cause something to break on some platform. Let's find out. --- src/allmydata/test/test_introducer.py | 15 ++++----------- 1 file changed, 4 insertions(+), 11 deletions(-) diff --git a/src/allmydata/test/test_introducer.py b/src/allmydata/test/test_introducer.py index b14b66ffb..ac587e51b 100644 --- a/src/allmydata/test/test_introducer.py +++ b/src/allmydata/test/test_introducer.py @@ -15,7 +15,7 @@ from six import ensure_binary, ensure_text import os, re, itertools from base64 import b32decode import json -from mock import Mock, patch +from mock import Mock from testtools.matchers import ( Is, @@ -94,17 +94,10 @@ class Node(testutil.SignalMixin, testutil.ReallyEqualMixin, AsyncTestCase): f.write(u'---\n') os.chmod(yaml_fname, 0o000) self.addCleanup(lambda: os.chmod(yaml_fname, 0o700)) - # just mocking the yaml failure, as "yamlutil.safe_load" only - # returns None on some platforms for unreadable files - with patch("allmydata.client.yamlutil") as p: - p.safe_load = Mock(return_value=None) - - fake_tub = Mock() - config = read_config(basedir, "portnum") - - with self.assertRaises(EnvironmentError): - create_introducer_clients(config, fake_tub) + config = read_config(basedir, "portnum") + with self.assertRaises(EnvironmentError): + create_introducer_clients(config, Tub()) @defer.inlineCallbacks def test_furl(self): From 3513e9b4fce500a358f5c517708cc33837b4abf7 Mon Sep 17 00:00:00 2001 From: Jean-Paul Calderone Date: Wed, 16 Dec 2020 15:25:11 -0500 Subject: [PATCH 02/10] news fragment --- newsfragments/3534.minor | 0 1 file changed, 0 insertions(+), 0 deletions(-) create mode 100644 newsfragments/3534.minor diff --git a/newsfragments/3534.minor b/newsfragments/3534.minor new file mode 100644 index 000000000..e69de29bb From 60e401ca697abfd910dc841c900025524b90846f Mon Sep 17 00:00:00 2001 From: Jean-Paul Calderone Date: Wed, 16 Dec 2020 16:19:33 -0500 Subject: [PATCH 03/10] Make ObserverList synchronous, reentrant, and exception safe with tests --- src/allmydata/test/test_observer.py | 40 +++++++++++++++++++++++++++++ src/allmydata/util/observer.py | 15 ++++++++--- 2 files changed, 52 insertions(+), 3 deletions(-) diff --git a/src/allmydata/test/test_observer.py b/src/allmydata/test/test_observer.py index 0db13db58..73eece98e 100644 --- a/src/allmydata/test/test_observer.py +++ b/src/allmydata/test/test_observer.py @@ -101,3 +101,43 @@ class Observer(unittest.TestCase): d.addCallback(_step2) d.addCallback(_check2) return d + + def test_observer_list_reentrant(self): + """ + ``ObserverList`` is reentrant. + """ + observed = [] + + def observer_one(): + obs.unsubscribe(observer_one) + + def observer_two(): + observed.append(None) + + obs = observer.ObserverList() + obs.subscribe(observer_one) + obs.subscribe(observer_two) + obs.notify() + + self.assertEqual([None], observed) + + def test_observer_list_observer_errors(self): + """ + An error in an earlier observer does not prevent notification from being + delivered to a later observer. + """ + observed = [] + + def observer_one(): + raise Exception("Some problem here") + + def observer_two(): + observed.append(None) + + obs = observer.ObserverList() + obs.subscribe(observer_one) + obs.subscribe(observer_two) + obs.notify() + + self.assertEqual([None], observed) + self.assertEqual(1, len(self.flushLoggedErrors(Exception))) diff --git a/src/allmydata/util/observer.py b/src/allmydata/util/observer.py index 432aabb87..ad55e65a5 100644 --- a/src/allmydata/util/observer.py +++ b/src/allmydata/util/observer.py @@ -16,6 +16,9 @@ if PY2: import weakref from twisted.internet import defer from foolscap.api import eventually +from twisted.logger import ( + Logger, +) """The idiom we use is for the observed object to offer a method named 'when_something', which returns a deferred. That deferred will be fired when @@ -97,7 +100,10 @@ class LazyOneShotObserverList(OneShotObserverList): self._fire(self._get_result()) class ObserverList(object): - """A simple class to distribute events to a number of subscribers.""" + """ + Immediately distribute events to a number of subscribers. + """ + _logger = Logger() def __init__(self): self._watchers = [] @@ -109,8 +115,11 @@ class ObserverList(object): self._watchers.remove(observer) def notify(self, *args, **kwargs): - for o in self._watchers: - eventually(o, *args, **kwargs) + for o in self._watchers[:]: + try: + o(*args, **kwargs) + except: + self._logger.failure("While notifying {o!r}", o=o) class EventStreamObserver(object): """A simple class to distribute multiple events to a single subscriber. From b2c9296f6befc5e0046c165bb5b0d1b628cadf69 Mon Sep 17 00:00:00 2001 From: Jean-Paul Calderone Date: Wed, 16 Dec 2020 16:20:00 -0500 Subject: [PATCH 04/10] Use ObserverList instead of an ad hoc reimplementation --- src/allmydata/introducer/client.py | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/src/allmydata/introducer/client.py b/src/allmydata/introducer/client.py index fa1e1efe8..60a06ce2d 100644 --- a/src/allmydata/introducer/client.py +++ b/src/allmydata/introducer/client.py @@ -24,6 +24,9 @@ from allmydata.introducer.common import sign_to_foolscap, unsign_from_foolscap,\ get_tubid_string_from_ann from allmydata.util import log, yamlutil, connection_status from allmydata.util.rrefutil import add_version_to_remote_reference +from allmydata.util.observer import ( + ObserverList, +) from allmydata.crypto.error import BadSignature from allmydata.util.assertutil import precondition @@ -64,8 +67,7 @@ class IntroducerClient(service.Service, Referenceable): self._publisher = None self._since = None - self._local_subscribers = [] # (servicename,cb,args,kwargs) tuples - self._subscribed_service_names = set() + self._local_subscribers = {} # {servicename: ObserverList} self._subscriptions = set() # requests we've actually sent # _inbound_announcements remembers one announcement per @@ -179,21 +181,21 @@ class IntroducerClient(service.Service, Referenceable): return log.msg(*args, **kwargs) def subscribe_to(self, service_name, cb, *args, **kwargs): - self._local_subscribers.append( (service_name,cb,args,kwargs) ) - self._subscribed_service_names.add(service_name) + obs = self._local_subscribers.setdefault(service_name, ObserverList()) + obs.subscribe(lambda key_s, ann: cb(key_s, ann, *args, **kwargs)) self._maybe_subscribe() for index,(ann,key_s,when) in list(self._inbound_announcements.items()): precondition(isinstance(key_s, bytes), key_s) servicename = index[0] if servicename == service_name: - eventually(cb, key_s, ann, *args, **kwargs) + obs.notify(key_s, ann) def _maybe_subscribe(self): if not self._publisher: self.log("want to subscribe, but no introducer yet", level=log.NOISY) return - for service_name in self._subscribed_service_names: + for service_name in self._local_subscribers: if service_name in self._subscriptions: continue self._subscriptions.add(service_name) @@ -272,7 +274,7 @@ class IntroducerClient(service.Service, Referenceable): precondition(isinstance(key_s, bytes), key_s) self._debug_counts["inbound_announcement"] += 1 service_name = str(ann["service-name"]) - if service_name not in self._subscribed_service_names: + if service_name not in self._local_subscribers: self.log("announcement for a service we don't care about [%s]" % (service_name,), level=log.UNUSUAL, umid="dIpGNA") self._debug_counts["wrong_service"] += 1 @@ -343,9 +345,10 @@ class IntroducerClient(service.Service, Referenceable): def _deliver_announcements(self, key_s, ann): precondition(isinstance(key_s, bytes), key_s) service_name = str(ann["service-name"]) - for (service_name2,cb,args,kwargs) in self._local_subscribers: - if service_name2 == service_name: - eventually(cb, key_s, ann, *args, **kwargs) + + obs = self._local_subscribers.get(service_name) + if obs is not None: + obs.notify(key_s, ann) def connection_status(self): assert self.running # startService builds _introducer_reconnector From 98000c2b66a7a4a5b68e0abddb41abfc2ea9544d Mon Sep 17 00:00:00 2001 From: Jean-Paul Calderone Date: Wed, 16 Dec 2020 16:20:38 -0500 Subject: [PATCH 05/10] re-implement test_unsigned_announcement without mock and to make assertions about public behavior instead of private implementation details --- src/allmydata/test/test_introducer.py | 57 +++++++++++++++++++++------ 1 file changed, 45 insertions(+), 12 deletions(-) diff --git a/src/allmydata/test/test_introducer.py b/src/allmydata/test/test_introducer.py index ac587e51b..668d577db 100644 --- a/src/allmydata/test/test_introducer.py +++ b/src/allmydata/test/test_introducer.py @@ -15,7 +15,12 @@ from six import ensure_binary, ensure_text import os, re, itertools from base64 import b32decode import json -from mock import Mock +from operator import ( + setitem, +) +from functools import ( + partial, +) from testtools.matchers import ( Is, @@ -84,7 +89,8 @@ class Node(testutil.SignalMixin, testutil.ReallyEqualMixin, AsyncTestCase): def test_introducer_clients_unloadable(self): """ - Error if introducers.yaml exists but we can't read it + ``create_introducer_clients`` raises ``EnvironmentError`` if + ``introducers.yaml`` exists but we can't read it. """ basedir = u"introducer.IntroducerNode.test_introducer_clients_unloadable" os.mkdir(basedir) @@ -1030,23 +1036,50 @@ class Signatures(SyncTestCase): unsign_from_foolscap, (bad_msg, sig, b"v999-key")) def test_unsigned_announcement(self): - ed25519.verifying_key_from_string(b"pub-v0-wodst6ly4f7i7akt2nxizsmmy2rlmer6apltl56zctn67wfyu5tq") - mock_tub = Mock() + """ + An incorrectly signed announcement is not delivered to subscribers. + """ + private_key, public_key = ed25519.create_signing_keypair() + public_key_str = ed25519.string_from_verifying_key(public_key) + ic = IntroducerClient( - mock_tub, + Tub(), "pb://", u"fake_nick", "0.0.0", "1.2.3", (0, u"i am a nonce"), - "invalid", + FilePath(self.mktemp()), + ) + received = {} + ic.subscribe_to("good-stuff", partial(setitem, received)) + + # Deliver a good message to prove our test code is valid. + ann = {"service-name": "good-stuff", "payload": "hello"} + ann_t = sign_to_foolscap(ann, private_key) + ic.got_announcements([ann_t]) + + self.assertEqual( + {public_key_str[len("pub-"):]: ann}, + received, + ) + received.clear() + + # Now deliver one without a valid signature and observe that it isn't + # delivered to the subscriber. + ann = {"service-name": "good-stuff", "payload": "bad stuff"} + (msg, sig, key) = sign_to_foolscap(ann, private_key) + # Invalidate the signature a little + sig = sig.replace(b"2", b"3") + ann_t = (msg, sig, key) + ic.got_announcements([ann_t]) + + # The received announcements dict should remain empty because we + # should not receive the announcement with the invalid signature. + self.assertEqual( + {}, + received, ) - self.assertEqual(0, ic._debug_counts["inbound_announcement"]) - ic.got_announcements([ - (b"message", b"v0-aaaaaaa", b"v0-wodst6ly4f7i7akt2nxizsmmy2rlmer6apltl56zctn67wfyu5tq") - ]) - # we should have rejected this announcement due to a bad signature - self.assertEqual(0, ic._debug_counts["inbound_announcement"]) # add tests of StorageFarmBroker: if it receives duplicate announcements, it From b200075246ee8d69e19d852fd44cdaa87b117908 Mon Sep 17 00:00:00 2001 From: Jean-Paul Calderone Date: Wed, 16 Dec 2020 16:23:05 -0500 Subject: [PATCH 06/10] whitespace --- src/allmydata/introducer/client.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/allmydata/introducer/client.py b/src/allmydata/introducer/client.py index 60a06ce2d..af84d12c3 100644 --- a/src/allmydata/introducer/client.py +++ b/src/allmydata/introducer/client.py @@ -345,7 +345,6 @@ class IntroducerClient(service.Service, Referenceable): def _deliver_announcements(self, key_s, ann): precondition(isinstance(key_s, bytes), key_s) service_name = str(ann["service-name"]) - obs = self._local_subscribers.get(service_name) if obs is not None: obs.notify(key_s, ann) From 4117beba6a3eb2f142e089c55fc04a840d1d1e24 Mon Sep 17 00:00:00 2001 From: Jean-Paul Calderone Date: Wed, 16 Dec 2020 16:25:51 -0500 Subject: [PATCH 07/10] remove unused import yaaay --- src/allmydata/introducer/client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/allmydata/introducer/client.py b/src/allmydata/introducer/client.py index af84d12c3..de24f9cf3 100644 --- a/src/allmydata/introducer/client.py +++ b/src/allmydata/introducer/client.py @@ -16,7 +16,7 @@ from six import ensure_text import time from zope.interface import implementer from twisted.application import service -from foolscap.api import Referenceable, eventually +from foolscap.api import Referenceable from allmydata.interfaces import InsufficientVersionError from allmydata.introducer.interfaces import IIntroducerClient, \ RIIntroducerSubscriberClient_v2 From a223f6bb60a9e9c3e7a25613c479c36d2d2ca74f Mon Sep 17 00:00:00 2001 From: Jean-Paul Calderone Date: Wed, 16 Dec 2020 17:31:06 -0500 Subject: [PATCH 08/10] More reliably corrupt the signature --- src/allmydata/test/test_introducer.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/allmydata/test/test_introducer.py b/src/allmydata/test/test_introducer.py index 668d577db..1ba928257 100644 --- a/src/allmydata/test/test_introducer.py +++ b/src/allmydata/test/test_introducer.py @@ -1069,8 +1069,11 @@ class Signatures(SyncTestCase): # delivered to the subscriber. ann = {"service-name": "good-stuff", "payload": "bad stuff"} (msg, sig, key) = sign_to_foolscap(ann, private_key) - # Invalidate the signature a little - sig = sig.replace(b"2", b"3") + # Drop a base32 word from the middle of the key to invalidate the + # signature. + sig_l = list(sig) + sig_l[20:22] = [] + sig = b"".join(sig_l) ann_t = (msg, sig, key) ic.got_announcements([ann_t]) From 895ba55cf7759ed7a76eb77b435bbd038fe6a759 Mon Sep 17 00:00:00 2001 From: Jean-Paul Calderone Date: Wed, 16 Dec 2020 18:17:14 -0500 Subject: [PATCH 09/10] Python 3 compatibility --- src/allmydata/test/test_introducer.py | 6 +++--- src/allmydata/util/base32.py | 4 +++- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/src/allmydata/test/test_introducer.py b/src/allmydata/test/test_introducer.py index 1ba928257..0475d3f6c 100644 --- a/src/allmydata/test/test_introducer.py +++ b/src/allmydata/test/test_introducer.py @@ -1071,9 +1071,9 @@ class Signatures(SyncTestCase): (msg, sig, key) = sign_to_foolscap(ann, private_key) # Drop a base32 word from the middle of the key to invalidate the # signature. - sig_l = list(sig) - sig_l[20:22] = [] - sig = b"".join(sig_l) + sig_a = bytearray(sig) + sig_a[20:22] = [] + sig = bytes(sig_a) ann_t = (msg, sig, key) ic.got_announcements([ann_t]) diff --git a/src/allmydata/util/base32.py b/src/allmydata/util/base32.py index 287d214ea..41c0f0413 100644 --- a/src/allmydata/util/base32.py +++ b/src/allmydata/util/base32.py @@ -140,7 +140,9 @@ def a2b(cs): # Add padding back, to make Python's base64 module happy: while (len(cs) * 5) % 8 != 0: cs += b"=" - return base64.b32decode(cs) + # Let newbytes come through and still work on Python 2, where the base64 + # module gets confused by them. + return base64.b32decode(backwardscompat_bytes(cs)) __all__ = ["b2a", "a2b", "b2a_or_none", "BASE32CHAR_3bits", "BASE32CHAR_1bits", "BASE32CHAR", "BASE32STR_anybytes", "could_be_base32_encoded"] From 0ffbc7870ee4f0f319f594acd8cb47aa289d51e7 Mon Sep 17 00:00:00 2001 From: Jean-Paul Calderone Date: Wed, 16 Dec 2020 20:32:04 -0500 Subject: [PATCH 10/10] Okay, let KeyboardInterrupt through --- src/allmydata/test/test_observer.py | 13 +++++++++++++ src/allmydata/util/observer.py | 2 +- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/src/allmydata/test/test_observer.py b/src/allmydata/test/test_observer.py index 73eece98e..134876be3 100644 --- a/src/allmydata/test/test_observer.py +++ b/src/allmydata/test/test_observer.py @@ -141,3 +141,16 @@ class Observer(unittest.TestCase): self.assertEqual([None], observed) self.assertEqual(1, len(self.flushLoggedErrors(Exception))) + + def test_observer_list_propagate_keyboardinterrupt(self): + """ + ``KeyboardInterrupt`` escapes ``ObserverList.notify``. + """ + def observer_one(): + raise KeyboardInterrupt() + + obs = observer.ObserverList() + obs.subscribe(observer_one) + + with self.assertRaises(KeyboardInterrupt): + obs.notify() diff --git a/src/allmydata/util/observer.py b/src/allmydata/util/observer.py index ad55e65a5..4a39fe014 100644 --- a/src/allmydata/util/observer.py +++ b/src/allmydata/util/observer.py @@ -118,7 +118,7 @@ class ObserverList(object): for o in self._watchers[:]: try: o(*args, **kwargs) - except: + except Exception: self._logger.failure("While notifying {o!r}", o=o) class EventStreamObserver(object):