forked-synapse/tests/config/test_server.py

#
# This file is licensed under the Affero General Public License (AGPL) version 3.
#
# Copyright (C) 2023 New Vector, Ltd
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# See the GNU Affero General Public License for more details:
# <https://www.gnu.org/licenses/agpl-3.0.html>.
#
# Originally licensed under the Apache License, Version 2.0:
# <http://www.apache.org/licenses/LICENSE-2.0>.
#
# [This file includes modifications made by New Vector Limited]
#
#

import yaml

from synapse.config._base import ConfigError
from synapse.config.server import ServerConfig, generate_ip_set, is_threepid_reserved

from tests import unittest


class ServerConfigTestCase(unittest.TestCase):
    def test_is_threepid_reserved(self) -> None:
        user1 = {"medium": "email", "address": "user1@example.com"}
        user2 = {"medium": "email", "address": "user2@example.com"}
        user3 = {"medium": "email", "address": "user3@example.com"}
        user1_msisdn = {"medium": "msisdn", "address": "447700000000"}
        config = [user1, user2]

        self.assertTrue(is_threepid_reserved(config, user1))
        self.assertFalse(is_threepid_reserved(config, user3))
        self.assertFalse(is_threepid_reserved(config, user1_msisdn))

    def test_unsecure_listener_no_listeners_open_private_ports_false(self) -> None:
        conf = yaml.safe_load(
            ServerConfig().generate_config_section(
                "CONFDIR", "/data_dir_path", "che.org", False, None
            )
        )

        expected_listeners = [
            {
                "port": 8008,
                "tls": False,
                "type": "http",
                "x_forwarded": True,
                "bind_addresses": ["::1", "127.0.0.1"],
                "resources": [{"names": ["client", "federation"], "compress": False}],
            }
        ]

        self.assertEqual(conf["listeners"], expected_listeners)

    def test_unsecure_listener_no_listeners_open_private_ports_true(self) -> None:
        conf = yaml.safe_load(
            ServerConfig().generate_config_section(
                "CONFDIR", "/data_dir_path", "che.org", True, None
            )
        )

        expected_listeners = [
            {
                "port": 8008,
                "tls": False,
                "type": "http",
                "x_forwarded": True,
                "resources": [{"names": ["client", "federation"], "compress": False}],
            }
        ]

        self.assertEqual(conf["listeners"], expected_listeners)

    def test_listeners_set_correctly_open_private_ports_false(self) -> None:
        listeners = [
            {
                "port": 8448,
                "resources": [{"names": ["federation"]}],
                "tls": True,
                "type": "http",
            },
            {
                "port": 443,
                "resources": [{"names": ["client"]}],
                "tls": False,
                "type": "http",
            },
        ]

        conf = yaml.safe_load(
            ServerConfig().generate_config_section(
                "CONFDIR", "/data_dir_path", "this.one.listens", True, listeners
            )
        )

        self.assertEqual(conf["listeners"], listeners)

    def test_listeners_set_correctly_open_private_ports_true(self) -> None:
        listeners = [
            {
                "port": 8448,
                "resources": [{"names": ["federation"]}],
                "tls": True,
                "type": "http",
            },
            {
                "port": 443,
                "resources": [{"names": ["client"]}],
                "tls": False,
                "type": "http",
            },
            {
                "port": 1243,
                "resources": [{"names": ["client"]}],
                "tls": False,
                "type": "http",
                "bind_addresses": ["this_one_is_bound"],
            },
        ]

        expected_listeners = listeners.copy()
        expected_listeners[1]["bind_addresses"] = ["::1", "127.0.0.1"]

        conf = yaml.safe_load(
            ServerConfig().generate_config_section(
                "CONFDIR", "/data_dir_path", "this.one.listens", True, listeners
            )
        )

        self.assertEqual(conf["listeners"], expected_listeners)


class GenerateIpSetTestCase(unittest.TestCase):
    def test_empty(self) -> None:
        ip_set = generate_ip_set(())
        self.assertFalse(ip_set)

        ip_set = generate_ip_set((), ())
        self.assertFalse(ip_set)

    def test_generate(self) -> None:
        """Check adding IPv4 and IPv6 addresses."""
        # IPv4 address
        ip_set = generate_ip_set(("1.2.3.4",))
        self.assertEqual(len(ip_set.iter_cidrs()), 4)

        # IPv4 CIDR
        ip_set = generate_ip_set(("1.2.3.4/24",))
        self.assertEqual(len(ip_set.iter_cidrs()), 4)

        # IPv6 address
        ip_set = generate_ip_set(("2001:db8::8a2e:370:7334",))
        self.assertEqual(len(ip_set.iter_cidrs()), 1)

        # IPv6 CIDR
        ip_set = generate_ip_set(("2001:db8::/104",))
        self.assertEqual(len(ip_set.iter_cidrs()), 1)

        # The addresses can overlap OK.
        ip_set = generate_ip_set(("1.2.3.4", "::1.2.3.4"))
        self.assertEqual(len(ip_set.iter_cidrs()), 4)

    def test_extra(self) -> None:
        """Extra IP addresses are treated the same."""
        ip_set = generate_ip_set((), ("1.2.3.4",))
        self.assertEqual(len(ip_set.iter_cidrs()), 4)

        ip_set = generate_ip_set(("1.1.1.1",), ("1.2.3.4",))
        self.assertEqual(len(ip_set.iter_cidrs()), 8)

        # They can duplicate without error.
        ip_set = generate_ip_set(("1.2.3.4",), ("1.2.3.4",))
        self.assertEqual(len(ip_set.iter_cidrs()), 4)

    def test_bad_value(self) -> None:
        """An error should be raised if a bad value is passed in."""
        with self.assertRaises(ConfigError):
            generate_ip_set(("not-an-ip",))

        with self.assertRaises(ConfigError):
            generate_ip_set(("1.2.3.4/128",))

        with self.assertRaises(ConfigError):
            generate_ip_set((":::",))

        # The following get treated as empty data.
        self.assertFalse(generate_ip_set(None))
        self.assertFalse(generate_ip_set({}))