particle-os/debian-forge-composer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
debian-forge-composer/tools/image-info
Tomas Hozza a2d4ba715f Extend information gathered by image-info tool 
This is a preparation for defining EC2 images for RHEL-8.5.0. These
extensions to image-info tool represent modifications done to the
official EC2 images currently produced as RHEL release. It is important
to be able to analyse these aspects of images, before we define them in
osbuild-composer, to ensure that the resulting images will be consistent
with the current state.

- Read non-empty lines from /etc/hosts file and add them to the report.
- Read content of /etc/machine-id and add it to the report.
- Read uncommented key/values from /etc/systemd/logind.conf and add them
  to the report.
- Read all ifcfg-* files from /etc/sysconfig/network-scripts/ and add
  their values to the report.
- Read content of /etc/locale.conf and add it to the report.
- Read SELinux configuration from /etc/selinux/config and add it to the
  report.
- Inspect the filesystem tree for SELinux context mismatches and add
  them to the report.
- Read configuration files from /etc/modprobe.d/ and for now report only
  all blacklisted kernel modules.
- Read RHSM configuration from /etc/rhsm/rhsm.conf and add it to the
  report.
- Read cloud-init configuration from /etc/cloud/cloud.conf and add it to
  the report.
- Read all *.conf files from /etc/dracut.conf.d/ and add their content to
  the report.
- Read VC and X11 keyboard configuration and add it to the report.
- Read specific configuration directives from Chrony configuration and
  add them to the report. Specifically 'server', 'pool', 'peer' and
  'leapsectz'.
- Read drop-in configurations for *.service unit files from
  /etc/systemd/system/ and add them to the report.
- Read all configuration files from /etc/tmpfiles.d/ and add them to the
  report.
- Read all configuration files from /etc/sysctl.d/ and add them to the
  report.
- Read the Tuned active profile and profile mode and add them to the
  report.
- Read all configuration files from /etc/security/limits.d and add them
  to the report.
- Read sudoers configuration from /etc/sudoers and files in
  /etc/sudoers.d/ and add uncommented lines to the report. No
  sophisticated parsing is done, because the configuration format
  grammar is too complicated for the purpose of image-info.
- Read udev rules configuration files from /etc/udev/rules.d/ and add
  them to the report.
- Read DNF configuration and defined vars and add them to the report.
- Read profile ID and enabled features used by authselect.
- Enable SELinux, extended attributes and POSIX ACLs support when
  unpacking 'tar' image type to prevent potential mismatches

Regenerate all image test cases to reflect changes in the image-info
output. Modify the distro-arch-imagetype-map.json to cover all
combinations currently covered by existing image test cases.

Add doc strings to all read_* functions.

Signed-off-by: Tomas Hozza <thozza@redhat.com>
2021-06-09 18:23:00 +02:00

1982 lines
62 KiB
Python
Executable file
Raw Blame History

