Compare commits
6 Commits
a42dd4ba54
...
v2.2.0
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
69779919e5 | ||
|
|
05e9129f4e | ||
|
|
963f0a8da6 | ||
|
|
853a503a3f | ||
|
|
b4f6ec4424 | ||
|
|
b88878efec |
17
.github/workflows/test.yml
vendored
17
.github/workflows/test.yml
vendored
@@ -18,5 +18,22 @@ jobs:
|
|||||||
- uses: actions/setup-python@v5
|
- uses: actions/setup-python@v5
|
||||||
with:
|
with:
|
||||||
python-version: "3.12"
|
python-version: "3.12"
|
||||||
|
# tor lets the deterministic offline onion-keygen test run (it validates
|
||||||
|
# the real production keygen path); that test needs no network.
|
||||||
|
- run: sudo apt-get update && sudo apt-get install -y tor
|
||||||
- run: pip install pytest pyyaml
|
- run: pip install pytest pyyaml
|
||||||
- run: make test
|
- run: make test
|
||||||
|
|
||||||
|
tor-network-e2e:
|
||||||
|
# The live onion round-trip depends on the public Tor network, so it is
|
||||||
|
# informational (continue-on-error) rather than a blocking merge gate.
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
continue-on-error: true
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@v4
|
||||||
|
- uses: actions/setup-python@v5
|
||||||
|
with:
|
||||||
|
python-version: "3.12"
|
||||||
|
- run: sudo apt-get update && sudo apt-get install -y tor
|
||||||
|
- run: pip install pytest pyyaml
|
||||||
|
- run: LIM_E2E_TOR=1 make test-tor PYTHON=python
|
||||||
|
|||||||
14
CHANGELOG.md
14
CHANGELOG.md
@@ -1,5 +1,19 @@
|
|||||||
# Changelog
|
# Changelog
|
||||||
|
|
||||||
|
## [2.2.0] - 2026-07-21
|
||||||
|
|
||||||
|
- Optional remote LUKS unlock via a Tor onion service in the initramfs.
|
||||||
|
Encrypted image setups can bake *tor* into the initramfs (mkinitcpio hooks
|
||||||
|
ordered *netconf tor dropbear encryptssh*), generate the v3 onion keys
|
||||||
|
offline in the image chroot, and print the stable *.onion* address. Unlock
|
||||||
|
with *torsocks ssh root@<onion-address>*. Useful behind NAT/CGNAT or a
|
||||||
|
dynamic IP, where the direct *ssh root@<ip>* unlock cannot reach the host.
|
||||||
|
- End-to-end tests: a rootless one (*LIM_E2E_TOR=1*) that round-trips a
|
||||||
|
passphrase through a real Tor onion service, and a full virtualized one
|
||||||
|
(*LIM_E2E_QEMU=1*) that builds a LUKS image, boots it in QEMU, and unlocks
|
||||||
|
the root over Tor. The offline onion keygen runs in CI; *make test-tor* /
|
||||||
|
*make test-qemu* / *make test-all* run the opt-in stages.
|
||||||
|
|
||||||
## [2.1.0] - 2026-07-14
|
## [2.1.0] - 2026-07-14
|
||||||
|
|
||||||
Initial PyPI release 🥳
|
Initial PyPI release 🥳
|
||||||
|
|||||||
22
Makefile
22
Makefile
@@ -1,6 +1,9 @@
|
|||||||
.PHONY: install test
|
.PHONY: install test test-tor test-qemu test-all
|
||||||
|
|
||||||
PREFIX ?= $(HOME)/.local
|
PREFIX ?= $(HOME)/.local
|
||||||
|
# Interpreter used for the tests. Override if your active venv lacks pytest,
|
||||||
|
# e.g. make test PYTHON=/usr/bin/python3
|
||||||
|
PYTHON ?= python3
|
||||||
|
|
||||||
install:
|
install:
|
||||||
chmod +x main.py
|
chmod +x main.py
|
||||||
@@ -8,5 +11,20 @@ install:
|
|||||||
ln -sf $(CURDIR)/main.py $(PREFIX)/bin/lim
|
ln -sf $(CURDIR)/main.py $(PREFIX)/bin/lim
|
||||||
@echo "Installed lim to $(PREFIX)/bin/lim"
|
@echo "Installed lim to $(PREFIX)/bin/lim"
|
||||||
|
|
||||||
|
# Default suite: fully mocked, no root/QEMU/network. The opt-in e2e tests skip.
|
||||||
test:
|
test:
|
||||||
python3 -m pytest
|
$(PYTHON) -m pytest
|
||||||
|
|
||||||
|
# Rootless Tor onion unlock e2e (needs the `tor` binary and network access).
|
||||||
|
test-tor:
|
||||||
|
LIM_E2E_TOR=1 $(PYTHON) -m pytest tests/e2e/test_tor_unlock_e2e.py -v
|
||||||
|
|
||||||
|
# Full virtualized build+boot+unlock e2e. Needs qemu, arch-install-scripts,
|
||||||
|
# cryptsetup, tor and ncat; the build stage runs under sudo. Set CHUTNEY_PATH
|
||||||
|
# to use a private, offline Tor network instead of the public one.
|
||||||
|
test-qemu:
|
||||||
|
LIM_E2E_QEMU=1 $(PYTHON) -m pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
|
||||||
|
|
||||||
|
# Everything: mocked suite plus both opt-in e2e stages, in one run.
|
||||||
|
test-all:
|
||||||
|
LIM_E2E_TOR=1 LIM_E2E_QEMU=1 $(PYTHON) -m pytest -v -s
|
||||||
|
|||||||
75
README.md
75
README.md
@@ -10,6 +10,7 @@ Linux Image Manager (lim) is a Python tool for downloading, configuring, and man
|
|||||||
|
|
||||||
- **Image Download & Setup:** Automatically download, verify (checksum + GPG signature) and prepare Linux distributions.
|
- **Image Download & Setup:** Automatically download, verify (checksum + GPG signature) and prepare Linux distributions.
|
||||||
- **Encrypted Storage:** Configure LUKS encryption for secure image management.
|
- **Encrypted Storage:** Configure LUKS encryption for secure image management.
|
||||||
|
- **Tor Onion Unlock:** Optionally bake a Tor onion service into the initramfs so the dropbear LUKS unlock shell stays reachable behind NAT or dynamic IPs.
|
||||||
- **Virtual RAID1:** Easily set up virtual Btrfs RAID1 for data redundancy.
|
- **Virtual RAID1:** Easily set up virtual Btrfs RAID1 for data redundancy.
|
||||||
- **Backup & Restore:** Create image backups from devices using dd.
|
- **Backup & Restore:** Create image backups from devices using dd.
|
||||||
- **Chroot Environment:** Easily enter a chroot shell to maintain or modify Linux images.
|
- **Chroot Environment:** Easily enter a chroot shell to maintain or modify Linux images.
|
||||||
@@ -26,7 +27,7 @@ package-manager install lim
|
|||||||
|
|
||||||
This command makes Linux Image Manager globally available as `lim` in your terminal. The `lim` alias points to the **main.py** entry point.
|
This command makes Linux Image Manager globally available as `lim` in your terminal. The `lim` alias points to the **main.py** entry point.
|
||||||
|
|
||||||
There are no Python dependencies beyond the standard library. The commands call the usual system tools (`cryptsetup`, `fdisk`, `dd`, `rsync`, `wget`, `gpg`, `encfs`, `pv`, `bsdtar`, ...), so those need to be installed for the command you use.
|
The only Python dependency is **PyYAML** (`>=6`), used to read the image catalog. The pip/wheel install pulls it in automatically; on the `package-manager`/symlink install path make sure it is present (`python-yaml` on Arch). The commands also call the usual system tools (`cryptsetup`, `fdisk`, `dd`, `rsync`, `wget`, `gpg`, `encfs`, `pv`, `bsdtar`, ...), so those need to be installed for the command you use.
|
||||||
|
|
||||||
## Usage ⚙️
|
## Usage ⚙️
|
||||||
|
|
||||||
@@ -94,6 +95,8 @@ lim/
|
|||||||
distributions.yml # single point of truth for the image catalog
|
distributions.yml # single point of truth for the image catalog
|
||||||
configuration/ # package collections used during image setup
|
configuration/ # package collections used during image setup
|
||||||
tests/unit/ # unit tests (all external commands mocked)
|
tests/unit/ # unit tests (all external commands mocked)
|
||||||
|
tests/e2e/ # Tor onion unlock end-to-end tests (rootless + QEMU, opt-in)
|
||||||
|
tests/e2e/qemu/ # full virtualized build+boot+unlock harness
|
||||||
tests/lint/ # architecture guards (e.g. max file length)
|
tests/lint/ # architecture guards (e.g. max file length)
|
||||||
```
|
```
|
||||||
|
|
||||||
@@ -102,15 +105,83 @@ tests/lint/ # architecture guards (e.g. max file length)
|
|||||||
Customize your environment in the `lim/configuration/` folder:
|
Customize your environment in the `lim/configuration/` folder:
|
||||||
- **General Packages:** Contains common packages for all setup scripts.
|
- **General Packages:** Contains common packages for all setup scripts.
|
||||||
- **Server LUKS Packages:** Contains packages needed for setting up LUKS encryption on servers.
|
- **Server LUKS Packages:** Contains packages needed for setting up LUKS encryption on servers.
|
||||||
|
- **initcpio Hooks:** The mkinitcpio install/runtime hooks and torrc baked into images for the Tor onion unlock.
|
||||||
|
|
||||||
|
## Remote LUKS Unlock via Tor 🧅
|
||||||
|
|
||||||
|
When you answer yes to *"Should the system be remotely unlockable via a Tor
|
||||||
|
onion service?"* during an encrypted `lim --type image` setup, the image gets:
|
||||||
|
|
||||||
|
- `tor` and `busybox` installed, plus a custom `tor` mkinitcpio hook ordered
|
||||||
|
between `netconf` and `dropbear`.
|
||||||
|
- A v3 onion service key generated **offline** inside the image chroot; the
|
||||||
|
resulting `.onion` address is printed at the end of the setup and stays
|
||||||
|
stable across reboots and IP changes.
|
||||||
|
- An NTP clock sync in early userspace (boards like the Raspberry Pi have no
|
||||||
|
RTC and would otherwise boot with a clock Tor rejects). Override the server
|
||||||
|
with the kernel parameter `tor_ntp=<host>`.
|
||||||
|
|
||||||
|
After booting the device, unlock it from any machine with a running Tor
|
||||||
|
client:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
torsocks ssh root@<onion-address>
|
||||||
|
```
|
||||||
|
|
||||||
|
Typing the LUKS passphrase into that shell resumes the boot. The direct
|
||||||
|
`ssh root@<ip>` unlock keeps working as before; Tor is an additional path,
|
||||||
|
useful behind NAT/CGNAT where no port forwarding is possible.
|
||||||
|
|
||||||
|
**Security note:** the onion private key is stored in the initramfs on the
|
||||||
|
unencrypted boot partition. Anyone with physical access to the SD card can
|
||||||
|
read it and impersonate the onion service — treat the address as
|
||||||
|
non-secret. Access control remains the SSH public key
|
||||||
|
(`/etc/dropbear/root_key`), exactly as with the direct unlock.
|
||||||
|
|
||||||
## Development & Tests 🧪
|
## Development & Tests 🧪
|
||||||
|
|
||||||
The test suite mocks all external commands, so it runs safely on any machine:
|
The unit suite mocks all external commands, so it runs safely on any machine:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
pytest
|
pytest
|
||||||
```
|
```
|
||||||
|
|
||||||
|
### Tor unlock end-to-end test
|
||||||
|
|
||||||
|
`tests/e2e/test_tor_unlock_e2e.py` reproduces the operator-visible half of the
|
||||||
|
Tor decryption process **rootless** — no block devices, no `cryptsetup`, no
|
||||||
|
`chroot`. It generates the v3 onion keys offline (exactly as `lim.image.tor`
|
||||||
|
does inside the image), starts a real Tor onion service from a torrc mirroring
|
||||||
|
the baked-in one, then connects to the `.onion` address through Tor and
|
||||||
|
delivers a passphrase to a dropbear stand-in, asserting it arrives and the
|
||||||
|
endpoint "unlocks".
|
||||||
|
|
||||||
|
It needs the real `tor` binary and live Tor network access (onion bootstrap is
|
||||||
|
slow), so it is opt-in and skipped otherwise:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
LIM_E2E_TOR=1 pytest tests/e2e/test_tor_unlock_e2e.py -v
|
||||||
|
```
|
||||||
|
|
||||||
|
The offline-only checks in that file (onion keygen, production-flag guard) run
|
||||||
|
without network. The physical flashing half (`dd`, `cryptsetup luksFormat`,
|
||||||
|
`mount`, `chroot`, `mkinitcpio`) requires root and a matching CPU/qemu setup and
|
||||||
|
is not covered by this rootless test.
|
||||||
|
|
||||||
|
### Full virtualized unlock (QEMU)
|
||||||
|
|
||||||
|
For the *whole* process — build an encrypted image, boot it, run the real
|
||||||
|
initramfs Tor+dropbear chain, and unlock it — `tests/e2e/qemu/` drives a QEMU
|
||||||
|
virtual machine end to end and asserts the real system booted only after the
|
||||||
|
passphrase was delivered over Tor. It is opt-in (`LIM_E2E_QEMU=1`) and needs
|
||||||
|
QEMU, `arch-install-scripts`, `cryptsetup`, `tor` and `ncat`; the build stage
|
||||||
|
needs root (ideally inside a throwaway VM to keep the host rootless). A private,
|
||||||
|
offline Tor network via [chutney](https://gitlab.torproject.org/tpo/core/chutney)
|
||||||
|
is supported. See [tests/e2e/qemu/README.md](tests/e2e/qemu/README.md).
|
||||||
|
|
||||||
|
The harness's pure logic and drift guards run in the normal suite
|
||||||
|
(`test_qemu_harness_unit.py`) — no QEMU, root, or network needed.
|
||||||
|
|
||||||
## License 📜
|
## License 📜
|
||||||
|
|
||||||
This project is licensed under the GNU General Public License Version 3. See the [LICENSE.txt](./LICENSE.txt) file for details.
|
This project is licensed under the GNU General Public License Version 3. See the [LICENSE.txt](./LICENSE.txt) file for details.
|
||||||
|
|||||||
47
lim/configuration/initcpio/tor_hook
Normal file
47
lim/configuration/initcpio/tor_hook
Normal file
@@ -0,0 +1,47 @@
|
|||||||
|
#!/usr/bin/ash
|
||||||
|
# mkinitcpio runtime hook: start Tor so the dropbear unlock shell is
|
||||||
|
# reachable as an onion service while encryptssh waits for the passphrase.
|
||||||
|
# Installed to /etc/initcpio/hooks/tor by linux-image-manager.
|
||||||
|
|
||||||
|
# netconf's ipconfig drops its DHCP lease data (incl. DNS) into
|
||||||
|
# /tmp/net-*.conf; busybox's resolver only reads /etc/resolv.conf.
|
||||||
|
_tor_write_resolv_conf() {
|
||||||
|
[ -s /etc/resolv.conf ] && return 0
|
||||||
|
local conf dns
|
||||||
|
for conf in /tmp/net-*.conf; do
|
||||||
|
[ -f "$conf" ] || continue
|
||||||
|
# Extract ONLY the DNS fields with sed; never source these files —
|
||||||
|
# they hold attacker-controllable DHCP option strings (hostname,
|
||||||
|
# domain, rootpath), and sourcing would run them as root pre-boot.
|
||||||
|
for dns in $(sed -n 's/^IPV4DNS[01]=//p' "$conf"); do
|
||||||
|
[ -n "$dns" ] && [ "$dns" != "0.0.0.0" ] \
|
||||||
|
&& echo "nameserver $dns" >> /etc/resolv.conf
|
||||||
|
done
|
||||||
|
done
|
||||||
|
[ -s /etc/resolv.conf ]
|
||||||
|
}
|
||||||
|
|
||||||
|
run_hook() {
|
||||||
|
# Tor rejects consensus documents when the clock is far off; boards
|
||||||
|
# without an RTC boot in 1970, so sync before starting Tor. Bounded:
|
||||||
|
# a failed sync must never block the boot.
|
||||||
|
msg "tor: syncing clock via NTP..."
|
||||||
|
# Best-effort DNS; a tor_ntp IP literal needs none, so never gate on it.
|
||||||
|
_tor_write_resolv_conf || msg "tor: no DNS from DHCP (fine if tor_ntp is an IP)"
|
||||||
|
/usr/local/bin/busybox timeout 30 \
|
||||||
|
/usr/local/bin/busybox ntpd -n -q -p "${tor_ntp:-pool.ntp.org}" \
|
||||||
|
|| msg "tor: NTP sync failed, keeping current clock"
|
||||||
|
|
||||||
|
msg "tor: starting onion service for remote unlock..."
|
||||||
|
mkdir -p /var/lib/tor
|
||||||
|
chmod 0700 /var/lib/tor /etc/tor/onion
|
||||||
|
chmod 0600 /etc/tor/onion/hs_ed25519_secret_key
|
||||||
|
tor -f /etc/tor/torrc --RunAsDaemon 1 --Log "notice file /tmp/tor.log" \
|
||||||
|
|| msg "tor: failed to start, unlock stays reachable via direct IP"
|
||||||
|
}
|
||||||
|
|
||||||
|
run_cleanuphook() {
|
||||||
|
# Nothing from the initramfs may keep running after the pivot.
|
||||||
|
/usr/local/bin/busybox killall tor 2>/dev/null
|
||||||
|
return 0
|
||||||
|
}
|
||||||
37
lim/configuration/initcpio/tor_install
Normal file
37
lim/configuration/initcpio/tor_install
Normal file
@@ -0,0 +1,37 @@
|
|||||||
|
#!/bin/bash
|
||||||
|
# mkinitcpio install hook: Tor onion service for remote LUKS unlock.
|
||||||
|
# Installed to /etc/initcpio/install/tor by linux-image-manager.
|
||||||
|
|
||||||
|
build() {
|
||||||
|
add_binary /usr/bin/tor
|
||||||
|
# Full busybox for the ntpd applet: boards without an RTC (e.g. most
|
||||||
|
# Raspberry Pis) wake up in 1970, which Tor's consensus checks reject.
|
||||||
|
# A different target path keeps mkinitcpio's own busybox untouched.
|
||||||
|
add_binary /usr/bin/busybox /usr/local/bin/busybox
|
||||||
|
# glibc resolves the NTP server hostname via getaddrinfo(), which dlopen()s
|
||||||
|
# these NSS modules at runtime — add_binary only follows NEEDED libs, so
|
||||||
|
# without them DNS silently fails, ntpd never syncs, and the clock stays
|
||||||
|
# at 1970. glibc's built-in default (no nsswitch.conf) is "dns ... files".
|
||||||
|
add_binary /usr/lib/libnss_dns.so.2
|
||||||
|
add_binary /usr/lib/libnss_files.so.2
|
||||||
|
add_file /etc/tor/initramfs-torrc /etc/tor/torrc
|
||||||
|
add_file /etc/tor/initramfs-onion/hostname /etc/tor/onion/hostname
|
||||||
|
add_file /etc/tor/initramfs-onion/hs_ed25519_public_key /etc/tor/onion/hs_ed25519_public_key
|
||||||
|
add_file /etc/tor/initramfs-onion/hs_ed25519_secret_key /etc/tor/onion/hs_ed25519_secret_key
|
||||||
|
add_runscript
|
||||||
|
}
|
||||||
|
|
||||||
|
help() {
|
||||||
|
cat <<HELPEOF
|
||||||
|
Starts a Tor onion service in early userspace so the dropbear unlock
|
||||||
|
shell stays reachable under the .onion address baked into the image,
|
||||||
|
even behind NAT or a dynamic IP.
|
||||||
|
|
||||||
|
Place it between netconf and dropbear:
|
||||||
|
HOOKS=(... netconf tor dropbear encryptssh ...)
|
||||||
|
|
||||||
|
Optional kernel parameter:
|
||||||
|
tor_ntp=<server> NTP server used to set the clock before Tor starts
|
||||||
|
(default: pool.ntp.org)
|
||||||
|
HELPEOF
|
||||||
|
}
|
||||||
7
lim/configuration/initcpio/torrc
Normal file
7
lim/configuration/initcpio/torrc
Normal file
@@ -0,0 +1,7 @@
|
|||||||
|
# Tor configuration for the initramfs onion unlock service.
|
||||||
|
# Baked into the initramfs as /etc/tor/torrc by the "tor" mkinitcpio hook;
|
||||||
|
# the onion keys come from /etc/tor/initramfs-onion on the system.
|
||||||
|
DataDirectory /var/lib/tor
|
||||||
|
HiddenServiceDir /etc/tor/onion
|
||||||
|
HiddenServicePort 22 127.0.0.1:22
|
||||||
|
SocksPort 0
|
||||||
3
lim/configuration/packages/server/tor.txt
Normal file
3
lim/configuration/packages/server/tor.txt
Normal file
@@ -0,0 +1,3 @@
|
|||||||
|
# Packages for the Tor onion unlock service in the initramfs
|
||||||
|
tor
|
||||||
|
busybox # full applet set: ntpd for clock sync on RTC-less boards
|
||||||
@@ -9,6 +9,7 @@ from pathlib import Path
|
|||||||
from lim import catalog, fsutil, packages, runner, ui
|
from lim import catalog, fsutil, packages, runner, ui
|
||||||
from lim.image.plan import ImagePlan
|
from lim.image.plan import ImagePlan
|
||||||
from lim.image.session import ImageSession, chroot_bash, install_packages
|
from lim.image.session import ImageSession, chroot_bash, install_packages
|
||||||
|
from lim.image.tor import configure_tor_unlock
|
||||||
|
|
||||||
MKINITCPIO_HOOKS_PREFIX = (
|
MKINITCPIO_HOOKS_PREFIX = (
|
||||||
"base udev autodetect microcode modconf kms keyboard keymap consolefont block"
|
"base udev autodetect microcode modconf kms keyboard keymap consolefont block"
|
||||||
@@ -32,9 +33,10 @@ def _configure_mkinitcpio(plan: ImagePlan, root: Path) -> None:
|
|||||||
fsutil.replace_in_file(
|
fsutil.replace_in_file(
|
||||||
"BINARIES=()", "BINARIES=(/usr/lib/libgcc_s.so.1)", mkinitcpio_path
|
"BINARIES=()", "BINARIES=(/usr/lib/libgcc_s.so.1)", mkinitcpio_path
|
||||||
)
|
)
|
||||||
|
tor_hook = "tor " if plan.tor_unlock else ""
|
||||||
fsutil.replace_in_file(
|
fsutil.replace_in_file(
|
||||||
f"HOOKS=({MKINITCPIO_HOOKS_PREFIX} {MKINITCPIO_HOOKS_SUFFIX})",
|
f"HOOKS=({MKINITCPIO_HOOKS_PREFIX} {MKINITCPIO_HOOKS_SUFFIX})",
|
||||||
f"HOOKS=({MKINITCPIO_HOOKS_PREFIX} sleep netconf dropbear encryptssh "
|
f"HOOKS=({MKINITCPIO_HOOKS_PREFIX} sleep netconf {tor_hook}dropbear encryptssh "
|
||||||
f"{MKINITCPIO_HOOKS_SUFFIX})",
|
f"{MKINITCPIO_HOOKS_SUFFIX})",
|
||||||
mkinitcpio_path,
|
mkinitcpio_path,
|
||||||
)
|
)
|
||||||
@@ -90,9 +92,15 @@ def _configure_bootloader(plan: ImagePlan, session: ImageSession, root: Path) ->
|
|||||||
else:
|
else:
|
||||||
cmdline_txt_path = session.boot_mount_path / "cmdline.txt"
|
cmdline_txt_path = session.boot_mount_path / "cmdline.txt"
|
||||||
ui.info(f"Configuring {cmdline_txt_path}...")
|
ui.info(f"Configuring {cmdline_txt_path}...")
|
||||||
|
# Firmware-boot boards (e.g. RPi4 cmdline.txt) need the same early
|
||||||
|
# networking as the boot.txt branch, or the initramfs netconf hook
|
||||||
|
# brings up no interface and remote unlock (dropbear + tor) is
|
||||||
|
# unreachable — the onion never even publishes.
|
||||||
|
hostname = (root / "etc/hostname").read_text().strip()
|
||||||
fsutil.replace_in_file(
|
fsutil.replace_in_file(
|
||||||
"root=/dev/mmcblk0p2",
|
"root=/dev/mmcblk0p2",
|
||||||
f"{cryptdevice} rootfstype={plan.root_filesystem}",
|
f"{cryptdevice} rootfstype={plan.root_filesystem} "
|
||||||
|
f"ip=::::{hostname}:eth0:dhcp net.ifnames=0 biosdevname=0",
|
||||||
cmdline_txt_path,
|
cmdline_txt_path,
|
||||||
)
|
)
|
||||||
ui.info(f"Content of {cmdline_txt_path}:{cmdline_txt_path.read_text()}")
|
ui.info(f"Content of {cmdline_txt_path}:{cmdline_txt_path.read_text()}")
|
||||||
@@ -112,6 +120,10 @@ def configure_encryption(
|
|||||||
ui.info(f"Adding {authorized_keys} to dropbear...")
|
ui.info(f"Adding {authorized_keys} to dropbear...")
|
||||||
runner.run(["cp", "-v", str(authorized_keys), str(dropbear_root_key_path)], sudo=True)
|
runner.run(["cp", "-v", str(authorized_keys), str(dropbear_root_key_path)], sudo=True)
|
||||||
|
|
||||||
|
if plan.tor_unlock:
|
||||||
|
# Hook files and onion keys must exist before mkinitcpio bakes the image.
|
||||||
|
configure_tor_unlock(plan, root)
|
||||||
|
|
||||||
_configure_mkinitcpio(plan, root)
|
_configure_mkinitcpio(plan, root)
|
||||||
_register_encrypted_root(plan, session, root)
|
_register_encrypted_root(plan, session, root)
|
||||||
_configure_bootloader(plan, session, root)
|
_configure_bootloader(plan, session, root)
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ class ImagePlan:
|
|||||||
luks_memory_cost: str | None = None
|
luks_memory_cost: str | None = None
|
||||||
raspberry_pi_version: str | None = None
|
raspberry_pi_version: str | None = None
|
||||||
encrypt_system: bool = False
|
encrypt_system: bool = False
|
||||||
|
tor_unlock: bool = False
|
||||||
root_filesystem: str = ""
|
root_filesystem: str = ""
|
||||||
image_folder: Path = field(default_factory=Path)
|
image_folder: Path = field(default_factory=Path)
|
||||||
|
|
||||||
|
|||||||
@@ -49,6 +49,10 @@ def _choose_and_verify_image(plan: ImagePlan) -> None:
|
|||||||
if plan.operation_system == "linux":
|
if plan.operation_system == "linux":
|
||||||
choosers.choose_linux_image(plan)
|
choosers.choose_linux_image(plan)
|
||||||
plan.encrypt_system = ui.confirm("Should the system be encrypted?")
|
plan.encrypt_system = ui.confirm("Should the system be encrypted?")
|
||||||
|
if plan.encrypt_system:
|
||||||
|
plan.tor_unlock = ui.confirm(
|
||||||
|
"Should the system be remotely unlockable via a Tor onion service?"
|
||||||
|
)
|
||||||
ui.info("Generating os-image...")
|
ui.info("Generating os-image...")
|
||||||
transfer.download_image(plan)
|
transfer.download_image(plan)
|
||||||
else:
|
else:
|
||||||
|
|||||||
79
lim/image/tor.py
Normal file
79
lim/image/tor.py
Normal file
@@ -0,0 +1,79 @@
|
|||||||
|
"""Tor onion service in the initramfs for remote LUKS unlock.
|
||||||
|
|
||||||
|
The onion keys are generated offline inside the image chroot
|
||||||
|
(``tor --DisableNetwork 1`` writes them without touching the network)
|
||||||
|
and baked into the initramfs by the "tor" mkinitcpio hook, so the
|
||||||
|
dropbear unlock shell stays reachable under a stable .onion address
|
||||||
|
even behind NAT or a dynamic IP.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from lim import config, packages, runner, ui
|
||||||
|
from lim.errors import LimError
|
||||||
|
from lim.image.plan import ImagePlan
|
||||||
|
from lim.image.session import chroot_bash, install_packages
|
||||||
|
|
||||||
|
# Paths inside the image (relative to the mounted root).
|
||||||
|
ONION_DIR = "etc/tor/initramfs-onion"
|
||||||
|
TORRC_PATH = "etc/tor/initramfs-torrc"
|
||||||
|
|
||||||
|
# An empty -f torrc keeps the image's /etc/tor/torrc (User tor, ...) out of
|
||||||
|
# the keygen run; with DisableNetwork the keys appear within a second, the
|
||||||
|
# loop only cushions slow qemu-emulated chroots.
|
||||||
|
_KEYGEN_SCRIPT = f"""
|
||||||
|
mkdir -p /{ONION_DIR}
|
||||||
|
chmod 0700 /{ONION_DIR}
|
||||||
|
: > /tmp/tor-keygen-torrc
|
||||||
|
tor -f /tmp/tor-keygen-torrc --DisableNetwork 1 \\
|
||||||
|
--DataDirectory /tmp/tor-keygen-data \\
|
||||||
|
--HiddenServiceDir /{ONION_DIR} \\
|
||||||
|
--HiddenServicePort "22 127.0.0.1:22" \\
|
||||||
|
--SocksPort 0 --RunAsDaemon 0 --Log "notice stderr" &
|
||||||
|
tor_pid=$!
|
||||||
|
for _ in $(seq 1 30); do
|
||||||
|
[ -s /{ONION_DIR}/hostname ] && break
|
||||||
|
sleep 1
|
||||||
|
done
|
||||||
|
kill "$tor_pid" 2>/dev/null || true
|
||||||
|
rm -rf /tmp/tor-keygen-data /tmp/tor-keygen-torrc
|
||||||
|
[ -s /{ONION_DIR}/hostname ]
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
def _install_initcpio_files(root: Path) -> None:
|
||||||
|
source_dir = config.CONFIGURATION_PATH / "initcpio"
|
||||||
|
for source, target in (
|
||||||
|
(source_dir / "tor_install", root / "etc/initcpio/install/tor"),
|
||||||
|
(source_dir / "tor_hook", root / "etc/initcpio/hooks/tor"),
|
||||||
|
(source_dir / "torrc", root / TORRC_PATH),
|
||||||
|
):
|
||||||
|
ui.info(f"Installing {target}...")
|
||||||
|
runner.run(
|
||||||
|
["install", "-D", "-m", "0644", str(source), str(target)], sudo=True
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _generate_onion_keys(root: Path) -> str:
|
||||||
|
hostname_path = root / ONION_DIR / "hostname"
|
||||||
|
if hostname_path.is_file():
|
||||||
|
ui.info("Onion keys already exist, keeping the existing address.")
|
||||||
|
else:
|
||||||
|
ui.info("Generating onion service keys (offline, inside the chroot)...")
|
||||||
|
chroot_bash(root, _KEYGEN_SCRIPT, error_msg="Onion key generation failed.")
|
||||||
|
if not hostname_path.is_file():
|
||||||
|
raise LimError(f"Onion key generation produced no {hostname_path}.")
|
||||||
|
return hostname_path.read_text().strip()
|
||||||
|
|
||||||
|
|
||||||
|
def configure_tor_unlock(plan: ImagePlan, root: Path) -> str:
|
||||||
|
"""Install everything the "tor" mkinitcpio hook bakes in; return the onion address."""
|
||||||
|
ui.info("Setting up remote unlock via Tor onion service...")
|
||||||
|
install_packages(
|
||||||
|
plan.distribution, root, " ".join(packages.get_packages("server/tor"))
|
||||||
|
)
|
||||||
|
_install_initcpio_files(root)
|
||||||
|
onion_address = _generate_onion_keys(root)
|
||||||
|
ui.success(f"Onion unlock address: {onion_address}")
|
||||||
|
ui.info(f"Unlock later with: torsocks ssh root@{onion_address}")
|
||||||
|
return onion_address
|
||||||
@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "linux-image-manager"
|
name = "linux-image-manager"
|
||||||
version = "2.1.0"
|
version = "2.2.0"
|
||||||
description = "CLI (`lim`) for downloading, verifying and flashing Linux images, LUKS/Btrfs encrypted storage (single drive and RAID1), image backups and chroot maintenance"
|
description = "CLI (`lim`) for downloading, verifying and flashing Linux images, LUKS/Btrfs encrypted storage (single drive and RAID1), image backups and chroot maintenance"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
requires-python = ">=3.10"
|
requires-python = ">=3.10"
|
||||||
@@ -28,7 +28,11 @@ lim = "lim.cli:main"
|
|||||||
include = ["lim*"]
|
include = ["lim*"]
|
||||||
|
|
||||||
[tool.setuptools.package-data]
|
[tool.setuptools.package-data]
|
||||||
lim = ["distributions.yml", "configuration/packages/**/*.txt"]
|
lim = [
|
||||||
|
"distributions.yml",
|
||||||
|
"configuration/packages/**/*.txt",
|
||||||
|
"configuration/initcpio/*",
|
||||||
|
]
|
||||||
|
|
||||||
[tool.pytest.ini_options]
|
[tool.pytest.ini_options]
|
||||||
testpaths = ["tests"]
|
testpaths = ["tests"]
|
||||||
|
|||||||
0
tests/e2e/__init__.py
Normal file
0
tests/e2e/__init__.py
Normal file
85
tests/e2e/qemu/README.md
Normal file
85
tests/e2e/qemu/README.md
Normal file
@@ -0,0 +1,85 @@
|
|||||||
|
# Virtualized end-to-end Tor unlock harness
|
||||||
|
|
||||||
|
This harness models the **entire** encrypted-image process in a virtual
|
||||||
|
machine: it builds a LUKS image carrying the real `lim` Tor-in-initramfs
|
||||||
|
artifacts, boots it in QEMU, lets the real `netconf → tor → dropbear →
|
||||||
|
encryptssh` chain run in early userspace, then unlocks the root over Tor and
|
||||||
|
proves the real system booted.
|
||||||
|
|
||||||
|
It complements the lightweight, always-rootless
|
||||||
|
[`test_tor_unlock_e2e.py`](../test_tor_unlock_e2e.py): that one round-trips a
|
||||||
|
passphrase through Tor against a dropbear *stand-in*; this one runs the real
|
||||||
|
dropbear, real cryptsetup, real mkinitcpio initramfs, and a real tor daemon in
|
||||||
|
a booted machine.
|
||||||
|
|
||||||
|
## Why a virtio machine and not an emulated Raspberry Pi
|
||||||
|
|
||||||
|
QEMU's `raspi3b`/`raspi4b` machines do not emulate the Pi's USB-gadget
|
||||||
|
Ethernet — exactly the network path the Pi images use in the initramfs
|
||||||
|
(`g_ether`/`smsc95xx`/`lan78xx`). Without early networking there is no Tor and
|
||||||
|
no unlock, so emulating the literal board is pointless here. Instead we
|
||||||
|
reproduce the same *software stack* (same HOOKS chain, same tor hook, same
|
||||||
|
torrc and onion keys) on a QEMU-friendly `virt`/`q35` machine with
|
||||||
|
`virtio-net`. The only deltas from a real Pi image are the NIC driver and the
|
||||||
|
CPU architecture.
|
||||||
|
|
||||||
|
## Stages
|
||||||
|
|
||||||
|
| Stage | File | Privilege |
|
||||||
|
|---|---|---|
|
||||||
|
| Build encrypted image | `build_image.sh` | **root** (loop, cryptsetup, chroot) |
|
||||||
|
| Tor network | `tor_net.py` | rootless |
|
||||||
|
| Boot + unlock | `boot_unlock.py` | rootless (QEMU user-mode net) |
|
||||||
|
| Orchestration | `harness.py` | mixed (uses `sudo` for the build only) |
|
||||||
|
| Pure command builders | `config.py` | none — unit-tested offline |
|
||||||
|
|
||||||
|
## Keeping the host rootless
|
||||||
|
|
||||||
|
Only `build_image.sh` needs root (there is no rootless dm-crypt). To keep your
|
||||||
|
host unprivileged, run the whole harness **inside a throwaway VM** that has
|
||||||
|
`/dev/kvm`, and drive it from there. QEMU itself, the tor client and the unlock
|
||||||
|
are rootless: user-mode networking (`-netdev user`) needs no tap device, and a
|
||||||
|
Tor onion service needs no inbound port forward (its rendezvous is outbound
|
||||||
|
from both ends).
|
||||||
|
|
||||||
|
When run directly on the host, `harness.py` invokes the build via `sudo -E`
|
||||||
|
and then `chown`s the artifacts back so rootless QEMU can read them.
|
||||||
|
|
||||||
|
## Requirements
|
||||||
|
|
||||||
|
- `qemu-system-x86_64` (or `-aarch64` for `LIM_E2E_ARCH=aarch64`)
|
||||||
|
- `arch-install-scripts` (`pacstrap`, `arch-chroot`)
|
||||||
|
- `cryptsetup`, `mkinitcpio`, `tor`, `openssh`, `ncat` (SOCKS5 ProxyCommand)
|
||||||
|
- Network access to the public Tor network, **or** `chutney` for a private one
|
||||||
|
- `/dev/kvm` recommended (TCG works but is slow; aarch64-on-x86 is always TCG)
|
||||||
|
|
||||||
|
## Running
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Public Tor network (simplest; non-deterministic, needs internet):
|
||||||
|
LIM_E2E_QEMU=1 pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
|
||||||
|
|
||||||
|
# Private, offline, deterministic Tor network via chutney:
|
||||||
|
git clone https://gitlab.torproject.org/tpo/core/chutney
|
||||||
|
CHUTNEY_PATH=$PWD/chutney LIM_E2E_QEMU=1 \
|
||||||
|
pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
|
||||||
|
```
|
||||||
|
|
||||||
|
Environment knobs: `LIM_E2E_ARCH` (`x86_64` default, or `aarch64`),
|
||||||
|
`CHUTNEY_PATH` (enables the private network), `CHUTNEY_PATH` unset → public.
|
||||||
|
|
||||||
|
## Determinism note (chutney)
|
||||||
|
|
||||||
|
For a fully private loop the **guest image** must trust the same authorities as
|
||||||
|
the host client. `ChutneyNetwork.test_network_conf()` distils the
|
||||||
|
`TestingTorNetwork`/`DirAuthority` lines (rewriting `127.0.0.1` to the QEMU
|
||||||
|
user-net host alias `10.0.2.2`) into a file, which `build_image.sh` appends to
|
||||||
|
the baked-in torrc via `TOR_TEST_NETWORK_CONF`. The offline onion key
|
||||||
|
generation is unaffected — it never touches the network either way.
|
||||||
|
|
||||||
|
## What runs in CI without any of this
|
||||||
|
|
||||||
|
The pure logic (`config.py`, the env parser, the chutney torrc parsing) and the
|
||||||
|
drift guards that keep `build_image.sh` aligned with what the harness expects
|
||||||
|
are covered by [`test_qemu_harness_unit.py`](../test_qemu_harness_unit.py),
|
||||||
|
which runs in the normal `pytest` suite — no QEMU, root, or network.
|
||||||
0
tests/e2e/qemu/__init__.py
Normal file
0
tests/e2e/qemu/__init__.py
Normal file
164
tests/e2e/qemu/boot_unlock.py
Normal file
164
tests/e2e/qemu/boot_unlock.py
Normal file
@@ -0,0 +1,164 @@
|
|||||||
|
"""Boot a built image in QEMU and unlock its LUKS root over Tor.
|
||||||
|
|
||||||
|
This is the live half of the harness. QEMU runs rootless (user-mode net);
|
||||||
|
the passphrase is delivered exactly as an operator would: SSH through Tor to
|
||||||
|
the initramfs dropbear, piping the passphrase into the encryptssh prompt.
|
||||||
|
Success is observed on the serial console, not inferred.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import contextlib
|
||||||
|
import subprocess
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from tests.e2e.qemu import config
|
||||||
|
from tests.e2e.qemu.config import QemuSpec
|
||||||
|
|
||||||
|
# One generous budget covers guest boot + tor bootstrap + onion publish +
|
||||||
|
# unlock. TCG (no KVM) is slow, so keep it roomy.
|
||||||
|
UNLOCK_TIMEOUT = 420
|
||||||
|
DELIVERY_INTERVAL = 8
|
||||||
|
_SSH_ATTEMPT_TIMEOUT = 45
|
||||||
|
_POLL_INTERVAL = 2
|
||||||
|
_PROMPT_WAIT = 4 # let the encryptssh prompt appear before sending
|
||||||
|
|
||||||
|
|
||||||
|
def _serial_contains(spec: QemuSpec, needle: str) -> bool:
|
||||||
|
log = spec.serial_log
|
||||||
|
return log.is_file() and needle in log.read_text(errors="replace")
|
||||||
|
|
||||||
|
|
||||||
|
def _stream_new_serial(spec: QemuSpec, offset: int) -> int:
|
||||||
|
"""Print serial output written since ``offset``; return the new offset.
|
||||||
|
|
||||||
|
Streams the guest's boot messages (netconf, the tor hook, dropbear,
|
||||||
|
encryptssh) live so the operator sees the VM working, not a black box.
|
||||||
|
"""
|
||||||
|
log = spec.serial_log
|
||||||
|
if not log.is_file():
|
||||||
|
return offset
|
||||||
|
data = log.read_bytes()
|
||||||
|
if len(data) > offset:
|
||||||
|
print(data[offset:].decode(errors="replace"), end="", flush=True)
|
||||||
|
return len(data)
|
||||||
|
|
||||||
|
|
||||||
|
def _serial_tail(spec: QemuSpec, lines: int = 15) -> str:
|
||||||
|
if not spec.serial_log.is_file():
|
||||||
|
return "(no serial output captured)"
|
||||||
|
tail = spec.serial_log.read_text(errors="replace").splitlines()[-lines:]
|
||||||
|
return "serial tail:\n " + "\n ".join(tail)
|
||||||
|
|
||||||
|
|
||||||
|
def _qemu_stderr_path(spec: QemuSpec) -> Path:
|
||||||
|
return spec.work_dir / "qemu.stderr.log"
|
||||||
|
|
||||||
|
|
||||||
|
def _qemu_stderr_tail(spec: QemuSpec, lines: int = 10) -> str:
|
||||||
|
path = _qemu_stderr_path(spec)
|
||||||
|
if not path.is_file() or not path.read_text(errors="replace").strip():
|
||||||
|
return ""
|
||||||
|
tail = path.read_text(errors="replace").splitlines()[-lines:]
|
||||||
|
return "\nqemu stderr:\n " + "\n ".join(tail)
|
||||||
|
|
||||||
|
|
||||||
|
def _deliver_worker(spec: QemuSpec) -> None:
|
||||||
|
"""Feed the passphrase over an SSH-through-Tor session, held open.
|
||||||
|
|
||||||
|
Runs in a background thread so the main loop keeps streaming serial. The
|
||||||
|
channel is deliberately kept OPEN after sending: closing stdin immediately
|
||||||
|
(as a plain pipe does) tears the session down before cryptsetup finishes
|
||||||
|
its argon2 KDF, killing the unlock mid-flight. We wait for the remote to
|
||||||
|
close it once the boot proceeds and dropbear is torn down.
|
||||||
|
"""
|
||||||
|
proc = subprocess.Popen(
|
||||||
|
config.ssh_argv(spec),
|
||||||
|
stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
||||||
|
text=True,
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
time.sleep(_PROMPT_WAIT)
|
||||||
|
if proc.poll() is None and proc.stdin is not None:
|
||||||
|
try:
|
||||||
|
proc.stdin.write(spec.passphrase + "\n")
|
||||||
|
proc.stdin.flush()
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
proc.wait(timeout=_SSH_ATTEMPT_TIMEOUT)
|
||||||
|
except subprocess.TimeoutExpired:
|
||||||
|
proc.kill()
|
||||||
|
finally:
|
||||||
|
if proc.stdin is not None:
|
||||||
|
with contextlib.suppress(OSError):
|
||||||
|
proc.stdin.close()
|
||||||
|
|
||||||
|
|
||||||
|
def boot_and_unlock(spec: QemuSpec) -> None:
|
||||||
|
"""Boot the image, unlock it over Tor, and confirm the real system booted.
|
||||||
|
|
||||||
|
Returns normally only when the boot-ok marker appears on the serial
|
||||||
|
console; the passphrase is (re)delivered on each poll until then. Raises
|
||||||
|
with the serial tail on QEMU death or timeout.
|
||||||
|
"""
|
||||||
|
if spec.serial_log.exists():
|
||||||
|
spec.serial_log.unlink()
|
||||||
|
print(f"\n========== booting in QEMU ({spec.arch}, accel={spec.accel}) ==========")
|
||||||
|
print(f"Target onion: {spec.onion_address}:22 (unlock budget {UNLOCK_TIMEOUT}s)")
|
||||||
|
print("---- guest serial console ----", flush=True)
|
||||||
|
# QEMU's own diagnostics (bad -machine/-cpu, /dev/kvm denied, image busy)
|
||||||
|
# go to stderr; capture them so an early exit is debuggable, not opaque.
|
||||||
|
qemu_stderr = _qemu_stderr_path(spec).open("wb")
|
||||||
|
qemu = subprocess.Popen(
|
||||||
|
config.qemu_argv(spec),
|
||||||
|
stdout=subprocess.DEVNULL, stderr=qemu_stderr,
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
deadline = time.monotonic() + UNLOCK_TIMEOUT
|
||||||
|
offset = 0
|
||||||
|
next_delivery = 0.0
|
||||||
|
delivery: threading.Thread | None = None
|
||||||
|
while time.monotonic() < deadline:
|
||||||
|
offset = _stream_new_serial(spec, offset)
|
||||||
|
if _serial_contains(spec, config.BOOT_OK_MARKER):
|
||||||
|
print("\n[harness] boot-ok marker seen — LUKS root unlocked and booted.")
|
||||||
|
return
|
||||||
|
if qemu.poll() is not None:
|
||||||
|
raise RuntimeError(
|
||||||
|
f"QEMU exited early (code {qemu.returncode}).\n"
|
||||||
|
f"{_serial_tail(spec)}{_qemu_stderr_tail(spec)}"
|
||||||
|
)
|
||||||
|
# One delivery at a time; each holds its SSH session open in the
|
||||||
|
# background while we keep streaming serial and watching the marker.
|
||||||
|
idle = delivery is None or not delivery.is_alive()
|
||||||
|
if idle and time.monotonic() >= next_delivery:
|
||||||
|
print(f"\n[harness] delivering passphrase over Tor to "
|
||||||
|
f"{spec.onion_address}:22 (holding session open) ...", flush=True)
|
||||||
|
delivery = threading.Thread(
|
||||||
|
target=_deliver_worker, args=(spec,), daemon=True
|
||||||
|
)
|
||||||
|
delivery.start()
|
||||||
|
next_delivery = time.monotonic() + DELIVERY_INTERVAL
|
||||||
|
time.sleep(_POLL_INTERVAL)
|
||||||
|
raise RuntimeError(
|
||||||
|
f"Boot-ok marker never appeared within {UNLOCK_TIMEOUT}s.\n"
|
||||||
|
f"{_serial_tail(spec)}{_qemu_stderr_tail(spec)}"
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
qemu.terminate()
|
||||||
|
try:
|
||||||
|
qemu.wait(timeout=15)
|
||||||
|
except subprocess.TimeoutExpired:
|
||||||
|
qemu.kill()
|
||||||
|
qemu.wait(timeout=10)
|
||||||
|
qemu_stderr.close()
|
||||||
|
|
||||||
|
|
||||||
|
def parse_env_file(path: Path) -> dict[str, str]:
|
||||||
|
"""Read the KEY=VALUE image.env written by build_image.sh into a dict."""
|
||||||
|
result: dict[str, str] = {}
|
||||||
|
for line in path.read_text().splitlines():
|
||||||
|
if "=" in line and not line.startswith("#"):
|
||||||
|
key, value = line.split("=", 1)
|
||||||
|
result[key.strip()] = value.strip()
|
||||||
|
return result
|
||||||
182
tests/e2e/qemu/build_image.sh
Executable file
182
tests/e2e/qemu/build_image.sh
Executable file
@@ -0,0 +1,182 @@
|
|||||||
|
#!/bin/bash
|
||||||
|
# Build a QEMU-bootable, LUKS-encrypted image that reproduces lim's Tor
|
||||||
|
# unlock stack: the REAL initcpio files from lim/configuration/initcpio/,
|
||||||
|
# the tor hook wired between netconf and dropbear, offline onion keys, and
|
||||||
|
# a boot-ok marker service. Boots on a virtio machine (we model the stack,
|
||||||
|
# not the Raspberry Pi board, whose USB-gadget net QEMU cannot emulate).
|
||||||
|
#
|
||||||
|
# Needs root (loop device, cryptsetup, chroot). To keep a host rootless,
|
||||||
|
# run this inside a throwaway VM — see tests/e2e/qemu/README.md.
|
||||||
|
#
|
||||||
|
# Required environment:
|
||||||
|
# WORK_DIR writable scratch directory (outputs land here)
|
||||||
|
# REPO_ROOT linux-image-manager checkout (for the real initcpio files)
|
||||||
|
# ARCH x86_64 (default) or aarch64
|
||||||
|
# PASSPHRASE LUKS passphrase the harness will send over Tor
|
||||||
|
# SSH_PUBKEY path to the public key authorized for dropbear
|
||||||
|
#
|
||||||
|
# Writes $WORK_DIR/image.env with LUKS_UUID / MAPPER_NAME / KERNEL / INITRAMFS
|
||||||
|
# / IMAGE for the Python harness to consume.
|
||||||
|
|
||||||
|
set -euo pipefail
|
||||||
|
|
||||||
|
: "${WORK_DIR:?set WORK_DIR}"
|
||||||
|
: "${REPO_ROOT:?set REPO_ROOT}"
|
||||||
|
: "${PASSPHRASE:?set PASSPHRASE}"
|
||||||
|
: "${SSH_PUBKEY:?set SSH_PUBKEY}"
|
||||||
|
ARCH="${ARCH:-x86_64}"
|
||||||
|
MAPPER_NAME="cryptroot"
|
||||||
|
|
||||||
|
ROOTFS="$WORK_DIR/rootfs"
|
||||||
|
IMAGE="$WORK_DIR/image.raw"
|
||||||
|
INITCPIO_SRC="$REPO_ROOT/lim/configuration/initcpio"
|
||||||
|
|
||||||
|
case "$ARCH" in
|
||||||
|
x86_64) KERNEL_PKG=linux; CONSOLE=ttyS0 ;;
|
||||||
|
aarch64) KERNEL_PKG=linux; CONSOLE=ttyAMA0 ;;
|
||||||
|
*) echo "Unsupported ARCH: $ARCH" >&2; exit 2 ;;
|
||||||
|
esac
|
||||||
|
|
||||||
|
banner() { printf "\n========== %s ==========\n" "$1"; }
|
||||||
|
|
||||||
|
cleanup() {
|
||||||
|
set +e
|
||||||
|
mountpoint -q "$WORK_DIR/mnt" && umount -R "$WORK_DIR/mnt"
|
||||||
|
[ -e "/dev/mapper/$MAPPER_NAME" ] && cryptsetup close "$MAPPER_NAME"
|
||||||
|
[ -n "${LOOP:-}" ] && losetup -d "$LOOP"
|
||||||
|
}
|
||||||
|
trap cleanup EXIT
|
||||||
|
|
||||||
|
banner "bootstrapping rootfs with pacstrap"
|
||||||
|
mkdir -p "$ROOTFS"
|
||||||
|
# -c reuses the host package cache (fast). If a cached package is corrupt,
|
||||||
|
# clear it first: sudo rm -f /var/cache/pacman/pkg/<pkg>-*.pkg.tar.zst*
|
||||||
|
# A file of blank lines on stdin takes the default answer for every provider
|
||||||
|
# and confirmation prompt (e.g. the `linux` virtual package), keeping the
|
||||||
|
# bootstrap non-interactive without the SIGPIPE/pipefail pitfalls of `yes`.
|
||||||
|
printf '\n%.0s' $(seq 100) > "$WORK_DIR/pacman-answers"
|
||||||
|
pacstrap -c "$ROOTFS" \
|
||||||
|
base "$KERNEL_PKG" mkinitcpio cryptsetup \
|
||||||
|
busybox tor openssh \
|
||||||
|
mkinitcpio-utils mkinitcpio-netconf mkinitcpio-dropbear \
|
||||||
|
< "$WORK_DIR/pacman-answers"
|
||||||
|
|
||||||
|
banner "configuring mkinitcpio (virtio modules, tor between netconf/dropbear)"
|
||||||
|
# No `autodetect`: it narrows the initramfs to the *build host's* modules and
|
||||||
|
# would strip the virtio drivers the guest VM needs for early networking.
|
||||||
|
cat > "$ROOTFS/etc/mkinitcpio.conf" <<EOF
|
||||||
|
MODULES=(virtio_pci virtio_blk virtio_net)
|
||||||
|
BINARIES=(/usr/lib/libgcc_s.so.1)
|
||||||
|
FILES=()
|
||||||
|
HOOKS=(base udev modconf keyboard block ptsmount netconf tor dropbear encryptssh filesystems fsck)
|
||||||
|
EOF
|
||||||
|
|
||||||
|
banner "adding a devpts-mount hook (dropbear needs a PTY for encryptssh)"
|
||||||
|
# Without a mounted devpts, dropbear cannot allocate /dev/pts/0, so the
|
||||||
|
# interactive encryptssh passphrase prompt never runs and the SSH session
|
||||||
|
# just exits ("TIOCSCTTY: Input/output error", "/dev/pts/0: No such file").
|
||||||
|
cat > "$ROOTFS/etc/initcpio/install/ptsmount" <<'EOF'
|
||||||
|
#!/bin/bash
|
||||||
|
build() { add_runscript; }
|
||||||
|
help() { echo "Mount devpts so dropbear can allocate PTYs for encryptssh."; }
|
||||||
|
EOF
|
||||||
|
cat > "$ROOTFS/etc/initcpio/hooks/ptsmount" <<'EOF'
|
||||||
|
#!/usr/bin/ash
|
||||||
|
run_hook() {
|
||||||
|
mkdir -p /dev/pts
|
||||||
|
mount -t devpts devpts /dev/pts -o mode=620,gid=5,ptmxmode=666 2>/dev/null
|
||||||
|
[ -e /dev/ptmx ] || ln -s /dev/pts/ptmx /dev/ptmx
|
||||||
|
}
|
||||||
|
EOF
|
||||||
|
|
||||||
|
banner "installing the real lim initcpio files"
|
||||||
|
install -D -m0644 "$INITCPIO_SRC/tor_install" "$ROOTFS/etc/initcpio/install/tor"
|
||||||
|
install -D -m0644 "$INITCPIO_SRC/tor_hook" "$ROOTFS/etc/initcpio/hooks/tor"
|
||||||
|
install -D -m0644 "$INITCPIO_SRC/torrc" "$ROOTFS/etc/tor/initramfs-torrc"
|
||||||
|
|
||||||
|
if [ -n "${TOR_TEST_NETWORK_CONF:-}" ]; then
|
||||||
|
banner "wiring the image to a private (chutney) Tor network"
|
||||||
|
cat "$TOR_TEST_NETWORK_CONF" >> "$ROOTFS/etc/tor/initramfs-torrc"
|
||||||
|
fi
|
||||||
|
|
||||||
|
banner "generating onion keys offline"
|
||||||
|
install -d -m0700 "$ROOTFS/etc/tor/initramfs-onion"
|
||||||
|
: > "$WORK_DIR/keygen-torrc"
|
||||||
|
tor -f "$WORK_DIR/keygen-torrc" --DisableNetwork 1 \
|
||||||
|
--DataDirectory "$WORK_DIR/keygen-data" \
|
||||||
|
--HiddenServiceDir "$ROOTFS/etc/tor/initramfs-onion" \
|
||||||
|
--HiddenServicePort "22 127.0.0.1:22" \
|
||||||
|
--SocksPort 0 --RunAsDaemon 0 --Log "notice stderr" &
|
||||||
|
keygen_pid=$!
|
||||||
|
for _ in $(seq 1 30); do
|
||||||
|
[ -s "$ROOTFS/etc/tor/initramfs-onion/hostname" ] && break
|
||||||
|
sleep 1
|
||||||
|
done
|
||||||
|
kill "$keygen_pid" 2>/dev/null || true
|
||||||
|
ONION_ADDRESS="$(cat "$ROOTFS/etc/tor/initramfs-onion/hostname")"
|
||||||
|
|
||||||
|
banner "authorizing the ssh key for dropbear"
|
||||||
|
install -D -m0600 "$SSH_PUBKEY" "$ROOTFS/etc/dropbear/root_key"
|
||||||
|
|
||||||
|
banner "baking the boot-ok marker service"
|
||||||
|
cat > "$ROOTFS/etc/systemd/system/lim-boot-ok.service" <<'EOF'
|
||||||
|
[Unit]
|
||||||
|
Description=Signal a successful LUKS-unlock boot to the serial console
|
||||||
|
After=multi-user.target
|
||||||
|
[Service]
|
||||||
|
Type=oneshot
|
||||||
|
ExecStart=/bin/sh -c 'echo LIM_UNLOCK_BOOT_OK > /dev/console'
|
||||||
|
[Install]
|
||||||
|
WantedBy=multi-user.target
|
||||||
|
EOF
|
||||||
|
ln -sf /etc/systemd/system/lim-boot-ok.service \
|
||||||
|
"$ROOTFS/etc/systemd/system/multi-user.target.wants/lim-boot-ok.service"
|
||||||
|
ln -sf "/usr/lib/systemd/system/getty@.service" \
|
||||||
|
"$ROOTFS/etc/systemd/system/getty.target.wants/getty@$CONSOLE.service"
|
||||||
|
|
||||||
|
banner "generating the initramfs"
|
||||||
|
# arch-chroot binds /proc, /sys, /dev itself; no pre-mount needed.
|
||||||
|
arch-chroot "$ROOTFS" mkinitcpio -P
|
||||||
|
|
||||||
|
banner "creating the LUKS image"
|
||||||
|
truncate -s 4G "$IMAGE"
|
||||||
|
LOOP="$(losetup -f --show "$IMAGE")"
|
||||||
|
# Cap the argon2 memory so the unlock fits in the guest's RAM (see QemuSpec).
|
||||||
|
printf '%s' "$PASSPHRASE" | cryptsetup luksFormat --type luks2 --batch-mode \
|
||||||
|
--pbkdf-memory 262144 "$LOOP" -
|
||||||
|
LUKS_UUID="$(cryptsetup luksUUID "$LOOP")"
|
||||||
|
printf '%s' "$PASSPHRASE" | cryptsetup open "$LOOP" "$MAPPER_NAME" -
|
||||||
|
mkfs.ext4 -q "/dev/mapper/$MAPPER_NAME"
|
||||||
|
|
||||||
|
banner "copying the rootfs into the encrypted volume"
|
||||||
|
mkdir -p "$WORK_DIR/mnt"
|
||||||
|
mount "/dev/mapper/$MAPPER_NAME" "$WORK_DIR/mnt"
|
||||||
|
cp -a "$ROOTFS/." "$WORK_DIR/mnt/"
|
||||||
|
echo "/dev/mapper/$MAPPER_NAME / ext4 defaults,noatime 0 1" > "$WORK_DIR/mnt/etc/fstab"
|
||||||
|
# Kernel/initramfs filenames differ per distro (Arch: vmlinuz-linux /
|
||||||
|
# initramfs-linux.img; Manjaro: vmlinuz-6.1-x86_64 / initramfs-6.1-x86_64.img).
|
||||||
|
# Discover them instead of hard-coding, and skip the larger fallback image.
|
||||||
|
KERNEL_IMG="$(find "$ROOTFS/boot" -maxdepth 1 -name 'vmlinuz-*' | sort | head -1)"
|
||||||
|
INITRAMFS_IMG="$(find "$ROOTFS/boot" -maxdepth 1 -name 'initramfs-*.img' \
|
||||||
|
! -name '*fallback*' | sort | head -1)"
|
||||||
|
[ -n "$KERNEL_IMG" ] || { echo "No kernel image in $ROOTFS/boot" >&2; exit 1; }
|
||||||
|
[ -n "$INITRAMFS_IMG" ] || { echo "No initramfs in $ROOTFS/boot" >&2; exit 1; }
|
||||||
|
echo "Using kernel $KERNEL_IMG and initramfs $INITRAMFS_IMG"
|
||||||
|
cp "$KERNEL_IMG" "$WORK_DIR/kernel"
|
||||||
|
cp "$INITRAMFS_IMG" "$WORK_DIR/initramfs"
|
||||||
|
sync
|
||||||
|
umount -R "$WORK_DIR/mnt"
|
||||||
|
cryptsetup close "$MAPPER_NAME"
|
||||||
|
losetup -d "$LOOP"
|
||||||
|
LOOP=""
|
||||||
|
|
||||||
|
banner "writing image.env"
|
||||||
|
cat > "$WORK_DIR/image.env" <<EOF
|
||||||
|
LUKS_UUID=$LUKS_UUID
|
||||||
|
MAPPER_NAME=$MAPPER_NAME
|
||||||
|
ONION_ADDRESS=$ONION_ADDRESS
|
||||||
|
IMAGE=$IMAGE
|
||||||
|
KERNEL=$WORK_DIR/kernel
|
||||||
|
INITRAMFS=$WORK_DIR/initramfs
|
||||||
|
EOF
|
||||||
|
echo "Build complete. Onion address: $ONION_ADDRESS"
|
||||||
132
tests/e2e/qemu/config.py
Normal file
132
tests/e2e/qemu/config.py
Normal file
@@ -0,0 +1,132 @@
|
|||||||
|
"""Pure, side-effect-free builders for the QEMU unlock harness.
|
||||||
|
|
||||||
|
Everything here is deterministic string/list construction so it can be
|
||||||
|
unit-tested without QEMU, root, or a network (see test_qemu_harness_unit.py).
|
||||||
|
The stages that actually touch the system live in build_image.sh,
|
||||||
|
chutney.py and boot_unlock.py.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
# Printed to /dev/console by a oneshot service baked into the image; its
|
||||||
|
# appearance on the serial log is the proof that the LUKS root actually
|
||||||
|
# unlocked and the real system finished booting.
|
||||||
|
BOOT_OK_MARKER = "LIM_UNLOCK_BOOT_OK"
|
||||||
|
|
||||||
|
# The HOOKS the built image must carry — the tor hook sits between netconf
|
||||||
|
# and dropbear, exactly as lim.image.encryption wires it for real images.
|
||||||
|
EXPECTED_HOOKS_ORDER = ("netconf", "tor", "dropbear", "encryptssh")
|
||||||
|
|
||||||
|
# Serial console device per architecture (kernel console= and getty).
|
||||||
|
_SERIAL_CONSOLE = {"x86_64": "ttyS0", "aarch64": "ttyAMA0"}
|
||||||
|
_QEMU_BINARY = {"x86_64": "qemu-system-x86_64", "aarch64": "qemu-system-aarch64"}
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class QemuSpec:
|
||||||
|
"""Everything needed to boot one built image and unlock it."""
|
||||||
|
|
||||||
|
arch: str
|
||||||
|
work_dir: Path
|
||||||
|
image_path: Path
|
||||||
|
kernel_path: Path
|
||||||
|
initramfs_path: Path
|
||||||
|
luks_uuid: str
|
||||||
|
mapper_name: str
|
||||||
|
passphrase: str
|
||||||
|
ssh_key_path: Path
|
||||||
|
onion_address: str = ""
|
||||||
|
socks_port: int = 9050
|
||||||
|
# 2 GiB so the LUKS argon2 KDF has enough memory to unlock in the guest
|
||||||
|
# (build_image.sh also caps --pbkdf-memory to keep this headroom).
|
||||||
|
memory_mb: int = 2048
|
||||||
|
accel: str = "tcg" # "kvm" when the host arch matches and /dev/kvm exists
|
||||||
|
|
||||||
|
@property
|
||||||
|
def serial_log(self) -> Path:
|
||||||
|
return self.work_dir / "serial.log"
|
||||||
|
|
||||||
|
|
||||||
|
def qemu_binary(arch: str) -> str:
|
||||||
|
"""Return the qemu-system-* binary for an arch (raises on unsupported)."""
|
||||||
|
if arch not in _QEMU_BINARY:
|
||||||
|
raise ValueError(f"Unsupported arch: {arch}")
|
||||||
|
return _QEMU_BINARY[arch]
|
||||||
|
|
||||||
|
|
||||||
|
def kernel_cmdline(spec: QemuSpec) -> str:
|
||||||
|
"""Kernel command line unlocking the LUKS root over the mkinitcpio hooks.
|
||||||
|
|
||||||
|
``ip=dhcp`` is what the netconf hook consumes to bring up virtio-net in
|
||||||
|
early userspace; the Pi images use USB-gadget networking instead, which
|
||||||
|
QEMU cannot emulate — hence virtio here (we model the stack, not the board).
|
||||||
|
"""
|
||||||
|
console = _SERIAL_CONSOLE[spec.arch]
|
||||||
|
# The netconf hook needs the explicit ip= form
|
||||||
|
# <client>:<server>:<gw>:<netmask>:<host>:<device>:<proto> — the device
|
||||||
|
# name is mandatory (as lim's own boot config uses it); a bare "ip=dhcp"
|
||||||
|
# leaves netconf looping on "SIOCGIFFLAGS: No such device". net.ifnames=0
|
||||||
|
# keeps the virtio NIC named eth0.
|
||||||
|
# The LAST console= owns /dev/console, which the boot-ok marker echoes to
|
||||||
|
# and QEMU captures via -serial file, so the serial console must come last.
|
||||||
|
return (
|
||||||
|
f"cryptdevice=UUID={spec.luks_uuid}:{spec.mapper_name} "
|
||||||
|
f"root=/dev/mapper/{spec.mapper_name} rw "
|
||||||
|
f"ip=::::lim-e2e:eth0:dhcp net.ifnames=0 console=tty0 console={console}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def qemu_argv(spec: QemuSpec) -> list[str]:
|
||||||
|
"""Full QEMU command: direct kernel boot, virtio disk/net, serial to log.
|
||||||
|
|
||||||
|
User-mode networking (``-netdev user``) needs no root and no tap device;
|
||||||
|
it is enough for a Tor onion service, whose rendezvous is outbound from
|
||||||
|
both ends, so no inbound host port forward is required.
|
||||||
|
"""
|
||||||
|
argv = [qemu_binary(spec.arch)] # validates the arch
|
||||||
|
net_device = "virtio-net-pci" if spec.arch == "x86_64" else "virtio-net-device"
|
||||||
|
if spec.arch == "x86_64":
|
||||||
|
argv += ["-machine", "q35"]
|
||||||
|
else:
|
||||||
|
argv += ["-machine", "virt", "-cpu", "cortex-a72"]
|
||||||
|
if spec.accel == "kvm":
|
||||||
|
argv += ["-cpu", "host", "-accel", "kvm"]
|
||||||
|
elif spec.arch == "x86_64":
|
||||||
|
argv += ["-cpu", "max", "-accel", "tcg"]
|
||||||
|
argv += [
|
||||||
|
"-m", str(spec.memory_mb),
|
||||||
|
"-kernel", str(spec.kernel_path),
|
||||||
|
"-initrd", str(spec.initramfs_path),
|
||||||
|
"-append", kernel_cmdline(spec),
|
||||||
|
"-drive", f"file={spec.image_path},if=virtio,format=raw",
|
||||||
|
"-netdev", "user,id=net0",
|
||||||
|
"-device", f"{net_device},netdev=net0",
|
||||||
|
"-nographic",
|
||||||
|
"-serial", f"file:{spec.serial_log}",
|
||||||
|
"-no-reboot",
|
||||||
|
]
|
||||||
|
return argv
|
||||||
|
|
||||||
|
|
||||||
|
def ssh_proxy_command(socks_port: int) -> str:
|
||||||
|
"""ncat-based SOCKS5 ProxyCommand routing the SSH stream through Tor."""
|
||||||
|
return f"ncat --proxy 127.0.0.1:{socks_port} --proxy-type socks5 %h %p"
|
||||||
|
|
||||||
|
|
||||||
|
def ssh_argv(spec: QemuSpec) -> list[str]:
|
||||||
|
"""Non-interactive SSH into the initramfs dropbear via the onion address.
|
||||||
|
|
||||||
|
``-tt`` forces a pty so the cryptsetup askpass inside the encryptssh
|
||||||
|
session reads the passphrase we pipe on stdin. Host-key checking is off
|
||||||
|
because a throwaway onion has no known_hosts entry.
|
||||||
|
"""
|
||||||
|
return [
|
||||||
|
"ssh", "-tt",
|
||||||
|
"-o", "StrictHostKeyChecking=no",
|
||||||
|
"-o", "UserKnownHostsFile=/dev/null",
|
||||||
|
"-o", "BatchMode=yes",
|
||||||
|
"-o", f"ProxyCommand={ssh_proxy_command(spec.socks_port)}",
|
||||||
|
"-i", str(spec.ssh_key_path),
|
||||||
|
f"root@{spec.onion_address}",
|
||||||
|
]
|
||||||
180
tests/e2e/qemu/harness.py
Normal file
180
tests/e2e/qemu/harness.py
Normal file
@@ -0,0 +1,180 @@
|
|||||||
|
"""Orchestrate the full virtualized unlock: build -> tor net -> boot -> unlock.
|
||||||
|
|
||||||
|
Wires the stages together and owns their teardown. Kept thin so each stage
|
||||||
|
stays independently testable; the pure command construction lives in
|
||||||
|
config.py and is unit-tested without any of this running.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import os
|
||||||
|
import platform
|
||||||
|
import subprocess
|
||||||
|
import threading
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from tests.e2e.qemu import boot_unlock, tor_net
|
||||||
|
from tests.e2e.qemu.config import QemuSpec
|
||||||
|
|
||||||
|
IMAGE_ENV_NAME = "image.env"
|
||||||
|
_SUDO_REFRESH_INTERVAL = 60
|
||||||
|
|
||||||
|
DEFAULT_PASSPHRASE = "lim-e2e-luks-passphrase" # noqa: S105 (throwaway test secret)
|
||||||
|
DEFAULT_ARCH = "x86_64"
|
||||||
|
DEFAULT_SOCKS_PORT = 9052
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class HarnessConfig:
|
||||||
|
repo_root: Path
|
||||||
|
work_dir: Path
|
||||||
|
arch: str = DEFAULT_ARCH
|
||||||
|
passphrase: str = DEFAULT_PASSPHRASE
|
||||||
|
chutney_path: Path | None = None
|
||||||
|
chutney_network: str = "networks/basic"
|
||||||
|
|
||||||
|
|
||||||
|
def choose_accel(arch: str) -> str:
|
||||||
|
"""Use KVM only when the guest arch matches the host and /dev/kvm exists."""
|
||||||
|
host = "x86_64" if platform.machine() in ("x86_64", "AMD64") else platform.machine()
|
||||||
|
if arch == host and Path("/dev/kvm").exists():
|
||||||
|
return "kvm"
|
||||||
|
return "tcg"
|
||||||
|
|
||||||
|
|
||||||
|
def _generate_ssh_key(work_dir: Path) -> Path:
|
||||||
|
key_path = work_dir / "unlock_key"
|
||||||
|
subprocess.run(
|
||||||
|
["ssh-keygen", "-t", "ed25519", "-N", "", "-f", str(key_path), "-q"],
|
||||||
|
check=True,
|
||||||
|
)
|
||||||
|
return key_path
|
||||||
|
|
||||||
|
|
||||||
|
class _SudoKeepalive:
|
||||||
|
"""Refresh the cached sudo credential so the long build never re-prompts.
|
||||||
|
|
||||||
|
The password is entered once, up front (before the slow tor bootstrap),
|
||||||
|
then a background thread keeps the timestamp warm until stop().
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self) -> None:
|
||||||
|
self._stop = threading.Event()
|
||||||
|
self._thread = threading.Thread(target=self._loop, daemon=True)
|
||||||
|
|
||||||
|
def _loop(self) -> None:
|
||||||
|
while not self._stop.wait(_SUDO_REFRESH_INTERVAL):
|
||||||
|
subprocess.run(["sudo", "-n", "-v"], check=False)
|
||||||
|
|
||||||
|
def start(self) -> None:
|
||||||
|
self._thread.start()
|
||||||
|
|
||||||
|
def stop(self) -> None:
|
||||||
|
self._stop.set()
|
||||||
|
self._thread.join(timeout=5)
|
||||||
|
|
||||||
|
|
||||||
|
def _acquire_sudo() -> _SudoKeepalive | None:
|
||||||
|
"""Prompt for sudo once now; return a keepalive (None when already root)."""
|
||||||
|
if os.geteuid() == 0:
|
||||||
|
return None
|
||||||
|
print("\n[harness] The build stage needs root — enter your sudo password now.")
|
||||||
|
subprocess.run(["sudo", "-v"], check=True)
|
||||||
|
keepalive = _SudoKeepalive()
|
||||||
|
keepalive.start()
|
||||||
|
return keepalive
|
||||||
|
|
||||||
|
|
||||||
|
def _build_image(cfg: HarnessConfig, ssh_key: Path, test_network_conf: Path | None) -> None:
|
||||||
|
script = cfg.repo_root / "tests/e2e/qemu/build_image.sh"
|
||||||
|
env = {
|
||||||
|
**os.environ,
|
||||||
|
"WORK_DIR": str(cfg.work_dir),
|
||||||
|
"REPO_ROOT": str(cfg.repo_root),
|
||||||
|
"ARCH": cfg.arch,
|
||||||
|
"PASSPHRASE": cfg.passphrase,
|
||||||
|
"SSH_PUBKEY": f"{ssh_key}.pub",
|
||||||
|
}
|
||||||
|
if test_network_conf is not None:
|
||||||
|
env["TOR_TEST_NETWORK_CONF"] = str(test_network_conf)
|
||||||
|
# -n: never prompt here; the credential was primed by _acquire_sudo().
|
||||||
|
prefix = [] if os.geteuid() == 0 else ["sudo", "-n", "-E"]
|
||||||
|
subprocess.run([*prefix, "bash", str(script)], env=env, check=True)
|
||||||
|
# Hand the root-built artifacts back so rootless QEMU can read them.
|
||||||
|
_reclaim_work_dir(cfg, required=True)
|
||||||
|
|
||||||
|
|
||||||
|
def _reclaim_work_dir(cfg: HarnessConfig, *, required: bool = False) -> None:
|
||||||
|
"""Reclaim the root-built work dir for the invoking user via chown.
|
||||||
|
|
||||||
|
Needed before boot so rootless QEMU can read the image, and again in the
|
||||||
|
finally so a failed build never leaves root-owned files that the pytest
|
||||||
|
tmp cleanup then cannot remove.
|
||||||
|
"""
|
||||||
|
if os.geteuid() == 0 or not cfg.work_dir.exists():
|
||||||
|
return
|
||||||
|
subprocess.run(
|
||||||
|
["sudo", "-n", "chown", "-R", str(os.getuid()), str(cfg.work_dir)],
|
||||||
|
check=required,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _make_tor_net(cfg: HarnessConfig):
|
||||||
|
if cfg.chutney_path is not None:
|
||||||
|
return tor_net.ChutneyNetwork(cfg.chutney_path, cfg.work_dir, cfg.chutney_network)
|
||||||
|
return tor_net.PublicTorClient(cfg.work_dir, DEFAULT_SOCKS_PORT)
|
||||||
|
|
||||||
|
|
||||||
|
def _load_spec(cfg: HarnessConfig, ssh_key: Path, socks_port: int) -> QemuSpec:
|
||||||
|
env = boot_unlock.parse_env_file(cfg.work_dir / IMAGE_ENV_NAME)
|
||||||
|
return QemuSpec(
|
||||||
|
arch=cfg.arch,
|
||||||
|
work_dir=cfg.work_dir,
|
||||||
|
image_path=Path(env["IMAGE"]),
|
||||||
|
kernel_path=Path(env["KERNEL"]),
|
||||||
|
initramfs_path=Path(env["INITRAMFS"]),
|
||||||
|
luks_uuid=env["LUKS_UUID"],
|
||||||
|
mapper_name=env["MAPPER_NAME"],
|
||||||
|
onion_address=env["ONION_ADDRESS"],
|
||||||
|
passphrase=cfg.passphrase,
|
||||||
|
ssh_key_path=ssh_key,
|
||||||
|
socks_port=socks_port,
|
||||||
|
accel=choose_accel(cfg.arch),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def run(cfg: HarnessConfig) -> QemuSpec:
|
||||||
|
"""Execute every stage; return the QemuSpec on a fully verified unlock.
|
||||||
|
|
||||||
|
A chutney network must exist before the build so the image can be wired
|
||||||
|
to its authorities; a public client only needs to be up before boot.
|
||||||
|
"""
|
||||||
|
cfg.work_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
ssh_key = _generate_ssh_key(cfg.work_dir)
|
||||||
|
# Prime sudo before the slow tor bootstrap so the prompt is the first thing
|
||||||
|
# the operator sees, not a surprise five minutes in.
|
||||||
|
sudo = _acquire_sudo()
|
||||||
|
net = _make_tor_net(cfg)
|
||||||
|
try:
|
||||||
|
print("\n[harness] bootstrapping Tor client (~30-60s)...", flush=True)
|
||||||
|
net.start()
|
||||||
|
print(f"[harness] Tor client ready on socks port {net.socks_port}.", flush=True)
|
||||||
|
print("[harness] building encrypted image "
|
||||||
|
"(pacstrap + mkinitcpio, several minutes)...", flush=True)
|
||||||
|
_build_image(cfg, ssh_key, net.test_network_conf)
|
||||||
|
spec = _load_spec(cfg, ssh_key, net.socks_port)
|
||||||
|
boot_unlock.boot_and_unlock(spec)
|
||||||
|
return spec
|
||||||
|
finally:
|
||||||
|
# Reclaim ownership FIRST so a slow/failing teardown can never leave
|
||||||
|
# root-owned files behind, and isolate each step so none masks the
|
||||||
|
# real error propagating out of the try or skips the others.
|
||||||
|
_reclaim_work_dir(cfg)
|
||||||
|
try:
|
||||||
|
net.stop()
|
||||||
|
except Exception as exc: # noqa: BLE001 - teardown must not mask the real failure
|
||||||
|
print(f"[harness] tor network teardown failed: {exc}", flush=True)
|
||||||
|
if sudo is not None:
|
||||||
|
try:
|
||||||
|
sudo.stop()
|
||||||
|
except Exception as exc: # noqa: BLE001 - teardown must not mask the real failure
|
||||||
|
print(f"[harness] sudo keepalive teardown failed: {exc}", flush=True)
|
||||||
146
tests/e2e/qemu/tor_net.py
Normal file
146
tests/e2e/qemu/tor_net.py
Normal file
@@ -0,0 +1,146 @@
|
|||||||
|
"""Tor network providers for the QEMU unlock harness.
|
||||||
|
|
||||||
|
Two interchangeable ways to give the harness a working SOCKS port that can
|
||||||
|
reach the guest's onion service:
|
||||||
|
|
||||||
|
* PublicTorClient - a local tor client on the public network (default, both
|
||||||
|
the guest's baked-in torrc and this client use the real Tor authorities).
|
||||||
|
* ChutneyNetwork - a private, deterministic, offline Tor network via the
|
||||||
|
official `chutney` tool. Fully private operation also needs the image
|
||||||
|
built against the same authorities; test_network_conf() emits the file
|
||||||
|
build_image.sh consumes for that (see README.md).
|
||||||
|
|
||||||
|
Both expose ``socks_port``, ``start()`` and ``stop()``.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import subprocess
|
||||||
|
import time
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
BOOTSTRAP_TIMEOUT = 180
|
||||||
|
_GUEST_HOST_ALIAS = "10.0.2.2" # QEMU user-net alias for the host's 127.0.0.1
|
||||||
|
|
||||||
|
|
||||||
|
class PublicTorClient:
|
||||||
|
"""A throwaway tor client bootstrapped against the public network."""
|
||||||
|
|
||||||
|
def __init__(self, work_dir: Path, socks_port: int) -> None:
|
||||||
|
self.socks_port = socks_port
|
||||||
|
self.test_network_conf: Path | None = None
|
||||||
|
self._work_dir = work_dir
|
||||||
|
self._log = work_dir / "tor-client.log"
|
||||||
|
self._data = work_dir / "tor-client-data"
|
||||||
|
self._process: subprocess.Popen | None = None
|
||||||
|
|
||||||
|
def start(self) -> None:
|
||||||
|
self._data.mkdir(parents=True, exist_ok=True)
|
||||||
|
self._data.chmod(0o700)
|
||||||
|
# An empty -f keeps the system /etc/tor/torrc (User tor, ...) out of
|
||||||
|
# this rootless client, which would otherwise refuse to start.
|
||||||
|
empty_torrc = self._work_dir / "tor-client-torrc"
|
||||||
|
empty_torrc.write_text("")
|
||||||
|
self._process = subprocess.Popen(
|
||||||
|
[
|
||||||
|
"tor", "-f", str(empty_torrc),
|
||||||
|
"--SocksPort", str(self.socks_port),
|
||||||
|
"--DataDirectory", str(self._data),
|
||||||
|
"--Log", f"notice file {self._log}",
|
||||||
|
],
|
||||||
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
||||||
|
)
|
||||||
|
_wait_for_bootstrap(self._log, self._process)
|
||||||
|
|
||||||
|
def stop(self) -> None:
|
||||||
|
if self._process is None:
|
||||||
|
return
|
||||||
|
self._process.terminate()
|
||||||
|
try:
|
||||||
|
self._process.wait(timeout=15)
|
||||||
|
except subprocess.TimeoutExpired:
|
||||||
|
# Never leak a tor holding the fixed SOCKS port into the next run.
|
||||||
|
self._process.kill()
|
||||||
|
self._process.wait(timeout=5)
|
||||||
|
|
||||||
|
|
||||||
|
class ChutneyNetwork:
|
||||||
|
"""A private Tor network managed by the chutney tool."""
|
||||||
|
|
||||||
|
def __init__(self, chutney_path: Path, work_dir: Path, network: str) -> None:
|
||||||
|
self._chutney = chutney_path
|
||||||
|
self._network = network
|
||||||
|
self._work_dir = work_dir
|
||||||
|
self.socks_port = 0
|
||||||
|
self.test_network_conf: Path | None = None
|
||||||
|
|
||||||
|
def _run(self, *args: str) -> None:
|
||||||
|
subprocess.run(
|
||||||
|
[str(self._chutney / "chutney"), *args, self._network],
|
||||||
|
cwd=self._chutney, check=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
def start(self) -> None:
|
||||||
|
self._run("configure")
|
||||||
|
self._run("start")
|
||||||
|
self._run("wait_for_bootstrap")
|
||||||
|
client_torrc = _find_client_torrc(self._chutney / "net" / "nodes")
|
||||||
|
self.socks_port = _parse_socks_port(client_torrc)
|
||||||
|
self.test_network_conf = self._emit_network_conf(client_torrc)
|
||||||
|
|
||||||
|
def _emit_network_conf(self, client_torrc: Path) -> Path:
|
||||||
|
"""Distil the authority/test-network lines the guest image needs.
|
||||||
|
|
||||||
|
127.0.0.1 authority addresses are rewritten to the QEMU user-net host
|
||||||
|
alias so the guest's initramfs tor can reach them through user-mode
|
||||||
|
networking.
|
||||||
|
"""
|
||||||
|
wanted = ("TestingTorNetwork", "DirAuthority", "DirServer", "AlternateDirAuthority")
|
||||||
|
lines = [
|
||||||
|
line.replace("127.0.0.1", _GUEST_HOST_ALIAS)
|
||||||
|
for line in client_torrc.read_text().splitlines()
|
||||||
|
if line.strip().startswith(wanted)
|
||||||
|
]
|
||||||
|
conf = self._work_dir / "tor-test-network.conf"
|
||||||
|
conf.write_text("\n".join(lines) + "\n")
|
||||||
|
return conf
|
||||||
|
|
||||||
|
def stop(self) -> None:
|
||||||
|
self._run("stop")
|
||||||
|
|
||||||
|
|
||||||
|
def _wait_for_bootstrap(log_path: Path, process: subprocess.Popen) -> None:
|
||||||
|
deadline = time.monotonic() + BOOTSTRAP_TIMEOUT
|
||||||
|
while time.monotonic() < deadline:
|
||||||
|
if process.poll() is not None:
|
||||||
|
raise RuntimeError(
|
||||||
|
f"tor exited early (code {process.returncode}).\n{_log_tail(log_path)}"
|
||||||
|
)
|
||||||
|
if log_path.is_file() and "Bootstrapped 100%" in log_path.read_text():
|
||||||
|
return
|
||||||
|
time.sleep(1)
|
||||||
|
raise RuntimeError(f"tor client did not bootstrap in time.\n{_log_tail(log_path)}")
|
||||||
|
|
||||||
|
|
||||||
|
def _log_tail(log_path: Path, lines: int = 8) -> str:
|
||||||
|
if not log_path.is_file():
|
||||||
|
return "(no tor log written)"
|
||||||
|
tail = log_path.read_text().splitlines()[-lines:]
|
||||||
|
return "tor log:\n " + "\n ".join(tail)
|
||||||
|
|
||||||
|
|
||||||
|
def _find_client_torrc(nodes_dir: Path) -> Path:
|
||||||
|
for torrc in sorted(nodes_dir.glob("*/torrc")):
|
||||||
|
if _parse_socks_port(torrc):
|
||||||
|
return torrc
|
||||||
|
raise RuntimeError(f"No chutney client with a SocksPort under {nodes_dir}.")
|
||||||
|
|
||||||
|
|
||||||
|
def _parse_socks_port(torrc: Path) -> int:
|
||||||
|
if not torrc.is_file():
|
||||||
|
return 0
|
||||||
|
for line in torrc.read_text().splitlines():
|
||||||
|
parts = line.split()
|
||||||
|
if len(parts) >= 2 and parts[0] == "SocksPort" and parts[1].isdigit():
|
||||||
|
port = int(parts[1])
|
||||||
|
if port:
|
||||||
|
return port
|
||||||
|
return 0
|
||||||
181
tests/e2e/test_qemu_harness_unit.py
Normal file
181
tests/e2e/test_qemu_harness_unit.py
Normal file
@@ -0,0 +1,181 @@
|
|||||||
|
"""Always-on unit tests for the QEMU harness logic.
|
||||||
|
|
||||||
|
No QEMU, root, or network: they exercise the pure command builders and guard
|
||||||
|
the root build script against drifting away from what the harness expects.
|
||||||
|
Run as part of the normal suite.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from lim import config as lim_config
|
||||||
|
from tests.e2e.qemu import boot_unlock, config, harness, tor_net
|
||||||
|
from tests.e2e.qemu.config import EXPECTED_HOOKS_ORDER, QemuSpec
|
||||||
|
|
||||||
|
BUILD_SCRIPT = Path(__file__).resolve().parents[2] / "tests/e2e/qemu/build_image.sh"
|
||||||
|
|
||||||
|
|
||||||
|
def _spec(**overrides) -> QemuSpec:
|
||||||
|
base = {
|
||||||
|
"arch": "x86_64",
|
||||||
|
"work_dir": Path("/work"),
|
||||||
|
"image_path": Path("/work/image.raw"),
|
||||||
|
"kernel_path": Path("/work/kernel"),
|
||||||
|
"initramfs_path": Path("/work/initramfs"),
|
||||||
|
"luks_uuid": "1111-UUID",
|
||||||
|
"mapper_name": "cryptroot",
|
||||||
|
"passphrase": "secret",
|
||||||
|
"ssh_key_path": Path("/work/unlock_key"),
|
||||||
|
"onion_address": "abcd.onion",
|
||||||
|
"socks_port": 9052,
|
||||||
|
}
|
||||||
|
base.update(overrides)
|
||||||
|
return QemuSpec(**base)
|
||||||
|
|
||||||
|
|
||||||
|
class TestKernelCmdline:
|
||||||
|
def test_unlocks_the_luks_root_over_dhcp(self):
|
||||||
|
line = config.kernel_cmdline(_spec())
|
||||||
|
assert "cryptdevice=UUID=1111-UUID:cryptroot" in line
|
||||||
|
assert "root=/dev/mapper/cryptroot" in line
|
||||||
|
# Explicit device in the ip= form, else netconf loops on "No such device".
|
||||||
|
assert ":eth0:dhcp" in line
|
||||||
|
assert "console=ttyS0" in line
|
||||||
|
|
||||||
|
def test_serial_console_is_last_so_marker_reaches_serial_log(self):
|
||||||
|
line = config.kernel_cmdline(_spec())
|
||||||
|
assert line.index("console=tty0") < line.index("console=ttyS0")
|
||||||
|
|
||||||
|
def test_aarch64_uses_amba_serial(self):
|
||||||
|
assert "console=ttyAMA0" in config.kernel_cmdline(_spec(arch="aarch64"))
|
||||||
|
|
||||||
|
|
||||||
|
class TestQemuArgv:
|
||||||
|
def test_x86_uses_virtio_net_and_direct_kernel_boot(self):
|
||||||
|
argv = config.qemu_argv(_spec())
|
||||||
|
assert argv[0] == "qemu-system-x86_64"
|
||||||
|
assert "virtio-net-pci,netdev=net0" in argv
|
||||||
|
assert "-kernel" in argv
|
||||||
|
assert "/work/kernel" in argv
|
||||||
|
assert "-initrd" in argv
|
||||||
|
assert "/work/initramfs" in argv
|
||||||
|
assert "file=/work/image.raw,if=virtio,format=raw" in argv
|
||||||
|
assert f"file:{_spec().serial_log}" in argv
|
||||||
|
assert "user,id=net0" in argv
|
||||||
|
|
||||||
|
def test_aarch64_uses_virt_machine_and_net_device(self):
|
||||||
|
argv = config.qemu_argv(_spec(arch="aarch64"))
|
||||||
|
assert argv[0] == "qemu-system-aarch64"
|
||||||
|
assert "virt" in argv
|
||||||
|
assert "virtio-net-device,netdev=net0" in argv
|
||||||
|
|
||||||
|
def test_kvm_accel_is_emitted_when_requested(self):
|
||||||
|
assert "kvm" in config.qemu_argv(_spec(accel="kvm"))
|
||||||
|
assert "kvm" not in config.qemu_argv(_spec(accel="tcg"))
|
||||||
|
|
||||||
|
def test_unknown_arch_raises(self):
|
||||||
|
with pytest.raises(ValueError, match="Unsupported arch"):
|
||||||
|
config.qemu_argv(_spec(arch="riscv64"))
|
||||||
|
|
||||||
|
|
||||||
|
class TestQemuBinary:
|
||||||
|
def test_derives_binary_per_arch(self):
|
||||||
|
assert config.qemu_binary("x86_64") == "qemu-system-x86_64"
|
||||||
|
assert config.qemu_binary("aarch64") == "qemu-system-aarch64"
|
||||||
|
|
||||||
|
def test_unknown_arch_raises(self):
|
||||||
|
with pytest.raises(ValueError, match="Unsupported arch"):
|
||||||
|
config.qemu_binary("riscv64")
|
||||||
|
|
||||||
|
|
||||||
|
class TestSshInvocation:
|
||||||
|
def test_proxy_command_routes_through_socks5(self):
|
||||||
|
proxy = config.ssh_proxy_command(9052)
|
||||||
|
assert "--proxy 127.0.0.1:9052" in proxy
|
||||||
|
assert "--proxy-type socks5" in proxy
|
||||||
|
|
||||||
|
def test_ssh_argv_targets_onion_via_key_and_proxy(self):
|
||||||
|
argv = config.ssh_argv(_spec())
|
||||||
|
assert "root@abcd.onion" in argv
|
||||||
|
assert "-i" in argv
|
||||||
|
assert "/work/unlock_key" in argv
|
||||||
|
assert any("ProxyCommand=" in part for part in argv)
|
||||||
|
assert "StrictHostKeyChecking=no" in argv
|
||||||
|
|
||||||
|
|
||||||
|
class TestBuildScriptStaysAligned:
|
||||||
|
"""Guards: the root build script must match what the harness assumes."""
|
||||||
|
|
||||||
|
def _script(self) -> str:
|
||||||
|
return BUILD_SCRIPT.read_text()
|
||||||
|
|
||||||
|
def test_hooks_place_tor_between_netconf_and_dropbear(self):
|
||||||
|
hooks_line = next(
|
||||||
|
line for line in self._script().splitlines() if line.startswith("HOOKS=")
|
||||||
|
)
|
||||||
|
positions = [hooks_line.index(hook) for hook in EXPECTED_HOOKS_ORDER]
|
||||||
|
assert positions == sorted(positions), hooks_line
|
||||||
|
|
||||||
|
def test_installs_the_real_lim_initcpio_files(self):
|
||||||
|
script = self._script()
|
||||||
|
for name in ("tor_install", "tor_hook", "torrc"):
|
||||||
|
assert f"$INITCPIO_SRC/{name}" in script
|
||||||
|
assert "etc/dropbear/root_key" in script
|
||||||
|
|
||||||
|
def test_uses_virtio_net_and_emits_boot_marker(self):
|
||||||
|
script = self._script()
|
||||||
|
assert "virtio_net" in script
|
||||||
|
assert config.BOOT_OK_MARKER in script
|
||||||
|
|
||||||
|
def test_mounts_devpts_before_dropbear_for_pty(self):
|
||||||
|
hooks_line = next(
|
||||||
|
line for line in self._script().splitlines() if line.startswith("HOOKS=")
|
||||||
|
)
|
||||||
|
assert "ptsmount" in hooks_line
|
||||||
|
assert hooks_line.index("ptsmount") < hooks_line.index("dropbear")
|
||||||
|
assert "mount -t devpts" in self._script()
|
||||||
|
|
||||||
|
def test_real_initcpio_source_directory_exists(self):
|
||||||
|
assert (lim_config.CONFIGURATION_PATH / "initcpio" / "tor_hook").is_file()
|
||||||
|
|
||||||
|
|
||||||
|
class TestOrchestratorGlue:
|
||||||
|
def test_load_spec_reads_image_env(self, tmp_path):
|
||||||
|
(tmp_path / "image.env").write_text(
|
||||||
|
"LUKS_UUID=abc-123\n"
|
||||||
|
"MAPPER_NAME=cryptroot\n"
|
||||||
|
"ONION_ADDRESS=xyz.onion\n"
|
||||||
|
"IMAGE=/w/image.raw\n"
|
||||||
|
"KERNEL=/w/kernel\n"
|
||||||
|
"INITRAMFS=/w/initramfs\n"
|
||||||
|
)
|
||||||
|
cfg = harness.HarnessConfig(repo_root=Path("/repo"), work_dir=tmp_path)
|
||||||
|
spec = harness._load_spec(cfg, ssh_key=tmp_path / "k", socks_port=9052)
|
||||||
|
assert spec.luks_uuid == "abc-123"
|
||||||
|
assert spec.onion_address == "xyz.onion"
|
||||||
|
assert spec.image_path == Path("/w/image.raw")
|
||||||
|
assert spec.socks_port == 9052
|
||||||
|
|
||||||
|
def test_env_parser_ignores_comments_and_blanks(self, tmp_path):
|
||||||
|
env = tmp_path / "image.env"
|
||||||
|
env.write_text("# a comment\n\nKEY=value\n")
|
||||||
|
assert boot_unlock.parse_env_file(env) == {"KEY": "value"}
|
||||||
|
|
||||||
|
def test_choose_accel_falls_back_to_tcg_for_foreign_arch(self):
|
||||||
|
assert harness.choose_accel("riscv64") == "tcg"
|
||||||
|
|
||||||
|
|
||||||
|
class TestChutneyParsing:
|
||||||
|
def test_parse_socks_port_picks_first_nonzero(self, tmp_path):
|
||||||
|
torrc = tmp_path / "torrc"
|
||||||
|
torrc.write_text("SocksPort 0\nSocksPort 9008\n")
|
||||||
|
assert tor_net._parse_socks_port(torrc) == 9008
|
||||||
|
|
||||||
|
def test_find_client_torrc_skips_relays(self, tmp_path):
|
||||||
|
(tmp_path / "000a").mkdir()
|
||||||
|
(tmp_path / "000a" / "torrc").write_text("SocksPort 0\n")
|
||||||
|
(tmp_path / "001c").mkdir()
|
||||||
|
(tmp_path / "001c" / "torrc").write_text("SocksPort 9010\n")
|
||||||
|
found = tor_net._find_client_torrc(tmp_path)
|
||||||
|
assert found == tmp_path / "001c" / "torrc"
|
||||||
72
tests/e2e/test_qemu_unlock_e2e.py
Normal file
72
tests/e2e/test_qemu_unlock_e2e.py
Normal file
@@ -0,0 +1,72 @@
|
|||||||
|
"""Full virtualized end-to-end test: build -> boot in QEMU -> unlock over Tor.
|
||||||
|
|
||||||
|
Models the whole process the way lim runs it, on a QEMU-bootable virtio
|
||||||
|
machine (the Raspberry Pi's USB-gadget net can't be emulated, so we model
|
||||||
|
the software stack, not the board):
|
||||||
|
|
||||||
|
1. build a LUKS image carrying the REAL lim initcpio tor artifacts,
|
||||||
|
2. boot it in QEMU (rootless, user-mode net),
|
||||||
|
3. its initramfs runs the real netconf/tor/dropbear/encryptssh chain and
|
||||||
|
publishes the onion service,
|
||||||
|
4. deliver the LUKS passphrase by SSHing to that onion through Tor,
|
||||||
|
5. assert the boot-ok marker appears on the serial console — i.e. the real
|
||||||
|
root actually unlocked and booted.
|
||||||
|
|
||||||
|
Heavy and privileged (QEMU + a tor network; the build needs root, ideally in
|
||||||
|
a throwaway VM). Opt-in and skipped unless LIM_E2E_QEMU=1:
|
||||||
|
|
||||||
|
LIM_E2E_QEMU=1 pytest tests/e2e/test_qemu_unlock_e2e.py -v -s
|
||||||
|
|
||||||
|
Environment knobs: LIM_E2E_ARCH (x86_64|aarch64), CHUTNEY_PATH (use a private
|
||||||
|
chutney network instead of the public one). See tests/e2e/qemu/README.md.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import os
|
||||||
|
import shutil
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from tests.e2e.qemu import config as qemu_config
|
||||||
|
from tests.e2e.qemu.harness import HarnessConfig, run
|
||||||
|
|
||||||
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||||
|
|
||||||
|
# The QEMU binary is arch-specific, so derive it from LIM_E2E_ARCH instead of
|
||||||
|
# hardcoding x86_64 — otherwise the documented aarch64 run is never reachable.
|
||||||
|
_ARCH = os.environ.get("LIM_E2E_ARCH", "x86_64")
|
||||||
|
_REQUIRED = (
|
||||||
|
qemu_config.qemu_binary(_ARCH),
|
||||||
|
"cryptsetup", "pacstrap", "tor", "ssh", "ncat",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _missing_tools() -> list[str]:
|
||||||
|
return [tool for tool in _REQUIRED if shutil.which(tool) is None]
|
||||||
|
|
||||||
|
|
||||||
|
pytestmark = pytest.mark.skipif(
|
||||||
|
os.environ.get("LIM_E2E_QEMU") != "1" or bool(_missing_tools()),
|
||||||
|
reason=(
|
||||||
|
"needs LIM_E2E_QEMU=1 and "
|
||||||
|
+ ", ".join(_REQUIRED)
|
||||||
|
+ " (build stage needs root; slow)."
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_full_build_boot_and_tor_unlock(tmp_path):
|
||||||
|
chutney_path = os.environ.get("CHUTNEY_PATH")
|
||||||
|
cfg = HarnessConfig(
|
||||||
|
repo_root=REPO_ROOT,
|
||||||
|
work_dir=tmp_path / "qemu-e2e",
|
||||||
|
arch=_ARCH,
|
||||||
|
chutney_path=Path(chutney_path) if chutney_path else None,
|
||||||
|
)
|
||||||
|
|
||||||
|
spec = run(cfg)
|
||||||
|
|
||||||
|
# run() only returns after the marker was observed; re-assert for clarity.
|
||||||
|
assert spec.onion_address.endswith(".onion")
|
||||||
|
serial = spec.serial_log.read_text(errors="replace")
|
||||||
|
assert qemu_config.BOOT_OK_MARKER in serial
|
||||||
114
tests/e2e/test_tor_unlock_e2e.py
Normal file
114
tests/e2e/test_tor_unlock_e2e.py
Normal file
@@ -0,0 +1,114 @@
|
|||||||
|
"""End-to-end test of the Tor onion LUKS-unlock path (rootless).
|
||||||
|
|
||||||
|
It reproduces the operator-visible half of the decryption process that the
|
||||||
|
"tor" mkinitcpio hook drives at boot:
|
||||||
|
|
||||||
|
1. generate the v3 onion keys offline (as lim.image.tor does in the chroot),
|
||||||
|
2. start a real Tor onion service from a torrc mirroring the baked-in one,
|
||||||
|
forwarding the virtual port 22 to a local dropbear stand-in,
|
||||||
|
3. connect to the .onion address through Tor and deliver the passphrase,
|
||||||
|
4. assert the passphrase reached the unlock endpoint and it "unlocked".
|
||||||
|
|
||||||
|
The physical flashing half (loop device, cryptsetup, mount, chroot,
|
||||||
|
mkinitcpio) needs root and is out of scope here by design.
|
||||||
|
|
||||||
|
Requires the real `tor` binary and live Tor network access, so it is
|
||||||
|
opt-in and skipped unless LIM_E2E_TOR=1:
|
||||||
|
|
||||||
|
LIM_E2E_TOR=1 pytest tests/e2e/test_tor_unlock_e2e.py -v
|
||||||
|
"""
|
||||||
|
|
||||||
|
import os
|
||||||
|
import shutil
|
||||||
|
import time
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from lim.image import tor as tor_module
|
||||||
|
from tests.e2e import tor_harness
|
||||||
|
|
||||||
|
# The offline checks only need the tor binary (no network) and are
|
||||||
|
# deterministic, so they run in CI whenever tor is installed. Only the live
|
||||||
|
# onion round-trip needs the public Tor network, so it stays opt-in behind
|
||||||
|
# LIM_E2E_TOR=1 to keep an external, occasionally-flaky dependency out of the
|
||||||
|
# blocking gate.
|
||||||
|
_needs_tor = pytest.mark.skipif(
|
||||||
|
shutil.which("tor") is None, reason="needs the tor binary"
|
||||||
|
)
|
||||||
|
_needs_tor_network = pytest.mark.skipif(
|
||||||
|
shutil.which("tor") is None or os.environ.get("LIM_E2E_TOR") != "1",
|
||||||
|
reason="needs the tor binary and LIM_E2E_TOR=1 (live Tor network, slow)",
|
||||||
|
)
|
||||||
|
|
||||||
|
PASSPHRASE = b"correct horse battery staple"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def workdir(tmp_path):
|
||||||
|
(tmp_path / "onion").mkdir()
|
||||||
|
(tmp_path / "data").mkdir()
|
||||||
|
return tmp_path
|
||||||
|
|
||||||
|
|
||||||
|
def test_offline_keygen_matches_production_flags():
|
||||||
|
"""Guard: the harness keygen mirrors the flags production actually runs."""
|
||||||
|
script = tor_module._KEYGEN_SCRIPT
|
||||||
|
assert "--DisableNetwork 1" in script
|
||||||
|
assert "HiddenServicePort" in script
|
||||||
|
assert "22 127.0.0.1:22" in script
|
||||||
|
assert "--SocksPort 0" in script
|
||||||
|
|
||||||
|
|
||||||
|
@_needs_tor
|
||||||
|
def test_onion_keygen_is_deterministic_and_offline(workdir):
|
||||||
|
"""Keys generate without network and the .onion address is stable."""
|
||||||
|
address = tor_harness.generate_onion_keys(
|
||||||
|
workdir / "onion", workdir / "data"
|
||||||
|
)
|
||||||
|
assert address.endswith(".onion")
|
||||||
|
assert len(address) == len("v" * 56) + len(".onion") # v3 = 56 base32 chars
|
||||||
|
for name in ("hs_ed25519_secret_key", "hs_ed25519_public_key", "hostname"):
|
||||||
|
assert (workdir / "onion" / name).is_file()
|
||||||
|
|
||||||
|
|
||||||
|
@_needs_tor_network
|
||||||
|
def test_unlock_passphrase_travels_over_onion(workdir):
|
||||||
|
"""Full rootless round-trip: client -> Tor -> onion -> dropbear stand-in."""
|
||||||
|
onion_dir = workdir / "onion"
|
||||||
|
data_dir = workdir / "data"
|
||||||
|
onion_address = tor_harness.generate_onion_keys(onion_dir, data_dir)
|
||||||
|
|
||||||
|
backend_port = tor_harness.free_port()
|
||||||
|
socks_port = tor_harness.free_port()
|
||||||
|
torrc = workdir / "torrc"
|
||||||
|
log_path = workdir / "tor.log"
|
||||||
|
tor_harness.write_test_torrc(torrc, data_dir, onion_dir, backend_port)
|
||||||
|
|
||||||
|
dropbear = tor_harness.FakeDropbear(backend_port)
|
||||||
|
dropbear.start()
|
||||||
|
service = tor_harness.start_onion_service(torrc, socks_port, log_path)
|
||||||
|
try:
|
||||||
|
tor_harness.wait_bootstrapped(log_path, service)
|
||||||
|
|
||||||
|
# Onion descriptors need a moment to publish after bootstrap; retry.
|
||||||
|
last_error = None
|
||||||
|
for _ in range(5):
|
||||||
|
try:
|
||||||
|
stream = tor_harness.socks5_connect(socks_port, onion_address, 22)
|
||||||
|
break
|
||||||
|
except (OSError, RuntimeError) as exc:
|
||||||
|
last_error = exc
|
||||||
|
time.sleep(2)
|
||||||
|
else:
|
||||||
|
pytest.fail(f"Could not reach {onion_address} via Tor: {last_error}")
|
||||||
|
|
||||||
|
with stream:
|
||||||
|
stream.sendall(PASSPHRASE + b"\n")
|
||||||
|
reply = stream.recv(64)
|
||||||
|
finally:
|
||||||
|
service.terminate()
|
||||||
|
service.wait(timeout=15)
|
||||||
|
dropbear.stop()
|
||||||
|
|
||||||
|
assert reply.strip() == b"UNLOCKED"
|
||||||
|
assert dropbear.received == PASSPHRASE
|
||||||
178
tests/e2e/tor_harness.py
Normal file
178
tests/e2e/tor_harness.py
Normal file
@@ -0,0 +1,178 @@
|
|||||||
|
"""Rootless helpers to stand up a real Tor onion service and reach it.
|
||||||
|
|
||||||
|
These reproduce, without root/block-devices/chroot, what the "tor"
|
||||||
|
mkinitcpio hook does at LUKS-decryption time: generate v3 onion keys
|
||||||
|
offline, run a real Tor onion service that forwards the virtual port 22
|
||||||
|
to a local "dropbear", and let a client reach that service through Tor.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import socket
|
||||||
|
import struct
|
||||||
|
import subprocess
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
# Matches lim.image.tor._KEYGEN_SCRIPT: an empty -f torrc keeps the host's
|
||||||
|
# /etc/tor/torrc out of the run, --DisableNetwork 1 makes the keys appear
|
||||||
|
# without ever touching the network.
|
||||||
|
KEYGEN_TIMEOUT = 30
|
||||||
|
BOOTSTRAP_TIMEOUT = 180
|
||||||
|
ONION_CONNECT_TIMEOUT = 90
|
||||||
|
|
||||||
|
|
||||||
|
def free_port() -> int:
|
||||||
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||||||
|
sock.bind(("127.0.0.1", 0))
|
||||||
|
return sock.getsockname()[1]
|
||||||
|
|
||||||
|
|
||||||
|
def generate_onion_keys(onion_dir: Path, data_dir: Path) -> str:
|
||||||
|
"""Create v3 onion keys offline; return the .onion hostname.
|
||||||
|
|
||||||
|
Mirrors lim.image.tor._KEYGEN_SCRIPT, run directly on the host instead
|
||||||
|
of inside the image chroot (the tor invocation is identical).
|
||||||
|
"""
|
||||||
|
onion_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
onion_dir.chmod(0o700)
|
||||||
|
empty_torrc = data_dir.parent / "keygen-torrc"
|
||||||
|
empty_torrc.write_text("")
|
||||||
|
process = subprocess.Popen(
|
||||||
|
[
|
||||||
|
"tor", "-f", str(empty_torrc), "--DisableNetwork", "1",
|
||||||
|
"--DataDirectory", str(data_dir),
|
||||||
|
"--HiddenServiceDir", str(onion_dir),
|
||||||
|
"--HiddenServicePort", "22 127.0.0.1:22",
|
||||||
|
"--SocksPort", "0", "--RunAsDaemon", "0", "--Log", "notice stderr",
|
||||||
|
],
|
||||||
|
stderr=subprocess.DEVNULL,
|
||||||
|
)
|
||||||
|
hostname = onion_dir / "hostname"
|
||||||
|
try:
|
||||||
|
deadline = time.monotonic() + KEYGEN_TIMEOUT
|
||||||
|
while time.monotonic() < deadline:
|
||||||
|
if hostname.is_file() and hostname.read_text().strip():
|
||||||
|
break
|
||||||
|
time.sleep(0.5)
|
||||||
|
finally:
|
||||||
|
process.terminate()
|
||||||
|
process.wait(timeout=10)
|
||||||
|
if not (hostname.is_file() and hostname.read_text().strip()):
|
||||||
|
raise RuntimeError("Offline onion key generation produced no hostname.")
|
||||||
|
return hostname.read_text().strip()
|
||||||
|
|
||||||
|
|
||||||
|
def start_onion_service(
|
||||||
|
torrc_path: Path, socks_port: int, log_path: Path
|
||||||
|
) -> subprocess.Popen:
|
||||||
|
"""Start Tor with the given torrc (service side) plus a SOCKS port.
|
||||||
|
|
||||||
|
The torrc carries the HiddenServiceDir/HiddenServicePort just like the
|
||||||
|
baked-in image torrc; --SocksPort lets the same instance also act as the
|
||||||
|
client's entry point so the test needs only one bootstrap.
|
||||||
|
"""
|
||||||
|
return subprocess.Popen(
|
||||||
|
[
|
||||||
|
"tor", "-f", str(torrc_path),
|
||||||
|
"--SocksPort", str(socks_port),
|
||||||
|
"--Log", f"notice file {log_path}",
|
||||||
|
],
|
||||||
|
stdout=subprocess.DEVNULL,
|
||||||
|
stderr=subprocess.DEVNULL,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def wait_bootstrapped(log_path: Path, process: subprocess.Popen) -> None:
|
||||||
|
deadline = time.monotonic() + BOOTSTRAP_TIMEOUT
|
||||||
|
while time.monotonic() < deadline:
|
||||||
|
if process.poll() is not None:
|
||||||
|
raise RuntimeError(f"Tor exited early (code {process.returncode}).")
|
||||||
|
if log_path.is_file() and "Bootstrapped 100%" in log_path.read_text():
|
||||||
|
return
|
||||||
|
time.sleep(1)
|
||||||
|
raise RuntimeError("Tor did not bootstrap in time.")
|
||||||
|
|
||||||
|
|
||||||
|
def _socks5_handshake(sock: socket.socket, onion_host: str, onion_port: int) -> None:
|
||||||
|
sock.sendall(b"\x05\x01\x00") # VER, 1 method, no-auth
|
||||||
|
if sock.recv(2) != b"\x05\x00":
|
||||||
|
raise RuntimeError("SOCKS5 handshake rejected.")
|
||||||
|
host = onion_host.encode()
|
||||||
|
request = (
|
||||||
|
b"\x05\x01\x00\x03"
|
||||||
|
+ bytes([len(host)]) + host
|
||||||
|
+ struct.pack(">H", onion_port)
|
||||||
|
)
|
||||||
|
sock.settimeout(ONION_CONNECT_TIMEOUT)
|
||||||
|
sock.sendall(request)
|
||||||
|
header = sock.recv(4)
|
||||||
|
if len(header) < 2 or header[1] != 0x00:
|
||||||
|
code = header[1] if len(header) >= 2 else -1
|
||||||
|
raise RuntimeError(f"SOCKS5 CONNECT failed (reply code {code}).")
|
||||||
|
# Drain the bound-address tail so the stream starts at real payload.
|
||||||
|
atyp = header[3]
|
||||||
|
tail = {0x01: 4, 0x04: 16}.get(atyp)
|
||||||
|
if tail is None: # domain name
|
||||||
|
tail = sock.recv(1)[0]
|
||||||
|
sock.recv(tail + 2)
|
||||||
|
|
||||||
|
|
||||||
|
def socks5_connect(socks_port: int, onion_host: str, onion_port: int) -> socket.socket:
|
||||||
|
"""Open a stream to onion_host:onion_port through Tor's SOCKS5 proxy."""
|
||||||
|
sock = socket.create_connection(("127.0.0.1", socks_port), timeout=30)
|
||||||
|
try:
|
||||||
|
_socks5_handshake(sock, onion_host, onion_port)
|
||||||
|
except (OSError, RuntimeError):
|
||||||
|
sock.close()
|
||||||
|
raise
|
||||||
|
return sock
|
||||||
|
|
||||||
|
|
||||||
|
class FakeDropbear:
|
||||||
|
r"""Stand-in for the initramfs dropbear+cryptsetup unlock endpoint.
|
||||||
|
|
||||||
|
Accepts one connection on 127.0.0.1:<port>, reads a passphrase line, and
|
||||||
|
replies "UNLOCKED\n" — the observable proof that a passphrase delivered
|
||||||
|
over the onion connection reached the unlock shell.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, port: int) -> None:
|
||||||
|
self.port = port
|
||||||
|
self.received: bytes | None = None
|
||||||
|
self._server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||||
|
self._server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||||
|
self._server.bind(("127.0.0.1", port))
|
||||||
|
self._server.listen(1)
|
||||||
|
self._thread = threading.Thread(target=self._serve, daemon=True)
|
||||||
|
|
||||||
|
def start(self) -> None:
|
||||||
|
self._thread.start()
|
||||||
|
|
||||||
|
def _serve(self) -> None:
|
||||||
|
try:
|
||||||
|
self._server.settimeout(BOOTSTRAP_TIMEOUT + ONION_CONNECT_TIMEOUT)
|
||||||
|
conn, _ = self._server.accept()
|
||||||
|
with conn:
|
||||||
|
conn.settimeout(ONION_CONNECT_TIMEOUT)
|
||||||
|
self.received = conn.recv(4096).strip()
|
||||||
|
conn.sendall(b"UNLOCKED\n")
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
|
||||||
|
def stop(self) -> None:
|
||||||
|
self._server.close()
|
||||||
|
self._thread.join(timeout=5)
|
||||||
|
|
||||||
|
|
||||||
|
def write_test_torrc(path: Path, data_dir: Path, onion_dir: Path, backend_port: int) -> None:
|
||||||
|
"""Write a torrc mirroring the baked-in one, with test-writable paths.
|
||||||
|
|
||||||
|
The virtual onion port stays 22 (as in production); only the loopback
|
||||||
|
backend and the data/key directories move to rootless temp locations.
|
||||||
|
"""
|
||||||
|
path.write_text(
|
||||||
|
f"DataDirectory {data_dir}\n"
|
||||||
|
f"HiddenServiceDir {onion_dir}\n"
|
||||||
|
f"HiddenServicePort 22 127.0.0.1:{backend_port}\n"
|
||||||
|
"SocksPort 0\n"
|
||||||
|
)
|
||||||
137
tests/unit/test_tor.py
Normal file
137
tests/unit/test_tor.py
Normal file
@@ -0,0 +1,137 @@
|
|||||||
|
import pytest
|
||||||
|
|
||||||
|
from lim import config
|
||||||
|
from lim.device import Device
|
||||||
|
from lim.errors import LimError
|
||||||
|
from lim.image import encryption, tor
|
||||||
|
from lim.image.plan import ImagePlan
|
||||||
|
from lim.image.session import ImageSession
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def plan():
|
||||||
|
return ImagePlan(distribution="arch", raspberry_pi_version="4", tor_unlock=True)
|
||||||
|
|
||||||
|
|
||||||
|
def _write_onion_hostname(root, address="abcdefghijklmnop.onion"):
|
||||||
|
onion_dir = root / tor.ONION_DIR
|
||||||
|
onion_dir.mkdir(parents=True)
|
||||||
|
(onion_dir / "hostname").write_text(f"{address}\n")
|
||||||
|
|
||||||
|
|
||||||
|
class TestConfigureTorUnlock:
|
||||||
|
def test_installs_hooks_and_returns_address(self, tmp_path, plan, fake_runner):
|
||||||
|
_write_onion_hostname(tmp_path, "stableaddress.onion")
|
||||||
|
|
||||||
|
address = tor.configure_tor_unlock(plan, tmp_path)
|
||||||
|
|
||||||
|
assert address == "stableaddress.onion"
|
||||||
|
package_installs = [
|
||||||
|
input_text
|
||||||
|
for _, _, input_text in fake_runner.calls
|
||||||
|
if input_text and "pacman" in input_text and "tor busybox" in input_text
|
||||||
|
]
|
||||||
|
assert len(package_installs) == 1
|
||||||
|
assert len(fake_runner.find("install", "tor_install", "etc/initcpio/install/tor")) == 1
|
||||||
|
assert len(fake_runner.find("install", "tor_hook", "etc/initcpio/hooks/tor")) == 1
|
||||||
|
assert len(fake_runner.find("install", "torrc", "etc/tor/initramfs-torrc")) == 1
|
||||||
|
# Existing keys must be kept: no keygen chroot run.
|
||||||
|
assert fake_runner.find("chroot", "DisableNetwork") == []
|
||||||
|
|
||||||
|
def test_generates_keys_when_missing(self, tmp_path, plan, fake_runner):
|
||||||
|
with pytest.raises(LimError, match="produced no"):
|
||||||
|
tor.configure_tor_unlock(plan, tmp_path)
|
||||||
|
# FakeRunner executes nothing, so the hostname file never appears —
|
||||||
|
# but the keygen chroot script must have been issued exactly once.
|
||||||
|
keygen_calls = [
|
||||||
|
(kind, cmd, input_text)
|
||||||
|
for kind, cmd, input_text in fake_runner.calls
|
||||||
|
if input_text and "DisableNetwork" in input_text
|
||||||
|
]
|
||||||
|
assert len(keygen_calls) == 1
|
||||||
|
assert "HiddenServiceDir" in keygen_calls[0][2]
|
||||||
|
|
||||||
|
def test_hook_resources_exist(self):
|
||||||
|
source_dir = config.CONFIGURATION_PATH / "initcpio"
|
||||||
|
for name in ("tor_install", "tor_hook", "torrc"):
|
||||||
|
assert (source_dir / name).is_file()
|
||||||
|
|
||||||
|
|
||||||
|
class TestInitcpioHookHardening:
|
||||||
|
"""Guards for review findings in the shipped initramfs hooks."""
|
||||||
|
|
||||||
|
def _read(self, name):
|
||||||
|
return (config.CONFIGURATION_PATH / "initcpio" / name).read_text()
|
||||||
|
|
||||||
|
def test_install_bakes_the_dns_resolver(self):
|
||||||
|
# Without the NSS module the NTP hostname never resolves and the clock
|
||||||
|
# stays at 1970, so Tor rejects the consensus and never publishes.
|
||||||
|
assert "libnss_dns.so.2" in self._read("tor_install")
|
||||||
|
|
||||||
|
def test_hook_never_sources_dhcp_lease_files(self):
|
||||||
|
# Sourcing /tmp/net-*.conf would run attacker-controlled DHCP option
|
||||||
|
# strings as root before the LUKS root is unlocked.
|
||||||
|
hook = self._read("tor_hook")
|
||||||
|
assert '. "$conf"' not in hook
|
||||||
|
assert "sed -n 's/^IPV4DNS" in hook
|
||||||
|
|
||||||
|
def test_hook_attempts_ntp_without_gating_on_dhcp_dns(self):
|
||||||
|
# NTP must run even when tor_ntp is an IP literal (no DNS needed).
|
||||||
|
hook = self._read("tor_hook")
|
||||||
|
assert "skipping NTP sync" not in hook
|
||||||
|
|
||||||
|
|
||||||
|
class TestBootloaderNetworking:
|
||||||
|
"""The cmdline.txt boot path (RPi4-class) must set ip= for remote unlock."""
|
||||||
|
|
||||||
|
def _session(self, tmp_path):
|
||||||
|
session = ImageSession(Device("mmcblk0"))
|
||||||
|
session.root_partition_uuid = "ROOT-UUID"
|
||||||
|
session.root_mapper_name = "cryptroot"
|
||||||
|
session.root_mapper_path = "/dev/mapper/cryptroot"
|
||||||
|
session.boot_mount_path = tmp_path / "boot"
|
||||||
|
session.boot_mount_path.mkdir()
|
||||||
|
return session
|
||||||
|
|
||||||
|
def test_cmdline_txt_gets_network_params(self, tmp_path, plan):
|
||||||
|
root = tmp_path / "root"
|
||||||
|
(root / "etc").mkdir(parents=True)
|
||||||
|
(root / "etc" / "hostname").write_text("myhost\n")
|
||||||
|
session = self._session(tmp_path)
|
||||||
|
(session.boot_mount_path / "cmdline.txt").write_text(
|
||||||
|
"root=/dev/mmcblk0p2 rw rootwait\n"
|
||||||
|
)
|
||||||
|
|
||||||
|
encryption._configure_bootloader(plan, session, root)
|
||||||
|
|
||||||
|
content = (session.boot_mount_path / "cmdline.txt").read_text()
|
||||||
|
assert "ip=::::myhost:eth0:dhcp" in content
|
||||||
|
assert "net.ifnames=0" in content
|
||||||
|
assert "cryptdevice=UUID=ROOT-UUID:cryptroot" in content
|
||||||
|
assert "root=/dev/mmcblk0p2" not in content
|
||||||
|
|
||||||
|
|
||||||
|
class TestMkinitcpioHooksLine:
|
||||||
|
def _mkinitcpio_conf(self, root):
|
||||||
|
path = root / "etc/mkinitcpio.conf"
|
||||||
|
path.parent.mkdir(parents=True)
|
||||||
|
path.write_text(
|
||||||
|
"MODULES=()\n"
|
||||||
|
"BINARIES=()\n"
|
||||||
|
f"HOOKS=({encryption.MKINITCPIO_HOOKS_PREFIX} "
|
||||||
|
f"{encryption.MKINITCPIO_HOOKS_SUFFIX})\n"
|
||||||
|
)
|
||||||
|
return path
|
||||||
|
|
||||||
|
def test_tor_hook_between_netconf_and_dropbear(self, tmp_path, plan, fake_runner):
|
||||||
|
path = self._mkinitcpio_conf(tmp_path)
|
||||||
|
encryption._configure_mkinitcpio(plan, tmp_path)
|
||||||
|
assert "netconf tor dropbear encryptssh" in path.read_text()
|
||||||
|
|
||||||
|
def test_no_tor_hook_when_disabled(self, tmp_path, plan, fake_runner):
|
||||||
|
plan.tor_unlock = False
|
||||||
|
path = self._mkinitcpio_conf(tmp_path)
|
||||||
|
encryption._configure_mkinitcpio(plan, tmp_path)
|
||||||
|
content = path.read_text()
|
||||||
|
assert "netconf dropbear encryptssh" in content
|
||||||
|
assert " tor " not in content
|
||||||
Reference in New Issue
Block a user