1046 lines
37 KiB
Python
1046 lines
37 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import shutil
|
|
import shlex
|
|
import stat
|
|
import subprocess # nosec
|
|
import time
|
|
from dataclasses import asdict
|
|
from typing import Any, Dict, List, Optional, Set, Tuple
|
|
|
|
from . import accounts as _accounts
|
|
from . import systemd as _systemd
|
|
from .fsutil import stat_dir_triplet
|
|
from .platform import detect_platform, get_backend
|
|
from .ignore import IgnorePolicy
|
|
from .harvest_safety import ensure_private_empty_dir, prepare_new_private_dir
|
|
from .pathfilter import PathFilter
|
|
from .version import get_enroll_version
|
|
from .state import write_state
|
|
from .harvest_collectors.context import HarvestContext
|
|
from .harvest_types import (
|
|
EtcCustomSnapshot,
|
|
ExcludedFile,
|
|
FirewallRuntimeSnapshot,
|
|
ManagedDir,
|
|
ManagedFile,
|
|
PackageSnapshot,
|
|
ServiceSnapshot,
|
|
SysctlSnapshot,
|
|
)
|
|
|
|
from .capture import capture_file, write_bytes_into_bundle
|
|
from . import system_paths
|
|
from .package_hints import package_section_from_installations, safe_name
|
|
|
|
UnitQueryError = _systemd.UnitQueryError
|
|
|
|
|
|
def list_enabled_services() -> List[str]:
|
|
return _systemd.list_enabled_services()
|
|
|
|
|
|
def list_enabled_timers() -> List[str]:
|
|
return _systemd.list_enabled_timers()
|
|
|
|
|
|
def get_unit_info(unit: str) -> Any:
|
|
return _systemd.get_unit_info(unit)
|
|
|
|
|
|
def get_timer_info(timer: str) -> Any:
|
|
return _systemd.get_timer_info(timer)
|
|
|
|
|
|
def collect_non_system_users() -> List[Any]:
|
|
return _accounts.collect_non_system_users()
|
|
|
|
|
|
def _merge_parent_dirs(
|
|
existing_dirs: List[ManagedDir],
|
|
managed_files: List[ManagedFile],
|
|
*,
|
|
policy: IgnorePolicy,
|
|
extra_paths: Optional[List[str]] = None,
|
|
) -> List[ManagedDir]:
|
|
"""Ensure parent directories for managed_files are present in managed_dirs.
|
|
|
|
This is used so the Ansible manifest can create destination directories with
|
|
explicit owner/group/mode (ansible-lint friendly) without needing a separate
|
|
"mkdir without perms" task.
|
|
|
|
We only add the immediate parent directory for each managed file. For
|
|
explicit directory includes (extra_paths), existing_dirs will already
|
|
contain the walked directory tree.
|
|
"""
|
|
by_path: Dict[str, ManagedDir] = {
|
|
d.path: d for d in (existing_dirs or []) if d.path
|
|
}
|
|
|
|
def _iter_paths() -> List[str]:
|
|
paths: List[str] = []
|
|
for mf in managed_files or []:
|
|
if mf and mf.path:
|
|
paths.append(str(mf.path))
|
|
for p in extra_paths or []:
|
|
if p:
|
|
paths.append(str(p))
|
|
return paths
|
|
|
|
for p0 in _iter_paths():
|
|
p = str(p0 or "").rstrip("/")
|
|
if not p:
|
|
continue
|
|
dpath = os.path.dirname(p)
|
|
if not dpath or dpath == "/":
|
|
continue
|
|
if dpath in by_path:
|
|
continue
|
|
|
|
# Directory-deny logic: newer IgnorePolicy implementations provide
|
|
# deny_reason_dir(). Older/simple policies (including unit tests) may
|
|
# only implement deny_reason(), which is file-oriented and may return
|
|
# "not_regular_file" for directories.
|
|
deny = None
|
|
deny_dir = getattr(policy, "deny_reason_dir", None)
|
|
if callable(deny_dir):
|
|
deny = deny_dir(dpath)
|
|
else:
|
|
deny = policy.deny_reason(dpath)
|
|
if deny in ("not_regular_file", "not_file", "not_regular"):
|
|
deny = None
|
|
if deny:
|
|
# If the file itself was captured, its parent directory is likely safe,
|
|
# but still respect deny globs for directories to avoid managing
|
|
# sensitive/forbidden trees.
|
|
continue
|
|
|
|
try:
|
|
owner, group, mode = stat_dir_triplet(dpath)
|
|
except OSError:
|
|
continue
|
|
|
|
by_path[dpath] = ManagedDir(
|
|
path=dpath,
|
|
owner=owner,
|
|
group=group,
|
|
mode=mode,
|
|
reason="parent_of_managed_file",
|
|
)
|
|
|
|
return [by_path[k] for k in sorted(by_path)]
|
|
|
|
|
|
_FIREWALL_CAPTURE_COMMANDS: Dict[str, Tuple[str, ...]] = {
|
|
"ipset_save": ("ipset", "save"),
|
|
"iptables_v4_save": ("iptables-save",),
|
|
"iptables_v6_save": ("ip6tables-save",),
|
|
"sysctl_all": ("sysctl", "-a"),
|
|
}
|
|
|
|
|
|
def _run_capture_command(
|
|
command_key: str, *, timeout: int = 10
|
|
) -> tuple[Optional[str], Optional[str]]:
|
|
"""Return (stdout, error_note) for an allowlisted local state command.
|
|
|
|
The command key is resolved through ``_FIREWALL_CAPTURE_COMMANDS`` so this
|
|
helper never executes caller-supplied argv. Commands are run with
|
|
``shell=False`` explicitly to avoid shell interpretation.
|
|
"""
|
|
argv = _FIREWALL_CAPTURE_COMMANDS.get(command_key)
|
|
if argv is None:
|
|
return None, f"Unknown capture command: {command_key}"
|
|
|
|
exe = argv[0]
|
|
if shutil.which(exe) is None:
|
|
return None, f"{exe} not found on PATH."
|
|
|
|
try:
|
|
proc = subprocess.run( # nosec
|
|
argv,
|
|
shell=False,
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
timeout=timeout,
|
|
)
|
|
except Exception as e: # noqa: BLE001
|
|
return None, f"{' '.join(argv)} failed: {e!r}"
|
|
|
|
if proc.returncode != 0:
|
|
stderr = (proc.stderr or "").strip()
|
|
if len(stderr) > 300:
|
|
stderr = stderr[:297] + "..."
|
|
return (
|
|
None,
|
|
f"{' '.join(argv)} exited {proc.returncode}: {stderr or '(no stderr)'}",
|
|
)
|
|
|
|
return proc.stdout or "", None
|
|
|
|
|
|
def _write_generated_artifact(
|
|
bundle_dir: str, role_name: str, src_rel: str, content: str
|
|
) -> None:
|
|
"""Write a generated harvest artifact that did not exist as a file on disk."""
|
|
|
|
# Keep generated artifacts on the same no-follow/exclusive-create path as
|
|
# copied source artifacts. This preserves the invariant that nothing under
|
|
# artifacts/ is written via a plain open() that could follow a symlink if a
|
|
# directory were unexpectedly reused or raced.
|
|
write_bytes_into_bundle(bundle_dir, role_name, src_rel, content.encode("utf-8"))
|
|
|
|
|
|
_SYSCTL_KEY_RE = re.compile(r"^[A-Za-z0-9_.-]+$")
|
|
_SYSCTL_GENERATED_DEST = "/etc/sysctl.d/99-enroll.conf"
|
|
_SYSCTL_GENERATED_SRC_REL = "sysctl/99-enroll.conf"
|
|
|
|
# Writable-looking action/identity keys that are poor candidates for persistent
|
|
# config. This avoids generating a file that tries to replay one-shot triggers or
|
|
# host identity that should be managed elsewhere (e.g. /etc/hostname).
|
|
_SYSCTL_VOLATILE_KEYS = {
|
|
"fs.binfmt_misc.status",
|
|
"kernel.domainname",
|
|
"kernel.hostname",
|
|
"kernel.kexec_load_disabled",
|
|
"kernel.kexec_load_limit_panic",
|
|
"kernel.kexec_load_limit_reboot",
|
|
"kernel.max_rcu_stall_to_panic",
|
|
"kernel.modules_disabled",
|
|
"kernel.ns_last_pid",
|
|
"net.ipv4.route.flush",
|
|
"net.ipv6.route.flush",
|
|
"vm.compact_memory",
|
|
"vm.drop_caches",
|
|
"vm.stat_refresh",
|
|
}
|
|
|
|
_SYSCTL_VOLATILE_PREFIXES = (
|
|
"fs.binfmt_misc.",
|
|
"kernel.sched_domain.",
|
|
)
|
|
|
|
# These are paired with ratio/byte counterparts. The inactive side appears as 0
|
|
# when read; replaying that 0 through sysctl -p is noisy and can be rejected by
|
|
# kernels that enforce minimum values.
|
|
_SYSCTL_SKIP_ZERO_VALUE_KEYS = {
|
|
"vm.dirty_background_bytes",
|
|
"vm.dirty_background_ratio",
|
|
"vm.dirty_bytes",
|
|
"vm.dirty_ratio",
|
|
}
|
|
|
|
|
|
def _sysctl_proc_path(key: str) -> str:
|
|
return "/proc/sys/" + key.replace(".", "/")
|
|
|
|
|
|
def _sysctl_key_is_persistable(key: str) -> tuple[bool, str]:
|
|
if not key or not _SYSCTL_KEY_RE.fullmatch(key):
|
|
return False, "invalid key"
|
|
if key in _SYSCTL_VOLATILE_KEYS or any(
|
|
key.startswith(prefix) for prefix in _SYSCTL_VOLATILE_PREFIXES
|
|
):
|
|
return False, "volatile/action key"
|
|
|
|
proc_path = _sysctl_proc_path(key)
|
|
try:
|
|
st = os.stat(proc_path)
|
|
except OSError:
|
|
return False, "no /proc/sys entry"
|
|
|
|
if not stat.S_ISREG(st.st_mode):
|
|
return False, "not a regular /proc/sys entry"
|
|
if (stat.S_IMODE(st.st_mode) & 0o222) == 0:
|
|
return False, "read-only /proc/sys entry"
|
|
return True, ""
|
|
|
|
|
|
def _sysctl_entry_is_persistable(key: str, value: str) -> tuple[bool, str]:
|
|
ok, reason = _sysctl_key_is_persistable(key)
|
|
if not ok:
|
|
return ok, reason
|
|
|
|
if key in _SYSCTL_SKIP_ZERO_VALUE_KEYS and str(value).strip() == "0":
|
|
return False, "inactive mutually-exclusive zero value"
|
|
|
|
return True, ""
|
|
|
|
|
|
def _parse_sysctl_a_output(
|
|
text: str,
|
|
*,
|
|
require_persistable: bool = True,
|
|
) -> tuple[Dict[str, str], Dict[str, int]]:
|
|
"""Parse `sysctl -a` output into persistable key/value pairs.
|
|
|
|
`sysctl -a` includes read-only, write-only, multiline, action-like, and
|
|
host-identity values. Persisting those can create noisy or failing Ansible
|
|
runs, so the default parser keeps only single-line writable-looking keys.
|
|
"""
|
|
|
|
out: Dict[str, str] = {}
|
|
skipped: Dict[str, int] = {
|
|
"malformed": 0,
|
|
"empty_value": 0,
|
|
"non_persistable": 0,
|
|
"duplicate": 0,
|
|
}
|
|
|
|
for raw in (text or "").splitlines():
|
|
line = raw.strip()
|
|
if not line:
|
|
continue
|
|
if " = " in line:
|
|
key, value = line.split(" = ", 1)
|
|
elif "=" in line:
|
|
key, value = line.split("=", 1)
|
|
else:
|
|
skipped["malformed"] += 1
|
|
continue
|
|
|
|
key = key.strip()
|
|
value = value.strip()
|
|
if not key:
|
|
skipped["malformed"] += 1
|
|
continue
|
|
if value == "":
|
|
skipped["empty_value"] += 1
|
|
continue
|
|
if key in out:
|
|
skipped["duplicate"] += 1
|
|
continue
|
|
if require_persistable:
|
|
ok, _reason = _sysctl_entry_is_persistable(key, value)
|
|
if not ok:
|
|
skipped["non_persistable"] += 1
|
|
continue
|
|
out[key] = value
|
|
|
|
return dict(sorted(out.items())), skipped
|
|
|
|
|
|
def _render_sysctl_conf(parameters: Dict[str, str], notes: List[str]) -> str:
|
|
lines = [
|
|
"# Generated by Enroll from live sysctl state.",
|
|
"# Review before applying broadly; runtime sysctl state can be host/kernel-specific.",
|
|
]
|
|
for note in notes:
|
|
lines.append(f"# {note}")
|
|
lines.append("")
|
|
for key, value in sorted((parameters or {}).items()):
|
|
safe_value = str(value).replace("\n", " ").strip()
|
|
lines.append(f"{key} = {safe_value}")
|
|
lines.append("")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _collect_sysctl_snapshot(bundle_dir: str) -> SysctlSnapshot:
|
|
role_name = "sysctl"
|
|
notes: List[str] = []
|
|
managed_files: List[ManagedFile] = []
|
|
|
|
out, err = _run_capture_command("sysctl_all", timeout=20)
|
|
if err:
|
|
notes.append(err)
|
|
return SysctlSnapshot(role_name=role_name, notes=notes)
|
|
|
|
parameters, skipped = _parse_sysctl_a_output(out or "")
|
|
if not parameters:
|
|
notes.append("No persistable live sysctl parameters were detected.")
|
|
return SysctlSnapshot(role_name=role_name, parameters=parameters, notes=notes)
|
|
|
|
notes.append(f"Captured {len(parameters)} live writable sysctl parameter(s).")
|
|
skipped_total = sum(skipped.values())
|
|
if skipped_total:
|
|
details = ", ".join(f"{k}={v}" for k, v in sorted(skipped.items()) if v)
|
|
notes.append(
|
|
"Skipped "
|
|
f"{skipped_total} sysctl entr{'y' if skipped_total == 1 else 'ies'} "
|
|
f"that were not suitable for persistence ({details})."
|
|
)
|
|
|
|
_write_generated_artifact(
|
|
bundle_dir,
|
|
role_name,
|
|
_SYSCTL_GENERATED_SRC_REL,
|
|
_render_sysctl_conf(parameters, notes),
|
|
)
|
|
managed_files.append(
|
|
ManagedFile(
|
|
path=_SYSCTL_GENERATED_DEST,
|
|
src_rel=_SYSCTL_GENERATED_SRC_REL,
|
|
owner="root",
|
|
group="root",
|
|
mode="0644",
|
|
reason="system_sysctl",
|
|
)
|
|
)
|
|
return SysctlSnapshot(
|
|
role_name=role_name,
|
|
managed_files=managed_files,
|
|
parameters=parameters,
|
|
notes=notes,
|
|
)
|
|
|
|
|
|
def _ipset_save_has_state(text: str) -> bool:
|
|
for raw in (text or "").splitlines():
|
|
line = raw.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
if line.startswith(("create ", "add ")):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _parse_ipset_set_names(text: str) -> List[str]:
|
|
names: List[str] = []
|
|
seen: Set[str] = set()
|
|
for raw in (text or "").splitlines():
|
|
line = raw.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
try:
|
|
toks = shlex.split(line)
|
|
except ValueError:
|
|
toks = line.split()
|
|
if len(toks) >= 2 and toks[0] == "create" and toks[1] not in seen:
|
|
seen.add(toks[1])
|
|
names.append(toks[1])
|
|
return names
|
|
|
|
|
|
def _iptables_save_has_state(text: str) -> bool:
|
|
"""Return True when iptables-save output contains non-default state."""
|
|
for raw in (text or "").splitlines():
|
|
line = raw.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
if line.startswith("*") or line == "COMMIT":
|
|
continue
|
|
if line.startswith(":"):
|
|
parts = line.split()
|
|
chain_name = parts[0][1:] if parts else ""
|
|
policy = parts[1] if len(parts) >= 2 else ""
|
|
# Built-in empty chains usually look like ':INPUT ACCEPT [0:0]'.
|
|
# A changed policy, or any custom chain, is meaningful state.
|
|
if policy not in ("ACCEPT", "-"):
|
|
return True
|
|
if policy == "-" and chain_name:
|
|
return True
|
|
continue
|
|
if line.startswith(("-A ", "-I ", "-N ", "-P ", "-R ")):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _collect_firewall_runtime_snapshot(
|
|
bundle_dir: str,
|
|
*,
|
|
persistent_ipset_files: Optional[List[str]] = None,
|
|
persistent_iptables_v4_files: Optional[List[str]] = None,
|
|
persistent_iptables_v6_files: Optional[List[str]] = None,
|
|
) -> FirewallRuntimeSnapshot:
|
|
"""Capture live kernel firewall state only when no persistent config exists.
|
|
|
|
Enroll also harvests persistent firewall files such as
|
|
/etc/iptables/rules.v4, /etc/iptables/rules.v6, and /etc/ipset.conf as
|
|
managed files. The generated runtime restore role is therefore a fallback:
|
|
it captures each firewall family only when that family has no persistent
|
|
file to avoid generating two roles that try to manage the same state.
|
|
"""
|
|
role_name = "firewall_runtime"
|
|
packages: Set[str] = set()
|
|
notes: List[str] = []
|
|
ipset_save_rel: Optional[str] = None
|
|
ipset_sets: List[str] = []
|
|
iptables_v4_rel: Optional[str] = None
|
|
iptables_v6_rel: Optional[str] = None
|
|
|
|
persistent_ipset_files = persistent_ipset_files or []
|
|
persistent_iptables_v4_files = persistent_iptables_v4_files or []
|
|
persistent_iptables_v6_files = persistent_iptables_v6_files or []
|
|
|
|
if persistent_ipset_files:
|
|
notes.append(
|
|
"Live ipset runtime capture skipped because persistent ipset "
|
|
f"configuration was found: {', '.join(persistent_ipset_files)}"
|
|
)
|
|
else:
|
|
ipset_out, ipset_err = _run_capture_command("ipset_save")
|
|
if ipset_err:
|
|
notes.append(ipset_err)
|
|
elif ipset_out is not None and _ipset_save_has_state(ipset_out):
|
|
ipset_save_rel = "firewall/ipset.save"
|
|
_write_generated_artifact(bundle_dir, role_name, ipset_save_rel, ipset_out)
|
|
ipset_sets = _parse_ipset_set_names(ipset_out)
|
|
packages.add("ipset")
|
|
|
|
if persistent_iptables_v4_files:
|
|
notes.append(
|
|
"Live IPv4 iptables runtime capture skipped because persistent "
|
|
f"IPv4 iptables configuration was found: {', '.join(persistent_iptables_v4_files)}"
|
|
)
|
|
else:
|
|
ipt4_out, ipt4_err = _run_capture_command("iptables_v4_save")
|
|
if ipt4_err:
|
|
notes.append(ipt4_err)
|
|
elif ipt4_out is not None and _iptables_save_has_state(ipt4_out):
|
|
iptables_v4_rel = "firewall/iptables.v4"
|
|
_write_generated_artifact(bundle_dir, role_name, iptables_v4_rel, ipt4_out)
|
|
packages.add("iptables")
|
|
|
|
if persistent_iptables_v6_files:
|
|
notes.append(
|
|
"Live IPv6 iptables runtime capture skipped because persistent "
|
|
f"IPv6 iptables configuration was found: {', '.join(persistent_iptables_v6_files)}"
|
|
)
|
|
else:
|
|
ipt6_out, ipt6_err = _run_capture_command("iptables_v6_save")
|
|
if ipt6_err:
|
|
notes.append(ipt6_err)
|
|
elif ipt6_out is not None and _iptables_save_has_state(ipt6_out):
|
|
iptables_v6_rel = "firewall/iptables.v6"
|
|
_write_generated_artifact(bundle_dir, role_name, iptables_v6_rel, ipt6_out)
|
|
packages.add("iptables")
|
|
|
|
# Package names are intentionally added only when matching live state was
|
|
# captured. Merely having iptables/ipset installed should not create a role.
|
|
|
|
return FirewallRuntimeSnapshot(
|
|
role_name=role_name,
|
|
packages=sorted(packages),
|
|
ipset_save=ipset_save_rel,
|
|
ipset_sets=ipset_sets,
|
|
iptables_v4_save=iptables_v4_rel,
|
|
iptables_v6_save=iptables_v6_rel,
|
|
notes=notes,
|
|
)
|
|
|
|
|
|
def harvest(
|
|
bundle_dir: str,
|
|
policy: Optional[IgnorePolicy] = None,
|
|
*,
|
|
dangerous: bool = False,
|
|
include_paths: Optional[List[str]] = None,
|
|
exclude_paths: Optional[List[str]] = None,
|
|
allow_existing_output: bool = False,
|
|
) -> str:
|
|
# If a policy is not supplied, build one. `--dangerous` relaxes secret
|
|
# detection and deny-glob skipping.
|
|
if policy is None:
|
|
policy = IgnorePolicy(dangerous=dangerous)
|
|
elif dangerous:
|
|
# If callers explicitly provided a policy but also requested
|
|
# dangerous behaviour, honour the CLI intent.
|
|
policy.dangerous = True
|
|
bundle_path = (
|
|
ensure_private_empty_dir(bundle_dir, label="harvest output")
|
|
if allow_existing_output
|
|
else prepare_new_private_dir(bundle_dir, label="harvest output")
|
|
)
|
|
bundle_dir = str(bundle_path)
|
|
|
|
# User-provided includes/excludes. Excludes apply to all harvesting;
|
|
# includes are harvested into an extra role.
|
|
path_filter = PathFilter(include=include_paths or (), exclude=exclude_paths or ())
|
|
|
|
from .harvest_collectors.container_images import ContainerImagesCollector
|
|
from .harvest_collectors.cron_logrotate import CronLogrotateCollector
|
|
from .harvest_collectors.package_manager import PackageManagerConfigCollector
|
|
from .harvest_collectors.paths import ExtraPathsCollector, UsrLocalCustomCollector
|
|
from .harvest_collectors.runtime import RuntimeStateCollector
|
|
from .harvest_collectors.services import ServicePackageCollector
|
|
from .harvest_collectors.users import UsersCollector
|
|
|
|
if hasattr(os, "geteuid") and os.geteuid() != 0:
|
|
print(
|
|
"Warning: not running as root; harvest may miss files or metadata.",
|
|
flush=True,
|
|
)
|
|
|
|
platform = detect_platform()
|
|
backend = get_backend(platform)
|
|
|
|
owned_etc, etc_owner_map, topdir_to_pkgs, pkg_to_etc_paths = (
|
|
backend.build_etc_index()
|
|
)
|
|
|
|
# Global de-duplication across roles: each absolute path is captured at most once.
|
|
# This avoids multiple Ansible roles managing the same destination file.
|
|
captured_global: Set[str] = set()
|
|
|
|
# -------------------------
|
|
# Cron / logrotate unification
|
|
#
|
|
# If cron/logrotate are installed, capture all related configuration/state into
|
|
# dedicated package roles ("cron" and "logrotate") so the same destination path
|
|
# is never managed by unrelated roles.
|
|
#
|
|
# This includes user-specific crontabs under /var/spool, which means the cron role
|
|
# should be applied after users have been created (handled in manifest ordering).
|
|
# -------------------------
|
|
|
|
installed_pkgs = backend.installed_packages() or {}
|
|
installed_names: Set[str] = set(installed_pkgs.keys())
|
|
|
|
persistent_ipset_files = system_paths.persistent_firewall_files(
|
|
system_paths.persistent_ipset_globs()
|
|
)
|
|
persistent_iptables_v4_files = system_paths.persistent_firewall_files(
|
|
system_paths.persistent_iptables_v4_globs()
|
|
)
|
|
persistent_iptables_v6_files = system_paths.persistent_firewall_files(
|
|
system_paths.persistent_iptables_v6_globs()
|
|
)
|
|
|
|
context = HarvestContext(
|
|
bundle_dir=bundle_dir,
|
|
policy=policy,
|
|
path_filter=path_filter,
|
|
platform=platform,
|
|
backend=backend,
|
|
installed_pkgs=installed_pkgs,
|
|
installed_names=installed_names,
|
|
owned_etc=owned_etc,
|
|
etc_owner_map=etc_owner_map,
|
|
topdir_to_pkgs=topdir_to_pkgs,
|
|
pkg_to_etc_paths=pkg_to_etc_paths,
|
|
captured_global=captured_global,
|
|
)
|
|
|
|
runtime_collection = RuntimeStateCollector(
|
|
context,
|
|
persistent_ipset_files=persistent_ipset_files,
|
|
persistent_iptables_v4_files=persistent_iptables_v4_files,
|
|
persistent_iptables_v6_files=persistent_iptables_v6_files,
|
|
).collect()
|
|
firewall_runtime_snapshot = runtime_collection.firewall_runtime_snapshot
|
|
sysctl_snapshot = runtime_collection.sysctl_snapshot
|
|
|
|
# The generated sysctl role owns /etc/sysctl.d/99-enroll.conf; do not also
|
|
# capture an existing file at that path into etc_custom/package roles.
|
|
for mf in sysctl_snapshot.managed_files:
|
|
captured_global.add(mf.path)
|
|
|
|
cron_logrotate_collection = CronLogrotateCollector(context).collect()
|
|
cron_pkg = cron_logrotate_collection.cron_pkg
|
|
logrotate_pkg = cron_logrotate_collection.logrotate_pkg
|
|
cron_snapshot = cron_logrotate_collection.cron_snapshot
|
|
logrotate_snapshot = cron_logrotate_collection.logrotate_snapshot
|
|
|
|
service_package_collection = ServicePackageCollector(
|
|
context,
|
|
cron_snapshot=cron_snapshot,
|
|
logrotate_snapshot=logrotate_snapshot,
|
|
cron_pkg=cron_pkg,
|
|
logrotate_pkg=logrotate_pkg,
|
|
).collect()
|
|
service_snaps = service_package_collection.service_snaps
|
|
pkg_snaps = service_package_collection.pkg_snaps
|
|
manual_pkgs = service_package_collection.manual_pkgs
|
|
service_role_aliases = service_package_collection.service_role_aliases
|
|
seen_by_role = service_package_collection.seen_by_role
|
|
|
|
# -------------------------
|
|
# Users role, Flatpak and Snap state
|
|
# -------------------------
|
|
users_collection = UsersCollector(context, seen_by_role).collect()
|
|
users_snapshot = users_collection.users_snapshot
|
|
flatpak_snapshot = users_collection.flatpak_snapshot
|
|
snap_snapshot = users_collection.snap_snapshot
|
|
|
|
# -------------------------
|
|
# Container image inventory (Docker/Podman image caches)
|
|
# -------------------------
|
|
container_images_snapshot = ContainerImagesCollector(context).collect()
|
|
|
|
# -------------------------
|
|
# Package manager config role
|
|
# - Debian: apt_config
|
|
# - Fedora/RHEL-like: dnf_config
|
|
# -------------------------
|
|
package_manager_config = PackageManagerConfigCollector(
|
|
context, seen_by_role
|
|
).collect()
|
|
apt_config_snapshot = package_manager_config.apt_config_snapshot
|
|
dnf_config_snapshot = package_manager_config.dnf_config_snapshot
|
|
|
|
# -------------------------
|
|
# etc_custom role (unowned /etc files not already attributed elsewhere)
|
|
# -------------------------
|
|
etc_notes: List[str] = []
|
|
etc_excluded: List[ExcludedFile] = []
|
|
etc_managed: List[ManagedFile] = []
|
|
etc_role_name = "etc_custom"
|
|
|
|
# Files already captured by earlier roles. Use the global set so we never
|
|
# end up with the same destination path managed by multiple roles.
|
|
already: Set[str] = captured_global
|
|
|
|
# Maps for re-attributing shared snippets (cron.d/logrotate.d) to existing roles.
|
|
svc_by_role: Dict[str, ServiceSnapshot] = {s.role_name: s for s in service_snaps}
|
|
pkg_by_role: Dict[str, PackageSnapshot] = {p.role_name: p for p in pkg_snaps}
|
|
|
|
# Package name -> role_name for manually-installed package roles.
|
|
pkg_name_to_role: Dict[str, str] = {p.package: p.role_name for p in pkg_snaps}
|
|
|
|
# Package name -> list of service role names that reference it.
|
|
pkg_to_service_roles: Dict[str, List[str]] = {}
|
|
for s in service_snaps:
|
|
for pkg in s.packages:
|
|
pkg_to_service_roles.setdefault(pkg, []).append(s.role_name)
|
|
|
|
# Alias -> role mapping used as a fallback when package ownership is missing.
|
|
# Prefer service roles over package roles when both would match.
|
|
alias_ranked: Dict[str, tuple[int, str]] = {}
|
|
|
|
def _add_alias(alias: str, role_name: str, *, priority: int) -> None:
|
|
key = safe_name(alias)
|
|
if not key:
|
|
return
|
|
cur = alias_ranked.get(key)
|
|
if (
|
|
cur is None
|
|
or priority < cur[0]
|
|
or (priority == cur[0] and role_name < cur[1])
|
|
):
|
|
alias_ranked[key] = (priority, role_name)
|
|
|
|
for role_name, aliases in service_role_aliases.items():
|
|
for a in aliases:
|
|
_add_alias(a, role_name, priority=0)
|
|
|
|
for p in pkg_snaps:
|
|
_add_alias(p.package, p.role_name, priority=1)
|
|
|
|
def _target_role_for_shared_snippet(path: str) -> Optional[tuple[str, str]]:
|
|
"""If `path` is a shared snippet, return (role_name, reason) to attach to.
|
|
|
|
This is used primarily for /etc/logrotate.d/* and /etc/cron.d/* where
|
|
files are "owned" by many packages but people tend to reason about them
|
|
per service.
|
|
|
|
Resolution order:
|
|
1) package owner -> service role (if any service references the package)
|
|
2) package owner -> package role (manual package role exists)
|
|
3) basename/stem alias match -> preferred role
|
|
"""
|
|
if path.startswith("/etc/logrotate.d/"):
|
|
tag = "logrotate_snippet"
|
|
elif path.startswith("/etc/cron.d/"):
|
|
tag = "cron_snippet"
|
|
else:
|
|
return None
|
|
|
|
base = os.path.basename(path)
|
|
candidates: List[str] = [base]
|
|
if "." in base:
|
|
candidates.append(base.split(".", 1)[0])
|
|
|
|
seen: Set[str] = set()
|
|
uniq: List[str] = []
|
|
for c in candidates:
|
|
if c and c not in seen:
|
|
seen.add(c)
|
|
uniq.append(c)
|
|
|
|
pkg = backend.owner_of_path(path)
|
|
if pkg:
|
|
svc_roles = sorted(set(pkg_to_service_roles.get(pkg, [])))
|
|
if svc_roles:
|
|
# If multiple service roles reference the same package, prefer
|
|
# the role that most closely matches the snippet name (basename
|
|
# or stem). This avoids surprising attributions such as an
|
|
# AppArmor loader role "claiming" a cron/logrotate snippet
|
|
# that is clearly named after another package/service.
|
|
if len(svc_roles) > 1:
|
|
# Direct role-name matches first.
|
|
for c in [pkg, *uniq]:
|
|
rn = safe_name(c)
|
|
if rn in svc_roles:
|
|
return (rn, tag)
|
|
# Next, use the alias map if it points at one of the roles.
|
|
for c in [pkg, *uniq]:
|
|
hit = alias_ranked.get(safe_name(c))
|
|
if hit is not None and hit[1] in svc_roles:
|
|
return (hit[1], tag)
|
|
|
|
# Deterministic fallback: lowest role name.
|
|
return (svc_roles[0], tag)
|
|
pkg_role = pkg_name_to_role.get(pkg)
|
|
if pkg_role:
|
|
return (pkg_role, tag)
|
|
|
|
for c in uniq:
|
|
key = safe_name(c)
|
|
hit = alias_ranked.get(key)
|
|
if hit is not None:
|
|
return (hit[1], tag)
|
|
|
|
return None
|
|
|
|
def _lists_for_role(role_name: str) -> tuple[List[ManagedFile], List[ExcludedFile]]:
|
|
if role_name in svc_by_role:
|
|
snap = svc_by_role[role_name]
|
|
return (snap.managed_files, snap.excluded)
|
|
if role_name in pkg_by_role:
|
|
snap = pkg_by_role[role_name]
|
|
return (snap.managed_files, snap.excluded)
|
|
# Fallback (shouldn't normally happen): attribute to etc_custom.
|
|
return (etc_managed, etc_excluded)
|
|
|
|
# Capture essential system config/state (even if package-owned).
|
|
etc_role_seen = seen_by_role.setdefault(etc_role_name, set())
|
|
for path, reason in system_paths.iter_system_capture_paths():
|
|
if path in already:
|
|
continue
|
|
|
|
target = _target_role_for_shared_snippet(path)
|
|
if target is not None:
|
|
role_for_copy, reason_for_role = target
|
|
managed_out, excluded_out = _lists_for_role(role_for_copy)
|
|
role_seen = seen_by_role.setdefault(role_for_copy, set())
|
|
else:
|
|
role_for_copy, reason_for_role = (etc_role_name, reason)
|
|
managed_out, excluded_out = (etc_managed, etc_excluded)
|
|
role_seen = etc_role_seen
|
|
|
|
capture_file(
|
|
bundle_dir=bundle_dir,
|
|
role_name=role_for_copy,
|
|
abs_path=path,
|
|
reason=reason_for_role,
|
|
policy=policy,
|
|
path_filter=path_filter,
|
|
managed_out=managed_out,
|
|
excluded_out=excluded_out,
|
|
seen_role=role_seen,
|
|
seen_global=captured_global,
|
|
)
|
|
|
|
# Walk /etc for remaining unowned config-ish files
|
|
scanned = 0
|
|
for dirpath, _, filenames in os.walk("/etc"):
|
|
for fn in filenames:
|
|
path = os.path.join(dirpath, fn)
|
|
if backend.is_pkg_config_path(path):
|
|
continue
|
|
if path in already:
|
|
continue
|
|
if path in owned_etc:
|
|
continue
|
|
if not os.path.isfile(path) or os.path.islink(path):
|
|
continue
|
|
if not system_paths.is_confish(path):
|
|
continue
|
|
|
|
target = _target_role_for_shared_snippet(path)
|
|
if target is not None:
|
|
role_for_copy, reason_for_role = target
|
|
managed_out, excluded_out = _lists_for_role(role_for_copy)
|
|
role_seen = seen_by_role.setdefault(role_for_copy, set())
|
|
else:
|
|
role_for_copy, reason_for_role = (etc_role_name, "custom_unowned")
|
|
managed_out, excluded_out = (etc_managed, etc_excluded)
|
|
role_seen = etc_role_seen
|
|
|
|
if capture_file(
|
|
bundle_dir=bundle_dir,
|
|
role_name=role_for_copy,
|
|
abs_path=path,
|
|
reason=reason_for_role,
|
|
policy=policy,
|
|
path_filter=path_filter,
|
|
managed_out=managed_out,
|
|
excluded_out=excluded_out,
|
|
seen_role=role_seen,
|
|
seen_global=captured_global,
|
|
):
|
|
scanned += 1
|
|
if scanned >= system_paths.MAX_FILES_CAP:
|
|
etc_notes.append(
|
|
f"Reached file cap ({system_paths.MAX_FILES_CAP}) while scanning /etc for unowned files."
|
|
)
|
|
break
|
|
if scanned >= system_paths.MAX_FILES_CAP:
|
|
break
|
|
|
|
etc_custom_snapshot = EtcCustomSnapshot(
|
|
role_name=etc_role_name,
|
|
managed_files=etc_managed,
|
|
excluded=etc_excluded,
|
|
notes=etc_notes,
|
|
)
|
|
|
|
# -------------------------
|
|
# usr_local_custom and extra_paths roles
|
|
# -------------------------
|
|
already_all: Set[str] = set(already)
|
|
for mf in etc_managed:
|
|
already_all.add(mf.path)
|
|
|
|
usr_local_custom_snapshot = UsrLocalCustomCollector(
|
|
context,
|
|
seen_by_role,
|
|
already_all,
|
|
).collect()
|
|
|
|
extra_paths_snapshot = ExtraPathsCollector(
|
|
context,
|
|
seen_by_role,
|
|
already_all,
|
|
include_paths=include_paths,
|
|
exclude_paths=exclude_paths,
|
|
).collect()
|
|
|
|
# -------------------------
|
|
# Inventory: packages (SBOM-ish)
|
|
# -------------------------
|
|
installed = installed_pkgs
|
|
|
|
manual_set: Set[str] = set(manual_pkgs or [])
|
|
|
|
pkg_units: Dict[str, Set[str]] = {}
|
|
pkg_roles_map: Dict[str, Set[str]] = {}
|
|
|
|
for svc in service_snaps:
|
|
for p in svc.packages:
|
|
pkg_units.setdefault(p, set()).add(svc.unit)
|
|
pkg_roles_map.setdefault(p, set()).add(svc.role_name)
|
|
|
|
pkg_role_names: Dict[str, List[str]] = {}
|
|
for ps in pkg_snaps:
|
|
pkg_roles_map.setdefault(ps.package, set()).add(ps.role_name)
|
|
pkg_role_names.setdefault(ps.package, []).append(ps.role_name)
|
|
|
|
pkg_names: Set[str] = set()
|
|
pkg_names |= manual_set
|
|
pkg_names |= set(pkg_units.keys())
|
|
pkg_names |= {ps.package for ps in pkg_snaps}
|
|
pkg_names |= set(firewall_runtime_snapshot.packages or [])
|
|
|
|
packages_inventory: Dict[str, Dict[str, object]] = {}
|
|
for pkg in sorted(pkg_names):
|
|
installs = installed.get(pkg, []) or []
|
|
arches = sorted({i.get("arch") for i in installs if i.get("arch")})
|
|
vers = sorted({i.get("version") for i in installs if i.get("version")})
|
|
version: Optional[str] = vers[0] if len(vers) == 1 else None
|
|
section = package_section_from_installations(installs)
|
|
|
|
observed: List[Dict[str, str]] = []
|
|
if pkg in manual_set:
|
|
observed.append({"kind": "user_installed"})
|
|
for unit in sorted(pkg_units.get(pkg, set())):
|
|
observed.append({"kind": "systemd_unit", "ref": unit})
|
|
for rn in sorted(set(pkg_role_names.get(pkg, []))):
|
|
observed.append({"kind": "package_role", "ref": rn})
|
|
if pkg in set(firewall_runtime_snapshot.packages or []):
|
|
observed.append(
|
|
{"kind": "firewall_runtime", "ref": firewall_runtime_snapshot.role_name}
|
|
)
|
|
pkg_roles_map.setdefault(pkg, set()).add(
|
|
firewall_runtime_snapshot.role_name
|
|
)
|
|
|
|
roles = sorted(pkg_roles_map.get(pkg, set()))
|
|
|
|
packages_inventory[pkg] = {
|
|
"version": version,
|
|
"arches": arches,
|
|
"installations": installs,
|
|
"section": section,
|
|
"observed_via": observed,
|
|
"roles": roles,
|
|
}
|
|
|
|
# Ensure every role has explicit managed_dirs for parent directories of managed files.
|
|
# This lets the manifest create directories with owner/group/mode (ansible-lint friendly)
|
|
# without a separate "mkdir without perms" task.
|
|
users_snapshot.managed_dirs = _merge_parent_dirs(
|
|
users_snapshot.managed_dirs, users_snapshot.managed_files, policy=policy
|
|
)
|
|
for s in service_snaps:
|
|
s.managed_dirs = _merge_parent_dirs(
|
|
s.managed_dirs,
|
|
s.managed_files,
|
|
policy=policy,
|
|
extra_paths=[ml.path for ml in (s.managed_links or [])],
|
|
)
|
|
for p in pkg_snaps:
|
|
p.managed_dirs = _merge_parent_dirs(
|
|
p.managed_dirs,
|
|
p.managed_files,
|
|
policy=policy,
|
|
extra_paths=[ml.path for ml in (p.managed_links or [])],
|
|
)
|
|
|
|
if apt_config_snapshot:
|
|
apt_config_snapshot.managed_dirs = _merge_parent_dirs(
|
|
apt_config_snapshot.managed_dirs,
|
|
apt_config_snapshot.managed_files,
|
|
policy=policy,
|
|
)
|
|
if dnf_config_snapshot:
|
|
dnf_config_snapshot.managed_dirs = _merge_parent_dirs(
|
|
dnf_config_snapshot.managed_dirs,
|
|
dnf_config_snapshot.managed_files,
|
|
policy=policy,
|
|
)
|
|
if etc_custom_snapshot:
|
|
etc_custom_snapshot.managed_dirs = _merge_parent_dirs(
|
|
etc_custom_snapshot.managed_dirs,
|
|
etc_custom_snapshot.managed_files,
|
|
policy=policy,
|
|
)
|
|
if usr_local_custom_snapshot:
|
|
usr_local_custom_snapshot.managed_dirs = _merge_parent_dirs(
|
|
usr_local_custom_snapshot.managed_dirs,
|
|
usr_local_custom_snapshot.managed_files,
|
|
policy=policy,
|
|
)
|
|
if extra_paths_snapshot:
|
|
extra_paths_snapshot.managed_dirs = _merge_parent_dirs(
|
|
extra_paths_snapshot.managed_dirs,
|
|
extra_paths_snapshot.managed_files,
|
|
policy=policy,
|
|
)
|
|
|
|
state = {
|
|
"enroll": {
|
|
"version": get_enroll_version(),
|
|
"harvest_time": time.time_ns(),
|
|
},
|
|
"host": {
|
|
"hostname": os.uname().nodename,
|
|
"os": platform.os_family,
|
|
"pkg_backend": backend.name,
|
|
"os_release": platform.os_release,
|
|
},
|
|
"inventory": {
|
|
"packages": packages_inventory,
|
|
},
|
|
"roles": {
|
|
"users": asdict(users_snapshot),
|
|
"flatpak": asdict(flatpak_snapshot),
|
|
"snap": asdict(snap_snapshot),
|
|
"container_images": asdict(container_images_snapshot),
|
|
"services": [asdict(s) for s in service_snaps],
|
|
"packages": [asdict(p) for p in pkg_snaps],
|
|
"apt_config": asdict(apt_config_snapshot),
|
|
"dnf_config": asdict(dnf_config_snapshot),
|
|
"firewall_runtime": asdict(firewall_runtime_snapshot),
|
|
"sysctl": asdict(sysctl_snapshot),
|
|
"etc_custom": asdict(etc_custom_snapshot),
|
|
"usr_local_custom": asdict(usr_local_custom_snapshot),
|
|
"extra_paths": asdict(extra_paths_snapshot),
|
|
},
|
|
}
|
|
|
|
return str(write_state(bundle_dir, state))
|