mirror of
https://github.com/freedombox/FreedomBox.git
synced 2026-01-21 07:55:00 +00:00
46f162d093
Also maintain a global list of apps Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org> Reviewed-by: James Valleroy <jvalleroy@mailbox.org>
309 lines
9.8 KiB
Python
309 lines
9.8 KiB
Python
#
|
|
# This file is part of FreedomBox.
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as
|
|
# published by the Free Software Foundation, either version 3 of the
|
|
# License, or (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
#
|
|
"""
|
|
FreedomBox app to manage storage.
|
|
"""
|
|
import logging
|
|
import subprocess
|
|
|
|
import psutil
|
|
from django.utils.translation import ugettext_lazy as _
|
|
|
|
from plinth import action_utils, actions
|
|
from plinth import app as app_module
|
|
from plinth import cfg, menu
|
|
from plinth import service as service_module
|
|
from plinth import utils
|
|
from plinth.errors import PlinthError
|
|
from plinth.utils import format_lazy, import_from_gi
|
|
|
|
version = 3
|
|
|
|
name = _('Storage')
|
|
|
|
managed_services = ['freedombox-udiskie']
|
|
|
|
managed_packages = ['parted', 'udiskie', 'gir1.2-udisks-2.0']
|
|
|
|
description = [
|
|
format_lazy(
|
|
_('This module allows you to manage storage media attached to your '
|
|
'{box_name}. You can view the storage media currently in use, mount '
|
|
'and unmount removable media, expand the root partition etc.'),
|
|
box_name=_(cfg.box_name))
|
|
]
|
|
|
|
service = None
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
manual_page = 'Storage'
|
|
|
|
is_essential = True
|
|
|
|
app = None
|
|
|
|
|
|
class StorageApp(app_module.App):
|
|
"""FreedomBox app for storage."""
|
|
|
|
app_id = 'storage'
|
|
|
|
def __init__(self):
|
|
"""Create components for the app."""
|
|
super().__init__()
|
|
menu_item = menu.Menu('menu-storage', name, None, 'fa-hdd-o',
|
|
'storage:index', parent_url_name='system')
|
|
self.add(menu_item)
|
|
|
|
|
|
def init():
|
|
"""Intialize the module."""
|
|
global app
|
|
app = StorageApp()
|
|
app.set_enabled(True)
|
|
|
|
|
|
def get_disks():
|
|
"""Returns list of disks by combining information from df and lsblk."""
|
|
disks = _get_disks_from_udisks()
|
|
disks_from_df = _get_disks_from_df()
|
|
|
|
# Add size info from df to the disks from udisks based on mount point.
|
|
for disk_from_udisks in disks:
|
|
for disk_from_df in disks_from_df:
|
|
if disk_from_udisks['mount_point'] == disk_from_df['mount_point']:
|
|
disk_from_udisks.update(disk_from_df)
|
|
|
|
return sorted(disks, key=lambda disk: disk['device'])
|
|
|
|
|
|
def _get_disks_from_udisks():
|
|
"""List devices that can be ejected."""
|
|
udisks = utils.import_from_gi('UDisks', '2.0')
|
|
client = udisks.Client.new_sync()
|
|
object_manager = client.get_object_manager()
|
|
devices = []
|
|
|
|
for obj in object_manager.get_objects():
|
|
|
|
if not obj.get_block():
|
|
continue
|
|
|
|
block = obj.get_block()
|
|
|
|
if block.props.id_usage != 'filesystem':
|
|
continue
|
|
|
|
device = {
|
|
'device': block.props.device,
|
|
'label': block.props.id_label,
|
|
'size': format_bytes(block.props.size),
|
|
'filesystem_type': block.props.id_type,
|
|
'is_removable': not block.props.hint_system
|
|
}
|
|
try:
|
|
device['mount_point'] = obj.get_filesystem().props.mount_points[0]
|
|
except Exception:
|
|
continue
|
|
devices.append(device)
|
|
|
|
return devices
|
|
|
|
|
|
def _get_disks_from_df():
|
|
"""Return the list of disks and free space available using 'df'."""
|
|
try:
|
|
output = actions.superuser_run('storage', ['usage-info'])
|
|
except subprocess.CalledProcessError as exception:
|
|
logger.exception('Error getting disk information: %s', exception)
|
|
return []
|
|
|
|
disks = []
|
|
for line in output.splitlines()[1:]:
|
|
parts = line.split(maxsplit=6)
|
|
keys = ('device', 'file_system_type', 'size', 'used', 'free',
|
|
'percent_used', 'mount_point')
|
|
disk = dict(zip(keys, parts))
|
|
disk['percent_used'] = int(disk['percent_used'].rstrip('%'))
|
|
disk['size'] = int(disk['size'])
|
|
disk['used'] = int(disk['used'])
|
|
disk['free'] = int(disk['free'])
|
|
disk['size_str'] = format_bytes(disk['size'])
|
|
disk['used_str'] = format_bytes(disk['used'])
|
|
disk['free_str'] = format_bytes(disk['free'])
|
|
disks.append(disk)
|
|
|
|
return disks
|
|
|
|
|
|
def get_filesystem_type(mount_point='/'):
|
|
"""Returns the type of the filesystem mounted at mountpoint."""
|
|
for partition in psutil.disk_partitions():
|
|
if partition.mountpoint == mount_point:
|
|
return partition.fstype
|
|
|
|
raise ValueError('No such mount point')
|
|
|
|
|
|
def get_disk_info(mount_point):
|
|
"""Get information about the free space of a drive"""
|
|
disks = get_disks()
|
|
list_root = [disk for disk in disks if disk['mount_point'] == mount_point]
|
|
if not list_root:
|
|
raise PlinthError('Mount point {} not found.'.format(mount_point))
|
|
|
|
percent_used = list_root[0]['percent_used']
|
|
free_bytes = list_root[0]['free']
|
|
free_gib = free_bytes / (1024**3)
|
|
return {
|
|
'percent_used': percent_used,
|
|
'free_bytes': free_bytes,
|
|
'free_gib': free_gib
|
|
}
|
|
|
|
|
|
def get_root_device(disks):
|
|
"""Return the root partition's device from list of partitions."""
|
|
for disk in disks:
|
|
if disk['mount_point'] == '/':
|
|
return disk['device']
|
|
return None
|
|
|
|
|
|
def is_expandable(device):
|
|
"""Return the list of partitions that can be expanded."""
|
|
if not device:
|
|
return False
|
|
|
|
try:
|
|
output = actions.superuser_run(
|
|
'storage', ['is-partition-expandable', device], log_error=False)
|
|
except actions.ActionError:
|
|
return False
|
|
|
|
return int(output.strip())
|
|
|
|
|
|
def expand_partition(device):
|
|
"""Expand a partition."""
|
|
actions.superuser_run('storage', ['expand-partition', device])
|
|
|
|
|
|
def format_bytes(size):
|
|
"""Return human readable disk size from bytes."""
|
|
if not size:
|
|
return size
|
|
|
|
if size < 1024:
|
|
return _('{disk_size:.1f} bytes').format(disk_size=size)
|
|
|
|
if size < 1024**2:
|
|
size /= 1024
|
|
return _('{disk_size:.1f} KiB').format(disk_size=size)
|
|
|
|
if size < 1024**3:
|
|
size /= 1024**2
|
|
return _('{disk_size:.1f} MiB').format(disk_size=size)
|
|
|
|
if size < 1024**4:
|
|
size /= 1024**3
|
|
return _('{disk_size:.1f} GiB').format(disk_size=size)
|
|
|
|
size /= 1024**4
|
|
return _('{disk_size:.1f} TiB').format(disk_size=size)
|
|
|
|
|
|
def get_error_message(error):
|
|
"""Return an error message given an exception."""
|
|
udisks = import_from_gi('UDisks', '2.0')
|
|
if error.matches(udisks.Error.quark(), udisks.Error.FAILED):
|
|
message = _('The operation failed.')
|
|
elif error.matches(udisks.Error.quark(), udisks.Error.CANCELLED):
|
|
message = _('The operation was cancelled.')
|
|
elif error.matches(udisks.Error.quark(), udisks.Error.ALREADY_UNMOUNTING):
|
|
message = _('The device is already unmounting.')
|
|
elif error.matches(udisks.Error.quark(), udisks.Error.NOT_SUPPORTED):
|
|
message = _('The operation is not supported due to '
|
|
'missing driver/tool support.')
|
|
elif error.matches(udisks.Error.quark(), udisks.Error.TIMED_OUT):
|
|
message = _('The operation timed out.')
|
|
elif error.matches(udisks.Error.quark(), udisks.Error.WOULD_WAKEUP):
|
|
message = _('The operation would wake up a disk that is '
|
|
'in a deep-sleep state.')
|
|
elif error.matches(udisks.Error.quark(), udisks.Error.DEVICE_BUSY):
|
|
message = _('Attempting to unmount a device that is busy.')
|
|
elif error.matches(udisks.Error.quark(), udisks.Error.ALREADY_CANCELLED):
|
|
message = _('The operation has already been cancelled.')
|
|
elif error.matches(udisks.Error.quark(), udisks.Error.NOT_AUTHORIZED) or \
|
|
error.matches(udisks.Error.quark(),
|
|
udisks.Error.NOT_AUTHORIZED_CAN_OBTAIN) or \
|
|
error.matches(udisks.Error.quark(),
|
|
udisks.Error.NOT_AUTHORIZED_DISMISSED):
|
|
message = _('Not authorized to perform the requested operation.')
|
|
elif error.matches(udisks.Error.quark(), udisks.Error.ALREADY_MOUNTED):
|
|
message = _('The device is already mounted.')
|
|
elif error.matches(udisks.Error.quark(), udisks.Error.NOT_MOUNTED):
|
|
message = _('The device is not mounted.')
|
|
elif error.matches(udisks.Error.quark(),
|
|
udisks.Error.OPTION_NOT_PERMITTED):
|
|
message = _('Not permitted to use the requested option.')
|
|
elif error.matches(udisks.Error.quark(),
|
|
udisks.Error.MOUNTED_BY_OTHER_USER):
|
|
message = _('The device is mounted by another user.')
|
|
else:
|
|
message = error.message
|
|
|
|
return message
|
|
|
|
|
|
def is_running():
|
|
"""Return whether the service is running."""
|
|
return action_utils.service_is_running('freedombox-udiskie')
|
|
|
|
|
|
def is_enabled():
|
|
"""Return whether the module is enabled."""
|
|
return action_utils.service_is_enabled('freedombox-udiskie')
|
|
|
|
|
|
def enable():
|
|
"""Enable the module."""
|
|
actions.superuser_run('udiskie', ['enable'])
|
|
app.enable()
|
|
|
|
|
|
def disable():
|
|
"""Disable the module."""
|
|
actions.superuser_run('udiskie', ['disable'])
|
|
app.disable()
|
|
|
|
|
|
def setup(helper, old_version=None):
|
|
"""Install and configure the module."""
|
|
helper.install(managed_packages, skip_recommends=True)
|
|
helper.call('post', actions.superuser_run, 'udiskie', ['enable'])
|
|
global service
|
|
if service is None:
|
|
service = service_module.Service(
|
|
managed_services[0], name, is_enabled=is_enabled, enable=enable,
|
|
disable=disable, is_running=is_running)
|
|
disks = get_disks()
|
|
root_device = get_root_device(disks)
|
|
if is_expandable(root_device):
|
|
expand_partition(root_device)
|