1#!/usr/local/bin/python3.8 2# vim:fileencoding=UTF-8:ts=4:sw=4:sta:et:sts=4:ai 3# License: GPLv3 Copyright: 2010, Kovid Goyal <kovid at kovidgoyal.net> 4 5 6from contextlib import suppress 7 8from calibre.constants import isbsd, islinux, iswindows 9from calibre.utils.config_base import tweaks 10 11 12class LinuxNetworkStatus: 13 14 # Map of NetworkManager connectivity values to their XDP/GLib equivalents 15 NM_XDP_CONNECTIVITY_MAP = { 16 0: 4, # NM_CONNECTIVITY_UNKNOWN → Full network 17 1: 1, # NM_CONNECTIVITY_NONE → Local only 18 2: 3, # NM_CONNECTIVITY_PORTAL → Captive portal 19 3: 2, # NM_CONNECTIVITY_LIMITED → Limited connectivity 20 4: 4, # NM_CONNECTIVITY_FULL → Full network 21 } 22 23 def __init__(self): 24 from jeepney import DBusAddress, Properties, new_method_call 25 # Prefer desktop portal interface here since it can theoretically 26 # work with network management solutions other than NetworkManager 27 # and is controlled by the current desktop session 28 # 29 # There is no difference in terms of “features” provided between 30 # the two APIs from our point of view. 31 self.xdp_call = lambda : new_method_call(DBusAddress( 32 '/org/freedesktop/portal/desktop', 33 bus_name='org.freedesktop.portal.Desktop', 34 interface="org.freedesktop.portal.NetworkMonitor"), 'GetConnectivity') 35 self.nm_call = lambda : Properties(DBusAddress('/org/freedesktop/NetworkManager', 36 bus_name='org.freedesktop.NetworkManager', 37 interface="org.freedesktop.NetworkManager")).get('Connectivity') 38 39 if self.xdp() is not None: 40 self.get_connectivity = self.xdp 41 elif self.nm() is not None: 42 self.get_connectivity = self.nm 43 else: 44 self.get_connectivity = lambda : 4 45 46 def connect(self, which='SESSION'): 47 from jeepney.io.blocking import open_dbus_connection 48 if not hasattr(self, 'connection'): 49 self.connection = open_dbus_connection(which) 50 51 def xdp(self): 52 with suppress(Exception): 53 self.connect('SESSION') 54 return self.send(self.xdp_call()) 55 if hasattr(self, 'connection'): 56 self.connection.close() 57 del self.connection 58 59 def nm(self): 60 with suppress(Exception): 61 self.connect('SYSTEM') 62 return self.NM_XDP_CONNECTIVITY_MAP.get(self.send(self.nm_call()), 4) 63 if hasattr(self, 'connection'): 64 self.connection.close() 65 del self.connection 66 67 def send(self, msg): 68 from jeepney import DBusErrorResponse, MessageType 69 reply = self.connection.send_and_get_reply(msg) 70 if reply.header.message_type is MessageType.error: 71 raise DBusErrorResponse(reply) 72 return reply.body[0] 73 74 def __call__(self): 75 with suppress(Exception): 76 # Meanings of returned XDP/GLib connectivity values: 77 # * 1: Local only. The host is not configured with a route to the internet. 78 # * 2: Limited connectivity. The host is connected to a network, but can't reach the full internet. 79 # * 3: Captive portal. The host is behind a captive portal and cannot reach the full internet. 80 # * 4: Full network. The host connected to a network, and can reach the full internet. 81 return self.get_connectivity() == 4 82 return True 83 84 85class WindowsNetworkStatus: 86 87 def __init__(self): 88 from calibre_extensions import winutil 89 self.winutil = winutil 90 91 def __call__(self): 92 if self.winutil is None: 93 return True 94 return self.winutil.internet_connected() 95 96 97class DummyNetworkStatus: 98 99 def __call__(self): 100 return True 101 102 103def internet_connected(): 104 if tweaks['skip_network_check']: 105 return True 106 if not hasattr(internet_connected, 'checker'): 107 internet_connected.checker = WindowsNetworkStatus() if iswindows else \ 108 LinuxNetworkStatus() if (islinux or isbsd) else \ 109 DummyNetworkStatus() 110 111 return internet_connected.checker() 112