From feb29e278fd889315209275d866f550a4b186391 Mon Sep 17 00:00:00 2001 From: Samuel FORESTIER <dev@samuel.domains> Date: Fri, 9 Oct 2020 20:23:46 +0200 Subject: [PATCH] [WAN_IP] Reworks for configuration, DRY and performance purposes + Fixes IPv6 detection in some particular cases (behind a NAT46) + Speeds up `dig` usage failing when no IPv6 are available + DNS or HTTP external requests can now be disabled from configuration (fixes #81) + DNS or HTTP resolvers can now be changed from configuration + Factorizes internal code to DRY + DNS and HTTP timeouts are now uncoupled and may be different [skip ci] --- README.md | 30 +++- archey/entries/wan_ip.py | 112 ++++++++------ archey/test/entries/test_archey_wan_ip.py | 178 +++++++++++----------- config.json | 17 ++- 4 files changed, 188 insertions(+), 149 deletions(-) diff --git a/README.md b/README.md index c5d3550..9df18cb 100644 --- a/README.md +++ b/README.md @@ -314,15 +314,31 @@ Below stand further descriptions for each available (default) option : }, { "type": "WAN_IP", - // Set to `false` to only display IPv4 WAN addresses. - "ipv6_support": true, - // Some timeouts you can adjust as default ones might be undersized for your connectivity (seconds). - "ipv4_timeout_secs": 1, - "ipv6_timeout_secs": 1, // // As explained above, you may temporary hide entries as you wish. // See below example to hide your public IP addresses before posting your configuration on Internet. - //"disabled": true + //"disabled": true, + // + // Below are settings relative to IPv4/IPv6 public addresses retrieval. + // I hope options are self-explanatory. + // You may set `dns_query` (or `http_url`) to `false` to disable them. + // You may directly set `ipv4` or `ipv6` fields to `false` to completely disable them. + // + // <https://ident.me/> server sources : <https://github.com/pcarrier/identme>. + "ipv4": { + "dns_query": "myip.opendns.com", + "dns_resolver": "resolver1.opendns.com", + "dns_timeout": 1, + "http_url": "https://v4.ident.me/", + "http_timeout": 1 + }, + "ipv6": { + "dns_query": "myip.opendns.com", + "dns_resolver": "resolver1.opendns.com", + "dns_timeout": 1, + "http_url": "https://v6.ident.me/", + "http_timeout": 1 + } } ], "default_strings": { @@ -357,5 +373,3 @@ Any improvement would be appreciated. * If you had to tweak this project to make it work on your system, please **[open a pull request](https://github.com/HorlogeSkynet/archey4/pulls)** so as to share your modifications with the rest of the world and participate in this project ! You should also check [Info for contributors](https://github.com/HorlogeSkynet/archey4/wiki/Info-for-contributors). * If your distribution is not (currently) supported, please check [How do I add a distribution to Archey?](https://github.com/HorlogeSkynet/archey4/wiki/How-do-I-add-a-distribution-to-Archey%3F). - -* When looking up your public IP address (**WAN\_IP**), Archey will try at first to run a DNS query for `myip.opendns.com`, against OpenDNS's resolver(s). On error, it would fall back on regular HTTPS request(s) to <https://ident.me> ([server sources](https://github.com/pcarrier/identme)). diff --git a/archey/entries/wan_ip.py b/archey/entries/wan_ip.py index aa2f46e..4d6c364 100644 --- a/archey/entries/wan_ip.py +++ b/archey/entries/wan_ip.py @@ -15,62 +15,82 @@ class WanIP(Entry): self.value = [] - self._retrieve_ipv4_address() + ipv4_addr = self._retrieve_ip_address(4) + if ipv4_addr: + self.value.append(ipv4_addr) - # IPv6 address retrieval (unless the user doesn't want it). - if self.options.get('ipv6_support', True): - self._retrieve_ipv6_address() + ipv6_addr = self._retrieve_ip_address(6) + if ipv6_addr: + self.value.append(ipv6_addr) - self.value = list(filter(None, self.value)) - def _retrieve_ipv4_address(self): + def _retrieve_ip_address(self, ip_version): + """ + Best effort to retrieve public IP address based on corresponding options. + We are trying special DNS resolutions first for performance and (system) caching purposes. + """ + options = self.options.get('ipv{}'.format(ip_version), {}) + + # Is retrieval enabled for this IP version ? + if not options and options != {}: + return None + + # Is retrieval via DNS query enabled ? + dns_query = options.get('dns_query', 'myip.opendns.com') + if dns_query: + # Run the DNS query. + try: + ip_address = self._run_dns_query( + dns_query, + options.get('dns_resolver', 'resolver1.opendns.com'), + ('AAAA' if ip_version == 6 else 'A'), + options.get('dns_timeout', 1) + ) + except FileNotFoundError: + # DNS lookup tool does not seem to be available. + pass + else: + return ip_address + + # Is retrieval via HTTP(S) request enabled ? + http_url = options.get('http_url', 'https://v{}.ident.me/'.format(ip_version)) + if not http_url: + return None + + # Run the HTTP(S) request. + return self._run_http_request( + http_url, + options.get('http_timeout', 1) + ) + + + @staticmethod + def _run_dns_query(query, resolver, query_type, timeout): + """Simple wrapper to `dig` command to perform DNS queries""" try: - ipv4_addr = check_output( - [ - 'dig', '+short', '-4', 'A', 'myip.opendns.com', - '@resolver1.opendns.com' - ], - timeout=self.options.get('ipv4_timeout_secs', 1), + ip_address = check_output( + ['dig', '+short', query_type, query, '@' + resolver], + timeout=timeout, stderr=DEVNULL, universal_newlines=True ).rstrip() - except (FileNotFoundError, TimeoutExpired, CalledProcessError): - try: - ipv4_addr = urlopen( - 'https://v4.ident.me/', - timeout=self.options.get('ipv4_timeout_secs', 1) - ) - except (HTTPError, URLError, SocketTimeoutError): - # The machine does not seem to be connected to Internet... - return + except (TimeoutExpired, CalledProcessError): + return None - ipv4_addr = ipv4_addr.read().decode().strip() + # `ip_address` might be empty here. + return ip_address - self.value.append(ipv4_addr) - - def _retrieve_ipv6_address(self): + @staticmethod + def _run_http_request(server_url, timeout): + """Simple wrapper to `urllib` module to perform HTTP requests""" try: - ipv6_addr = check_output( - [ - 'dig', '+short', '-6', 'AAAA', 'myip.opendns.com', - '@resolver1.ipv6-sandbox.opendns.com' - ], - timeout=self.options.get('ipv6_timeout_secs', 1), - stderr=DEVNULL, universal_newlines=True - ).rstrip() - except (FileNotFoundError, TimeoutExpired, CalledProcessError): - try: - response = urlopen( - 'https://v6.ident.me/', - timeout=self.options.get('ipv6_timeout_secs', 1) - ) - except (HTTPError, URLError, SocketTimeoutError): - # It looks like this machine doesn't have any IPv6 address... - # ... or is not connected to Internet. - return + http_request = urlopen( + server_url, + timeout=timeout + ) + except (HTTPError, URLError, SocketTimeoutError): + return None - ipv6_addr = response.read().decode().strip() - - self.value.append(ipv6_addr) + return http_request.read().decode().strip() def output(self, output): diff --git a/archey/test/entries/test_archey_wan_ip.py b/archey/test/entries/test_archey_wan_ip.py index 3e5f215..379a78b 100644 --- a/archey/test/entries/test_archey_wan_ip.py +++ b/archey/test/entries/test_archey_wan_ip.py @@ -5,7 +5,6 @@ from unittest.mock import MagicMock, patch from socket import timeout as SocketTimeoutError from subprocess import TimeoutExpired -from urllib.error import URLError from archey.test import CustomAssertions from archey.entries.wan_ip import WanIP @@ -15,121 +14,116 @@ from archey.constants import DEFAULT_CONFIG class TestWanIPEntry(unittest.TestCase, CustomAssertions): """ - Here, we mock calls to `dig` or `urlopen`. + Here, we end up mocking calls to `dig` or `urlopen`. """ - @patch( - 'archey.entries.wan_ip.check_output', - side_effect=[ - 'XXX.YY.ZZ.TTT\n', - '0123::4567:89a:dead:beef\n' - ] - ) - def test_ipv6_and_ipv4(self, _): - """Test the regular case : Both IPv4 and IPv6 are retrieved""" - self.assertEqual( - WanIP(options={ - 'ipv6_support': True - }).value, - ['XXX.YY.ZZ.TTT', '0123::4567:89a:dead:beef'] - ) - - @patch( - 'archey.entries.wan_ip.check_output', - return_value='XXX.YY.ZZ.TTT' - ) - def test_ipv4_only(self, _): - """Test only public IPv4 detection""" - self.assertEqual( - WanIP(options={ - 'ipv6_support': False - }).value, - ['XXX.YY.ZZ.TTT'] - ) + def setUp(self): + """We use these mocks so often, it's worth defining them here.""" + self.wan_ip_mock = HelperMethods.entry_mock(WanIP) + self.output_mock = MagicMock() @patch( 'archey.entries.wan_ip.check_output', side_effect=[ - 'XXX.YY.ZZ.TTT', # The IPv4 address is detected - TimeoutExpired('dig', 1) # `check_output` call will fail + TimeoutExpired('dig', 1), # `check_output` call will hard-fail. + '0123::4567:89a:dead:beef\n', ] ) @patch('archey.entries.wan_ip.urlopen') - def test_ipv6_timeout(self, urlopen_mock, _): + def test_ipv4_ko_and_ipv6_ok(self, urlopen_mock, _): + """Test fallback on HTTP method only when DNS lookup failed""" + # `urlopen` will hard-fail. + urlopen_mock.return_value.read.side_effect = SocketTimeoutError(0) + + # IPv4 retrieval failed. + self.assertFalse( + WanIP._retrieve_ip_address(self.wan_ip_mock, 4), # pylint: disable=protected-access + ) + + # IPv6 worked like a (almost !) charm. + self.assertEqual( + WanIP._retrieve_ip_address(self.wan_ip_mock, 6), # pylint: disable=protected-access + '0123::4567:89a:dead:beef' + ) + + @patch( + 'archey.entries.wan_ip.check_output', + side_effect=[ + '\n', # `check_output` call will soft-fail. + FileNotFoundError('dig') # `check_output` call will hard-fail. + ] + ) + @patch('archey.entries.wan_ip.urlopen') + def test_proper_http_fallback(self, urlopen_mock, _): + """Test fallback on HTTP method only when DNS lookup failed""" + urlopen_mock.return_value.read.return_value = b'XXX.YY.ZZ.TTT\n' + + # HTTP back-end was not called, we trust DNS lookup tool which failed. + self.assertFalse( + WanIP._retrieve_ip_address(self.wan_ip_mock, 4), # pylint: disable=protected-access + ) + + # New try: HTTP method has been called ! + self.assertEqual( + WanIP._retrieve_ip_address(self.wan_ip_mock, 4), # pylint: disable=protected-access + 'XXX.YY.ZZ.TTT' + ) + + def test_retrieval_disabled(self): + """Test behavior when both IPv4 and IPv6 retrievals are purposely disabled""" + self.wan_ip_mock.options = { + 'ipv4': False, + 'ipv6': False + } + + # Both retrievals fail. + self.assertFalse( + WanIP._retrieve_ip_address(self.wan_ip_mock, 4) # pylint: disable=protected-access + ) + self.assertFalse( + WanIP._retrieve_ip_address(self.wan_ip_mock, 6) # pylint: disable=protected-access + ) + + def test_method_disabled(self): + """Check whether user could disable resolver back-ends from configuration""" + self.wan_ip_mock.options = { + 'ipv4': { + 'dns_query': False, + 'http_url': False + } + } + + # Internal method doesn't return any address. + self.assertFalse( + WanIP._retrieve_ip_address(self.wan_ip_mock, 4) # pylint: disable=protected-access + ) + + def test_two_addresses(self): """ - Test when `dig` call timeout for the IPv6 detection. + Test when both IPv4 and IPv6 addresses could be retrieved. Additionally check the `output` method behavior. """ - urlopen_mock.return_value.read.return_value = b'0123::4567:89a:dead:beef\n' + self.wan_ip_mock.value = ['XXX.YY.ZZ.TTT', '0123::4567:89a:dead:beef'] - wan_ip = WanIP(options={ - 'ipv6_support': True - }) + WanIP.output(self.wan_ip_mock, self.output_mock) - output_mock = MagicMock() - wan_ip.output(output_mock) - - self.assertListEqual( - wan_ip.value, - ['XXX.YY.ZZ.TTT', '0123::4567:89a:dead:beef'] - ) self.assertEqual( - output_mock.append.call_args[0][1], - 'XXX.YY.ZZ.TTT, 0123::4567:89a:dead:beef' + self.output_mock.append.call_args[0][1], + "XXX.YY.ZZ.TTT, 0123::4567:89a:dead:beef" ) - @patch( - 'archey.entries.wan_ip.check_output', - side_effect=TimeoutExpired('dig', 1) # `check_output` call will fail - ) - @patch( - 'archey.entries.wan_ip.urlopen', - # `urlopen` call will fail - side_effect=URLError('<urlopen error timed out>') - ) - def test_ipv4_timeout_twice(self, _, __): - """Test when both `dig` and `URLOpen` trigger timeouts...""" - self.assertListEmpty(WanIP(options={ - 'ipv6_support': False - }).value) - - @patch( - 'archey.entries.wan_ip.check_output', - side_effect=TimeoutExpired('dig', 1) # `check_output` call will fail - ) - @patch( - 'archey.entries.wan_ip.urlopen', - side_effect=SocketTimeoutError(1) # `urlopen` call will fail - ) - def test_ipv4_timeout_twice_socket_error(self, _, __): - """Test when both `dig` timeouts and `URLOpen` raises `socket.timeout`...""" - self.assertListEmpty(WanIP(options={ - 'ipv6_support': False - }).value) - - @patch( - 'archey.entries.wan_ip.check_output', - return_value='' # No address will be returned - ) - @patch( - 'urllib.request.urlopen', - return_value=None # No object will be returned - ) @HelperMethods.patch_clean_configuration - def test_no_address(self, _, __): + def test_no_address(self): """ Test when no address could be retrieved. Additionally check the `output` method behavior. """ - wan_ip = WanIP(options={ - 'ipv6_support': False - }) + self.wan_ip_mock.value = [] - output_mock = MagicMock() - wan_ip.output(output_mock) + WanIP.output(self.wan_ip_mock, self.output_mock) - self.assertListEmpty(wan_ip.value) + self.assertListEmpty(self.wan_ip_mock.value) self.assertEqual( - output_mock.append.call_args[0][1], + self.output_mock.append.call_args[0][1], DEFAULT_CONFIG['default_strings']['no_address'] ) diff --git a/config.json b/config.json index f65c5f8..f8689c9 100644 --- a/config.json +++ b/config.json @@ -56,9 +56,20 @@ }, { "type": "WAN_IP", - "ipv6_support": true, - "ipv4_timeout_secs": 1, - "ipv6_timeout_secs": 1 + "ipv4": { + "dns_query": "myip.opendns.com", + "dns_resolver": "resolver1.opendns.com", + "dns_timeout": 1, + "http_url": "https://v4.ident.me/", + "http_timeout": 1 + }, + "ipv6": { + "dns_query": "myip.opendns.com", + "dns_resolver": "resolver1.opendns.com", + "dns_timeout": 1, + "http_url": "https://v6.ident.me/", + "http_timeout": 1 + } } ], "default_strings": {