Don't start or announce anonymous access if config says not to

src/allmydata/client.py

@@ -861,33 +861,32 @@ class _Client(node.Node, pollmixin.PollMixin):
                              "is not listening ('tub.port=' is empty)")
 
         ss = self.get_anonymous_storage_server()
-        furl_file = self.config.get_private_path("storage.furl").encode(get_filesystem_encoding())
-        furl = self.tub.registerReference(ss, furlFile=furl_file)
-
-        anonymous_announcement = {
-            "anonymous-storage-FURL": furl,
+        announcement = {
             "permutation-seed-base32": self._init_permutation_seed(ss),
         }
 
+        if anonymous_storage_enabled(self.config):
+            furl_file = self.config.get_private_path("storage.furl").encode(get_filesystem_encoding())
+            furl = self.tub.registerReference(ss, furlFile=furl_file)
+            announcement["anonymous-storage-FURL"] = furl
+
         enabled_storage_servers = self._enable_storage_servers(
             announceable_storage_servers,
         )
-        plugins_announcement = {}
         storage_options = list(
             storage_server.announcement
             for storage_server
             in enabled_storage_servers
         )
+        plugins_announcement = {}
         if storage_options:
             # Only add the new key if there are any plugins enabled.
             plugins_announcement[u"storage-options"] = storage_options
 
-        total_announcement = {}
-        total_announcement.update(anonymous_announcement)
-        total_announcement.update(plugins_announcement)
+        announcement.update(plugins_announcement)
 
         for ic in self.introducer_clients:
-            ic.publish("storage", total_announcement, self._node_private_key)
+            ic.publish("storage", announcement, self._node_private_key)
 
 
     def _enable_storage_servers(self, announceable_storage_servers):

src/allmydata/test/test_client.py

@@ -28,6 +28,8 @@ from testtools.matchers import (
     MatchesListwise,
     MatchesDict,
     Always,
+    Is,
+    raises,
 )
 from testtools.twistedsupport import (
     succeeded,
@@ -757,6 +759,129 @@ def flush_but_dont_ignore(res):
     return d
 
 
+class AnonymousStorage(SyncTestCase):
+    """
+    Tests for behaviors of the client object with respect to the anonymous
+    storage service.
+    """
+    @defer.inlineCallbacks
+    def test_anonymous_storage_enabled(self):
+        """
+        If anonymous storage access is enabled then the client announces it.
+        """
+        basedir = self.id()
+        os.makedirs(basedir + b"/private")
+        config = client.config_from_string(
+            basedir,
+            b"tub.port",
+            BASECONFIG_I % (SOME_FURL,) + (
+                b"[storage]\n"
+                b"enabled = true\n"
+                b"anonymous = true\n"
+            )
+        )
+        node = yield client.create_client_from_config(
+            config,
+            _introducer_factory=MemoryIntroducerClient,
+        )
+        self.assertThat(
+            get_published_announcements(node),
+            MatchesListwise([
+                matches_storage_announcement(
+                    basedir,
+                    anonymous=True,
+                ),
+            ]),
+        )
+
+    @defer.inlineCallbacks
+    def test_anonymous_storage_disabled(self):
+        """
+        If anonymous storage access is disabled then the client does not announce
+        it nor does it write a fURL for it to beneath the node directory.
+        """
+        basedir = self.id()
+        os.makedirs(basedir + b"/private")
+        config = client.config_from_string(
+            basedir,
+            b"tub.port",
+            BASECONFIG_I % (SOME_FURL,) + (
+                b"[storage]\n"
+                b"enabled = true\n"
+                b"anonymous = false\n"
+            )
+        )
+        node = yield client.create_client_from_config(
+            config,
+            _introducer_factory=MemoryIntroducerClient,
+        )
+        self.expectThat(
+            get_published_announcements(node),
+            MatchesListwise([
+                matches_storage_announcement(
+                    basedir,
+                    anonymous=False,
+                ),
+            ]),
+        )
+        self.expectThat(
+            config.get_private_config(b"storage.furl", default=None),
+            Is(None),
+        )
+
+    @defer.inlineCallbacks
+    def test_anonymous_storage_enabled_then_disabled(self):
+        """
+        If a node is run with anonymous storage enabled and then later anonymous
+        storage is disabled in the configuration for that node, it is not
+        possible to reach the anonymous storage server via the originally
+        published fURL.
+        """
+        basedir = self.id()
+        os.makedirs(basedir + b"/private")
+        enabled_config = client.config_from_string(
+            basedir,
+            b"tub.port",
+            BASECONFIG_I % (SOME_FURL,) + (
+                b"[storage]\n"
+                b"enabled = true\n"
+                b"anonymous = true\n"
+            )
+        )
+        node = yield client.create_client_from_config(
+            enabled_config,
+            _introducer_factory=MemoryIntroducerClient,
+        )
+        anonymous_storage_furl = enabled_config.get_private_config(b"storage.furl")
+        def check_furl():
+            return node.tub.getReferenceForURL(anonymous_storage_furl)
+        # Perform a sanity check that our test code makes sense: is this a
+        # legit way to verify whether a fURL will refer to an object?
+        self.assertThat(
+            check_furl(),
+            # If it doesn't raise a KeyError we're in business.
+            Always(),
+        )
+
+        disabled_config = client.config_from_string(
+            basedir,
+            b"tub.port",
+            BASECONFIG_I % (SOME_FURL,) + (
+                b"[storage]\n"
+                b"enabled = true\n"
+                b"anonymous = false\n"
+            )
+        )
+        node = yield client.create_client_from_config(
+            disabled_config,
+            _introducer_factory=MemoryIntroducerClient,
+        )
+        self.assertThat(
+            check_furl,
+            raises(KeyError),
+        )
+
+
 class IntroducerClients(unittest.TestCase):
 
     def test_invalid_introducer_furl(self):