ogclone-engine/ogclient/lib/python3/InventoryLib.py

531 lines
22 KiB
Python

#/**
#@file InventoryLib.py
#@brief Librería o clase Inventory
#@class Inventory
#@brief Funciones para recogida de datos de inventario de hardware y software de los clientes.
#@warning License: GNU GPLv3+
#*/
import platform
import sys
import os
import subprocess
import re
import json
import shutil
import glob
import plistlib
#import bsddb
import ogGlobals
import SystemLib
import FileSystemLib
import RegistryLib
import FileLib
#/**
# ogGetArch
#@brief Devuelve el tipo de arquitectura del cliente.
#@return str_arch - Arquitectura (i386 para 32 bits, x86_64 para 64 bits).
#*/
def ogGetArch():
if os.path.isdir ('/lib64'):
return 'x86_64'
else:
return 'i386'
#/**
# ogGetOsType int_ndisk int_npartition
#@brief Devuelve el tipo del sistema operativo instalado.
#@param int_ndisk nº de orden del disco
#@param int_npartition nº de orden de la partición
#@return OSType - Tipo de sistema operativo.
#@see ogGetOsVersion
#*/ ##
def ogGetOsType(disk, partition):
try:
os_version = ogGetOsVersion(disk, partition)
if os_version:
return os_version.split(":", 1)[0]
else:
return None
except Exception as e:
print(f"Error en ogGetOsType: {e}")
return "Unknown"
#/**
# ogGetOsUuid int_ndisk int_nfilesys
#@brief Devuelve el UUID del sistema operativo instalado en un sistema de archivos.
#@param int_ndisk nº de orden del disco
#@param int_nfilesys nº de orden de la partición
#@return str_uuid - UUID del sistema operativo.
#@exception OG_ERR_FORMAT Formato incorrecto.
#@exception OG_ERR_NOTFOUND Disco o partición no corresponden con un dispositiv
#*/ ##
def ogGetOsUuid (disk, par):
mntdir = FileSystemLib.ogMount (disk, par)
if not mntdir: return None
# Obtener UUID según el tipo de sistema operativo.
os_type = ogGetOsType (disk, par)
if 'Linux' == os_type:
# Leer el UUID del sistema de ficheros raíz o el fichero de identificador.
uuid = subprocess.run (['findmnt', '-no', 'UUID', mntdir], capture_output=True, text=True).stdout.strip()
if not uuid:
uuid = open (os.path.join (mntdir, 'etc', 'machine-id')).read().strip()
return uuid
elif 'Windows' == os_type:
# Leer identificador en clave de registro.
uuid = RegistryLib.ogGetRegistryValue (mntdir, 'SOFTWARE', r'\Microsoft\Cryptography\MachineGuid')
return uuid
#/**
# ogGetSerialNumber
#@brief Obtiene el nº de serie del cliente.
#*/ ##
def ogGetSerialNumber():
SERIALNO = subprocess.check_output(["dmidecode", "-s", "system-serial-number"]).decode().strip()
SERIALNO = re.sub(r"(not specified|to be filled|invalid entry|default string)", "", SERIALNO, flags=re.IGNORECASE)
SERIALNO = SERIALNO.replace(" ", "")
SERIALNO = SERIALNO[:25] if len(SERIALNO) > 25 else SERIALNO
if SERIALNO: return SERIALNO
return None
#/**
# ogIsEfiActive
#@brief Comprueba si el sistema tiene activo el arranque EFI.
#*/ ##
def ogIsEfiActive():
return os.path.isdir("/sys/firmware/efi")
#/**
# ogListHardwareInfo
#@brief Lista el inventario de hardware de la máquina cliente.
#@return TipoDispositivo:Modelo (por determinar)
#@warning Se ignoran los parámetros de entrada.
#@note TipoDispositivo = { bio, boa, bus, cha, cdr, cpu, dis, fir, mem, mod, mul, net, sto, usb, vga }
#@note Requisitos: dmidecode, lshw, awk
#*/ ##
def ogListHardwareInfo():
ret = ''
# Ejecutar dmidecode y obtener tipo de chasis
dmi_out = subprocess.run (['dmidecode', '-s', 'chassis-type'], capture_output=True, text=True).stdout
dmi_out = '\n'.join ([ x for x in dmi_out.splitlines() if 'Other' not in x ])
ret += f'cha={dmi_out}\n'
if os.path.exists ('/sys/firmware/efi'):
ret += f'boo=UEFI\n'
else:
ret += f'boo=BIOS\n'
awk_script = r'''
BEGIN {type="mod";}
/product:/ {sub(/ *product: */,""); prod=$0;}
/vendor:/ {sub(/ *vendor: */,""); vend=$0;}
/version:/ {sub(/ *version: */,"v.");vers=$0;}
/size:/ {size=$2;}
/clock:/ {clock=$2;}
/slot:/ {sub(/ *slot: */,""); slot=$0;}
/\*-/ {if (type=="mem"){
if (size!=""){
numbank++;
print type"="vend,prod,size,clock" ("slot")";}
}else{
if (type=="totalmem"){
if (size!=""){
totalmemory="mem="size;}
}else{
if (type!="" && prod!=""){
if (prod=="v."vers)
vers="";
print type"="vend,prod,size,vers;} }
}
type=prod=vend=vers=size=clock=slot="";}
$1~/-core/ {type="boa";}
$1~/-firmware/ {type="bio";}
$1~/-cpu/ {type="cpu";}
$1~/-bank/ {type="mem";}
$1~/-memory/ {type="totalmem";}
$1~/-ide/ {type="ide";}
$1~/-storage/ {type="sto";}
$1~/-disk/ {type="dis";}
$1~/-cdrom/ {type="cdr";}
$1~/-display/ {type="vga";}
$1~/-network/ {type="net";}
$1~/-multimedia/ {type="mul";}
$1~/-usb/ {type="usb";}
$1~/-firewire/ {type="fir";}
$1~/-serial/ {type="bus";}
END {if (type!="" && prod!="")
print type"="vend,prod,size,vers;
if (length(numbank)==0 && length(totalmemory)>=4)
print totalmemory; }
'''
lshw_out = subprocess.run (['lshw'], capture_output=True, text=True).stdout
awk_out = subprocess.run (['awk', awk_script], input=lshw_out, capture_output=True, text=True).stdout
ret += awk_out
return ret
#/**
# ogListSoftware int_ndisk int_npartition
#@brief Lista el inventario de software instalado en un sistema operativo.
#@param int_ndisk nº de orden del disco
#@param int_npartition nº de orden de la partición
#@return programa versión ...
#@warning Se ignoran los parámetros de entrada.
#@note Requisitos: ...
#@todo Detectar software en Linux
#*/ ##
def ogListSoftware (disk, par):
mntdir = FileSystemLib.ogMount (disk, par)
if not mntdir: return None
ostype = ogGetOsType (disk, par)
if not ostype: return None
apps = []
if 'Linux' == ostype:
# Procesar paquetes dpkg.
pkgdir = f'{mntdir}/var/lib/dpkg'
if os.path.exists (pkgdir):
status_file = os.path.join(pkgdir, "status")
awk_script = '''
/Package:/ {if (pack!="") print pack,vers;
sub(/-dev$/,"",$2);
pack=$2}
/Version:/ {sub(/^.*:/,"",$2); sub(/-.*$/,"",$2);
vers=$2}
/Status:/ {if ($2!="install") pack=vers=""}
END {if (pack!="") print pack,vers}
'''
awk_out = subprocess.run (['awk', awk_script, status_file], capture_output=True, text=True).stdout
apps = awk_out.splitlines()
# Procesar paquetes RPM.
pkgdir = f'{mntdir}/var/lib/rpm'
if os.path.exists (pkgdir):
if shutil.which ('rpm'):
for f in glob.glob (f'{pkgdir}/__db.*'):
os.unlink (f)
rpm_out = subprocess.run (['rpm', '--dbpath', pkgdir, '-qa', '--qf', '%{NAME} %{VERSION}\n'], capture_output=True, text=True).stdout
for l in rpm_out.splitlines():
words = l.split()
if (not re.search ('-devel$', words[0])):
words[1] = re.sub ('-.*', '', words[1])
apps.append (' '.join (words))
for f in glob.glob (f'{pkgdir}/__db.*'):
os.unlink (f)
else:
pass
#db = bsddb.hashopen (f'{pkgdir}/Name', 'r');
#for k in db.keys():
# apps.append (re.sub ('-devel$', '', k))
# Procesar paquetes pacman.
pkgdir = f'{mntdir}/var/lib/pacman/local'
if os.path.exists (pkgdir):
for f in glob.glob (f'{pkgdir}/*'):
if '-' not in f: continue
idx = f[0:f.rfind ('-')].rfind ('-') ## index of 2nd-to-last dash
apps.append (f[0:idx] + ' ' + f[idx+1:])
# Procesar aplicaciones Snappy.
pkgdir = f'{mntdir}/snap'
out = ''
awk_script = '''
/name:/ {pack=$2}
/version:/ {vers=$2}
END {if (pack!="") print pack,"(snap)",vers}
'''
files = subprocess.run (['find', f'{pkgdir}/*/current/meta', '-name', 'snap.yaml'], capture_output=True, text=True).stdout.splitlines()
for f in files:
awk_out = subprocess.run (['awk', awk_script, f], capture_output=True, text=True).stdout
out += awk_out
apps += out.splitlines()
# Procesar aplicaciones Flatpak.
pkgdir = f'{mntdir}/var/lib/flatpak'
files = glob.glob (f'{pkgdir}/app/*/current/active/deploy/*')
for f in files:
p = open (f.strip()).read().split ('\0')
try:
if (p[0] != 'flathub'): raise ValueError
apps.append ('{} (flatpak) {}'.format (p[p.index('appdata-name') + 4], p[p.index('appdata-version') + 1]))
except ValueError:
pass
elif 'Windows' == ostype:
if shutil.which ('hivexregedit'):
hive = RegistryLib.ogGetHivePath (mntdir, 'software')
if hive:
cmd1_out = subprocess.run (['hivexregedit', '--unsafe-printable-strings', '--export', hive, r'\Microsoft\Windows\CurrentVersion\Uninstall'], capture_output=True, text=True).stdout
cmd1_out += subprocess.run (['hivexregedit', '--unsafe-printable-strings', '--export', hive, r'\Wow6432Node\Microsoft\Windows\CurrentVersion\Uninstall'], capture_output=True, text=True).stdout
out = name = ''
for l in cmd1_out.splitlines():
words = l.split ('"')
if len(words) < 4: continue
if (re.match (r'\[', words[0])): name=''
if (re.search ('DisplayName', words[1])): name=words[3]
if (re.search ('DisplayVersion', words[1])): apps.append (f'{name} {words[3]}')
else:
keys = RegistryLib.ogListRegistryKeys (mntdir, 'software', r'\Microsoft\Windows\CurrentVersion\Uninstall')
keys32 = RegistryLib.ogListRegistryKeys (mntdir, 'software', r'\Wow6432Node\Microsoft\Windows\CurrentVersion\Uninstall')
for k in keys:
prog = RegistryLib.ogGetRegistryValue (mntdir, 'software', rf'\Microsoft\Windows\CurrentVersion\Uninstall\{k}\DisplayName')
if prog:
vers = RegistryLib.ogGetRegistryValue (mntdir, 'software', rf'\Microsoft\Windows\CurrentVersion\Uninstall\{k}\DisplayVersion')
apps.append (f'{prog} {vers}')
for k in keys32:
prog = RegistryLib.gGetRegistryValue (mntdir, 'software', rf'\Wow6432Node\Microsoft\Windows\CurrentVersion\Uninstall\{k}\DisplayName')
if prog:
vers = RegistryLib.ogGetRegistryValue (mntdir, 'software', rf'\Wow6432Node\Microsoft\Windows\CurrentVersion\Uninstall\{k}\DisplayVersion')
apps.append (f'{prog} {vers}')
elif 'MacOS' == ostype:
files = subprocess.run (['find', f'{mntdir}/Applications', '-type', 'd', '-name', '*.app', '-prune', '-print'], capture_output=True, text=True).stdout.splitlines()
for k in files:
FILE = f'{k}/Contents/version.plist'
if not os.stat (FILE).st_size:
FILE = f'{k}/Contents/version.plist.uncompress'
if os.stat (FILE).st_size:
VERSION = subprocess.run (['awk', '-F[<>]', '/ShortVersionString/ {getline;v=$3} END {print v}', FILE], capture_output=True, text=True).stdout.strip()
bn = os.path.basename (k)
bn = re.sub ('.app$', '', bn)
apps.append (f'{bn} {VERSION}')
elif 'BSD' == ostype:
sqlite_out = subprocess.run (['sqlite3', f'{mntdir}/var/db/pkg/local.sqlite'], input='SELECT name FROM pkg_search;\n', capture_output=True, text=True).stdout
for l in sqlite_out.splitlines():
apps.append (' '.join (re.search ('(.*)-(.*)', l).groups()))
else:
SystemLib.ogRaiseError ([], ogGlobals.OG_ERR_NOTOS, f'{disk}, {par} ({ostype})')
return None
os_version = ogGetOsVersion (disk, par).split (':')[1]
return [os_version] + sorted (set (apps))
## https://stackoverflow.com/questions/2522651/find-a-key-inside-a-deeply-nested-dictionary/2522706#2522706
def _find_key_recursive(plist_dict, key_substr):
for k in plist_dict.keys():
if key_substr in k: return plist_dict[k]
for k, v in plist_dict.items():
if type(v) is dict: # Only recurse if we hit a dict value
value = _find_key_recursive(v, key_substr)
if value:
return value
return ''
#/**
# ogGetOsVersion int_ndisk int_nfilesys
#@brief Devuelve la versión del sistema operativo instalado en un sistema de archivos.
#@param int_ndisk nº de orden del disco
#@param int_nfilesys nº de orden de la partición
#@return OSType:OSVersion - tipo y versión del sistema operativo.
#@note OSType = { Android, BSD, GrubLoader, Hurd, Linux, MacOS, Solaris, Windows, WinLoader }
#@note Requisitos: awk, head, chroot
#@exception OG_ERR_FORMAT Formato incorrecto.
#@exception OG_ERR_NOTFOUND Disco o partición no corresponden con un dispositiv
#@exception OG_ERR_PARTITION Fallo al montar el sistema de archivos.
#*/ ##
#ogGetOsVersion ("1", "2") => "Linux:Ubuntu precise (12.04 LTS) 64 bits"
def ogGetOsVersion(disk, part):
mntdir = FileSystemLib.ogMount (disk, part)
if not mntdir:
return None
type = version = None
is64bit = ''
# Buscar tipo de sistema operativo.
# Para GNU/Linux: leer descripción.
os_release = os.path.join(mntdir, "etc/os-release")
if os.path.isfile(os_release):
type = 'Linux'
with open(os_release, "r") as f:
for line in f:
if line.startswith("PRETTY_NAME"):
version = line.split("=", 1)[1].strip().strip('"')
break
if not version:
lsb_release = os.path.join(mntdir, "etc/lsb-release")
if os.path.isfile(lsb_release):
type = 'Linux'
with open(lsb_release, "r") as f:
for line in f:
if line.startswith("DISTRIB_DESCRIPTION"):
version = line.split("=", 1)[1].strip().strip('"')
break
if not version:
for distrib in ["redhat", "SuSE", "mandrake", "gentoo"]:
distrib_file = os.path.join(mntdir, f"etc/{distrib}-release")
if os.path.isfile(distrib_file):
type = 'Linux'
with open(distrib_file, "r") as f:
version = f.readline().strip()
break
if not version:
arch_release_file = os.path.join(mntdir, "etc/arch-release")
if os.path.isfile(arch_release_file):
type = 'Linux'
version = "Arch Linux"
if not version:
slack_release_file = os.path.join(mntdir, "slackware-version")
if os.path.isfile(slack_release_file):
type = 'Linux'
with open (slack_release_file, 'r') as fd:
c = fd.read()
version = "Slackware {c}"
# Si no se encuentra, intentar ejecutar "lsb_release".
if not version:
out = subprocess.run (['chroot', mntdir, 'lsb_release', '-d'], capture_output=True, text=True).stdout
m = re.search (':\t(.*)', out)
if m:
type = 'Linux'
version = m.group(1)
# Comprobar Linux de 64 bits.
if version and os.path.exists(os.path.join(mntdir, "lib64")):
is64bit = ogGlobals.lang.MSG_64BIT
# Para Android, leer fichero de propiedades.
if not version:
type = 'Android'
files = glob.glob (os.path.join (mntdir, 'android*/system/build.prop'))
if files and os.path.isfile (files[0]):
v = []
with open (files[0], 'r') as f:
for line in f:
if 'product.brand' in line or 'build.version.release' in line:
v.append (line.split('=')[1].strip())
version = ' '.join (v)
if os.path.exists(os.path.join(mntdir, "lib64")):
is64bit = ogGlobals.lang.MSG_64BIT
# Para GNU/Hurd, comprobar fichero de inicio (basado en os-prober).
if not version:
type = 'Hurd'
if os.path.exists(os.path.join(mntdir, "hurd/init")):
version = 'GNU/Hurd'
# Para Windows: leer la version del registro.
if not version:
type = 'Windows'
build = 0
file = RegistryLib.ogGetHivePath (mntdir, 'SOFTWARE')
if file:
# Nuevo método más rápido para acceder al registro de Windows..
i = '\n'.join ([
f'load {file}',
r'cd \Microsoft\Windows NT\CurrentVersion',
'lsval ProductName',
'lsval DisplayVersion',
])
version = subprocess.run (['hivexsh'], input=i, capture_output=True, text=True).stdout
version = version.replace ('\n', ' ')
# Recoge el valor del número de compilación para ver si es Windows 10/11
i = '\n'.join ([
f'load {file}',
r'cd \Microsoft\Windows NT\CurrentVersion',
'lsval CurrentBuildNumber',
])
build = subprocess.run (['hivexsh'], input=i, capture_output=True, text=True).stdout
if subprocess.run (['reglookup', '-H', '-p', 'Microsoft/Windows/CurrentVersion/ProgramW6432Dir', file], capture_output=True, text=True).stdout:
is64bit = ogGlobals.lang.MSG_64BIT
if not version:
# Compatibilidad con métrodo antiguo y más lento de acceder al registro.
version = RegistryLib.ogGetRegistryValue (mntdir, 'software', r'\Microsoft\Windows NT\CurrentVersion\ProductName')
if RegistryLib.ogGetRegistryValue (mntdir, 'software', r'\Microsoft\Windows\CurrentVersion\ProgramW6432Dir'):
is64bit = ogGlobals.lang.MSG_64BIT
# Si la compilación es mayor o igual a 22000 es Windows 11
if int (build) >= 22000:
version = version.replace ('10', '11')
# Para cargador Windows: buscar versión en fichero BCD (basado en os-prober).
if not version:
type = 'WinLoader'
file = FileLib.ogGetPath (file=f'{mntdir}/boot/bcd')
if not file:
file = FileLib.ogGetPath (file=f'{mntdir}/EFI/Microsoft/boot/bcd')
if file:
for distrib in 'Windows Recovery', 'Windows Boot':
with open (file, 'rb') as fd:
contents = fd.read()
distrib_utf16_regex = re.sub (r'(.)', '\\1.', distrib)
distrib_utf16_regex = bytes (distrib_utf16_regex, 'ascii')
if re.search (distrib_utf16_regex, contents):
version = f'{distrib} loader'
# Para macOS: detectar kernel y completar con fichero plist de información del sistema.
if not version:
type = 'MacOS'
# Kernel de Mac OS (no debe ser fichero de texto).
file = f'{mntdir}/mach_kernel'
out = subprocess.run (['file', '--brief', file], capture_output=True, text=True).stdout
if not 'text' in out:
# Obtener tipo de kernel.
if 'Mach-O' in out: version = 'macOS'
if 'Mach-O 64-bit' in out: is64bit = ogGlobals.lang.MSG_64BIT
# Datos de configuración de versión de Mac OS.
file = f'{mntdir}/System/Library/CoreServices/SystemVersion.plist'
if os.path.exists (file):
with open (file, 'rb') as fd:
contents = fd.read()
plist_dict = plistlib.loads (contents)
n = _find_key_recursive (plist_dict, 'ProductName')
v = _find_key_recursive (plist_dict, 'ProductVersion')
version = f'{n} {v}'.strip()
# Datos de recuperación de macOS.
if version and os.path.exists (f'{mntdir}/com.apple.recovery.boot'):
version += ' recovery'
# Para FreeBSD: obtener datos del Kernel.
### TODO Revisar solución.
if not version:
type = 'BSD'
file = f'{mntdir}/boot/kernel/kernel'
if os.path.exists (file):
lines = subprocess.run (['strings', file], capture_output=True, text=True).stdout.splitlines()
release_search = list (filter (lambda x: re.search ('@.*RELEASE', x), lines))
if release_search:
first, second, *rest = release_search[0].split()
first = first.replace ('@(#)', '')
version = f'{first} {second}'
out = subprocess.run (['file', '--brief', file], capture_output=True, text=True).stdout
if 'x86-64' in out: is64bit = ogGlobals.lang.MSG_64BIT
# Para Solaris: leer el fichero de versión.
### TODO Revisar solución.
if not version:
type = 'Solaris'
file = f'{mntdir}/etc/release'
if os.path.exists (file):
with open (file, 'r') as fd:
version = fd.readline().strip
# Para cargador GRUB, comprobar fichero de configuración.
if not version:
type = 'GrubLoader'
for file in f'{mntdir}/grub/menu.lst', f'{mntdir}/boot/grub/menu.lst':
if os.path.exists (file):
VERSION = 'GRUB Loader'
for entry in f'{mntdir}/grub/grub.cfg', f'{mntdir}/grub2/grub.cfg', f'{mntdir}/EFI/*/grub.cfg', f'{mntdir}/boot/grub/grub.cfg', f'{mntdir}/boot/grub2/grub.cfg', f'{mntdir}/boot/EFI/*/grub.cfg':
for file in glob.glob (entry):
if os.path.exists (file):
version = 'GRUB2 Loader'
# Mostrar resultado y salir sin errores.
if version: return f"{type}:{version} {is64bit}"
return None