Files
archy/image-recipe/host-policy/provision.py
T

162 lines
8.9 KiB
Python

#!/usr/bin/env python3
"""Provision only an offline Debian image root; never invoke APT/nft/systemctl."""
import argparse
import json
import os
from pathlib import Path
import stat
import sys
sys.path.insert(0, str(Path(__file__).resolve().parent))
import firewall
MASKED = ('apt-daily.timer', 'apt-daily-upgrade.timer', 'apt-daily.service',
'apt-daily-upgrade.service', 'unattended-upgrades.service',
'archipelago-update.timer', 'archipelago-update.service', 'nftables.service')
DEPENDENTS = ('NetworkManager.service', 'systemd-networkd.service', 'networking.service',
'archipelago.service', 'nginx.service', 'ssh.service', 'sshd.service',
'archipelago-fips.service', 'archipelago-wg.service', 'tailscaled.service',
'ssh.socket', 'sshd.socket', 'podman.socket', 'docker.socket',
'nostr-relay.service', 'archipelago-first-boot-containers.service')
BUILD_MARKER = b'# Archipelago image build sources\n'
def checked_path(root, relative):
path = root / relative
current = root
for part in path.relative_to(root).parts[:-1]:
current /= part
if current.exists() or current.is_symlink():
mode = current.lstat().st_mode
if not stat.S_ISDIR(mode) or current.lstat().st_uid != os.geteuid() or mode & 0o022:
raise ValueError('foreign image path preserved: ' + str(current))
return path
def provision(root, payload):
root = root.absolute()
if root == Path('/') or root.resolve() != root or not root.is_dir():
raise ValueError('an existing offline image directory is required; / is forbidden')
info = root.stat()
if info.st_uid != os.geteuid() or info.st_mode & 0o022:
raise ValueError('unprotected offline image directory')
release = (root / 'etc/os-release').resolve()
if not release.is_relative_to(root):
raise ValueError('OS release escaped the image')
values = dict(line.split('=', 1) for line in release.read_text().splitlines() if '=' in line)
suite = values.get('VERSION_CODENAME', '').strip('"')
if values.get('ID', '').strip('"') != 'debian' or suite not in ('bookworm', 'trixie'):
raise ValueError('only reviewed Debian suites are supported')
keyring = root / 'usr/share/keyrings/debian-archive-keyring.gpg'
if not keyring.resolve().is_relative_to(root) or not keyring.is_file() or not keyring.stat().st_size:
raise ValueError('Debian archive keyring required')
ports = json.loads((payload / 'app-ports.json').read_text())
files = firewall.render(ports)
files['usr/libexec/archipelago-host-firewall'] = (payload / 'firewall.py').read_bytes()
files['etc/systemd/system/archipelago-host-firewall.service'] = (payload / 'archipelago-host-firewall.service').read_bytes()
files['usr/libexec/archipelago-app-profiles'] = (payload / 'load-app-profiles.py').read_bytes()
files['etc/systemd/system/archipelago-app-profiles.service'] = (payload / 'archipelago-app-profiles.service').read_bytes()
for profile in sorted((payload / 'apparmor').iterdir()):
if profile.is_symlink() or not profile.is_file():
raise ValueError('invalid application profile asset')
files['usr/share/archipelago/apparmor/' + profile.name] = profile.read_bytes()
for unit in ('archipelago.service', 'podman.service', 'docker.service', 'archipelago-first-boot-containers.service'):
files['etc/systemd/system/' + unit + '.d/21-app-profiles.conf'] = b'[Unit]\nRequires=archipelago-app-profiles.service\nAfter=archipelago-app-profiles.service\n'
dependency = b'[Unit]\nRequires=archipelago-host-firewall.service\nAfter=archipelago-host-firewall.service\n'
for unit in DEPENDENTS:
files['etc/systemd/system/' + unit + '.d/20-host-firewall.conf'] = dependency
files['etc/apt/apt.conf.d/99archipelago-d09'] = b'''// Archipelago: package changes require reviewed signed D09 authority.
APT::Periodic::Enable "0";
APT::Periodic::Update-Package-Lists "0";
APT::Periodic::Download-Upgradeable-Packages "0";
APT::Periodic::Unattended-Upgrade "0";
Unattended-Upgrade::Automatic-Reboot "false";
'''
files['etc/apt/sources.list.d/archipelago-runtime.sources'] = f'''# Archipelago runtime sources; no unattended package mutation.
Types: deb
URIs: https://deb.debian.org/debian
Suites: {suite} {suite}-updates
Components: main contrib non-free non-free-firmware
Signed-By: /usr/share/keyrings/debian-archive-keyring.gpg
Types: deb
URIs: https://deb.debian.org/debian-security
Suites: {suite}-security
Components: main contrib non-free non-free-firmware
Signed-By: /usr/share/keyrings/debian-archive-keyring.gpg
'''.encode()
files['etc/archipelago/os-update-policy.json'] = (json.dumps({
'version': 1, 'package_writer': 'reviewed-signed-D09',
'unattended_package_changes': False, 'automatic_reboot': False,
'review_cadence': 'not-appointed', 'suite': suite,
}, indent=2) + '\n').encode()
links = {'etc/systemd/system/' + name: '/dev/null' for name in MASKED}
links['etc/systemd/system/multi-user.target.wants/archipelago-host-firewall.service'] = '../archipelago-host-firewall.service'
links['etc/systemd/system/multi-user.target.wants/archipelago-app-profiles.service'] = '../archipelago-app-profiles.service'
# Validate every destination before writing anything; no foreign overwrite.
for name, contents in files.items():
path = checked_path(root, name)
if path.exists() or path.is_symlink():
metadata = path.lstat()
if (not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1
or metadata.st_uid != os.geteuid() or metadata.st_mode & 0o022
or path.read_bytes() != contents):
raise ValueError('different existing image asset preserved: ' + name)
for name, target in links.items():
path = checked_path(root, name)
if (path.exists() or path.is_symlink()) and (not path.is_symlink() or os.readlink(path) != target):
raise ValueError('foreign systemd policy preserved: ' + name)
build = checked_path(root, 'etc/apt/sources.list.d/archipelago-build.sources')
if build.exists() or build.is_symlink():
metadata = build.lstat()
if (not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1
or metadata.st_uid != os.geteuid() or metadata.st_mode & 0o022
or not build.read_bytes().startswith(BUILD_MARKER)):
raise ValueError('foreign build source file preserved')
index_hook = checked_path(root, 'etc/apt/apt.conf.d/98archipelago-build-indexes')
if index_hook.exists() or index_hook.is_symlink():
metadata = index_hook.lstat()
if (not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1
or metadata.st_uid != os.geteuid() or metadata.st_mode & 0o022
or not index_hook.read_bytes().startswith(b'// Archipelago image build index lock\n')):
raise ValueError('foreign build index hook preserved')
# A stale snapshot/default source must not survive silently alongside the
# runtime pockets. Foreign source files require an explicit image decision.
source_dir = root / 'etc/apt/sources.list.d'
candidates = [root / 'etc/apt/sources.list']
if source_dir.exists():
candidates += [p for p in source_dir.iterdir() if p.suffix in ('.list', '.sources')]
for path in candidates:
if path in (build, root / 'etc/apt/sources.list.d/archipelago-runtime.sources'):
continue
if path.is_symlink() or (path.exists() and any(line.strip() and not line.lstrip().startswith('#') for line in path.read_text().splitlines())):
raise ValueError('foreign APT sources preserved; image source reconciliation required')
for name, contents in files.items():
path = root / name
path.parent.mkdir(parents=True, exist_ok=True, mode=0o755)
if not path.exists():
with path.open('xb') as target:
target.write(contents)
path.chmod(0o755 if name.startswith('usr/libexec/') else 0o644)
for name, target in links.items():
path = root / name
path.parent.mkdir(parents=True, exist_ok=True, mode=0o755)
if not path.is_symlink():
path.symlink_to(target)
if build.exists():
build.unlink()
if index_hook.exists():
index_hook.unlink()
return {'state': 'IMAGE_POLICY_PROVISIONED_NOT_RUNTIME_QUALIFIED', 'suite': suite,
'firewall_tables': list(firewall.TABLES), 'masked_units': list(MASKED),
'review_cadence': 'not-appointed'}
if __name__ == '__main__':
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--root', type=Path, required=True)
args = parser.parse_args()
if os.geteuid() != 0:
parser.error('image provisioning requires the image build/deployment principal')
print(json.dumps(provision(args.root, Path(__file__).resolve().parent)))