203 lines
7.2 KiB
Python
Executable file
203 lines
7.2 KiB
Python
Executable file
#!/usr/bin/env python
|
|
# Copyright (C) 2021 Wes Barnett
|
|
|
|
# This program is free software; you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation; either version 2 of the License, or
|
|
# (at your option) any later version.
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
|
|
# You should have received a copy of the GNU General Public License along
|
|
# with this program; if not, write to the Free Software Foundation, Inc.,
|
|
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
|
|
"""Script for taking pre/post snapshots; run from pacman hooks."""
|
|
|
|
from argparse import ArgumentParser
|
|
from configparser import ConfigParser
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
|
|
|
|
logging.basicConfig(format="%(message)s", level=logging.INFO)
|
|
|
|
|
|
class SnapperCmd:
|
|
|
|
def __init__(self, config, snapshot_type, cleanup_algorithm, description="",
|
|
nodbus=False, pre_number=None, userdata=""):
|
|
self.cmd = ["snapper"]
|
|
if nodbus:
|
|
self.cmd.append("--no-dbus")
|
|
self.cmd.extend([
|
|
f"--config {config} create",
|
|
f"--cleanup-algorithm {cleanup_algorithm}",
|
|
"--print-number"
|
|
])
|
|
if description:
|
|
self.cmd.append(f"--description \"{description}\"")
|
|
if userdata:
|
|
self.cmd.append(f"--userdata \"{userdata}\"")
|
|
if snapshot_type == "post":
|
|
if pre_number is not None:
|
|
self.cmd.append(f"--pre-number {pre_number}")
|
|
else:
|
|
logging.debug("snapshot type specified as 'post' but no pre snapshot number, "
|
|
"so setting snapshot type to 'single'. If installing "
|
|
"snap-pac this is normal.")
|
|
snapshot_type = "single"
|
|
self.cmd.append(f"--type {snapshot_type}")
|
|
|
|
def __call__(self):
|
|
return os.popen(self.__str__()).read().rstrip("\n")
|
|
|
|
def __str__(self):
|
|
return " ".join(self.cmd)
|
|
|
|
|
|
class ConfigProcessor:
|
|
|
|
def __init__(self, ini_file, snapshot_type, parent_cmd=None, packages=None):
|
|
"""Set up defaults for snap-pac configuration."""
|
|
|
|
if parent_cmd is None:
|
|
self.parent_cmd = os.popen(f"ps -p {os.getppid()} -o args=").read().strip()
|
|
else:
|
|
self.parent_cmd = parent_cmd
|
|
|
|
if packages is None:
|
|
self.packages = [line.rstrip("\n") for line in sys.stdin]
|
|
else:
|
|
self.packages = packages
|
|
|
|
self.snapshot_type = snapshot_type
|
|
|
|
self.config = ConfigParser()
|
|
self.config["DEFAULT"] = {
|
|
"snapshot": False,
|
|
"cleanup_algorithm": "number",
|
|
"pre_description": self.parent_cmd,
|
|
"post_description": " ".join(self.packages),
|
|
"desc_limit": 72,
|
|
"important_packages": [],
|
|
"important_commands": [],
|
|
"userdata": []
|
|
}
|
|
self.config["root"] = {
|
|
"snapshot": True
|
|
}
|
|
self.config.read(ini_file)
|
|
|
|
def get_cleanup_algorithm(self, section):
|
|
return self.config.get(section, "cleanup_algorithm")
|
|
|
|
def get_description(self, section):
|
|
desc_limit = self.config.getint(section, "desc_limit")
|
|
return self.config.get(section, f"{self.snapshot_type}_description")[:desc_limit]
|
|
|
|
def check_important_commands(self, section):
|
|
return self.parent_cmd in json.loads(self.config.get(section, "important_commands"))
|
|
|
|
def check_important_packages(self, section):
|
|
important_packages = json.loads(self.config.get(section, "important_packages"))
|
|
return any(x in important_packages for x in self.packages)
|
|
|
|
def check_important(self, section):
|
|
return (self.check_important_commands(section) or
|
|
self.check_important_packages(section))
|
|
|
|
def get_userdata(self, section):
|
|
userdata = set(json.loads(self.config.get(section, "userdata")))
|
|
if self.check_important(section):
|
|
userdata.add("important=yes")
|
|
return ",".join(sorted(list(userdata)))
|
|
|
|
def __call__(self, section):
|
|
if section not in self.config:
|
|
self.config.add_section(section)
|
|
return {
|
|
"description": self.get_description(section),
|
|
"cleanup_algorithm": self.get_cleanup_algorithm(section),
|
|
"userdata": self.get_userdata(section),
|
|
"snapshot": self.config.getboolean(section, "snapshot")
|
|
}
|
|
|
|
|
|
def get_snapper_configs(conf_file):
|
|
"""Get the snapper configurations."""
|
|
for line in conf_file.read_text().split("\n"):
|
|
if line.startswith("SNAPPER_CONFIGS"):
|
|
line = line.rstrip("\n").rstrip("\"").split("=")
|
|
return line[1].lstrip("\"").split()
|
|
|
|
|
|
class Prefile:
|
|
"""Handles reading and writing of pre snapshot number."""
|
|
def __init__(self, snapper_config, snapshot_type):
|
|
self.file = Path(tempfile.gettempdir()) / f"snap-pac-pre_{snapper_config}"
|
|
self.snapshot_type = snapshot_type
|
|
|
|
def read(self):
|
|
if self.snapshot_type == "pre":
|
|
pre_number = None
|
|
else:
|
|
try:
|
|
pre_number = self.file.read_text()
|
|
except FileNotFoundError:
|
|
pre_number = None
|
|
logging.debug(f"prefile {self.file} not found. Ensure you have run the pre snapshot first. "
|
|
"If installing snap-pac this is normal.")
|
|
else:
|
|
self.file.unlink()
|
|
return pre_number
|
|
|
|
def write(self, num):
|
|
if self.snapshot_type == "pre":
|
|
self.file.write_text(num)
|
|
|
|
|
|
def check_skip():
|
|
return os.getenv("SNAP_PAC_SKIP", "n").lower() in ["y", "yes", "true", "1"]
|
|
|
|
|
|
def parse_args():
|
|
parser = ArgumentParser(description="Script for taking pre/post snapper snapshots. Used with pacman hooks.")
|
|
parser.add_argument(dest="type", choices=["pre", "post"], help="snapper snapshot type")
|
|
parser.add_argument(
|
|
"--ini", dest="snap_pac_ini", type=Path,
|
|
default=Path("/etc/snap-pac.ini"), help="snap-pac ini file path"
|
|
)
|
|
parser.add_argument(
|
|
"--conf", dest="snapper_conf_file", type=Path,
|
|
default=Path("/etc/conf.d/snapper"), help="snapper configuration file path"
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
if check_skip():
|
|
logging.warning("snapper snapshots skipped")
|
|
quit()
|
|
|
|
args = parse_args()
|
|
config_processor = ConfigProcessor(args.snap_pac_ini, args.type)
|
|
chroot = os.stat("/") != os.stat("/proc/1/root/.")
|
|
|
|
for snapper_config in get_snapper_configs(args.snapper_conf_file):
|
|
|
|
data = config_processor(snapper_config)
|
|
if data["snapshot"]:
|
|
prefile = Prefile(snapper_config, args.type)
|
|
pre_number = prefile.read()
|
|
num = SnapperCmd(snapper_config, args.type, data["cleanup_algorithm"],
|
|
data["description"], chroot, pre_number, data["userdata"])()
|
|
logging.info(f"==> {snapper_config}: {num}")
|
|
prefile.write(num)
|