config/chroot_local-includes/usr/lib/python3/dist-packages/tailsgreeter/settings/persistence.py

   1 # Copyright 2012-2016 Tails developers <tails@boum.org>
   2 # Copyright 2011 Max <govnototalitarizm@gmail.com>
   3 # Copyright 2011 Martin Owens
   4 #
   5 # This program is free software: you can redistribute it and/or modify
   6 #  it under the terms of the GNU General Public License as published by
   7 #  the Free Software Foundation, either version 3 of the License, or
   8 #  (at your option) any later version.
   9 #
  10 #  This program is distributed in the hope that it will be useful,
  11 #  but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 #  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 #  GNU General Public License for more details.
  14 #
  15 #  You should have received a copy of the GNU General Public License
  16 #  along with this program.  If not, see <http://www.gnu.org/licenses/>
  17 #
  18 """Persistent Storage handling"""
  19 import gettext
  20 import logging
  21 import os
  22 from typing import Optional
  23
  24 from gi.repository import Gio, GLib
  25
  26 import tailsgreeter.errors
  27 import tps.dbus.errors as tps_errors
  28 from tailsgreeter import config  # NOQA: E402
  29 from tps import InvalidBootDeviceErrorType
  30
  31
  32 _ = gettext.gettext
  33
  34 BUS_NAME = "org.boum.tails.PersistentStorage"
  35 OBJECT_PATH = "/org/boum/tails/PersistentStorage"
  36 INTERFACE_NAME = "org.boum.tails.PersistentStorage"
  37
  38
  39 class PersistentStorageSettings:
  40     """Controller for settings related to Persistent Storage"""
  41
  42     def __init__(self) -> None:
  43         self.failed_with_unrecoverable_error = False
  44         self.cleartext_name = "TailsData_unlocked"
  45         self.cleartext_device = "/dev/mapper/" + self.cleartext_name
  46         self.service_proxy = Gio.DBusProxy.new_sync(
  47             Gio.bus_get_sync(Gio.BusType.SYSTEM, None),
  48             Gio.DBusProxyFlags.NONE,
  49             None,
  50             BUS_NAME,
  51             OBJECT_PATH,
  52             INTERFACE_NAME,
  53             None,
  54         )  # type: Gio.DBusProxy
  55         device_variant = self.service_proxy.get_cached_property("Device")  # type: GLib.Variant
  56         self.device = device_variant.get_string() if device_variant else "/"
  57         self.is_unlocked = False
  58         self.is_created = self.service_proxy.get_cached_property("IsCreated")
  59         self.can_unlock = self.service_proxy.get_cached_property("CanUnlock")
  60         self.is_upgraded = self.service_proxy.get_cached_property("IsUpgraded")
  61         self.error: GLib.Variant = self.service_proxy.get_cached_property("Error")
  62         self.error_type: InvalidBootDeviceErrorType | None = None
  63         if self.error:
  64             self.error_type = InvalidBootDeviceErrorType(self.error.get_uint32())
  65         self.service_proxy.connect("g-properties-changed", self.on_properties_changed)
  66
  67     def on_properties_changed(
  68         self,
  69         proxy: Gio.DBusProxy,
  70         changed_properties: GLib.Variant,
  71         invalidated_properties: list[str],
  72     ):
  73         """Callback for when the Persistent Storage properties change"""
  74         logging.debug("changed properties: %s", changed_properties)
  75         keys = set(changed_properties.keys())
  76
  77         if "CanUnlock" in keys:
  78             self.can_unlock = changed_properties["CanUnlock"]
  79         if "Error" in keys:
  80             self.error = changed_properties["Error"]
  81             if self.error:
  82                 self.error_type = InvalidBootDeviceErrorType(self.error.get_uint32())
  83         if "IsCreated" in keys:
  84             self.is_created = changed_properties["IsCreated"]
  85         if "IsUpgraded" in keys:
  86             self.is_upgraded = changed_properties["IsUpgraded"]
  87         if "Device" in keys:
  88             self.device = changed_properties["Device"]
  89
  90     def unlock(self, passphrase: str, forceful_fsck: bool = False):
  91         """Unlock the Persistent Storage partition
  92
  93         Raises:
  94             * WrongPassphraseError if the passphrase is incorrect.
  95             * FilesystemErrorsLeftUncorrectedError if `e2fsck -f -p`
  96               found some errors that it could not correct.
  97             * PersistentStorageError if something else went wrong."""
  98         logging.debug("Unlocking Persistent Storage")
  99
 100         try:
 101             self.service_proxy.call_sync(
 102                 method_name="Unlock",
 103                 parameters=GLib.Variant("(sb)", (passphrase, forceful_fsck)),
 104                 flags=Gio.DBusCallFlags.NONE,
 105                 # In some cases, the default timeout of 25 seconds was not
 106                 # enough, especially since we now run fsck as part of the unlock
 107                 # operation, so we use a larger timeout instead.
 108                 timeout_msec=480000,
 109             )
 110         except GLib.GError as err:
 111             if tps_errors.IncorrectPassphraseError.is_instance(err):
 112                 raise tailsgreeter.errors.WrongPassphraseError from err
 113
 114             # All of the following errors are considered unrecoverable
 115             # (so we don't ask the user to unlock before starting Tails)
 116             self.failed_with_unrecoverable_error = True
 117
 118             if tps_errors.IOErrorsDetectedError.is_instance(err):
 119                 raise tailsgreeter.errors.IOErrorsDetectedError from err
 120
 121             if tps_errors.FilesystemErrorsLeftUncorrectedError.is_instance(err):
 122                 raise tailsgreeter.errors.FilesystemErrorsLeftUncorrectedError from err
 123
 124             raise tailsgreeter.errors.PersistentStorageError(
 125                 _("Error unlocking Persistent Storage: {}").format(err)
 126             ) from err
 127         self.is_unlocked = True
 128
 129     def upgrade_luks(self, passphrase):
 130         """Upgrade the Persistent Storage to the latest format
 131
 132         Raises:
 133             WrongPassphraseError if the passphrase is incorrect.
 134             PersistentStorageError if something else went wrong."""
 135         logging.debug("Upgrading Persistent Storage")
 136         try:
 137             self.service_proxy.call_sync(
 138                 method_name="UpgradeLUKS",
 139                 parameters=GLib.Variant("(s)", (passphrase,)),
 140                 flags=Gio.DBusCallFlags.NONE,
 141                 # GLib.MAXINT (largest 32-bit signed integer) disables
 142                 # the timeout
 143                 timeout_msec=GLib.MAXINT,
 144             )
 145         except GLib.GError as err:
 146             if tps_errors.IncorrectPassphraseError.is_instance(err):
 147                 raise tailsgreeter.errors.WrongPassphraseError from err
 148             self.failed_with_unrecoverable_error = True
 149             raise tailsgreeter.errors.PersistentStorageError(
 150                 _("Error upgrading Persistent Storage: {}").format(err)
 151             ) from err
 152
 153     def activate_persistent_storage(self):
 154         """Activate the already unlocked Persistent Storage"""
 155         try:
 156             self.service_proxy.call_sync(
 157                 method_name="Activate",
 158                 parameters=None,
 159                 flags=Gio.DBusCallFlags.NONE,
 160                 # We have seen this take almost 4 minutes (tails/tails#19944)
 161                 # so let's try 5 minutes for now. We don't want to disable
 162                 # the timeout completely (yet) since we still want to catch
 163                 # situations where e.g. some of our on-activation scripts
 164                 # gets in an infinite loop or similar.
 165                 timeout_msec=300000,
 166             )
 167         except GLib.GError as err:
 168             if tps_errors.FeatureActivationFailedError.is_instance(err):
 169                 tps_errors.FeatureActivationFailedError.strip_remote_error(err)
 170                 features = err.message.split(":")
 171                 # translate feature names
 172                 features = [config.gettext(feature) for feature in features]
 173                 # Translators: Don't translate {features}, it's a placeholder
 174                 # and will be replaced.
 175                 msg = config.gettext(
 176                     "Failed to activate some features of the Persistent Storage: {features}."
 177                 ).format(features=", ".join(features))
 178                 raise tailsgreeter.errors.FeatureActivationFailedError(msg) from err
 179             self.failed_with_unrecoverable_error = True
 180             raise tailsgreeter.errors.PersistentStorageError(
 181                 _("Error activating Persistent Storage: {}").format(err)
 182             ) from err
 183
 184     def abort_repair_filesystem(self):
 185         """Abort any ongoing filesystem check on the Persistent
 186         Storage.
 187
 188         Raises PersistentStorageError if something went wrong."""
 189
 190         try:
 191             self.service_proxy.call_sync(
 192                 method_name="AbortRepairFilesystem",
 193                 parameters=None,
 194                 flags=Gio.DBusCallFlags.NONE,
 195                 # GLib.MAXINT (largest 32-bit signed integer) disables
 196                 # the timeout
 197                 timeout_msec=GLib.MAXINT,
 198             )
 199         except GLib.GError as err:
 200             raise tailsgreeter.errors.PersistentStorageError(
 201                 _(
 202                     "Failed to abort when repairing Persistent Storage filesystem: {}"
 203                 ).format(err)
 204             ) from err
 205
 206     def repair_filesystem(self, finish_callback: callable) -> Gio.Cancellable:
 207         """Asynchronously start a forceful filesystem check (e2fsck -f -y)
 208         on the Persistent Storage.
 209
 210         The finish_callback is called once the operation has finished
 211         and will receive any errors raised during the operation.
 212
 213         Returns a Gio.Cancellable so the caller can abort the
 214         filesystem check by calling its cancel() method.
 215
 216         Raises PersistentStorageError if something went wrong."""
 217
 218         cancellable = Gio.Cancellable()
 219
 220         self.service_proxy.call(
 221             method_name="RepairFilesystem",
 222             parameters=None,
 223             flags=Gio.DBusCallFlags.NONE,
 224             # GLib.MAXINT (largest 32-bit signed integer) disables
 225             # the timeout
 226             timeout_msec=GLib.MAXINT,
 227             cancellable=cancellable,
 228             callback=finish_callback,
 229         )
 230
 231         return cancellable