mirror of
https://mau.dev/maunium/synapse.git
synced 2024-12-15 17:53:51 +01:00
d2f7c4e6b1
Allow for the creation of a support user. A support user can access the server, join rooms, interact with other users, but does not appear in the user directory nor does it contribute to monthly active user limits.
321 lines
11 KiB
Python
321 lines
11 KiB
Python
# -*- 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
|
|
|
|
from synapse.api.constants import UserTypes
|
|
from synapse.rest.client.v1.admin import register_servlets
|
|
|
|
from tests import unittest
|
|
|
|
|
|
class UserRegisterTestCase(unittest.HomeserverTestCase):
|
|
|
|
servlets = [register_servlets]
|
|
|
|
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=[])
|
|
|
|
self.secrets = Mock()
|
|
|
|
self.hs = self.setup_test_homeserver()
|
|
|
|
self.hs.config.registration_shared_secret = u"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": u"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": u"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"])
|