mirrors/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2024-12-14 11:57:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Allow admin users to create or modify users without a shared secret (#6495)

Browse source 
Signed-off-by: Manuel Stahl <manuel.stahl@awesome-technologies.de>
This commit is contained in:
Manuel Stahl 2020-01-09 14:31:00 +01:00 committed by Richard van der Hoff
parent d773290cb1
commit d2906fe666
9 changed files with 655 additions and 339 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
changelog.d/5742.feature Normal file
View file

@ -0,0 +1 @@
Allow admin to create or modify a user. Contributed by Awesome Technologies Innovationslabor GmbH.

33
docs/admin_api/user_admin_api.rst
View file

@ -1,3 +1,33 @@
Create or modify Account
========================
This API allows an administrator to create or modify a user account with a
specific ``user_id``.
This api is::
PUT /_synapse/admin/v2/users/<user_id>
with a body of:
.. code:: json
{
"password": "user_password",
"displayname": "User",
"avatar_url": "<avatar_url>",
"admin": false,
"deactivated": false
}
including an ``access_token`` of a server admin.
The parameter ``displayname`` is optional and defaults to ``user_id``.
The parameter ``avatar_url`` is optional.
The parameter ``admin`` is optional and defaults to 'false'.
The parameter ``deactivated`` is optional and defaults to 'false'.
If the user already exists then optional parameters default to the current value.
List Accounts
=============
@ -50,7 +80,8 @@ This API returns information about a specific user account.
The api is::
GET /_synapse/admin/v1/whois/<user_id>
GET /_synapse/admin/v1/whois/<user_id> (deprecated)
GET /_synapse/admin/v2/users/<user_id>
including an ``access_token`` of a server admin.

9
synapse/handlers/admin.py
View file

@ -51,6 +51,15 @@ class AdminHandler(BaseHandler):
return ret
async def get_user(self, user):
"""Function to get user details"""
ret = await self.store.get_user_by_id(user.to_string())
if ret:
profile = await self.store.get_profileinfo(user.localpart)
ret["displayname"] = profile.display_name
ret["avatar_url"] = profile.avatar_url
return ret
async def get_users(self):
"""Function to retrieve a list of users in users table.

2
synapse/rest/admin/__init__.py
View file

@ -38,6 +38,7 @@ from synapse.rest.admin.users import (
SearchUsersRestServlet,
UserAdminServlet,
UserRegisterServlet,
UserRestServletV2,
UsersRestServlet,
UsersRestServletV2,
WhoisRestServlet,
@ -191,6 +192,7 @@ def register_servlets(hs, http_server):
SendServerNoticeServlet(hs).register(http_server)
VersionServlet(hs).register(http_server)
UserAdminServlet(hs).register(http_server)
UserRestServletV2(hs).register(http_server)
UsersRestServletV2(hs).register(http_server)

142
synapse/rest/admin/users.py
View file

@ -102,6 +102,148 @@ class UsersRestServletV2(RestServlet):
return 200, ret
class UserRestServletV2(RestServlet):
PATTERNS = (re.compile("^/_synapse/admin/v2/users/(?P<user_id>@[^/]+)$"),)
"""Get request to list user details.
This needs user to have administrator access in Synapse.
GET /_synapse/admin/v2/users/<user_id>
returns:
200 OK with user details if success otherwise an error.
Put request to allow an administrator to add or modify a user.
This needs user to have administrator access in Synapse.
We use PUT instead of POST since we already know the id of the user
object to create. POST could be used to create guests.
PUT /_synapse/admin/v2/users/<user_id>
{
"password": "secret",
"displayname": "User"
}
returns:
201 OK with new user object if user was created or
200 OK with modified user object if user was modified
otherwise an error.
"""
def __init__(self, hs):
self.hs = hs
self.auth = hs.get_auth()
self.admin_handler = hs.get_handlers().admin_handler
self.profile_handler = hs.get_profile_handler()
self.set_password_handler = hs.get_set_password_handler()
self.deactivate_account_handler = hs.get_deactivate_account_handler()
self.registration_handler = hs.get_registration_handler()
async def on_GET(self, request, user_id):
await assert_requester_is_admin(self.auth, request)
target_user = UserID.from_string(user_id)
if not self.hs.is_mine(target_user):
raise SynapseError(400, "Can only lookup local users")
ret = await self.admin_handler.get_user(target_user)
return 200, ret
async def on_PUT(self, request, user_id):
await assert_requester_is_admin(self.auth, request)
target_user = UserID.from_string(user_id)
body = parse_json_object_from_request(request)
if not self.hs.is_mine(target_user):
raise SynapseError(400, "This endpoint can only be used with local users")
user = await self.admin_handler.get_user(target_user)
if user: # modify user
requester = await self.auth.get_user_by_req(request)
if "displayname" in body:
await self.profile_handler.set_displayname(
target_user, requester, body["displayname"], True
)
if "avatar_url" in body:
await self.profile_handler.set_avatar_url(
target_user, requester, body["avatar_url"], True
)
if "admin" in body:
set_admin_to = bool(body["admin"])
if set_admin_to != user["admin"]:
auth_user = requester.user
if target_user == auth_user and not set_admin_to:
raise SynapseError(400, "You may not demote yourself.")
await self.admin_handler.set_user_server_admin(
target_user, set_admin_to
)
if "password" in body:
if (
not isinstance(body["password"], text_type)
or len(body["password"]) > 512
):
raise SynapseError(400, "Invalid password")
else:
new_password = body["password"]
await self._set_password_handler.set_password(
target_user, new_password, requester
)
if "deactivated" in body:
deactivate = bool(body["deactivated"])
if deactivate and not user["deactivated"]:
result = await self.deactivate_account_handler.deactivate_account(
target_user.to_string(), False
)
if not result:
raise SynapseError(500, "Could not deactivate user")
user = await self.admin_handler.get_user(target_user)
return 200, user
else: # create user
if "password" not in body:
raise SynapseError(
400, "password must be specified", errcode=Codes.BAD_JSON
)
elif (
not isinstance(body["password"], text_type)
or len(body["password"]) > 512
):
raise SynapseError(400, "Invalid password")
admin = body.get("admin", None)
user_type = body.get("user_type", None)
displayname = body.get("displayname", None)
if user_type is not None and user_type not in UserTypes.ALL_USER_TYPES:
raise SynapseError(400, "Invalid user type")
user_id = await self.registration_handler.register_user(
localpart=target_user.localpart,
password=body["password"],
admin=bool(admin),
default_display_name=displayname,
user_type=user_type,
)
if "avatar_url" in body:
await self.profile_handler.set_avatar_url(
user_id, requester, body["avatar_url"], True
)
ret = await self.admin_handler.get_user(target_user)
return 201, ret
class UserRegisterServlet(RestServlet):
"""
Attributes:

2
synapse/storage/data_stores/main/registration.py
View file

@ -52,11 +52,13 @@ class RegistrationWorkerStore(SQLBaseStore):
"name",
"password_hash",
"is_guest",
"admin",
"consent_version",
"consent_server_notice_sent",
"appservice_id",
"creation_ts",
"user_type",
"deactivated",
],
allow_none=True,
desc="get_user_by_id",

