[Common] Add is_interface to validate network interfaces
Libtorrent now supports interface names instead of just IP address so add new common functions to validate user input. * Added is_interface that will verify if a libtorrent interface of name or IP address. * Added is_interface_name to verify that the name supplied is a valid network interface name in the operating system. On Windows sock.if_nameindex() is only supported on 3.8+ and does not return a uuid (required by libtorrent) so use ifaddr package. Using git commit version for ifaddr due to adapter name decode bug in v0.1.7. On other OSes attempt to use stdlib and fallback to ifaddr if installed otherwiser return True. * Added tests for is_interface & is_interface_name * Updated UIs with change from address to interface * Updated is_ipv6 and is_ipv4 to used inet_pton; now supported on Windows. Ref: https://github.com/pydron/ifaddr/pull/32 Closes: https://github.com/deluge-torrent/deluge/pull/338
This commit is contained in:
parent
d8acadb085
commit
540d557cb2
|
@ -28,6 +28,7 @@ All modules will require the [common](#common) section dependencies.
|
|||
- [setproctitle] - Optional: Renaming processes.
|
||||
- [Pillow] - Optional: Support for resizing tracker icons.
|
||||
- [dbus-python] - Optional: Show item location in filemanager.
|
||||
- [ifaddr] - Optional: Verify network interfaces.
|
||||
|
||||
### Linux and BSD
|
||||
|
||||
|
@ -96,3 +97,4 @@ All modules will require the [common](#common) section dependencies.
|
|||
[libnotify]: https://developer.gnome.org/libnotify/
|
||||
[python-appindicator]: https://packages.ubuntu.com/xenial/python-appindicator
|
||||
[librsvg]: https://wiki.gnome.org/action/show/Projects/LibRsvg
|
||||
[ifaddr]: https://pypi.org/project/ifaddr/
|
||||
|
|
|
@ -17,6 +17,7 @@ import numbers
|
|||
import os
|
||||
import platform
|
||||
import re
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import tarfile
|
||||
|
@ -44,6 +45,11 @@ if platform.system() in ('Windows', 'Microsoft'):
|
|||
|
||||
os.environ['SSL_CERT_FILE'] = where()
|
||||
|
||||
try:
|
||||
import ifaddr
|
||||
except ImportError:
|
||||
ifaddr = None
|
||||
|
||||
|
||||
if platform.system() not in ('Windows', 'Microsoft', 'Darwin'):
|
||||
# gi makes dbus available on Window but don't import it as unused.
|
||||
|
@ -900,6 +906,29 @@ def free_space(path):
|
|||
return disk_data.f_bavail * block_size
|
||||
|
||||
|
||||
def is_interface(interface):
|
||||
"""Check if interface is a valid IP or network adapter.
|
||||
|
||||
Args:
|
||||
interface (str): The IP or interface name to test.
|
||||
|
||||
Returns:
|
||||
bool: Whether interface is valid is not.
|
||||
|
||||
Examples:
|
||||
Windows:
|
||||
>>> is_interface('{7A30AE62-23ZA-3744-Z844-A5B042524871}')
|
||||
>>> is_interface('127.0.0.1')
|
||||
True
|
||||
Linux:
|
||||
>>> is_interface('lo')
|
||||
>>> is_interface('127.0.0.1')
|
||||
True
|
||||
|
||||
"""
|
||||
return is_ip(interface) or is_interface_name(interface)
|
||||
|
||||
|
||||
def is_ip(ip):
|
||||
"""A test to see if 'ip' is a valid IPv4 or IPv6 address.
|
||||
|
||||
|
@ -935,15 +964,12 @@ def is_ipv4(ip):
|
|||
|
||||
"""
|
||||
|
||||
import socket
|
||||
|
||||
try:
|
||||
if windows_check():
|
||||
return socket.inet_aton(ip)
|
||||
else:
|
||||
return socket.inet_pton(socket.AF_INET, ip)
|
||||
socket.inet_pton(socket.AF_INET, ip)
|
||||
except OSError:
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
|
||||
def is_ipv6(ip):
|
||||
|
@ -962,23 +988,51 @@ def is_ipv6(ip):
|
|||
"""
|
||||
|
||||
try:
|
||||
import ipaddress
|
||||
except ImportError:
|
||||
import socket
|
||||
|
||||
try:
|
||||
return socket.inet_pton(socket.AF_INET6, ip)
|
||||
except (OSError, AttributeError):
|
||||
if windows_check():
|
||||
log.warning('Unable to verify IPv6 Address on Windows.')
|
||||
return True
|
||||
socket.inet_pton(socket.AF_INET6, ip)
|
||||
except OSError:
|
||||
return False
|
||||
else:
|
||||
try:
|
||||
return ipaddress.IPv6Address(decode_bytes(ip))
|
||||
except ipaddress.AddressValueError:
|
||||
pass
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def is_interface_name(name):
|
||||
"""Returns True if an interface name exists.
|
||||
|
||||
Args:
|
||||
name (str): The Interface to test. eg. eth0 linux. GUID on Windows.
|
||||
|
||||
Returns:
|
||||
bool: Whether name is valid or not.
|
||||
|
||||
Examples:
|
||||
>>> is_interface_name("eth0")
|
||||
True
|
||||
>>> is_interface_name("{7A30AE62-23ZA-3744-Z844-A5B042524871}")
|
||||
True
|
||||
|
||||
"""
|
||||
|
||||
if not windows_check():
|
||||
try:
|
||||
socket.if_nametoindex(name)
|
||||
except OSError:
|
||||
pass
|
||||
else:
|
||||
return True
|
||||
|
||||
if ifaddr:
|
||||
try:
|
||||
adapters = ifaddr.get_adapters()
|
||||
except OSError:
|
||||
return True
|
||||
else:
|
||||
return any([name == a.name for a in adapters])
|
||||
|
||||
if windows_check():
|
||||
regex = '^{[0-9A-Z]{8}-([0-9A-Z]{4}-){3}[0-9A-Z]{12}}$'
|
||||
return bool(re.search(regex, str(name)))
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def decode_bytes(byte_str, encoding='utf8'):
|
||||
|
|
|
@ -164,19 +164,25 @@ class Core(component.Component):
|
|||
# store the one in the config so we can restore it on shutdown
|
||||
self._old_listen_interface = None
|
||||
if listen_interface:
|
||||
if deluge.common.is_ip(listen_interface):
|
||||
if deluge.common.is_interface(listen_interface):
|
||||
self._old_listen_interface = self.config['listen_interface']
|
||||
self.config['listen_interface'] = listen_interface
|
||||
else:
|
||||
log.error(
|
||||
'Invalid listen interface (must be IP Address): %s',
|
||||
'Invalid listen interface (must be IP Address or Interface Name): %s',
|
||||
listen_interface,
|
||||
)
|
||||
|
||||
self._old_outgoing_interface = None
|
||||
if outgoing_interface:
|
||||
self._old_outgoing_interface = self.config['outgoing_interface']
|
||||
self.config['outgoing_interface'] = outgoing_interface
|
||||
if deluge.common.is_interface(outgoing_interface):
|
||||
self._old_outgoing_interface = self.config['outgoing_interface']
|
||||
self.config['outgoing_interface'] = outgoing_interface
|
||||
else:
|
||||
log.error(
|
||||
'Invalid outgoing interface (must be IP Address or Interface Name): %s',
|
||||
outgoing_interface,
|
||||
)
|
||||
|
||||
# New release check information
|
||||
self.__new_release = None
|
||||
|
|
|
@ -21,6 +21,8 @@ from deluge.common import (
|
|||
ftime,
|
||||
get_path_size,
|
||||
is_infohash,
|
||||
is_interface,
|
||||
is_interface_name,
|
||||
is_ip,
|
||||
is_ipv4,
|
||||
is_ipv6,
|
||||
|
@ -116,6 +118,55 @@ class CommonTestCase(unittest.TestCase):
|
|||
self.assertTrue(is_ipv6('2001:db8::'))
|
||||
self.assertFalse(is_ipv6('2001:db8:'))
|
||||
|
||||
def get_windows_interface_name(self):
|
||||
import winreg
|
||||
|
||||
# find a network card in the registery
|
||||
with winreg.OpenKey(
|
||||
winreg.HKEY_LOCAL_MACHINE,
|
||||
r'SOFTWARE\Microsoft\Windows NT\CurrentVersion\NetworkCards',
|
||||
) as key:
|
||||
self.assertTrue(
|
||||
winreg.QueryInfoKey(key)[0] > 0
|
||||
) # must have at least 1 network card
|
||||
network_card = winreg.EnumKey(key, 0)
|
||||
# get GUID of network card
|
||||
with winreg.OpenKey(
|
||||
winreg.HKEY_LOCAL_MACHINE,
|
||||
fr'SOFTWARE\Microsoft\Windows NT\CurrentVersion\NetworkCards\{network_card}',
|
||||
) as key:
|
||||
for i in range(1):
|
||||
value = winreg.EnumValue(key, i)
|
||||
if value[0] == 'ServiceName':
|
||||
interface_name = value[1]
|
||||
return interface_name
|
||||
|
||||
def test_is_interface_name(self):
|
||||
if windows_check():
|
||||
interface_name = self.get_windows_interface_name()
|
||||
self.assertFalse(is_interface_name('2001:db8:'))
|
||||
self.assertFalse(
|
||||
is_interface_name('{THIS0000-IS00-ONLY-FOR0-TESTING00000}')
|
||||
)
|
||||
self.assertTrue(is_interface_name(interface_name))
|
||||
else:
|
||||
self.assertTrue(is_interface_name('lo'))
|
||||
self.assertFalse(is_interface_name('127.0.0.1'))
|
||||
self.assertFalse(is_interface_name('eth01101'))
|
||||
|
||||
def test_is_interface(self):
|
||||
if windows_check():
|
||||
interface_name = self.get_windows_interface_name()
|
||||
self.assertTrue(is_interface('127.0.0.1'))
|
||||
self.assertTrue(is_interface(interface_name))
|
||||
self.assertFalse(is_interface('127'))
|
||||
self.assertFalse(is_interface('{THIS0000-IS00-ONLY-FOR0-TESTING00000}'))
|
||||
else:
|
||||
self.assertTrue(is_interface('lo'))
|
||||
self.assertTrue(is_interface('127.0.0.1'))
|
||||
self.assertFalse(is_interface('127.'))
|
||||
self.assertFalse(is_interface('eth01101'))
|
||||
|
||||
def test_version_split(self):
|
||||
self.assertTrue(VersionSplit('1.2.2') == VersionSplit('1.2.2'))
|
||||
self.assertTrue(VersionSplit('1.2.1') < VersionSplit('1.2.2'))
|
||||
|
|
|
@ -8,7 +8,7 @@
|
|||
|
||||
import logging
|
||||
|
||||
from deluge.common import is_ip
|
||||
from deluge.common import is_interface
|
||||
from deluge.decorators import overrides
|
||||
from deluge.i18n import get_languages
|
||||
from deluge.ui.client import client
|
||||
|
@ -91,11 +91,12 @@ class BasePreferencePane(BaseInputPane, BaseWindow, PopupsHandler):
|
|||
)
|
||||
elif ipt.name == 'listen_interface':
|
||||
listen_interface = ipt.get_value().strip()
|
||||
if is_ip(listen_interface) or not listen_interface:
|
||||
if is_interface(listen_interface) or not listen_interface:
|
||||
conf_dict['listen_interface'] = listen_interface
|
||||
elif ipt.name == 'outgoing_interface':
|
||||
outgoing_interface = ipt.get_value().strip()
|
||||
conf_dict['outgoing_interface'] = outgoing_interface
|
||||
if is_interface(outgoing_interface) or not outgoing_interface:
|
||||
conf_dict['outgoing_interface'] = outgoing_interface
|
||||
elif ipt.name.startswith('proxy_'):
|
||||
if ipt.name == 'proxy_type':
|
||||
conf_dict.setdefault('proxy', {})['type'] = ipt.get_value()
|
||||
|
|
|
@ -2573,8 +2573,8 @@ used sparingly.</property>
|
|||
<object class="GtkEntry" id="entry_interface">
|
||||
<property name="visible">True</property>
|
||||
<property name="can_focus">True</property>
|
||||
<property name="tooltip_text" translatable="yes">The IP address of the interface to listen for incoming bittorrent connections on. Leave this empty if you want to use the default.</property>
|
||||
<property name="max_length">15</property>
|
||||
<property name="tooltip_text" translatable="yes">IP address or network interface name to listen for incoming BitTorrent connections. Leave empty to use system default.</property>
|
||||
<property name="max_length">40</property>
|
||||
<property name="width_chars">15</property>
|
||||
<property name="truncate_multiline">True</property>
|
||||
<property name="primary_icon_activatable">False</property>
|
||||
|
@ -2587,7 +2587,7 @@ used sparingly.</property>
|
|||
<object class="GtkLabel" id="label110">
|
||||
<property name="visible">True</property>
|
||||
<property name="can_focus">False</property>
|
||||
<property name="label" translatable="yes">Incoming Address</property>
|
||||
<property name="label" translatable="yes">Incoming Interface</property>
|
||||
<attributes>
|
||||
<attribute name="weight" value="bold"/>
|
||||
</attributes>
|
||||
|
@ -2812,9 +2812,9 @@ used sparingly.</property>
|
|||
<property name="visible">True</property>
|
||||
<property name="can_focus">True</property>
|
||||
<property name="tooltip_text" translatable="yes">
|
||||
The network interface name or IP address for outgoing BitTorrent connections. (Leave empty for default.)
|
||||
IP address or network interface name for outgoing BitTorrent connections. Leave empty to use system default.
|
||||
</property>
|
||||
<property name="max_length">15</property>
|
||||
<property name="max_length">40</property>
|
||||
<property name="invisible_char">●</property>
|
||||
<property name="width_chars">15</property>
|
||||
<property name="truncate_multiline">True</property>
|
||||
|
|
|
@ -671,11 +671,15 @@ class Preferences(component.Component):
|
|||
'chk_random_outgoing_ports'
|
||||
).get_active()
|
||||
incoming_address = self.builder.get_object('entry_interface').get_text().strip()
|
||||
if deluge.common.is_ip(incoming_address) or not incoming_address:
|
||||
if deluge.common.is_interface(incoming_address) or not incoming_address:
|
||||
new_core_config['listen_interface'] = incoming_address
|
||||
new_core_config['outgoing_interface'] = (
|
||||
outgoing_address = (
|
||||
self.builder.get_object('entry_outgoing_interface').get_text().strip()
|
||||
)
|
||||
if deluge.common.is_interface(outgoing_address) or not outgoing_address:
|
||||
new_core_config['outgoing_interface'] = (
|
||||
self.builder.get_object('entry_outgoing_interface').get_text().strip()
|
||||
)
|
||||
new_core_config['peer_tos'] = self.builder.get_object(
|
||||
'entry_peer_tos'
|
||||
).get_text()
|
||||
|
|
|
@ -6,7 +6,7 @@ from PyInstaller.utils.hooks import collect_all, collect_submodules, copy_metada
|
|||
|
||||
datas = []
|
||||
binaries = []
|
||||
hiddenimports = ['pygame']
|
||||
hiddenimports = ['pygame','ifaddr']
|
||||
|
||||
# Collect Meta Data
|
||||
datas += copy_metadata('deluge', recursive=True)
|
||||
|
|
|
@ -13,3 +13,4 @@ windows-curses; sys_platform == 'win32'
|
|||
zope.interface>=4.4.2
|
||||
distro; 'linux' in sys_platform or 'bsd' in sys_platform
|
||||
pygeoip
|
||||
https://github.com/pydron/ifaddr/archive/37cb5334f392f12811d38d90ec891746e3247c76.zip
|
||||
|
|
Loading…
Reference in New Issue