#!/usr/bin/env python3
##
## -----------------------------------------------------------------
## This file is part of WAPT Software Deployment
## Copyright (C) 2012 - 2026 Tranquil IT https://www.tranquil.it
## All Rights Reserved.
##
## WAPT helps systems administrators to efficiently deploy
## setup, update and configure applications.
## ------------------------------------------------------------------
##
import os
import socket
import platform
import subprocess
import logging
import glob
import datetime
import re
import locale
import threading
import grp
import psutil
import shutil
import tempfile
import configparser
import shlex
from setuphelpers_unix import host_info_common_unix,dmi_info_common_unix
from waptutils import isfile,fixed_columns_to_dicts,ensure_unicode,run
from contextlib import contextmanager
LOCALE_LOCK = threading.Lock()
@contextmanager
def setlocale(name):
with LOCALE_LOCK:
saved = locale.setlocale(locale.LC_ALL)
try:
yield locale.setlocale(locale.LC_ALL, name)
finally:
locale.setlocale(locale.LC_ALL, saved)
try:
import distro
linux_distribution = list(distro.linux_distribution(False))
linux_distribution[2] = distro.codename()
linux_distribution = tuple(linux_distribution)
except ImportError:
linux_distribution = (None, None, None)
logger = logging.getLogger('waptcore')
[docs]def is_linux64():
return platform.machine().endswith('64')
isLinux64 = is_linux64
[docs]def get_distrib_version():
return linux_distribution[1]
[docs]def get_os_version():
return get_distrib_version()
[docs]def get_distrib_linux():
return linux_distribution[0]
[docs]def get_code_name_version():
return linux_distribution[2]
[docs]def get_hostname():
try:
return socket.getfqdn().lower()
except:
return ""
[docs]def type_debian():
return isfile('/etc/debian_version')
is_debian_based=type_debian
[docs]def is_debian():
return linux_distribution is not None and linux_distribution[0].lower() == 'debian'
[docs]def get_debian_version():
if isfile('/etc/debian_version'):
with open('/etc/debian_version') as f:
debian_version = f.read().strip()
return debian_version
[docs]def type_redhat():
return isfile('/etc/redhat-release')
def type_rhel():
return linux_distribution is not None and linux_distribution[0].lower() in ('rhel', 'centos', 'oracle','almalinux','rocky','fedora')
is_rhel_based=type_rhel
is_redhat_based=type_redhat
[docs]def local_users():
result = []
with open('/etc/passwd') as f:
for u in [entry.split(':',1) for entry in f.read().splitlines()]:
if not u[0]:
continue
result.append(u[0])
return result
[docs]def host_info():
info = host_info_common_unix()
try:
if dmi_info:
dmi = dmi_info()
info['system_manufacturer'] = dmi.get('System',{}).get('Manufacturer')
info['system_productname'] = dmi.get('System',{}).get('ProductName')
info['serial_nr'] = dmi.get('System',{}).get('Serial')
except Exception as e:
logger.warning('Error while getting dmi info: %s' % e)
pass
info['os_release_name'] = get_code_name_version().lower()
if info['os_release_name'] == 'core':
info['os_release_name'] = get_distrib_linux() + ' ' + get_distrib_version()
info['platform'] = platform.system()
info['os_name'] = get_distrib_linux()
info['os_version'] = get_distrib_version()
if is_debian():
info['debian_version'] = get_debian_version()
if type_debian():
info['reboot_needed'] = isfile('/var/run/reboot-required')
info['reboot-required.pkgs'] = []
if isfile('/var/run/reboot-required.pkgs'):
with open('/var/run/reboot-required.pkgs','r') as f:
info['reboot-required.pkgs'] = f.read().split('\n')
info['linux64'] = isLinux64()
with open('/etc/group') as f:
info['local_groups'] = {group.split(':')[0]:[m for m in group.split(':')[3].split(',') if m != '' ] for group in f.read().split('\n') if ':' in group and group.split(':')[0] != ''}
info['local_users'] = []
with open('/etc/passwd') as f:
for u in [entry.split(':') for entry in f.read().split('\n')]:
if not u[0]:
continue
info['local_users'].append(u[0])
gr_struct = grp.getgrgid(int(u[3]))
if gr_struct.gr_name in info['local_groups']:
if u[0] not in info['local_groups'][gr_struct.gr_name]:
info['local_groups'][gr_struct.gr_name].append(u[0])
return info
[docs]def dmi_info():
return dmi_info_common_unix()
def listcontrol_to_dict(cmd=None, first_entry=None):
if os.path.isfile(cmd):
with open(cmd,'r') as f :
list_package = '\n' + f.read()
else:
list_package = '\n' + run(cmd)
list_control=[]
for i in list_package.split(first_entry):
if not i.strip('\n'):
continue
list_control.append(first_entry.replace('\n','') + i)
list_package = []
for pkg in list_control:
entry = {}
gpg = False
for l in pkg.split('\n'):
if '-----BEGIN PGP PUBLIC KEY BLOCK-----' in l:
gpg = True
continue
if '-----END PGP PUBLIC KEY BLOCK-----' in l:
gpg = False
continue
if gpg :
continue
if not l:
continue
if (':' in l) and (not '::' in l) and (not l.startswith(' ')) :
key = l.split(':',1)[0].strip().lower()
value = l.split(':',1)[1].strip()
entry[key] = value.strip('\n')
else:
entry[key] = entry[key].strip('\n') + '\n' + l
list_package.append(entry)
return list_package
[docs]def installed_softwares(keywords=None, name=None, ignore_empty_names=True):
""" Return list of installed software from apt or rpm
Args:
keywords (str or list): string to lookup in key, display_name or publisher fields
Returns:
list of dicts: [{'key', 'name', 'version', 'install_date', 'install_location'
'uninstall_string', 'publisher','system_component'}]
"""
name_re = re.compile(name) if name is not None else None
list_installed_softwares = []
if isinstance(keywords, str):
keywords = keywords.lower().split()
elif isinstance(keywords, bytes):
keywords = str(keywords).lower().split()
elif keywords is not None:
keywords = [ensure_unicode(k).lower() for k in keywords]
else:
keywords = None
def check_words(target, words):
mywords = target.lower()
result = not words or mywords
for w in words:
result = result and w in mywords
return result
if type_debian():
try:
r = listcontrol_to_dict('dpkg-query -s','\nPackage: ')
except:
r = listcontrol_to_dict('/var/lib/dpkg/status','\nPackage: ')
dict_info_test = {}
for fileinfotest in glob.glob("/var/lib/dpkg/info/*.list"):
dict_info_test[fileinfotest.rsplit(':',1)[0] + '.list'] = fileinfotest
for pkg in r:
path_dpkg_info = "/var/lib/dpkg/info/"
if pkg['status'] in ["install ok installed","hold ok installed"] and (not ignore_empty_names or pkg.get('package', '') != '') and (
(name_re is None or name_re.match(pkg['package'])) and
(keywords is None or check_words(' '.join([pkg['package'], pkg['version'] , pkg.get('homepage', '') ]), keywords))):
path_dpkg_info_full = path_dpkg_info + pkg['package'] +'.list'
if not os.path.isfile(path_dpkg_info_full):
if path_dpkg_info_full in dict_info_test:
path_dpkg_info_full = dict_info_test[path_dpkg_info_full]
try:
install_date = os.path.getctime(path_dpkg_info_full)
install_date = datetime.datetime.fromtimestamp(install_date).strftime('%Y-%m-%d %H:%M:%S')
except:
install_date = ''
list_installed_softwares.append({
'key': pkg['package'],
'name': pkg['package'],
'version': pkg['version'],
'install_date': install_date,
'install_location': '',
'uninstall_string': '',
'publisher': pkg.get('homepage',''),
'system_component': ''})
elif type_redhat():
for header in listcontrol_to_dict('LANG="en_US.UTF-8" rpm -qai','Name : '):
if (not ignore_empty_names or header.get('name', '') != '') and (
(name_re is None or name_re.match(ensure_unicode(header['name']))) and
(keywords is None or check_words(' '.join([ensure_unicode(header['name']), ensure_unicode(header.get('url', ''))]), keywords))):
with setlocale('C'):
try:
dateformat = datetime.datetime.strptime(header['install date'],'%a %b %d %H:%M:%S %Y').strftime('%Y-%m-%d %H:%M:%S')
except ValueError:
dateformat = datetime.datetime.strptime(header['install date'].rsplit(' ',1)[0],'%a %d %b %Y %H:%M:%S %p').strftime('%Y-%m-%d %H:%M:%S')
list_installed_softwares.append({
'key': '%s_%s' % (ensure_unicode(header['name']),ensure_unicode(header['version'])),
'name': ensure_unicode(header['name']),
'version': ensure_unicode(header['version']),
'install_date': dateformat,
'install_location': '',
'uninstall_string': '',
'publisher': ensure_unicode(header.get('url','')),
'system_component': ''})
else:
list_installed_softwares.append({'key': 'Distribution not supported yet', 'name': 'Distribution not supported yet', 'version': 'Distribution not supported yet', 'install_date': 'Distribution not supported yet',
'install_location': 'Distribution not supported yet', 'uninstall_string': 'Distribution not supported yet', 'publisher': 'Distribution not supported yet', 'system_component': 'Distribution not supported yet'})
try:
for snap in get_snap_softwares():
if (not ignore_empty_names or snap['name'] != '') and (
(name_re is None or name_re.match(snap['name'])) and
(keywords is None or check_words(snap['name'], keywords))):
list_installed_softwares.append(snap)
except Exception as e:
logger.warning('Error with snap list %s' % e)
pass
try:
for appimage in get_appimage_softwares():
if (not ignore_empty_names or appimage['name'] != '') and (
(name_re is None or name_re.match(appimage['name'])) and
(keywords is None or check_words(' '.join([appimage['key'], appimage['name'], appimage['install_location']]), keywords))):
list_installed_softwares.append(appimage)
except Exception as e:
logger.warning('Error with AppImage list %s' % e)
return list_installed_softwares
def get_snap_softwares():
list_installed_softwares=[]
cmd = "snap list --color=never --unicode=always"
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, shell=True, universal_newlines=True)
stdout, stderr = proc.communicate()
retcode = proc.returncode
if retcode == 0:
i=0
for line in stdout.split("\n"):
# Skip header
if i==0:
i+=1
continue
else:
i+=1
package=line.split()
if package != []:
snap = {}
snap['name'] = package[0]
snap['version'] = package[1]
snap['revision'] = package[2]
snap['tracking'] = package[3]
snap['publisher'] = 'snap/' + package[4]
snap['notes'] = package[5]
if os.path.isfile(os.path.join("/var/lib/snapd/snaps/", package[0] + "_" + package[2] + ".snap")):
snap['install_path'] = os.path.join("/var/lib/snapd/snaps/", package[0] + "_" + package[2] + ".snap")
snap['install_date'] = os.path.getctime(snap['install_path'])
snap['install_date'] = datetime.datetime.fromtimestamp(snap['install_date']).strftime('%Y-%m-%d %H:%M:%S')
else:
snap['install_path'] = ''
snap['install_date'] = ''
list_installed_softwares.append({
'key':'snap_' + snap['name'],
'name':snap['name'],
'version':snap['version']+"-"+snap['revision'],
'install_date':snap['install_date'],
'install_location':snap['install_path'],
'uninstall_string':"snap",
'publisher':snap['publisher'],
'system_component':''})
return list_installed_softwares
[docs]def uninstall_key_exists(uninstallkey):
uninstallkey = uninstallkey.lower()
for software in installed_softwares():
if software['name'].lower()==uninstallkey:
return True
return False
[docs]def install_apt(package, allow_unauthenticated=False, force_confold=True, force_confnew=False):
"""
Install package from APT repositories
"""
update_apt()
dpkg_opts = ['-o Dpkg::Options::="--force-confdef"']
if force_confold:
dpkg_opts.append('-o Dpkg::Options::="--force-confold"')
elif force_confnew:
dpkg_opts.append('-o Dpkg::Options::="--force-confnew"')
cmd_parts = [
'LANG=C',
'DEBIAN_FRONTEND=noninteractive',
'apt-get install -y',
*dpkg_opts
]
if allow_unauthenticated:
cmd_parts.append('--allow-unauthenticated')
cmd_parts.append(package)
return run(" ".join(cmd_parts))
[docs]def uninstall_apt(package, autoremove=False):
"""
Remove package with APT
"""
if autoremove:
return run('LANG=C DEBIAN_FRONTEND=noninteractive apt-get remove --autoremove -y %s' % package)
else:
return run('LANG=C DEBIAN_FRONTEND=noninteractive apt-get remove -y %s' % package)
[docs]def install_deb(paths):
"""
Install one or multiple .deb packages using apt.
Accepts either a single file path (string) or a list of file paths.
Automatically normalizes paths so apt treats them as local packages.
"""
if isinstance(paths, str):
paths = [paths]
else:
paths = list(paths)
def normalize_path(p):
# Ensure apt recognizes the path as a local file
if p.startswith(("/", "./")):
return p
return "./" + p
deb_paths = [normalize_path(p) for p in paths]
return install_apt(" ".join(deb_paths))
[docs]def install_package_if_needed(package_name):
if not uninstall_key_exists(package_name):
if type_debian():
return install_apt(package_name)
if type_redhat():
return install_yum(package_name)
else:
raise Exception('not supported')
[docs]def purge_deb(deb_name):
return run('LANG=C DEBIAN_FRONTEND=noninteractive dpkg --purge %s' % deb_name)
[docs]def install_required_dependencies_apt():
return run('LANG=C DEBIAN_FRONTEND=noninteractive apt-get -f -y install')
[docs]def autoremove_apt():
return run('LANG=C DEBIAN_FRONTEND=noninteractive apt-get -y autoremove')
[docs]def install_yum(package):
return run('LANG=C yum install -y %s' % (package))
def install_dnf(package):
return install_yum(package)
[docs]def uninstall_yum(package):
return run('LANG=C yum remove -y %s' % package)
def uninstall_dnf(package):
return uninstall_yum(package)
[docs]def autoremove_yum():
return run('LANG=C yum autoremove -y')
def autoremove_dnf():
return autoremove_yum()
[docs]def update_apt():
return run('LANG=C DEBIAN_FRONTEND=noninteractive apt-get -y update')
[docs]def upgrade_apt(force_confold=True, force_confnew=False):
dpkg_opts = ['-o Dpkg::Options::="--force-confdef"']
if force_confold:
dpkg_opts.append('-o Dpkg::Options::="--force-confold"')
elif force_confnew:
dpkg_opts.append('-o Dpkg::Options::="--force-confnew"')
cmd_parts = [
'LANG=C',
'DEBIAN_FRONTEND=noninteractive',
'apt-get upgrade -y',
*dpkg_opts
]
return run(" ".join(cmd_parts))
[docs]def update_yum():
return run('LANG=C yum update -y')
def update_dnf():
return update_yum()
[docs]def upgrade_yum():
return run('LANG=C yum upgrade -y')
def upgrade_dnf():
return upgrade_yum()
[docs]def install_rpm(package):
return run('LANG=C yum localinstall -y %s' % (package))
[docs]def systemd_start_service(servicename):
return run('LANG=C systemctl start %s' % (servicename))
[docs]def systemd_stop_service(servicename):
return run('LANG=C systemctl stop %s' % (servicename))
[docs]def systemd_restart_service(servicename):
return run('LANG=C systemctl restart %s' % (servicename))
[docs]def systemd_status_service(servicename):
return run('LANG=C systemctl status %s' % (servicename))
[docs]def systemd_enable_start_service(servicename):
return run('LANG=C systemctl enable --now %s' % (servicename))
[docs]def systemd_enable_service(servicename):
return run('LANG=C systemctl enable %s' % (servicename))
[docs]def systemd_disable_service(servicename):
return run('LANG=C systemctl disable %s' % (servicename))
[docs]def systemd_is_running_service(servicename):
return run('LANG=C systemctl is-active --quiet %s' % (servicename), accept_returncodes=[0, 3, 4]).returncode == 0
[docs]def systemd_daemon_reload():
return run('LANG=C systemctl daemon-reload')
[docs]def service_list():
result = {}
systemd_list =run('LANG=C systemctl list-units')
units = fixed_columns_to_dicts(systemd_list)
for unit in units:
if unit['UNIT'].endswith('.service'):
result[unit['UNIT']] = unit
return result
[docs]def running_on_ac():
try:
t = psutil.sensors_battery()
if not t:
return True
return t.power_plugged
except:
return
def _read_desktop_file(desktop_file):
desktop = configparser.RawConfigParser(delimiters=('=',), strict=False)
desktop.optionxform = str
desktop.read(desktop_file, encoding='utf-8')
return desktop
def _find_appimage_icon(squashfs_dir, icon_name):
for name in (glob.escape(icon_name), '*'):
for ext in ('svg', 'png'):
icons = glob.glob(os.path.join(squashfs_dir, '%s.%s' % (name, ext)))
if icons:
return icons[0]
return None
[docs]def install_appimage(appimage, install_dir, binaliasname=None, version=None):
"""Install an AppImage in install_dir and add its .desktop launchers to /usr/share/applications
Any previous install in install_dir is removed first.
The installed AppImage is then listed by installed_softwares().
Args:
appimage (str): path to the .AppImage file
install_dir (str): installation directory
binaliasname (str): name of an optional symlink to the AppImage created in /usr/bin
version (str): version reported by installed_softwares(), stored as X-AppImage-Version in the .desktop if the AppImage does not define it
>>> install_appimage(glob.glob('*.AppImage')[0], '/opt/kate', version=control.get_software_version())
"""
appimage = os.path.abspath(appimage)
appimage_dst = os.path.join(install_dir, os.path.basename(appimage))
if os.path.isdir(install_dir):
uninstall_appimage(install_dir, binaliasname)
os.makedirs(install_dir)
shutil.copy2(appimage, appimage_dst)
os.chmod(appimage_dst, 0o755)
extract_dir = tempfile.mkdtemp()
try:
run('"%s" --appimage-extract' % appimage_dst, cwd=extract_dir)
squashfs_dir = os.path.join(extract_dir, 'squashfs-root')
for desktop_src in glob.glob(os.path.join(squashfs_dir, '*.desktop')):
desktop = _read_desktop_file(desktop_src)
# point the launchers to the installed AppImage, keeping their arguments (ex: %U)
for section in desktop.sections():
if desktop.has_option(section, 'Exec'):
args = [appimage_dst] + shlex.split(desktop.get(section, 'Exec'))[1:]
desktop.set(section, 'Exec', shlex.join(args))
if desktop.has_option('Desktop Entry', 'TryExec'):
desktop.set('Desktop Entry', 'TryExec', appimage_dst)
if version and not desktop.has_option('Desktop Entry', 'X-AppImage-Version'):
desktop.set('Desktop Entry', 'X-AppImage-Version', version)
if desktop.has_option('Desktop Entry', 'Icon'):
icon_src = _find_appimage_icon(squashfs_dir, desktop.get('Desktop Entry', 'Icon'))
if icon_src:
icon_dst = os.path.join(install_dir, os.path.basename(icon_src))
shutil.copyfile(icon_src, icon_dst)
desktop.set('Desktop Entry', 'Icon', icon_dst)
desktop_dst = os.path.join(install_dir, os.path.basename(desktop_src))
with open(desktop_dst, 'w', encoding='utf-8') as f:
desktop.write(f, space_around_delimiters=False)
menu_desktop = os.path.join('/usr/share/applications', os.path.basename(desktop_dst))
if os.path.lexists(menu_desktop):
os.remove(menu_desktop)
os.symlink(desktop_dst, menu_desktop)
finally:
shutil.rmtree(extract_dir, ignore_errors=True)
if binaliasname:
bin_alias = os.path.join('/usr/bin', binaliasname)
if os.path.lexists(bin_alias):
os.remove(bin_alias)
os.symlink(appimage_dst, bin_alias)
[docs]def uninstall_appimage(install_dir, binaliasname=None):
"""Remove an AppImage installed with install_appimage, its .desktop launchers and its /usr/bin symlink
Args:
install_dir (str): installation directory
binaliasname (str): name of the symlink to remove from /usr/bin
>>> uninstall_appimage('/opt/kate')
"""
install_dir = os.path.realpath(install_dir)
# only remove links pointing to install_dir, never files installed by something else
links = glob.glob(os.path.join('/usr/share/applications', '*.desktop'))
if binaliasname:
links.append(os.path.join('/usr/bin', binaliasname))
for link in links:
if os.path.islink(link) and os.path.realpath(link).startswith(install_dir + os.sep):
os.remove(link)
if os.path.isdir(install_dir):
shutil.rmtree(install_dir)
[docs]def get_appimage_softwares():
"""Return the AppImages installed like install_appimage does
They are found from the /usr/share/applications links to a .desktop whose Exec is an .AppImage.
"""
list_installed_softwares = []
install_locations = []
for menu_desktop in glob.glob(os.path.join('/usr/share/applications', '*.desktop')):
if not os.path.islink(menu_desktop):
continue
try:
desktop = _read_desktop_file(os.path.realpath(menu_desktop))
exec_args = shlex.split(desktop.get('Desktop Entry', 'Exec', fallback=''))
except (configparser.Error, ValueError) as e:
logger.warning('Error reading %s: %s' % (menu_desktop, e))
continue
appimage = exec_args[0] if exec_args else ''
if not appimage.lower().endswith('.appimage') or not os.path.isfile(appimage):
continue
install_location = os.path.dirname(appimage)
# an AppImage can have several .desktop launchers
if install_location in install_locations:
continue
install_locations.append(install_location)
install_date = datetime.datetime.fromtimestamp(os.path.getctime(appimage)).isoformat(sep=' ', timespec='seconds')
version = desktop.get('Desktop Entry', 'X-AppImage-Version', fallback='')
list_installed_softwares.append(
{
'key': 'appimage_' + os.path.basename(install_location),
'name': desktop.get('Desktop Entry', 'Name', fallback=os.path.basename(install_location)),
'version': version,
'install_date': install_date,
'install_location': install_location,
'uninstall_string': 'appimage',
'publisher': '',
'system_component': ''
}
)
return list_installed_softwares