Miguel Jacq
a2be708a31
Introduce --ask-become-pass or -K to support password-required sudo on remote hosts, just like Ansible. It will also fall back to this prompt if a password is required but the arg wasn't passed in. With thanks to slhck from HN for the initial patch, advice and feedback.
937 lines
34 KiB
Python
937 lines
34 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import configparser
|
|
import os
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from .cache import new_harvest_cache_dir
|
|
from .diff import compare_harvests, format_report, post_webhook, send_email
|
|
from .harvest import harvest
|
|
from .manifest import manifest
|
|
from .remote import remote_harvest, RemoteSudoPasswordRequired
|
|
from .sopsutil import SopsError, encrypt_file_binary
|
|
from .version import get_enroll_version
|
|
|
|
|
|
def _discover_config_path(argv: list[str]) -> Optional[Path]:
|
|
"""Return the config path to use, if any.
|
|
|
|
Precedence:
|
|
1) --no-config disables loading.
|
|
2) --config PATH (or -c PATH)
|
|
3) $ENROLL_CONFIG
|
|
4) ./enroll.ini, ./.enroll.ini
|
|
5) $XDG_CONFIG_HOME/enroll/enroll.ini (or ~/.config/enroll/enroll.ini)
|
|
|
|
The config file is optional; if no file is found, returns None.
|
|
"""
|
|
|
|
# Quick scan for explicit flags without needing to build the full parser.
|
|
if "--no-config" in argv:
|
|
return None
|
|
|
|
def _value_after(flag: str) -> Optional[str]:
|
|
try:
|
|
i = argv.index(flag)
|
|
except ValueError:
|
|
return None
|
|
if i + 1 >= len(argv):
|
|
return None
|
|
return argv[i + 1]
|
|
|
|
p = _value_after("--config") or _value_after("-c")
|
|
if p:
|
|
return Path(p).expanduser()
|
|
|
|
envp = os.environ.get("ENROLL_CONFIG")
|
|
if envp:
|
|
return Path(envp).expanduser()
|
|
|
|
cwd = Path.cwd()
|
|
for name in ("enroll.ini", ".enroll.ini"):
|
|
cp = cwd / name
|
|
if cp.exists() and cp.is_file():
|
|
return cp
|
|
|
|
xdg = os.environ.get("XDG_CONFIG_HOME")
|
|
if xdg:
|
|
base = Path(xdg).expanduser()
|
|
else:
|
|
base = Path.home() / ".config"
|
|
cp = base / "enroll" / "enroll.ini"
|
|
if cp.exists() and cp.is_file():
|
|
return cp
|
|
|
|
return None
|
|
|
|
|
|
def _parse_bool(s: str) -> Optional[bool]:
|
|
v = str(s).strip().lower()
|
|
if v in {"1", "true", "yes", "y", "on"}:
|
|
return True
|
|
if v in {"0", "false", "no", "n", "off"}:
|
|
return False
|
|
return None
|
|
|
|
|
|
def _action_lookup(p: argparse.ArgumentParser) -> dict[str, argparse.Action]:
|
|
"""Map config keys -> argparse actions for a parser.
|
|
|
|
Accepts both dest names and long option names without leading dashes,
|
|
normalized with '-' -> '_'.
|
|
"""
|
|
|
|
m: dict[str, argparse.Action] = {}
|
|
for a in p._actions: # noqa: SLF001 (argparse internal)
|
|
if not getattr(a, "dest", None):
|
|
continue
|
|
dest = str(a.dest).strip().lower()
|
|
if dest:
|
|
m[dest] = a
|
|
for opt in getattr(a, "option_strings", []) or []:
|
|
k = opt.lstrip("-").strip().lower()
|
|
if k:
|
|
m[k.replace("-", "_")] = a
|
|
m[k] = a
|
|
return m
|
|
|
|
|
|
def _choose_flag(a: argparse.Action) -> Optional[str]:
|
|
# Prefer a long flag if available (e.g. --dangerous over -d)
|
|
for s in getattr(a, "option_strings", []) or []:
|
|
if s.startswith("--"):
|
|
return s
|
|
for s in getattr(a, "option_strings", []) or []:
|
|
return s
|
|
return None
|
|
|
|
|
|
def _split_list_value(v: str) -> list[str]:
|
|
# Support comma-separated and/or multi-line lists.
|
|
raw = str(v)
|
|
if "\n" in raw:
|
|
parts = [p.strip() for p in raw.splitlines()]
|
|
return [p for p in parts if p]
|
|
if "," in raw:
|
|
parts = [p.strip() for p in raw.split(",")]
|
|
return [p for p in parts if p]
|
|
raw = raw.strip()
|
|
return [raw] if raw else []
|
|
|
|
|
|
def _section_to_argv(
|
|
p: argparse.ArgumentParser, cfg: configparser.ConfigParser, section: str
|
|
) -> list[str]:
|
|
"""Translate an INI section into argv tokens for this parser."""
|
|
if not cfg.has_section(section):
|
|
return []
|
|
|
|
lookup = _action_lookup(p)
|
|
out: list[str] = []
|
|
|
|
for k, v in cfg.items(section):
|
|
key = str(k).strip().lower().replace("-", "_")
|
|
# Avoid recursion / confusing self-configuration.
|
|
if key in {"config", "no_config"}:
|
|
continue
|
|
|
|
a = lookup.get(key)
|
|
if not a:
|
|
# Unknown keys are ignored (but we try to be helpful).
|
|
print(
|
|
f"warning: config [{section}] contains unknown option '{k}' (ignored)",
|
|
file=sys.stderr,
|
|
)
|
|
continue
|
|
|
|
flag = _choose_flag(a)
|
|
if not flag:
|
|
continue
|
|
|
|
# Boolean flags
|
|
if isinstance(a, argparse._StoreTrueAction): # noqa: SLF001
|
|
b = _parse_bool(v)
|
|
if b is True:
|
|
out.append(flag)
|
|
continue
|
|
if isinstance(a, argparse._StoreFalseAction): # noqa: SLF001
|
|
b = _parse_bool(v)
|
|
if b is False:
|
|
out.append(flag)
|
|
continue
|
|
|
|
# Repeated options
|
|
if isinstance(a, argparse._AppendAction): # noqa: SLF001
|
|
for item in _split_list_value(v):
|
|
out.extend([flag, item])
|
|
continue
|
|
|
|
# Count flags (rare, but easy to support)
|
|
if isinstance(a, argparse._CountAction): # noqa: SLF001
|
|
b = _parse_bool(v)
|
|
if b is True:
|
|
out.append(flag)
|
|
else:
|
|
try:
|
|
n = int(str(v).strip())
|
|
except ValueError:
|
|
n = 0
|
|
out.extend([flag] * max(0, n))
|
|
continue
|
|
|
|
# Standard scalar options
|
|
sval = str(v).strip()
|
|
if sval:
|
|
out.extend([flag, sval])
|
|
|
|
return out
|
|
|
|
|
|
def _inject_config_argv(
|
|
argv: list[str],
|
|
*,
|
|
cfg_path: Optional[Path],
|
|
root_parser: argparse.ArgumentParser,
|
|
subparsers: dict[str, argparse.ArgumentParser],
|
|
) -> list[str]:
|
|
"""Return argv with config-derived tokens inserted.
|
|
|
|
We insert:
|
|
- [enroll] options before the subcommand
|
|
- [<subcommand>] options immediately after the subcommand token
|
|
|
|
CLI flags always win because they come later in argv.
|
|
"""
|
|
|
|
if not cfg_path:
|
|
return argv
|
|
cfg_path = Path(cfg_path).expanduser()
|
|
if not (cfg_path.exists() and cfg_path.is_file()):
|
|
return argv
|
|
|
|
cfg = configparser.ConfigParser()
|
|
try:
|
|
cfg.read(cfg_path, encoding="utf-8")
|
|
except (OSError, configparser.Error) as e:
|
|
raise SystemExit(f"error: failed to read config file {cfg_path}: {e}")
|
|
|
|
global_tokens = _section_to_argv(root_parser, cfg, "enroll")
|
|
|
|
# Find the subcommand token position.
|
|
cmd_pos: Optional[int] = None
|
|
cmd_name: Optional[str] = None
|
|
for i, tok in enumerate(argv):
|
|
if tok in subparsers:
|
|
cmd_pos = i
|
|
cmd_name = tok
|
|
break
|
|
if cmd_pos is None or cmd_name is None:
|
|
# No subcommand found (argparse will handle the error); only apply global.
|
|
return global_tokens + argv
|
|
|
|
cmd_tokens = _section_to_argv(subparsers[cmd_name], cfg, cmd_name)
|
|
# Also accept section names with '_' in place of '-' (e.g. [single_shot])
|
|
if "-" in cmd_name:
|
|
alt = cmd_name.replace("-", "_")
|
|
if alt != cmd_name:
|
|
cmd_tokens += _section_to_argv(subparsers[cmd_name], cfg, alt)
|
|
|
|
return global_tokens + argv[: cmd_pos + 1] + cmd_tokens + argv[cmd_pos + 1 :]
|
|
|
|
|
|
def _resolve_sops_out_file(out: Optional[str], *, hint: str) -> Path:
|
|
"""Resolve an output *file* path for --sops mode.
|
|
|
|
If `out` looks like a directory (or points to an existing directory), we
|
|
place the encrypted harvest inside it as harvest.tar.gz.sops.
|
|
"""
|
|
if out:
|
|
p = Path(out).expanduser()
|
|
if p.exists() and p.is_dir():
|
|
return p / "harvest.tar.gz.sops"
|
|
# Heuristic: treat paths with a suffix as files; otherwise directories.
|
|
if p.suffix:
|
|
return p
|
|
return p / "harvest.tar.gz.sops"
|
|
|
|
# Default: use a secure cache directory.
|
|
d = new_harvest_cache_dir(hint=hint).dir
|
|
return d / "harvest.tar.gz.sops"
|
|
|
|
|
|
def _tar_dir_to(path_dir: Path, tar_path: Path) -> None:
|
|
tar_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with tarfile.open(tar_path, mode="w:gz") as tf:
|
|
# Keep a stable on-disk layout when extracted: state.json + artifacts/
|
|
tf.add(str(path_dir), arcname=".")
|
|
|
|
|
|
def _encrypt_harvest_dir_to_sops(
|
|
bundle_dir: Path, out_file: Path, fps: list[str]
|
|
) -> Path:
|
|
out_file = Path(out_file)
|
|
out_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Create the tarball alongside the output file (keeps filesystem permissions/locality sane).
|
|
fd, tmp_tgz = tempfile.mkstemp(
|
|
prefix=".enroll-harvest-", suffix=".tar.gz", dir=str(out_file.parent)
|
|
)
|
|
os.close(fd)
|
|
try:
|
|
_tar_dir_to(bundle_dir, Path(tmp_tgz))
|
|
encrypt_file_binary(Path(tmp_tgz), out_file, pgp_fingerprints=fps, mode=0o600)
|
|
finally:
|
|
try:
|
|
os.unlink(tmp_tgz)
|
|
except FileNotFoundError:
|
|
pass
|
|
return out_file
|
|
|
|
|
|
def _add_common_manifest_args(p: argparse.ArgumentParser) -> None:
|
|
p.add_argument(
|
|
"--fqdn",
|
|
help="Host FQDN/name for site-mode output (creates inventory/, inventory/host_vars/, playbooks/).",
|
|
)
|
|
g = p.add_mutually_exclusive_group()
|
|
g.add_argument(
|
|
"--jinjaturtle",
|
|
action="store_true",
|
|
help="Attempt jinjaturtle template integration (it will error if jinjaturtle is not found on PATH).",
|
|
)
|
|
g.add_argument(
|
|
"--no-jinjaturtle",
|
|
action="store_true",
|
|
help="Do not use jinjaturtle integration, even if it is installed.",
|
|
)
|
|
|
|
|
|
def _jt_mode(args: argparse.Namespace) -> str:
|
|
if getattr(args, "jinjaturtle", False):
|
|
return "on"
|
|
if getattr(args, "no_jinjaturtle", False):
|
|
return "off"
|
|
return "auto"
|
|
|
|
|
|
def _add_config_args(p: argparse.ArgumentParser) -> None:
|
|
p.add_argument(
|
|
"-c",
|
|
"--config",
|
|
help=(
|
|
"Path to an INI config file for default options. If omitted, enroll will look for "
|
|
"./enroll.ini, ./.enroll.ini, or ~/.config/enroll/enroll.ini (or $XDG_CONFIG_HOME/enroll/enroll.ini)."
|
|
),
|
|
)
|
|
p.add_argument(
|
|
"--no-config",
|
|
action="store_true",
|
|
help="Do not load any INI config file (even if one would be auto-discovered).",
|
|
)
|
|
|
|
|
|
def _add_remote_args(p: argparse.ArgumentParser) -> None:
|
|
p.add_argument(
|
|
"--remote-host",
|
|
help="SSH host to run harvesting on (if set, harvest runs remotely and is pulled locally).",
|
|
)
|
|
p.add_argument(
|
|
"--remote-port",
|
|
type=int,
|
|
default=22,
|
|
help="SSH port for --remote-host (default: 22).",
|
|
)
|
|
p.add_argument(
|
|
"--remote-user",
|
|
default=os.environ.get("USER") or None,
|
|
help="SSH username for --remote-host (default: local $USER).",
|
|
)
|
|
|
|
# Align terminology with Ansible: "become" == sudo.
|
|
p.add_argument(
|
|
"--ask-become-pass",
|
|
"-K",
|
|
action="store_true",
|
|
help=(
|
|
"Prompt for the remote sudo (become) password when using --remote-host "
|
|
"(similar to ansible --ask-become-pass)."
|
|
),
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser(prog="enroll")
|
|
ap.add_argument(
|
|
"-v",
|
|
"--version",
|
|
action="version",
|
|
version=f"{get_enroll_version()}",
|
|
)
|
|
_add_config_args(ap)
|
|
sub = ap.add_subparsers(dest="cmd", required=True)
|
|
|
|
h = sub.add_parser("harvest", help="Harvest service/package/config state")
|
|
_add_config_args(h)
|
|
_add_remote_args(h)
|
|
h.add_argument(
|
|
"--out",
|
|
help=(
|
|
"Harvest output directory. If --sops is set, this may be either a directory "
|
|
"(an encrypted file named harvest.tar.gz.sops will be created inside) or a file path."
|
|
),
|
|
)
|
|
h.add_argument(
|
|
"--dangerous",
|
|
action="store_true",
|
|
help="Collect files more aggressively (may include secrets). Disables secret-avoidance checks.",
|
|
)
|
|
h.add_argument(
|
|
"--include-path",
|
|
action="append",
|
|
default=[],
|
|
metavar="PATTERN",
|
|
help=(
|
|
"Include extra file paths to harvest (repeatable). Supports globs (including '**') and regex via 're:<regex>'. "
|
|
"Included files are still filtered by IgnorePolicy unless --dangerous is used."
|
|
),
|
|
)
|
|
h.add_argument(
|
|
"--exclude-path",
|
|
action="append",
|
|
default=[],
|
|
metavar="PATTERN",
|
|
help=(
|
|
"Exclude file paths from harvesting (repeatable). Supports globs (including '**') and regex via 're:<regex>'. "
|
|
"Excludes apply to all harvesting, including defaults."
|
|
),
|
|
)
|
|
|
|
h.add_argument(
|
|
"--sops",
|
|
nargs="+",
|
|
metavar="GPG_FINGERPRINT",
|
|
help=(
|
|
"Encrypt the harvest output as a SOPS-encrypted tarball using the given GPG fingerprint(s). "
|
|
"Requires `sops` on PATH."
|
|
),
|
|
)
|
|
h.add_argument(
|
|
"--no-sudo",
|
|
action="store_true",
|
|
help="Don't use sudo on the remote host (when using --remote options). This may result in a limited harvest due to permission restrictions.",
|
|
)
|
|
|
|
m = sub.add_parser("manifest", help="Render Ansible roles from a harvest")
|
|
_add_config_args(m)
|
|
m.add_argument(
|
|
"--harvest",
|
|
required=True,
|
|
help=(
|
|
"Path to the directory created by the harvest command, or (with --sops) "
|
|
"a SOPS-encrypted harvest tarball."
|
|
),
|
|
)
|
|
m.add_argument(
|
|
"--out",
|
|
required=True,
|
|
help=(
|
|
"Output location for the generated manifest. In plain mode this is a directory. "
|
|
"In --sops mode this may be either a directory (an encrypted file named manifest.tar.gz.sops will be created inside) "
|
|
"or a file path."
|
|
),
|
|
)
|
|
m.add_argument(
|
|
"--sops",
|
|
nargs="+",
|
|
metavar="GPG_FINGERPRINT",
|
|
help=(
|
|
"In --sops mode, decrypt the harvest using `sops -d` (if the harvest is an encrypted file) "
|
|
"and then bundle+encrypt the entire generated manifest output into a single SOPS-encrypted tarball "
|
|
"(binary) using the given GPG fingerprint(s). Requires `sops` on PATH."
|
|
),
|
|
)
|
|
_add_common_manifest_args(m)
|
|
|
|
s = sub.add_parser(
|
|
"single-shot", help="Harvest state, then manifest Ansible code, in one shot"
|
|
)
|
|
_add_config_args(s)
|
|
_add_remote_args(s)
|
|
s.add_argument(
|
|
"--harvest",
|
|
help=(
|
|
"Where to place the harvest. In plain mode this is a directory; in --sops mode this may be "
|
|
"a directory or a file path (an encrypted file is produced)."
|
|
),
|
|
)
|
|
s.add_argument(
|
|
"--dangerous",
|
|
action="store_true",
|
|
help="Collect files more aggressively (may include secrets). Disables secret-avoidance checks.",
|
|
)
|
|
s.add_argument(
|
|
"--include-path",
|
|
action="append",
|
|
default=[],
|
|
metavar="PATTERN",
|
|
help=(
|
|
"Include extra file paths to harvest (repeatable). Supports globs (including '**') and regex via 're:<regex>'. "
|
|
"Included files are still filtered by IgnorePolicy unless --dangerous is used."
|
|
),
|
|
)
|
|
s.add_argument(
|
|
"--exclude-path",
|
|
action="append",
|
|
default=[],
|
|
metavar="PATTERN",
|
|
help=(
|
|
"Exclude file paths from harvesting (repeatable). Supports globs (including '**') and regex via 're:<regex>'. "
|
|
"Excludes apply to all harvesting, including defaults."
|
|
),
|
|
)
|
|
|
|
s.add_argument(
|
|
"--sops",
|
|
nargs="+",
|
|
metavar="GPG_FINGERPRINT",
|
|
help=(
|
|
"Encrypt the harvest as a SOPS-encrypted tarball, and bundle+encrypt the manifest output in --out "
|
|
"(same behaviour as `harvest --sops` and `manifest --sops`)."
|
|
),
|
|
)
|
|
s.add_argument(
|
|
"--no-sudo",
|
|
action="store_true",
|
|
help="Don't use sudo on the remote host (when using --remote options). This may result in a limited harvest due to permission restrictions.",
|
|
)
|
|
s.add_argument(
|
|
"--out",
|
|
required=True,
|
|
help=(
|
|
"Output location for the generated manifest. In plain mode this is a directory. "
|
|
"In --sops mode this may be either a directory (an encrypted file named manifest.tar.gz.sops will be created inside) "
|
|
"or a file path."
|
|
),
|
|
)
|
|
_add_common_manifest_args(s)
|
|
|
|
d = sub.add_parser("diff", help="Compare two harvests and report differences")
|
|
_add_config_args(d)
|
|
d.add_argument(
|
|
"--old",
|
|
required=True,
|
|
help=(
|
|
"Old/baseline harvest (directory, a path to state.json, a tarball, or a SOPS-encrypted bundle)."
|
|
),
|
|
)
|
|
d.add_argument(
|
|
"--new",
|
|
required=True,
|
|
help=(
|
|
"New/current harvest (directory, a path to state.json, a tarball, or a SOPS-encrypted bundle)."
|
|
),
|
|
)
|
|
d.add_argument(
|
|
"--sops",
|
|
action="store_true",
|
|
help="Allow SOPS-encrypted harvest bundle inputs (requires `sops` on PATH).",
|
|
)
|
|
d.add_argument(
|
|
"--format",
|
|
choices=["text", "markdown", "json"],
|
|
default="text",
|
|
help="Report output format (default: text).",
|
|
)
|
|
d.add_argument(
|
|
"--out",
|
|
help="Write the report to this file instead of stdout.",
|
|
)
|
|
d.add_argument(
|
|
"--exit-code",
|
|
action="store_true",
|
|
help="Exit with status 2 if differences are detected.",
|
|
)
|
|
d.add_argument(
|
|
"--notify-always",
|
|
action="store_true",
|
|
help="Send webhook/email even when there are no differences.",
|
|
)
|
|
d.add_argument(
|
|
"--webhook",
|
|
help="POST the report to this URL (only when differences are detected, unless --notify-always).",
|
|
)
|
|
d.add_argument(
|
|
"--webhook-format",
|
|
choices=["json", "text", "markdown"],
|
|
default="json",
|
|
help="Payload format for --webhook (default: json).",
|
|
)
|
|
d.add_argument(
|
|
"--webhook-header",
|
|
action="append",
|
|
default=[],
|
|
metavar="K:V",
|
|
help="Extra HTTP header for --webhook (repeatable), e.g. 'Authorization: Bearer ...'.",
|
|
)
|
|
d.add_argument(
|
|
"--email-to",
|
|
action="append",
|
|
default=[],
|
|
help="Email the report to this address (repeatable; only when differences are detected unless --notify-always).",
|
|
)
|
|
d.add_argument(
|
|
"--email-from",
|
|
help="From address for --email-to (default: enroll@<hostname>).",
|
|
)
|
|
d.add_argument(
|
|
"--email-subject",
|
|
help="Subject for --email-to (default: 'enroll diff report').",
|
|
)
|
|
d.add_argument(
|
|
"--smtp",
|
|
help="SMTP server host[:port] for --email-to. If omitted, uses local sendmail.",
|
|
)
|
|
d.add_argument(
|
|
"--smtp-user",
|
|
help="SMTP username (optional).",
|
|
)
|
|
d.add_argument(
|
|
"--smtp-password-env",
|
|
help="Environment variable containing SMTP password (optional).",
|
|
)
|
|
|
|
argv = sys.argv[1:]
|
|
cfg_path = _discover_config_path(argv)
|
|
argv = _inject_config_argv(
|
|
argv,
|
|
cfg_path=cfg_path,
|
|
root_parser=ap,
|
|
subparsers={
|
|
"harvest": h,
|
|
"manifest": m,
|
|
"single-shot": s,
|
|
"diff": d,
|
|
},
|
|
)
|
|
args = ap.parse_args(argv)
|
|
|
|
try:
|
|
if args.cmd == "harvest":
|
|
sops_fps = getattr(args, "sops", None)
|
|
if args.remote_host:
|
|
if sops_fps:
|
|
out_file = _resolve_sops_out_file(args.out, hint=args.remote_host)
|
|
with tempfile.TemporaryDirectory(prefix="enroll-harvest-") as td:
|
|
tmp_bundle = Path(td) / "bundle"
|
|
tmp_bundle.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
os.chmod(tmp_bundle, 0o700)
|
|
except OSError:
|
|
pass
|
|
remote_harvest(
|
|
ask_become_pass=args.ask_become_pass,
|
|
local_out_dir=tmp_bundle,
|
|
remote_host=args.remote_host,
|
|
remote_port=int(args.remote_port),
|
|
remote_user=args.remote_user,
|
|
dangerous=bool(args.dangerous),
|
|
no_sudo=bool(args.no_sudo),
|
|
include_paths=list(getattr(args, "include_path", []) or []),
|
|
exclude_paths=list(getattr(args, "exclude_path", []) or []),
|
|
)
|
|
_encrypt_harvest_dir_to_sops(
|
|
tmp_bundle, out_file, list(sops_fps)
|
|
)
|
|
print(str(out_file))
|
|
else:
|
|
out_dir = (
|
|
Path(args.out)
|
|
if args.out
|
|
else new_harvest_cache_dir(hint=args.remote_host).dir
|
|
)
|
|
state = remote_harvest(
|
|
ask_become_pass=args.ask_become_pass,
|
|
local_out_dir=out_dir,
|
|
remote_host=args.remote_host,
|
|
remote_port=int(args.remote_port),
|
|
remote_user=args.remote_user,
|
|
dangerous=bool(args.dangerous),
|
|
no_sudo=bool(args.no_sudo),
|
|
include_paths=list(getattr(args, "include_path", []) or []),
|
|
exclude_paths=list(getattr(args, "exclude_path", []) or []),
|
|
)
|
|
print(str(state))
|
|
else:
|
|
if sops_fps:
|
|
out_file = _resolve_sops_out_file(args.out, hint="local")
|
|
with tempfile.TemporaryDirectory(prefix="enroll-harvest-") as td:
|
|
tmp_bundle = Path(td) / "bundle"
|
|
tmp_bundle.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
os.chmod(tmp_bundle, 0o700)
|
|
except OSError:
|
|
pass
|
|
harvest(
|
|
str(tmp_bundle),
|
|
dangerous=bool(args.dangerous),
|
|
include_paths=list(getattr(args, "include_path", []) or []),
|
|
exclude_paths=list(getattr(args, "exclude_path", []) or []),
|
|
)
|
|
_encrypt_harvest_dir_to_sops(
|
|
tmp_bundle, out_file, list(sops_fps)
|
|
)
|
|
print(str(out_file))
|
|
else:
|
|
if args.out:
|
|
out_dir = args.out
|
|
else:
|
|
out_dir = (
|
|
Path(args.out)
|
|
if args.out
|
|
else new_harvest_cache_dir(hint=args.remote_host).dir
|
|
)
|
|
path = harvest(
|
|
out_dir,
|
|
dangerous=bool(args.dangerous),
|
|
include_paths=list(getattr(args, "include_path", []) or []),
|
|
exclude_paths=list(getattr(args, "exclude_path", []) or []),
|
|
)
|
|
print(path)
|
|
elif args.cmd == "manifest":
|
|
out_enc = manifest(
|
|
args.harvest,
|
|
args.out,
|
|
fqdn=args.fqdn,
|
|
jinjaturtle=_jt_mode(args),
|
|
sops_fingerprints=getattr(args, "sops", None),
|
|
)
|
|
if getattr(args, "sops", None) and out_enc:
|
|
print(str(out_enc))
|
|
elif args.cmd == "diff":
|
|
report, has_changes = compare_harvests(
|
|
args.old,
|
|
args.new,
|
|
sops_mode=bool(getattr(args, "sops", False)),
|
|
)
|
|
|
|
txt = format_report(report, fmt=str(getattr(args, "format", "text")))
|
|
out_path = getattr(args, "out", None)
|
|
if out_path:
|
|
p = Path(out_path).expanduser()
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
p.write_text(txt, encoding="utf-8")
|
|
else:
|
|
print(txt, end="" if txt.endswith("\n") else "\n")
|
|
|
|
should_notify = has_changes or bool(getattr(args, "notify_always", False))
|
|
|
|
webhook = getattr(args, "webhook", None)
|
|
if webhook and should_notify:
|
|
wf = str(getattr(args, "webhook_format", "json"))
|
|
payload = format_report(report, fmt=wf)
|
|
body = payload.encode("utf-8")
|
|
headers = {}
|
|
if wf == "json":
|
|
headers["Content-Type"] = "application/json"
|
|
else:
|
|
headers["Content-Type"] = "text/plain; charset=utf-8"
|
|
for hv in getattr(args, "webhook_header", []) or []:
|
|
if ":" in hv:
|
|
k, v = hv.split(":", 1)
|
|
headers[k.strip()] = v.strip()
|
|
status, _resp = post_webhook(webhook, body, headers=headers)
|
|
if status and status >= 400:
|
|
raise SystemExit(f"error: webhook returned HTTP {status}")
|
|
|
|
to_addrs = getattr(args, "email_to", []) or []
|
|
if to_addrs and should_notify:
|
|
subject = getattr(args, "email_subject", None) or "enroll diff report"
|
|
smtp_pw = None
|
|
pw_env = getattr(args, "smtp_password_env", None)
|
|
if pw_env:
|
|
smtp_pw = os.environ.get(str(pw_env))
|
|
send_email(
|
|
to_addrs=list(to_addrs),
|
|
subject=str(subject),
|
|
body=txt,
|
|
from_addr=getattr(args, "email_from", None),
|
|
smtp=getattr(args, "smtp", None),
|
|
smtp_user=getattr(args, "smtp_user", None),
|
|
smtp_password=smtp_pw,
|
|
)
|
|
|
|
if getattr(args, "exit_code", False) and has_changes:
|
|
raise SystemExit(2)
|
|
elif args.cmd == "single-shot":
|
|
sops_fps = getattr(args, "sops", None)
|
|
if args.remote_host:
|
|
if sops_fps:
|
|
out_file = _resolve_sops_out_file(
|
|
args.harvest, hint=args.remote_host
|
|
)
|
|
with tempfile.TemporaryDirectory(prefix="enroll-harvest-") as td:
|
|
tmp_bundle = Path(td) / "bundle"
|
|
tmp_bundle.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
os.chmod(tmp_bundle, 0o700)
|
|
except OSError:
|
|
pass
|
|
remote_harvest(
|
|
ask_become_pass=args.ask_become_pass,
|
|
local_out_dir=tmp_bundle,
|
|
remote_host=args.remote_host,
|
|
remote_port=int(args.remote_port),
|
|
remote_user=args.remote_user,
|
|
dangerous=bool(args.dangerous),
|
|
no_sudo=bool(args.no_sudo),
|
|
include_paths=list(getattr(args, "include_path", []) or []),
|
|
exclude_paths=list(getattr(args, "exclude_path", []) or []),
|
|
)
|
|
_encrypt_harvest_dir_to_sops(
|
|
tmp_bundle, out_file, list(sops_fps)
|
|
)
|
|
|
|
manifest(
|
|
str(out_file),
|
|
args.out,
|
|
fqdn=args.fqdn,
|
|
jinjaturtle=_jt_mode(args),
|
|
sops_fingerprints=list(sops_fps),
|
|
)
|
|
if not args.harvest:
|
|
print(str(out_file))
|
|
else:
|
|
harvest_dir = (
|
|
Path(args.harvest)
|
|
if args.harvest
|
|
else new_harvest_cache_dir(hint=args.remote_host).dir
|
|
)
|
|
remote_harvest(
|
|
ask_become_pass=args.ask_become_pass,
|
|
local_out_dir=harvest_dir,
|
|
remote_host=args.remote_host,
|
|
remote_port=int(args.remote_port),
|
|
remote_user=args.remote_user,
|
|
dangerous=bool(args.dangerous),
|
|
no_sudo=bool(args.no_sudo),
|
|
include_paths=list(getattr(args, "include_path", []) or []),
|
|
exclude_paths=list(getattr(args, "exclude_path", []) or []),
|
|
)
|
|
manifest(
|
|
str(harvest_dir),
|
|
args.out,
|
|
fqdn=args.fqdn,
|
|
jinjaturtle=_jt_mode(args),
|
|
)
|
|
# For usability (when --harvest wasn't provided), print the harvest path.
|
|
if not args.harvest:
|
|
print(str(harvest_dir / "state.json"))
|
|
else:
|
|
if sops_fps:
|
|
out_file = _resolve_sops_out_file(args.harvest, hint="local")
|
|
with tempfile.TemporaryDirectory(prefix="enroll-harvest-") as td:
|
|
tmp_bundle = Path(td) / "bundle"
|
|
tmp_bundle.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
os.chmod(tmp_bundle, 0o700)
|
|
except OSError:
|
|
pass
|
|
harvest(
|
|
str(tmp_bundle),
|
|
dangerous=bool(args.dangerous),
|
|
include_paths=list(getattr(args, "include_path", []) or []),
|
|
exclude_paths=list(getattr(args, "exclude_path", []) or []),
|
|
)
|
|
_encrypt_harvest_dir_to_sops(
|
|
tmp_bundle, out_file, list(sops_fps)
|
|
)
|
|
|
|
manifest(
|
|
str(out_file),
|
|
args.out,
|
|
fqdn=args.fqdn,
|
|
jinjaturtle=_jt_mode(args),
|
|
sops_fingerprints=list(sops_fps),
|
|
)
|
|
if not args.harvest:
|
|
print(str(out_file))
|
|
else:
|
|
if not args.harvest:
|
|
raise SystemExit(
|
|
"error: --harvest is required unless --remote-host is set"
|
|
)
|
|
harvest(
|
|
args.harvest,
|
|
dangerous=bool(args.dangerous),
|
|
include_paths=list(getattr(args, "include_path", []) or []),
|
|
exclude_paths=list(getattr(args, "exclude_path", []) or []),
|
|
)
|
|
manifest(
|
|
args.harvest,
|
|
args.out,
|
|
fqdn=args.fqdn,
|
|
jinjaturtle=_jt_mode(args),
|
|
)
|
|
elif args.cmd == "diff":
|
|
report, has_changes = compare_harvests(
|
|
args.old, args.new, sops_mode=bool(getattr(args, "sops", False))
|
|
)
|
|
|
|
rendered = format_report(report, fmt=str(args.format))
|
|
if args.out:
|
|
Path(args.out).expanduser().write_text(rendered, encoding="utf-8")
|
|
else:
|
|
print(rendered, end="")
|
|
|
|
do_notify = bool(has_changes or getattr(args, "notify_always", False))
|
|
|
|
if do_notify and getattr(args, "webhook", None):
|
|
wf = str(getattr(args, "webhook_format", "json"))
|
|
body = format_report(report, fmt=wf).encode("utf-8")
|
|
headers = {"User-Agent": "enroll"}
|
|
if wf == "json":
|
|
headers["Content-Type"] = "application/json"
|
|
else:
|
|
headers["Content-Type"] = "text/plain; charset=utf-8"
|
|
for hv in getattr(args, "webhook_header", []) or []:
|
|
if ":" not in hv:
|
|
raise SystemExit(
|
|
"error: --webhook-header must be in the form 'K:V'"
|
|
)
|
|
k, v = hv.split(":", 1)
|
|
headers[k.strip()] = v.strip()
|
|
status, _ = post_webhook(str(args.webhook), body, headers=headers)
|
|
if status and status >= 400:
|
|
raise SystemExit(f"error: webhook returned HTTP {status}")
|
|
|
|
if do_notify and (getattr(args, "email_to", []) or []):
|
|
subject = getattr(args, "email_subject", None) or "enroll diff report"
|
|
smtp_password = None
|
|
pw_env = getattr(args, "smtp_password_env", None)
|
|
if pw_env:
|
|
smtp_password = os.environ.get(str(pw_env))
|
|
send_email(
|
|
to_addrs=list(getattr(args, "email_to", []) or []),
|
|
subject=str(subject),
|
|
body=rendered,
|
|
from_addr=getattr(args, "email_from", None),
|
|
smtp=getattr(args, "smtp", None),
|
|
smtp_user=getattr(args, "smtp_user", None),
|
|
smtp_password=smtp_password,
|
|
)
|
|
|
|
if getattr(args, "exit_code", False) and has_changes:
|
|
raise SystemExit(2)
|
|
except RemoteSudoPasswordRequired:
|
|
raise SystemExit(
|
|
"error: remote sudo requires a password. Re-run with --ask-become-pass."
|
|
) from None
|
|
except RuntimeError as e:
|
|
raise SystemExit(f"error: {e}") from None
|
|
except SopsError as e:
|
|
raise SystemExit(f"error: {e}") from None
|