diff --git a/changelog.d/16132.misc b/changelog.d/16132.misc
new file mode 100644
index 0000000000..aca26079d8
--- /dev/null
+++ b/changelog.d/16132.misc
@@ -0,0 +1 @@
+MSC3861: allow impersonation by an admin user using `_oidc_admin_impersonate_user_id` query parameter.
diff --git a/synapse/api/auth/msc3861_delegated.py b/synapse/api/auth/msc3861_delegated.py
index 18875f2c81..4bdfe31b22 100644
--- a/synapse/api/auth/msc3861_delegated.py
+++ b/synapse/api/auth/msc3861_delegated.py
@@ -246,7 +246,7 @@ class MSC3861DelegatedAuth(BaseAuth):
return introspection_token
async def is_server_admin(self, requester: Requester) -> bool:
- return "urn:synapse:admin:*" in requester.scope
+ return SCOPE_SYNAPSE_ADMIN in requester.scope
async def get_user_by_req(
self,
@@ -263,6 +263,25 @@ class MSC3861DelegatedAuth(BaseAuth):
# so that we don't provision the user if they don't have enough permission:
requester = await self.get_user_by_access_token(access_token, allow_expired)
+ # Allow impersonation by an admin user using `_oidc_admin_impersonate_user_id` query parameter
+ if request.args is not None:
+ user_id_params = request.args.get(b"_oidc_admin_impersonate_user_id")
+ if user_id_params:
+ if await self.is_server_admin(requester):
+ user_id_str = user_id_params[0].decode("ascii")
+ impersonated_user_id = UserID.from_string(user_id_str)
+ logging.info(f"Admin impersonation of user {user_id_str}")
+ requester = create_requester(
+ user_id=impersonated_user_id,
+ scope=[SCOPE_MATRIX_API],
+ authenticated_entity=requester.user.to_string(),
+ )
+ else:
+ raise AuthError(
+ 401,
+ "Impersonation not possible by a non admin user",
+ )
+
# Deny the request if the user account is locked.
if not allow_locked and await self.store.get_user_locked_status(
requester.user.to_string()
@@ -290,14 +309,14 @@ class MSC3861DelegatedAuth(BaseAuth):
# XXX: This is a temporary solution so that the admin API can be called by
# the OIDC provider. This will be removed once we have OIDC client
# credentials grant support in matrix-authentication-service.
- logging.info("Admin toked used")
+ logging.info("Admin token used")
# XXX: that user doesn't exist and won't be provisioned.
# This is mostly fine for admin calls, but we should also think about doing
# requesters without a user_id.
admin_user = UserID("__oidc_admin", self._hostname)
return create_requester(
user_id=admin_user,
- scope=["urn:synapse:admin:*"],
+ scope=[SCOPE_SYNAPSE_ADMIN],
)
try:
diff --git a/tests/handlers/test_oauth_delegation.py b/tests/handlers/test_oauth_delegation.py
index 82c26e303f..1456b675a7 100644
--- a/tests/handlers/test_oauth_delegation.py
+++ b/tests/handlers/test_oauth_delegation.py
@@ -340,6 +340,41 @@ class MSC3861OAuthDelegation(HomeserverTestCase):
get_awaitable_result(self.auth.is_server_admin(requester)), False
)
+ def test_active_user_admin_impersonation(self) -> None:
+ """The handler should return a requester with normal user rights
+ and an user ID matching the one specified in query param `user_id`"""
+
+ self.http_client.request = simple_async_mock(
+ return_value=FakeResponse.json(
+ code=200,
+ payload={
+ "active": True,
+ "sub": SUBJECT,
+ "scope": " ".join([SYNAPSE_ADMIN_SCOPE, MATRIX_USER_SCOPE]),
+ "username": USERNAME,
+ },
+ )
+ )
+ request = Mock(args={})
+ request.args[b"access_token"] = [b"mockAccessToken"]
+ impersonated_user_id = f"@{USERNAME}:{SERVER_NAME}"
+ request.args[b"_oidc_admin_impersonate_user_id"] = [
+ impersonated_user_id.encode("ascii")
+ ]
+ request.requestHeaders.getRawHeaders = mock_getRawHeaders()
+ requester = self.get_success(self.auth.get_user_by_req(request))
+ self.http_client.get_json.assert_called_once_with(WELL_KNOWN)
+ self.http_client.request.assert_called_once_with(
+ method="POST", uri=INTROSPECTION_ENDPOINT, data=ANY, headers=ANY
+ )
+ self._assertParams()
+ self.assertEqual(requester.user.to_string(), impersonated_user_id)
+ self.assertEqual(requester.is_guest, False)
+ self.assertEqual(requester.device_id, None)
+ self.assertEqual(
+ get_awaitable_result(self.auth.is_server_admin(requester)), False
+ )
+
def test_active_user_with_device(self) -> None:
"""The handler should return a requester with normal user rights and a device ID."""
|
