mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-01 10:22:08 +00:00
173 lines
7.0 KiB
Python
173 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
|
|
|
# ==========================================================
|
|
# ProxMenux - A menu-driven script for Proxmox VE management
|
|
# ==========================================================
|
|
# Author : jcastro
|
|
# GitHub : github.com/jcastro
|
|
# Maintainer : MacRimi
|
|
# Copyright : (c) 2026 MacRimi & contributors
|
|
# License : (GPL-3.0) (https://github.com/MacRimi/ProxMenux/blob/main/LICENSE)
|
|
# Version : 1.0
|
|
# Last Updated: 22/10/2026
|
|
# ==========================================================
|
|
# Description:
|
|
# Preserve original OCI image files outside managed-volume mountpoints
|
|
# before attaching persistent disks, so ZFS snapshot backups can mount them.
|
|
# Only operate on a stopped container's freshly extracted raw rootfs.
|
|
# ==========================================================
|
|
|
|
"""Keep raw OCI image defaults out of future managed-volume mountpoints.
|
|
|
|
Call only on a stopped, freshly extracted rootfs, after capturing volume seeds
|
|
and before attaching persistent disks. Reused disks are never opened here.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import re
|
|
import stat
|
|
|
|
ARCHIVE = '.proxmenux-image-defaults'
|
|
MARKER = '.proxmenux-image-defaults-v1'
|
|
FLAGS = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
|
|
|
|
|
|
def path_parts(value):
|
|
if (not isinstance(value, str) or not value.startswith('/') or value == '/'
|
|
or any(c in value for c in '\0\n\r')
|
|
or any(p in ('.', '..', '') for p in value.split('/')[1:])):
|
|
raise ValueError(f'Unsafe image mountpoint: {value!r}')
|
|
parts = PurePosixPath(value).parts[1:]
|
|
if parts[0] == ARCHIVE:
|
|
raise ValueError('The image-default archive is a reserved path')
|
|
return parts
|
|
|
|
|
|
def mounted_paths():
|
|
def unescape(text):
|
|
return re.sub(r'\\([0-7]{3})', lambda m: chr(int(m[1], 8)), text)
|
|
return [Path(unescape(line.split()[4]))
|
|
for line in Path('/proc/self/mountinfo').read_text().splitlines()]
|
|
|
|
|
|
def open_directory(root_fd, parts):
|
|
current = os.dup(root_fd)
|
|
try:
|
|
for component in parts:
|
|
child = os.open(component, FLAGS, dir_fd=current)
|
|
os.close(current)
|
|
current = child
|
|
return current
|
|
except BaseException:
|
|
os.close(current)
|
|
raise
|
|
|
|
|
|
def normalize(root, paths, mounts=None):
|
|
root = Path(root)
|
|
if root.is_symlink() or not root.is_absolute() or root.resolve() != root:
|
|
raise ValueError('The raw rootfs must be a real absolute directory')
|
|
paths = sorted(set(paths), key=lambda p: (-len(path_parts(p)), p))
|
|
mounts = mounted_paths() if mounts is None else list(map(Path, mounts))
|
|
root_fd = os.open(root, FLAGS)
|
|
targets, result = [], []
|
|
try:
|
|
root_stat = os.fstat(root_fd)
|
|
# Validate every path before moving any entry. A mount on an ancestor
|
|
# or descendant means this is not exclusively the raw image anymore.
|
|
for value in paths:
|
|
parts = path_parts(value)
|
|
target = root.joinpath(*parts)
|
|
for mount in mounts:
|
|
if mount != root and mount.is_relative_to(root) and (mount == target or target.is_relative_to(mount)
|
|
or mount.is_relative_to(target)):
|
|
raise ValueError(f'An attached filesystem intersects {value}')
|
|
try:
|
|
fd = open_directory(root_fd, parts)
|
|
except FileNotFoundError:
|
|
continue
|
|
except OSError as error:
|
|
raise ValueError(f'Image mountpoint is not a real directory: {value}') from error
|
|
if os.fstat(fd).st_dev != root_stat.st_dev:
|
|
os.close(fd)
|
|
raise ValueError(f'Image mountpoint crosses a filesystem: {value}')
|
|
targets.append((value, fd))
|
|
nonempty = [(p, fd) for p, fd in targets if os.listdir(fd)]
|
|
if not nonempty:
|
|
return []
|
|
existing_archive = False
|
|
try:
|
|
os.stat(ARCHIVE, dir_fd=root_fd, follow_symlinks=False)
|
|
except FileNotFoundError:
|
|
pass
|
|
else:
|
|
existing_archive = True
|
|
if not existing_archive:
|
|
os.mkdir(ARCHIVE, 0o755, dir_fd=root_fd)
|
|
archive_fd = os.open(ARCHIVE, FLAGS, dir_fd=root_fd)
|
|
try:
|
|
if existing_archive:
|
|
marker = os.open(MARKER, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=archive_fd)
|
|
try:
|
|
if (not stat.S_ISREG(os.fstat(marker).st_mode)
|
|
or os.read(marker, 32) != b'proxmenux-image-defaults-v1\n'):
|
|
raise ValueError('The image-default archive already exists')
|
|
finally:
|
|
os.close(marker)
|
|
else:
|
|
os.fchown(archive_fd, root_stat.st_uid, root_stat.st_gid)
|
|
marker = os.open(MARKER, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644, dir_fd=archive_fd)
|
|
try:
|
|
os.fchown(marker, root_stat.st_uid, root_stat.st_gid)
|
|
os.write(marker, b'proxmenux-image-defaults-v1\n')
|
|
finally:
|
|
os.close(marker)
|
|
# Validate all destinations before moving entries; never overwrite
|
|
# a prior saved directory (including one shipped by the image).
|
|
for value, _ in nonempty:
|
|
name = hashlib.sha256(value.encode()).hexdigest()
|
|
try:
|
|
os.stat(name, dir_fd=archive_fd, follow_symlinks=False)
|
|
except FileNotFoundError:
|
|
continue
|
|
raise ValueError(f'Image-default destination already exists: {value}')
|
|
for value, fd in nonempty:
|
|
name = hashlib.sha256(value.encode()).hexdigest()
|
|
os.mkdir(name, 0o700, dir_fd=archive_fd)
|
|
saved_fd = os.open(name, FLAGS, dir_fd=archive_fd)
|
|
try:
|
|
owner = os.fstat(fd)
|
|
os.fchown(saved_fd, owner.st_uid, owner.st_gid)
|
|
os.fchmod(saved_fd, stat.S_IMODE(owner.st_mode))
|
|
for child in os.listdir(fd):
|
|
os.rename(child, child, src_dir_fd=fd, dst_dir_fd=saved_fd)
|
|
finally:
|
|
os.close(saved_fd)
|
|
result.append({'target': value, 'saved': f'/{ARCHIVE}/{name}'})
|
|
finally:
|
|
os.close(archive_fd)
|
|
return result
|
|
finally:
|
|
for _, fd in targets:
|
|
os.close(fd)
|
|
os.close(root_fd)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('rootfs')
|
|
parser.add_argument('deployment')
|
|
args = parser.parse_args()
|
|
deployment = json.loads(Path(args.deployment).read_text())
|
|
paths = [m['container_path'] for m in deployment.get('mounts', [])
|
|
if m.get('type') == 'managed-volume']
|
|
for item in normalize(Path(args.rootfs), paths):
|
|
print(f"Image defaults preserved: {item['target']} -> {item['saved']}")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|