mirror of
https://github.com/HorlogeSkynet/archey4
synced 2025-04-11 12:00:19 +02:00
[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]
This commit is contained in:
parent
1d7a29e0b3
commit
feb29e278f
30
README.md
30
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)).
|
||||
|
@ -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):
|
||||
|
@ -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']
|
||||
)
|
||||
|
||||
|
17
config.json
17
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": {
|
||||
|
Loading…
x
Reference in New Issue
Block a user