Mirrors
/
synapse
mirror of https://github.com/element-hq/synapse.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix userconsent on Python 3 (#3938)

This commit is contained in:
Amber Brown 2018-10-02 00:11:58 +10:00 committed by GitHub
parent 8f5c23d0cd
commit 6e05fd032c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 240 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
changelog.d/3938.bugfix Normal file
View File

@ -0,0 +1 @@
Sending server notices regarding user consent now works on Python 3.

2
synapse/api/urls.py
View File

@ -64,7 +64,7 @@ class ConsentURIBuilder(object):
"""
mac = hmac.new(
key=self._hmac_secret,
msg=user_id,
msg=user_id.encode('ascii'),
digestmod=sha256,
).hexdigest()
consent_uri = "%s_matrix/consent?%s" % (

100
tests/server_notices/test_consent.py Normal file
View File

@ -0,0 +1,100 @@
# -*- 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.
from synapse.rest.client.v1 import admin, login, room
from synapse.rest.client.v2_alpha import sync
from tests import unittest
class ConsentNoticesTests(unittest.HomeserverTestCase):
servlets = [
sync.register_servlets,
admin.register_servlets,
login.register_servlets,
room.register_servlets,
]
def make_homeserver(self, reactor, clock):
self.consent_notice_message = "consent %(consent_uri)s"
config = self.default_config()
config.user_consent_version = "1"
config.user_consent_server_notice_content = {
"msgtype": "m.text",
"body": self.consent_notice_message,
}
config.public_baseurl = "https://example.com/"
config.form_secret = "123abc"
config.server_notices_mxid = "@notices:test"
config.server_notices_mxid_display_name = "test display name"
config.server_notices_mxid_avatar_url = None
config.server_notices_room_name = "Server Notices"
hs = self.setup_test_homeserver(config=config)
return hs
def prepare(self, reactor, clock, hs):
self.user_id = self.register_user("bob", "abc123")
self.access_token = self.login("bob", "abc123")
def test_get_sync_message(self):
"""
When user consent server notices are enabled, a sync will cause a notice
to fire (in a room which the user is invited to). The notice contains
the notice URL + an authentication code.
"""
# Initial sync, to get the user consent room invite
request, channel = self.make_request(
"GET", "/_matrix/client/r0/sync", access_token=self.access_token
)
self.render(request)
self.assertEqual(channel.code, 200)
# Get the Room ID to join
room_id = list(channel.json_body["rooms"]["invite"].keys())[0]
# Join the room
request, channel = self.make_request(
"POST",
"/_matrix/client/r0/rooms/" + room_id + "/join",
access_token=self.access_token,
)
self.render(request)
self.assertEqual(channel.code, 200)
# Sync again, to get the message in the room
request, channel = self.make_request(
"GET", "/_matrix/client/r0/sync", access_token=self.access_token
)
self.render(request)
self.assertEqual(channel.code, 200)
# Get the message
room = channel.json_body["rooms"]["join"][room_id]
messages = [
x for x in room["timeline"]["events"] if x["type"] == "m.room.message"
]
# One message, with the consent URL
self.assertEqual(len(messages), 1)
self.assertTrue(
messages[0]["content"]["body"].startswith(
"consent https://example.com/_matrix/consent"
)
)

48
tests/storage/test_client_ips.py
View File

@ -14,10 +14,6 @@
# 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 twisted.internet import defer
@ -145,34 +141,8 @@ class ClientIpAuthTestCase(unittest.HomeserverTestCase):
return hs
def prepare(self, hs, reactor, clock):
self.hs.config.registration_shared_secret = u"shared"
self.store = self.hs.get_datastore()
# Create the user
request, channel = self.make_request("GET", "/_matrix/client/r0/admin/register")
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", "/_matrix/client/r0/admin/register", body.encode('utf8')
)
self.render(request)
self.assertEqual(channel.code, 200)
self.user_id = channel.json_body["user_id"]
self.user_id = self.register_user("bob", "abc123", True)
def test_request_with_xforwarded(self):
"""
@ -194,20 +164,7 @@ class ClientIpAuthTestCase(unittest.HomeserverTestCase):
def _runtest(self, headers, expected_ip, make_request_args):
device_id = "bleb"
body = json.dumps(
{
"type": "m.login.password",
"user": "bob",
"password": "abc123",
"device_id": device_id,
}
)
request, channel = self.make_request(
"POST", "/_matrix/client/r0/login", body.encode('utf8'), **make_request_args
)
self.render(request)
self.assertEqual(channel.code, 200)
access_token = channel.json_body["access_token"].encode('ascii')
access_token = self.login("bob", "abc123", device_id=device_id)
# Advance to a known time
self.reactor.advance(123456 - self.reactor.seconds())
@ -215,7 +172,6 @@ class ClientIpAuthTestCase(unittest.HomeserverTestCase):
request, channel = self.make_request(
"GET",
"/_matrix/client/r0/admin/users/" + self.user_id,
body.encode('utf8'),
access_token=access_token,
**make_request_args
)

