diff --git a/synapse/rest/admin/users.py b/synapse/rest/admin/users.py
index 8efefbc0a0..2d8d963062 100644
--- a/synapse/rest/admin/users.py
+++ b/synapse/rest/admin/users.py
@@ -16,6 +16,7 @@ import hashlib
import hmac
import logging
from http import HTTPStatus
+from typing import TYPE_CHECKING
from synapse.api.constants import UserTypes
from synapse.api.errors import Codes, NotFoundError, SynapseError
@@ -35,6 +36,9 @@ from synapse.rest.admin._base import (
)
from synapse.types import UserID
+if TYPE_CHECKING:
+ from synapse.server import HomeServer
+
logger = logging.getLogger(__name__)
@@ -708,3 +712,42 @@ class UserMembershipRestServlet(RestServlet):
ret = {"joined_rooms": list(room_ids), "total": len(room_ids)}
return 200, ret
+
+
+class UserTokenRestServlet(RestServlet):
+ """An admin API for logging in as a user.
+ """
+
+ PATTERNS = admin_patterns("/users/(?P<user_id>[^/]*)/login$")
+
+ def __init__(self, hs: "HomeServer"):
+ self.hs = hs
+ self.store = hs.get_datastore()
+ self.auth = hs.get_auth()
+ self.auth_handler = hs.get_auth_handler()
+
+ async def on_PUT(self, request, user_id):
+ requester = await self.auth.get_user_by_req(request)
+ await assert_user_is_admin(self.auth, requester.user)
+ auth_user = requester.user
+
+ if not self.hs.is_mine_id(user_id):
+ raise SynapseError(400, "Only local users can be logged in as")
+
+ body = parse_json_object_from_request(request)
+
+ valid_until_ms = body.get("valid_until_ms")
+ if valid_until_ms and not isinstance(valid_until_ms, int):
+ raise SynapseError(400, "'valid_until_ms' parameter must be an int")
+
+ if auth_user.to_string() == user_id:
+ raise SynapseError(400, "Cannot use admin API to login as self")
+
+ token = await self.auth_handler.get_access_token_for_user_id(
+ user_id=auth_user.to_string(),
+ device_id=None,
+ valid_until_ms=valid_until_ms,
+ puppets_user_id=user_id,
+ )
+
+ return 200, {"access_token": token}
|
