diff --git a/changelog.d/11009.bugfix b/changelog.d/11009.bugfix
new file mode 100644
index 0000000000..13b8e5983b
--- /dev/null
+++ b/changelog.d/11009.bugfix
@@ -0,0 +1 @@
+Fix a long-standing bug which meant that events received over federation were sometimes incorrectly accepted into the room state.
diff --git a/synapse/event_auth.py b/synapse/event_auth.py
index ca0293a3dc..e885961698 100644
--- a/synapse/event_auth.py
+++ b/synapse/event_auth.py
@@ -14,7 +14,7 @@
 # limitations under the License.
 
 import logging
-from typing import Any, Dict, List, Optional, Set, Tuple, Union
+from typing import Any, Dict, Iterable, List, Optional, Set, Tuple, Union
 
 from canonicaljson import encode_canonical_json
 from signedjson.key import decode_verify_key_bytes
@@ -113,7 +113,7 @@ def validate_event_for_room_version(
 
 
 def check_auth_rules_for_event(
-    room_version_obj: RoomVersion, event: EventBase, auth_events: StateMap[EventBase]
+    room_version_obj: RoomVersion, event: EventBase, auth_events: Iterable[EventBase]
 ) -> None:
     """Check that an event complies with the auth rules
 
@@ -137,8 +137,6 @@ def check_auth_rules_for_event(
     Raises:
         AuthError if the checks fail
     """
-    assert isinstance(auth_events, dict)
-
     # We need to ensure that the auth events are actually for the same room, to
     # stop people from using powers they've been granted in other rooms for
     # example.
@@ -147,7 +145,7 @@ def check_auth_rules_for_event(
     # the state res algorithm isn't silly enough to give us events from different rooms.
     # Still, it's easier to do it anyway.
     room_id = event.room_id
-    for auth_event in auth_events.values():
+    for auth_event in auth_events:
         if auth_event.room_id != room_id:
             raise AuthError(
                 403,
@@ -186,8 +184,10 @@ def check_auth_rules_for_event(
         logger.debug("Allowing! %s", event)
         return
 
+    auth_dict = {(e.type, e.state_key): e for e in auth_events}
+
     # 3. If event does not have a m.room.create in its auth_events, reject.
-    creation_event = auth_events.get((EventTypes.Create, ""), None)
+    creation_event = auth_dict.get((EventTypes.Create, ""), None)
     if not creation_event:
         raise AuthError(403, "No create event in auth events")
 
@@ -195,7 +195,7 @@ def check_auth_rules_for_event(
     creating_domain = get_domain_from_id(event.room_id)
     originating_domain = get_domain_from_id(event.sender)
     if creating_domain != originating_domain:
-        if not _can_federate(event, auth_events):
+        if not _can_federate(event, auth_dict):
             raise AuthError(403, "This room has been marked as unfederatable.")
 
     # 4. If type is m.room.aliases
@@ -217,23 +217,20 @@ def check_auth_rules_for_event(
         logger.debug("Allowing! %s", event)
         return
 
-    if logger.isEnabledFor(logging.DEBUG):
-        logger.debug("Auth events: %s", [a.event_id for a in auth_events.values()])
-
     # 5. If type is m.room.membership
     if event.type == EventTypes.Member:
-        _is_membership_change_allowed(room_version_obj, event, auth_events)
+        _is_membership_change_allowed(room_version_obj, event, auth_dict)
         logger.debug("Allowing! %s", event)
         return
 
-    _check_event_sender_in_room(event, auth_events)
+    _check_event_sender_in_room(event, auth_dict)
 
     # Special case to allow m.room.third_party_invite events wherever
     # a user is allowed to issue invites.  Fixes
     # https://github.com/vector-im/vector-web/issues/1208 hopefully
     if event.type == EventTypes.ThirdPartyInvite:
-        user_level = get_user_power_level(event.user_id, auth_events)
-        invite_level = get_named_level(auth_events, "invite", 0)
+        user_level = get_user_power_level(event.user_id, auth_dict)
+        invite_level = get_named_level(auth_dict, "invite", 0)
 
         if user_level < invite_level:
             raise AuthError(403, "You don't have permission to invite users")
@@ -241,20 +238,20 @@ def check_auth_rules_for_event(
             logger.debug("Allowing! %s", event)
             return
 
-    _can_send_event(event, auth_events)
+    _can_send_event(event, auth_dict)
 
     if event.type == EventTypes.PowerLevels:
-        _check_power_levels(room_version_obj, event, auth_events)
+        _check_power_levels(room_version_obj, event, auth_dict)
 
     if event.type == EventTypes.Redaction:
-        check_redaction(room_version_obj, event, auth_events)
+        check_redaction(room_version_obj, event, auth_dict)
 
     if (
         event.type == EventTypes.MSC2716_INSERTION
         or event.type == EventTypes.MSC2716_BATCH
         or event.type == EventTypes.MSC2716_MARKER
     ):
-        check_historical(room_version_obj, event, auth_events)
+        check_historical(room_version_obj, event, auth_dict)
 
     logger.debug("Allowing! %s", event)
 
diff --git a/synapse/handlers/event_auth.py b/synapse/handlers/event_auth.py
index d089c56286..365063ebdf 100644
--- a/synapse/handlers/event_auth.py
+++ b/synapse/handlers/event_auth.py
@@ -55,8 +55,7 @@ class EventAuthHandler:
         """Check an event passes the auth rules at its own auth events"""
         auth_event_ids = event.auth_event_ids()
         auth_events_by_id = await self._store.get_events(auth_event_ids)
-        auth_events = {(e.type, e.state_key): e for e in auth_events_by_id.values()}
-        check_auth_rules_for_event(room_version_obj, event, auth_events)
+        check_auth_rules_for_event(room_version_obj, event, auth_events_by_id.values())
 
     def compute_auth_events(
         self,
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index e072efad16..69f1ef3afa 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -1167,13 +1167,11 @@ class FederationHandler:
                 logger.info("Failed to find auth event %r", e_id)
 
         for e in itertools.chain(auth_events, state, [event]):
-            auth_for_e = {
-                (event_map[e_id].type, event_map[e_id].state_key): event_map[e_id]
-                for e_id in e.auth_event_ids()
-                if e_id in event_map
-            }
+            auth_for_e = [
+                event_map[e_id] for e_id in e.auth_event_ids() if e_id in event_map
+            ]
             if create_event:
-                auth_for_e[(EventTypes.Create, "")] = create_event
+                auth_for_e.append(create_event)
 
             try:
                 validate_event_for_room_version(room_version, e)
diff --git a/synapse/handlers/federation_event.py b/synapse/handlers/federation_event.py
index b8ce0006bb..1705432d7c 100644
--- a/synapse/handlers/federation_event.py
+++ b/synapse/handlers/federation_event.py
@@ -1203,7 +1203,7 @@ class FederationEventHandler:
 
         def prep(event: EventBase) -> Optional[Tuple[EventBase, EventContext]]:
             with nested_logging_context(suffix=event.event_id):
-                auth = {}
+                auth = []
                 for auth_event_id in event.auth_event_ids():
                     ae = persisted_events.get(auth_event_id)
                     if not ae:
@@ -1216,7 +1216,7 @@ class FederationEventHandler:
                         # exist, which means it is premature to reject `event`. Instead we
                         # just ignore it for now.
                         return None
-                    auth[(ae.type, ae.state_key)] = ae
+                    auth.append(ae)
 
                 context = EventContext.for_outlier()
                 try:
@@ -1305,7 +1305,9 @@ class FederationEventHandler:
             auth_events_for_auth = calculated_auth_event_map
 
         try:
-            check_auth_rules_for_event(room_version_obj, event, auth_events_for_auth)
+            check_auth_rules_for_event(
+                room_version_obj, event, auth_events_for_auth.values()
+            )
         except AuthError as e:
             logger.warning("Failed auth resolution for %r because %s", event, e)
             context.rejected = RejectedReason.AUTH_ERROR
@@ -1403,11 +1405,9 @@ class FederationEventHandler:
         current_state_ids_list = [
             e for k, e in current_state_ids.items() if k in auth_types
         ]
-
-        auth_events_map = await self._store.get_events(current_state_ids_list)
-        current_auth_events = {
-            (e.type, e.state_key): e for e in auth_events_map.values()
-        }
+        current_auth_events = await self._store.get_events_as_list(
+            current_state_ids_list
+        )
 
         try:
             check_auth_rules_for_event(room_version_obj, event, current_auth_events)
diff --git a/synapse/state/v1.py b/synapse/state/v1.py
index ffe6207a3c..6edadea550 100644
--- a/synapse/state/v1.py
+++ b/synapse/state/v1.py
@@ -332,7 +332,7 @@ def _resolve_auth_events(
             event_auth.check_auth_rules_for_event(
                 RoomVersions.V1,
                 event,
-                auth_events,
+                auth_events.values(),
             )
             prev_event = event
         except AuthError:
@@ -350,7 +350,7 @@ def _resolve_normal_events(
             event_auth.check_auth_rules_for_event(
                 RoomVersions.V1,
                 event,
-                auth_events,
+                auth_events.values(),
             )
             return event
         except AuthError:
diff --git a/synapse/state/v2.py b/synapse/state/v2.py
index bd18eefd58..c618df2fde 100644
--- a/synapse/state/v2.py
+++ b/synapse/state/v2.py
@@ -549,7 +549,7 @@ async def _iterative_auth_checks(
             event_auth.check_auth_rules_for_event(
                 room_version,
                 event,
-                auth_events,
+                auth_events.values(),
             )
 
             resolved_state[(event.type, event.state_key)] = event_id
diff --git a/tests/test_event_auth.py b/tests/test_event_auth.py
index cf407c51cf..e2c506e5a4 100644
--- a/tests/test_event_auth.py
+++ b/tests/test_event_auth.py
@@ -24,6 +24,47 @@ from synapse.types import JsonDict, get_domain_from_id
 
 
 class EventAuthTestCase(unittest.TestCase):
+    def test_rejected_auth_events(self):
+        """
+        Events that refer to rejected events in their auth events are rejected
+        """
+        creator = "@creator:example.com"
+        auth_events = [
+            _create_event(creator),
+            _join_event(creator),
+        ]
+
+        # creator should be able to send state
+        event_auth.check_auth_rules_for_event(
+            RoomVersions.V9,
+            _random_state_event(creator),
+            auth_events,
+        )
+
+        # ... but a rejected join_rules event should cause it to be rejected
+        rejected_join_rules = _join_rules_event(creator, "public")
+        rejected_join_rules.rejected_reason = "stinky"
+        auth_events.append(rejected_join_rules)
+
+        self.assertRaises(
+            AuthError,
+            event_auth.check_auth_rules_for_event,
+            RoomVersions.V9,
+            _random_state_event(creator),
+            auth_events,
+        )
+
+        # ... even if there is *also* a good join rules
+        auth_events.append(_join_rules_event(creator, "public"))
+
+        self.assertRaises(
+            AuthError,
+            event_auth.check_auth_rules_for_event,
+            RoomVersions.V9,
+            _random_state_event(creator),
+            auth_events,
+        )
+
     def test_random_users_cannot_send_state_before_first_pl(self):
         """
         Check that, before the first PL lands, the creator is the only user
@@ -31,11 +72,11 @@ class EventAuthTestCase(unittest.TestCase):
         """
         creator = "@creator:example.com"
         joiner = "@joiner:example.com"
-        auth_events = {
-            ("m.room.create", ""): _create_event(creator),
-            ("m.room.member", creator): _join_event(creator),
-            ("m.room.member", joiner): _join_event(joiner),
-        }
+        auth_events = [
+            _create_event(creator),
+            _join_event(creator),
+            _join_event(joiner),
+        ]
 
         # creator should be able to send state
         event_auth.check_auth_rules_for_event(
@@ -62,15 +103,15 @@ class EventAuthTestCase(unittest.TestCase):
         pleb = "@joiner:example.com"
         king = "@joiner2:example.com"
 
-        auth_events = {
-            ("m.room.create", ""): _create_event(creator),
-            ("m.room.member", creator): _join_event(creator),
-            ("m.room.power_levels", ""): _power_levels_event(
+        auth_events = [
+            _create_event(creator),
+            _join_event(creator),
+            _power_levels_event(
                 creator, {"state_default": "30", "users": {pleb: "29", king: "30"}}
             ),
-            ("m.room.member", pleb): _join_event(pleb),
-            ("m.room.member", king): _join_event(king),
-        }
+            _join_event(pleb),
+            _join_event(king),
+        ]
 
         # pleb should not be able to send state
         self.assertRaises(
@@ -92,10 +133,10 @@ class EventAuthTestCase(unittest.TestCase):
         """Alias events have special behavior up through room version 6."""
         creator = "@creator:example.com"
         other = "@other:example.com"
-        auth_events = {
-            ("m.room.create", ""): _create_event(creator),
-            ("m.room.member", creator): _join_event(creator),
-        }
+        auth_events = [
+            _create_event(creator),
+            _join_event(creator),
+        ]
 
         # creator should be able to send aliases
         event_auth.check_auth_rules_for_event(
@@ -131,10 +172,10 @@ class EventAuthTestCase(unittest.TestCase):
         """After MSC2432, alias events have no special behavior."""
         creator = "@creator:example.com"
         other = "@other:example.com"
-        auth_events = {
-            ("m.room.create", ""): _create_event(creator),
-            ("m.room.member", creator): _join_event(creator),
-        }
+        auth_events = [
+            _create_event(creator),
+            _join_event(creator),
+        ]
 
         # creator should be able to send aliases
         event_auth.check_auth_rules_for_event(
@@ -170,14 +211,14 @@ class EventAuthTestCase(unittest.TestCase):
         creator = "@creator:example.com"
         pleb = "@joiner:example.com"
 
-        auth_events = {
-            ("m.room.create", ""): _create_event(creator),
-            ("m.room.member", creator): _join_event(creator),
-            ("m.room.power_levels", ""): _power_levels_event(
+        auth_events = [
+            _create_event(creator),
+            _join_event(creator),
+            _power_levels_event(
                 creator, {"state_default": "30", "users": {pleb: "30"}}
             ),
-            ("m.room.member", pleb): _join_event(pleb),
-        }
+            _join_event(pleb),
+        ]
 
         # pleb should be able to modify the notifications power level.
         event_auth.check_auth_rules_for_event(
@@ -211,7 +252,7 @@ class EventAuthTestCase(unittest.TestCase):
         event_auth.check_auth_rules_for_event(
             RoomVersions.V6,
             _join_event(pleb),
-            auth_events,
+            auth_events.values(),
         )
 
         # A user cannot be force-joined to a room.
@@ -219,7 +260,7 @@ class EventAuthTestCase(unittest.TestCase):
             event_auth.check_auth_rules_for_event(
                 RoomVersions.V6,
                 _member_event(pleb, "join", sender=creator),
-                auth_events,
+                auth_events.values(),
             )
 
         # Banned should be rejected.
@@ -228,7 +269,7 @@ class EventAuthTestCase(unittest.TestCase):
             event_auth.check_auth_rules_for_event(
                 RoomVersions.V6,
                 _join_event(pleb),
-                auth_events,
+                auth_events.values(),
             )
 
         # A user who left can re-join.
@@ -236,7 +277,7 @@ class EventAuthTestCase(unittest.TestCase):
         event_auth.check_auth_rules_for_event(
             RoomVersions.V6,
             _join_event(pleb),
-            auth_events,
+            auth_events.values(),
         )
 
         # A user can send a join if they're in the room.
@@ -244,7 +285,7 @@ class EventAuthTestCase(unittest.TestCase):
         event_auth.check_auth_rules_for_event(
             RoomVersions.V6,
             _join_event(pleb),
-            auth_events,
+            auth_events.values(),
         )
 
         # A user can accept an invite.
@@ -254,7 +295,7 @@ class EventAuthTestCase(unittest.TestCase):
         event_auth.check_auth_rules_for_event(
             RoomVersions.V6,
             _join_event(pleb),
-            auth_events,
+            auth_events.values(),
         )
 
     def test_join_rules_invite(self):
@@ -275,7 +316,7 @@ class EventAuthTestCase(unittest.TestCase):
             event_auth.check_auth_rules_for_event(
                 RoomVersions.V6,
                 _join_event(pleb),
-                auth_events,
+                auth_events.values(),
             )
 
         # A user cannot be force-joined to a room.
@@ -283,7 +324,7 @@ class EventAuthTestCase(unittest.TestCase):
             event_auth.check_auth_rules_for_event(
                 RoomVersions.V6,
                 _member_event(pleb, "join", sender=creator),
-                auth_events,
+                auth_events.values(),
             )
 
         # Banned should be rejected.
@@ -292,7 +333,7 @@ class EventAuthTestCase(unittest.TestCase):
             event_auth.check_auth_rules_for_event(
                 RoomVersions.V6,
                 _join_event(pleb),
-                auth_events,
+                auth_events.values(),
             )
 
         # A user who left cannot re-join.
@@ -301,7 +342,7 @@ class EventAuthTestCase(unittest.TestCase):
             event_auth.check_auth_rules_for_event(
                 RoomVersions.V6,
                 _join_event(pleb),
-                auth_events,
+                auth_events.values(),
             )
 
         # A user can send a join if they're in the room.
@@ -309,7 +350,7 @@ class EventAuthTestCase(unittest.TestCase):
         event_auth.check_auth_rules_for_event(
             RoomVersions.V6,
             _join_event(pleb),
-            auth_events,
+            auth_events.values(),
         )
 
         # A user can accept an invite.
@@ -319,7 +360,7 @@ class EventAuthTestCase(unittest.TestCase):
         event_auth.check_auth_rules_for_event(
             RoomVersions.V6,
             _join_event(pleb),
-            auth_events,
+            auth_events.values(),
         )
 
     def test_join_rules_msc3083_restricted(self):
@@ -347,7 +388,7 @@ class EventAuthTestCase(unittest.TestCase):
             event_auth.check_auth_rules_for_event(
                 RoomVersions.V6,
                 _join_event(pleb),
-                auth_events,
+                auth_events.values(),
             )
 
         # A properly formatted join event should work.
@@ -360,7 +401,7 @@ class EventAuthTestCase(unittest.TestCase):
         event_auth.check_auth_rules_for_event(
             RoomVersions.V8,
             authorised_join_event,
-            auth_events,
+            auth_events.values(),
         )
 
         # A join issued by a specific user works (i.e. the power level checks
@@ -380,7 +421,7 @@ class EventAuthTestCase(unittest.TestCase):
                     EventContentFields.AUTHORISING_USER: "@inviter:foo.test"
                 },
             ),
-            pl_auth_events,
+            pl_auth_events.values(),
         )
 
         # A join which is missing an authorised server is rejected.
@@ -388,7 +429,7 @@ class EventAuthTestCase(unittest.TestCase):
             event_auth.check_auth_rules_for_event(
                 RoomVersions.V8,
                 _join_event(pleb),
-                auth_events,
+                auth_events.values(),
             )
 
         # An join authorised by a user who is not in the room is rejected.
@@ -405,7 +446,7 @@ class EventAuthTestCase(unittest.TestCase):
                         EventContentFields.AUTHORISING_USER: "@other:example.com"
                     },
                 ),
-                auth_events,
+                auth_events.values(),
             )
 
         # A user cannot be force-joined to a room. (This uses an event which
@@ -421,7 +462,7 @@ class EventAuthTestCase(unittest.TestCase):
                         EventContentFields.AUTHORISING_USER: "@inviter:foo.test"
                     },
                 ),
-                auth_events,
+                auth_events.values(),
             )
 
         # Banned should be rejected.
@@ -430,7 +471,7 @@ class EventAuthTestCase(unittest.TestCase):
             event_auth.check_auth_rules_for_event(
                 RoomVersions.V8,
                 authorised_join_event,
-                auth_events,
+                auth_events.values(),
             )
 
         # A user who left can re-join.
@@ -438,7 +479,7 @@ class EventAuthTestCase(unittest.TestCase):
         event_auth.check_auth_rules_for_event(
             RoomVersions.V8,
             authorised_join_event,
-            auth_events,
+            auth_events.values(),
         )
 
         # A user can send a join if they're in the room. (This doesn't need to
@@ -447,7 +488,7 @@ class EventAuthTestCase(unittest.TestCase):
         event_auth.check_auth_rules_for_event(
             RoomVersions.V8,
             _join_event(pleb),
-            auth_events,
+            auth_events.values(),
         )
 
         # A user can accept an invite. (This doesn't need to be authorised since
@@ -458,7 +499,7 @@ class EventAuthTestCase(unittest.TestCase):
         event_auth.check_auth_rules_for_event(
             RoomVersions.V8,
             _join_event(pleb),
-            auth_events,
+            auth_events.values(),
         )
 
 
@@ -473,6 +514,7 @@ def _create_event(user_id: str) -> EventBase:
             "room_id": TEST_ROOM_ID,
             "event_id": _get_event_id(),
             "type": "m.room.create",
+            "state_key": "",
             "sender": user_id,
             "content": {"creator": user_id},
         }