80
tests/unittest.py
View File

@ -14,6 +14,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import hashlib
import hmac
import logging
from mock import Mock
@ -32,6 +34,7 @@ from synapse.types import UserID, create_requester
from synapse.util.logcontext import LoggingContextFilter
from tests.server import get_clock, make_request, render, setup_test_homeserver
from tests.utils import default_config
# Set up putting Synapse's logs into Trial's.
rootLogger = logging.getLogger()
@ -121,7 +124,7 @@ class TestCase(unittest.TestCase):
try:
self.assertEquals(attrs[key], getattr(obj, key))
except AssertionError as e:
raise (type(e))(str(e) + " for '.%s'" % key)
raise (type(e))(e.message + " for '.%s'" % key)
def assert_dict(self, required, actual):
"""Does a partial assert of a dict.
@ -223,6 +226,15 @@ class HomeserverTestCase(TestCase):
hs = self.setup_test_homeserver()
return hs
def default_config(self, name="test"):
"""
Get a default HomeServer config object.
Args:
name (str): The homeserver name/domain.
"""
return default_config(name)
def prepare(self, reactor, clock, homeserver):
"""
Prepare for the test. This involves things like mocking out parts of
@ -297,3 +309,69 @@ class HomeserverTestCase(TestCase):
return d
self.pump()
return self.successResultOf(d)
def register_user(self, username, password, admin=False):
"""
Register a user. Requires the Admin API be registered.
Args:
username (bytes/unicode): The user part of the new user.
password (bytes/unicode): The password of the new user.
admin (bool): Whether the user should be created as an admin
or not.
Returns:
The MXID of the new user (unicode).
"""
self.hs.config.registration_shared_secret = u"shared"
# Create the user
request, channel = self.make_request("GET", "/_matrix/client/r0/admin/register")
self.render(request)
nonce = channel.json_body["nonce"]
want_mac = hmac.new(key=b"shared", digestmod=hashlib.sha1)
nonce_str = b"\x00".join([username.encode('utf8'), password.encode('utf8')])
if admin:
nonce_str += b"\x00admin"
else:
nonce_str += b"\x00notadmin"
want_mac.update(nonce.encode('ascii') + b"\x00" + nonce_str)
want_mac = want_mac.hexdigest()
body = json.dumps(
{
"nonce": nonce,
"username": username,
"password": password,
"admin": admin,
"mac": want_mac,
}
)
request, channel = self.make_request(
"POST", "/_matrix/client/r0/admin/register", body.encode('utf8')
)
self.render(request)
self.assertEqual(channel.code, 200)
user_id = channel.json_body["user_id"]
return user_id
def login(self, username, password, device_id=None):
"""
Log in a user, and get an access token. Requires the Login API be
registered.
"""
body = {"type": "m.login.password", "user": username, "password": password}
if device_id:
body["device_id"] = device_id
request, channel = self.make_request(
"POST", "/_matrix/client/r0/login", json.dumps(body).encode('utf8')
)
self.render(request)
self.assertEqual(channel.code, 200)
access_token = channel.json_body["access_token"].encode('ascii')
return access_token

61
tests/utils.py
View File

@ -96,34 +96,10 @@ def setupdb():
atexit.register(_cleanup)
class TestHomeServer(HomeServer):
DATASTORE_CLASS = DataStore
@defer.inlineCallbacks
def setup_test_homeserver(
cleanup_func,
name="test",
datastore=None,
config=None,
reactor=None,
homeserverToUse=TestHomeServer,
**kargs
):
def default_config(name):
"""
Setup a homeserver suitable for running tests against. Keyword arguments
are passed to the Homeserver constructor.
If no datastore is supplied, one is created and given to the homeserver.
Args:
cleanup_func : The function used to register a cleanup routine for
after the test.
Create a reasonable test config.
"""
if reactor is None:
from twisted.internet import reactor
if config is None:
config = Mock()
config.signing_key = [MockKey()]
config.event_cache_size = 1
@ -173,6 +149,39 @@ def setup_test_homeserver(
config.is_threepid_reserved.side_effect = is_threepid_reserved
return config
class TestHomeServer(HomeServer):
DATASTORE_CLASS = DataStore
@defer.inlineCallbacks
def setup_test_homeserver(
cleanup_func,
name="test",
datastore=None,
config=None,
reactor=None,
homeserverToUse=TestHomeServer,
**kargs
):
"""
Setup a homeserver suitable for running tests against. Keyword arguments
are passed to the Homeserver constructor.
If no datastore is supplied, one is created and given to the homeserver.
Args:
cleanup_func : The function used to register a cleanup routine for
after the test.
"""
if reactor is None:
from twisted.internet import reactor
if config is None:
config = default_config(name)
config.use_frozen_dicts = True
config.ldap_enabled = False