Remove pushers when deleting 3pid from account (#10581)

When a user deletes an email from their account it will
now also remove all pushers for that email and that user
(even if these pushers were created by a different client)

7 files changed, 143 insertions, 1 deletions

diff --git a/CHANGES.md b/CHANGES.md
index f8da8771aa..24f3d53a6d 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,3 +1,5 @@
+Users will stop receiving message updates via email for addresses that were previously linked to their account
+
 Synapse 1.41.0 (2021-08-24)
 ===========================
 
diff --git a/changelog.d/10581.bugfix b/changelog.d/10581.bugfix
new file mode 100644
index 0000000000..15c7da4497
--- /dev/null
+++ b/changelog.d/10581.bugfix
@@ -0,0 +1 @@
+Remove pushers when deleting a 3pid from an account. Pushers for old unlinked emails will also be deleted.
\ No newline at end of file
diff --git a/docs/upgrade.md b/docs/upgrade.md
index 6d4b8cb48e..dcf0a7db5b 100644
--- a/docs/upgrade.md
+++ b/docs/upgrade.md
@@ -107,6 +107,11 @@ This may affect you if you make use of custom HTML templates for the
 The template is now provided an `error` variable if the authentication
 process failed. See the default templates linked above for an example.
 
+# Upgrading to v1.42.0
+
+## Removal of out-of-date email pushers
+Users will stop receiving message updates via email for addresses that were
+once, but not still, linked to their account.
 
 # Upgrading to v1.41.0
 
diff --git a/synapse/handlers/auth.py b/synapse/handlers/auth.py
index 98d3d2d97f..34725324a6 100644
--- a/synapse/handlers/auth.py
+++ b/synapse/handlers/auth.py
@@ -1464,6 +1464,10 @@ class AuthHandler(BaseHandler):
         )
 
         await self.store.user_delete_threepid(user_id, medium, address)
+        if medium == "email":
+            await self.store.delete_pusher_by_app_id_pushkey_user_id(
+                app_id="m.email", pushkey=address, user_id=user_id
+            )
         return result
 
     async def hash(self, password: str) -> str:
@@ -1732,7 +1736,6 @@ class AuthHandler(BaseHandler):
 
 @attr.s(slots=True)
 class MacaroonGenerator:
-
     hs = attr.ib()
 
     def generate_guest_access_token(self, user_id: str) -> str:
diff --git a/synapse/storage/databases/main/pusher.py b/synapse/storage/databases/main/pusher.py
index b48fe086d4..e47caa2125 100644
--- a/synapse/storage/databases/main/pusher.py
+++ b/synapse/storage/databases/main/pusher.py
@@ -48,6 +48,11 @@ class PusherWorkerStore(SQLBaseStore):
             self._remove_stale_pushers,
         )
 
