# -*- coding: utf-8 -*- # Import Python libs from __future__ import absolute_import, unicode_literals, print_function import logging import socket import textwrap # Import Salt Testing libs from tests.support.unit import skipIf from tests.support.unit import TestCase from tests.support.mock import ( MagicMock, mock_open, patch, NO_MOCK, NO_MOCK_REASON, ) # Import salt libs import salt.utils.network as network log = logging.getLogger(__name__) LINUX = '''\ eth0 Link encap:Ethernet HWaddr e0:3f:49:85:6a:af inet addr:10.10.10.56 Bcast:10.10.10.255 Mask:255.255.252.0 inet6 addr: fe80::e23f:49ff:fe85:6aaf/64 Scope:Link UP BROADCAST RUNNING MULTICAST MTU:1500 Metric:1 RX packets:643363 errors:0 dropped:0 overruns:0 frame:0 TX packets:196539 errors:0 dropped:0 overruns:0 carrier:0 collisions:0 txqueuelen:1000 RX bytes:386388355 (368.4 MiB) TX bytes:25600939 (24.4 MiB) lo Link encap:Local Loopback inet addr:127.0.0.1 Mask:255.0.0.0 inet6 addr: ::1/128 Scope:Host UP LOOPBACK RUNNING MTU:65536 Metric:1 RX packets:548901 errors:0 dropped:0 overruns:0 frame:0 TX packets:548901 errors:0 dropped:0 overruns:0 carrier:0 collisions:0 txqueuelen:0 RX bytes:613479895 (585.0 MiB) TX bytes:613479895 (585.0 MiB) ''' FREEBSD = ''' em0: flags=8843 metric 0 mtu 1500 options=4219b ether 00:30:48:ff:ff:ff inet 10.10.10.250 netmask 0xffffffe0 broadcast 10.10.10.255 inet 10.10.10.56 netmask 0xffffffc0 broadcast 10.10.10.63 media: Ethernet autoselect (1000baseT ) status: active em1: flags=8c02 metric 0 mtu 1500 options=4219b ether 00:30:48:aa:aa:aa media: Ethernet autoselect status: no carrier plip0: flags=8810 metric 0 mtu 1500 lo0: flags=8049 metric 0 mtu 16384 options=3 inet6 fe80::1%lo0 prefixlen 64 scopeid 0x8 inet6 ::1 prefixlen 128 inet 127.0.0.1 netmask 0xff000000 nd6 options=3 tun0: flags=8051 metric 0 mtu 1500 options=80000 inet 10.12.0.1 --> 10.12.0.2 netmask 0xffffffff Opened by PID 1964 ''' SOLARIS = '''\ lo0: flags=2001000849 mtu 8232 index 1 inet 127.0.0.1 netmask ff000000 net0: flags=100001100943 mtu 1500 index 2 inet 10.10.10.38 netmask ffffffe0 broadcast 10.10.10.63 ilbint0: flags=110001100843 mtu 1500 index 3 inet 10.6.0.11 netmask ffffff00 broadcast 10.6.0.255 ilbext0: flags=110001100843 mtu 1500 index 4 inet 10.10.11.11 netmask ffffffe0 broadcast 10.10.11.31 ilbext0:1: flags=110001100843 mtu 1500 index 4 inet 10.10.11.12 netmask ffffffe0 broadcast 10.10.11.31 vpn0: flags=1000011008d1 mtu 1480 index 5 inet tunnel src 10.10.11.12 tunnel dst 10.10.5.5 tunnel hop limit 64 inet 10.6.0.14 --> 10.6.0.15 netmask ff000000 lo0: flags=2002000849 mtu 8252 index 1 inet6 ::1/128 net0: flags=120002004941 mtu 1500 index 2 inet6 fe80::221:9bff:fefd:2a22/10 ilbint0: flags=120002000840 mtu 1500 index 3 inet6 ::/0 ilbext0: flags=120002000840 mtu 1500 index 4 inet6 ::/0 vpn0: flags=120002200850 mtu 1480 index 5 inet tunnel src 10.10.11.12 tunnel dst 10.10.5.5 tunnel hop limit 64 inet6 ::/0 --> fe80::b2d6:7c10 ''' FREEBSD_SOCKSTAT = '''\ USER COMMAND PID FD PROTO LOCAL ADDRESS FOREIGN ADDRESS root python2.7 1294 41 tcp4 127.0.0.1:61115 127.0.0.1:4506 ''' FREEBSD_SOCKSTAT_WITH_FAT_PID = '''\ USER COMMAND PID FD PROTO LOCAL ADDRESS FOREIGN ADDRESS salt-master python2.781106 35 tcp4 127.0.0.1:61115 127.0.0.1:4506 ''' IPV4_SUBNETS = {True: ('10.10.0.0/24',), False: ('10.10.0.0', '10.10.0.0/33', 'FOO', 9, '0.9.800.1000/24')} IPV6_SUBNETS = {True: ('::1/128',), False: ('::1', '::1/129', 'FOO', 9, 'aj01::feac/64')} @skipIf(NO_MOCK, NO_MOCK_REASON) class NetworkTestCase(TestCase): def test_sanitize_host(self): ret = network.sanitize_host('10.1./2.$3') self.assertEqual(ret, '10.1.2.3') def test_host_to_ips(self): ''' NOTE: When this test fails it's usually because the IP address has changed. In these cases, we just need to update the IP address in the assertion. ''' def _side_effect(host, *args): try: return { 'github.com': [ (2, 1, 6, '', ('192.30.255.112', 0)), (2, 1, 6, '', ('192.30.255.113', 0)), ], 'ipv6host.foo': [ (socket.AF_INET6, 1, 6, '', ('2001:a71::1', 0, 0, 0)), ], }[host] except KeyError: raise socket.gaierror(-2, 'Name or service not known') getaddrinfo_mock = MagicMock(side_effect=_side_effect) with patch.object(socket, 'getaddrinfo', getaddrinfo_mock): # Test host that can be resolved ret = network.host_to_ips('github.com') self.assertEqual(ret, ['192.30.255.112', '192.30.255.113']) # Test ipv6 ret = network.host_to_ips('ipv6host.foo') self.assertEqual(ret, ['2001:a71::1']) # Test host that can't be resolved ret = network.host_to_ips('someothersite.com') self.assertEqual(ret, None) def test_generate_minion_id(self): self.assertTrue(network.generate_minion_id()) def test__generate_minion_id_with_unicode_in_etc_hosts(self): ''' Test that unicode in /etc/hosts doesn't raise an error when _generate_minion_id() helper is called to gather the hosts. ''' content = textwrap.dedent('''\ # 以下为主机名解析 ## ccc 127.0.0.1 localhost thisismyhostname # 本机 ''') fopen_mock = mock_open(read_data={'/etc/hosts': content}) with patch('salt.utils.files.fopen', fopen_mock): assert 'thisismyhostname' in network._generate_minion_id() def test_is_ip(self): self.assertTrue(network.is_ip('10.10.0.3')) self.assertFalse(network.is_ip('0.9.800.1000')) def test_is_ipv4(self): self.assertTrue(network.is_ipv4('10.10.0.3')) self.assertFalse(network.is_ipv4('10.100.1')) self.assertFalse(network.is_ipv4('2001:db8:0:1:1:1:1:1')) def test_is_ipv6(self): self.assertTrue(network.is_ipv6('2001:db8:0:1:1:1:1:1')) self.assertTrue(network.is_ipv6('0:0:0:0:0:0:0:1')) self.assertTrue(network.is_ipv6('::1')) self.assertTrue(network.is_ipv6('::')) self.assertTrue(network.is_ipv6('2001:0db8:85a3:0000:0000:8a2e:0370:7334')) self.assertTrue(network.is_ipv6('2001:0db8:85a3::8a2e:0370:7334')) self.assertFalse(network.is_ipv6('2001:0db8:0370:7334')) self.assertFalse(network.is_ipv6('2001:0db8:::0370:7334')) self.assertFalse(network.is_ipv6('10.0.1.2')) self.assertFalse(network.is_ipv6('2001.0db8.85a3.0000.0000.8a2e.0370.7334')) def test_is_subnet(self): for subnet_data in (IPV4_SUBNETS, IPV6_SUBNETS): for item in subnet_data[True]: log.debug('Testing that %s is a valid subnet', item) self.assertTrue(network.is_subnet(item)) for item in subnet_data[False]: log.debug('Testing that %s is not a valid subnet', item) self.assertFalse(network.is_subnet(item)) def test_is_ipv4_subnet(self): for item in IPV4_SUBNETS[True]: log.debug('Testing that %s is a valid subnet', item) self.assertTrue(network.is_ipv4_subnet(item)) for item in IPV4_SUBNETS[False]: log.debug('Testing that %s is not a valid subnet', item) self.assertFalse(network.is_ipv4_subnet(item)) def test_is_ipv6_subnet(self): for item in IPV6_SUBNETS[True]: log.debug('Testing that %s is a valid subnet', item) self.assertTrue(network.is_ipv6_subnet(item)) for item in IPV6_SUBNETS[False]: log.debug('Testing that %s is not a valid subnet', item) self.assertFalse(network.is_ipv6_subnet(item)) def test_cidr_to_ipv4_netmask(self): self.assertEqual(network.cidr_to_ipv4_netmask(24), '255.255.255.0') self.assertEqual(network.cidr_to_ipv4_netmask(21), '255.255.248.0') self.assertEqual(network.cidr_to_ipv4_netmask(17), '255.255.128.0') self.assertEqual(network.cidr_to_ipv4_netmask(9), '255.128.0.0') self.assertEqual(network.cidr_to_ipv4_netmask(36), '') self.assertEqual(network.cidr_to_ipv4_netmask('lol'), '') def test_number_of_set_bits_to_ipv4_netmask(self): set_bits_to_netmask = network._number_of_set_bits_to_ipv4_netmask(0xffffff00) self.assertEqual(set_bits_to_netmask, '255.255.255.0') set_bits_to_netmask = network._number_of_set_bits_to_ipv4_netmask(0xffff6400) def test_hex2ip(self): self.assertEqual(network.hex2ip('0x4A7D2B63'), '74.125.43.99') self.assertEqual(network.hex2ip('0x4A7D2B63', invert=True), '99.43.125.74') def test_interfaces_ifconfig_linux(self): interfaces = network._interfaces_ifconfig(LINUX) self.assertEqual(interfaces, {'eth0': {'hwaddr': 'e0:3f:49:85:6a:af', 'inet': [{'address': '10.10.10.56', 'broadcast': '10.10.10.255', 'netmask': '255.255.252.0'}], 'inet6': [{'address': 'fe80::e23f:49ff:fe85:6aaf', 'prefixlen': '64', 'scope': 'link'}], 'up': True}, 'lo': {'inet': [{'address': '127.0.0.1', 'netmask': '255.0.0.0'}], 'inet6': [{'address': '::1', 'prefixlen': '128', 'scope': 'host'}], 'up': True}} ) def test_interfaces_ifconfig_freebsd(self): interfaces = network._interfaces_ifconfig(FREEBSD) self.assertEqual(interfaces, {'': {'up': False}, 'em0': {'hwaddr': '00:30:48:ff:ff:ff', 'inet': [{'address': '10.10.10.250', 'broadcast': '10.10.10.255', 'netmask': '255.255.255.224'}, {'address': '10.10.10.56', 'broadcast': '10.10.10.63', 'netmask': '255.255.255.192'}], 'up': True}, 'em1': {'hwaddr': '00:30:48:aa:aa:aa', 'up': False}, 'lo0': {'inet': [{'address': '127.0.0.1', 'netmask': '255.0.0.0'}], 'inet6': [{'address': 'fe80::1', 'prefixlen': '64', 'scope': '0x8'}, {'address': '::1', 'prefixlen': '128', 'scope': None}], 'up': True}, 'plip0': {'up': False}, 'tun0': {'inet': [{'address': '10.12.0.1', 'netmask': '255.255.255.255'}], 'up': True}} ) def test_interfaces_ifconfig_solaris(self): with patch('salt.utils.platform.is_sunos', lambda: True): interfaces = network._interfaces_ifconfig(SOLARIS) expected_interfaces = {'ilbint0': {'inet6': [], 'inet': [{'broadcast': '10.6.0.255', 'netmask': '255.255.255.0', 'address': '10.6.0.11'}], 'up': True}, 'lo0': {'inet6': [{'prefixlen': '128', 'address': '::1'}], 'inet': [{'netmask': '255.0.0.0', 'address': '127.0.0.1'}], 'up': True}, 'ilbext0': {'inet6': [], 'inet': [{'broadcast': '10.10.11.31', 'netmask': '255.255.255.224', 'address': '10.10.11.11'}, {'broadcast': '10.10.11.31', 'netmask': '255.255.255.224', 'address': '10.10.11.12'}], 'up': True}, 'vpn0': {'inet6': [], 'inet': [{'netmask': '255.0.0.0', 'address': '10.6.0.14'}], 'up': True}, 'net0': {'inet6': [{'prefixlen': '10', 'address': 'fe80::221:9bff:fefd:2a22'}], 'inet': [{'broadcast': '10.10.10.63', 'netmask': '255.255.255.224', 'address': '10.10.10.38'}], 'up': True}} self.assertEqual(interfaces, expected_interfaces) def test_freebsd_remotes_on(self): with patch('salt.utils.platform.is_sunos', lambda: False): with patch('salt.utils.platform.is_freebsd', lambda: True): with patch('subprocess.check_output', return_value=FREEBSD_SOCKSTAT): remotes = network._freebsd_remotes_on('4506', 'remote') self.assertEqual(remotes, set(['127.0.0.1'])) def test_freebsd_remotes_on_with_fat_pid(self): with patch('salt.utils.platform.is_sunos', lambda: False): with patch('salt.utils.platform.is_freebsd', lambda: True): with patch('subprocess.check_output', return_value=FREEBSD_SOCKSTAT_WITH_FAT_PID): remotes = network._freebsd_remotes_on('4506', 'remote') self.assertEqual(remotes, set(['127.0.0.1'])) def test_generate_minion_id_distinct(self): ''' Test if minion IDs are distinct in the pool. :return: ''' with patch('platform.node', MagicMock(return_value='nodename')), \ patch('socket.gethostname', MagicMock(return_value='hostname')), \ patch('socket.getfqdn', MagicMock(return_value='hostname.domainname.blank')), \ patch('socket.getaddrinfo', MagicMock(return_value=[(2, 3, 0, 'attrname', ('127.0.1.1', 0))])), \ patch('salt.utils.files.fopen', mock_open()), \ patch('salt.utils.network.ip_addrs', MagicMock(return_value=['1.2.3.4', '5.6.7.8'])): self.assertEqual(network._generate_minion_id(), ['hostname.domainname.blank', 'nodename', 'hostname', '1.2.3.4', '5.6.7.8']) def test_generate_minion_id_127_name(self): ''' Test if minion IDs can be named 127.foo :return: ''' with patch('platform.node', MagicMock(return_value='127')), \ patch('socket.gethostname', MagicMock(return_value='127')), \ patch('socket.getfqdn', MagicMock(return_value='127.domainname.blank')), \ patch('socket.getaddrinfo', MagicMock(return_value=[(2, 3, 0, 'attrname', ('127.0.1.1', 0))])), \ patch('salt.utils.files.fopen', mock_open()), \ patch('salt.utils.network.ip_addrs', MagicMock(return_value=['1.2.3.4', '5.6.7.8'])): self.assertEqual(network._generate_minion_id(), ['127.domainname.blank', '127', '1.2.3.4', '5.6.7.8']) def test_generate_minion_id_127_name_startswith(self): ''' Test if minion IDs can be named starting from "127" :return: ''' with patch('platform.node', MagicMock(return_value='127890')), \ patch('socket.gethostname', MagicMock(return_value='127890')), \ patch('socket.getfqdn', MagicMock(return_value='127890.domainname.blank')), \ patch('socket.getaddrinfo', MagicMock(return_value=[(2, 3, 0, 'attrname', ('127.0.1.1', 0))])), \ patch('salt.utils.files.fopen', mock_open()), \ patch('salt.utils.network.ip_addrs', MagicMock(return_value=['1.2.3.4', '5.6.7.8'])): self.assertEqual(network._generate_minion_id(), ['127890.domainname.blank', '127890', '1.2.3.4', '5.6.7.8']) def test_generate_minion_id_duplicate(self): ''' Test if IP addresses in the minion IDs are distinct in the pool :return: ''' with patch('platform.node', MagicMock(return_value='hostname')), \ patch('socket.gethostname', MagicMock(return_value='hostname')), \ patch('socket.getfqdn', MagicMock(return_value='hostname')), \ patch('socket.getaddrinfo', MagicMock(return_value=[(2, 3, 0, 'hostname', ('127.0.1.1', 0))])), \ patch('salt.utils.files.fopen', mock_open()), \ patch('salt.utils.network.ip_addrs', MagicMock(return_value=['1.2.3.4', '1.2.3.4', '1.2.3.4'])): self.assertEqual(network._generate_minion_id(), ['hostname', '1.2.3.4']) def test_generate_minion_id_platform_used(self): ''' Test if platform.node is used for the first occurrence. The platform.node is most common hostname resolver before anything else. :return: ''' with patch('platform.node', MagicMock(return_value='very.long.and.complex.domain.name')), \ patch('socket.gethostname', MagicMock(return_value='hostname')), \ patch('socket.getfqdn', MagicMock(return_value='')), \ patch('socket.getaddrinfo', MagicMock(return_value=[(2, 3, 0, 'hostname', ('127.0.1.1', 0))])), \ patch('salt.utils.files.fopen', mock_open()), \ patch('salt.utils.network.ip_addrs', MagicMock(return_value=['1.2.3.4', '1.2.3.4', '1.2.3.4'])): self.assertEqual(network.generate_minion_id(), 'very.long.and.complex.domain.name') def test_generate_minion_id_platform_localhost_filtered(self): ''' Test if localhost is filtered from the first occurrence. :return: ''' with patch('platform.node', MagicMock(return_value='localhost')), \ patch('socket.gethostname', MagicMock(return_value='pick.me')), \ patch('socket.getfqdn', MagicMock(return_value='hostname.domainname.blank')), \ patch('socket.getaddrinfo', MagicMock(return_value=[(2, 3, 0, 'hostname', ('127.0.1.1', 0))])), \ patch('salt.utils.files.fopen', mock_open()), \ patch('salt.utils.network.ip_addrs', MagicMock(return_value=['1.2.3.4', '1.2.3.4', '1.2.3.4'])): self.assertEqual(network.generate_minion_id(), 'hostname.domainname.blank') def test_generate_minion_id_platform_localhost_filtered_all(self): ''' Test if any of the localhost is filtered from everywhere. :return: ''' with patch('platform.node', MagicMock(return_value='localhost')), \ patch('socket.gethostname', MagicMock(return_value='ip6-loopback')), \ patch('socket.getfqdn', MagicMock(return_value='ip6-localhost')), \ patch('socket.getaddrinfo', MagicMock(return_value=[(2, 3, 0, 'localhost', ('127.0.1.1', 0))])), \ patch('salt.utils.files.fopen', mock_open()), \ patch('salt.utils.network.ip_addrs', MagicMock(return_value=['127.0.0.1', '::1', 'fe00::0', 'fe02::1', '1.2.3.4'])): self.assertEqual(network.generate_minion_id(), '1.2.3.4') def test_generate_minion_id_platform_localhost_only(self): ''' Test if there is no other choice but localhost. :return: ''' with patch('platform.node', MagicMock(return_value='localhost')), \ patch('socket.gethostname', MagicMock(return_value='ip6-loopback')), \ patch('socket.getfqdn', MagicMock(return_value='ip6-localhost')), \ patch('socket.getaddrinfo', MagicMock(return_value=[(2, 3, 0, 'localhost', ('127.0.1.1', 0))])), \ patch('salt.utils.files.fopen', mock_open()), \ patch('salt.utils.network.ip_addrs', MagicMock(return_value=['127.0.0.1', '::1', 'fe00::0', 'fe02::1'])): self.assertEqual(network.generate_minion_id(), 'localhost') def test_generate_minion_id_platform_fqdn(self): ''' Test if fqdn is picked up. :return: ''' with patch('platform.node', MagicMock(return_value='localhost')), \ patch('socket.gethostname', MagicMock(return_value='ip6-loopback')), \ patch('socket.getfqdn', MagicMock(return_value='pick.me')), \ patch('socket.getaddrinfo', MagicMock(return_value=[(2, 3, 0, 'localhost', ('127.0.1.1', 0))])), \ patch('salt.utils.files.fopen', mock_open()), \ patch('salt.utils.network.ip_addrs', MagicMock(return_value=['127.0.0.1', '::1', 'fe00::0', 'fe02::1'])): self.assertEqual(network.generate_minion_id(), 'pick.me') def test_generate_minion_id_platform_localhost_addrinfo(self): ''' Test if addinfo is picked up. :return: ''' with patch('platform.node', MagicMock(return_value='localhost')), \ patch('socket.gethostname', MagicMock(return_value='ip6-loopback')), \ patch('socket.getfqdn', MagicMock(return_value='ip6-localhost')), \ patch('socket.getaddrinfo', MagicMock(return_value=[(2, 3, 0, 'pick.me', ('127.0.1.1', 0))])), \ patch('salt.utils.files.fopen', mock_open()), \ patch('salt.utils.network.ip_addrs', MagicMock(return_value=['127.0.0.1', '::1', 'fe00::0', 'fe02::1'])): self.assertEqual(network.generate_minion_id(), 'pick.me') def test_generate_minion_id_platform_ip_addr_only(self): ''' Test if IP address is the only what is used as a Minion ID in case no DNS name. :return: ''' with patch('platform.node', MagicMock(return_value='localhost')), \ patch('socket.gethostname', MagicMock(return_value='ip6-loopback')), \ patch('socket.getfqdn', MagicMock(return_value='ip6-localhost')), \ patch('socket.getaddrinfo', MagicMock(return_value=[(2, 3, 0, 'localhost', ('127.0.1.1', 0))])), \ patch('salt.utils.files.fopen', mock_open()), \ patch('salt.utils.network.ip_addrs', MagicMock(return_value=['127.0.0.1', '::1', 'fe00::0', 'fe02::1', '1.2.3.4'])): self.assertEqual(network.generate_minion_id(), '1.2.3.4') def test_gen_mac(self): with patch('random.randint', return_value=1) as random_mock: self.assertEqual(random_mock.return_value, 1) ret = network.gen_mac('00:16:3E') expected_mac = '00:16:3E:01:01:01' self.assertEqual(ret, expected_mac) def test_mac_str_to_bytes(self): self.assertRaises(ValueError, network.mac_str_to_bytes, '31337') self.assertRaises(ValueError, network.mac_str_to_bytes, '0001020304056') self.assertRaises(ValueError, network.mac_str_to_bytes, '00:01:02:03:04:056') self.assertRaises(ValueError, network.mac_str_to_bytes, 'a0:b0:c0:d0:e0:fg') self.assertEqual(b'\x10\x08\x06\x04\x02\x00', network.mac_str_to_bytes('100806040200')) self.assertEqual(b'\xf8\xe7\xd6\xc5\xb4\xa3', network.mac_str_to_bytes('f8e7d6c5b4a3'))