1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
# -*- coding: utf-8 -*-
# Copyright 2019 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 mock import Mock
from twisted.internet import defer
import synapse.rest.admin
from synapse.rest.client.v1 import login
from synapse.rest.client.v2_alpha import account
from tests import unittest
class ThreepidISRewrittenURLTestCase(unittest.HomeserverTestCase):
servlets = [
synapse.rest.admin.register_servlets_for_client_rest_resource,
login.register_servlets,
account.register_servlets,
]
def make_homeserver(self, reactor, clock):
self.address = "test@test"
self.is_server_name = "testis"
self.is_server_url = "https://testis"
self.rewritten_is_url = "https://int.testis"
config = self.default_config()
config["trusted_third_party_id_servers"] = [self.is_server_name]
config["rewrite_identity_server_urls"] = {
self.is_server_url: self.rewritten_is_url
}
mock_http_client = Mock(spec=["get_json", "post_json_get_json"])
mock_http_client.get_json.side_effect = defer.succeed({})
mock_http_client.post_json_get_json.return_value = defer.succeed(
{"address": self.address, "medium": "email"}
)
self.hs = self.setup_test_homeserver(
config=config, simple_http_client=mock_http_client
)
mock_blacklisting_http_client = Mock(spec=["get_json", "post_json_get_json"])
mock_blacklisting_http_client.get_json.side_effect = defer.succeed({})
mock_blacklisting_http_client.post_json_get_json.return_value = defer.succeed(
{"address": self.address, "medium": "email"}
)
# TODO: This class does not use a singleton to get it's http client
# This should be fixed for easier testing
# https://github.com/matrix-org/synapse-dinsic/issues/26
self.hs.get_handlers().identity_handler.blacklisting_http_client = (
mock_blacklisting_http_client
)
return self.hs
def prepare(self, reactor, clock, hs):
self.user_id = self.register_user("kermit", "monkey")
def test_rewritten_id_server(self):
"""
Tests that, when validating a 3PID association while rewriting the IS's server
name:
* the bind request is done against the rewritten hostname
* the original, non-rewritten, server name is stored in the database
"""
handler = self.hs.get_handlers().identity_handler
post_json_get_json = handler.blacklisting_http_client.post_json_get_json
store = self.hs.get_datastore()
creds = {"sid": "123", "client_secret": "some_secret"}
# Make sure processing the mocked response goes through.
data = self.get_success(
handler.bind_threepid(
client_secret=creds["client_secret"],
sid=creds["sid"],
mxid=self.user_id,
id_server=self.is_server_name,
use_v2=False,
)
)
self.assertEqual(data.get("address"), self.address)
# Check that the request was done against the rewritten server name.
post_json_get_json.assert_called_once_with(
"%s/_matrix/identity/api/v1/3pid/bind" % (self.rewritten_is_url,),
{
"sid": creds["sid"],
"client_secret": creds["client_secret"],
"mxid": self.user_id,
},
headers={},
)
# Check that the original server name is saved in the database instead of the
# rewritten one.
id_servers = self.get_success(
store.get_id_servers_user_bound(self.user_id, "email", self.address)
)
self.assertEqual(id_servers, [self.is_server_name])
|
