12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811 |
- # -*- 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
- import urllib.parse
- from binascii import unhexlify
- from mock import Mock
- import synapse.rest.admin
- from synapse.api.constants import UserTypes
- from synapse.api.errors import Codes, HttpResponseException, ResourceLimitError
- from synapse.rest.client.v1 import login, logout, profile, room
- from synapse.rest.client.v2_alpha import devices, sync
- from tests import unittest
- from tests.test_utils import make_awaitable
- from tests.unittest import override_config
- class UserRegisterTestCase(unittest.HomeserverTestCase):
- servlets = [
- synapse.rest.admin.register_servlets_for_client_rest_resource,
- profile.register_servlets,
- ]
- def make_homeserver(self, reactor, clock):
- self.url = "/_synapse/admin/v1/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
- channel = self.make_request("POST", self.url, b"{}")
- 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)
- channel = self.make_request("GET", self.url)
- 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).
- """
- channel = self.make_request("GET", self.url)
- nonce = channel.json_body["nonce"]
- # 59 seconds
- self.reactor.advance(59)
- body = json.dumps({"nonce": nonce})
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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)
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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.
- """
- channel = self.make_request("GET", self.url)
- 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,
- }
- )
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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.
- """
- channel = self.make_request("GET", self.url)
- 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,
- }
- )
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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.
- """
- channel = self.make_request("GET", self.url)
- 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,
- }
- )
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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():
- channel = self.make_request("GET", self.url)
- return channel.json_body["nonce"]
- #
- # Nonce check
- #
- # Must be present
- body = json.dumps({})
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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()})
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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})
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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"})
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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})
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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"})
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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})
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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"})
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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})
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- 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",
- }
- )
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("Invalid user type", channel.json_body["error"])
- def test_displayname(self):
- """
- Test that displayname of new user is set
- """
- # set no displayname
- channel = self.make_request("GET", self.url)
- nonce = channel.json_body["nonce"]
- want_mac = hmac.new(key=b"shared", digestmod=hashlib.sha1)
- want_mac.update(nonce.encode("ascii") + b"\x00bob1\x00abc123\x00notadmin")
- want_mac = want_mac.hexdigest()
- body = json.dumps(
- {"nonce": nonce, "username": "bob1", "password": "abc123", "mac": want_mac}
- )
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@bob1:test", channel.json_body["user_id"])
- channel = self.make_request("GET", "/profile/@bob1:test/displayname")
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("bob1", channel.json_body["displayname"])
- # displayname is None
- channel = self.make_request("GET", self.url)
- nonce = channel.json_body["nonce"]
- want_mac = hmac.new(key=b"shared", digestmod=hashlib.sha1)
- want_mac.update(nonce.encode("ascii") + b"\x00bob2\x00abc123\x00notadmin")
- want_mac = want_mac.hexdigest()
- body = json.dumps(
- {
- "nonce": nonce,
- "username": "bob2",
- "displayname": None,
- "password": "abc123",
- "mac": want_mac,
- }
- )
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@bob2:test", channel.json_body["user_id"])
- channel = self.make_request("GET", "/profile/@bob2:test/displayname")
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("bob2", channel.json_body["displayname"])
- # displayname is empty
- channel = self.make_request("GET", self.url)
- nonce = channel.json_body["nonce"]
- want_mac = hmac.new(key=b"shared", digestmod=hashlib.sha1)
- want_mac.update(nonce.encode("ascii") + b"\x00bob3\x00abc123\x00notadmin")
- want_mac = want_mac.hexdigest()
- body = json.dumps(
- {
- "nonce": nonce,
- "username": "bob3",
- "displayname": "",
- "password": "abc123",
- "mac": want_mac,
- }
- )
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@bob3:test", channel.json_body["user_id"])
- channel = self.make_request("GET", "/profile/@bob3:test/displayname")
- self.assertEqual(404, int(channel.result["code"]), msg=channel.result["body"])
- # set displayname
- channel = self.make_request("GET", self.url)
- nonce = channel.json_body["nonce"]
- want_mac = hmac.new(key=b"shared", digestmod=hashlib.sha1)
- want_mac.update(nonce.encode("ascii") + b"\x00bob4\x00abc123\x00notadmin")
- want_mac = want_mac.hexdigest()
- body = json.dumps(
- {
- "nonce": nonce,
- "username": "bob4",
- "displayname": "Bob's Name",
- "password": "abc123",
- "mac": want_mac,
- }
- )
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@bob4:test", channel.json_body["user_id"])
- channel = self.make_request("GET", "/profile/@bob4:test/displayname")
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("Bob's Name", channel.json_body["displayname"])
- @override_config(
- {"limit_usage_by_mau": True, "max_mau_value": 2, "mau_trial_days": 0}
- )
- def test_register_mau_limit_reached(self):
- """
- Check we can register a user via the shared secret registration API
- even if the MAU limit is reached.
- """
- handler = self.hs.get_registration_handler()
- store = self.hs.get_datastore()
- # Set monthly active users to the limit
- store.get_monthly_active_count = Mock(
- return_value=make_awaitable(self.hs.config.max_mau_value)
- )
- # Check that the blocking of monthly active users is working as expected
- # The registration of a new user fails due to the limit
- self.get_failure(
- handler.register_user(localpart="local_part"), ResourceLimitError
- )
- # Register new user with admin API
- channel = self.make_request("GET", self.url)
- 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,
- }
- )
- channel = self.make_request("POST", self.url, body.encode("utf8"))
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@bob:test", channel.json_body["user_id"])
- 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.
- """
- channel = self.make_request("GET", self.url, b"{}")
- 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.
- """
- channel = self.make_request(
- "GET",
- self.url + "?deactivated=true",
- b"{}",
- access_token=self.admin_user_tok,
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(3, len(channel.json_body["users"]))
- self.assertEqual(3, channel.json_body["total"])
- class UserRestTestCase(unittest.HomeserverTestCase):
- servlets = [
- synapse.rest.admin.register_servlets,
- login.register_servlets,
- sync.register_servlets,
- ]
- def prepare(self, reactor, clock, hs):
- self.store = hs.get_datastore()
- 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")
- self.url_other_user = "/_synapse/admin/v2/users/%s" % urllib.parse.quote(
- self.other_user
- )
- def test_requester_is_no_admin(self):
- """
- If the user is not a server admin, an error is returned.
- """
- url = "/_synapse/admin/v2/users/@bob:test"
- channel = self.make_request("GET", url, access_token=self.other_user_token,)
- self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("You are not a server admin", channel.json_body["error"])
- channel = self.make_request(
- "PUT", url, access_token=self.other_user_token, content=b"{}",
- )
- 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_user_does_not_exist(self):
- """
- Tests that a lookup for a user that does not exist returns a 404
- """
- channel = self.make_request(
- "GET",
- "/_synapse/admin/v2/users/@unknown_person:test",
- access_token=self.admin_user_tok,
- )
- self.assertEqual(404, channel.code, msg=channel.json_body)
- self.assertEqual("M_NOT_FOUND", channel.json_body["errcode"])
- def test_create_server_admin(self):
- """
- Check that a new admin user is created successfully.
- """
- url = "/_synapse/admin/v2/users/@bob:test"
- # Create user (server admin)
- body = json.dumps(
- {
- "password": "abc123",
- "admin": True,
- "displayname": "Bob's name",
- "threepids": [{"medium": "email", "address": "bob@bob.bob"}],
- "avatar_url": "mxc://fibble/wibble",
- }
- )
- channel = self.make_request(
- "PUT",
- url,
- access_token=self.admin_user_tok,
- content=body.encode(encoding="utf_8"),
- )
- self.assertEqual(201, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@bob:test", channel.json_body["name"])
- self.assertEqual("Bob's name", channel.json_body["displayname"])
- self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
- self.assertEqual("bob@bob.bob", channel.json_body["threepids"][0]["address"])
- self.assertEqual(True, channel.json_body["admin"])
- self.assertEqual("mxc://fibble/wibble", channel.json_body["avatar_url"])
- # Get user
- channel = self.make_request("GET", url, access_token=self.admin_user_tok,)
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@bob:test", channel.json_body["name"])
- self.assertEqual("Bob's name", channel.json_body["displayname"])
- self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
- self.assertEqual("bob@bob.bob", channel.json_body["threepids"][0]["address"])
- self.assertEqual(True, channel.json_body["admin"])
- self.assertEqual(False, channel.json_body["is_guest"])
- self.assertEqual(False, channel.json_body["deactivated"])
- self.assertEqual("mxc://fibble/wibble", channel.json_body["avatar_url"])
- def test_create_user(self):
- """
- Check that a new regular user is created successfully.
- """
- url = "/_synapse/admin/v2/users/@bob:test"
- # Create user
- body = json.dumps(
- {
- "password": "abc123",
- "admin": False,
- "displayname": "Bob's name",
- "threepids": [{"medium": "email", "address": "bob@bob.bob"}],
- "avatar_url": "mxc://fibble/wibble",
- }
- )
- channel = self.make_request(
- "PUT",
- url,
- access_token=self.admin_user_tok,
- content=body.encode(encoding="utf_8"),
- )
- self.assertEqual(201, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@bob:test", channel.json_body["name"])
- self.assertEqual("Bob's name", channel.json_body["displayname"])
- self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
- self.assertEqual("bob@bob.bob", channel.json_body["threepids"][0]["address"])
- self.assertEqual(False, channel.json_body["admin"])
- self.assertEqual("mxc://fibble/wibble", channel.json_body["avatar_url"])
- # Get user
- channel = self.make_request("GET", url, access_token=self.admin_user_tok,)
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@bob:test", channel.json_body["name"])
- self.assertEqual("Bob's name", channel.json_body["displayname"])
- self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
- self.assertEqual("bob@bob.bob", channel.json_body["threepids"][0]["address"])
- self.assertEqual(False, channel.json_body["admin"])
- self.assertEqual(False, channel.json_body["is_guest"])
- self.assertEqual(False, channel.json_body["deactivated"])
- self.assertEqual("mxc://fibble/wibble", channel.json_body["avatar_url"])
- @override_config(
- {"limit_usage_by_mau": True, "max_mau_value": 2, "mau_trial_days": 0}
- )
- def test_create_user_mau_limit_reached_active_admin(self):
- """
- Check that an admin can register a new user via the admin API
- even if the MAU limit is reached.
- Admin user was active before creating user.
- """
- handler = self.hs.get_registration_handler()
- # Sync to set admin user to active
- # before limit of monthly active users is reached
- channel = self.make_request("GET", "/sync", access_token=self.admin_user_tok)
- if channel.code != 200:
- raise HttpResponseException(
- channel.code, channel.result["reason"], channel.result["body"]
- )
- # Set monthly active users to the limit
- self.store.get_monthly_active_count = Mock(
- return_value=make_awaitable(self.hs.config.max_mau_value)
- )
- # Check that the blocking of monthly active users is working as expected
- # The registration of a new user fails due to the limit
- self.get_failure(
- handler.register_user(localpart="local_part"), ResourceLimitError
- )
- # Register new user with admin API
- url = "/_synapse/admin/v2/users/@bob:test"
- # Create user
- body = json.dumps({"password": "abc123", "admin": False})
- channel = self.make_request(
- "PUT",
- url,
- access_token=self.admin_user_tok,
- content=body.encode(encoding="utf_8"),
- )
- self.assertEqual(201, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@bob:test", channel.json_body["name"])
- self.assertEqual(False, channel.json_body["admin"])
- @override_config(
- {"limit_usage_by_mau": True, "max_mau_value": 2, "mau_trial_days": 0}
- )
- def test_create_user_mau_limit_reached_passive_admin(self):
- """
- Check that an admin can register a new user via the admin API
- even if the MAU limit is reached.
- Admin user was not active before creating user.
- """
- handler = self.hs.get_registration_handler()
- # Set monthly active users to the limit
- self.store.get_monthly_active_count = Mock(
- return_value=make_awaitable(self.hs.config.max_mau_value)
- )
- # Check that the blocking of monthly active users is working as expected
- # The registration of a new user fails due to the limit
- self.get_failure(
- handler.register_user(localpart="local_part"), ResourceLimitError
- )
- # Register new user with admin API
- url = "/_synapse/admin/v2/users/@bob:test"
- # Create user
- body = json.dumps({"password": "abc123", "admin": False})
- channel = self.make_request(
- "PUT",
- url,
- access_token=self.admin_user_tok,
- content=body.encode(encoding="utf_8"),
- )
- # Admin user is not blocked by mau anymore
- self.assertEqual(201, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@bob:test", channel.json_body["name"])
- self.assertEqual(False, channel.json_body["admin"])
- @override_config(
- {
- "email": {
- "enable_notifs": True,
- "notif_for_new_users": True,
- "notif_from": "test@example.com",
- },
- "public_baseurl": "https://example.com",
- }
- )
- def test_create_user_email_notif_for_new_users(self):
- """
- Check that a new regular user is created successfully and
- got an email pusher.
- """
- url = "/_synapse/admin/v2/users/@bob:test"
- # Create user
- body = json.dumps(
- {
- "password": "abc123",
- "threepids": [{"medium": "email", "address": "bob@bob.bob"}],
- }
- )
- channel = self.make_request(
- "PUT",
- url,
- access_token=self.admin_user_tok,
- content=body.encode(encoding="utf_8"),
- )
- self.assertEqual(201, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@bob:test", channel.json_body["name"])
- self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
- self.assertEqual("bob@bob.bob", channel.json_body["threepids"][0]["address"])
- pushers = self.get_success(
- self.store.get_pushers_by({"user_name": "@bob:test"})
- )
- pushers = list(pushers)
- self.assertEqual(len(pushers), 1)
- self.assertEqual("@bob:test", pushers[0].user_name)
- @override_config(
- {
- "email": {
- "enable_notifs": False,
- "notif_for_new_users": False,
- "notif_from": "test@example.com",
- },
- "public_baseurl": "https://example.com",
- }
- )
- def test_create_user_email_no_notif_for_new_users(self):
- """
- Check that a new regular user is created successfully and
- got not an email pusher.
- """
- url = "/_synapse/admin/v2/users/@bob:test"
- # Create user
- body = json.dumps(
- {
- "password": "abc123",
- "threepids": [{"medium": "email", "address": "bob@bob.bob"}],
- }
- )
- channel = self.make_request(
- "PUT",
- url,
- access_token=self.admin_user_tok,
- content=body.encode(encoding="utf_8"),
- )
- self.assertEqual(201, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@bob:test", channel.json_body["name"])
- self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
- self.assertEqual("bob@bob.bob", channel.json_body["threepids"][0]["address"])
- pushers = self.get_success(
- self.store.get_pushers_by({"user_name": "@bob:test"})
- )
- pushers = list(pushers)
- self.assertEqual(len(pushers), 0)
- def test_set_password(self):
- """
- Test setting a new password for another user.
- """
- # Change password
- body = json.dumps({"password": "hahaha"})
- channel = self.make_request(
- "PUT",
- self.url_other_user,
- access_token=self.admin_user_tok,
- content=body.encode(encoding="utf_8"),
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- def test_set_displayname(self):
- """
- Test setting the displayname of another user.
- """
- # Modify user
- body = json.dumps({"displayname": "foobar"})
- channel = self.make_request(
- "PUT",
- self.url_other_user,
- access_token=self.admin_user_tok,
- content=body.encode(encoding="utf_8"),
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@user:test", channel.json_body["name"])
- self.assertEqual("foobar", channel.json_body["displayname"])
- # Get user
- channel = self.make_request(
- "GET", self.url_other_user, access_token=self.admin_user_tok,
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@user:test", channel.json_body["name"])
- self.assertEqual("foobar", channel.json_body["displayname"])
- def test_set_threepid(self):
- """
- Test setting threepid for an other user.
- """
- # Delete old and add new threepid to user
- body = json.dumps(
- {"threepids": [{"medium": "email", "address": "bob3@bob.bob"}]}
- )
- channel = self.make_request(
- "PUT",
- self.url_other_user,
- access_token=self.admin_user_tok,
- content=body.encode(encoding="utf_8"),
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@user:test", channel.json_body["name"])
- self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
- self.assertEqual("bob3@bob.bob", channel.json_body["threepids"][0]["address"])
- # Get user
- channel = self.make_request(
- "GET", self.url_other_user, access_token=self.admin_user_tok,
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@user:test", channel.json_body["name"])
- self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
- self.assertEqual("bob3@bob.bob", channel.json_body["threepids"][0]["address"])
- def test_deactivate_user(self):
- """
- Test deactivating another user.
- """
- # Deactivate user
- body = json.dumps({"deactivated": True})
- channel = self.make_request(
- "PUT",
- self.url_other_user,
- access_token=self.admin_user_tok,
- content=body.encode(encoding="utf_8"),
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@user:test", channel.json_body["name"])
- self.assertEqual(True, channel.json_body["deactivated"])
- # the user is deactivated, the threepid will be deleted
- # Get user
- channel = self.make_request(
- "GET", self.url_other_user, access_token=self.admin_user_tok,
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@user:test", channel.json_body["name"])
- self.assertEqual(True, channel.json_body["deactivated"])
- def test_reactivate_user(self):
- """
- Test reactivating another user.
- """
- # Deactivate the user.
- channel = self.make_request(
- "PUT",
- self.url_other_user,
- access_token=self.admin_user_tok,
- content=json.dumps({"deactivated": True}).encode(encoding="utf_8"),
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self._is_erased("@user:test", False)
- d = self.store.mark_user_erased("@user:test")
- self.assertIsNone(self.get_success(d))
- self._is_erased("@user:test", True)
- # Attempt to reactivate the user (without a password).
- channel = self.make_request(
- "PUT",
- self.url_other_user,
- access_token=self.admin_user_tok,
- content=json.dumps({"deactivated": False}).encode(encoding="utf_8"),
- )
- self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
- # Reactivate the user.
- channel = self.make_request(
- "PUT",
- self.url_other_user,
- access_token=self.admin_user_tok,
- content=json.dumps({"deactivated": False, "password": "foo"}).encode(
- encoding="utf_8"
- ),
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- # Get user
- channel = self.make_request(
- "GET", self.url_other_user, access_token=self.admin_user_tok,
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@user:test", channel.json_body["name"])
- self.assertEqual(False, channel.json_body["deactivated"])
- self._is_erased("@user:test", False)
- def test_set_user_as_admin(self):
- """
- Test setting the admin flag on a user.
- """
- # Set a user as an admin
- body = json.dumps({"admin": True})
- channel = self.make_request(
- "PUT",
- self.url_other_user,
- access_token=self.admin_user_tok,
- content=body.encode(encoding="utf_8"),
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@user:test", channel.json_body["name"])
- self.assertEqual(True, channel.json_body["admin"])
- # Get user
- channel = self.make_request(
- "GET", self.url_other_user, access_token=self.admin_user_tok,
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual("@user:test", channel.json_body["name"])
- self.assertEqual(True, channel.json_body["admin"])
- def test_accidental_deactivation_prevention(self):
- """
- Ensure an account can't accidentally be deactivated by using a str value
- for the deactivated body parameter
- """
- url = "/_synapse/admin/v2/users/@bob:test"
- # Create user
- body = json.dumps({"password": "abc123"})
- channel = self.make_request(
- "PUT",
- url,
- access_token=self.admin_user_tok,
- content=body.encode(encoding="utf_8"),
- )
- 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
- channel = self.make_request("GET", url, access_token=self.admin_user_tok,)
- 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(0, channel.json_body["deactivated"])
- # Change password (and use a str for deactivate instead of a bool)
- body = json.dumps({"password": "abc123", "deactivated": "false"}) # oops!
- channel = self.make_request(
- "PUT",
- url,
- access_token=self.admin_user_tok,
- content=body.encode(encoding="utf_8"),
- )
- self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
- # Check user is not deactivated
- channel = self.make_request("GET", url, access_token=self.admin_user_tok,)
- 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"])
- # Ensure they're still alive
- self.assertEqual(0, channel.json_body["deactivated"])
- def _is_erased(self, user_id, expect):
- """Assert that the user is erased or not
- """
- d = self.store.is_user_erased(user_id)
- if expect:
- self.assertTrue(self.get_success(d))
- else:
- self.assertFalse(self.get_success(d))
- class UserMembershipRestTestCase(unittest.HomeserverTestCase):
- servlets = [
- synapse.rest.admin.register_servlets,
- login.register_servlets,
- room.register_servlets,
- ]
- def prepare(self, reactor, clock, hs):
- self.store = hs.get_datastore()
- 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.url = "/_synapse/admin/v1/users/%s/joined_rooms" % urllib.parse.quote(
- self.other_user
- )
- def test_no_auth(self):
- """
- Try to list rooms of an user without authentication.
- """
- channel = self.make_request("GET", self.url, b"{}")
- self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
- def test_requester_is_no_admin(self):
- """
- If the user is not a server admin, an error is returned.
- """
- other_user_token = self.login("user", "pass")
- channel = self.make_request("GET", self.url, access_token=other_user_token,)
- self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
- def test_user_does_not_exist(self):
- """
- Tests that a lookup for a user that does not exist returns a 404
- """
- url = "/_synapse/admin/v1/users/@unknown_person:test/joined_rooms"
- channel = self.make_request("GET", url, access_token=self.admin_user_tok,)
- self.assertEqual(404, channel.code, msg=channel.json_body)
- self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
- def test_user_is_not_local(self):
- """
- Tests that a lookup for a user that is not a local returns a 400
- """
- url = "/_synapse/admin/v1/users/@unknown_person:unknown_domain/joined_rooms"
- channel = self.make_request("GET", url, access_token=self.admin_user_tok,)
- self.assertEqual(400, channel.code, msg=channel.json_body)
- self.assertEqual("Can only lookup local users", channel.json_body["error"])
- def test_no_memberships(self):
- """
- Tests that a normal lookup for rooms is successfully
- if user has no memberships
- """
- # Get rooms
- channel = self.make_request("GET", self.url, access_token=self.admin_user_tok,)
- self.assertEqual(200, channel.code, msg=channel.json_body)
- self.assertEqual(0, channel.json_body["total"])
- self.assertEqual(0, len(channel.json_body["joined_rooms"]))
- def test_get_rooms(self):
- """
- Tests that a normal lookup for rooms is successfully
- """
- # Create rooms and join
- other_user_tok = self.login("user", "pass")
- number_rooms = 5
- for n in range(number_rooms):
- self.helper.create_room_as(self.other_user, tok=other_user_tok)
- # Get rooms
- channel = self.make_request("GET", self.url, access_token=self.admin_user_tok,)
- self.assertEqual(200, channel.code, msg=channel.json_body)
- self.assertEqual(number_rooms, channel.json_body["total"])
- self.assertEqual(number_rooms, len(channel.json_body["joined_rooms"]))
- class PushersRestTestCase(unittest.HomeserverTestCase):
- servlets = [
- synapse.rest.admin.register_servlets,
- login.register_servlets,
- ]
- def prepare(self, reactor, clock, hs):
- self.store = hs.get_datastore()
- 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.url = "/_synapse/admin/v1/users/%s/pushers" % urllib.parse.quote(
- self.other_user
- )
- def test_no_auth(self):
- """
- Try to list pushers of an user without authentication.
- """
- channel = self.make_request("GET", self.url, b"{}")
- self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
- def test_requester_is_no_admin(self):
- """
- If the user is not a server admin, an error is returned.
- """
- other_user_token = self.login("user", "pass")
- channel = self.make_request("GET", self.url, access_token=other_user_token,)
- self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
- def test_user_does_not_exist(self):
- """
- Tests that a lookup for a user that does not exist returns a 404
- """
- url = "/_synapse/admin/v1/users/@unknown_person:test/pushers"
- channel = self.make_request("GET", url, access_token=self.admin_user_tok,)
- self.assertEqual(404, channel.code, msg=channel.json_body)
- self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
- def test_user_is_not_local(self):
- """
- Tests that a lookup for a user that is not a local returns a 400
- """
- url = "/_synapse/admin/v1/users/@unknown_person:unknown_domain/pushers"
- channel = self.make_request("GET", url, access_token=self.admin_user_tok,)
- self.assertEqual(400, channel.code, msg=channel.json_body)
- self.assertEqual("Can only lookup local users", channel.json_body["error"])
- def test_get_pushers(self):
- """
- Tests that a normal lookup for pushers is successfully
- """
- # Get pushers
- channel = self.make_request("GET", self.url, access_token=self.admin_user_tok,)
- self.assertEqual(200, channel.code, msg=channel.json_body)
- self.assertEqual(0, channel.json_body["total"])
- # Register the pusher
- other_user_token = self.login("user", "pass")
- user_tuple = self.get_success(
- self.store.get_user_by_access_token(other_user_token)
- )
- token_id = user_tuple.token_id
- self.get_success(
- self.hs.get_pusherpool().add_pusher(
- user_id=self.other_user,
- access_token=token_id,
- kind="http",
- app_id="m.http",
- app_display_name="HTTP Push Notifications",
- device_display_name="pushy push",
- pushkey="a@example.com",
- lang=None,
- data={"url": "https://example.com/_matrix/push/v1/notify"},
- )
- )
- # Get pushers
- channel = self.make_request("GET", self.url, access_token=self.admin_user_tok,)
- self.assertEqual(200, channel.code, msg=channel.json_body)
- self.assertEqual(1, channel.json_body["total"])
- for p in channel.json_body["pushers"]:
- self.assertIn("pushkey", p)
- self.assertIn("kind", p)
- self.assertIn("app_id", p)
- self.assertIn("app_display_name", p)
- self.assertIn("device_display_name", p)
- self.assertIn("profile_tag", p)
- self.assertIn("lang", p)
- self.assertIn("url", p["data"])
- class UserMediaRestTestCase(unittest.HomeserverTestCase):
- servlets = [
- synapse.rest.admin.register_servlets,
- login.register_servlets,
- ]
- def prepare(self, reactor, clock, hs):
- self.store = hs.get_datastore()
- self.media_repo = hs.get_media_repository_resource()
- 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.url = "/_synapse/admin/v1/users/%s/media" % urllib.parse.quote(
- self.other_user
- )
- def test_no_auth(self):
- """
- Try to list media of an user without authentication.
- """
- channel = self.make_request("GET", self.url, b"{}")
- self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
- def test_requester_is_no_admin(self):
- """
- If the user is not a server admin, an error is returned.
- """
- other_user_token = self.login("user", "pass")
- channel = self.make_request("GET", self.url, access_token=other_user_token,)
- self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
- def test_user_does_not_exist(self):
- """
- Tests that a lookup for a user that does not exist returns a 404
- """
- url = "/_synapse/admin/v1/users/@unknown_person:test/media"
- channel = self.make_request("GET", url, access_token=self.admin_user_tok,)
- self.assertEqual(404, channel.code, msg=channel.json_body)
- self.assertEqual(Codes.NOT_FOUND, channel.json_body["errcode"])
- def test_user_is_not_local(self):
- """
- Tests that a lookup for a user that is not a local returns a 400
- """
- url = "/_synapse/admin/v1/users/@unknown_person:unknown_domain/media"
- channel = self.make_request("GET", url, access_token=self.admin_user_tok,)
- self.assertEqual(400, channel.code, msg=channel.json_body)
- self.assertEqual("Can only lookup local users", channel.json_body["error"])
- def test_limit(self):
- """
- Testing list of media with limit
- """
- number_media = 20
- other_user_tok = self.login("user", "pass")
- self._create_media(other_user_tok, number_media)
- channel = self.make_request(
- "GET", self.url + "?limit=5", access_token=self.admin_user_tok,
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(channel.json_body["total"], number_media)
- self.assertEqual(len(channel.json_body["media"]), 5)
- self.assertEqual(channel.json_body["next_token"], 5)
- self._check_fields(channel.json_body["media"])
- def test_from(self):
- """
- Testing list of media with a defined starting point (from)
- """
- number_media = 20
- other_user_tok = self.login("user", "pass")
- self._create_media(other_user_tok, number_media)
- channel = self.make_request(
- "GET", self.url + "?from=5", access_token=self.admin_user_tok,
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(channel.json_body["total"], number_media)
- self.assertEqual(len(channel.json_body["media"]), 15)
- self.assertNotIn("next_token", channel.json_body)
- self._check_fields(channel.json_body["media"])
- def test_limit_and_from(self):
- """
- Testing list of media with a defined starting point and limit
- """
- number_media = 20
- other_user_tok = self.login("user", "pass")
- self._create_media(other_user_tok, number_media)
- channel = self.make_request(
- "GET", self.url + "?from=5&limit=10", access_token=self.admin_user_tok,
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(channel.json_body["total"], number_media)
- self.assertEqual(channel.json_body["next_token"], 15)
- self.assertEqual(len(channel.json_body["media"]), 10)
- self._check_fields(channel.json_body["media"])
- def test_limit_is_negative(self):
- """
- Testing that a negative limit parameter returns a 400
- """
- channel = self.make_request(
- "GET", self.url + "?limit=-5", access_token=self.admin_user_tok,
- )
- self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(Codes.INVALID_PARAM, channel.json_body["errcode"])
- def test_from_is_negative(self):
- """
- Testing that a negative from parameter returns a 400
- """
- channel = self.make_request(
- "GET", self.url + "?from=-5", access_token=self.admin_user_tok,
- )
- self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(Codes.INVALID_PARAM, channel.json_body["errcode"])
- def test_next_token(self):
- """
- Testing that `next_token` appears at the right place
- """
- number_media = 20
- other_user_tok = self.login("user", "pass")
- self._create_media(other_user_tok, number_media)
- # `next_token` does not appear
- # Number of results is the number of entries
- channel = self.make_request(
- "GET", self.url + "?limit=20", access_token=self.admin_user_tok,
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(channel.json_body["total"], number_media)
- self.assertEqual(len(channel.json_body["media"]), number_media)
- self.assertNotIn("next_token", channel.json_body)
- # `next_token` does not appear
- # Number of max results is larger than the number of entries
- channel = self.make_request(
- "GET", self.url + "?limit=21", access_token=self.admin_user_tok,
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(channel.json_body["total"], number_media)
- self.assertEqual(len(channel.json_body["media"]), number_media)
- self.assertNotIn("next_token", channel.json_body)
- # `next_token` does appear
- # Number of max results is smaller than the number of entries
- channel = self.make_request(
- "GET", self.url + "?limit=19", access_token=self.admin_user_tok,
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(channel.json_body["total"], number_media)
- self.assertEqual(len(channel.json_body["media"]), 19)
- self.assertEqual(channel.json_body["next_token"], 19)
- # Check
- # Set `from` to value of `next_token` for request remaining entries
- # `next_token` does not appear
- channel = self.make_request(
- "GET", self.url + "?from=19", access_token=self.admin_user_tok,
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(channel.json_body["total"], number_media)
- self.assertEqual(len(channel.json_body["media"]), 1)
- self.assertNotIn("next_token", channel.json_body)
- def test_user_has_no_media(self):
- """
- Tests that a normal lookup for media is successfully
- if user has no media created
- """
- channel = self.make_request("GET", self.url, access_token=self.admin_user_tok,)
- self.assertEqual(200, channel.code, msg=channel.json_body)
- self.assertEqual(0, channel.json_body["total"])
- self.assertEqual(0, len(channel.json_body["media"]))
- def test_get_media(self):
- """
- Tests that a normal lookup for media is successfully
- """
- number_media = 5
- other_user_tok = self.login("user", "pass")
- self._create_media(other_user_tok, number_media)
- channel = self.make_request("GET", self.url, access_token=self.admin_user_tok,)
- self.assertEqual(200, channel.code, msg=channel.json_body)
- self.assertEqual(number_media, channel.json_body["total"])
- self.assertEqual(number_media, len(channel.json_body["media"]))
- self.assertNotIn("next_token", channel.json_body)
- self._check_fields(channel.json_body["media"])
- def _create_media(self, user_token, number_media):
- """
- Create a number of media for a specific user
- """
- upload_resource = self.media_repo.children[b"upload"]
- for i in range(number_media):
- # file size is 67 Byte
- image_data = unhexlify(
- b"89504e470d0a1a0a0000000d4948445200000001000000010806"
- b"0000001f15c4890000000a49444154789c63000100000500010d"
- b"0a2db40000000049454e44ae426082"
- )
- # Upload some media into the room
- self.helper.upload_media(
- upload_resource, image_data, tok=user_token, expect_code=200
- )
- def _check_fields(self, content):
- """Checks that all attributes are present in content
- """
- for m in content:
- self.assertIn("media_id", m)
- self.assertIn("media_type", m)
- self.assertIn("media_length", m)
- self.assertIn("upload_name", m)
- self.assertIn("created_ts", m)
- self.assertIn("last_access_ts", m)
- self.assertIn("quarantined_by", m)
- self.assertIn("safe_from_quarantine", m)
- class UserTokenRestTestCase(unittest.HomeserverTestCase):
- """Test for /_synapse/admin/v1/users/<user>/login
- """
- servlets = [
- synapse.rest.admin.register_servlets,
- login.register_servlets,
- sync.register_servlets,
- room.register_servlets,
- devices.register_servlets,
- logout.register_servlets,
- ]
- def prepare(self, reactor, clock, hs):
- self.store = hs.get_datastore()
- 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_tok = self.login("user", "pass")
- self.url = "/_synapse/admin/v1/users/%s/login" % urllib.parse.quote(
- self.other_user
- )
- def _get_token(self) -> str:
- channel = self.make_request(
- "POST", self.url, b"{}", access_token=self.admin_user_tok
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- return channel.json_body["access_token"]
- def test_no_auth(self):
- """Try to login as a user without authentication.
- """
- channel = self.make_request("POST", self.url, b"{}")
- self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
- def test_not_admin(self):
- """Try to login as a user as a non-admin user.
- """
- channel = self.make_request(
- "POST", self.url, b"{}", access_token=self.other_user_tok
- )
- self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
- def test_send_event(self):
- """Test that sending event as a user works.
- """
- # Create a room.
- room_id = self.helper.create_room_as(self.other_user, tok=self.other_user_tok)
- # Login in as the user
- puppet_token = self._get_token()
- # Test that sending works, and generates the event as the right user.
- resp = self.helper.send_event(room_id, "com.example.test", tok=puppet_token)
- event_id = resp["event_id"]
- event = self.get_success(self.store.get_event(event_id))
- self.assertEqual(event.sender, self.other_user)
- def test_devices(self):
- """Tests that logging in as a user doesn't create a new device for them.
- """
- # Login in as the user
- self._get_token()
- # Check that we don't see a new device in our devices list
- channel = self.make_request(
- "GET", "devices", b"{}", access_token=self.other_user_tok
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- # We should only see the one device (from the login in `prepare`)
- self.assertEqual(len(channel.json_body["devices"]), 1)
- def test_logout(self):
- """Test that calling `/logout` with the token works.
- """
- # Login in as the user
- puppet_token = self._get_token()
- # Test that we can successfully make a request
- channel = self.make_request("GET", "devices", b"{}", access_token=puppet_token)
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- # Logout with the puppet token
- channel = self.make_request("POST", "logout", b"{}", access_token=puppet_token)
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- # The puppet token should no longer work
- channel = self.make_request("GET", "devices", b"{}", access_token=puppet_token)
- self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
- # .. but the real user's tokens should still work
- channel = self.make_request(
- "GET", "devices", b"{}", access_token=self.other_user_tok
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- def test_user_logout_all(self):
- """Tests that the target user calling `/logout/all` does *not* expire
- the token.
- """
- # Login in as the user
- puppet_token = self._get_token()
- # Test that we can successfully make a request
- channel = self.make_request("GET", "devices", b"{}", access_token=puppet_token)
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- # Logout all with the real user token
- channel = self.make_request(
- "POST", "logout/all", b"{}", access_token=self.other_user_tok
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- # The puppet token should still work
- channel = self.make_request("GET", "devices", b"{}", access_token=puppet_token)
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- # .. but the real user's tokens shouldn't
- channel = self.make_request(
- "GET", "devices", b"{}", access_token=self.other_user_tok
- )
- self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
- def test_admin_logout_all(self):
- """Tests that the admin user calling `/logout/all` does expire the
- token.
- """
- # Login in as the user
- puppet_token = self._get_token()
- # Test that we can successfully make a request
- channel = self.make_request("GET", "devices", b"{}", access_token=puppet_token)
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- # Logout all with the admin user token
- channel = self.make_request(
- "POST", "logout/all", b"{}", access_token=self.admin_user_tok
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- # The puppet token should no longer work
- channel = self.make_request("GET", "devices", b"{}", access_token=puppet_token)
- self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
- # .. but the real user's tokens should still work
- channel = self.make_request(
- "GET", "devices", b"{}", access_token=self.other_user_tok
- )
- self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
- @unittest.override_config(
- {
- "public_baseurl": "https://example.org/",
- "user_consent": {
- "version": "1.0",
- "policy_name": "My Cool Privacy Policy",
- "template_dir": "/",
- "require_at_registration": True,
- "block_events_error": "You should accept the policy",
- },
- "form_secret": "123secret",
- }
- )
- def test_consent(self):
- """Test that sending a message is not subject to the privacy policies.
- """
- # Have the admin user accept the terms.
- self.get_success(self.store.user_set_consent_version(self.admin_user, "1.0"))
- # First, cheekily accept the terms and create a room
- self.get_success(self.store.user_set_consent_version(self.other_user, "1.0"))
- room_id = self.helper.create_room_as(self.other_user, tok=self.other_user_tok)
- self.helper.send_event(room_id, "com.example.test", tok=self.other_user_tok)
- # Now unaccept it and check that we can't send an event
- self.get_success(self.store.user_set_consent_version(self.other_user, "0.0"))
- self.helper.send_event(
- room_id, "com.example.test", tok=self.other_user_tok, expect_code=403
- )
- # Login in as the user
- puppet_token = self._get_token()
- # Sending an event on their behalf should work fine
- self.helper.send_event(room_id, "com.example.test", tok=puppet_token)
- @override_config(
- {"limit_usage_by_mau": True, "max_mau_value": 1, "mau_trial_days": 0}
- )
- def test_mau_limit(self):
- # Create a room as the admin user. This will bump the monthly active users to 1.
- room_id = self.helper.create_room_as(self.admin_user, tok=self.admin_user_tok)
- # Trying to join as the other user should fail due to reaching MAU limit.
- self.helper.join(
- room_id, user=self.other_user, tok=self.other_user_tok, expect_code=403
- )
- # Logging in as the other user and joining a room should work, even
- # though the MAU limit would stop the user doing so.
- puppet_token = self._get_token()
- self.helper.join(room_id, user=self.other_user, tok=puppet_token)
- class WhoisRestTestCase(unittest.HomeserverTestCase):
- servlets = [
- synapse.rest.admin.register_servlets,
- login.register_servlets,
- ]
- def prepare(self, reactor, clock, hs):
- self.store = hs.get_datastore()
- 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.url1 = "/_synapse/admin/v1/whois/%s" % urllib.parse.quote(self.other_user)
- self.url2 = "/_matrix/client/r0/admin/whois/%s" % urllib.parse.quote(
- self.other_user
- )
- def test_no_auth(self):
- """
- Try to get information of an user without authentication.
- """
- channel = self.make_request("GET", self.url1, b"{}")
- self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
- channel = self.make_request("GET", self.url2, b"{}")
- self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
- def test_requester_is_not_admin(self):
- """
- If the user is not a server admin, an error is returned.
- """
- self.register_user("user2", "pass")
- other_user2_token = self.login("user2", "pass")
- channel = self.make_request("GET", self.url1, access_token=other_user2_token,)
- self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
- channel = self.make_request("GET", self.url2, access_token=other_user2_token,)
- self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"])
- self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
- def test_user_is_not_local(self):
- """
- Tests that a lookup for a user that is not a local returns a 400
- """
- url1 = "/_synapse/admin/v1/whois/@unknown_person:unknown_domain"
- url2 = "/_matrix/client/r0/admin/whois/@unknown_person:unknown_domain"
- channel = self.make_request("GET", url1, access_token=self.admin_user_tok,)
- self.assertEqual(400, channel.code, msg=channel.json_body)
- self.assertEqual("Can only whois a local user", channel.json_body["error"])
- channel = self.make_request("GET", url2, access_token=self.admin_user_tok,)
- self.assertEqual(400, channel.code, msg=channel.json_body)
- self.assertEqual("Can only whois a local user", channel.json_body["error"])
- def test_get_whois_admin(self):
- """
- The lookup should succeed for an admin.
- """
- channel = self.make_request("GET", self.url1, access_token=self.admin_user_tok,)
- self.assertEqual(200, channel.code, msg=channel.json_body)
- self.assertEqual(self.other_user, channel.json_body["user_id"])
- self.assertIn("devices", channel.json_body)
- channel = self.make_request("GET", self.url2, access_token=self.admin_user_tok,)
- self.assertEqual(200, channel.code, msg=channel.json_body)
- self.assertEqual(self.other_user, channel.json_body["user_id"])
- self.assertIn("devices", channel.json_body)
- def test_get_whois_user(self):
- """
- The lookup should succeed for a normal user looking up their own information.
- """
- other_user_token = self.login("user", "pass")
- channel = self.make_request("GET", self.url1, access_token=other_user_token,)
- self.assertEqual(200, channel.code, msg=channel.json_body)
- self.assertEqual(self.other_user, channel.json_body["user_id"])
- self.assertIn("devices", channel.json_body)
- channel = self.make_request("GET", self.url2, access_token=other_user_token,)
- self.assertEqual(200, channel.code, msg=channel.json_body)
- self.assertEqual(self.other_user, channel.json_body["user_id"])
- self.assertIn("devices", channel.json_body)
|