Content server: Listen for all incoming IPv4 and v6 connections by default

Clean up previous PR. Remains to do something about zeroconf, it should
advertise both IP addresses when the server is listening on both.
This commit is contained in:
Kovid Goyal 2023-12-04 16:31:14 +05:30
parent 725adf696f
commit f637aa59e5
No known key found for this signature in database
GPG Key ID: 06BC317B515ACE7C
6 changed files with 23 additions and 44 deletions

View File

@ -17,23 +17,15 @@ from calibre.utils.smtp import config as email_config
def local_url_for_content_server():
from calibre.srv.opts import server_config
from calibre.utils.network import is_ipv6_addr, get_fallback_server_addr
from calibre.utils.network import format_addr_for_url, get_fallback_server_addr
opts = server_config()
interface = opts.listen_on or get_fallback_server_addr()
addr_map = {'0.0.0.0': '127.0.0.1',
'::': '::1'}
if interface in addr_map:
interface = addr_map[interface]
addr = opts.listen_on or get_fallback_server_addr()
addr = {'0.0.0.0': '127.0.0.1', '::': '::1'}.get(addr, addr)
protocol = 'https' if opts.ssl_certfile and opts.ssl_keyfile else 'http'
prefix = opts.url_prefix or ''
port = opts.port
addr = f'[{interface}]' if is_ipv6_addr(interface) else f'{interface}'
return f'{protocol}://{addr}:{port}{prefix}'
return f'{protocol}://{format_addr_for_url(addr)}:{port}{prefix}'
def open_in_browser():

View File

@ -1299,23 +1299,12 @@ class ConfigWidget(ConfigWidgetBase):
self.stopping_msg.accept()
def test_server(self):
from calibre.utils.network import is_ipv6_addr, get_fallback_server_addr
from calibre.utils.network import format_addr_for_url, get_fallback_server_addr
prefix = self.advanced_tab.get('url_prefix') or ''
protocol = 'https' if self.advanced_tab.has_ssl else 'http'
lo = self.advanced_tab.get('listen_on') or get_fallback_server_addr()
addr_map = {'0.0.0.0': '127.0.0.1',
'::': '::1'}
if lo in addr_map:
lo = addr_map[lo]
if is_ipv6_addr(lo):
lo = f'[{lo}]'
url = '{protocol}://{interface}:{port}{prefix}'.format(
protocol=protocol, interface=lo,
port=self.main_tab.opt_port.value(), prefix=prefix)
addr = self.advanced_tab.get('listen_on') or get_fallback_server_addr()
addr = {'0.0.0.0': '127.0.0.1', '::': '::1'}.get(addr, addr)
url = f'{protocol}://{format_addr_for_url(addr)}:{self.main_tab.opt_port.value()}{prefix}'
open_url(QUrl(url))
def view_server_logs(self):

View File

@ -509,17 +509,14 @@ class ServerLoop:
self.setup_socket()
def serve(self):
from calibre.utils.network import is_ipv6_addr
from calibre.utils.network import format_addr_for_url
self.connection_map = {}
if not self.socket_was_preactivated:
self.socket.listen(min(socket.SOMAXCONN, 128))
self.bound_address = ba = self.socket.getsockname()
if isinstance(ba, tuple):
if is_ipv6_addr(ba[0]):
addr = f'[{ba[0]}]'
else:
addr = f'{ba[0]}'
addr = format_addr_for_url(str(ba[0]))
ba_str = f'{addr}:' + ':'.join(map(str, ba[1:]))
self.pool.start()
with TemporaryDirectory(prefix='srv-') as tdir:

View File

@ -112,8 +112,8 @@ raw_options = (
_('The interface on which to listen for connections'),
'listen_on', None,
_('The default is to listen on all available IPv6 and IPv4 interfaces. You can change this to, for'
' example, "127.0.0.1" to only listen for connections from the local machine, or'
' to "::" to listen to all incoming IPv6 and IPv4 connections.'),
' example, "127.0.0.1" to only listen for IPv4 connections from the local machine, or'
' to "0.0.0.0" to listen to all incoming IPv4 connections.'),
_('Fallback to auto-detected interface'),
'fallback_to_detected_interface', True,

View File

@ -77,9 +77,6 @@ def _get_external_ip():
def verify_ipV4_address(ip_address):
if ip_address == None:
return None
result = None
if ip_address != '0.0.0.0' and ip_address != '::':
# do some more sanity checks on the address
@ -87,7 +84,7 @@ def verify_ipV4_address(ip_address):
socket.inet_aton(ip_address)
if len(ip_address.split('.')) == 4:
result = ip_address
except OSError:
except Exception:
# Not legal ip address
pass
return result

View File

@ -113,13 +113,17 @@ def is_ipv6_addr(addr):
import socket
try:
socket.inet_pton(socket.AF_INET6, addr)
return True
except OSError:
except Exception:
return False
return True
def format_addr_for_url(addr):
if is_ipv6_addr(addr):
addr = f'[{addr}]'
return addr
def get_fallback_server_addr():
from socket import has_dualstack_ipv6
if has_dualstack_ipv6():
return '::'
else:
return '0.0.0.0'
return '::' if has_dualstack_ipv6() else '0.0.0.0'