Compare commits

..

10 Commits

Author SHA1 Message Date
admin 4a1a84941c D43 = A run close-out: REPORT and CONTEXT. Docs only.
gates / gates (push) Successful in 1m40s
2026-10-24 21:04:19 +02:00
admin c8d3c17b13 D43 = A (R-1257): the root-key-only sshd drop-in only on an appliance; on BYO nothing is written and our copy is removed; the in-force check reads the LAN addresses (X3); a failing sshd -t undoes only the SSH step (X4). Commit only, not released.
gates / gates (push) Successful in 1m32s
2026-10-24 20:47:17 +02:00
admin f5c3d77a61 Review round 2: .claude/rules/unprompted-work.md aligned with its four siblings (the missing 'A version number is pushed once' fact paragraph); no code change
gates / gates (push) Successful in 1m32s
2026-10-24 15:50:59 +02:00
admin 5e05627c4e REPORT: R-1250 test on request, live proof on demo-hp
gates / gates (push) Successful in 1m41s
2026-10-24 13:48:07 +02:00
admin 1e087bb158 linetest: waiting_for_quiet needs the full watched window (R-1250)
gates / gates (push) Successful in 1m35s
The live proof found a fresh agent answering waiting before it had watched one minute; a quiet
box's torrent app would be held for nothing.
2026-10-24 13:24:41 +02:00
admin 29bcf4403e linetest: a test on request for a box that seeds all day (R-1250)
gates / gates (push) Successful in 1m37s
GET /host/line-test reports waiting_for_quiet (due, in the window, tunnel up, only the quiet
missing); POST /host/line-test/run runs one measurement with a 5 s settle and a 10 s quiet
check instead of 10 quiet minutes, runs 5 s apart (~55 s). Every other weekly rule holds.
Commit only; the release rides the next train.
2026-10-24 13:18:59 +02:00
admin 08ca67861f CHANGELOG: v0.164.0 released (the weekly upload line test, R-1247), delivered to the three test boxes, vouched by the operator; REPORT
gates / gates (push) Successful in 1m32s
2026-10-24 09:46:00 +02:00
admin f7e1dcbb27 R-1247: the weekly upload line test — the agent's own iperf3 client to ep0's tunnel-only server, Tuesday 10:00-15:59 after 10 quiet minutes, under the heavy-op gate, best of 3; line_test stanza + GET /host/line-test; 15 tests red-proved; unreleased (next v0.164.0)
gates / gates (push) Successful in 1m35s
2026-10-24 08:59:42 +02:00
admin 2d4b1930df REPORT: day 2026-10-23 — agent 0.163.0 released, delivered (step bundle), vouched by the operator
gates / gates (push) Successful in 1m48s
2026-10-23 11:08:14 +02:00
admin b4721eee11 CHANGELOG: v0.163.0 released (D30, D4 root SSH by key only, R-1204)
gates / gates (push) Successful in 2m5s
2026-10-23 08:44:19 +02:00
20 changed files with 2146 additions and 42 deletions
+6
View File
@@ -50,6 +50,12 @@ it **first** in the morning note. A decision you cannot write in that shape is o
on Tester 1, demo-hp and demo-felhom through the normal update path and the box walk's checks are green
on all three — not at night unless the night brief says so. **Agent releases still stop for the
operator's vouch.** The golden is baked weekly and vouched by the operator; CC never vouches. **No `--no-verify`.**
**A version number is pushed once** — `felhom-controller/controller/build.sh` now refuses a second
push of a tag that exists, and a registry it cannot reach is a refusal too. This is a FACT about
what the build does, not a new rule: a fix for something already pushed is the next patch version,
and a box on a bad image is moved by the normal update path, never by re-tagging. Earned on
2026-10-15, when 0.315.0 was pushed three times from three different commits and a cache kept
serving the first one (`09-update-architecture.md` §0).
7. **An enumerated gap becomes a row in the same session — or, if it is small, is fixed in it** (the size rule).
Prose is not a record.
8. **Hungarian text is searched with ASCII fragments**, with a positive and a negative control.
+91 -3
View File
@@ -1,4 +1,92 @@
## Unreleased (2026-10-22/23 night) — the tunnel heals itself after a clock that ran ahead (D30, R-1204) — COMMITTED, NOT RELEASED
## Unreleased (next: v0.165.0) — D43 = A: BYO hosts keep their own SSH settings (R-1257) (2026-10-24 night) — committed, NOT released (the release + the operator's vouch are Monday's)
- **The root-key-only sshd drop-in (`/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf`, R-982 / D4) goes only to an
appliance** — operator ruling D43 = A (2026-10-24): a BYO host belongs to its owner (`11` §1). The table entry's policy
is now `appliance`: written only when the ROOT-OWNED install record (`/var/lib/felhom-install/state.json`) says
`appliance`. On a `byo` record nothing is written, and a copy WE wrote (this exact path, beginning with our header
line) is REMOVED, its previous copy kept for the undo, then `sshd -t` and a reload. A box whose record is absent or
agent-writable proves nothing and is left exactly as it is. The bundle report lists `removed`; the record says
`removed` / `skipped`. The installer records the mode before it installs the bundle, so a new appliance still gets it.
- **X3: the „in force" check asks sshd about each of the host's own LAN addresses** (`ip -4 -o addr show scope
global`), not only 192.0.2.1 — a `Match Address <LAN>` block that re-allows root passwords no longer reads as in force.
- **X4: a failing `sshd -t` puts back ONLY the SSH step** (`undo_sshd_step`: the previous copy, or none; sshd is not
reloaded, so it never ran the refused state) and the rest of the bundle stays; it used to refuse and undo the whole
bundle. The report carries `sshd_step_undone`; the record names what is on disk.
- Tests (`configs/test_felhom_config_bundle.py`, class `ByoKeepsItsSSH`, 7 tests; each rule seen failing with it taken
out): a BYO host gets no drop-in; our copy comes off a BYO host; the owner's own file stays; no record → left as it
is; a LAN `Match` block is not in force; every address key-only passes; a failing `sshd -t` undoes only the SSH step.
The fake box now carries an appliance record by default (every earlier test unchanged).
## 2026-10-24 (review round 2, unattended) — `.claude/rules/unprompted-work.md` aligned with its four siblings
- The copy lacked item 6's paragraph „A version number is pushed once …" that the felhom.eu and felhom-controller copies
have carried since 2026-10-15; the file's header says „change all five or none". The paragraph states a fact
(`felhom-controller/controller/build.sh` refuses an existing tag and an unreachable registry). Now byte-identical to
the other four (md5 b1891bd4). No code change. Record: felhom.eu `documentation/audits/review-round-2-2026-10-24/EDITS.md`.
## Unreleased (next: v0.165.0) — the line test for a box that seeds all day (R-1250) (2026-10-24) — committed, not released
Attended brief 2026-10-24, Part 4 (commit only; the release rides the next normal train). The weekly test (v0.164.0)
starts only after 10 quiet minutes; a torrent app that seeds all day never gives them, so that box never got a tested
line and qBittorrent never got its 50 % limit. Now:
- **`linetest.Tester.WaitingForQuiet`** — due, inside the window, tries left, tunnel up, and only the quiet missing.
`GET /host/line-test` carries it as `waiting_for_quiet`; a controller with a seeding torrent app acts only on it.
It needs the full watched window (11 minute samples): a freshly started agent is not known to be busy, so it never
invites a hold (found in the live proof, 2026-10-24: the first build said `true` before one minute was watched;
`TestFreshTesterIsNotWaiting`).
- **`linetest.Tester.RunNow`** and **`POST /host/line-test/run`** (synchronous, about 55 s) — a test on request while the
controller holds its app's upload low: every weekly rule holds (due, window, `MaxTriesPerDay`, tunnel, the one heavy-op
gate) except the 10 quiet minutes, which become a 5 s settle and a 10 s quiet check (below 2 Mbit/s, so a request
never measures the family's own upload); the three runs are 5 s apart instead of 30. A refusal answers its code
(`not-due`, `outside-window`, `tries-used`, `no-tunnel`, `heavy-op`, `not-quiet`, `server-busy`); a failed request
spends one of the day's tries. `Tick` and `RunNow` now share one `ready()` check.
- Tests (each seen failing): `TestSeedingBoxIsWaitingForQuietAndRunNowMeasuresIt` (the sleeps, the 60 s bound),
`TestRunNowRefusesAnUplinkThatIsStillBusy`, `TestRunNowKeepsTheWeeklyRules`, `TestRunNowFailuresSpendTheDaysTries`,
`TestQuietBoxIsNotWaiting`, `TestLineTestRunRoute`; a compile-time assertion that the Tester serves the request half.
**No sudoers or config-bundle change. MinController: none** — an older controller never calls the route.
## v0.164.0 — the weekly upload line test (R-1247) (2026-10-24) — RELEASED on the operator's ruling (attended release run); vouched by the operator 09:44:58 CEST
Binary sha256 `1b803d8b70f8afbe1546f6989f35b6b0e29c9ed63f6acb3843d3c04ef0d24b2b`; config bundle sha256
`1d7b54d39d0c1aff682ba0c0a7f91aea0f9407a10d2f5a8bbfdc858d32c13341` (configs unchanged — no bundle job sent). Built from
`f7e1dcb` (CI 2277 green) in a fresh clone by `scripts/release-agent.sh`. Delivered by one signed `agent_update` per box
(07:25Z): Tester 1, demo-hp and demo-felhom — COMPLETED, self-check 76/76, committed after the 1-minute dwell, the line
test started (felhom.eu `audits/line-test-2026-10-24/agent-0164-boxes.txt`).
Operator ruling 2026-10-24 07:26: once a week the box's HOST measures its upload line with a 10-second iperf3 upload to an
iperf3 server on ep0 that answers **only inside the WireGuard tunnel** (felhom.eu `scripts/ep0-iperf3/`; installed the same
day). The agent speaks the iperf3 protocol itself (operator's pick, 2026-10-24: nothing is installed on any host).
**No sudoers or config-bundle change.** **MinController: none** — the controller (≥ 0.342.0) reads the new route and
treats a 404 as „not measured".
- **`internal/linetest`** — `Client` (the iperf3 3.x client protocol: cookie, params, N data streams, TEST_END, results
exchange; the value is the SERVER's received bytes) and `Tester` (the schedule): due weekly from Tuesday 10:00 local,
starts 10:00–15:59 only (never in the night chain), from a per-host slot spread over 10:00–13:59; only after 10 quiet
minutes (every one-minute transmit rate of the host's physical uplink below 2 Mbit/s); under the one-heavy-op gate
(`line-test`); 3 runs of 10 s with 4 streams, 30 s apart, the highest counts; each run bounded by 25 s; „server busy" →
tried again at the next tick the same day, at most 6 tries a day; no operational tunnel → no test, no value. The last
result is persisted (`<oob state dir>/line-test.json`).
- **Report:** a `line_test` stanza (`schedule`, `last_ok_at`, `mbps`, `runs_mbps`, `server`, `last_attempt_at`,
`last_error`). **Local API:** `GET /host/line-test` → `{present, mbps, measured_at, method:"iperf3-tunnel"}`.
- `wgtunnel.Manager.LineTestTarget()` — the desired block's `pbs_tunnel_ip`, only while the tunnel is operational
(cached, no exec). Config `line_test.disable` (default on).
- **Proven live** (felhom.eu `audits/line-test-2026-10-24/`): a probe built from this package on demo-hp's host read
**90.7 Mbit/s** in 10.4 s (the push floor said „at least 10"); a second test 2 s later → „server busy" in 0.1 s while the
first went on undisturbed (90.6).
- Tests, each red-proved by a mutation (15): `TestClientSpeaksTheProtocolAndCountsTheServersBytes` (against a fake
iperf3 server), `TestClientBusyServerIsErrBusy`, `TestClientHonoursTheDeadlineAgainstAStalledServer`,
`TestEligibleWindowNeverInTheNightChain`, `TestNeverTestsInTheNightOrBeforeTuesday`, `TestBusyServerIsRetriedLaterTheSameDay`,
`TestTriesPerDayAreBounded`, `TestNoTunnelNoTest`, `TestBusyUplinkOrHeavyOpWaits`, `TestTickMeasuresTheHighestOfThreeRuns`,
`TestStateSurvivesARestartAndReportsIt`, `TestUplinkCountsTheBridgesPhysicalPortsOnly`, `TestLineTestRoute`,
`TestLineTestTargetOnlyWhileOperational`, `TestMainWiresTheLineTest`.
## v0.163.0 — the tunnel heals itself after a clock that ran ahead (D30, R-1204); root SSH from the LAN by key only (D4, R-982) (2026-10-23) — RELEASED on the operator's yes (day brief); the vouch is the operator's
Binary sha256 `561b5316cd511de9fa9d8b8be4553da7add684f715d3bea533ca330ff7f3763c`; config bundle sha256 `3afb88d9fbbc4adc16083fa52a98cd4974583c6378a77a1a6e0ed8ae6d1b80c8`. Built from `1109092` (CI 2246 green). Both sections below are in it.
### The tunnel re-key (D30) and the D4 bundle drop-in
- **D30 (operator 2026-10-22: the agent makes a new key and registers it on the existing peer path).** When the latest
handshake stays stamped in the future for 15 minutes (the R-1204 sign: ep0 keeps the future timestamp and refuses
@@ -32,9 +120,9 @@
- **Known gap, unchanged (filed):** after a reboot the kernel forgets the future stamp (latest handshake 0), so neither
this re-key nor the hub's alarm sees a box whose clock was corrected and which then restarted (R-1223).
## Unreleased (2026-10-22 night) — the tunnel's future-stamped handshake is reported (R-1204) — COMMITTED, NOT RELEASED (no agent release tonight); root SSH by key only (R-982, D4) is a READY PATCH, not on main
### The future-stamped handshake is reported (R-1204) — first written on the 2026-10-22 night
- **R-982 / D4 — a ready patch, NOT on main.** Built and proven the same night (felhom.eu
- **R-982 / D4 — a ready patch on that night; on `main` since `1109092` and released in v0.163.0 (see above).** Built and proven the same night (felhom.eu
`audits/night-2026-10-22/P2-D4/`): a bundle drop-in `/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf`
(`PermitRootLogin prohibit-password`), a `sshd` content check (exactly `prohibit-password`; `sshd -t -f`), and after
an install that wrote it `sshd -t` + `systemctl reload ssh`; tests `RootKeyOnly` (3), red-proved. Live on a fresh
+11
View File
@@ -1,5 +1,16 @@
# CONTEXT — felhom-agent working state
> **2026-10-24 night (unattended) — D43 = A on `main` (`c8d3c17`), NOT released (next v0.165.0, with R-1250).** The sshd
> drop-in's bundle policy is `appliance` (install record `/var/lib/felhom-install/state.json` mode); BYO: nothing written,
> our copy removed; no proof: untouched. X3: `lan_addresses()` for `sshd -T`. X4: `undo_sshd_step` — only the SSH file is
> put back. Release + the operator's vouch: Monday's list (felhom.eu STATUS).
> **2026-10-24 (attended release run) — v0.164.0: the weekly upload line test (R-1247; operator ruling 07:26).**
> `internal/linetest`: the agent's own iperf3 client (operator's pick — nothing installed on hosts) to ep0's tunnel-only
> server `10.77.0.1:5201`, Tuesday 10:00-15:59, after 10 quiet minutes, under the heavy-op gate, best of 3; `line_test`
> stanza + `GET /host/line-test` (the controller's qBittorrent limit, D41 = 50 %). demo-hp measured 90.7 Mbit/s. Release
> needs the operator's vouch.
> **2026-10-22/23 night — D30 on `main` (`ae69ad4`), unreleased; the D4 patches v2 ready.** Operator ruling D30
> (2026-10-22): a handshake stamped in the future for 15 min → a new WG key pair (`wgtunnel/rekey.go`; old key kept as
> `wg/private.key.prev`), registered on `POST /hosts/<id>/wg` (the hub re-keys in place; its full-list push swaps the
+14 -21
View File
@@ -1,23 +1,16 @@
# REPORT — night 2026-10-22 → 23: D30 on `main`, the D4 patches v2, nothing released
# REPORT — D43 = A: BYO hosts keep their own SSH settings (R-1257), 2026-10-24 night (unattended; commit only)
**What:** the night brief's agent parts. No release, no vouch. Design read: `03-host-agent.md` (the tunnel, the key, the
bundle), `06-offsite-connectivity.md` (the full-list push to ep0).
Brief: the restore-holes run, Part 4. Starting the run was the operator's ruling **D43 = A**. Previous report (R-1250's
line hold, attended, same day): `git show f5c3d77:REPORT.md`.
- **D30** (`ae69ad4`, hardened `39fb9cb`, `b9b831f`): a handshake stamped in the future for 15 min → a new key pair,
registered on `POST /hosts/<id>/wg` (the hub re-keys in place; its full-list push swaps the key on ep0 — nothing new
on ep0). The key file is replaced by an fsynced rename over it (never missing or empty); the first old key stays
`private.key.prev`, later ones dated (3 kept); the once-an-hour bound survives a restart; a wait stamp dated after now
is dropped. Tests `TestD30_*` (7), each red-proved.
- **Live** (felhom.eu `audits/night-2026-10-23/P2-D30/`, test box Tester4-16fdef, fixture `ae69ad4`): with ep0 free to
start a handshake, a 2-hour clock jump healed by itself and no re-key ran (correct); with ep0's starts dropped on the
box, the tunnel stayed dead, the agent re-keyed at 15 min, the hub took the key (`existed=false generation=3
sync=ok`), the box switched at its next desired-state fetch and shook hands with the new key.
- **D4 patches v2** (felhom.eu `audits/night-2026-10-23/P3-review/`), after a fresh review (A1–A7): `try-reload-or-restart`,
an `sshd -T` read-back reported as `sshd_dropin_not_in_force`, a reload in undo, the right citation; installer: a warning
when the reload fails. RootKeyOnly 6/6 (two new, red-proved); installer tests 40/40 with the patch. Both apply to
tonight's mains. Live by hand on the test box: root password SSH refused, key works, web password 200.
- **Gap filed:** a reboot after the clock was corrected forgets the future stamp (felhom.eu R-1223).
CI by commit: `ae69ad4` 2229, `39fb9cb` 2235 green; `b9b831f` and this docs commit: see felhom.eu
`REPORT-night-2026-10-23.md`. Fixture names on the test box: `0.163.0-night.1` (= `ae69ad4`), then `0.162.99`
(= `b9b831f`; a plain number, because the controller cannot compare a pre-release string — the off-site unlock refused).
- **`configs/felhom-os-apply` (commit `c8d3c17`):** the root-key-only sshd drop-in is an `appliance`-policy file —
written only when the root-owned install record says `appliance`; on `byo` nothing is written and a copy we wrote
(exact path, our header line) is removed (the previous copy kept for the undo, then `sshd -t` and a reload); a box
whose record proves nothing is left as it is. **X3:** the in-force check asks `sshd -T` for each of the host's own
global IPv4 addresses. **X4:** a failing `sshd -t` puts back only the SSH file (`sshd_step_undone` in the report);
the rest of the bundle stays.
- **Tests:** `configs/test_felhom_config_bundle.py` class `ByoKeepsItsSSH` (7), each rule seen failing with it taken out
(policy back to `replace`; the probe back to 192.0.2.1; `sshd -t` back to `must`). The fake box now carries an
appliance record by default; all 68 bundle tests and 146 os-apply tests pass; `agent_gates.py --fast` OK.
- **Not done here (Monday):** the agent release v0.165.0 and the operator's vouch; a signed bundle on a BYO test host.
- Docs: felhom.eu `03` (the D4 paragraph) and `11` §1 carry the ruling; register R-1257 → VERIFY.
+1
View File
@@ -171,6 +171,7 @@
| `localapi.GuestAPI` / `BackupService` / `BackupStore` / `TokenAuthority` | internal/localapi/server.go | `*proxmox.Client`, `*backup.BackupRunner`, `*backup.Store`, `*TokenStore` | `fakeGuests`/`fakeBackups`/`fakeStore` internal/localapi/server_test.go |
| `backup.InFlight` | internal/backup/inflight.go | `TryAcquire(what) (release, busy, ok)` / `Busy()` | THE host-wide "one heavy guest operation at a time" gate — shared by the local-API backup path and the restore-test scheduler (R-85) | A **LINK** guard, not a lock one: the scratch VMID never touches the live guest's vzdump lock, but an offsite restore PULLS multi-GB over the tunnel a backup PUSHES one. Callers **DEFER, never cancel** — a deferred restore-test costs coverage, a cancelled backup costs the backup. A nil gate is ungated (pre-R-85 callers). |
| `fstrim.Trimmer` (R-444) | internal/fstrim/fstrim.go | `New(runner, guests, gate, statePath, logger)` / `Pass(ctx)` / `GuestDiskTrimStatus(ctx)` / `ParseTrimmed(out)` | the weekly `pct fstrim <vmid>` of owned running guests (Wednesday from 10:00 local, starts 10:00-20:59 only), under `backup.InFlight`; last result per guest persisted and reported as `guest_disk_trim` | A busy gate DEFERS to the next hourly tick, never waits; a failed trim retries at most `MaxAttemptsPerWeek`; the report reads the persisted record, it never runs pct |
| `linetest.Tester` + `linetest.Client` (R-1247) | internal/linetest/linetest.go · internal/linetest/iperf.go | `New(runner, target, gate, uplink, hostID, statePath, logger)` / `Tick(ctx)` / `Value()` / `LineTestStatus(ctx)`; `Client{Streams, Time}.Run(ctx, addr)` | the weekly upload line test: a 10-second iperf3 upload (the agent speaks the iperf3 protocol — nothing installed on the host) to ep0's tunnel-only server (`wgtunnel.Manager.LineTestTarget`), Tuesday 10:00-15:59 local from a per-host slot, after 10 quiet minutes of the PHYSICAL uplink (`SysUplinkTx`: the bridge's ports with a `device` link — the bridge counter misses bridged guest traffic), under `backup.InFlight`; best of 3; reported as `line_test`, served as GET /host/line-test | **The value is the SERVER's received bytes**, never the client's written ones (the send buffer). `Run` refuses a context without a deadline. iperf3 numbers streams 1, 3, 4, 5 — the server matches results by those ids. „server busy" (ACCESS_DENIED) stops the measurement and is retried at the next tick, at most `MaxTriesPerDay` |
| `capability` store-grant probe (`storeGrantStatuses` / `storeGrantVerdict` / `Client.Permissions`) | cmd/felhom-agent/main.go, internal/proxmox/query.go | *"may the agent READ this backup tier?"*, one `capability.Status` per configured tier | R-185. **Never infer permission from an empty content listing** — `{"data":[]}` is what a FORBIDDEN tier and a NEWBORN tier both return, and that ambiguity hid an unreadable host tier on both demo boxes. Ask `/access/permissions` **as the agent's own token** (root always says yes). **The ungranted answer is not empty and not a 403** — it carries the privileges inherited from the box-wide `/` grant, so test for **`Datastore.AllocateSpace`** specifically; path-presence or `Datastore.Audit` reports a blinded storage healthy. Probed set comes from `BackupTiers()`, never a fixed list. Critical except the `local` fallback. Composes AROUND the sudo prober (the `poolReadStatus` precedent); `Status`'s wire shape is untouched so the hub alert is free. Unreachable PVE ⇒ degraded, never ok. |
| `backup.RestoreTestState` | internal/backup/restoretest_state.go | `RecordSuccess(target,archive,tier,verified,t)` / `ProvenArchive(target)` / `ProvenRestoreTests(ctx)` / `LastSuccess(target)` / `OldestFirst(targets)` | Per-tier restore-test PROOF state, persisted (atomic tmp+rename) — **which archive** was proven, and when (R-86) | **Credit ONLY on success** — a permanently failing tier must keep sorting first, or it looks freshly proven and stops being retried. Ties break on target id: without it, two tiers proven in the same second rotate by Go's randomised map order. **This one NEEDS persistence unlike R-84** — R-84 had ground truth to consult (the archive is still on the storage); a restore-test destroys its scratch and leaves no artifact. **R-86: the ARCHIVE is the state, the time is metadata** — a time alone cannot answer "have we proven THIS archive", which is the due-check's whole question. A pre-R-86 file (bare RFC3339 per target) keeps its time and yields NO proven archive, so each tier is due once after the upgrade; reading a legacy time as proof of the current archive would invent a guarantee. **R-189: it is also the REPORTABLE half of the restore-test signal.** The in-memory `backup.Store` holds only this process's latest run, and under per-archive due-ness the agent will not re-test a proven archive — so a proof lost to a restart is not repeated for a whole archive generation (observed live: a passing 14.5 GB offsite restore reached no host-report). `ProvenRestoreTests` renders the stored proofs as `hub.RestoreTest` entries and the collector merges them; a record missing the archive or the tier is NOT emitted, because an unproven tier reading as proven is worse than the defect. **Only successes are stored, deliberately:** a success suppresses future work, a failure causes it. |
| `backup.BackupSuccessState` | internal/backup/backup_state.go | `RecordBackupSuccess(target, b)` / `LastKnownSuccess(target, vmid)` | Newest SUCCESSFUL backup per tier+guest, persisted (atomic tmp+rename) — the due-check's fallback when the tier's storage cannot be read after a restart (R-894) | **Read ONLY when the storage cannot be read** — a storage that answers is the ground truth (R-84), and an archive absent there must make the tier due even when this file remembers one. A saved copy older than the cadence still reads due. Only successes are written. |
+72
View File
@@ -0,0 +1,72 @@
package main
import (
"go/ast"
"go/parser"
"go/token"
"testing"
"gitea.dooplex.hu/admin/felhom-agent/internal/linetest"
"gitea.dooplex.hu/admin/felhom-agent/internal/localapi"
)
// R-1250: SetLineTest takes a LineTestReader and finds the request half by a type assertion, so a Tester that lost
// WaitingForQuiet or RunNow would compile and silently answer 503 to every seeding box. This line makes it a build
// failure instead.
var _ localapi.LineTestRequester = (*linetest.Tester)(nil)
// R-1247: the weekly line test has the fstrim shape (component + reporter seam + goroutine) plus a local-API reader,
// so its wiring is asserted from the AST. It must share the ONE heavy-op gate (heavyOps) — or it could run beside a
// backup — and take its server from the tunnel manager, never a literal address.
func TestMainWiresTheLineTest(t *testing.T) {
fset := token.NewFileSet()
f, err := parser.ParseFile(fset, "main.go", nil, 0)
if err != nil {
t.Fatalf("parse main.go: %v", err)
}
var withGate, withTunnelTarget, reporterWired, localAPIWired, started bool
ast.Inspect(f, func(n ast.Node) bool {
switch node := n.(type) {
case *ast.CallExpr:
if fn, ok := node.Fun.(*ast.SelectorExpr); ok {
switch fn.Sel.Name {
case "New":
if pkg, ok := fn.X.(*ast.Ident); ok && pkg.Name == "linetest" && len(node.Args) >= 3 {
if id, ok := node.Args[2].(*ast.Ident); ok && id.Name == "heavyOps" {
withGate = true
}
if sel, ok := node.Args[1].(*ast.SelectorExpr); ok && sel.Sel.Name == "LineTestTarget" {
withTunnelTarget = true
}
}
case "SetLineTestReporter":
reporterWired = true
case "SetLineTest":
localAPIWired = true
}
}
case *ast.GoStmt:
if sel, ok := node.Call.Fun.(*ast.SelectorExpr); ok && sel.Sel.Name == "Run" {
if id, ok := sel.X.(*ast.Ident); ok && id.Name == "lineTest" {
started = true
}
}
}
return true
})
if !withGate {
t.Error("main.go never calls linetest.New(..., heavyOps, ...) — no line test, or one outside the heavy-op gate")
}
if !withTunnelTarget {
t.Error("linetest.New does not take its server from wgMgr.LineTestTarget")
}
if !reporterWired {
t.Error("main.go never calls collector.SetLineTestReporter — the line_test stanza never reaches the hub")
}
if !localAPIWired {
t.Error("main.go never calls localSrv.SetLineTest — the controller's GET /host/line-test always reads present:false")
}
if !started {
t.Error("main.go never starts the line test with `go lineTest.Run(ctx)`")
}
}
+19
View File
@@ -43,6 +43,7 @@ import (
"gitea.dooplex.hu/admin/felhom-agent/internal/guestnet"
"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
"gitea.dooplex.hu/admin/felhom-agent/internal/lanresolver"
"gitea.dooplex.hu/admin/felhom-agent/internal/linetest"
"gitea.dooplex.hu/admin/felhom-agent/internal/localapi"
applog "gitea.dooplex.hu/admin/felhom-agent/internal/log"
"gitea.dooplex.hu/admin/felhom-agent/internal/mgmtplane"
@@ -1641,6 +1642,24 @@ func runDaemon(cfg config.Config, logger *slog.Logger, logRing *applog.Ring) int
} else {
logger.Info("fstrim: weekly guest disk trim disabled by config (disk_trim.disable)")
}
// R-1247 (operator ruling 2026-10-24 07:26): the weekly upload line test — a 10-second iperf3 upload from this host
// to ep0's tunnel-only server, Tuesday 10:00-15:59 local, after 10 quiet minutes, under the one-heavy-op gate. Only
// with the offsite tunnel (the server is reachable nowhere else). Not part of the errc fan-out.
switch {
case !cfg.LineTest.Enabled():
logger.Info("linetest: weekly upload line test disabled by config (line_test.disable)")
case wgMgr == nil:
logger.Info("linetest: weekly upload line test off — no offsite tunnel on this host")
default:
lineTest := linetest.New(linetest.Client{Streams: linetest.Streams, Time: linetest.TestTime}, wgMgr.LineTestTarget,
heavyOps, linetest.SysUplinkTx("/"), cfg.Hub.HostID,
filepath.Join(cfg.OOB.WithDefaults().StateDir, "line-test.json"), logger)
collector.SetLineTestReporter(lineTest)
if localSrv != nil {
localSrv.SetLineTest(lineTest)
}
go lineTest.Run(ctx)
}
if lanLoop != nil {
lanServers = 1
go func() { errc <- lanLoop.Run(ctx) }()
+103 -16
View File
@@ -168,7 +168,10 @@ KERNEL_ACTIVE = ("staged", "oneshot", "judging", "reverting")
# agent-unit → User=felhom-agent; dropin → a [Unit] section (SF-3); nft → nft -c -f; plain → none;
# sshd → `sshd -t -f` on the snippet alone (R-982).
# policy: replace → always written when it differs; if-absent → only when the box has none (a setting the operator may
# have tuned); oob → only on a box with the OOB belt (/etc/felhom-sshd exists — the installer's --no-oob leaves none).
# have tuned); oob → only on a box with the OOB belt (/etc/felhom-sshd exists — the installer's --no-oob leaves none);
# appliance → only on a box whose ROOT-OWNED install record says appliance (D43 = A, operator 2026-10-24: a BYO
# host keeps its own settings, `11` §1). On a BYO box nothing is written and a copy WE wrote (this exact path, our
# first line) is removed; a box with no readable record is left exactly as it is (it cannot prove which it is).
# Order matters: the sudoers files come LAST, so a referenced wrapper is in place before the line that allows it.
# R-861 (agent v0.146.0): the signed agent update, verified here (mode agent_update), then the A/B flip.
SELFUPDATE_OP = "agent_update"
@@ -187,6 +190,8 @@ BUNDLE_RE = re.compile(r"^bundle-[0-9]+\.[0-9]+\.[0-9]+(-[0-9A-Za-z.]+)?\.json$"
OOB_DIR = "/etc/felhom-sshd"
# R-982: the host's own sshd drop-in (after_install reloads sshd for it; undo reloads after putting it back).
SSHD_DROPIN = "/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf"
# D43 = A (R-1257): the first line every copy of the drop-in we ship carries — what proves a file on a BYO host is ours.
SSHD_DROPIN_HEADER = b"# MANAGED BY FELHOM (the agent's config bundle) \xe2\x80\x94 R-982"
BUNDLE_FILES = [
("/usr/local/sbin/felhom-mkfs-guarded", "felhom-mkfs-guarded.sh", 0o755, "bash", "replace"),
@@ -227,7 +232,8 @@ BUNDLE_FILES = [
("/etc/systemd/system/felhom-oob-nft.service", "felhom-oob-nft.service", 0o644, "unit", "oob"),
# R-982, operator ruling D4 (2026-10-22): the host's own sshd (port 22, the LAN): root by key only. The console
# password stays; felhom-sshd (the operator's door) reads nothing from sshd_config.d. Reloaded in after_install.
("/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf", "felhom-root-ssh-key-only.conf", 0o644, "sshd", "replace"),
# D43 = A (operator 2026-10-24, R-1257): appliances only — a BYO host keeps its own SSH settings (`11` §1).
("/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf", "felhom-root-ssh-key-only.conf", 0o644, "sshd", "appliance"),
("/etc/sudoers.d/felhom-op", "felhom-op.sudoers", 0o440, "sudoers", "oob"),
("/etc/sudoers.d/felhom-agent", "felhom-agent.sudoers", 0o440, "sudoers", "replace"),
]
@@ -1873,6 +1879,7 @@ class Bundle:
self.a, self.r = apply, apply.r
self.report = apply.report
self.sshd_dropin_not_in_force = False
self.sshd_step_undone = None # X4 (R-1257): why the SSH step alone was put back ("" = not), the rest stays
# ---------- reading and checking ----------
def load_file(self, path):
@@ -1982,14 +1989,40 @@ class Bundle:
return b"", None, None # a symlink or a directory: always replaced by the real file
return self.r.read_bytes(dest), stat.S_IMODE(st.st_mode), st.st_uid
def install_mode(self):
"""The box's ROOT-OWNED install record's mode ("appliance", "byo", …), or None when it proves nothing (absent,
agent-writable, unreadable) — the same proof check_appliance asks for the host layer (R12)."""
try:
st = self.r.stat(INSTALL_STATE)
if st.st_uid != 0 or (st.st_mode & 0o022):
return None
mode = json.loads(self.r.read_file(INSTALL_STATE)).get("mode")
except (OSError, ValueError, AttributeError):
return None
return mode if isinstance(mode, str) and mode else None
def plan_writes(self, files):
"""Decide per file: write / same / kept (if-absent and present) / skipped (oob on a box without the belt)."""
"""Decide per file: write / same / kept (if-absent and present) / skipped (oob on a box without the belt; an
appliance-only file on a box that is not one) / remove (an appliance-only file WE wrote, on a BYO box)."""
oob = self.r.isdir(OOB_DIR)
install_mode = self.install_mode()
plan = []
for dest, content, mode, check, policy in files:
if policy == "oob" and not oob:
plan.append((dest, content, mode, "skipped", None, None))
continue
if policy == "appliance" and install_mode != "appliance":
old, old_mode, _ = self.live(dest)
ours = old is not None and old.startswith(SSHD_DROPIN_HEADER)
if install_mode == "byo" and ours:
self.r.log(f"os-apply: BUNDLE {dest}: a BYO host keeps its own settings (D43) — removing the copy we wrote")
plan.append((dest, content, mode, "remove", old, old_mode))
else:
why = "a BYO host keeps its own settings (D43)" if install_mode == "byo" else \
f"the install record proves no appliance (mode={install_mode!r}) — left as it is"
self.r.log(f"os-apply: BUNDLE {dest} skipped: {why}")
plan.append((dest, content, mode, "skipped", None, None))
continue
old, old_mode, old_uid = self.live(dest)
if policy == "if-absent" and old is not None:
plan.append((dest, content, mode, "kept", old, old_mode))
@@ -2031,7 +2064,7 @@ class Bundle:
def install(self, ver, sha, files, authority, bootstrap):
log = self.r.log
plan = self.plan_writes(files)
counts = {k: sum(1 for p in plan if p[3] == k) for k in ("write", "same", "kept", "skipped")}
counts = {k: sum(1 for p in plan if p[3] == k) for k in ("write", "same", "kept", "skipped", "remove")}
log(f"os-apply: BUNDLE START agent={ver} sha={sha[:16]} authority={authority} files={len(plan)} "
f"write={counts['write']} same={counts['same']} kept={counts['kept']} skipped={counts['skipped']}")
stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime(self.r.now()))
@@ -2039,10 +2072,17 @@ class Bundle:
done = [] # (dest, old bytes or None, old mode)
rep = {"agent_version": ver, "sha256": sha, "authority": authority, "prev_dir": prev,
"written": [p[0] for p in plan if p[3] == "write"], "kept": [p[0] for p in plan if p[3] == "kept"],
"skipped": [p[0] for p in plan if p[3] == "skipped"], "same": counts["same"]}
"skipped": [p[0] for p in plan if p[3] == "skipped"], "same": counts["same"],
"removed": [p[0] for p in plan if p[3] == "remove"]}
self.report["bundle"] = rep
try:
for dest, content, mode, action, old, old_mode in plan:
if action == "remove": # D43 = A: our copy off a BYO host (the previous copy kept, the undo restores it)
self.r.put_file(prev + dest, old, 0o600)
done.append((dest, old, old_mode))
self.r.remove(dest)
log(f"os-apply: BUNDLE REMOVED {dest} (ours, on a BYO host — D43)")
continue
if action != "write":
continue
if old is not None:
@@ -2062,22 +2102,56 @@ class Bundle:
rep["signers_created"] = False
if self.sshd_dropin_not_in_force:
rep["sshd_dropin_not_in_force"] = True # R-982 A2: written and reloaded, but sshd -T shows root passwords allowed
if self.sshd_step_undone:
rep["sshd_step_undone"] = self.sshd_step_undone[:300] # X4: only the SSH step was put back
if bootstrap and not self.r.lexists(TRUST_SIGNERS):
self.r.put_file(TRUST_SIGNERS, pinned_signers_line().encode(), 0o644)
rep["signers_created"] = True
log(f"os-apply: BUNDLE created {TRUST_SIGNERS} with exactly the pinned key {PINNED_OPERATOR_KEY_ID}")
record = {"format": BUNDLE_FORMAT, "agent_version": ver, "bundle_sha256": sha, "authority": authority,
"installed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(self.r.now())),
"files": {p[0]: (sha256_hex(p[1]) if p[3] in ("write", "same") else
(sha256_hex(p[4]) if p[3] == "kept" else "skipped")) for p in plan}}
"files": {p[0]: self.record_entry(p) for p in plan}}
self.r.put_file(BUNDLE_RECORD, (json.dumps(record, indent=2, sort_keys=True) + "\n").encode(), 0o644)
self.prune()
log(f"os-apply: BUNDLE DONE agent={ver} written={counts['write']} same={counts['same']} "
f"self-check=ok signers-created={rep['signers_created']}")
return 0
def record_entry(self, p):
"""The record's word for one plan entry: the sha it installed, the sha it kept, or what it did instead."""
dest, content, _, action, old, _ = p
if dest == SSHD_DROPIN and self.sshd_step_undone:
return sha256_hex(old) if old is not None else "absent" # X4: the SSH step was put back
if action in ("write", "same"):
return sha256_hex(content)
if action == "kept":
return sha256_hex(old)
return "removed" if action == "remove" else "skipped"
def lan_addresses(self):
"""X3 (R-1257): the host's own global IPv4 addresses — the ones a LAN client reaches sshd on."""
rc, out, _ = self.r.host(["ip", "-4", "-o", "addr", "show", "scope", "global"], 30)
if rc != 0:
return []
return sorted(set(re.findall(r"\binet (\d+\.\d+\.\d+\.\d+)/", out)))
def undo_sshd_step(self, plan, why):
"""X4 (R-1257): `sshd -t` refused the host's whole config with our step in it — put back ONLY that file (the
previous copy, or none) and keep the rest of the bundle. sshd was not reloaded, so it never ran the refused state."""
for dest, _, _, action, old, old_mode in plan:
if dest != SSHD_DROPIN or action not in ("write", "remove"):
continue
if old is None:
self.r.remove(dest)
else:
self.r.put_file(dest, old, old_mode if old_mode is not None else 0o644)
self.sshd_step_undone = why
self.r.log(f"os-apply: BUNDLE WARNING sshd -t refused the host's config with the SSH step — ONLY that step is put "
f"back, the rest of the bundle stays: {why[:200]} (R-1257)")
def after_install(self, plan):
written = {p[0] for p in plan if p[3] == "write"}
removed = {p[0] for p in plan if p[3] == "remove"}
dests = {p[0] for p in plan if p[3] != "skipped"}
if any(d.startswith("/etc/systemd/system/") for d in written):
self.must(["systemctl", "daemon-reload"], "systemctl daemon-reload")
@@ -2098,16 +2172,29 @@ class Bundle:
# felhom-sshd is not touched. Then the POSITIVE check: sshd's own effective view for root. A box whose
# sshd_config has no Include of sshd_config.d, an earlier drop-in or a Match block can keep the password route
# — that is said loudly in the log and the report, never silently recorded as done (A2).
if SSHD_DROPIN in written:
self.must(["/usr/sbin/sshd", "-t"], "sshd -t (the host's whole sshd config)")
# R-1257 X4: a failing `sshd -t` puts back ONLY the SSH step (undo_sshd_step); it used to refuse and undo the
# whole bundle. X3: the in-force check asks sshd about each of the host's own LAN addresses — a probe from a
# documentation address (192.0.2.1) read a `Match Address <LAN>` block that re-allows passwords as in force.
if SSHD_DROPIN in written or SSHD_DROPIN in removed:
rc, out, err = self.r.host(["/usr/sbin/sshd", "-t"], 120)
if rc != 0:
self.undo_sshd_step(plan, f"sshd -t failed (rc={rc}): {(out + err).strip()[-200:]}")
return
self.must(["systemctl", "try-reload-or-restart", "ssh"], "systemctl try-reload-or-restart ssh")
rc, out, err = self.r.host(["/usr/sbin/sshd", "-T", "-C", "user=root,host=felhom-check,addr=192.0.2.1"], 60)
eff = re.findall(r"^permitrootlogin\s+(\S+)", out, re.M | re.I)
if rc != 0 or eff != ["without-password"]:
self.r.log(f"os-apply: BUNDLE WARNING the root-key-only drop-in is NOT in force: sshd -T says "
f"permitrootlogin={eff or '?'} (rc={rc}) — check sshd_config's Include, an earlier drop-in "
f"or a Match block (R-982)")
self.sshd_dropin_not_in_force = True
if SSHD_DROPIN in written:
addrs = self.lan_addresses()
if not addrs:
self.r.log("os-apply: BUNDLE WARNING no global IPv4 address read for the sshd in-force check — "
"probing 192.0.2.1 only (R-1257)")
addrs = ["192.0.2.1"]
for addr in addrs:
rc, out, err = self.r.host(["/usr/sbin/sshd", "-T", "-C", f"user=root,host=felhom-check,addr={addr}"], 60)
eff = re.findall(r"^permitrootlogin\s+(\S+)", out, re.M | re.I)
if rc != 0 or eff != ["without-password"]:
self.r.log(f"os-apply: BUNDLE WARNING the root-key-only drop-in is NOT in force for {addr}: sshd -T "
f"says permitrootlogin={eff or '?'} (rc={rc}) — check sshd_config's Include, an earlier "
f"drop-in or a Match block (R-982, R-1257)")
self.sshd_dropin_not_in_force = True
def must(self, argv, what):
rc, out, err = self.r.host(argv, 120)
+113 -1
View File
@@ -45,8 +45,10 @@ class St:
class Box:
"""An in-memory host. files: path -> bytes; modes/uids per path; dirs: a set."""
def __init__(self, bundle_bytes, signed, oob=False, signers=True):
def __init__(self, bundle_bytes, signed, oob=False, signers=True, install_mode="appliance"):
self.files, self.modes, self.uids = {}, {}, {}
if install_mode is not None: # the installer's ROOT-OWNED record (D43 = A reads it for the sshd drop-in)
self.put(osapply.INSTALL_STATE, json.dumps({"completed": [], "mode": install_mode}).encode(), 0o644)
self.dirs = {osapply.OOB_DIR} if oob else set()
self.put(osapply.TRUST_FILE, json.dumps({"host_id": HOST, "ring0_slow_lane": False}).encode(), 0o644)
if signers:
@@ -513,6 +515,116 @@ class RootKeyOnly(unittest.TestCase):
reloads = [i for i, c in enumerate(box.calls) if c == ["systemctl", "try-reload-or-restart", "ssh"]]
self.assertEqual(len(reloads), 2, box.calls) # after_install, then the undo
class ByoKeepsItsSSH(unittest.TestCase):
"""D43 = A (operator 2026-10-24, R-1257): the root-key-only drop-in goes ONLY to a box whose root-owned install record
says appliance; on a BYO host nothing is written and a copy WE wrote is removed; a box with no record is left as it
is. X3: the in-force check asks sshd about each of the host's own LAN addresses. X4: a failing `sshd -t` puts back
only the SSH step. RED-PROOF (2026-10-24, felhom-agent f5c3d77 + these tests): with the drop-in's policy back to
`replace`, test_a_byo_host_gets_no_dropin and test_our_copy_comes_off_a_byo_host fail; with the probe back to
192.0.2.1 only, test_a_match_block_for_the_lan_is_not_in_force fails; with `sshd -t` back to `must`,
test_a_failing_sshd_t_undoes_only_the_ssh_step fails (rc 3, the whole bundle undone)."""
DEST = "/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf"
OURS = (REPO / "configs" / "felhom-root-ssh-key-only.conf").read_bytes()
def test_a_byo_host_gets_no_dropin(self):
data, sha = real_bundle()
box = Box(data, signed_job(sha), install_mode="byo")
rc, rep = run(box)
self.assertEqual(rc, 0, rep)
self.assertNotIn(self.DEST, box.files)
self.assertNotIn(["/usr/sbin/sshd", "-t"], box.calls)
self.assertNotIn(["systemctl", "try-reload-or-restart", "ssh"], box.calls)
self.assertIn(self.DEST, rep["bundle"]["skipped"])
rec = json.loads(box.files[osapply.BUNDLE_RECORD])
self.assertEqual(rec["files"][self.DEST], "skipped")
def test_our_copy_comes_off_a_byo_host(self):
data, sha = real_bundle()
box = Box(data, signed_job(sha), install_mode="byo")
box.put(self.DEST, self.OURS, 0o644) # an earlier bundle wrote it (before D43)
rc, rep = run(box)
self.assertEqual(rc, 0, rep)
self.assertNotIn(self.DEST, box.files)
self.assertEqual(rep["bundle"]["removed"], [self.DEST])
self.assertTrue(any(k.endswith(self.DEST) and k.startswith(osapply.BUNDLE_PREV_DIR) for k in box.files),
"the previous copy must be kept for the undo")
self.assertLess(box.calls.index(["/usr/sbin/sshd", "-t"]), box.calls.index(["systemctl", "try-reload-or-restart", "ssh"]))
self.assertEqual(json.loads(box.files[osapply.BUNDLE_RECORD])["files"][self.DEST], "removed")
def test_the_owners_own_file_stays_on_a_byo_host(self):
data, sha = real_bundle()
box = Box(data, signed_job(sha), install_mode="byo")
box.put(self.DEST, b"# my own\nPermitRootLogin yes\n", 0o644)
rc, rep = run(box)
self.assertEqual(rc, 0, rep)
self.assertEqual(box.files[self.DEST], b"# my own\nPermitRootLogin yes\n")
self.assertNotIn(["systemctl", "try-reload-or-restart", "ssh"], box.calls)
def test_a_box_with_no_record_is_left_as_it_is(self):
for setup in ("none", "agent-writable"):
data, sha = real_bundle()
box = Box(data, signed_job(sha), install_mode=None if setup == "none" else "byo")
if setup == "agent-writable":
box.modes[osapply.INSTALL_STATE] = 0o666 # proves nothing
box.put(self.DEST, self.OURS, 0o644)
rc, rep = run(box)
self.assertEqual(rc, 0, rep)
self.assertEqual(box.files.get(self.DEST), self.OURS, setup)
self.assertIn(self.DEST, rep["bundle"]["skipped"], setup)
def _fake(self, box, addrs=("192.168.0.50",), match=None, sshd_t_rc=0):
orig = box.host
def host(argv, timeout=600, stdin=None):
if argv[:3] == ["ip", "-4", "-o"]:
box.calls.append(argv)
return 0, "".join(f"2: vmbr0 inet {a}/24 brd x scope global vmbr0\\ valid_lft forever\n" for a in addrs), ""
if argv[:2] == ["/usr/sbin/sshd", "-T"]:
box.calls.append(argv)
val = "yes" if match and f"addr={match}" in argv[-1] else "without-password"
return 0, f"port 22\npermitrootlogin {val}\n", ""
if argv == ["/usr/sbin/sshd", "-t"]:
box.calls.append(argv)
return (sshd_t_rc, "", "/etc/ssh/sshd_config line 9: Bad configuration option") if sshd_t_rc else (0, "", "")
return orig(argv, timeout, stdin)
box.host = host
def test_a_match_block_for_the_lan_is_not_in_force(self):
data, sha = real_bundle()
box = Box(data, signed_job(sha))
self._fake(box, addrs=("192.168.0.50", "10.77.0.4"), match="192.168.0.50")
rc, rep = run(box)
self.assertEqual(rc, 0, rep)
probed = [c[-1] for c in box.calls if c[:2] == ["/usr/sbin/sshd", "-T"]]
self.assertTrue(any("addr=192.168.0.50" in p for p in probed) and any("addr=10.77.0.4" in p for p in probed), probed)
self.assertFalse(any("addr=192.0.2.1" in p for p in probed), probed)
self.assertTrue(rep["bundle"].get("sshd_dropin_not_in_force"), rep)
self.assertTrue(any("NOT in force for 192.168.0.50" in l for l in box.logs), box.logs)
def test_the_lan_check_passes_when_every_address_is_key_only(self):
data, sha = real_bundle()
box = Box(data, signed_job(sha))
self._fake(box, addrs=("192.168.0.50", "10.77.0.4"))
rc, rep = run(box)
self.assertEqual(rc, 0, rep)
self.assertNotIn("sshd_dropin_not_in_force", rep["bundle"])
def test_a_failing_sshd_t_undoes_only_the_ssh_step(self):
for before in (None, b"# MANAGED BY FELHOM (the agent's config bundle) \xe2\x80\x94 R-982 old\nPermitRootLogin prohibit-password\n"):
data, sha = real_bundle()
box = Box(data, signed_job(sha))
if before is not None:
box.put(self.DEST, before, 0o644)
self._fake(box, sshd_t_rc=255)
rc, rep = run(box)
self.assertEqual(rc, 0, rep) # the bundle stays
self.assertEqual(box.files.get(self.DEST), before, "only the SSH step is put back")
self.assertIn("/usr/local/sbin/felhom-os-apply", box.files, "the rest of the bundle stays")
self.assertIn("sshd -t failed", rep["bundle"].get("sshd_step_undone", ""), rep)
self.assertNotIn(["systemctl", "try-reload-or-restart", "ssh"], box.calls, "sshd never ran the refused state")
want = "absent" if before is None else hashlib.sha256(before).hexdigest()
self.assertEqual(json.loads(box.files[osapply.BUNDLE_RECORD])["files"][self.DEST], want)
class TrustBootstrap(unittest.TestCase):
def test_missing_signers_verifies_against_the_pinned_key_and_creates_it(self):
data, sha = real_bundle()
+11
View File
@@ -34,6 +34,7 @@ type Config struct {
WGTunnel WGTunnelConfig `json:"wg_tunnel"`
GuestNet GuestNetConfig `json:"guest_net"`
DiskTrim DiskTrimConfig `json:"disk_trim"`
LineTest LineTestConfig `json:"line_test"`
OOB OOBConfig `json:"oob"`
SelfUpdate SelfUpdateConfig `json:"selfupdate"`
LogLevel string `json:"log_level"` // debug|info|warn|error (default info)
@@ -150,6 +151,16 @@ type DiskTrimConfig struct {
// Enabled reports whether the weekly trim should run.
func (d DiskTrimConfig) Enabled() bool { return !d.Disable }
// LineTestConfig configures the R-1247 weekly upload line test (internal/linetest). DEFAULT-ON: it reaches only ep0's
// tunnel-only iperf3 server, so it runs only on a box that already has an operational offsite tunnel (operator
// ruling 2026-10-24 07:26: every box with a tunnel measures its line weekly).
type LineTestConfig struct {
Disable bool `json:"disable"`
}
// Enabled reports whether the weekly line test should run.
func (l LineTestConfig) Enabled() bool { return !l.Disable }
// GuestNetConfig configures the R-54 guest-network watchdog (internal/guestnet).
//
// **This is the repo's first DEFAULT-ON feature gate, and the inversion is deliberate.** Every other
+17
View File
@@ -104,6 +104,12 @@ type GuestDiskTrimReporter interface {
GuestDiskTrimStatus(ctx context.Context) *GuestDiskTrimStatus
}
// LineTestReporter is the R-1247 seam the weekly line test plugs into (hub does not import linetest). nil → no
// line_test stanza.
type LineTestReporter interface {
LineTestStatus(ctx context.Context) *LineTestStatus
}
// Collector builds a HostReport from read-only sources. All deps are behind narrow
// interfaces for unit testing.
type Collector struct {
@@ -124,6 +130,7 @@ type Collector struct {
ctrlSup ControllerSupervisorReporter // R-523: in-guest controller supervisor (nil → stanza omitted)
guestNet GuestNetReporter // R-54: per-guest network watchdog (nil → stanza omitted)
diskTrim GuestDiskTrimReporter // R-444: weekly guest disk trim (nil → stanza omitted)
lineTest LineTestReporter // R-1247: weekly upload line test (nil → stanza omitted)
deliberateOff func() *DeliberateOff // R-998: household power-off record of this boot (nil → omitted)
foreignKey ForeignKeyArchiveReporter // R-366 slice 2 (nil → omitted)
selfUpdate SelfUpdateReporter // D1: agent self-update pending status (nil → false)
@@ -259,6 +266,12 @@ func (c *Collector) SetGuestDiskTrimReporter(r GuestDiskTrimReporter) *Collector
return c
}
// SetLineTestReporter wires the R-1247 weekly line test as a report source (nil-safe → stanza omitted).
func (c *Collector) SetLineTestReporter(r LineTestReporter) *Collector {
c.lineTest = r
return c
}
// SetDeliberateOffSource wires the R-998 power-off record (poweroff.State.Current, adapted).
func (c *Collector) SetDeliberateOffSource(f func() *DeliberateOff) *Collector {
c.deliberateOff = f
@@ -427,6 +440,10 @@ func (c *Collector) Collect(ctx context.Context) (*HostReport, error) {
if c.diskTrim != nil {
report.GuestDiskTrim = c.diskTrim.GuestDiskTrimStatus(ctx)
}
// R-1247: the last weekly line test (nil reporter = not wired → stanza omitted).
if c.lineTest != nil {
report.LineTest = c.lineTest.LineTestStatus(ctx)
}
// R-366 slice 2: archives the restore-test skipped as another key's (nil → not evaluated yet → omitted).
if c.foreignKey != nil {
report.ForeignKeyArchives = c.foreignKey.ForeignKeyArchives(ctx)
+18
View File
@@ -129,6 +129,11 @@ type HostReport struct {
// wired; an empty `guests` list means the job runs and no guest has been trimmed yet. No secret.
GuestDiskTrim *GuestDiskTrimStatus `json:"guest_disk_trim,omitempty"`
// LineTest is the weekly upload line test stanza (R-1247; operator ruling 2026-10-24): the schedule, the last
// measured upload line (best of the runs, Mbit/s, as ep0's iperf3 server counted it) and the last attempt. Present
// only when the job is wired. No secret.
LineTest *LineTestStatus `json:"line_test,omitempty"`
// ForeignKeyArchives (R-366 slice 2, `09` §3 decision 168): per backup tier, the whole-guest archives the
// restore-test SKIPPED because they were written with another key (an earlier install of this box). This box
// cannot open them; the hub turns a CHANGE of this list into one operator event. Absent = not evaluated yet since
@@ -232,6 +237,19 @@ type GuestDiskTrimStatus struct {
Guests []GuestDiskTrim `json:"guests,omitempty"`
}
// LineTestStatus is the R-1247 weekly upload line test stanza. `last_ok_at` + `mbps` are the last SUCCESSFUL
// measurement ("" / 0 = never); `last_attempt_at` + `last_error` the last attempt, whatever its verdict — never read
// the attempt time as a result.
type LineTestStatus struct {
Schedule string `json:"schedule"`
LastOKAt string `json:"last_ok_at,omitempty"`
Mbps float64 `json:"mbps,omitempty"`
RunsMbps []float64 `json:"runs_mbps,omitempty"`
Server string `json:"server,omitempty"`
LastAttemptAt string `json:"last_attempt_at,omitempty"`
LastError string `json:"last_error,omitempty"`
}
// GuestDiskTrim is one guest's LAST trim attempt. `ok` with `last_attempt_at` is the verdict of that attempt — never
// read the time alone as success; `last_ok_at` is the last attempt that succeeded ("" = never). `bytes_trimmed` is
// the sum of the "(N bytes) trimmed" lines `pct fstrim` printed, over `mounts` mount points.
+294
View File
@@ -0,0 +1,294 @@
package linetest
import (
"context"
"crypto/rand"
"encoding/binary"
"encoding/json"
"errors"
"fmt"
"io"
"net"
"strconv"
"sync"
"sync/atomic"
"time"
)
// A minimal iperf3 CLIENT for one TCP upload test (client → server), speaking the iperf3 control protocol (3.x) to the
// stock `iperf3 --server` on ep0 (operator ruling 2026-10-24: the agent speaks iperf3 itself, nothing is installed on
// the host). The number that counts is the SERVER's: the bytes it received, from its results JSON — bytes the client
// handed to its socket include what still sat in the send buffer when the timer ended.
//
// The protocol, as the server drives it (one state byte at a time on the control connection):
//
// client → 37-byte cookie
// server → PARAM_EXCHANGE(9) client → params JSON (4-byte big-endian length + bytes)
// server → CREATE_STREAMS(10) client opens N data connections, each starts with the same cookie
// server → TEST_START(1), TEST_RUNNING(2) client sends for the test time, then → TEST_END(4)
// server → EXCHANGE_RESULTS(13) client → its results JSON; server → its results JSON
// server → DISPLAY_RESULTS(14) client → IPERF_DONE(16)
//
// A server already running a test answers ACCESS_DENIED(-1) and closes: that is ErrBusy (another box is testing; try
// again later the same day). SERVER_ERROR(-3) / SERVER_TERMINATE(-2) end the test with an error.
const (
stParamExchange = 9
stCreateStreams = 10
stTestStart = 1
stTestRunning = 2
stTestEnd = 4
stExchangeResults = 13
stDisplayResults = 14
stIperfDone = 16
stAccessDenied = -1
stServerTerminate = -2
stServerError = -3
cookieSize = 37 // 36 characters + NUL
blockSize = 128 * 1024
maxJSONBytes = 1 << 20
)
// ErrBusy: the server is running another client's test.
var ErrBusy = errors.New("iperf3 server busy")
// Result is one upload test.
type Result struct {
Mbps float64 // the server's received bits per second / 1e6
ServerBytes int64 // bytes the server received, all streams
ClientBytes int64 // bytes the client wrote (diagnostics only)
ServerSeconds float64 // the server's measured duration
}
// Dialer opens a TCP connection (net.Dialer in production; a seam for tests).
type Dialer interface {
DialContext(ctx context.Context, network, address string) (net.Conn, error)
}
// Client runs one test. The whole run is bounded by ctx; Run sets every connection's deadline from it.
type Client struct {
Dialer Dialer
Streams int // parallel streams (-P)
Time time.Duration // test time (-t)
}
func (c Client) dialer() Dialer {
if c.Dialer != nil {
return c.Dialer
}
return &net.Dialer{Timeout: 5 * time.Second}
}
func newCookie() ([]byte, error) {
const alphabet = "abcdefghijklmnopqrstuvwxyz234567"
b := make([]byte, cookieSize)
if _, err := rand.Read(b[:cookieSize-1]); err != nil {
return nil, err
}
for i := 0; i < cookieSize-1; i++ {
b[i] = alphabet[int(b[i])%len(alphabet)]
}
b[cookieSize-1] = 0
return b, nil
}
func writeJSON(w io.Writer, v any) error {
data, err := json.Marshal(v)
if err != nil {
return err
}
var hdr [4]byte
binary.BigEndian.PutUint32(hdr[:], uint32(len(data)))
if _, err := w.Write(append(hdr[:], data...)); err != nil {
return err
}
return nil
}
func readJSON(r io.Reader, v any) error {
var hdr [4]byte
if _, err := io.ReadFull(r, hdr[:]); err != nil {
return err
}
n := binary.BigEndian.Uint32(hdr[:])
if n == 0 || n > maxJSONBytes {
return fmt.Errorf("iperf3: results JSON of %d bytes refused", n)
}
buf := make([]byte, n)
if _, err := io.ReadFull(r, buf); err != nil {
return err
}
return json.Unmarshal(buf, v)
}
type streamResult struct {
ID int `json:"id"`
Bytes int64 `json:"bytes"`
Retransmits int64 `json:"retransmits"`
Jitter float64 `json:"jitter"`
Errors int64 `json:"errors"`
OmittedErrors int64 `json:"omitted_errors"`
Packets int64 `json:"packets"`
OmittedPackets int64 `json:"omitted_packets"`
StartTime float64 `json:"start_time"`
EndTime float64 `json:"end_time"`
}
type results struct {
CPUTotal float64 `json:"cpu_util_total"`
CPUUser float64 `json:"cpu_util_user"`
CPUSystem float64 `json:"cpu_util_system"`
SenderHasRetransmits int `json:"sender_has_retransmits"`
CongestionUsed string `json:"congestion_used,omitempty"`
Streams []streamResult `json:"streams"`
}
// streamIDs are the ids iperf3 gives its streams: the first is 1, every later one is (streams so far) + 2 — so 1, 3,
// 4, 5 … (iperf_add_stream). The server matches our results by these ids.
func streamIDs(n int) []int {
ids := make([]int, n)
for i := range ids {
if i == 0 {
ids[i] = 1
} else {
ids[i] = i + 2
}
}
return ids
}
// Run performs one upload test to addr ("10.77.0.1:5201").
func (c Client) Run(ctx context.Context, addr string) (Result, error) {
var res Result
streams, dur := c.Streams, c.Time
if streams < 1 {
streams = 1
}
if dur <= 0 {
dur = 10 * time.Second
}
deadline, ok := ctx.Deadline()
if !ok {
return res, errors.New("iperf3: Run needs a context with a deadline")
}
cookie, err := newCookie()
if err != nil {
return res, err
}
ctrl, err := c.dialer().DialContext(ctx, "tcp", addr)
if err != nil {
return res, fmt.Errorf("iperf3: connect: %w", err)
}
defer ctrl.Close()
ctrl.SetDeadline(deadline)
if _, err := ctrl.Write(cookie); err != nil {
return res, fmt.Errorf("iperf3: cookie: %w", err)
}
var data []net.Conn
defer func() {
for _, d := range data {
d.Close()
}
}()
var sent atomic.Int64
var server results
var wg sync.WaitGroup
for {
var st [1]byte
if _, err := io.ReadFull(ctrl, st[:]); err != nil {
return res, fmt.Errorf("iperf3: control read: %w", err)
}
switch int8(st[0]) {
case stParamExchange:
params := map[string]any{"tcp": true, "omit": 0, "time": int(dur / time.Second), "num": 0, "blockcount": 0,
"parallel": streams, "len": blockSize, "pacing_timer": 1000, "client_version": "3.18"}
if err := writeJSON(ctrl, params); err != nil {
return res, fmt.Errorf("iperf3: params: %w", err)
}
case stCreateStreams:
for i := 0; i < streams; i++ {
d, err := c.dialer().DialContext(ctx, "tcp", addr)
if err != nil {
return res, fmt.Errorf("iperf3: data stream %d: %w", i+1, err)
}
d.SetDeadline(deadline)
data = append(data, d)
if _, err := d.Write(cookie); err != nil {
return res, fmt.Errorf("iperf3: data stream %d cookie: %w", i+1, err)
}
}
case stTestStart:
case stTestRunning:
stopAt := time.Now().Add(dur)
if stopAt.After(deadline) {
return res, errors.New("iperf3: no time left for the test inside the deadline")
}
buf := make([]byte, blockSize)
for _, d := range data {
d.SetWriteDeadline(stopAt)
wg.Add(1)
go func(d net.Conn) {
defer wg.Done()
for {
n, err := d.Write(buf)
sent.Add(int64(n))
if err != nil {
return
}
}
}(d)
}
wg.Wait()
if _, err := ctrl.Write([]byte{stTestEnd}); err != nil {
return res, fmt.Errorf("iperf3: test end: %w", err)
}
case stExchangeResults:
ids := streamIDs(len(data))
mine := results{SenderHasRetransmits: -1, CongestionUsed: "cubic"}
per := sent.Load() / int64(max(len(data), 1))
for _, id := range ids {
mine.Streams = append(mine.Streams, streamResult{ID: id, Bytes: per, Retransmits: -1, EndTime: dur.Seconds()})
}
if err := writeJSON(ctrl, mine); err != nil {
return res, fmt.Errorf("iperf3: send results: %w", err)
}
if err := readJSON(ctrl, &server); err != nil {
return res, fmt.Errorf("iperf3: read server results: %w", err)
}
case stDisplayResults:
ctrl.Write([]byte{stIperfDone})
return summarize(server, sent.Load(), dur)
case stAccessDenied:
return res, ErrBusy
case stServerTerminate, stServerError:
var codes [8]byte
io.ReadFull(ctrl, codes[:])
return res, fmt.Errorf("iperf3: server error (state %d, iperf errno %d, errno %d)", int8(st[0]),
int32(binary.BigEndian.Uint32(codes[:4])), int32(binary.BigEndian.Uint32(codes[4:])))
default:
return res, fmt.Errorf("iperf3: unexpected state %s", strconv.Itoa(int(int8(st[0]))))
}
}
}
// summarize turns the server's results into a Result. No stream bytes is an error, never a zero line.
func summarize(server results, clientBytes int64, dur time.Duration) (Result, error) {
res := Result{ClientBytes: clientBytes}
for _, s := range server.Streams {
res.ServerBytes += s.Bytes
if s.EndTime > res.ServerSeconds {
res.ServerSeconds = s.EndTime
}
}
if res.ServerBytes <= 0 {
return res, errors.New("iperf3: the server reported no received bytes")
}
secs := res.ServerSeconds
if secs <= 0 {
secs = dur.Seconds()
}
res.Mbps = float64(res.ServerBytes) * 8 / secs / 1e6
return res, nil
}
+541
View File
@@ -0,0 +1,541 @@
// Package linetest is the weekly upload line-speed test (R-1247; operator ruling 2026-10-24 07:26).
//
// Why: the controller's upload watch and qBittorrent's default limit (50 % of the line, D41) need the household's
// upload line. The passive reading — the nightly off-site push — is only a low floor (2026-10-24 night: demo-hp „at
// least 10 Mbit/s"; this test read 90.7 on the same line). So once a week the HOST (the tunnel lives on the host)
// runs a 10-second iperf3 upload to ep0's tunnel-only iperf3 server (felhom.eu scripts/ep0-iperf3). No file is left
// anywhere and nothing counts against the household's storage.
//
// The rule, each part pinned by a test in linetest_test.go:
// - Weekly: due from Tuesday 10:00 local until one measurement succeeded since then (a box that was off catches up
// on a later day, inside the window).
// - Daytime only: a test starts only 10:00–15:59 local — never in the night chain's window (01:00–06:59), and each
// host starts from its own minute (SlotMinute, spread over 10:00–13:59) so the fleet does not test at once.
// - Never beside a backup, a restore-test, an OS update or a trim: the test holds the host-wide one-heavy-op gate.
// - Only after QuietMinutes of quiet uplink: every one-minute transmit rate of the host's physical uplink below
// QuietMbps — so neither the family's upload nor a restore is measured against itself.
// - Up to RunsPerMeasurement runs of TestTime each; the highest counts. A run is bounded by RunTimeout.
// - „Server busy" (another box is testing) → tried again at the next tick the same day; at most MaxTriesPerDay.
// - No tunnel (no operational wg-felhom with a server address) → no test and no value; the controller then keeps its
// passive floor and sets no qBittorrent limit.
// - The last result is persisted, so a restart neither loses it nor re-tests.
// - A box whose torrent app seeds all day is never quiet for 10 minutes (R-1250). For it the controller holds the
// app's upload limit very low and asks for a test on request (RunNow): every rule above except the 10 quiet
// minutes still holds, the quiet is checked over RequestQuietWindow instead, and the runs are RequestRunGap
// apart, so the whole request ends in about 55 s.
package linetest
import (
"context"
"encoding/json"
"errors"
"fmt"
"hash/fnv"
"log/slog"
"net/netip"
"os"
"path/filepath"
"strconv"
"strings"
"sync"
"time"
"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
)
const (
Weekday = time.Tuesday
StartHour = 10 // first eligible local hour (inclusive)
EndHour = 16 // first NOT-eligible local hour: the last start is 15:59
SlotSpreadMinutes = 240
QuietMinutes = 10
QuietMbps = 2.0
RunsPerMeasurement = 3
TestTime = 10 * time.Second
Streams = 4
RunTimeout = 25 * time.Second
RunGap = 30 * time.Second
MaxTriesPerDay = 6
ServerPort = 5201
TickInterval = time.Minute
FirstTickDelay = 5 * time.Minute
GateName = "line-test"
// A test on request (R-1250): the controller holds the app's limit while it waits, so it is short.
RequestSettle = 5 * time.Second // the held app's last packets leave before the quiet is read
RequestQuietWindow = 10 * time.Second // the uplink must send below QuietMbps over this window
RequestRunGap = 5 * time.Second
)
// ScheduleText is carried on the host report.
const ScheduleText = "weekly, due Tuesday from 10:00 host-local time; starts 10:00-15:59 after 10 quiet minutes; never beside a backup, restore-test or update"
// Gate is the host-wide one-heavy-operation gate (*backup.InFlight).
type Gate interface {
TryAcquire(what string) (release func(), busy string, ok bool)
}
// Target answers the iperf3 server's tunnel address, ok=false when the tunnel is not operational.
type Target func() (netip.Addr, bool)
// Runner runs one upload test (Client in production).
type Runner interface {
Run(ctx context.Context, addr string) (Result, error)
}
// UplinkTx reads the host's physical-uplink transmit byte counter.
type UplinkTx func() (uint64, error)
// State is what is persisted (line-test.json).
type State struct {
LastOKAt time.Time `json:"last_ok_at,omitempty"`
Mbps float64 `json:"mbps,omitempty"`
RunsMbps []float64 `json:"runs_mbps,omitempty"`
Server string `json:"server,omitempty"`
LastAttemptAt time.Time `json:"last_attempt_at,omitempty"`
LastError string `json:"last_error,omitempty"`
TriesDay string `json:"tries_day,omitempty"` // local date of TriesToday
TriesToday int `json:"tries_today,omitempty"`
}
type txSample struct {
at time.Time
tx uint64
}
// Tester is the weekly job.
type Tester struct {
runner Runner
target Target
gate Gate
uplink UplinkTx
statePath string
slot int // minutes after StartHour:00 this host may first start
logger *slog.Logger
loc *time.Location
now func() time.Time
sleep func(context.Context, time.Duration) bool
mu sync.Mutex
st State
samples []txSample
}
// New builds the job and loads the persisted state (missing = empty; corrupt = logged, empty — the cost is one extra
// test, never a lost value elsewhere).
func New(runner Runner, target Target, gate Gate, uplink UplinkTx, hostID, statePath string, logger *slog.Logger) *Tester {
if logger == nil {
logger = slog.Default()
}
t := &Tester{runner: runner, target: target, gate: gate, uplink: uplink, statePath: statePath,
slot: SlotMinute(hostID), logger: logger, loc: time.Local, now: time.Now, sleep: sleepCtx}
t.load()
return t
}
func sleepCtx(ctx context.Context, d time.Duration) bool {
tm := time.NewTimer(d)
defer tm.Stop()
select {
case <-ctx.Done():
return false
case <-tm.C:
return true
}
}
// SlotMinute spreads hosts over the first SlotSpreadMinutes of the window, stable per host id.
func SlotMinute(hostID string) int {
h := fnv.New32a()
h.Write([]byte(hostID))
return int(h.Sum32() % SlotSpreadMinutes)
}
func (t *Tester) load() {
data, err := os.ReadFile(t.statePath)
if err != nil {
if !os.IsNotExist(err) {
t.logger.Warn("linetest: state read failed — starting empty", "path", t.statePath, "err", err)
}
return
}
if err := json.Unmarshal(data, &t.st); err != nil {
t.logger.Warn("linetest: state file corrupt — starting empty", "path", t.statePath, "err", err)
t.st = State{}
}
}
func (t *Tester) saveLocked() error {
data, err := json.MarshalIndent(t.st, "", " ")
if err != nil {
return err
}
if err := os.MkdirAll(filepath.Dir(t.statePath), 0o755); err != nil {
return err
}
tmp := t.statePath + ".tmp"
if err := os.WriteFile(tmp, data, 0o600); err != nil {
os.Remove(tmp)
return err
}
return os.Rename(tmp, t.statePath)
}
// EligibleAt reports whether a test may START at local time lt for a host whose slot is slot minutes after
// StartHour:00 on the due day. On a catch-up day (anchorDay=false) the slot is not waited for again.
func EligibleAt(lt time.Time, slot int, anchorDay bool) bool {
h := lt.Hour()
if h < StartHour || h >= EndHour {
return false
}
if anchorDay {
mins := (h-StartHour)*60 + lt.Minute()
return mins >= slot
}
return true
}
// weekAnchor is the most recent Tuesday StartHour:00 at or before lt.
func weekAnchor(lt time.Time) time.Time {
daysBack := (int(lt.Weekday()) - int(Weekday) + 7) % 7
d := lt.AddDate(0, 0, -daysBack)
a := time.Date(d.Year(), d.Month(), d.Day(), StartHour, 0, 0, 0, lt.Location())
if a.After(lt) {
d = d.AddDate(0, 0, -7)
a = time.Date(d.Year(), d.Month(), d.Day(), StartHour, 0, 0, 0, lt.Location())
}
return a
}
// Due reports whether a measurement is due at lt given the state.
func Due(st State, lt time.Time) bool {
return st.LastOKAt.IsZero() || st.LastOKAt.In(lt.Location()).Before(weekAnchor(lt))
}
// sample records the uplink counter (once a minute) and keeps QuietMinutes+1 samples.
func (t *Tester) sample() {
if t.uplink == nil {
return
}
tx, err := t.uplink()
if err != nil {
t.logger.Debug("linetest: uplink counter unreadable", "err", err)
t.mu.Lock()
t.samples = nil // a gap is not quiet
t.mu.Unlock()
return
}
t.mu.Lock()
t.samples = append(t.samples, txSample{at: t.now(), tx: tx})
if len(t.samples) > QuietMinutes+1 {
t.samples = t.samples[len(t.samples)-(QuietMinutes+1):]
}
t.mu.Unlock()
}
// quiet reports whether the last QuietMinutes minutes each sent below QuietMbps, and the highest minute seen.
func quiet(samples []txSample) (bool, float64) {
if len(samples) < QuietMinutes+1 {
return false, 0
}
worst := 0.0
for i := 1; i < len(samples); i++ {
dt := samples[i].at.Sub(samples[i-1].at).Seconds()
if dt <= 0 || dt > 3*TickInterval.Seconds() || samples[i].tx < samples[i-1].tx {
return false, worst // a gap or a counter reset is not proof of quiet
}
mbps := float64(samples[i].tx-samples[i-1].tx) * 8 / dt / 1e6
if mbps > worst {
worst = mbps
}
}
return worst < QuietMbps, worst
}
// Run ticks every TickInterval until ctx ends. It never returns an error: a failed test is a reported fact.
func (t *Tester) Run(ctx context.Context) {
t.logger.Info("linetest: weekly upload line test starting", "schedule", ScheduleText, "slot_minute", t.slot)
if !t.sleep(ctx, FirstTickDelay) {
return
}
for {
t.sample()
t.Tick(ctx)
if !t.sleep(ctx, TickInterval) {
return
}
}
}
// Refusal is why a test on request did not run (RunNow). Code is stable for the controller.
type Refusal struct {
Code string // not-due | outside-window | tries-used | no-tunnel | heavy-op | not-quiet
Msg string
}
func (r *Refusal) Error() string { return "linetest: " + r.Code + ": " + r.Msg }
// ready applies every rule except the quiet one: due this week, inside the window (the slot on the due day), tries
// left today, the tunnel up. It answers the server address and the local day.
func (t *Tester) ready() (addr, day string, refusal *Refusal) {
lt := t.now().In(t.loc)
t.mu.Lock()
st := t.st
t.mu.Unlock()
if !Due(st, lt) {
return "", "", &Refusal{"not-due", "measured this week"}
}
anchor := weekAnchor(lt)
anchorDay := lt.YearDay() == anchor.YearDay() && lt.Year() == anchor.Year()
if !EligibleAt(lt, t.slot, anchorDay) {
return "", "", &Refusal{"outside-window", "a test starts 10:00-15:59 local, from this host's slot on the due day"}
}
day = lt.Format("2006-01-02")
if st.TriesDay == day && st.TriesToday >= MaxTriesPerDay {
return "", "", &Refusal{"tries-used", fmt.Sprintf("%d tries today", st.TriesToday)}
}
ip, ok := t.target()
if !ok {
return "", "", &Refusal{"no-tunnel", "no operational tunnel"}
}
return netip.AddrPortFrom(ip, ServerPort).String(), day, nil
}
// Tick is one look.
func (t *Tester) Tick(ctx context.Context) {
addr, day, refusal := t.ready()
if refusal != nil {
if refusal.Code == "no-tunnel" {
t.logger.Debug("linetest: no operational tunnel — no test")
}
return
}
t.mu.Lock()
samples := append([]txSample(nil), t.samples...)
t.mu.Unlock()
if q, worst := quiet(samples); !q {
t.logger.Debug("linetest: uplink not quiet — waiting", "worst_minute_mbps", fmt.Sprintf("%.1f", worst), "samples", len(samples))
return
}
release, busy, ok := t.gate.TryAcquire(GateName)
if !ok {
t.logger.Info("linetest: deferred — a heavy operation is in flight", "busy", busy)
return
}
defer release()
t.measure(ctx, addr, day, RunGap)
}
// WaitingForQuiet reports that a measurement is due and could start now but the uplink has not been quiet for
// QuietMinutes — the case a seeding torrent app makes permanent (R-1250). The controller reads it from
// GET /host/line-test and only then holds its app's limit and calls RunNow.
func (t *Tester) WaitingForQuiet() bool {
if _, _, refusal := t.ready(); refusal != nil {
return false
}
t.mu.Lock()
samples := append([]txSample(nil), t.samples...)
t.mu.Unlock()
if len(samples) < QuietMinutes+1 {
return false // not yet watched for 10 minutes (a fresh start): not known to be busy, so no hold
}
q, _ := quiet(samples)
return !q
}
// RunNow is a test on request (POST /host/line-test/run, R-1250): every rule of the weekly test except the 10 quiet
// minutes, which the caller replaces by holding its own upload down. The uplink must still be quiet over
// RequestQuietWindow (after RequestSettle), so a request never measures the family's upload against itself. A
// *Refusal says why nothing ran; any other error is a failed measurement (the same as a failed weekly attempt).
func (t *Tester) RunNow(ctx context.Context) (mbps float64, runs []float64, err error) {
addr, day, refusal := t.ready()
if refusal != nil {
return 0, nil, refusal
}
release, busy, ok := t.gate.TryAcquire(GateName)
if !ok {
return 0, nil, &Refusal{"heavy-op", busy}
}
defer release()
if t.uplink == nil {
return 0, nil, &Refusal{"not-quiet", "no uplink counter"}
}
if !t.sleep(ctx, RequestSettle) {
return 0, nil, ctx.Err()
}
tx0, err0 := t.uplink()
at0 := t.now()
if !t.sleep(ctx, RequestQuietWindow) {
return 0, nil, ctx.Err()
}
tx1, err1 := t.uplink()
if err0 != nil || err1 != nil || tx1 < tx0 {
return 0, nil, &Refusal{"not-quiet", "uplink counter unreadable or reset"}
}
secs := t.now().Sub(at0).Seconds()
if secs <= 0 {
secs = RequestQuietWindow.Seconds()
}
if m := float64(tx1-tx0) * 8 / secs / 1e6; m >= QuietMbps {
t.logger.Info("linetest: test on request refused — the uplink is not quiet", "mbps", fmt.Sprintf("%.1f", m))
return 0, nil, &Refusal{"not-quiet", fmt.Sprintf("the uplink sent %.1f Mbit/s while held", m)}
}
t.logger.Info("linetest: test on request (a held torrent app, R-1250)", "server", addr)
if err := t.measure(ctx, addr, day, RequestRunGap); err != nil {
return 0, nil, err
}
t.mu.Lock()
defer t.mu.Unlock()
return t.st.Mbps, append([]float64(nil), t.st.RunsMbps...), nil
}
// measure runs the runs, gap apart, and stores the result. It returns nil only when a value was stored.
func (t *Tester) measure(ctx context.Context, addr, day string, gap time.Duration) error {
start := t.now()
var runs []float64
var lastErr error
busy := false
for i := 0; i < RunsPerMeasurement; i++ {
if i > 0 && !t.sleep(ctx, gap) {
break
}
rctx, cancel := context.WithTimeout(ctx, RunTimeout)
r, err := t.runner.Run(rctx, addr)
cancel()
if errors.Is(err, ErrBusy) {
busy, lastErr = true, err
break
}
if err != nil {
lastErr = err
t.logger.Warn("linetest: run failed", "run", i+1, "server", addr, "err", err)
continue
}
runs = append(runs, r.Mbps)
t.logger.Info(fmt.Sprintf("linetest: run %d: %.1f Mbit/s", i+1, r.Mbps), "server", addr,
"server_bytes", r.ServerBytes, "server_seconds", r.ServerSeconds)
}
t.mu.Lock()
defer t.mu.Unlock()
if t.st.TriesDay != day {
t.st.TriesDay, t.st.TriesToday = day, 0
}
t.st.TriesToday++
t.st.LastAttemptAt = start.UTC()
if len(runs) > 0 {
best := runs[0]
for _, r := range runs {
if r > best {
best = r
}
}
t.st.LastOKAt, t.st.Mbps, t.st.RunsMbps, t.st.Server, t.st.LastError = start.UTC(), best, runs, addr, ""
t.logger.Info(fmt.Sprintf("linetest: upload line measured: %.1f Mbit/s (best of %d)", best, len(runs)),
"server", addr, "runs", len(runs), "duration_s", t.now().Sub(start).Seconds())
} else {
msg := "no run succeeded"
if lastErr != nil {
msg = lastErr.Error()
}
if len(msg) > 300 {
msg = msg[:300]
}
t.st.LastError = msg
if busy {
t.logger.Info("linetest: the test server is busy — trying again later today", "server", addr, "tries_today", t.st.TriesToday)
} else {
t.logger.Warn("linetest: measurement failed", "server", addr, "tries_today", t.st.TriesToday, "err", msg)
}
}
if err := t.saveLocked(); err != nil {
t.logger.Warn("linetest: state save failed — the result will not survive a restart", "path", t.statePath, "err", err)
}
if len(runs) > 0 {
return nil
}
if lastErr == nil {
lastErr = errors.New("linetest: no run succeeded")
}
return lastErr
}
// Value is the measured line (ok=false: never measured).
func (t *Tester) Value() (mbps float64, at time.Time, ok bool) {
t.mu.Lock()
defer t.mu.Unlock()
if t.st.LastOKAt.IsZero() || t.st.Mbps <= 0 {
return 0, time.Time{}, false
}
return t.st.Mbps, t.st.LastOKAt, true
}
// LineTestStatus implements hub.LineTestReporter: a pure read of the persisted state.
func (t *Tester) LineTestStatus(context.Context) *hub.LineTestStatus {
t.mu.Lock()
defer t.mu.Unlock()
out := &hub.LineTestStatus{Schedule: ScheduleText, Mbps: t.st.Mbps, RunsMbps: t.st.RunsMbps, Server: t.st.Server,
LastError: t.st.LastError}
if !t.st.LastOKAt.IsZero() {
out.LastOKAt = t.st.LastOKAt.UTC().Format(time.RFC3339)
}
if !t.st.LastAttemptAt.IsZero() {
out.LastAttemptAt = t.st.LastAttemptAt.UTC().Format(time.RFC3339)
}
return out
}
// ── the host's physical uplink counter ──────────────────────────────────────────────────────────────────────────
// SysUplinkTx reads the transmit bytes of the host's physical uplink under root (normally "/"): the interface of the
// default route in /proc/net/route, and — when it is a bridge (vmbr0), as on every Proxmox host — the sum of its
// PHYSICAL member ports (those with a `device` link). The bridge's own counter does not see the guests' bridged
// traffic, which is the traffic this check exists for.
func SysUplinkTx(root string) UplinkTx {
return func() (uint64, error) {
ifc, err := defaultRouteIface(filepath.Join(root, "proc/net/route"))
if err != nil {
return 0, err
}
sys := filepath.Join(root, "sys/class/net")
ports := []string{ifc}
if ents, err := os.ReadDir(filepath.Join(sys, ifc, "brif")); err == nil {
ports = ports[:0]
for _, e := range ents {
if _, err := os.Stat(filepath.Join(sys, e.Name(), "device")); err == nil {
ports = append(ports, e.Name())
}
}
if len(ports) == 0 {
return 0, fmt.Errorf("linetest: bridge %s has no physical port", ifc)
}
}
var sum uint64
for _, p := range ports {
b, err := os.ReadFile(filepath.Join(sys, p, "statistics/tx_bytes"))
if err != nil {
return 0, err
}
n, err := strconv.ParseUint(strings.TrimSpace(string(b)), 10, 64)
if err != nil {
return 0, err
}
sum += n
}
return sum, nil
}
}
func defaultRouteIface(path string) (string, error) {
b, err := os.ReadFile(path)
if err != nil {
return "", err
}
for _, line := range strings.Split(string(b), "\n")[1:] {
f := strings.Fields(line)
if len(f) >= 8 && f[1] == "00000000" && f[7] == "00000000" {
return f[0], nil
}
}
return "", errors.New("linetest: no default route")
}
+574
View File
@@ -0,0 +1,574 @@
package linetest
import (
"context"
"encoding/binary"
"encoding/json"
"errors"
"io"
"net"
"net/netip"
"os"
"path/filepath"
"sync"
"testing"
"time"
"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
)
// ── a fake iperf3 server (the server half of the protocol, as `iperf3 --server` drives it) ─────────────────────────
type fakeServer struct {
ln net.Listener
busy bool // answer ACCESS_DENIED
stall bool // accept the cookie, then never answer
recvSeconds float64 // what the server reports as its duration
gotIDs []int // the stream ids the client reported
gotParams map[string]any
mu sync.Mutex
received int64
}
func newFakeServer(t *testing.T) *fakeServer {
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
s := &fakeServer{ln: ln, recvSeconds: 1}
t.Cleanup(func() { ln.Close() })
go s.serve()
return s
}
func (s *fakeServer) addr() string { return s.ln.Addr().String() }
func wjson(w io.Writer, v any) {
b, _ := json.Marshal(v)
var h [4]byte
binary.BigEndian.PutUint32(h[:], uint32(len(b)))
w.Write(append(h[:], b...))
}
func (s *fakeServer) serve() {
ctrl, err := s.ln.Accept()
if err != nil {
return
}
defer ctrl.Close()
cookie := make([]byte, cookieSize)
if _, err := io.ReadFull(ctrl, cookie); err != nil {
return
}
if s.busy {
ctrl.Write([]byte{0xff})
return
}
if s.stall {
time.Sleep(5 * time.Second)
return
}
ctrl.Write([]byte{stParamExchange})
var params map[string]any
if readJSON(ctrl, &params) != nil {
return
}
s.gotParams = params
n := int(params["parallel"].(float64))
ctrl.Write([]byte{stCreateStreams})
var wg sync.WaitGroup
var conns []net.Conn
for i := 0; i < n; i++ {
d, err := s.ln.Accept()
if err != nil {
return
}
c := make([]byte, cookieSize)
io.ReadFull(d, c)
conns = append(conns, d)
wg.Add(1)
go func(d net.Conn) {
defer wg.Done()
k, _ := io.Copy(io.Discard, d)
s.mu.Lock()
s.received += k
s.mu.Unlock()
}(d)
}
ctrl.Write([]byte{stTestStart, stTestRunning})
var st [1]byte
if _, err := io.ReadFull(ctrl, st[:]); err != nil || st[0] != stTestEnd {
return
}
ctrl.Write([]byte{stExchangeResults})
var theirs results
if readJSON(ctrl, &theirs) != nil {
return
}
for _, x := range theirs.Streams {
s.gotIDs = append(s.gotIDs, x.ID)
}
// The server counts what it RECEIVED; report a fixed figure so the test can tell it from the client's own count.
wjson(ctrl, results{Streams: []streamResult{{ID: 1, Bytes: 12_500_000, EndTime: s.recvSeconds}}})
ctrl.Write([]byte{stDisplayResults})
io.ReadFull(ctrl, st[:])
for _, d := range conns {
d.Close()
}
wg.Wait()
}
func TestClientSpeaksTheProtocolAndCountsTheServersBytes(t *testing.T) {
s := newFakeServer(t)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
r, err := Client{Streams: 4, Time: 300 * time.Millisecond}.Run(ctx, s.addr())
if err != nil {
t.Fatalf("Run: %v", err)
}
if r.ServerBytes != 12_500_000 || r.Mbps != 100 {
t.Fatalf("result = %+v; want the SERVER's 12.5 MB in 1 s = 100 Mbit/s", r)
}
if want := []int{1, 3, 4, 5}; len(s.gotIDs) != 4 || s.gotIDs[0] != want[0] || s.gotIDs[1] != want[1] || s.gotIDs[3] != want[3] {
t.Fatalf("stream ids sent = %v, want %v (iperf3 numbers its streams 1, 3, 4, 5)", s.gotIDs, want)
}
if s.gotParams["tcp"] != true || s.gotParams["parallel"].(float64) != 4 {
t.Fatalf("params = %v", s.gotParams)
}
}
func TestClientBusyServerIsErrBusy(t *testing.T) {
s := newFakeServer(t)
s.busy = true
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if _, err := (Client{Streams: 4, Time: time.Second}).Run(ctx, s.addr()); !errors.Is(err, ErrBusy) {
t.Fatalf("err = %v, want ErrBusy", err)
}
}
func TestClientHonoursTheDeadlineAgainstAStalledServer(t *testing.T) {
s := newFakeServer(t)
s.stall = true
ctx, cancel := context.WithTimeout(context.Background(), 400*time.Millisecond)
defer cancel()
start := time.Now()
_, err := Client{Streams: 4, Time: time.Second}.Run(ctx, s.addr())
if err == nil {
t.Fatal("a stalled server gave no error")
}
if took := time.Since(start); took > 1500*time.Millisecond {
t.Fatalf("Run took %v against a stalled server; the deadline is 400 ms", took)
}
}
func TestClientRefusesAContextWithoutDeadline(t *testing.T) {
if _, err := (Client{}).Run(context.Background(), "127.0.0.1:1"); err == nil {
t.Fatal("Run without a deadline must refuse (an unbounded test is the thing RunTimeout exists to prevent)")
}
}
// ── the schedule ────────────────────────────────────────────────────────────────────────────────────────────────
func TestEligibleWindowNeverInTheNightChain(t *testing.T) {
loc := time.FixedZone("CET", 3600)
start := time.Date(2026, 10, 19, 0, 0, 0, 0, loc) // a Monday
for m := 0; m < 7*24*60; m++ {
lt := start.Add(time.Duration(m) * time.Minute)
for _, anchorDay := range []bool{true, false} {
if !EligibleAt(lt, 0, anchorDay) {
continue
}
if h := lt.Hour(); h < StartHour || h >= EndHour || (h >= 1 && h <= 6) {
t.Fatalf("eligible at %s — outside 10:00-15:59 or inside the night chain (01:00-06:59)", lt.Format("Mon 15:04"))
}
}
}
}
func TestSlotDelaysTheFirstStartOnTheDueDayOnly(t *testing.T) {
loc := time.UTC
tue := time.Date(2026, 10, 20, 10, 30, 0, 0, loc)
if EligibleAt(tue, 45, true) {
t.Fatal("10:30 with slot 45 must wait until 10:45 on the due day")
}
if !EligibleAt(tue.Add(15*time.Minute), 45, true) {
t.Fatal("10:45 with slot 45 must be eligible")
}
if !EligibleAt(tue.Add(24*time.Hour), 45, false) {
t.Fatal("a catch-up day does not wait for the slot again")
}
if a, b := SlotMinute("demo-hp-bb76ea"), SlotMinute("demo-felhom-8363b5"); a < 0 || a >= SlotSpreadMinutes || b < 0 || b >= SlotSpreadMinutes {
t.Fatalf("slots out of range: %d %d", a, b)
}
}
func TestDueWeeklyFromTuesday(t *testing.T) {
loc := time.UTC
mon := time.Date(2026, 10, 19, 12, 0, 0, 0, loc)
tue := time.Date(2026, 10, 20, 12, 0, 0, 0, loc)
if !Due(State{}, mon) {
t.Fatal("never measured is due")
}
st := State{LastOKAt: time.Date(2026, 10, 14, 11, 0, 0, 0, loc)} // last Wednesday
if Due(st, mon) {
t.Fatal("measured after last Tuesday 10:00 is not due on Monday")
}
if !Due(st, tue) {
t.Fatal("from Tuesday 10:00 a new measurement is due")
}
}
// ── the job ─────────────────────────────────────────────────────────────────────────────────────────────────────
type fakeRunner struct {
mu sync.Mutex
calls int
out []runOut
}
type runOut struct {
mbps float64
err error
}
func (f *fakeRunner) Run(ctx context.Context, addr string) (Result, error) {
f.mu.Lock()
defer f.mu.Unlock()
i := f.calls
f.calls++
if i >= len(f.out) {
return Result{}, errors.New("no more scripted runs")
}
return Result{Mbps: f.out[i].mbps, ServerBytes: 1}, f.out[i].err
}
type fakeGate struct{ busy string }
func (g *fakeGate) TryAcquire(what string) (func(), string, bool) {
if g.busy != "" {
return nil, g.busy, false
}
return func() {}, "", true
}
type rig struct {
t *Tester
run *fakeRunner
gate *fakeGate
now time.Time
tx uint64
tunnel bool
}
// newRig: Tuesday 12:00 UTC (eligible, any slot), tunnel up, 11 quiet samples taken.
func newRig(t *testing.T, outs ...runOut) *rig {
r := &rig{run: &fakeRunner{out: outs}, gate: &fakeGate{}, now: time.Date(2026, 10, 20, 12, 0, 0, 0, time.UTC), tunnel: true}
r.t = New(r.run, func() (netip.Addr, bool) { return netip.MustParseAddr("10.77.0.1"), r.tunnel }, r.gate,
func() (uint64, error) { return r.tx, nil }, "host-x", filepath.Join(t.TempDir(), "line-test.json"), nil)
r.t.loc = time.UTC
r.t.now = func() time.Time { return r.now }
r.t.sleep = func(context.Context, time.Duration) bool { return true }
r.quietMinutes(QuietMinutes + 1)
return r
}
func (r *rig) minute(bytes uint64) {
r.now = r.now.Add(time.Minute)
r.tx += bytes
r.t.sample()
}
func (r *rig) quietMinutes(n int) {
for i := 0; i < n; i++ {
r.minute(1_000_000) // 1 MB/min = 0.13 Mbit/s
}
}
func TestTickMeasuresTheHighestOfThreeRuns(t *testing.T) {
r := newRig(t, runOut{mbps: 88}, runOut{mbps: 91.5}, runOut{mbps: 90})
r.t.Tick(context.Background())
mbps, _, ok := r.t.Value()
if !ok || mbps != 91.5 || r.run.calls != RunsPerMeasurement {
t.Fatalf("value %.1f ok=%v calls=%d; want the highest of 3 runs = 91.5", mbps, ok, r.run.calls)
}
r.t.Tick(context.Background())
if r.run.calls != RunsPerMeasurement {
t.Fatal("a measured week must not test again")
}
}
func TestBusyServerIsRetriedLaterTheSameDay(t *testing.T) {
r := newRig(t, runOut{err: ErrBusy}, runOut{mbps: 90}, runOut{mbps: 89}, runOut{mbps: 92})
r.t.Tick(context.Background())
if _, _, ok := r.t.Value(); ok || r.run.calls != 1 {
t.Fatalf("busy: ok=%v calls=%d; want no value and the measurement stopped after the busy answer", ok, r.run.calls)
}
if r.t.st.TriesToday != 1 || r.t.st.LastError == "" {
t.Fatalf("busy attempt not recorded: %+v", r.t.st)
}
r.quietMinutes(1)
r.t.Tick(context.Background())
if mbps, _, ok := r.t.Value(); !ok || mbps != 92 {
t.Fatalf("retry later the same day: value %.1f ok=%v", mbps, ok)
}
}
func TestTriesPerDayAreBounded(t *testing.T) {
outs := make([]runOut, 20)
for i := range outs {
outs[i] = runOut{err: ErrBusy}
}
r := newRig(t, outs...)
for i := 0; i < 10; i++ {
r.quietMinutes(1)
r.t.Tick(context.Background())
}
if r.run.calls != MaxTriesPerDay {
t.Fatalf("calls = %d in one day, want at most %d", r.run.calls, MaxTriesPerDay)
}
}
func TestNoTunnelNoTest(t *testing.T) {
r := newRig(t, runOut{mbps: 90})
r.tunnel = false
r.t.Tick(context.Background())
if r.run.calls != 0 {
t.Fatal("tested without an operational tunnel")
}
if _, _, ok := r.t.Value(); ok {
t.Fatal("a value without a test")
}
}
func TestBusyUplinkOrHeavyOpWaits(t *testing.T) {
r := newRig(t, runOut{mbps: 90}, runOut{mbps: 90}, runOut{mbps: 90})
r.minute(60_000_000) // one minute at 8 Mbit/s: the family is uploading
r.t.Tick(context.Background())
if r.run.calls != 0 {
t.Fatal("tested while the uplink was busy in the last 10 minutes")
}
r.quietMinutes(QuietMinutes)
r.gate.busy = "backup:felhom-pbs"
r.t.Tick(context.Background())
if r.run.calls != 0 {
t.Fatal("tested beside a heavy operation")
}
r.gate.busy = ""
r.t.Tick(context.Background())
if r.run.calls == 0 {
t.Fatal("never tested once the uplink was quiet and the gate free")
}
}
func TestNeverTestsInTheNightOrBeforeTuesday(t *testing.T) {
r := newRig(t, runOut{mbps: 90}, runOut{mbps: 90}, runOut{mbps: 90})
r.now = time.Date(2026, 10, 20, 3, 0, 0, 0, time.UTC) // Tuesday 03:00 — the night chain
r.quietMinutes(QuietMinutes + 1)
r.t.Tick(context.Background())
r.t.st.LastOKAt = time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC)
r.now = time.Date(2026, 10, 19, 12, 0, 0, 0, time.UTC) // Monday noon, measured last week
r.quietMinutes(QuietMinutes + 1)
r.t.Tick(context.Background())
if r.run.calls != 0 {
t.Fatalf("tested %d times at night or when not due", r.run.calls)
}
}
func TestStateSurvivesARestartAndReportsIt(t *testing.T) {
r := newRig(t, runOut{mbps: 90.7}, runOut{err: errors.New("reset")}, runOut{mbps: 80})
r.t.Tick(context.Background())
again := New(r.run, nil, r.gate, nil, "host-x", r.t.statePath, nil)
mbps, at, ok := again.Value()
if !ok || mbps != 90.7 || at.IsZero() {
t.Fatalf("after a restart: %.1f %v %v", mbps, at, ok)
}
b, _ := json.Marshal(again.LineTestStatus(context.Background()))
var back hub.LineTestStatus
json.Unmarshal(b, &back)
if back.Mbps != 90.7 || back.LastOKAt == "" || len(back.RunsMbps) != 2 || back.Schedule == "" {
t.Fatalf("report stanza = %s", b)
}
}
// ── the uplink counter ──────────────────────────────────────────────────────────────────────────────────────────
func TestUplinkCountsTheBridgesPhysicalPortsOnly(t *testing.T) {
root := t.TempDir()
w := func(p, s string) {
os.MkdirAll(filepath.Dir(filepath.Join(root, p)), 0o755)
os.WriteFile(filepath.Join(root, p), []byte(s), 0o644)
}
w("proc/net/route", "Iface\tDestination\tGateway\tFlags\tRefCnt\tUse\tMetric\tMask\n"+
"wg-felhom\t0000004D\t00000000\t0001\t0\t0\t0\tFFFFFFFF\n"+
"vmbr0\t00000000\t0100A8C0\t0003\t0\t0\t0\t00000000\n")
w("sys/class/net/vmbr0/statistics/tx_bytes", "5\n")
w("sys/class/net/vmbr0/brif/enp1s0/.keep", "")
w("sys/class/net/vmbr0/brif/veth9201i0/.keep", "")
w("sys/class/net/enp1s0/device/.keep", "")
w("sys/class/net/enp1s0/statistics/tx_bytes", "1000\n")
w("sys/class/net/veth9201i0/statistics/tx_bytes", "777777\n")
got, err := SysUplinkTx(root)()
if err != nil || got != 1000 {
t.Fatalf("uplink tx = %d, %v; want 1000 (the physical port only — not the bridge, not the guest's veth)", got, err)
}
}
func TestQuietNeedsTenFullMinutes(t *testing.T) {
base := time.Date(2026, 10, 20, 12, 0, 0, 0, time.UTC)
var s []txSample
for i := 0; i <= QuietMinutes; i++ {
s = append(s, txSample{at: base.Add(time.Duration(i) * time.Minute), tx: uint64(i) * 1_000_000})
}
if q, _ := quiet(s); !q {
t.Fatal("ten quiet minutes must read quiet")
}
if q, _ := quiet(s[1:]); q {
t.Fatal("nine minutes of samples must not read quiet")
}
gap := append([]txSample(nil), s...)
gap[5].at = gap[5].at.Add(-10 * time.Minute)
if q, _ := quiet(gap); q {
t.Fatal("a gap in the samples must not read quiet")
}
}
// ── a test on request (R-1250): a torrent app that seeds all day ───────────────────────────────────────────────────
// seedingRig: due and eligible, but every minute of the last ten sent at the line's speed — the weekly Tick never runs.
func seedingRig(t *testing.T, outs ...runOut) *rig {
r := newRig(t, outs...)
for i := 0; i < QuietMinutes+1; i++ {
r.minute(600_000_000) // 80 Mbit/s: a seed at the line's speed
}
return r
}
func TestSeedingBoxIsWaitingForQuietAndRunNowMeasuresIt(t *testing.T) {
r := seedingRig(t, runOut{mbps: 88}, runOut{mbps: 92}, runOut{mbps: 90})
r.t.Tick(context.Background())
if r.run.calls != 0 {
t.Fatal("the weekly Tick tested a seeding box")
}
if !r.t.WaitingForQuiet() {
t.Fatal("a due, eligible, seeding box must report waiting_for_quiet")
}
var slept []time.Duration
r.t.sleep = func(_ context.Context, d time.Duration) bool { slept = append(slept, d); return true }
mbps, runs, err := r.t.RunNow(context.Background()) // the controller holds the app: the counter stands still
if err != nil || mbps != 92 || len(runs) != RunsPerMeasurement || r.run.calls != RunsPerMeasurement {
t.Fatalf("RunNow = %.1f %v %v, calls %d; want the best of 3 = 92", mbps, runs, err, r.run.calls)
}
want := []time.Duration{RequestSettle, RequestQuietWindow, RequestRunGap, RequestRunGap}
if len(slept) != len(want) {
t.Fatalf("slept %v, want %v", slept, want)
}
var total time.Duration
for i := range want {
if slept[i] != want[i] {
t.Fatalf("slept %v, want %v", slept, want)
}
total += slept[i]
}
if total+RunsPerMeasurement*TestTime > 60*time.Second {
t.Fatalf("a request holds the app %v + runs; it must end inside 60 s", total)
}
if v, _, ok := r.t.Value(); !ok || v != 92 {
t.Fatalf("the requested measurement was not stored: %.1f %v", v, ok)
}
if r.t.WaitingForQuiet() {
t.Fatal("measured this week: no longer waiting")
}
if _, _, err := r.t.RunNow(context.Background()); refusalCode(err) != "not-due" {
t.Fatalf("a second request this week: %v, want not-due", err)
}
}
func refusalCode(err error) string {
var ref *Refusal
if errors.As(err, &ref) {
return ref.Code
}
return ""
}
// The held app is not the only uploader: a request still refuses to measure a busy uplink.
func TestRunNowRefusesAnUplinkThatIsStillBusy(t *testing.T) {
r := seedingRig(t, runOut{mbps: 90}, runOut{mbps: 90}, runOut{mbps: 90})
r.t.uplink = func() (uint64, error) { r.tx += 20_000_000; return r.tx, nil } // 20 MB per 10 s = 16 Mbit/s
if _, _, err := r.t.RunNow(context.Background()); refusalCode(err) != "not-quiet" || r.run.calls != 0 {
t.Fatalf("RunNow on a busy uplink: %v, calls %d; want not-quiet and no run", err, r.run.calls)
}
}
// Every other rule of the weekly test holds for a request.
func TestRunNowKeepsTheWeeklyRules(t *testing.T) {
r := seedingRig(t)
r.gate.busy = "backup:felhom-pbs"
if _, _, err := r.t.RunNow(context.Background()); refusalCode(err) != "heavy-op" {
t.Fatalf("beside a backup: %v, want heavy-op", err)
}
r.gate.busy = ""
r.tunnel = false
if _, _, err := r.t.RunNow(context.Background()); refusalCode(err) != "no-tunnel" {
t.Fatalf("no tunnel: %v, want no-tunnel", err)
}
if r.t.WaitingForQuiet() {
t.Fatal("no tunnel: a request could not run, so it must not be invited")
}
r.tunnel = true
r.now = time.Date(2026, 10, 20, 3, 0, 0, 0, time.UTC) // Tuesday 03:00 — the night chain
if _, _, err := r.t.RunNow(context.Background()); refusalCode(err) != "outside-window" {
t.Fatalf("at night: %v, want outside-window", err)
}
if r.run.calls != 0 {
t.Fatalf("%d runs on refused requests", r.run.calls)
}
}
// A failed request counts as a try, so a controller cannot request more than MaxTriesPerDay tests a day.
func TestRunNowFailuresSpendTheDaysTries(t *testing.T) {
var outs []runOut
for i := 0; i < MaxTriesPerDay*RunsPerMeasurement; i++ {
outs = append(outs, runOut{err: errors.New("connection refused")})
}
r := seedingRig(t, outs...)
for i := 0; i < MaxTriesPerDay; i++ {
if _, _, err := r.t.RunNow(context.Background()); err == nil || refusalCode(err) != "" {
t.Fatalf("try %d: %v, want a failed measurement", i+1, err)
}
}
if _, _, err := r.t.RunNow(context.Background()); refusalCode(err) != "tries-used" {
t.Fatalf("after %d failures: %v, want tries-used", MaxTriesPerDay, err)
}
}
// A quiet box tests itself on the weekly Tick: it must not invite the controller to hold its app.
func TestQuietBoxIsNotWaiting(t *testing.T) {
r := newRig(t)
if r.t.WaitingForQuiet() {
t.Fatal("ten quiet minutes: the weekly Tick runs; waiting_for_quiet must be false")
}
}
// Live proof 2026-10-24 (demo-hp): a freshly started tester answered waiting_for_quiet before it had watched a single
// minute, so a quiet box's torrent app would be held for nothing. Waiting needs the full window, seen busy.
func TestFreshTesterIsNotWaiting(t *testing.T) {
r := &rig{run: &fakeRunner{}, gate: &fakeGate{}, now: time.Date(2026, 10, 20, 12, 0, 0, 0, time.UTC), tunnel: true}
r.t = New(r.run, func() (netip.Addr, bool) { return netip.MustParseAddr("10.77.0.1"), r.tunnel }, r.gate,
func() (uint64, error) { return r.tx, nil }, "host-x", filepath.Join(t.TempDir(), "line-test.json"), nil)
r.t.loc = time.UTC
r.t.now = func() time.Time { return r.now }
if r.t.WaitingForQuiet() {
t.Fatal("no minute watched yet: waiting_for_quiet must be false")
}
for i := 0; i < QuietMinutes; i++ {
r.minute(600_000_000)
}
if r.t.WaitingForQuiet() {
t.Fatal("10 samples are 9 minutes: still not a full window")
}
r.minute(600_000_000)
if !r.t.WaitingForQuiet() {
t.Fatal("a full busy window: waiting_for_quiet must be true")
}
}
+89
View File
@@ -0,0 +1,89 @@
package localapi
import (
"context"
"errors"
"net/http"
"time"
"gitea.dooplex.hu/admin/felhom-agent/internal/linetest"
)
// LineTestReader is the R-1247 weekly upload line test (internal/linetest.Tester) as the local API reads it.
type LineTestReader interface {
Value() (mbps float64, at time.Time, ok bool)
}
// LineTestRequester is the R-1250 half: a due test the uplink never lets start, and a test on request. The Tester
// satisfies it; a reader without it answers waiting_for_quiet:false and 503 on the POST.
type LineTestRequester interface {
WaitingForQuiet() bool
RunNow(ctx context.Context) (mbps float64, runs []float64, err error)
}
// LineTestResponse is GET /host/line-test. Present=false: never measured, or the job is not wired (no tunnel, disabled,
// or an agent before the route answers 404 — the controller reads all three as „unknown").
type LineTestResponse struct {
Present bool `json:"present"`
Mbps float64 `json:"mbps,omitempty"`
MeasuredAt string `json:"measured_at,omitempty"` // RFC3339 UTC
Method string `json:"method,omitempty"` // "iperf3-tunnel": a 10-second upload to ep0 inside the tunnel
// WaitingForQuiet: a measurement is due and could start now, but the uplink has not been quiet for 10 minutes
// (R-1250). The controller with a seeding torrent app then holds its limit and POSTs /host/line-test/run.
WaitingForQuiet bool `json:"waiting_for_quiet,omitempty"`
}
// LineTestRunResponse is POST /host/line-test/run. OK=false with Refused: nothing ran (Refused is linetest.Refusal's
// code, or "server-busy"); OK=false without it: the runs failed.
type LineTestRunResponse struct {
OK bool `json:"ok"`
Mbps float64 `json:"mbps,omitempty"`
RunsMbps []float64 `json:"runs_mbps,omitempty"`
Refused string `json:"refused,omitempty"`
Error string `json:"error,omitempty"`
}
// SetLineTest wires the weekly line test (main.go builds it after the local API server).
func (s *Server) SetLineTest(r LineTestReader) { s.lineTest = r }
func (s *Server) handleLineTest(w http.ResponseWriter, r *http.Request, vmid int) {
if s.lineTest == nil {
writeOK(w, LineTestResponse{Present: false})
return
}
var waiting bool
if q, ok := s.lineTest.(LineTestRequester); ok {
waiting = q.WaitingForQuiet()
}
mbps, at, ok := s.lineTest.Value()
if !ok {
writeOK(w, LineTestResponse{Present: false, WaitingForQuiet: waiting})
return
}
s.logger.Debug("local-api: /host/line-test served", "vmid", vmid, "mbps", mbps, "measured_at", at.UTC().Format(time.RFC3339), "waiting_for_quiet", waiting)
writeOK(w, LineTestResponse{Present: true, Mbps: mbps, MeasuredAt: at.UTC().Format(time.RFC3339), Method: "iperf3-tunnel", WaitingForQuiet: waiting})
}
// handleLineTestRun is POST /host/line-test/run (R-1250): a test on request, synchronous (about 55 s). The answer is
// always 200 with the outcome in the body, except 503 when no line test is wired.
func (s *Server) handleLineTestRun(w http.ResponseWriter, r *http.Request, vmid int) {
q, ok := s.lineTest.(LineTestRequester)
if s.lineTest == nil || !ok {
writeErr(w, http.StatusServiceUnavailable, "line test not configured")
return
}
s.logger.Info("local-api: line test requested", "vmid", vmid)
mbps, runs, err := q.RunNow(r.Context())
var ref *linetest.Refusal
switch {
case err == nil:
writeOK(w, LineTestRunResponse{OK: true, Mbps: mbps, RunsMbps: runs})
case errors.As(err, &ref):
s.logger.Info("local-api: line test request refused", "vmid", vmid, "reason", ref.Code, "detail", ref.Msg)
writeOK(w, LineTestRunResponse{Refused: ref.Code, Error: ref.Msg})
case errors.Is(err, linetest.ErrBusy):
writeOK(w, LineTestRunResponse{Refused: "server-busy", Error: err.Error()})
default:
writeOK(w, LineTestRunResponse{Error: err.Error()})
}
}
+116
View File
@@ -0,0 +1,116 @@
package localapi
import (
"context"
"encoding/json"
"errors"
"log/slog"
"net/http/httptest"
"strings"
"testing"
"time"
"gitea.dooplex.hu/admin/felhom-agent/internal/linetest"
)
type fixedLine struct {
mbps float64
ok bool
}
func (f fixedLine) Value() (float64, time.Time, bool) {
return f.mbps, time.Date(2026, 10, 24, 6, 31, 0, 0, time.UTC), f.ok
}
func lineTestBody(t *testing.T, s *Server) LineTestResponse {
t.Helper()
rec := httptest.NewRecorder()
s.handleLineTest(rec, httptest.NewRequest("GET", "/host/line-test", nil), 9201)
var env struct {
OK bool `json:"ok"`
Data LineTestResponse `json:"data"`
}
if err := json.Unmarshal(rec.Body.Bytes(), &env); err != nil {
t.Fatalf("decode %q: %v", rec.Body.String(), err)
}
return env.Data
}
// R-1247: the controller sets qBittorrent's limit only from a PRESENT value; never measured and not wired must both
// read present:false, never a zero line.
func TestLineTestRoute(t *testing.T) {
s := &Server{logger: slog.Default()}
if got := lineTestBody(t, s); got.Present {
t.Fatalf("not wired: %+v, want present:false", got)
}
s.SetLineTest(fixedLine{ok: false})
if got := lineTestBody(t, s); got.Present || got.Mbps != 0 {
t.Fatalf("never measured: %+v, want present:false", got)
}
s.SetLineTest(fixedLine{mbps: 90.7, ok: true})
got := lineTestBody(t, s)
if !got.Present || got.Mbps != 90.7 || got.Method != "iperf3-tunnel" || !strings.HasPrefix(got.MeasuredAt, "2026-10-24T06:31") {
t.Fatalf("measured: %+v", got)
}
}
type requestableLine struct {
fixedLine
waiting bool
mbps float64
err error
calls int
}
func (r *requestableLine) WaitingForQuiet() bool { return r.waiting }
func (r *requestableLine) RunNow(context.Context) (float64, []float64, error) {
r.calls++
if r.err != nil {
return 0, nil, r.err
}
return r.mbps, []float64{r.mbps - 1, r.mbps}, nil
}
func lineTestRun(t *testing.T, s *Server) (int, LineTestRunResponse) {
t.Helper()
rec := httptest.NewRecorder()
s.handleLineTestRun(rec, httptest.NewRequest("POST", "/host/line-test/run", nil), 9201)
var env struct {
OK bool `json:"ok"`
Data LineTestRunResponse `json:"data"`
}
_ = json.Unmarshal(rec.Body.Bytes(), &env)
return rec.Code, env.Data
}
// R-1250: the controller acts only on waiting_for_quiet, and reads a refusal by its code — never as a measurement.
func TestLineTestRunRoute(t *testing.T) {
s := &Server{logger: slog.Default()}
if code, _ := lineTestRun(t, s); code != 503 {
t.Fatalf("not wired: HTTP %d, want 503", code)
}
s.SetLineTest(fixedLine{ok: true, mbps: 90})
if code, _ := lineTestRun(t, s); code != 503 {
t.Fatalf("a reader without the request half: HTTP %d, want 503", code)
}
l := &requestableLine{fixedLine: fixedLine{ok: false}, waiting: true, mbps: 93.5}
s.SetLineTest(l)
if got := lineTestBody(t, s); got.Present || !got.WaitingForQuiet {
t.Fatalf("GET while waiting: %+v, want present:false waiting_for_quiet:true", got)
}
if code, got := lineTestRun(t, s); code != 200 || !got.OK || got.Mbps != 93.5 || len(got.RunsMbps) != 2 {
t.Fatalf("a measured request: HTTP %d %+v", code, got)
}
l.err = &linetest.Refusal{Code: "not-quiet", Msg: "the uplink sent 16.0 Mbit/s while held"}
if _, got := lineTestRun(t, s); got.OK || got.Refused != "not-quiet" || got.Mbps != 0 {
t.Fatalf("a refusal: %+v, want ok:false refused:not-quiet", got)
}
l.err = linetest.ErrBusy
if _, got := lineTestRun(t, s); got.OK || got.Refused != "server-busy" {
t.Fatalf("server busy: %+v, want refused:server-busy", got)
}
l.err = errors.New("dial tcp 10.77.0.1:5201: i/o timeout")
if _, got := lineTestRun(t, s); got.OK || got.Refused != "" || got.Error == "" {
t.Fatalf("a failed run: %+v, want ok:false, no refusal, the error", got)
}
}
+5
View File
@@ -336,6 +336,7 @@ type Server struct {
formatJobs *FormatJobStore // F20-BUG3 detached-format job record (optional)
staleLock StaleLockController // F2-b startup stale-lock recovery (optional)
partialSweep func(ctx context.Context, vmid int) // R-1006 start-up sweep of dead vzdump leftovers (optional)
lineTest LineTestReader // R-1247 weekly line test (optional; nil → present:false)
// guestPower (F-REBOOT) is per-guest start-attempt state for the guest-power watchdog.
// Guarded by guestPowerMu in guestpower.go; in-memory on purpose (see guestPowerState).
guestPower map[int]guestPowerState
@@ -573,6 +574,10 @@ func (s *Server) Handler() http.Handler {
// R-856 (`09` §3 decision 143): the host crash guard's record of the last HOST boot — the controller
// waits ~15 min with app mails after an unclean one. Read-only; a missing/garbled file = present:false.
mux.HandleFunc("GET /host/crash-guard", s.withGuest(s.handleCrashGuard))
// R-1247: the weekly upload line test's last measurement — the controller's qBittorrent limit and upload watch
// read it. Read-only; never measured (or the job not wired) = present:false.
mux.HandleFunc("GET /host/line-test", s.withGuest(s.handleLineTest))
mux.HandleFunc("POST /host/line-test/run", s.withGuest(s.handleLineTestRun)) // R-1250: a test on request
// Disk management (slice 8C) — self-scoped; format routes through the data-bearing classifier+gate.
mux.HandleFunc("GET /disks", s.withGuest(s.handleDisks))
mux.HandleFunc("GET /disks/candidates", s.withGuest(s.handleDiskCandidates))
+32
View File
@@ -0,0 +1,32 @@
package wgtunnel
import (
"context"
"testing"
)
// R-1247: the line test's server is the desired block's pbs_tunnel_ip, and ONLY while the tunnel is operational
// (registered + unit active) — „no tunnel → no test" rests on this answer.
func TestLineTestTargetOnlyWhileOperational(t *testing.T) {
m, _, active, _ := testManager(t, &fakeHub{})
ctx := context.Background()
if _, ok := m.LineTestTarget(); ok {
t.Fatal("a target before any Apply")
}
m.Apply(ctx, false, nil) // registers
pub := localPub(t, m)
m.Apply(ctx, true, testBlock(pub)) // installs; the unit is not active yet
if _, ok := m.LineTestTarget(); ok {
t.Fatal("a target while the unit is not active")
}
*active = true
m.Apply(ctx, true, testBlock(pub))
ip, ok := m.LineTestTarget()
if !ok || ip.String() != "10.77.0.1" {
t.Fatalf("operational tunnel: target %v ok=%v, want 10.77.0.1", ip, ok)
}
m.Apply(ctx, true, nil) // revoked
if _, ok := m.LineTestTarget(); ok {
t.Fatal("a target after the hub revoked the tunnel")
}
}
+19 -1
View File
@@ -128,6 +128,7 @@ type Manager struct {
type convSnapshot struct {
wgBlockDesired bool
operational bool
serverIP netip.Addr // the desired block's pbs_tunnel_ip (R-1247 line test target); invalid when none
}
// TunnelConvergence returns the cached (desired, operational) snapshot without any exec (fast-tick
@@ -139,6 +140,17 @@ func (m *Manager) TunnelConvergence() (desired, operational bool) {
return m.conv.wgBlockDesired, m.conv.operational
}
// LineTestTarget is the R-1247 line test's server: ep0's tunnel address, ok only while the tunnel is operational
// (registered + unit active, as of the last Apply). Cached — no exec.
func (m *Manager) LineTestTarget() (netip.Addr, bool) {
m.mu.Lock()
defer m.mu.Unlock()
if !m.conv.operational || !m.conv.serverIP.IsValid() {
return netip.Addr{}, false
}
return m.conv.serverIP, true
}
// NewManager builds a Manager. stateDir is the agent state dir (default /var/lib/felhom-agent —
// note the FELHOM_WG sudoers install entry hard-codes the staged path under it).
func NewManager(runner proxmox.Runner, registrar Registrar, stateDir string, logger *slog.Logger) *Manager {
@@ -348,7 +360,13 @@ func (m *Manager) Apply(ctx context.Context, fetched bool, block *hub.WireWiregu
operational = m.isActive(ctx)
}
}
m.conv = convSnapshot{wgBlockDesired: desired, operational: operational}
var server netip.Addr
if desired {
if a, err := netip.ParseAddr(block.Endpoint.PBSTunnelIP); err == nil && a.Is4() {
server = a
}
}
m.conv = convSnapshot{wgBlockDesired: desired, operational: operational, serverIP: server}
}()
keyExists := false