1
0

test_server.py 6.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188
  1. # Copyright 2019 New Vector Ltd
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. import yaml
  15. from synapse.config._base import ConfigError
  16. from synapse.config.server import ServerConfig, generate_ip_set, is_threepid_reserved
  17. from tests import unittest
  18. class ServerConfigTestCase(unittest.TestCase):
  19. def test_is_threepid_reserved(self):
  20. user1 = {"medium": "email", "address": "user1@example.com"}
  21. user2 = {"medium": "email", "address": "user2@example.com"}
  22. user3 = {"medium": "email", "address": "user3@example.com"}
  23. user1_msisdn = {"medium": "msisdn", "address": "447700000000"}
  24. config = [user1, user2]
  25. self.assertTrue(is_threepid_reserved(config, user1))
  26. self.assertFalse(is_threepid_reserved(config, user3))
  27. self.assertFalse(is_threepid_reserved(config, user1_msisdn))
  28. def test_unsecure_listener_no_listeners_open_private_ports_false(self):
  29. conf = yaml.safe_load(
  30. ServerConfig().generate_config_section(
  31. "che.org", "/data_dir_path", False, None
  32. )
  33. )
  34. expected_listeners = [
  35. {
  36. "port": 8008,
  37. "tls": False,
  38. "type": "http",
  39. "x_forwarded": True,
  40. "bind_addresses": ["::1", "127.0.0.1"],
  41. "resources": [{"names": ["client", "federation"], "compress": False}],
  42. }
  43. ]
  44. self.assertEqual(conf["listeners"], expected_listeners)
  45. def test_unsecure_listener_no_listeners_open_private_ports_true(self):
  46. conf = yaml.safe_load(
  47. ServerConfig().generate_config_section(
  48. "che.org", "/data_dir_path", True, None
  49. )
  50. )
  51. expected_listeners = [
  52. {
  53. "port": 8008,
  54. "tls": False,
  55. "type": "http",
  56. "x_forwarded": True,
  57. "resources": [{"names": ["client", "federation"], "compress": False}],
  58. }
  59. ]
  60. self.assertEqual(conf["listeners"], expected_listeners)
  61. def test_listeners_set_correctly_open_private_ports_false(self):
  62. listeners = [
  63. {
  64. "port": 8448,
  65. "resources": [{"names": ["federation"]}],
  66. "tls": True,
  67. "type": "http",
  68. },
  69. {
  70. "port": 443,
  71. "resources": [{"names": ["client"]}],
  72. "tls": False,
  73. "type": "http",
  74. },
  75. ]
  76. conf = yaml.safe_load(
  77. ServerConfig().generate_config_section(
  78. "this.one.listens", "/data_dir_path", True, listeners
  79. )
  80. )
  81. self.assertEqual(conf["listeners"], listeners)
  82. def test_listeners_set_correctly_open_private_ports_true(self):
  83. listeners = [
  84. {
  85. "port": 8448,
  86. "resources": [{"names": ["federation"]}],
  87. "tls": True,
  88. "type": "http",
  89. },
  90. {
  91. "port": 443,
  92. "resources": [{"names": ["client"]}],
  93. "tls": False,
  94. "type": "http",
  95. },
  96. {
  97. "port": 1243,
  98. "resources": [{"names": ["client"]}],
  99. "tls": False,
  100. "type": "http",
  101. "bind_addresses": ["this_one_is_bound"],
  102. },
  103. ]
  104. expected_listeners = listeners.copy()
  105. expected_listeners[1]["bind_addresses"] = ["::1", "127.0.0.1"]
  106. conf = yaml.safe_load(
  107. ServerConfig().generate_config_section(
  108. "this.one.listens", "/data_dir_path", True, listeners
  109. )
  110. )
  111. self.assertEqual(conf["listeners"], expected_listeners)
  112. class GenerateIpSetTestCase(unittest.TestCase):
  113. def test_empty(self):
  114. ip_set = generate_ip_set(())
  115. self.assertFalse(ip_set)
  116. ip_set = generate_ip_set((), ())
  117. self.assertFalse(ip_set)
  118. def test_generate(self):
  119. """Check adding IPv4 and IPv6 addresses."""
  120. # IPv4 address
  121. ip_set = generate_ip_set(("1.2.3.4",))
  122. self.assertEqual(len(ip_set.iter_cidrs()), 4)
  123. # IPv4 CIDR
  124. ip_set = generate_ip_set(("1.2.3.4/24",))
  125. self.assertEqual(len(ip_set.iter_cidrs()), 4)
  126. # IPv6 address
  127. ip_set = generate_ip_set(("2001:db8::8a2e:370:7334",))
  128. self.assertEqual(len(ip_set.iter_cidrs()), 1)
  129. # IPv6 CIDR
  130. ip_set = generate_ip_set(("2001:db8::/104",))
  131. self.assertEqual(len(ip_set.iter_cidrs()), 1)
  132. # The addresses can overlap OK.
  133. ip_set = generate_ip_set(("1.2.3.4", "::1.2.3.4"))
  134. self.assertEqual(len(ip_set.iter_cidrs()), 4)
  135. def test_extra(self):
  136. """Extra IP addresses are treated the same."""
  137. ip_set = generate_ip_set((), ("1.2.3.4",))
  138. self.assertEqual(len(ip_set.iter_cidrs()), 4)
  139. ip_set = generate_ip_set(("1.1.1.1",), ("1.2.3.4",))
  140. self.assertEqual(len(ip_set.iter_cidrs()), 8)
  141. # They can duplicate without error.
  142. ip_set = generate_ip_set(("1.2.3.4",), ("1.2.3.4",))
  143. self.assertEqual(len(ip_set.iter_cidrs()), 4)
  144. def test_bad_value(self):
  145. """An error should be raised if a bad value is passed in."""
  146. with self.assertRaises(ConfigError):
  147. generate_ip_set(("not-an-ip",))
  148. with self.assertRaises(ConfigError):
  149. generate_ip_set(("1.2.3.4/128",))
  150. with self.assertRaises(ConfigError):
  151. generate_ip_set((":::",))
  152. # The following get treated as empty data.
  153. self.assertFalse(generate_ip_set(None))
  154. self.assertFalse(generate_ip_set({}))