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