338
tests/rest/admin/test_admin.py
View file

@ -13,14 +13,11 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import hashlib
import hmac
import json
from mock import Mock
import synapse.rest.admin
from synapse.api.constants import UserTypes
from synapse.http.server import JsonResource
from synapse.rest.admin import VersionServlet
from synapse.rest.client.v1 import events, login, room
@ -47,341 +44,6 @@ class VersionTestCase(unittest.HomeserverTestCase):
)
class UserRegisterTestCase(unittest.HomeserverTestCase):
servlets = [synapse.rest.admin.register_servlets_for_client_rest_resource]
def make_homeserver(self, reactor, clock):
self.url = "/_matrix/client/r0/admin/register"
self.registration_handler = Mock()
self.identity_handler = Mock()
self.login_handler = Mock()
self.device_handler = Mock()
self.device_handler.check_device_registered = Mock(return_value="FAKE")
self.datastore = Mock(return_value=Mock())
self.datastore.get_current_state_deltas = Mock(return_value=(0, []))
self.secrets = Mock()
self.hs = self.setup_test_homeserver()
self.hs.config.registration_shared_secret = "shared"
self.hs.get_media_repository = Mock()
self.hs.get_deactivate_account_handler = Mock()
return self.hs
def test_disabled(self):
"""
If there is no shared secret, registration through this method will be
prevented.
"""
self.hs.config.registration_shared_secret = None
request, channel = self.make_request("POST", self.url, b"{}")
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual(
"Shared secret registration is not enabled", channel.json_body["error"]
)
def test_get_nonce(self):
"""
Calling GET on the endpoint will return a randomised nonce, using the
homeserver's secrets provider.
"""
secrets = Mock()
secrets.token_hex = Mock(return_value="abcd")
self.hs.get_secrets = Mock(return_value=secrets)
request, channel = self.make_request("GET", self.url)
self.render(request)
self.assertEqual(channel.json_body, {"nonce": "abcd"})
def test_expired_nonce(self):
"""
Calling GET on the endpoint will return a randomised nonce, which will
only last for SALT_TIMEOUT (60s).
"""
request, channel = self.make_request("GET", self.url)
self.render(request)
nonce = channel.json_body["nonce"]
# 59 seconds
self.reactor.advance(59)
body = json.dumps({"nonce": nonce})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("username must be specified", channel.json_body["error"])
# 61 seconds
self.reactor.advance(2)
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("unrecognised nonce", channel.json_body["error"])
def test_register_incorrect_nonce(self):
"""
Only the provided nonce can be used, as it's checked in the MAC.
"""
request, channel = self.make_request("GET", self.url)
self.render(request)
nonce = channel.json_body["nonce"]
want_mac = hmac.new(key=b"shared", digestmod=hashlib.sha1)
want_mac.update(b"notthenonce\x00bob\x00abc123\x00admin")
want_mac = want_mac.hexdigest()
body = json.dumps(
{
"nonce": nonce,
"username": "bob",
"password": "abc123",
"admin": True,
"mac": want_mac,
}
)
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("HMAC incorrect", channel.json_body["error"])
def test_register_correct_nonce(self):
"""
When the correct nonce is provided, and the right key is provided, the
user is registered.
"""
request, channel = self.make_request("GET", self.url)
self.render(request)
nonce = channel.json_body["nonce"]
want_mac = hmac.new(key=b"shared", digestmod=hashlib.sha1)
want_mac.update(
nonce.encode("ascii") + b"\x00bob\x00abc123\x00admin\x00support"
)
want_mac = want_mac.hexdigest()
body = json.dumps(
{
"nonce": nonce,
"username": "bob",
"password": "abc123",
"admin": True,
"user_type": UserTypes.SUPPORT,
"mac": want_mac,
}
)
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("@bob:test", channel.json_body["user_id"])
def test_nonce_reuse(self):
"""
A valid unrecognised nonce.
"""
request, channel = self.make_request("GET", self.url)
self.render(request)
nonce = channel.json_body["nonce"]
want_mac = hmac.new(key=b"shared", digestmod=hashlib.sha1)
want_mac.update(nonce.encode("ascii") + b"\x00bob\x00abc123\x00admin")
want_mac = want_mac.hexdigest()
body = json.dumps(
{
"nonce": nonce,
"username": "bob",
"password": "abc123",
"admin": True,
"mac": want_mac,
}
)
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("@bob:test", channel.json_body["user_id"])
# Now, try and reuse it
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("unrecognised nonce", channel.json_body["error"])
def test_missing_parts(self):
"""
Synapse will complain if you don't give nonce, username, password, and
mac. Admin and user_types are optional. Additional checks are done for length
and type.
"""
def nonce():
request, channel = self.make_request("GET", self.url)
self.render(request)
return channel.json_body["nonce"]
#
# Nonce check
#
# Must be present
body = json.dumps({})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("nonce must be specified", channel.json_body["error"])
#
# Username checks
#
# Must be present
body = json.dumps({"nonce": nonce()})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("username must be specified", channel.json_body["error"])
# Must be a string
body = json.dumps({"nonce": nonce(), "username": 1234})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("Invalid username", channel.json_body["error"])
# Must not have null bytes
body = json.dumps({"nonce": nonce(), "username": "abcd\u0000"})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("Invalid username", channel.json_body["error"])
# Must not have null bytes
body = json.dumps({"nonce": nonce(), "username": "a" * 1000})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("Invalid username", channel.json_body["error"])
#
# Password checks
#
# Must be present
body = json.dumps({"nonce": nonce(), "username": "a"})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("password must be specified", channel.json_body["error"])
# Must be a string
body = json.dumps({"nonce": nonce(), "username": "a", "password": 1234})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("Invalid password", channel.json_body["error"])
# Must not have null bytes
body = json.dumps({"nonce": nonce(), "username": "a", "password": "abcd\u0000"})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("Invalid password", channel.json_body["error"])
# Super long
body = json.dumps({"nonce": nonce(), "username": "a", "password": "A" * 1000})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("Invalid password", channel.json_body["error"])
#
# user_type check
#
# Invalid user_type
body = json.dumps(
{
"nonce": nonce(),
"username": "a",
"password": "1234",
"user_type": "invalid",
}
)
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("Invalid user type", channel.json_body["error"])
class UsersListTestCase(unittest.HomeserverTestCase):
servlets = [
synapse.rest.admin.register_servlets,
login.register_servlets,
]
url = "/_synapse/admin/v2/users"
def prepare(self, reactor, clock, hs):
self.admin_user = self.register_user("admin", "pass", admin=True)
self.admin_user_tok = self.login("admin", "pass")
self.register_user("user1", "pass1", admin=False)
self.register_user("user2", "pass2", admin=False)
def test_no_auth(self):
"""
Try to list users without authentication.
"""
request, channel = self.make_request("GET", self.url, b"{}")
self.render(request)
self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("M_MISSING_TOKEN", channel.json_body["errcode"])
def test_all_users(self):
"""
List all users, including deactivated users.
"""
request, channel = self.make_request(
"GET",
self.url + "?deactivated=true",
b"{}",
access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual(3, len(channel.json_body["users"]))
class ShutdownRoomTestCase(unittest.HomeserverTestCase):
servlets = [
synapse.rest.admin.register_servlets_for_client_rest_resource,

465
tests/rest/admin/test_user.py Normal file
View file

@ -0,0 +1,465 @@
# -*- coding: utf-8 -*-
# Copyright 2018 New Vector Ltd
#
# 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.
import hashlib
import hmac
import json
from mock import Mock
import synapse.rest.admin
from synapse.api.constants import UserTypes
from synapse.rest.client.v1 import login
from tests import unittest
class UserRegisterTestCase(unittest.HomeserverTestCase):
servlets = [synapse.rest.admin.register_servlets_for_client_rest_resource]
def make_homeserver(self, reactor, clock):
self.url = "/_matrix/client/r0/admin/register"
self.registration_handler = Mock()
self.identity_handler = Mock()
self.login_handler = Mock()
self.device_handler = Mock()
self.device_handler.check_device_registered = Mock(return_value="FAKE")
self.datastore = Mock(return_value=Mock())
self.datastore.get_current_state_deltas = Mock(return_value=(0, []))
self.secrets = Mock()
self.hs = self.setup_test_homeserver()
self.hs.config.registration_shared_secret = "shared"
self.hs.get_media_repository = Mock()
self.hs.get_deactivate_account_handler = Mock()
return self.hs
def test_disabled(self):
"""
If there is no shared secret, registration through this method will be
prevented.
"""
self.hs.config.registration_shared_secret = None
request, channel = self.make_request("POST", self.url, b"{}")
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual(
"Shared secret registration is not enabled", channel.json_body["error"]
)
def test_get_nonce(self):
"""
Calling GET on the endpoint will return a randomised nonce, using the
homeserver's secrets provider.
"""
secrets = Mock()
secrets.token_hex = Mock(return_value="abcd")
self.hs.get_secrets = Mock(return_value=secrets)
request, channel = self.make_request("GET", self.url)
self.render(request)
self.assertEqual(channel.json_body, {"nonce": "abcd"})
def test_expired_nonce(self):
"""
Calling GET on the endpoint will return a randomised nonce, which will
only last for SALT_TIMEOUT (60s).
"""
request, channel = self.make_request("GET", self.url)
self.render(request)
nonce = channel.json_body["nonce"]
# 59 seconds
self.reactor.advance(59)
body = json.dumps({"nonce": nonce})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("username must be specified", channel.json_body["error"])
# 61 seconds
self.reactor.advance(2)
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("unrecognised nonce", channel.json_body["error"])
def test_register_incorrect_nonce(self):
"""
Only the provided nonce can be used, as it's checked in the MAC.
"""
request, channel = self.make_request("GET", self.url)
self.render(request)
nonce = channel.json_body["nonce"]
want_mac = hmac.new(key=b"shared", digestmod=hashlib.sha1)
want_mac.update(b"notthenonce\x00bob\x00abc123\x00admin")
want_mac = want_mac.hexdigest()
body = json.dumps(
{
"nonce": nonce,
"username": "bob",
"password": "abc123",
"admin": True,
"mac": want_mac,
}
)
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("HMAC incorrect", channel.json_body["error"])
def test_register_correct_nonce(self):
"""
When the correct nonce is provided, and the right key is provided, the
user is registered.
"""
request, channel = self.make_request("GET", self.url)
self.render(request)
nonce = channel.json_body["nonce"]
want_mac = hmac.new(key=b"shared", digestmod=hashlib.sha1)
want_mac.update(
nonce.encode("ascii") + b"\x00bob\x00abc123\x00admin\x00support"
)
want_mac = want_mac.hexdigest()
body = json.dumps(
{
"nonce": nonce,
"username": "bob",
"password": "abc123",
"admin": True,
"user_type": UserTypes.SUPPORT,
"mac": want_mac,
}
)
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("@bob:test", channel.json_body["user_id"])
def test_nonce_reuse(self):
"""
A valid unrecognised nonce.
"""
request, channel = self.make_request("GET", self.url)
self.render(request)
nonce = channel.json_body["nonce"]
want_mac = hmac.new(key=b"shared", digestmod=hashlib.sha1)
want_mac.update(nonce.encode("ascii") + b"\x00bob\x00abc123\x00admin")
want_mac = want_mac.hexdigest()
body = json.dumps(
{
"nonce": nonce,
"username": "bob",
"password": "abc123",
"admin": True,
"mac": want_mac,
}
)
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("@bob:test", channel.json_body["user_id"])
# Now, try and reuse it
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("unrecognised nonce", channel.json_body["error"])
def test_missing_parts(self):
"""
Synapse will complain if you don't give nonce, username, password, and
mac. Admin and user_types are optional. Additional checks are done for length
and type.
"""
def nonce():
request, channel = self.make_request("GET", self.url)
self.render(request)
return channel.json_body["nonce"]
#
# Nonce check
#
# Must be present
body = json.dumps({})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("nonce must be specified", channel.json_body["error"])
#
# Username checks
#
# Must be present
body = json.dumps({"nonce": nonce()})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("username must be specified", channel.json_body["error"])
# Must be a string
body = json.dumps({"nonce": nonce(), "username": 1234})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("Invalid username", channel.json_body["error"])
# Must not have null bytes
body = json.dumps({"nonce": nonce(), "username": "abcd\u0000"})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("Invalid username", channel.json_body["error"])
# Must not have null bytes
body = json.dumps({"nonce": nonce(), "username": "a" * 1000})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("Invalid username", channel.json_body["error"])
#
# Password checks
#
# Must be present
body = json.dumps({"nonce": nonce(), "username": "a"})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("password must be specified", channel.json_body["error"])
# Must be a string
body = json.dumps({"nonce": nonce(), "username": "a", "password": 1234})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("Invalid password", channel.json_body["error"])
# Must not have null bytes
body = json.dumps({"nonce": nonce(), "username": "a", "password": "abcd\u0000"})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("Invalid password", channel.json_body["error"])
# Super long
body = json.dumps({"nonce": nonce(), "username": "a", "password": "A" * 1000})
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("Invalid password", channel.json_body["error"])
#
# user_type check
#
# Invalid user_type
body = json.dumps(
{
"nonce": nonce(),
"username": "a",
"password": "1234",
"user_type": "invalid",
}
)
request, channel = self.make_request("POST", self.url, body.encode("utf8"))
self.render(request)
self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("Invalid user type", channel.json_body["error"])
class UsersListTestCase(unittest.HomeserverTestCase):
servlets = [
synapse.rest.admin.register_servlets,
login.register_servlets,
]
url = "/_synapse/admin/v2/users"
def prepare(self, reactor, clock, hs):
self.admin_user = self.register_user("admin", "pass", admin=True)
self.admin_user_tok = self.login("admin", "pass")
self.register_user("user1", "pass1", admin=False)
self.register_user("user2", "pass2", admin=False)
def test_no_auth(self):
"""
Try to list users without authentication.
"""
request, channel = self.make_request("GET", self.url, b"{}")
self.render(request)
self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("M_MISSING_TOKEN", channel.json_body["errcode"])
def test_all_users(self):
"""
List all users, including deactivated users.
"""
request, channel = self.make_request(
"GET",
self.url + "?deactivated=true",
b"{}",
access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual(3, len(channel.json_body["users"]))
class UserRestTestCase(unittest.HomeserverTestCase):
servlets = [
synapse.rest.admin.register_servlets,
login.register_servlets,
]
def prepare(self, reactor, clock, hs):
self.store = hs.get_datastore()
self.url = "/_synapse/admin/v2/users/@bob:test"
self.admin_user = self.register_user("admin", "pass", admin=True)
self.admin_user_tok = self.login("admin", "pass")
self.other_user = self.register_user("user", "pass")
self.other_user_token = self.login("user", "pass")
def test_requester_is_no_admin(self):
"""
If the user is not a server admin, an error is returned.
"""
self.hs.config.registration_shared_secret = None
request, channel = self.make_request(
"GET", self.url, access_token=self.other_user_token,
)
self.render(request)
self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("You are not a server admin", channel.json_body["error"])
request, channel = self.make_request(
"PUT", self.url, access_token=self.other_user_token, content=b"{}",
)
self.render(request)
self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("You are not a server admin", channel.json_body["error"])
def test_requester_is_admin(self):
"""
If the user is a server admin, a new user is created.
"""
self.hs.config.registration_shared_secret = None
body = json.dumps({"password": "abc123", "admin": True})
# Create user
request, channel = self.make_request(
"PUT",
self.url,
access_token=self.admin_user_tok,
content=body.encode(encoding="utf_8"),
)
self.render(request)
self.assertEqual(201, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("@bob:test", channel.json_body["name"])
self.assertEqual("bob", channel.json_body["displayname"])
# Get user
request, channel = self.make_request(
"GET", self.url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("@bob:test", channel.json_body["name"])
self.assertEqual("bob", channel.json_body["displayname"])
self.assertEqual(1, channel.json_body["admin"])
self.assertEqual(0, channel.json_body["is_guest"])
self.assertEqual(0, channel.json_body["deactivated"])
# Modify user
body = json.dumps({"displayname": "foobar", "deactivated": True})
request, channel = self.make_request(
"PUT",
self.url,
access_token=self.admin_user_tok,
content=body.encode(encoding="utf_8"),
)
self.render(request)
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("@bob:test", channel.json_body["name"])
self.assertEqual("foobar", channel.json_body["displayname"])
self.assertEqual(True, channel.json_body["deactivated"])
# Get user
request, channel = self.make_request(
"GET", self.url, access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
self.assertEqual("@bob:test", channel.json_body["name"])
self.assertEqual("foobar", channel.json_body["displayname"])
self.assertEqual(1, channel.json_body["admin"])
self.assertEqual(0, channel.json_body["is_guest"])
self.assertEqual(1, channel.json_body["deactivated"])

2
tests/storage/test_registration.py
View file

@ -43,12 +43,14 @@ class RegistrationStoreTestCase(unittest.TestCase):
# TODO(paul): Surely this field should be 'user_id', not 'name'
"name": self.user_id,
"password_hash": self.pwhash,
"admin": 0,
"is_guest": 0,
"consent_version": None,
"consent_server_notice_sent": None,
"appservice_id": None,
"creation_ts": 1000,
"user_type": None,
"deactivated": 0,
},
(yield self.store.get_user_by_id(self.user_id)),
)