diff --git a/synapse/api/auth/msc3861_delegated.py b/synapse/api/auth/msc3861_delegated.py
index 4bdfe31b22..14cba50c90 100644
--- a/synapse/api/auth/msc3861_delegated.py
+++ b/synapse/api/auth/msc3861_delegated.py
@@ -438,3 +438,16 @@ class MSC3861DelegatedAuth(BaseAuth):
scope=scope,
is_guest=(has_guest_scope and not has_user_scope),
)
+
+ def invalidate_cached_tokens(self, keys: List[str]) -> None:
+ """
+ Invalidate the entry(s) in the introspection token cache corresponding to the given key
+ """
+ for key in keys:
+ self._token_cache.invalidate(key)
+
+ def invalidate_token_cache(self) -> None:
+ """
+ Invalidate the entire token cache.
+ """
+ self._token_cache.invalidate_all()
diff --git a/synapse/replication/tcp/client.py b/synapse/replication/tcp/client.py
index 139f57cf86..04e8cff6ea 100644
--- a/synapse/replication/tcp/client.py
+++ b/synapse/replication/tcp/client.py
@@ -26,6 +26,7 @@ from synapse.logging.context import PreserveLoggingContext, make_deferred_yielda
from synapse.metrics.background_process_metrics import run_as_background_process
from synapse.replication.tcp.streams import (
AccountDataStream,
+ CachesStream,
DeviceListsStream,
PushersStream,
PushRulesStream,
@@ -73,6 +74,7 @@ class ReplicationDataHandler:
self._instance_name = hs.get_instance_name()
self._typing_handler = hs.get_typing_handler()
self._state_storage_controller = hs.get_storage_controllers().state
+ self.auth = hs.get_auth()
self._notify_pushers = hs.config.worker.start_pushers
self._pusher_pool = hs.get_pusherpool()
@@ -218,6 +220,16 @@ class ReplicationDataHandler:
self._state_storage_controller.notify_event_un_partial_stated(
row.event_id
)
+ # invalidate the introspection token cache
+ elif stream_name == CachesStream.NAME:
+ for row in rows:
+ if row.cache_func == "introspection_token_invalidation":
+ if row.keys[0] is None:
+ # invalidate the whole cache
+ # mypy ignore - the token cache is defined on MSC3861DelegatedAuth
+ self.auth.invalidate_token_cache() # type: ignore[attr-defined]
+ else:
+ self.auth.invalidate_cached_tokens(row.keys) # type: ignore[attr-defined]
await self._presence_handler.process_replication_rows(
stream_name, instance_name, token, rows
diff --git a/synapse/rest/admin/__init__.py b/synapse/rest/admin/__init__.py
index fe8177ed4d..55e752fda8 100644
--- a/synapse/rest/admin/__init__.py
+++ b/synapse/rest/admin/__init__.py
@@ -47,6 +47,7 @@ from synapse.rest.admin.federation import (
ListDestinationsRestServlet,
)
from synapse.rest.admin.media import ListMediaInRoom, register_servlets_for_media_repo
+from synapse.rest.admin.oidc import OIDCTokenRevocationRestServlet
from synapse.rest.admin.registration_tokens import (
ListRegistrationTokensRestServlet,
NewRegistrationTokenRestServlet,
@@ -297,6 +298,8 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
BackgroundUpdateRestServlet(hs).register(http_server)
BackgroundUpdateStartJobRestServlet(hs).register(http_server)
ExperimentalFeaturesRestServlet(hs).register(http_server)
+ if hs.config.experimental.msc3861.enabled:
+ OIDCTokenRevocationRestServlet(hs).register(http_server)
def register_servlets_for_client_rest_resource(
diff --git a/synapse/rest/admin/oidc.py b/synapse/rest/admin/oidc.py
new file mode 100644
index 0000000000..64d2d40550
--- /dev/null
+++ b/synapse/rest/admin/oidc.py
@@ -0,0 +1,55 @@
+# Copyright 2023 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.
+from http import HTTPStatus
+from typing import TYPE_CHECKING, Dict, Tuple
+
+from synapse.http.servlet import RestServlet
+from synapse.http.site import SynapseRequest
+from synapse.rest.admin._base import admin_patterns, assert_requester_is_admin
+
+if TYPE_CHECKING:
+ from synapse.server import HomeServer
+
+
+class OIDCTokenRevocationRestServlet(RestServlet):
+ """
+ Delete a given token introspection response - identified by the `jti` field - from the
+ introspection token cache when a token is revoked at the authorizing server
+ """
+
+ PATTERNS = admin_patterns("/OIDC_token_revocation/(?P<token_id>[^/]*)")
+
+ def __init__(self, hs: "HomeServer"):
+ super().__init__()
+ auth = hs.get_auth()
+
+ # If this endpoint is loaded then we must have enabled delegated auth.
+ from synapse.api.auth.msc3861_delegated import MSC3861DelegatedAuth
+
+ assert isinstance(auth, MSC3861DelegatedAuth)
+
+ self.auth = auth
+ self.store = hs.get_datastores().main
+
+ async def on_DELETE(
+ self, request: SynapseRequest, token_id: str
+ ) -> Tuple[HTTPStatus, Dict]:
+ await assert_requester_is_admin(self.auth, request)
+
+ self.auth._token_cache.invalidate(token_id)
+
+ # make sure we invalidate the cache on any workers
+ await self.store.stream_introspection_token_invalidation((token_id,))
+
+ return HTTPStatus.OK, {}
diff --git a/synapse/storage/databases/main/cache.py b/synapse/storage/databases/main/cache.py
index 2fbd389c71..18905e07b6 100644
--- a/synapse/storage/databases/main/cache.py
+++ b/synapse/storage/databases/main/cache.py
@@ -584,6 +584,19 @@ class CacheInvalidationWorkerStore(SQLBaseStore):
else:
return 0
+ async def stream_introspection_token_invalidation(
+ self, key: Tuple[Optional[str]]
+ ) -> None:
+ """
+ Stream an invalidation request for the introspection token cache to workers
+
+ Args:
+ key: token_id of the introspection token to remove from the cache
+ """
+ await self.send_invalidation_to_replication(
+ "introspection_token_invalidation", key
+ )
+
@wrap_as_background_process("clean_up_old_cache_invalidations")
async def _clean_up_cache_invalidation_wrapper(self) -> None:
"""
diff --git a/synapse/storage/databases/main/devices.py b/synapse/storage/databases/main/devices.py
index e4162f846b..fa69a4a298 100644
--- a/synapse/storage/databases/main/devices.py
+++ b/synapse/storage/databases/main/devices.py
@@ -33,6 +33,7 @@ from typing_extensions import Literal
from synapse.api.constants import EduTypes
from synapse.api.errors import Codes, StoreError
+from synapse.config.homeserver import HomeServerConfig
from synapse.logging.opentracing import (
get_active_span_text_map,
set_tag,
@@ -1663,6 +1664,7 @@ class DeviceStore(DeviceWorkerStore, DeviceBackgroundUpdateStore):
self.device_id_exists_cache: LruCache[
Tuple[str, str], Literal[True]
] = LruCache(cache_name="device_id_exists", max_size=10000)
+ self.config: HomeServerConfig = hs.config
async def store_device(
self,
@@ -1784,6 +1786,13 @@ class DeviceStore(DeviceWorkerStore, DeviceBackgroundUpdateStore):
for device_id in device_ids:
self.device_id_exists_cache.invalidate((user_id, device_id))
+ # TODO: don't nuke the entire cache once there is a way to associate
+ # device_id -> introspection_token
+ if self.config.experimental.msc3861.enabled:
+ # mypy ignore - the token cache is defined on MSC3861DelegatedAuth
+ self.auth._token_cache.invalidate_all() # type: ignore[attr-defined]
+ await self.stream_introspection_token_invalidation((None,))
+
async def update_device(
self, user_id: str, device_id: str, new_display_name: Optional[str] = None
) -> None:
diff --git a/synapse/util/caches/expiringcache.py b/synapse/util/caches/expiringcache.py
index 01ad02af67..9a3e10ddee 100644
--- a/synapse/util/caches/expiringcache.py
+++ b/synapse/util/caches/expiringcache.py
@@ -140,6 +140,20 @@ class ExpiringCache(Generic[KT, VT]):
return value.value
+ def invalidate(self, key: KT) -> None:
+ """
+ Remove the given key from the cache.
+ """
+
+ value = self._cache.pop(key, None)
+ if value:
+ if self.iterable:
+ self.metrics.inc_evictions(
+ EvictionReason.invalidation, len(value.value)
+ )
+ else:
+ self.metrics.inc_evictions(EvictionReason.invalidation)
+
def __contains__(self, key: KT) -> bool:
return key in self._cache
@@ -193,6 +207,14 @@ class ExpiringCache(Generic[KT, VT]):
len(self),
)
+ def invalidate_all(self) -> None:
+ """
+ Remove all items from the cache.
+ """
+ keys = set(self._cache.keys())
+ for key in keys:
+ self._cache.pop(key)
+
def __len__(self) -> int:
if self.iterable:
return sum(len(entry.value) for entry in self._cache.values())
|