+        self.db_pool.updates.register_background_update_handler(
+            "remove_deleted_email_pushers",
+            self._remove_deleted_email_pushers,
+        )
+
     def _decode_pushers_rows(self, rows: Iterable[dict]) -> Iterator[PusherConfig]:
         """JSON-decode the data in the rows returned from the `pushers` table
 
@@ -388,6 +393,73 @@ class PusherWorkerStore(SQLBaseStore):
 
         return number_deleted
 
+    async def _remove_deleted_email_pushers(
+        self, progress: dict, batch_size: int
+    ) -> int:
+        """A background update that deletes all pushers for deleted email addresses.
+
+        In previous versions of synapse, when users deleted their email address, it didn't
+        also delete all the pushers for that email address. This background update removes
+        those to prevent unwanted emails. This should only need to be run once (when users
+        upgrade to v1.42.0
+
+        Args:
+            progress: dict used to store progress of this background update
+            batch_size: the maximum number of rows to retrieve in a single select query
+
+        Returns:
+            The number of deleted rows
+        """
+
+        last_pusher = progress.get("last_pusher", 0)
+
+        def _delete_pushers(txn) -> int:
+
+            sql = """
+                SELECT p.id, p.user_name, p.app_id, p.pushkey
+                FROM pushers AS p
+                    LEFT JOIN user_threepids AS t
+                        ON t.user_id = p.user_name
+                        AND t.medium = 'email'
+                        AND t.address = p.pushkey
+                WHERE t.user_id is NULL
+                    AND p.app_id = 'm.email'
+                    AND p.id > ?
+                ORDER BY p.id ASC
+                LIMIT ?
+            """
+
+            txn.execute(sql, (last_pusher, batch_size))
+
+            last = None
+            num_deleted = 0
+            for row in txn:
+                last = row[0]
+                num_deleted += 1
+                self.db_pool.simple_delete_txn(
+                    txn,
+                    "pushers",
+                    {"user_name": row[1], "app_id": row[2], "pushkey": row[3]},
+                )
+
+            if last is not None:
+                self.db_pool.updates._background_update_progress_txn(
+                    txn, "remove_deleted_email_pushers", {"last_pusher": last}
+                )
+
+            return num_deleted
+
+        number_deleted = await self.db_pool.runInteraction(
+            "_remove_deleted_email_pushers", _delete_pushers
+        )
+
+        if number_deleted < batch_size:
+            await self.db_pool.updates._end_background_update(
+                "remove_deleted_email_pushers"
+            )
+
+        return number_deleted
+
 
 class PusherStore(PusherWorkerStore):
     def get_pushers_stream_token(self) -> int:
diff --git a/synapse/storage/schema/main/delta/63/02delete_unlinked_email_pushers.sql b/synapse/storage/schema/main/delta/63/02delete_unlinked_email_pushers.sql
new file mode 100644
index 0000000000..611c4b95cf
--- /dev/null
+++ b/synapse/storage/schema/main/delta/63/02delete_unlinked_email_pushers.sql
@@ -0,0 +1,20 @@
+/* Copyright 2021 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+-- We may not have deleted all pushers for emails that are no longer linked
+-- to an account, so we set up a background job to delete them.
+INSERT INTO background_updates (ordering, update_name, progress_json) VALUES
+  (6302, 'remove_deleted_email_pushers', '{}');
diff --git a/tests/push/test_email.py b/tests/push/test_email.py
index e0a3342088..eea07485a0 100644
--- a/tests/push/test_email.py
+++ b/tests/push/test_email.py
@@ -125,6 +125,8 @@ class EmailPusherTests(HomeserverTestCase):
             )
         )
 
+        self.auth_handler = hs.get_auth_handler()
+
     def test_need_validated_email(self):
         """Test that we can only add an email pusher if the user has validated
         their email.
@@ -305,6 +307,43 @@ class EmailPusherTests(HomeserverTestCase):
         # We should get emailed about that message
         self._check_for_mail()
 
+    def test_no_email_sent_after_removed(self):
+        # Create a simple room with two users
+        room = self.helper.create_room_as(self.user_id, tok=self.access_token)
+        self.helper.invite(
+            room=room,
+            src=self.user_id,
+            tok=self.access_token,
+            targ=self.others[0].id,
+        )
+        self.helper.join(
+            room=room,
+            user=self.others[0].id,
+            tok=self.others[0].token,
+        )
+
+        # The other user sends a single message.
+        self.helper.send(room, body="Hi!", tok=self.others[0].token)
+
+        # We should get emailed about that message
+        self._check_for_mail()
+
+        # disassociate the user's email address
+        self.get_success(
+            self.auth_handler.delete_threepid(
+                user_id=self.user_id,
+                medium="email",
+                address="a@example.com",
+            )
+        )
+
+        # check that the pusher for that email address has been deleted
+        pushers = self.get_success(
+            self.hs.get_datastore().get_pushers_by({"user_name": self.user_id})
+        )
+        pushers = list(pushers)
+        self.assertEqual(len(pushers), 0)
+
     def _check_for_mail(self):
         """Check that the user receives an email notification"""