123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321 |
- # -*- 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"])
|