UI: Add a connection info expander

This commit adds a connection info expander to the UI.

Uses the NM api for querying, except where I could not find any
functions (e.g. rx and tx bytes, which are gathered from standard file
locations).

If it cannot find the data, as this is not foolproof, it will set it
to N/A, the application will keep working.

Note that glade changed the property names from `_` to `-`, but this
still works on the VMS. Apparently `_` are an old format that is not
used any more in glade.

Signed-off-by: jwijenbergh <[email protected]>

eduvpn/network.py

@@ -3,21 +3,16 @@
 import enum
 from functools import partial
 from time import sleep
-from gettext import gettext
 from . import nm
 from . import settings
 from .state_machine import BaseState
 from .app import Application
 from .server import ConfiguredServer as Server
-from .utils import run_in_background_thread
+from .utils import run_in_background_thread, translated_property
 
 logger = logging.getLogger(__name__)
 
 
-def translated_property(text):
-    return property(lambda self: gettext(text))  # type: ignore
-
-
 class StatusImage(enum.Enum):
     # The value is the image filename.
     DEFAULT = 'desktop-default.png'

eduvpn/nm.py

@@ -44,6 +44,73 @@ def get_mainloop():
     return GLib.MainLoop()
 
 
+def get_iface() -> Optional[str]:
+    """
+    Get the interface as a string if there is a master device
+    """
+    client = get_client()
+    active_connection = client.get_primary_connection()
+    if not active_connection:
+        return None
+
+    master = active_connection.get_master()
+    if not master:
+        return None
+
+    return master.get_ip_iface()
+
+
+def get_ipv4() -> Optional[str]:
+    """
+    Get the ipv4 address as a string if there is one
+    """
+    client = get_client()
+    active_connection = client.get_primary_connection()
+    if not active_connection:
+        return None
+
+    ip4_config = active_connection.get_ip4_config()
+    addresses = ip4_config.get_addresses()
+    if not addresses:
+        return None
+    return addresses[0].get_address()
+
+
+def get_ipv6() -> Optional[str]:
+    """
+    Get the ipv6 address as a string if there is one
+    """
+    client = get_client()
+    active_connection = client.get_primary_connection()
+    if not active_connection:
+        return None
+
+    ip6_config = active_connection.get_ip6_config()
+    addresses = ip6_config.get_addresses()
+    if not addresses:
+        return None
+    return addresses[0].get_address()
+
+
+def get_timestamp() -> Optional[int]:
+    """
+    Get the timestamp the connection was last activated
+    """
+    client = get_client()
+    active_connection = client.get_primary_connection()
+    if not active_connection:
+        return None
+
+    connection = active_connection.get_connection()
+    if not connection:
+        return None
+
+    setting_connection = connection.get_settings()
+    if not setting_connection:
+        return None
+    return setting_connection[0].get_timestamp()
+
+
 def nm_available() -> bool:
     """
     check if Network Manager is available

eduvpn/ui/stats.py

@@ -0,0 +1,145 @@
+import logging
+import time
+from datetime import timedelta
+from pathlib import Path
+from typing import Optional, TextIO
+
+from ..nm import get_iface, get_ipv4, get_ipv6, get_timestamp
+from ..utils import cache, get_human_readable_bytes, translated_property
+
+logger = logging.getLogger(__name__)
+
+LINUX_NET_FOLDER = Path("/sys/class/net")
+
+
+class NetworkStats:
+
+    default_text = translated_property("N/A")
+
+    # These properties define an LRU cache
+    # This cache is used so that we do not query these every second
+    @property  # type: ignore
+    @cache
+    def ipv4(self) -> str:
+        _ipv4 = get_ipv4()
+        if _ipv4 is None:
+            _ipv4 = self.default_text
+        return _ipv4
+
+    @property  # type: ignore
+    @cache
+    def ipv6(self) -> str:
+        _ipv6 = get_ipv6()
+        if _ipv6 is None:
+            _ipv6 = self.default_text
+        return _ipv6
+
+    @property  # type: ignore
+    @cache
+    def upload_file(self) -> Optional[TextIO]:
+        return self.open_file("tx_bytes")
+
+    @property  # type: ignore
+    @cache
+    def download_file(self) -> Optional[TextIO]:
+        return self.open_file("rx_bytes")
+
+    @property  # type: ignore
+    @cache
+    def start_bytes_upload(self) -> Optional[int]:
+        return self.get_file_bytes(self.upload_file)  # type: ignore
+
+    @property  # type: ignore
+    @cache
+    def start_bytes_download(self) -> Optional[int]:
+        return self.get_file_bytes(self.download_file)  # type: ignore
+
+    @property  # type: ignore
+    @cache
+    def iface(self) -> Optional[str]:
+        return get_iface()
+
+    @property  # type: ignore
+    @cache
+    def timestamp(self) -> Optional[int]:
+        return get_timestamp()
+
+    @property
+    def download(self) -> str:
+        """
+        Get the download as a human readable string
+        """
+        file_bytes_download = self.get_file_bytes(self.download_file)  # type: ignore
+        if file_bytes_download is None:
+            return self.default_text
+        if file_bytes_download <= self.start_bytes_download:  # type: ignore
+            return get_human_readable_bytes(0)
+        return get_human_readable_bytes(file_bytes_download - self.start_bytes_download)  # type: ignore
+
+    @property
+    def upload(self) -> str:
+        """
+        Get the upload as a human readable string
+        """
+        file_bytes_upload = self.get_file_bytes(self.upload_file)  # type: ignore
+        if file_bytes_upload is None:
+            return self.default_text
+        if file_bytes_upload <= self.start_bytes_upload:  # type: ignore
+            return get_human_readable_bytes(0)
+        return get_human_readable_bytes(file_bytes_upload - self.start_bytes_upload)  # type:ignore
+
+    @property
+    def duration(self) -> str:
+        """
+        Get the duration of the connection, in "HH:MM:SS"
+        """
+        if self.timestamp is None:
+            logger.warning("Network Stats: failed to get timestamp")
+            return self.default_text
+        now_unix_seconds = int(time.time())
+        duration = now_unix_seconds - self.timestamp  # type: ignore
+        return str(timedelta(seconds=duration))
+
+    def open_file(self, filename: str) -> Optional[TextIO]:
+        """
+        Helper function to open a statistics network file
+        """
+        if not self.iface:
+            logger.warning(f"Network Stats: {filename}, failed to get interface")
+            return None
+        filepath = LINUX_NET_FOLDER / self.iface / "statistics" / filename  # type: ignore
+        if not filepath.is_file():
+            logger.warning(f"Network Stats: {filepath} is not a file")
+            return None
+        return open(filepath, "r")
+
+    def get_file_bytes(self, filehandler: Optional[TextIO]) -> Optional[int]:
+        """
+        Helper function to get a statistics file to calculate the total data transfer
+        """
+        # If the interface is not set
+        # or the file is not present, we cannot get the stat
+        if not self.iface:
+            # Warning was already shown
+            return None
+        if not filehandler:
+            # Warning was already shown
+            return None
+
+        # Get the statistic from the file
+        # and go to the beginning
+        try:
+            stat = int(filehandler.readline())
+        except ValueError:
+            stat = 0
+        filehandler.seek(0)
+        return stat
+
+    def cleanup(self):
+        """
+        Cleanup the network stats by closing the files
+        """
+        if self.download_file:
+            self.download_file.close()
+        if self.upload_file:
+            self.upload_file.close()

eduvpn/ui/ui.py

@@ -29,6 +29,7 @@
     get_prefix, run_in_main_gtk_thread, run_periodically, cancel_at_context_end)
 from . import search
 from .utils import show_ui_component, link_markup, show_error_dialog
+from .stats import NetworkStats
 
 logger = logging.getLogger(__name__)
 
@@ -90,6 +91,7 @@ def setup(self, builder, application: Application):
             "on_search_changed": self.on_search_changed,
             "on_search_activate": self.on_search_activate,
             "on_switch_connection_state": self.on_switch_connection_state,
+            "on_toggle_connection_info": self.on_toggle_connection_info,
             "on_profile_selection_changed": self.on_profile_selection_changed,
             "on_location_selection_changed": self.on_location_selection_changed,
             "on_acknowledge_error": self.on_acknowledge_error,
@@ -134,6 +136,14 @@ def setup(self, builder, application: Application):
         self.connection_status_label = builder.get_object('connectionStatusLabel')
         self.connection_session_label = builder.get_object('connectionSessionLabel')
         self.connection_switch = builder.get_object('connectionSwitch')
+        self.connection_info_expander = builder.get_object('connectionInfoExpander')
+        self.connection_info_duration = builder.get_object('connectionInfoDurationText')
+        self.connection_info_downloaded = builder.get_object('connectionInfoDownloadedText')
+        self.connection_info_uploaded = builder.get_object('connectionInfoUploadedText')
+        self.connection_info_ipv4address = builder.get_object('connectionInfoIpv4AddressText')
+        self.connection_info_ipv6address = builder.get_object('connectionInfoIpv6AddressText')
+        self.connection_info_thread_cancel = None
+        self.connection_info_stats = None
 
         self.server_image = builder.get_object('serverImage')
         self.server_label = builder.get_object('serverLabel')
@@ -493,6 +503,12 @@ def enter_ConfiguringConnection(self, old_state, new_state):
     def exit_ConfiguringConnection(self, old_state, new_state):
         self.hide_loading_page()
 
+    @transition_edge_callback(ENTER, network_state.ConnectedState)
+    def enter_ConnectedState(self, old_state, new_state):
+        is_expanded = self.connection_info_expander.get_expanded()
+        if is_expanded:
+            self.start_connection_info()
+
     @transition_edge_callback(ENTER, interface_state.ConnectionStatus)
     def enter_ConnectionStatus(self, old_state, new_state):
         self.show_page(self.connection_page)
@@ -502,6 +518,7 @@ def enter_ConnectionStatus(self, old_state, new_state):
     @transition_edge_callback(EXIT, interface_state.ConnectionStatus)
     def exit_ConnectionStatus(self, old_state, new_state):
         self.hide_page(self.connection_page)
+        self.stop_connection_info()
 
     @transition_level_callback(interface_state.ConnectionStatus)
     def context_ConnectionStatus(self, state):
@@ -511,6 +528,10 @@ def context_ConnectionStatus(self, state):
             'update-validity',
         ))
 
+    @transition_edge_callback(ENTER, network_state.DisconnectedState)
+    def enter_DisconnectedState(self, old_state, new_state):
+        self.stop_connection_info()
+
     @transition_edge_callback(ENTER, interface_state.ErrorState)
     def enter_ErrorState(self, old_state, new_state):
         self.show_page(self.error_page)
@@ -595,6 +616,50 @@ def on_switch_connection_state(self, switch, state):
                 self.app.interface_transition('deactivate_connection')
         return True
 
+    def stop_connection_info(self):
+        if self.connection_info_thread_cancel:
+            self.connection_info_thread_cancel()
+            self.connection_info_thread = None
+
+        if self.connection_info_stats:
+            self.connection_info_stats.cleanup()
+            self.connection_info_stats = None
+
+    def start_connection_info(self):
+        if not self.app.network_state.has_transition('disconnect'):
+            logger.info("Connection Info: VPN is not active")
+            return
+
+        def update_connection_info_callback():
+            assert self.connection_info_stats
+            download = self.connection_info_stats.download
+            upload = self.connection_info_stats.upload
+            ipv4 = self.connection_info_stats.ipv4
+            ipv6 = self.connection_info_stats.ipv6
+            duration = self.connection_info_stats.duration
+            self.connection_info_downloaded.set_text(download)
+            self.connection_info_uploaded.set_text(upload)
+            self.connection_info_ipv4address.set_text(ipv4)
+            self.connection_info_ipv6address.set_text(ipv6)
+            self.connection_info_duration.set_text(duration)
+
+        if not self.connection_info_stats:
+            self.connection_info_stats = NetworkStats()
+
+        # Run every second in the background
+        self.connection_info_thread_cancel = run_periodically(
+            update_connection_info_callback, 1
+        )
+
+    def on_toggle_connection_info(self, _):
+        logger.debug("clicked on connection info")
+        was_expanded = self.connection_info_expander.get_expanded()
+
+        if not was_expanded:
+            self.start_connection_info()
+        else:
+            self.stop_connection_info()
+
     def on_profile_selection_changed(self, selection):
         logger.debug("selected profile")
         (model, tree_iter) = selection.get_selected()

eduvpn/utils.py

@@ -4,6 +4,7 @@
 from datetime import datetime
 from email.utils import parsedate_to_datetime
 from functools import lru_cache, partial, wraps
+from gettext import gettext
 from logging import getLogger
 from os import path, environ
 from sys import prefix
@@ -169,3 +170,25 @@ def parse_http_date_header(date: str) -> datetime:
 
 
 parse_http_expires_header = parse_http_date_header
+
+
+def get_human_readable_bytes(total_bytes: int) -> str:
+    """
+    Helper function to calculate the human readable bytes.
+    E.g. B, kB, MB, GB, TB.
+    """
+    suffix = ""
+    hr_bytes = float(total_bytes)
+    for suffix in ["B", "kB", "MB", "GB", "TB"]:
+        if hr_bytes < 1024.0:
+            break
+        if suffix != "TB":
+            hr_bytes /= 1024.0
+
+    if suffix == "B":
+        return f"{int(hr_bytes)} {suffix}"
+    return f"{hr_bytes:.2f} {suffix}"
+
+
+def translated_property(text):
+    return property(lambda self: gettext(text))  # type: ignore