#!/usr/bin/python3
import argparse
import configparser
import contextlib
import errno
import functools
import glob
import mimetypes
import json
import os
import platform
import re
import subprocess
import sys
import tempfile
import xml.etree.ElementTree
import yaml
from osbuild import loop
def run_ostree(*args, _input=None, _check=True, **kwargs):
args = list(args) + [f'--{k}={v}' for k, v in kwargs.items()]
print("ostree " + " ".join(args), file=sys.stderr)
res = subprocess.run(["ostree"] + args,
encoding="utf-8",
stdout=subprocess.PIPE,
input=_input,
check=_check)
return res
@contextlib.contextmanager
def loop_create_device(ctl, fd, offset=None, sizelimit=None):
while True:
lo = loop.Loop(ctl.get_unbound())
try:
lo.set_fd(fd)
except OSError as e:
lo.close()
if e.errno == errno.EBUSY:
continue
raise e
try:
lo.set_status(offset=offset, sizelimit=sizelimit, autoclear=True)
except BlockingIOError:
lo.clear_fd()
lo.close()
continue
break
try:
yield lo
finally:
lo.close()
@contextlib.contextmanager
def loop_open(ctl, image, *, offset=None, size=None):
with open(image, "rb") as f:
fd = f.fileno()
with loop_create_device(ctl, fd, offset=offset, sizelimit=size) as lo:
yield os.path.join("/dev", lo.devname)
@contextlib.contextmanager
def open_image(ctl, image, fmt):
with tempfile.TemporaryDirectory(dir="/var/tmp") as tmp:
if fmt["type"] != "raw":
target = os.path.join(tmp, "image.raw")
# A bug exists in qemu that causes the conversion to raw to fail
# on aarch64 systems with a LOT of CPUs. A workaround is to use
# a single coroutine to do the conversion. It doesn't slow down
# the conversion by much, but it hangs about half the time without
# the limit set. 😢
# Bug: https://bugs.launchpad.net/qemu/+bug/1805256
if platform.machine() == 'aarch64':
subprocess.run(
["qemu-img", "convert", "-m", "1", "-O", "raw", image, target],
check=True
)
else:
subprocess.run(
["qemu-img", "convert", "-O", "raw", image, target],
check=True
)
else:
target = image
size = os.stat(target).st_size
with loop_open(ctl, target, offset=0, size=size) as dev:
yield target, dev
@contextlib.contextmanager
def mount_at(device, mountpoint, options=[], extra=[]):
opts = ",".join(["ro"] + options)
subprocess.run(["mount", "-o", opts] + extra + [device, mountpoint], check=True)
try:
yield mountpoint
finally:
subprocess.run(["umount", "--lazy", mountpoint], check=True)
@contextlib.contextmanager
def mount(device):
with tempfile.TemporaryDirectory() as mountpoint:
subprocess.run(["mount", "-o", "ro", device, mountpoint], check=True)
try:
yield mountpoint
finally:
subprocess.run(["umount", "--lazy", mountpoint], check=True)
def parse_environment_vars(s):
r = {}
for line in s.split("\n"):
line = line.strip()
if not line:
continue
if line[0] == '#':
continue
key, value = line.split("=", 1)
r[key] = value.strip('"')
return r
# Parses output of `systemctl list-unit-files`
def parse_unit_files(s, expected_state):
r = []
for line in s.split("\n")[1:]:
try:
unit, state, *_ = line.split()
except ValueError:
pass
if state != expected_state:
continue
r.append(unit)
return r
def subprocess_check_output(argv, parse_fn=None):
try:
output = subprocess.check_output(argv, encoding="utf-8")
except subprocess.CalledProcessError as e:
sys.stderr.write(f"--- Output from {argv}:\n")
sys.stderr.write(e.stdout)
sys.stderr.write("\n--- End of the output\n")
raise
return parse_fn(output) if parse_fn else output
def read_image_format(device):
"""
Read image format.
Returns: dictionary with at least one key 'type'. 'type' value is a string
representing the format of the image. In case the type is 'qcow2', the returned
dictionary contains second key 'compat' with a string value representing
the compatibility version of the 'qcow2' image.
An example return value:
{
"compat": "1.1",
"type": "qcow2"
}
"""
qemu = subprocess_check_output(["qemu-img", "info", "--output=json", device], json.loads)
format = qemu["format"]
result = {"type": format}
if format == "qcow2":
result["compat"] = qemu["format-specific"]["data"]["compat"]
return result
def read_partition(device, partition):
"""
Read block device attributes using 'blkid' and extend the passed 'partition'
dictionary.
Returns: the 'partition' dictionary provided as an argument, extended with
'label', 'uuid' and 'fstype' keys and their values.
"""
res = subprocess.run(["blkid", "--output", "export", device],
check=False, encoding="utf-8",
stdout=subprocess.PIPE)
if res.returncode == 0:
blkid = parse_environment_vars(res.stdout)
else:
blkid = {}
partition["label"] = blkid.get("LABEL") # doesn't exist for mbr
partition["uuid"] = blkid.get("UUID")
partition["fstype"] = blkid.get("TYPE")
return partition
def read_partition_table(device):
"""
Read information related to found partitions and partitioning table from
the device.
Returns: dictionary with three keys - 'partition-table', 'partition-table-id'
and 'partitions'.
'partition-table' value is a string with the type of the partition table or 'None'.
'partition-table-id' value is a string with the ID of the partition table or 'None'.
'partitions' value is a list of dictionaries representing found partitions.
An example return value:
{
"partition-table": "gpt",
"partition-table-id": "DA237A6F-F0D4-47DF-BB50-007E00DB347C",
"partitions": [
{
"bootable": false,
"partuuid": "64AF1EC2-0328-406A-8F36-83016E6DD858",
"size": 1048576,
"start": 1048576,
"type": "21686148-6449-6E6F-744E-656564454649",
},
{
"bootable": false,
"partuuid": "D650D523-06F6-4B90-9204-8F998FE9703C",
"size": 6442450944,
"start": 2097152,
"type": "0FC63DAF-8483-4772-8E79-3D69D8477DE4",
}
]
}
"""
partitions = []
info = {"partition-table": None,
"partition-table-id": None,
"partitions": partitions}
try:
sfdisk = subprocess_check_output(["sfdisk", "--json", device], json.loads)
except subprocess.CalledProcessError:
partitions.append(read_partition(device, False))
return info
ptable = sfdisk["partitiontable"]
assert ptable["unit"] == "sectors"
is_dos = ptable["label"] == "dos"
ssize = ptable.get("sectorsize", 512)
for i, p in enumerate(ptable["partitions"]):
partuuid = p.get("uuid")
if not partuuid and is_dos:
# For dos/mbr partition layouts the partition uuid
# is generated. Normally this would be done by
# udev+blkid, when the partition table is scanned.
# 'sfdisk' prefixes the partition id with '0x' but
# 'blkid' does not; remove it to mimic 'blkid'
table_id = ptable['id'][2:]
partuuid = "%.33s-%02x" % (table_id, i+1)
partitions.append({
"bootable": p.get("bootable", False),
"type": p["type"],
"start": p["start"] * ssize,
"size": p["size"] * ssize,
"partuuid": partuuid
})
info["partition-table"] = ptable["label"]
info["partition-table-id"] = ptable["id"]
return info
def read_bootloader_type(device):
"""
Read bootloader type from the provided device.
Returns: string representing the found bootloader. Function can return two values:
- 'grub'
- 'unknown'
"""
with open(device, "rb") as f:
if b"GRUB" in f.read(512):
return "grub"
else:
return "unknown"
def read_boot_entries(boot_dir):
"""
Read boot entries.
Returns: list of dictionaries representing configured boot entries.
An example return value:
[
{
"grub_arg": "--unrestricted",
"grub_class": "kernel",
"grub_users": "$grub_users",
"id": "rhel-20210429130346-0-rescue-c116920b13f44c59846f90b1057605bc",
"initrd": "/boot/initramfs-0-rescue-c116920b13f44c59846f90b1057605bc.img",
"linux": "/boot/vmlinuz-0-rescue-c116920b13f44c59846f90b1057605bc",
"options": "$kernelopts",
"title": "Red Hat Enterprise Linux (0-rescue-c116920b13f44c59846f90b1057605bc) 8.4 (Ootpa)",
"version": "0-rescue-c116920b13f44c59846f90b1057605bc"
},
{
"grub_arg": "--unrestricted",
"grub_class": "kernel",
"grub_users": "$grub_users",
"id": "rhel-20210429130346-4.18.0-305.el8.x86_64",
"initrd": "/boot/initramfs-4.18.0-305.el8.x86_64.img $tuned_initrd",
"linux": "/boot/vmlinuz-4.18.0-305.el8.x86_64",
"options": "$kernelopts $tuned_params",
"title": "Red Hat Enterprise Linux (4.18.0-305.el8.x86_64) 8.4 (Ootpa)",
"version": "4.18.0-305.el8.x86_64"
}
]
"""
entries = []
for conf in glob.glob(f"{boot_dir}/loader/entries/*.conf"):
with open(conf) as f:
entries.append(dict(line.strip().split(" ", 1) for line in f))
return sorted(entries, key=lambda e: e["title"])
def rpm_verify(tree):
"""
Read the output of 'rpm --verify'.
Returns: dictionary with two keys 'changed' and 'missing'.
'changed' value is a dictionary with the keys representing modified files from
installed RPM packages and values representing types of applied modifications.
'missing' value is a list of strings prepresenting missing values owned by
installed RPM packages.
An example return value:
{
"changed": {
"/etc/chrony.conf": "S.5....T.",
"/etc/cloud/cloud.cfg": "S.5....T.",
"/etc/nsswitch.conf": "....L....",
"/etc/openldap/ldap.conf": ".......T.",
"/etc/pam.d/fingerprint-auth": "....L....",
"/etc/pam.d/password-auth": "....L....",
"/etc/pam.d/postlogin": "....L....",
"/etc/pam.d/smartcard-auth": "....L....",
"/etc/pam.d/system-auth": "....L....",
"/etc/rhsm/rhsm.conf": "..5....T.",
"/etc/sudoers": "S.5....T.",
"/etc/systemd/logind.conf": "S.5....T."
},
"missing": [
"/etc/udev/rules.d/70-persistent-ipoib.rules",
"/run/cloud-init",
"/run/rpcbind",
"/run/setrans",
"/run/tuned"
]
}
"""
# cannot use `rpm --root` here, because rpm uses passwd from the host to
# verify user and group ownership:
# https://github.com/rpm-software-management/rpm/issues/882
rpm = subprocess.Popen(["chroot", tree, "rpm", "--verify", "--all"],
stdout=subprocess.PIPE, encoding="utf-8")
changed = {}
missing = []
for line in rpm.stdout:
# format description in rpm(8), under `--verify`
attrs = line[:9]
if attrs == "missing ":
missing.append(line[12:].rstrip())
else:
changed[line[13:].rstrip()] = attrs
# ignore return value, because it returns non-zero when it found changes
rpm.wait()
return {
"missing": sorted(missing),
"changed": changed
}
def rpm_packages(tree, is_ostree):
"""
Read NVRs of RPM packages installed on the system.
Returns: sorted list of strings representing RPM packages installed
on the system.
An example return value:
[
"NetworkManager-1.30.0-7.el8.x86_64",
"PackageKit-glib-1.1.12-6.el8.x86_64",
"PackageKit-gtk3-module-1.1.12-6.el8.x86_64",
"abattis-cantarell-fonts-0.0.25-6.el8.noarch",
"acl-2.2.53-1.el8.x86_64",
"adobe-mappings-cmap-20171205-3.el8.noarch",
"adobe-mappings-cmap-deprecated-20171205-3.el8.noarch",
"adobe-mappings-pdf-20180407-1.el8.noarch",
"adwaita-cursor-theme-3.28.0-2.el8.noarch",
"adwaita-icon-theme-3.28.0-2.el8.noarch",
"alsa-lib-1.2.4-5.el8.x86_64"
]
"""
cmd = ["rpm", "--root", tree, "-qa"]
if is_ostree:
cmd += ["--dbpath", "/usr/share/rpm"]
pkgs = subprocess_check_output(cmd, str.split)
return list(sorted(pkgs))
@contextlib.contextmanager
def change_root(root):
real_root = os.open("/", os.O_RDONLY)
try:
os.chroot(root)
yield None
finally:
os.fchdir(real_root)
os.chroot(".")
os.close(real_root)
def read_services(tree, state):
"""
Read the list of systemd services on the system in the given state.
Returns: alphabetically sorted list of strings representing systemd services
in the given state.
The returned list may be empty.
An example return value:
[
"arp-ethers.service",
"canberra-system-bootup.service",
"canberra-system-shutdown-reboot.service",
"canberra-system-shutdown.service",
"chrony-dnssrv@.timer",
"chrony-wait.service"
]
"""
services_state = subprocess_check_output(["systemctl", f"--root={tree}", "list-unit-files"], (lambda s: parse_unit_files(s, state)))
# Since systemd v246, some services previously reported as "enabled" /
# "disabled" are now reported as "alias". There is no systemd command, that
# would take an "alias" unit and report its state as enabled/disabled
# and could run on a different tree (with "--root" option).
# To make the produced list of services in the given state consistent on
# pre/post v246 systemd versions, check all "alias" units and append them
# to the list, if their target is also listed in 'services_state'.
if state != "alias":
services_alias = subprocess_check_output(["systemctl", f"--root={tree}", "list-unit-files"], (lambda s: parse_unit_files(s, "alias")))
for alias in services_alias:
# The service may be in one of the following places (output of
# "systemd-analyze unit-paths", it should not change too often).
unit_paths = [
"/etc/systemd/system.control",
"/run/systemd/system.control",
"/run/systemd/transient",
"/run/systemd/generator.early",
"/etc/systemd/system",
"/run/systemd/system",
"/run/systemd/generator",
"/usr/local/lib/systemd/system",
"/usr/lib/systemd/system",
"/run/systemd/generator.late"
]
with change_root(tree):
for path in unit_paths:
unit_path = os.path.join(path, alias)
if os.path.exists(unit_path):
real_unit_path = os.path.realpath(unit_path)
# Skip the alias, if there was a symlink cycle.
# When symbolic link cycles occur, the returned path will
# be one member of the cycle, but no guarantee is made about
# which member that will be.
if os.path.islink(real_unit_path):
continue
# Append the alias unit to the list, if its target is
# already there.
if os.path.basename(real_unit_path) in services_state:
services_state.append(alias)
# deduplicate and sort
services_state = list(set(services_state))
services_state.sort()
return services_state
def read_default_target(tree):
"""
Read the default systemd target.
Returns: string representing the default systemd target.
An example return value:
"multi-user.target"
"""
return subprocess_check_output(["systemctl", f"--root={tree}", "get-default"]).rstrip()
def read_firewall_zone(tree):
"""
Read enabled services from the configuration of the default firewall zone.
Returns: list of strings representing enabled services in the firewall.
The returned list may be empty.
An example return value:
[
"ssh",
"dhcpv6-client",
"cockpit"
]
"""
try:
with open(f"{tree}/etc/firewalld/firewalld.conf") as f:
conf = parse_environment_vars(f.read())
default = conf["DefaultZone"]
except FileNotFoundError:
default = "public"
r = []
try:
root = xml.etree.ElementTree.parse(f"{tree}/etc/firewalld/zones/{default}.xml").getroot()
except FileNotFoundError:
root = xml.etree.ElementTree.parse(f"{tree}/usr/lib/firewalld/zones/{default}.xml").getroot()
for element in root.findall("service"):
r.append(element.get("name"))
return r
def read_fstab(tree):
"""
Read the content of /etc/fstab.
Returns: list of all uncommented lines read from the configuration file
represented as a list of values split by whitespaces.
The returned list may be empty.
An example return value:
[
[],
[
"UUID=6d066eb4-e4c1-4472-91f9-d167097f48d1",
"/",
"xfs",
"defaults",
"0",
"0"
]
]
"""
result = []
with contextlib.suppress(FileNotFoundError):
with open(f"{tree}/etc/fstab") as f:
result = sorted([line.split() for line in f if line and not line.startswith("#")])
return result
def read_rhsm(tree):
"""
Read configuration changes possible via org.osbuild.rhsm stage
and in addition also the whole content of /etc/rhsm/rhsm.conf.
Returns: returns dictionary with two keys - 'dnf-plugins' and 'rhsm.conf'.
'dnf-plugins' value represents configuration of 'product-id' and
'subscription-manager' DNF plugins.
'rhsm.conf' value is a dictionary representing the content of the RHSM
configuration file.
The returned dictionary may be empty.
An example return value:
{
"dnf-plugins": {
"product-id": {
"enabled": true
},
"subscription-manager": {
"enabled": true
}
},
"rhsm.conf": {
"logging": {
"default_log_level": "INFO"
},
"rhsm": {
"auto_enable_yum_plugins": "1",
"baseurl": "https://cdn.redhat.com",
"ca_cert_dir": "/etc/rhsm/ca/",
"consumercertdir": "/etc/pki/consumer",
"entitlementcertdir": "/etc/pki/entitlement",
"full_refresh_on_yum": "0",
"inotify": "1",
"manage_repos": "0",
"package_profile_on_trans": "0",
"pluginconfdir": "/etc/rhsm/pluginconf.d",
"plugindir": "/usr/share/rhsm-plugins",
"productcertdir": "/etc/pki/product",
"repo_ca_cert": "/etc/rhsm/ca/redhat-uep.pem",
"repomd_gpg_url": "",
"report_package_profile": "1"
},
"rhsmcertd": {
"auto_registration": "1",
"auto_registration_interval": "60",
"autoattachinterval": "1440",
"certcheckinterval": "240",
"disable": "0",
"splay": "1"
},
"server": {
"hostname": "subscription.rhsm.redhat.com",
"insecure": "0",
"no_proxy": "",
"port": "443",
"prefix": "/subscription",
"proxy_hostname": "",
"proxy_password": "",
"proxy_port": "",
"proxy_scheme": "http",
"proxy_user": "",
"ssl_verify_depth": "3"
}
}
}
"""
result = {}
# Check RHSM DNF plugins configuration and allowed options
dnf_plugins_config = {
"product-id": f"{tree}/etc/dnf/plugins/product-id.conf",
"subscription-manager": f"{tree}/etc/dnf/plugins/subscription-manager.conf"
}
for plugin_name, plugin_path in dnf_plugins_config.items():
with contextlib.suppress(FileNotFoundError):
with open(plugin_path) as f:
parser = configparser.ConfigParser()
parser.read_file(f)
# only read "enabled" option from "main" section
with contextlib.suppress(configparser.NoSectionError, configparser.NoOptionError):
# get the value as the first thing, in case it raises an exception
enabled = parser.getboolean("main", "enabled")
try:
dnf_plugins_dict = result["dnf-plugins"]
except KeyError as _:
dnf_plugins_dict = result["dnf-plugins"] = {}
try:
plugin_dict = dnf_plugins_dict[plugin_name]
except KeyError as _:
plugin_dict = dnf_plugins_dict[plugin_name] = {}
plugin_dict["enabled"] = enabled
with contextlib.suppress(FileNotFoundError):
rhsm_conf = {}
with open(f"{tree}/etc/rhsm/rhsm.conf") as f:
parser = configparser.ConfigParser()
parser.read_file(f)
for section in parser.sections():
section_dict = {}
section_dict.update(parser[section])
if section_dict:
rhsm_conf[section] = section_dict
result["rhsm.conf"] = rhsm_conf
return result
def read_sysconfig(tree):
"""
Read selected configuration files from /etc/sysconfig.
Currently supported sysconfig files are:
- 'kernel' - /etc/sysconfig/kernel
- 'network' - /etc/sysconfig/network
- 'network-scripts' - /etc/sysconfig/network-scripts/ifcfg-*
Returns: dictionary with the keys being the supported types of sysconfig
configurations read by the function. Values of 'kernel' and 'network' keys
are a dictionaries containing key/values read from the respective
configuration files. Value of 'network-scripts' key is a dictionary with
the keys corresponding to the suffix of each 'ifcfg-*' configuration file
and their values holding dictionaries with all key/values read from the
configuration file.
The returned dictionary may be empty.
An example return value:
{
"kernel": {
"DEFAULTKERNEL": "kernel",
"UPDATEDEFAULT": "yes"
},
"network": {
"NETWORKING": "yes",
"NOZEROCONF": "yes"
},
"network-scripts": {
"ens3": {
"BOOTPROTO": "dhcp",
"BROWSER_ONLY": "no",
"DEFROUTE": "yes",
"DEVICE": "ens3",
"IPV4_FAILURE_FATAL": "no",
"IPV6INIT": "yes",
"IPV6_AUTOCONF": "yes",
"IPV6_DEFROUTE": "yes",
"IPV6_FAILURE_FATAL": "no",
"NAME": "ens3",
"ONBOOT": "yes",
"PROXY_METHOD": "none",
"TYPE": "Ethernet",
"UUID": "106f1b31-7093-41d6-ae47-1201710d0447"
},
"eth0": {
"BOOTPROTO": "dhcp",
"DEVICE": "eth0",
"IPV6INIT": "no",
"ONBOOT": "yes",
"PEERDNS": "yes",
"TYPE": "Ethernet",
"USERCTL": "yes"
}
}
}
"""
result = {}
sysconfig_paths = {
"kernel": f"{tree}/etc/sysconfig/kernel",
"network": f"{tree}/etc/sysconfig/network"
}
# iterate through supported configs
for name, path in sysconfig_paths.items():
with contextlib.suppress(FileNotFoundError):
with open(path) as f:
# if file exists start with empty array of values
result[name] = parse_environment_vars(f.read())
# iterate through all files in /etc/sysconfig/network-scripts
network_scripts = {}
files = glob.glob(f"{tree}/etc/sysconfig/network-scripts/ifcfg-*")
for file in files:
ifname = os.path.basename(file).lstrip("ifcfg-")
with open(file) as f:
network_scripts[ifname] = parse_environment_vars(f.read())
if network_scripts:
result["network-scripts"] = network_scripts
return result
def read_hosts(tree):
"""
Read non-empty lines of /etc/hosts.
Returns: list of strings for all uncommented lines in the configuration file.
The returned list may be empty.
An example return value:
[
"127.0.0.1 localhost localhost.localdomain localhost4 localhost4.localdomain4",
"::1 localhost localhost.localdomain localhost6 localhost6.localdomain6"
]
"""
result = []
with contextlib.suppress(FileNotFoundError):
with open(f"{tree}/etc/hosts") as f:
for line in f:
line = line.strip()
if line:
result.append(line)
return result
def read_logind_conf(tree):
"""
Read all uncommented key/values set in /etc/systemd/logind.conf.
Returns: dictionary with key/values read from the configuration file.
The returned dictionary may be empty.
An example return value:
{
"NAutoVTs": "0"
}
"""
result = {}
with contextlib.suppress(FileNotFoundError):
with open(f"{tree}/etc/systemd/logind.conf") as f:
parser = configparser.RawConfigParser()
# prevent conversion of the option name to lowercase
parser.optionxform = lambda option: option
parser.read_file(f)
with contextlib.suppress(configparser.NoSectionError):
result.update(parser["Login"])
return result
def read_locale(tree):
"""
Read all uncommented key/values set in /etc/locale.conf.
Returns: dictionary with key/values read from the configuration file.
The returned dictionary may be empty.
An example return value:
{
"LANG": "en_US"
}
"""
with contextlib.suppress(FileNotFoundError):
with open(f"{tree}/etc/locale.conf") as f:
return parse_environment_vars(f.read())
def read_selinux_info(tree):
"""
Read information related to SELinux.
Returns: dictionary with two keys - 'policy' and 'context-mismatch'.
'policy' value corresponds to the value returned by read_selinux_conf().
'context-mismatch' value corresponds to the value returned by
read_selinux_ctx_mismatch().
The returned dictionary may be empty. Keys with empty values are omitted.
An example return value:
{
"context-mismatch": [
{
"actual": "system_u:object_r:root_t:s0",
"expected": "system_u:object_r:device_t:s0",
"filename": "/dev"
},
{
"actual": "system_u:object_r:root_t:s0",
"expected": "system_u:object_r:default_t:s0",
"filename": "/proc"
}
],
"policy": {
"SELINUX": "permissive",
"SELINUXTYPE": "targeted"
}
}
"""
result = {}
policy = read_selinux_conf(tree)
if policy:
result["policy"] = policy
with contextlib.suppress(subprocess.CalledProcessError):
ctx_mismatch = read_selinux_ctx_mismatch(tree)
if ctx_mismatch:
result["context-mismatch"] = ctx_mismatch
return result
def read_selinux_conf(tree):
"""
Read all uncommented key/values set in /etc/selinux/config.
Returns: dictionary with key/values read from the configuration
file.
An example of returned value:
{
"SELINUX": "enforcing",
"SELINUXTYPE": "targeted"
}
"""
with contextlib.suppress(FileNotFoundError):
with open(f"{tree}/etc/selinux/config") as f:
return parse_environment_vars(f.read())
def read_selinux_ctx_mismatch(tree):
"""
Read any mismatch in selinux context of files on the image.
Returns: list of dictionaries as described below. If there
are no mismatches between used and expected selinux context,
then an empty list is returned.
An example of returned value:
[
{
"actual": "system_u:object_r:root_t:s0",
"expected": "system_u:object_r:device_t:s0",
"filename": "/dev"
},
{
"actual": "system_u:object_r:root_t:s0",
"expected": "system_u:object_r:default_t:s0",
"filename": "/proc"
}
]
"""
result = []
# The binary policy that should be used is on the image and has name "policy.X"
# where the "X" is a number. There may be more than one policy files.
# In the usual case, the policy with the highest number suffix should be used.
policy_files = glob.glob(f"{tree}/etc/selinux/targeted/policy/policy.*")
policy_files = sorted(policy_files, reverse=True)
if policy_files:
CMD = [
"setfiles",
"-r", f"{tree}",
"-nvF",
"-c", policy_files[0], # take the policy with the highest number
f"{tree}/etc/selinux/targeted/contexts/files/file_contexts",
f"{tree}"
]
output = subprocess.check_output(CMD).decode()
# output are lines such as:
# Would relabel /tmp/tmpwrozmb47/dev from system_u:object_r:root_t:s0 to system_u:object_r:device_t:s0\n
setfiles_pattern = r"Would\s+relabel\s+(?P<filename>.+)\s+from\s+(?P<actual>.+)\s+to\s+(?P<expected>.+)"
setfiles_re = re.compile(setfiles_pattern)
for line in output.splitlines():
line = line.strip()
if not line:
continue
match = setfiles_re.match(line)
# do not silently ignore changes of 'setfiles' output
if not match:
raise RuntimeError(f"could not match line '{line}' with pattern '{setfiles_pattern}'")
parsed_line = {
"filename": match.group("filename")[len(tree):],
"actual": match.group("actual"),
"expected": match.group("expected")
}
result.append(parsed_line)
return result
def read_modprobe_config(tree):
"""
Read all /etc/modprobe.d/*.conf files and for now, extract only blacklisted
kernel modules.
Returns: dictionary with the keys corresponding to configuration file names
found in /etc/modprobe.d/. The value of each key is a dictionary with
a single key 'blacklist', containing list of kernel module names disallowed
by the configuration file.
An example return value:
{
"blacklist-nouveau.conf": {
"blacklist": [
"nouveau"
]
}
}
"""
result = {}
BLACKLIST_CMD = "blacklist"
files = glob.glob(f"{tree}/etc/modprobe.d/*.conf")
for file in files:
filename = os.path.basename(file)
with open(file) as f:
# The format of files under modprobe.d: one command per line,
# with blank lines and lines starting with '#' ignored.
# A '\' at the end of a line causes it to continue on the next line.
line_to_be_continued = ""
for line in f:
line = line.strip()
# line is not blank
if line:
# comment, skip it
if line[0] == "#":
continue
# this line continues on the following line
if line[-1] == "\\":
line_to_be_continued += line[:-1]
continue
# this line ends here
else:
# is this line continuation of the previous one?
if line_to_be_continued:
line = line_to_be_continued + line
line_to_be_continued = ""
cmd, cmd_args = line.split(' ', 1)
# we care only about blacklist command for now
if cmd == BLACKLIST_CMD:
try:
file_result = result[filename]
except KeyError:
file_result = result[filename] = {}
try:
modules_list = file_result[BLACKLIST_CMD]
except KeyError:
modules_list = file_result[BLACKLIST_CMD] = []
modules_list.append(cmd_args)
return result
def read_cloud_init_conf(tree):
"""
Read the cloud-init configuration from /etc/cloud/cloud.cfg
Returns: dictionary representing the cloud-init configuration.
An example return value:
{
"cloud_config_modules": [
"mounts",
"locale",
"set-passwords",
"rh_subscription",
"yum-add-repo",
"package-update-upgrade-install",
"timezone",
"puppet",
"chef",
"salt-minion",
"mcollective",
"disable-ec2-metadata",
"runcmd"
],
"cloud_final_modules": [
"rightscale_userdata",
"scripts-per-once",
"scripts-per-boot",
"scripts-per-instance",
"scripts-user",
"ssh-authkey-fingerprints",
"keys-to-console",
"phone-home",
"final-message",
"power-state-change"
],
"cloud_init_modules": [
"disk_setup",
"migrator",
"bootcmd",
"write-files",
"growpart",
"resizefs",
"set_hostname",
"update_hostname",
"update_etc_hosts",
"rsyslog",
"users-groups",
"ssh"
],
"disable_root": 1,
"disable_vmware_customization": false,
"mount_default_fields": [
null,
null,
"auto",
"defaults,nofail,x-systemd.requires=cloud-init.service",
"0",
"2"
],
"resize_rootfs_tmp": "/dev",
"ssh_deletekeys": 1,
"ssh_genkeytypes": null,
"ssh_pwauth": 0,
"syslog_fix_perms": null,
"system_info": {
"default_user": {
"gecos": "Cloud User",
"groups": [
"adm",
"systemd-journal"
],
"lock_passwd": true,
"name": "ec2-user",
"shell": "/bin/bash",
"sudo": [
"ALL=(ALL) NOPASSWD:ALL"
]
},
"distro": "rhel",
"paths": {
"cloud_dir": "/var/lib/cloud",
"templates_dir": "/etc/cloud/templates"
},
"ssh_svcname": "sshd"
},
"users": [
"default"
]
}
"""
result = {}
with contextlib.suppress(FileNotFoundError):
with open(f"{tree}/etc/cloud/cloud.cfg") as f:
config = yaml.safe_load(f)
result.update(config)
return result
def read_dracut_conf_d(tree):
"""
Read all *.conf files from /etc/dracut.conf.d/.
Returns: dictionary with the keys representing names of configuration files
from /etc/dracut.conf.d. Value of each key is a dictionary representing the
uncommented configuration options read from the file.
An example return value:
{
"sgdisk.conf": {
"install_items": " sgdisk "
},
"xen.conf": {
"add_drivers": " xen-netfront xen-blkfront "
}
}
"""
result = {}
# iterate through all *.conf files in /etc/dracut.conf.d/
files = glob.glob(f"{tree}/etc/dracut.conf.d/*.conf")
for file in files:
confname = os.path.basename(file)
config = {}
with open(file) as f:
# dracut configuration key/values delimiter is '=' or '+='
for line in f:
line = line.strip()
# A '#' indicates the beginning of a comment; following
# characters, up to the end of the line are not interpreted.
line_comment = line.split("#", 1)
line = line_comment[0]
if line:
key, value = line.split("=", 1)
if key[-1] == "+":
key = key[:-1]
config[key] = value.strip('"')
result[confname] = config
return result
def read_keyboard_conf(tree):
"""
Read keyboard configuration for vconsole and X11.
Returns: dictionary with at most two keys 'X11' and 'vconsole'.
'vconsole' value is a dictionary representing configuration read from
/etc/vconsole.conf.
'X11' value is a dictionary with at most two keys 'layout' and 'variant',
which values are extracted from X11 keyborad configuration.
An example return value:
{
"X11": {
"layout": "us"
},
"vconsole": {
"FONT": "eurlatgr",
"KEYMAP": "us"
}
}
"""
result = {}
# read virtual console configuration
with contextlib.suppress(FileNotFoundError):
with open(f"{tree}/etc/vconsole.conf") as f:
values = parse_environment_vars(f.read())
if values:
result["vconsole"] = values
# read X11 keyboard configuration
with contextlib.suppress(FileNotFoundError):
# Example file content:
#
# Section "InputClass"
# Identifier "system-keyboard"
# MatchIsKeyboard "on"
# Option "XkbLayout" "us,sk"
# Option "XkbVariant" ",qwerty"
# EndSection
x11_config = {}
match_options_dict = {
"layout": r'Section\s+"InputClass"\s+.*Option\s+"XkbLayout"\s+"([\w,-]+)"\s+.*EndSection',
"variant": r'Section\s+"InputClass"\s+.*Option\s+"XkbVariant"\s+"([\w,-]+)"\s+.*EndSection'
}
with open(f"{tree}/etc/X11/xorg.conf.d/00-keyboard.conf") as f:
config = f.read()
for option, pattern in match_options_dict.items():
match = re.search(pattern, config, re.DOTALL)
if match and match.group(1):
x11_config[option] = match.group(1)
if x11_config:
result["X11"] = x11_config
return result
def read_chrony_conf(tree):
"""
Read specific directives from Chrony configuration. Currently parsed
directives are:
- 'server'
- 'pool'
- 'peer'
- 'leapsectz'
Returns: dictionary with the keys representing parsed directives from Chrony
configuration. Value of each key is a list of strings containing arguments
provided with each occurance of the directive in the configuration.
An example return value:
{
"leapsectz": [
"right/UTC"
],
"pool": [
"2.rhel.pool.ntp.org iburst"
],
"server": [
"169.254.169.123 prefer iburst minpoll 4 maxpoll 4"
]
}
"""
result = {}
parsed_directives = ["server", "pool", "peer", "leapsectz"]
with contextlib.suppress(FileNotFoundError):
with open(f"{tree}/etc/chrony.conf") as f:
for line in f:
line = line.strip()
if not line:
continue
# skip comments
if line[0] in ["!", ";", "#", "%"]:
continue
split_line = line.split()
if split_line[0] in parsed_directives:
try:
directive_list = result[split_line[0]]
except KeyError:
directive_list = result[split_line[0]] = []
directive_list.append(" ".join(split_line[1:]))
return result
def read_systemd_service_dropins(tree):
"""
Read systemd *.service drop-in configuration from /etc/systemd/system.
Returns: dictionary with keys representing names of directories from
/etc/systemd/system/ containing drop-in configuration files for systemd
service unit files. Value of each key is a representation of the drop-in
configuration.
An example return value:
{
"nm-cloud-setup.service.d": {
"Service": {
"Environment": "NM_CLOUD_SETUP_EC2=yes"
}
}
}
"""
result = {}
# read all unit drop-in configurations
for directory in glob.glob(f"{tree}/etc/systemd/system/*.service.d"):
config_files = glob.glob(f"{directory}/**/*.conf", recursive=True)
parser = configparser.RawConfigParser()
# prevent conversion of the opion name to lowercase
parser.optionxform = lambda option: option
parser.read(config_files)
dropin_config = {}
for section in parser.sections():
section_config = {}
section_config.update(parser[section])
if section_config:
dropin_config[section] = section_config
if dropin_config:
result[os.path.basename(directory)] = dropin_config
return result
def read_tmpfilesd(tree):
"""
Read all configuration files from /etc/tmpfiles.d.
Returns: dictionary with the keys representing names of configuration files
from /etc/tmpfiles.d. Value of each key is a list of strings representing
uncommented lines read from the configuration file.
An example return value:
{
"sap.conf": [
"x /tmp/.sap*",
"x /tmp/.hdb*lock",
"x /tmp/.trex*lock"
]
}
"""
result = {}
for file in glob.glob(f"{tree}/etc/tmpfiles.d/*.conf"):
with open(file) as f:
file_lines = []
for line in f:
line = line.strip()
if not line:
continue
if line[0] == "#":
continue
file_lines.append(line)
if file_lines:
result[os.path.basename(file)] = file_lines
return result
def read_tuned_profile(tree):
"""
Read the Tuned active profile and profile mode.
Returns: dictionary with at most two keys 'active_profile' and 'profile_mode'.
Value of each key is a string representing respective tuned configuration
value.
An example return value:
{
"active_profile": "sap-hana",
"profile_mode": "manual"
}
"""
result = {}
config_files = ["active_profile", "profile_mode"]
with contextlib.suppress(FileNotFoundError):
for config_file in config_files:
with open(f"{tree}/etc/tuned/{config_file}") as f:
value = f.read()
value = value.strip()
if value:
result[config_file] = value
return result
def read_sysctld(tree):
"""
Read all configuration files from /etc/sysctl.d.
Returns: dictionary with the keys representing names of configuration files
from /etc/sysctl.d. Value of each key is a list of strings representing
uncommented lines read from the configuration file.
An example return value:
{
"sap.conf": [
"kernel.pid_max = 4194304",
"vm.max_map_count = 2147483647"
]
}
"""
result = {}
for file in glob.glob(f"{tree}/etc/sysctl.d/*.conf"):
with open(file) as f:
values = []
for line in f:
line = line.strip()
if not line:
continue
# skip comments
if line[0] in ["#", ";"]:
continue
values.append(line)
if values:
result[os.path.basename(file)] = values
return result
def read_limitsd(tree):
"""
Read all configuration files from /etc/security/limits.d.
Returns: dictionary with the keys representing names of configuration files
from /etc/security/limits.d. Value of each key is a dictionary representing
uncommented configuration values read from the configuration file.
An example return value:
{
"99-sap.conf": [
{
"domain": "@sapsys",
"item": "nofile",
"type": "hard",
"value": "65536"
},
{
"domain": "@sapsys",
"item": "nofile",
"type": "soft",
"value": "65536"
}
]
}
"""
result = {}
for file in glob.glob(f"{tree}/etc/security/limits.d/*.conf"):
with open(file) as f:
values = []
for line in f:
line = line.strip()
# the '#' character introduces a comment - after which the rest of the line is ignored
split_line = line.split("#", 1)
line = split_line[0]
if not line:
continue
# Syntax of a line is "<domain> <type> <item> <value>"
domain, limit_type, item, value = line.split()
values.append({
"domain": domain,
"type": limit_type,
"item": item,
"value": value
})
if values:
result[os.path.basename(file)] = values
return result
def read_sudoers(tree):
"""
Read uncommented lines from sudoers configuration file and /etc/sudoers.d
This functions does not actually do much of a parsing, as sudoers file
format grammar is a bit too much for our purpose.
Any #include or #includedir directives are ignored by this function.
Returns: dictionary with the keys representing names of read configuration
files, /etc/sudoers and files from /etc/sudoers.d. Value of each key is
a list of strings representing uncommented lines read from the configuration
file.
An example return value:
{
"/etc/sudoers": [
"Defaults !visiblepw",
"Defaults always_set_home",
"Defaults match_group_by_gid",
"Defaults always_query_group_plugin",
"Defaults env_reset",
"Defaults env_keep = \"COLORS DISPLAY HOSTNAME HISTSIZE KDEDIR LS_COLORS\"",
"Defaults env_keep += \"MAIL PS1 PS2 QTDIR USERNAME LANG LC_ADDRESS LC_CTYPE\"",
"Defaults env_keep += \"LC_COLLATE LC_IDENTIFICATION LC_MEASUREMENT LC_MESSAGES\"",
"Defaults env_keep += \"LC_MONETARY LC_NAME LC_NUMERIC LC_PAPER LC_TELEPHONE\"",
"Defaults env_keep += \"LC_TIME LC_ALL LANGUAGE LINGUAS _XKB_CHARSET XAUTHORITY\"",
"Defaults secure_path = /sbin:/bin:/usr/sbin:/usr/bin",
"root\tALL=(ALL) \tALL",
"%wheel\tALL=(ALL)\tALL",
"ec2-user\tALL=(ALL)\tNOPASSWD: ALL"
]
}
"""
result = {}
def _parse_sudoers_file(f):
lines = []
for line in f:
line = line.strip()
if not line:
continue
if line[0] == "#":
continue
lines.append(line)
return lines
with contextlib.suppress(FileNotFoundError):
with open(f"{tree}/etc/sudoers") as f:
lines = _parse_sudoers_file(f)
if lines:
result["/etc/sudoers"] = lines
sudoersd_result = {}
for file in glob.glob(f"{tree}/etc/sudoers.d/*"):
with open(file) as f:
lines = _parse_sudoers_file(f)
if lines:
result[os.path.basename(file)] = lines
if sudoersd_result:
result["/etc/sudoers.d"] = sudoersd_result
return result
def read_udev_rules(tree):
"""
Read udev rules defined in /etc/udev/rules.d.
Returns: dictionary with the keys representing names of files with udev
rules from /etc/udev/rules.d. Value of each key is a list of strings
representing uncommented lines read from the configuration file. If
the file is empty (e.g. because of masking udev configuration installed
by an RPM), an emptu list is returned as the respective value.
An example return value:
{
"80-net-name-slot.rules": []
}
"""
result = {}
for file in glob.glob(f"{tree}/etc/udev/rules.d/*.rules"):
with open(file) as f:
lines = []
for line in f:
line = line.strip()
if not line:
continue
if line[0] == "#":
continue
lines.append(line)
# include also empty files in the report
result[os.path.basename(file)] = lines
return result
def read_dnf_conf(tree):
"""
Read DNF configuration and defined variable files.
Returns: dictionary with at most two keys 'dnf.conf' and 'vars'.
'dnf.conf' value is a dictionary representing the DNF configuration
file content.
'vars' value is a dictionary which keys represent names of files from
/etc/dnf/vars/ and values are strings representing the file content.
An example return value:
{
"dnf.conf": {
"main": {
"installonly_limit": "3"
}
},
"vars": {
"releasever": "8.4"
}
}
"""
result = {}
with contextlib.suppress(FileNotFoundError):
with open(f"{tree}/etc/dnf/dnf.conf") as f:
parser = configparser.RawConfigParser()
# prevent conversion of the opion name to lowercase
parser.optionxform = lambda option: option
parser.read(f)
dnf_config = {}
for section in parser.sections():
section_config = {}
section_config.update(parser[section])
if section_config:
dnf_config[section] = section_config
if dnf_config:
result["dnf.conf"] = dnf_config
dnf_vars = {}
for file in glob.glob(f"{tree}/etc/dnf/vars/*"):
with open(file) as f:
dnf_vars[os.path.basename(file)] = f.read().strip()
if dnf_vars:
result["vars"] = dnf_vars
return result
def read_authselect_conf(tree):
"""
Read authselect configuration.
Returns: dictionary with two keys 'profile-id' and 'enabled-features'.
'profile-id' value is a string representing the configured authselect
profile.
'enabled-features' value is a list of strings representing enabled features
of the used authselect profile. In case there are no specific features
enabled, the list is empty.
An example return value:
{
"enabled-features": [],
"profile-id": "sssd"
}
"""
result = {}
with contextlib.suppress(FileNotFoundError):
with open(f"{tree}/etc/authselect/authselect.conf") as f:
# the first line is always the profile ID
# following lines are listing enabled features
# lines starting with '#' and empty lines are skipped
authselect_conf_lines = []
for line in f:
line = line.strip()
if not line:
continue
if line[0] == "#":
continue
authselect_conf_lines.append(line)
if authselect_conf_lines:
result["profile-id"] = authselect_conf_lines[0]
result["enabled-features"] = authselect_conf_lines[1:]
return result
def append_filesystem(report, tree, *, is_ostree=False):
if os.path.exists(f"{tree}/etc/os-release"):
report["packages"] = rpm_packages(tree, is_ostree)
if not is_ostree:
report["rpm-verify"] = rpm_verify(tree)
with open(f"{tree}/etc/os-release") as f:
report["os-release"] = parse_environment_vars(f.read())
report["services-enabled"] = read_services(tree, "enabled")
report["services-disabled"] = read_services(tree, "disabled")
default_target = read_default_target(tree)
if default_target:
report["default-target"] = default_target
with contextlib.suppress(FileNotFoundError):
with open(f"{tree}/etc/hostname") as f:
report["hostname"] = f.read().strip()
with contextlib.suppress(FileNotFoundError):
report["timezone"] = os.path.basename(os.readlink(f"{tree}/etc/localtime"))
authselect_conf = read_authselect_conf(tree)
if authselect_conf:
report["authselect"] = authselect_conf
chrony_conf = read_chrony_conf(tree)
if chrony_conf:
report["chrony"] = chrony_conf
cloud_init_conf = read_cloud_init_conf(tree)
if cloud_init_conf:
report["/etc/cloud/cloud.conf"] = cloud_init_conf
dnf_conf = read_dnf_conf(tree)
if dnf_conf:
report["dnf"] = dnf_conf
dracut_config = read_dracut_conf_d(tree)
if dracut_config:
report["/etc/dracut.conf.d"] = dracut_config
with contextlib.suppress(FileNotFoundError):
report["firewall-enabled"] = read_firewall_zone(tree)
fstab = read_fstab(tree)
if fstab:
report["fstab"] = fstab
hosts = read_hosts(tree)
if hosts:
report["hosts"] = hosts
keyboard = read_keyboard_conf(tree)
if keyboard:
report["keyboard"] = keyboard
limitsd_conf = read_limitsd(tree)
if limitsd_conf:
report["/etc/security/limits.d"] = limitsd_conf
locale = read_locale(tree)
if locale:
report["locale"] = locale
logind = read_logind_conf(tree)
if logind:
report["logind.conf"] = logind
with contextlib.suppress(FileNotFoundError):
with open(f"{tree}/etc/machine-id") as f:
report["machine-id"] = f.readline()
modprobe_config = read_modprobe_config(tree)
if modprobe_config:
report["/etc/modprobe.d"] = modprobe_config
tmpfilesd_config = read_tmpfilesd(tree)
if tmpfilesd_config:
report["/etc/tmpfiles.d"] = tmpfilesd_config
rhsm = read_rhsm(tree)
if rhsm:
report["rhsm"] = rhsm
selinux = read_selinux_info(tree)
if selinux:
report["selinux"] = selinux
sudoers_conf = read_sudoers(tree)
if sudoers_conf:
report["sudoers"] = sudoers_conf
sysconfig = read_sysconfig(tree)
if sysconfig:
report["sysconfig"] = sysconfig
sysctld_config = read_sysctld(tree)
if sysctld_config:
report["/etc/sysctl.d"] = sysctld_config
systemd_service_dropins = read_systemd_service_dropins(tree)
if systemd_service_dropins:
report["systemd-service-dropins"] = systemd_service_dropins
tuned_profile = read_tuned_profile(tree)
if tuned_profile:
report["tuned"] = tuned_profile
udev_rules = read_udev_rules(tree)
if udev_rules:
report["/etc/udev/rules.d"] = udev_rules
with open(f"{tree}/etc/passwd") as f:
report["passwd"] = sorted(f.read().strip().split("\n"))
with open(f"{tree}/etc/group") as f:
report["groups"] = sorted(f.read().strip().split("\n"))
if is_ostree:
with open(f"{tree}/usr/lib/passwd") as f:
report["passwd-system"] = sorted(f.read().strip().split("\n"))
with open(f"{tree}/usr/lib/group") as f:
report["groups-system"] = sorted(f.read().strip().split("\n"))
if os.path.exists(f"{tree}/boot") and len(os.listdir(f"{tree}/boot")) > 0:
assert "bootmenu" not in report
with contextlib.suppress(FileNotFoundError):
with open(f"{tree}/boot/grub2/grubenv") as f:
report["boot-environment"] = parse_environment_vars(f.read())
report["bootmenu"] = read_boot_entries(f"{tree}/boot")
elif len(glob.glob(f"{tree}/vmlinuz-*")) > 0:
assert "bootmenu" not in report
with open(f"{tree}/grub2/grubenv") as f:
report["boot-environment"] = parse_environment_vars(f.read())
report["bootmenu"] = read_boot_entries(tree)
elif len(glob.glob(f"{tree}/EFI")):
print("EFI partition", file=sys.stderr)
def partition_is_esp(partition):
return partition["type"] == "C12A7328-F81F-11D2-BA4B-00A0C93EC93B"
def find_esp(partitions):
for i, p in enumerate(partitions):
if partition_is_esp(p):
return p, i
return None, 0
def append_partitions(report, device, loctl):
partitions = report["partitions"]
esp, esp_id = find_esp(partitions)
with contextlib.ExitStack() as cm:
devices = {}
for n, part in enumerate(partitions):
start, size = part["start"], part["size"]
dev = cm.enter_context(loop_open(loctl, device, offset=start, size=size))
devices[n] = dev
read_partition(dev, part)
for n, part in enumerate(partitions):
if not part["fstype"]:
continue
with mount(devices[n]) as tree:
if esp and os.path.exists(f"{tree}/boot/efi"):
with mount_at(devices[esp_id], f"{tree}/boot/efi", options=['umask=077']):
append_filesystem(report, tree)
else:
append_filesystem(report, tree)
def analyse_image(image):
loctl = loop.LoopControl()
imgfmt = read_image_format(image)
report = {"image-format": imgfmt}
with open_image(loctl, image, imgfmt) as (_, device):
report["bootloader"] = read_bootloader_type(device)
report.update(read_partition_table(device))
if report["partition-table"]:
append_partitions(report, device, loctl)
else:
with mount(device) as tree:
append_filesystem(report, tree)
return report
def append_directory(report, tree):
if os.path.lexists(f"{tree}/ostree"):
os.makedirs(f"{tree}/etc", exist_ok=True)
with mount_at(f"{tree}/usr/etc", f"{tree}/etc", extra=["--bind"]):
append_filesystem(report, tree, is_ostree=True)
else:
append_filesystem(report, tree)
def append_ostree_repo(report, repo):
ostree = functools.partial(run_ostree, repo=repo)
r = ostree("config", "get", "core.mode")
report["ostree"] = {
"repo": {
"core.mode": r.stdout.strip()
}
}
r = ostree("refs")
refs = r.stdout.strip().split("\n")
report["ostree"]["refs"] = refs
resolved = {r: ostree("rev-parse", r).stdout.strip() for r in refs}
commit = resolved[refs[0]]
with tempfile.TemporaryDirectory(dir="/var/tmp") as tmpdir:
tree = os.path.join(tmpdir, "tree")
ostree("checkout", "--force-copy", commit, tree)
append_directory(report, tree)
def analyse_directory(path):
report = {}
if os.path.exists(os.path.join(path, "compose.json")):
report["type"] = "ostree/commit"
repo = os.path.join(path, "repo")
append_ostree_repo(report, repo)
elif os.path.isdir(os.path.join(path, "refs")):
report["type"] = "ostree/repo"
append_ostree_repo(report, repo)
else:
append_directory(report, path)
return report
def is_tarball(path):
mtype, encoding = mimetypes.guess_type(path)
return mtype == "application/x-tar"
def analyse_tarball(path):
with tempfile.TemporaryDirectory(dir="/var/tmp") as tmpdir:
tree = os.path.join(tmpdir, "root")
os.makedirs(tree)
command = [
"tar",
"--selinux",
"--xattrs",
"--acls",
"-x",
"--auto-compress",
"-f", path,
"-C", tree
]
subprocess.run(command,
stdout=sys.stderr,
check=True)
return analyse_directory(tree)
def is_compressed(path):
_, encoding = mimetypes.guess_type(path)
return encoding in ["xz", "gzip", "bzip2"]
def analyse_compressed(path):
_, encoding = mimetypes.guess_type(path)
if encoding == "xz":
command = ["unxz", "--force"]
elif encoding == "gzip":
command = ["gunzip", "--force"]
elif encoding == "bzip2":
command = ["bunzip2", "--force"]
else:
raise ValueError(f"Unsupported compression: {encoding}")
with tempfile.TemporaryDirectory(dir="/var/tmp") as tmpdir:
subprocess.run(["cp", "--reflink=auto", "-a", path, tmpdir],
check=True)
files = os.listdir(tmpdir)
archive = os.path.join(tmpdir, files[0])
subprocess.run(command + [archive], check=True)
files = os.listdir(tmpdir)
assert len(files) == 1
image = os.path.join(tmpdir, files[0])
return analyse_image(image)
def main():
parser = argparse.ArgumentParser(description="Inspect an image")
parser.add_argument("target", metavar="TARGET",
help="The file or directory to analyse",
type=os.path.abspath)
args = parser.parse_args()
target = args.target
if os.path.isdir(target):
report = analyse_directory(target)
elif is_tarball(target):
report = analyse_tarball(target)
elif is_compressed(target):
report = analyse_compressed(target)
else:
report = analyse_image(target)
json.dump(report, sys.stdout, sort_keys=True, indent=2)
if __name__ == "__main__":
main()
Reference in a new issue View git blame Copy permalink