mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-23 20:25:56 +00:00
Add independent bounded guest-write evidence for backup thaw
An OK PVE task and HTTP admission controls cannot establish thaw. Supply a credential-free guest-local fsync/readback witness and bounded offline record checks for the real #2439 job, without QGA probes, recovery actions or a replacement native platform. Require actual overlap, every covered filesystem and truthful monitoring resumption; retain native provenance and execution with Delivery. Contract-Neutral: Measurement support for the existing guest-agent backup safety obligation only; production APIs, polling, access boundaries and release gates are unchanged. Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
@@ -58,6 +58,15 @@ on a containing build remains required. Until then, stopping Pulse before a
|
||||
freeze-enabled backup and restarting only after verified thaw remains the safe
|
||||
workaround, with monitoring and alerts unavailable during that window.
|
||||
|
||||
The [bounded backup observation tools](../../../../../tests/qualification/guest-agent-backup/README.md)
|
||||
provide an independent guest filesystem writer (file/directory fsync and
|
||||
synthetic readback over non-QGA stdout) and a fail-closed checker for native
|
||||
overlap, truthful deferred readings/History and post-task writes/resumption.
|
||||
An OK task without fresh writes on every covered filesystem cannot pass those
|
||||
record checks. Local tests are not native freeze/thaw evidence; supplied
|
||||
platform/artifact provenance and mount completeness still need independent
|
||||
review. The tools add no runtime interface, recovery action or release gate.
|
||||
|
||||
### Exact TrueNAS subscription termination — issue #2396
|
||||
|
||||
JSON-RPC stream readers and the subscription-acknowledgement wait recognise
|
||||
|
||||
@@ -0,0 +1,160 @@
|
||||
# Guest-agent backup observation — #2439
|
||||
|
||||
This is a **measurement tool**, not a new release gate or a backup hook. It
|
||||
does not start Pulse, a backup, PVE or QGA. It never sends a guest-agent command,
|
||||
freezes/thaws a filesystem, changes ACLs, disables freezing or restarts/resets
|
||||
a guest. Local tests and an OK backup task do not establish native acceptance.
|
||||
|
||||
The reported job is PVE **9.2.21**, **HAOS with QGA**, and Pulse **6.4.5** using
|
||||
the same effective **Audit + FileRead** access: a backup task reported OK while
|
||||
the guest remained frozen. Pulse's containing-source mitigation starts at
|
||||
`3c57c2d20728f9e711ae2930e282b853eb218730` (with its ancestors). Source controls
|
||||
cover admission, lock checks, in-flight payload withdrawal, no replay and cache
|
||||
truth. They cannot atomically reserve the native PVE/QGA channel or prove thaw.
|
||||
|
||||
## Smallest useful native route
|
||||
|
||||
Delivery owns provisioning and native execution. Core supplies the independent
|
||||
guest writer and record checks here. The existing private CI VM can build these
|
||||
tools and judge returned records; guest sudo does not supply a PVE hypervisor,
|
||||
HAOS, QGA's real channel or permission to reach an existing host. An ordinary
|
||||
QGA subprocess, simulated PVE API or observation guest with PVE/QGA disabled
|
||||
cannot meet this job. Do not build a replacement platform just to call it native.
|
||||
|
||||
The missing resource is **one owned disposable PVE target with a real HAOS/QGA
|
||||
guest and backup storage**, isolated from production, plus verified official
|
||||
image/package inputs and a supported way for its owner to observe it. Use an
|
||||
already provisioned suitable target if available; otherwise Delivery returns
|
||||
the exact provisioning/input gap through its normal reviewed infrastructure
|
||||
route. This file is not permission to use a host, account or credential.
|
||||
|
||||
Before native execution the owner retains, privately:
|
||||
|
||||
- The exact inspected Pulse binary digest, source/tree, required guard ancestry,
|
||||
build/artifact provenance and process identity. Use a containing reviewed
|
||||
build, not a version string or an unverified rebuild. Preserve 6.4.5 as the
|
||||
reported source, not proof of the candidate. No new-source acceptance follows
|
||||
from old app/source receipts.
|
||||
- Actual PVE manager/kernel/package identities (reconcile what the reported
|
||||
`9.2.21` identifies), HAOS image and QGA executable/version/configuration.
|
||||
Record filesystem-freeze enablement and every writable filesystem covered by
|
||||
the backup. Map opaque filesystem IDs to these mounts; exclude tmpfs and
|
||||
container overlays that are not actually frozen. Stage the writer through
|
||||
the owner's supported guest-console/image route **before** the backup.
|
||||
- Effective privileges of the *same* Pulse token: Audit and FileRead on the
|
||||
target VM, QGA enabled, ordinary TLS verification, one Pulse process, no other
|
||||
manual QGA probes. A refusal stops the attempt; do not substitute root or
|
||||
broaden access. Never record the credential, full config or service environment.
|
||||
- Preselected observation window (at most ten minutes), backup task identity,
|
||||
Pulse polling/cache intervals, native command-trace coverage and guest/host
|
||||
clock calibration with at most one second relative error. Hash/link original
|
||||
provenance, clock/mount crosswalk and raw task/command traces separately.
|
||||
|
||||
Run a normal freeze-enabled backup with Pulse stopped first as the fixture
|
||||
control, observing the guest independently. Then, on a fresh equivalent
|
||||
disposable fixture, observe normal containing-build polling across a backup.
|
||||
For in-flight acceptance, require a **genuinely observed** Pulse read spanning
|
||||
lock acquisition, not a sleep or a claimed overlap. Never force overlap on a
|
||||
customer guest or queue manual QGA probes. If no overlap occurs within the
|
||||
chosen window, retain that inconclusive attempt; it is not an acceptance pass.
|
||||
No repeated run is prescribed merely to obtain a favourable result.
|
||||
|
||||
## Independent guest filesystem writer
|
||||
|
||||
Build with the declared Go compiler; stage the resulting Linux binary for the
|
||||
actual guest architecture. Python is **not** required in HAOS:
|
||||
|
||||
```sh
|
||||
CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /tmp/guest-writer \
|
||||
./tests/qualification/guest-agent-backup/guest-writer
|
||||
```
|
||||
|
||||
The native owner runs one writer on each mapped, freeze-covered filesystem:
|
||||
|
||||
```sh
|
||||
guest-writer --directory /owner-provisioned/scratch-on-covered-mount \
|
||||
--run-id run-01 --filesystem-id data --duration-seconds 600 --interval-ms 1000
|
||||
```
|
||||
|
||||
The directory must exist, be absolute and have no symlink components. Each run
|
||||
creates a new private child; repeated IDs refuse reuse. Each write uses an
|
||||
exclusive new 0600 file, writes 512 synthetic bytes, fsyncs both file and
|
||||
directory, reads back and checks the bytes, then emits their SHA-256. At most
|
||||
600 such files are left per filesystem; it deletes nothing. Only opaque IDs,
|
||||
fixed outcomes and timestamps reach stdout. Collect stdout/exit through an
|
||||
**independent console channel**, not QGA or the filesystem being frozen.
|
||||
|
||||
One write can be in flight. Heartbeats continue while it is blocked; no further
|
||||
write is queued. Interruption, output failure, readback/write error or a write
|
||||
pending at the deadline cannot pass. The program does not unblock a frozen
|
||||
filesystem. Preserve failed evidence and leave disposal/recovery to the fixture
|
||||
owner; do not turn the failure into a guest restart/reset workaround.
|
||||
|
||||
A successful write witnesses only that mapped filesystem at that instant, not
|
||||
all filesystems, application integrity or a successful backup restore. Require
|
||||
at least two completed baseline writes, liveness during the frozen interval and
|
||||
two **newly started** fsync/readback writes after task completion on every mapped
|
||||
filesystem. A pre-freeze write that finally returns is insufficient on its own.
|
||||
|
||||
## Bounded record checks
|
||||
|
||||
`check.py` reads a redacted JSON record and adjacent hash-bound writer JSONL
|
||||
files. It has no network, subprocess or guest-control capability. The shape is
|
||||
exercised by `record()` and `witness_events()` in `test_check.py`; those are
|
||||
**synthetic checker fixtures, never native evidence**. Fields are exact and
|
||||
extra fields (including private names/raw logs) are rejected.
|
||||
|
||||
The native owner normalises its actual observations into that shape, retaining
|
||||
underlying receipts separately. `identity` binds Pulse/PVE/QGA/HAOS digests;
|
||||
`preflight` and its mount-crosswalk hash cover all declared filesystem IDs.
|
||||
`backup` contains native start, lock, freeze, thaw-*request*, unlock and completed
|
||||
times. A thaw request is not a thaw result. `coverage` binds the complete
|
||||
bounded native command trace with zero dropped events. `commands` are **actual
|
||||
dispatched** reads, not planned or deferred calls; preserve adverse outcomes.
|
||||
The checker requires real in-flight overlap, no concurrent/duplicate dispatch
|
||||
and no dispatch inside the observed lock/clock-uncertainty interval.
|
||||
|
||||
`snapshots` are baseline, at least two locked polls and resumed readbacks of the
|
||||
same VM. Retain metadata/memory/disk payload hashes, guest-agent state, memory
|
||||
source, disk reason, latest memory/disk History times and CPU observation time.
|
||||
Obtain them from the existing read-state, memory diagnostics and History APIs;
|
||||
normalise only the target guest's fixed fields, without raw snapshots in public
|
||||
evidence. Locked values/History cannot be renewed as fresh; CPU must progress.
|
||||
Resumption needs available/agent memory, no disk deferral and fresh post-task
|
||||
memory/disk History. **Internal cache timestamp preservation is source-tested;
|
||||
these public readbacks do not directly inspect those internal timestamps.**
|
||||
|
||||
```sh
|
||||
python3 tests/qualification/guest-agent-backup/check.py \
|
||||
--record /private/new-native-record/record.json --expected-source EXACT_40_HEX_SHA
|
||||
```
|
||||
|
||||
Files are bounded, hash-bound, single-name regular files; symlinks, traversal,
|
||||
duplicate JSON keys, wrong identities, truncation, impossible order, replay,
|
||||
incomplete coverage and adverse witness exits fail closed. Evidence-class and
|
||||
provenance fields are owner observations, **not access grants or independent
|
||||
artifact verification by the checker**. Exit 0 means the supplied record's
|
||||
checks passed. Output deliberately retains `native_acceptance_complete: false`:
|
||||
review original native provenance, mount completeness and command/task traces
|
||||
before judging the job. It does not establish the reported command-ID cause,
|
||||
fleet-wide safety, shipment, restore integrity or release qualification.
|
||||
|
||||
## Source checks and next result
|
||||
|
||||
```sh
|
||||
python3 -m unittest discover -s tests/qualification/guest-agent-backup -v
|
||||
go test -race ./tests/qualification/guest-agent-backup/guest-writer -count=1
|
||||
```
|
||||
|
||||
Use the assigned source-proof VM for worker execution. The real local writer
|
||||
smoke is ordinary synthetic fsync/readback, **not** a filesystem-freeze test.
|
||||
Do not replay unchanged guard/app suites to disguise the missing native target.
|
||||
|
||||
Delivery's next material result is verified native inputs and a completed (or
|
||||
adverse/inconclusive) bounded run: independent writes/liveness, actual command
|
||||
overlap, truthful lock-time readings/History and normal cache resumption on the
|
||||
exact containing build. Coordinator/reviewer/publisher own protected landing;
|
||||
Delivery/steward own its containing all-main release. Community owns useful
|
||||
reporter contact after material evidence. Until native acceptance, preserve
|
||||
stop-Pulse-before-backup and restart-only-after-verified-thaw advice, explicitly
|
||||
including the monitoring/alert outage. An OK task or cleared lock is not enough.
|
||||
@@ -0,0 +1,230 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Check a bounded, redacted native backup record. No network or guest commands.
|
||||
|
||||
Supplied provenance is evidence for independent review, not authority or an
|
||||
independently verified artifact. A successful check is not release qualification.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import re
|
||||
import sys
|
||||
|
||||
MAX_BYTES = 2 * 1024 * 1024
|
||||
ID = re.compile(r"[a-z0-9][a-z0-9-]{0,31}\Z")
|
||||
HEX40 = re.compile(r"[0-9a-f]{40}\Z")
|
||||
HEX64 = re.compile(r"[0-9a-f]{64}\Z")
|
||||
METHODS = {"get-fsinfo", "file-read", "get-osinfo", "info", "network-get-interfaces"}
|
||||
|
||||
|
||||
class InvalidRecord(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def require(ok, message):
|
||||
if not ok:
|
||||
raise InvalidRecord(message)
|
||||
|
||||
|
||||
def shape(value, keys, label):
|
||||
require(type(value) is dict and set(value) == set(keys.split()), label + " shape")
|
||||
|
||||
|
||||
def integer(value):
|
||||
return type(value) is int and 0 <= value < 2**53
|
||||
|
||||
|
||||
def digest(value, pattern=HEX64):
|
||||
return type(value) is str and pattern.fullmatch(value) is not None
|
||||
|
||||
|
||||
def unique_object(pairs):
|
||||
result = {}
|
||||
for key, value in pairs:
|
||||
require(key not in result, "duplicate JSON field")
|
||||
result[key] = value
|
||||
return result
|
||||
|
||||
|
||||
def decode(raw):
|
||||
try:
|
||||
return json.loads(raw, object_pairs_hook=unique_object)
|
||||
except (ValueError, UnicodeError, RecursionError):
|
||||
raise InvalidRecord("invalid JSON") from None
|
||||
|
||||
|
||||
def read_bounded(path):
|
||||
require(not path.is_symlink() and path.is_file(), "regular evidence file required")
|
||||
with path.open("rb") as stream:
|
||||
raw = stream.read(MAX_BYTES + 1)
|
||||
require(0 < len(raw) <= MAX_BYTES, "evidence size bound")
|
||||
return raw
|
||||
|
||||
|
||||
def witness_events(root, entry, run_id, skew):
|
||||
shape(entry, "filesystem_id file sha256 exit_code", "witness")
|
||||
require(type(entry["filesystem_id"]) is str and ID.fullmatch(entry["filesystem_id"]), "filesystem ID")
|
||||
name = entry["file"]
|
||||
require(type(name) is str and re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_.-]{0,95}\.jsonl", name), "witness filename")
|
||||
require(digest(entry["sha256"]), "witness digest")
|
||||
require(type(entry["exit_code"]) is int and entry["exit_code"] == 0, "witness did not exit successfully")
|
||||
raw = read_bounded(root / name)
|
||||
require(hashlib.sha256(raw).hexdigest() == entry["sha256"], "witness content mismatch")
|
||||
require(raw.endswith(b"\n"), "incomplete witness line")
|
||||
lines = raw.splitlines()
|
||||
require(4 <= len(lines) <= 2000 and all(0 < len(line) <= 2048 for line in lines), "witness event bound")
|
||||
events = [decode(line) for line in lines]
|
||||
previous_ms, previous_elapsed, sequence, previous_write_ms = -1, -1, 0, -1
|
||||
for i, event in enumerate(events):
|
||||
kind = event.get("kind") if type(event) is dict else None
|
||||
require(type(kind) is str, "witness event kind")
|
||||
extra = {"start": "", "heartbeat": "", "write": " sequence started_unix_ms sha256", "stop": " result"}.get(kind)
|
||||
require(extra is not None, "witness event kind")
|
||||
shape(event, "schema_version kind run_id filesystem_id unix_ms elapsed_ms" + extra, "witness event")
|
||||
require(type(event["schema_version"]) is int and event["schema_version"] == 1, "witness schema")
|
||||
require(event["run_id"] == run_id and event["filesystem_id"] == entry["filesystem_id"], "witness identity mismatch")
|
||||
at, elapsed = event["unix_ms"], event["elapsed_ms"]
|
||||
require(integer(at) and integer(elapsed) and at >= previous_ms and elapsed >= previous_elapsed, "witness time order")
|
||||
require(elapsed <= 601000, "witness duration bound")
|
||||
if i:
|
||||
drift = (at - events[0]["unix_ms"]) - (elapsed - events[0]["elapsed_ms"])
|
||||
require(abs(drift) <= skew, "witness clock discontinuity")
|
||||
require((kind == "start") == (i == 0) and (kind == "stop") == (i == len(events) - 1), "witness lifecycle")
|
||||
if kind == "write":
|
||||
sequence += 1
|
||||
require(type(event["sequence"]) is int and event["sequence"] == sequence, "witness sequence")
|
||||
require(integer(event["started_unix_ms"]) and
|
||||
max(previous_write_ms, events[0]["unix_ms"]) <= event["started_unix_ms"] <= at, "witness write time")
|
||||
payload = f"Pulse disposable backup witness\n{run_id}\n{entry['filesystem_id']}\n{sequence}\n".encode()
|
||||
payload += b"x" * (512 - len(payload))
|
||||
require(digest(event["sha256"]) and event["sha256"] == hashlib.sha256(payload).hexdigest(), "witness readback digest")
|
||||
previous_write_ms = at
|
||||
if kind == "stop":
|
||||
require(event["result"] == "complete", "witness incomplete or failed")
|
||||
previous_ms, previous_elapsed = at, elapsed
|
||||
return events
|
||||
|
||||
|
||||
def check_record(record, root, expected_source):
|
||||
require(digest(expected_source, HEX40), "expected source SHA")
|
||||
shape(record, "schema_version run_id evidence_class identity preflight window backup coverage commands snapshots witnesses", "record")
|
||||
require(type(record["schema_version"]) is int and record["schema_version"] == 1, "record schema")
|
||||
require(type(record["run_id"]) is str and ID.fullmatch(record["run_id"]), "run ID")
|
||||
require(record["evidence_class"] == "native-disposable", "not a native disposable record")
|
||||
identity = record["identity"]
|
||||
shape(identity, "pulse_source pulse_binary_sha256 pve_packages_sha256 qga_binary_sha256 haos_image_sha256 provenance_receipt_sha256", "identity")
|
||||
require(identity["pulse_source"] == expected_source, "installed source mismatch")
|
||||
require(all(digest(identity[k]) for k in identity if k != "pulse_source"), "input identity digest")
|
||||
preflight = record["preflight"]
|
||||
shape(preflight, "pve_version guest_type audit file_read tls_verified qga_enabled freeze_enabled single_pulse_process filesystem_map_sha256 filesystem_ids", "preflight")
|
||||
require(preflight["pve_version"] == "9.2.21" and preflight["guest_type"] == "HAOS", "reported platform mismatch")
|
||||
require(all(preflight[k] is True for k in ("audit", "file_read", "tls_verified", "qga_enabled", "freeze_enabled", "single_pulse_process")), "preflight incomplete")
|
||||
require(digest(preflight["filesystem_map_sha256"]), "filesystem crosswalk digest")
|
||||
expected_filesystems = preflight["filesystem_ids"]
|
||||
require(type(expected_filesystems) is list and 1 <= len(expected_filesystems) <= 16 and
|
||||
all(type(v) is str and ID.fullmatch(v) for v in expected_filesystems) and
|
||||
len(set(expected_filesystems)) == len(expected_filesystems), "covered filesystem IDs")
|
||||
window = record["window"]
|
||||
shape(window, "start_ms end_ms clock_error_ms", "window")
|
||||
require(all(integer(v) for v in window.values()), "observation bounds")
|
||||
start, end, skew = window["start_ms"], window["end_ms"], window["clock_error_ms"]
|
||||
require(3000 <= end - start <= 600000 and skew <= 1000, "observation duration or clock bound")
|
||||
backup = record["backup"]
|
||||
shape(backup, "start_ms lock_start_ms freeze_ms thaw_request_ms lock_end_ms completed_ms status trace_sha256", "backup")
|
||||
times = [backup[k] for k in ("start_ms", "lock_start_ms", "freeze_ms", "thaw_request_ms", "lock_end_ms", "completed_ms")]
|
||||
require(all(integer(v) for v in times) and times == sorted(times) and start < times[0] < times[-1] < end, "backup time order")
|
||||
require(backup["lock_start_ms"] < backup["lock_end_ms"], "backup lock not observed")
|
||||
require(backup["status"] == "OK" and digest(backup["trace_sha256"]), "backup did not complete normally")
|
||||
coverage = record["coverage"]
|
||||
shape(coverage, "start_ms end_ms dropped_events trace_sha256", "coverage")
|
||||
require(all(integer(coverage[k]) for k in ("start_ms", "end_ms", "dropped_events")) and
|
||||
coverage["start_ms"] <= start and coverage["end_ms"] >= end and coverage["dropped_events"] == 0 and
|
||||
digest(coverage["trace_sha256"]), "native command coverage incomplete")
|
||||
|
||||
commands = record["commands"]
|
||||
require(type(commands) is list and 1 <= len(commands) <= 512, "native command bound")
|
||||
ids, intervals = set(), []
|
||||
overlap = False
|
||||
for command in commands:
|
||||
shape(command, "id method start_ms end_ms result", "command")
|
||||
require(type(command["id"]) is str and ID.fullmatch(command["id"]) and command["id"] not in ids, "duplicate or invalid command ID")
|
||||
ids.add(command["id"])
|
||||
require(type(command["method"]) is str and command["method"] in METHODS and command["result"] == "ok", "native command adverse or unsupported")
|
||||
began, finished = command["start_ms"], command["end_ms"]
|
||||
require(integer(began) and integer(finished) and start <= began < finished <= end, "command time order")
|
||||
require(not backup["lock_start_ms"] - skew <= began <= backup["lock_end_ms"] + skew, "command dispatched during lock or uncertainty")
|
||||
overlap |= began < backup["lock_start_ms"] - skew and finished > backup["lock_start_ms"] + skew
|
||||
intervals.append((began, finished))
|
||||
require(overlap, "no observed in-flight overlap")
|
||||
intervals.sort()
|
||||
require(all(a[1] <= b[0] for a, b in zip(intervals, intervals[1:])), "concurrent guest commands")
|
||||
|
||||
snapshots = record["snapshots"]
|
||||
require(type(snapshots) is list and 4 <= len(snapshots) <= 64, "monitoring snapshot bound")
|
||||
phases = [s.get("phase") if type(s) is dict else None for s in snapshots]
|
||||
require(phases == ["before"] + ["locked"] * (len(snapshots) - 2) + ["resumed"], "monitoring phases")
|
||||
for snapshot in snapshots:
|
||||
shape(snapshot, "phase at_ms guest_identity_sha256 guest_status metadata_sha256 memory_sha256 disk_sha256 memory_source disk_reason disk_history_latest_ms memory_history_latest_ms cpu_latest_ms", "snapshot")
|
||||
require(integer(snapshot["at_ms"]), "snapshot time")
|
||||
require(all(digest(snapshot[k]) for k in ("guest_identity_sha256", "metadata_sha256", "memory_sha256", "disk_sha256")), "snapshot digest")
|
||||
require(all(integer(snapshot[k]) and start <= snapshot[k] <= snapshot["at_ms"] <= end for k in
|
||||
("at_ms", "disk_history_latest_ms", "memory_history_latest_ms", "cpu_latest_ms")), "snapshot times")
|
||||
require(all(a["at_ms"] < b["at_ms"] for a, b in zip(snapshots, snapshots[1:])), "snapshot time order")
|
||||
before, resumed = snapshots[0], snapshots[-1]
|
||||
require(before["at_ms"] < backup["start_ms"] - skew and before["guest_status"] == "available" and
|
||||
before["memory_source"] == "guest-agent-meminfo" and before["disk_reason"] == "", "baseline guest truth missing")
|
||||
stable = ("guest_identity_sha256", "metadata_sha256", "memory_sha256", "disk_sha256", "disk_history_latest_ms", "memory_history_latest_ms")
|
||||
for locked in snapshots[1:-1]:
|
||||
require(backup["lock_start_ms"] + skew < locked["at_ms"] < backup["lock_end_ms"] - skew,
|
||||
"locked snapshot outside lock")
|
||||
require(locked["guest_status"] == "deferred" and locked["memory_source"] == "previous-snapshot" and
|
||||
locked["disk_reason"] == "prev-vm-locked", "deferral not visible")
|
||||
require(all(locked[k] == before[k] for k in stable), "retained evidence renewed or changed during lock")
|
||||
require(locked["cpu_latest_ms"] > before["cpu_latest_ms"], "non-QGA monitoring did not progress")
|
||||
require(resumed["guest_identity_sha256"] == before["guest_identity_sha256"] and resumed["guest_status"] == "available" and
|
||||
resumed["memory_source"] == "guest-agent-meminfo" and resumed["disk_reason"] == "", "guest did not resume")
|
||||
require(all(resumed[k] > backup["completed_ms"] + skew for k in ("disk_history_latest_ms", "memory_history_latest_ms", "cpu_latest_ms")),
|
||||
"fresh post-backup History absent")
|
||||
|
||||
witnesses = record["witnesses"]
|
||||
require(type(witnesses) is list and 1 <= len(witnesses) <= 16, "filesystem witness bound")
|
||||
filesystems = set()
|
||||
for witness in witnesses:
|
||||
events = witness_events(root, witness, record["run_id"], skew)
|
||||
require(witness["filesystem_id"] not in filesystems, "duplicate filesystem witness")
|
||||
filesystems.add(witness["filesystem_id"])
|
||||
require(events[0]["unix_ms"] < backup["start_ms"] - skew and end - skew <= events[-1]["unix_ms"] <= end + skew,
|
||||
"witness did not cover the observation window")
|
||||
writes = [e for e in events if e["kind"] == "write"]
|
||||
require(sum(e["unix_ms"] < backup["start_ms"] - skew for e in writes) >= 2, "baseline filesystem writes missing")
|
||||
require(sum(backup["completed_ms"] + skew < e["started_unix_ms"] <= e["unix_ms"] <= end + skew for e in writes) >= 2,
|
||||
"independent post-task fsync/readback missing")
|
||||
require(any(e["kind"] == "heartbeat" and backup["freeze_ms"] + skew < e["unix_ms"] < backup["thaw_request_ms"] - skew for e in events),
|
||||
"guest liveness during freeze missing")
|
||||
require(filesystems == set(expected_filesystems), "covered filesystem witness missing")
|
||||
return {"record_checks_passed": True, "native_acceptance_complete": False,
|
||||
"source": expected_source, "filesystem_witnesses": len(filesystems),
|
||||
"limits": "Review native provenance, covered-filesystem completeness and raw traces independently. This record does not reproduce the reported cause, certify all guests, establish delivery or qualify a release."}
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--record", type=Path, required=True)
|
||||
parser.add_argument("--expected-source", required=True)
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
result = check_record(decode(read_bounded(args.record)), args.record.parent, args.expected_source)
|
||||
except (InvalidRecord, OSError, TypeError, ValueError, RecursionError) as error:
|
||||
# The record may contain private or malformed input. Do not echo it or
|
||||
# exception paths. Detailed receipts remain with the native owner.
|
||||
failure = str(error) if isinstance(error, InvalidRecord) else "evidence read or schema failure"
|
||||
print(json.dumps({"record_checks_passed": False, "native_acceptance_complete": False, "failure": failure}), file=sys.stderr)
|
||||
return 1
|
||||
print(json.dumps(result, sort_keys=True))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,206 @@
|
||||
// guest-writer observes a disposable guest filesystem without using QGA.
|
||||
// It never starts a backup, freezes/thaws a filesystem or restarts a service.
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"os/signal"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"syscall"
|
||||
"time"
|
||||
)
|
||||
|
||||
var opaqueID = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,31}$`)
|
||||
|
||||
type options struct {
|
||||
directory, runID, filesystemID string
|
||||
duration, interval time.Duration
|
||||
}
|
||||
|
||||
func (o options) validate() error {
|
||||
if !opaqueID.MatchString(o.runID) || !opaqueID.MatchString(o.filesystemID) {
|
||||
return errors.New("opaque run and filesystem IDs required")
|
||||
}
|
||||
if o.duration < 3*time.Second || o.duration > 10*time.Minute || o.interval < time.Second || o.interval > 5*time.Second {
|
||||
return errors.New("duration must be 3..600 seconds and interval 1000..5000 milliseconds")
|
||||
}
|
||||
if !filepath.IsAbs(o.directory) || filepath.Clean(o.directory) != o.directory {
|
||||
return errors.New("absolute clean scratch directory required")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// The owner must map this directory to a filesystem actually covered by the
|
||||
// native backup. A tmpfs or container overlay is not an equivalent witness.
|
||||
func prepareWriter(o options) (func(int) (string, error), error) {
|
||||
if err := o.validate(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resolved, err := filepath.EvalSymlinks(o.directory)
|
||||
if err != nil || resolved != o.directory {
|
||||
return nil, errors.New("scratch directory unavailable or symlinked")
|
||||
}
|
||||
info, err := os.Stat(o.directory)
|
||||
if err != nil || !info.IsDir() {
|
||||
return nil, errors.New("scratch directory unavailable")
|
||||
}
|
||||
dir := filepath.Join(o.directory, "pulse-backup-witness-"+o.runID+"-"+o.filesystemID)
|
||||
// Never reuse a run or overwrite any pre-existing file. Retain synthetic
|
||||
// files on both success and failure; fixture teardown belongs to its owner.
|
||||
if err := os.Mkdir(dir, 0700); err != nil {
|
||||
return nil, errors.New("new witness directory could not be created")
|
||||
}
|
||||
return func(seq int) (string, error) {
|
||||
payload := []byte(fmt.Sprintf("Pulse disposable backup witness\n%s\n%s\n%d\n", o.runID, o.filesystemID, seq))
|
||||
payload = append(payload, bytes.Repeat([]byte{'x'}, 512-len(payload))...)
|
||||
path := filepath.Join(dir, fmt.Sprintf("witness-%06d", seq))
|
||||
f, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0600)
|
||||
if err != nil {
|
||||
return "", errors.New("witness write failed")
|
||||
}
|
||||
_, writeErr := f.Write(payload)
|
||||
syncErr := f.Sync()
|
||||
closeErr := f.Close()
|
||||
if errors.Join(writeErr, syncErr, closeErr) != nil {
|
||||
return "", errors.New("witness write failed")
|
||||
}
|
||||
d, err := os.Open(dir)
|
||||
if err != nil {
|
||||
return "", errors.New("witness directory sync failed")
|
||||
}
|
||||
syncErr, closeErr = d.Sync(), d.Close()
|
||||
if errors.Join(syncErr, closeErr) != nil {
|
||||
return "", errors.New("witness directory sync failed")
|
||||
}
|
||||
readback, err := os.ReadFile(path)
|
||||
if err != nil || !bytes.Equal(readback, payload) {
|
||||
return "", errors.New("witness readback failed")
|
||||
}
|
||||
digest := sha256.Sum256(readback)
|
||||
return hex.EncodeToString(digest[:]), nil
|
||||
}, nil
|
||||
}
|
||||
|
||||
type event struct {
|
||||
SchemaVersion int `json:"schema_version"`
|
||||
Kind string `json:"kind"`
|
||||
RunID string `json:"run_id"`
|
||||
FilesystemID string `json:"filesystem_id"`
|
||||
UnixMS int64 `json:"unix_ms"`
|
||||
ElapsedMS int64 `json:"elapsed_ms"`
|
||||
Sequence int `json:"sequence,omitempty"`
|
||||
StartedUnixMS int64 `json:"started_unix_ms,omitempty"`
|
||||
SHA256 string `json:"sha256,omitempty"`
|
||||
Result string `json:"result,omitempty"`
|
||||
}
|
||||
|
||||
type writeResult struct {
|
||||
sequence int
|
||||
started time.Time
|
||||
digest string
|
||||
err error
|
||||
}
|
||||
|
||||
func observe(ctx context.Context, o options, output io.Writer, write func(int) (string, error)) error {
|
||||
start := time.Now()
|
||||
deadline := start.Add(o.duration)
|
||||
encoder := json.NewEncoder(output)
|
||||
emit := func(e event) error {
|
||||
now := time.Now()
|
||||
e.SchemaVersion, e.RunID, e.FilesystemID = 1, o.runID, o.filesystemID
|
||||
e.UnixMS, e.ElapsedMS = now.UnixMilli(), now.Sub(start).Milliseconds()
|
||||
return encoder.Encode(e)
|
||||
}
|
||||
if err := emit(event{Kind: "start"}); err != nil {
|
||||
return err
|
||||
}
|
||||
timer := time.NewTimer(o.duration)
|
||||
defer timer.Stop()
|
||||
ticker := time.NewTicker(o.interval)
|
||||
defer ticker.Stop()
|
||||
results := make(chan writeResult, 1)
|
||||
sequence, pending := 0, false
|
||||
begin := func() {
|
||||
sequence++
|
||||
pending = true
|
||||
seq, began := sequence, time.Now()
|
||||
go func() {
|
||||
digest, err := write(seq)
|
||||
results <- writeResult{seq, began, digest, err}
|
||||
}()
|
||||
}
|
||||
stop := func(result string) error { return emit(event{Kind: "stop", Result: result}) }
|
||||
if ctx.Err() == nil {
|
||||
begin()
|
||||
}
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
_ = stop("interrupted")
|
||||
return errors.New("witness interrupted")
|
||||
case <-timer.C:
|
||||
if pending {
|
||||
_ = stop("write-pending")
|
||||
return errors.New("witness deadline with write pending")
|
||||
}
|
||||
return stop("complete")
|
||||
case <-ticker.C:
|
||||
if err := emit(event{Kind: "heartbeat"}); err != nil {
|
||||
return err
|
||||
}
|
||||
// A blocked filesystem write must not queue more work or prevent
|
||||
// independent guest liveness observations on stdout.
|
||||
if !pending && time.Now().Before(deadline) {
|
||||
begin()
|
||||
}
|
||||
case result := <-results:
|
||||
pending = false
|
||||
if result.err != nil {
|
||||
_ = stop("write-failed")
|
||||
return errors.New("witness write failed")
|
||||
}
|
||||
if err := emit(event{Kind: "write", Sequence: result.sequence, StartedUnixMS: result.started.UnixMilli(), SHA256: result.digest}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func main() {
|
||||
var o options
|
||||
var durationSeconds, intervalMS int
|
||||
flag.StringVar(&o.directory, "directory", "", "owner-provisioned scratch directory on the covered filesystem")
|
||||
flag.StringVar(&o.runID, "run-id", "", "opaque run ID; no names or paths")
|
||||
flag.StringVar(&o.filesystemID, "filesystem-id", "", "opaque filesystem ID from the private mount crosswalk")
|
||||
flag.IntVar(&durationSeconds, "duration-seconds", 600, "preselected observation duration, 3..600")
|
||||
flag.IntVar(&intervalMS, "interval-ms", 1000, "synthetic write/heartbeat interval, 1000..5000")
|
||||
flag.Parse()
|
||||
if flag.NArg() != 0 || durationSeconds < 3 || durationSeconds > 600 || intervalMS < 1000 || intervalMS > 5000 {
|
||||
fmt.Fprintln(os.Stderr, "invalid witness arguments")
|
||||
os.Exit(2)
|
||||
}
|
||||
o.duration, o.interval = time.Duration(durationSeconds)*time.Second, time.Duration(intervalMS)*time.Millisecond
|
||||
write, err := prepareWriter(o)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(2)
|
||||
}
|
||||
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer cancel()
|
||||
if err := observe(ctx, o, os.Stdout, write); err != nil {
|
||||
// Fixed errors only: never emit the private filesystem path or a
|
||||
// provider/service environment. An output error is still a failure.
|
||||
fmt.Fprintln(os.Stderr, "witness observation failed")
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,220 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func testOptions(t *testing.T) options {
|
||||
t.Helper()
|
||||
return options{directory: t.TempDir(), runID: "run-01", filesystemID: "data", duration: 3 * time.Second, interval: time.Second}
|
||||
}
|
||||
|
||||
func events(t *testing.T, output *bytes.Buffer) []event {
|
||||
t.Helper()
|
||||
var result []event
|
||||
decoder := json.NewDecoder(output)
|
||||
for {
|
||||
var e event
|
||||
if err := decoder.Decode(&e); err == io.EOF {
|
||||
break
|
||||
} else if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
result = append(result, e)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func TestWitnessWritesSyncsAndReadsOnlyNewSyntheticFiles(t *testing.T) {
|
||||
o := testOptions(t)
|
||||
untouched := filepath.Join(o.directory, "existing")
|
||||
if err := os.WriteFile(untouched, []byte("do not overwrite"), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
write, err := prepareWriter(o)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
dir := filepath.Join(o.directory, "pulse-backup-witness-run-01-data")
|
||||
for seq := 1; seq <= 2; seq++ {
|
||||
digest, err := write(seq)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
path := filepath.Join(dir, fmt.Sprintf("witness-%06d", seq))
|
||||
payload, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
hash := sha256.Sum256(payload)
|
||||
if len(payload) != 512 || digest != hex.EncodeToString(hash[:]) || !bytes.Contains(payload, []byte("\nrun-01\ndata\n")) {
|
||||
t.Fatal("witness did not read back the new synthetic payload")
|
||||
}
|
||||
info, err := os.Stat(path)
|
||||
if err != nil || info.Mode().Perm() != 0600 {
|
||||
t.Fatal("witness file permissions")
|
||||
}
|
||||
}
|
||||
if _, err := write(1); err == nil {
|
||||
t.Fatal("repeated sequence overwrote existing evidence")
|
||||
}
|
||||
if _, err := prepareWriter(o); err == nil {
|
||||
t.Fatal("repeated run reused existing evidence")
|
||||
}
|
||||
data, err := os.ReadFile(untouched)
|
||||
if err != nil || string(data) != "do not overwrite" {
|
||||
t.Fatal("unrelated file changed")
|
||||
}
|
||||
info, err := os.Stat(dir)
|
||||
if err != nil || info.Mode().Perm() != 0700 {
|
||||
t.Fatal("witness directory permissions")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWitnessRejectsInvalidInputsBeforeCreatingFiles(t *testing.T) {
|
||||
changes := map[string]func(*options){
|
||||
"relative": func(o *options) { o.directory = "relative" },
|
||||
"run traversal": func(o *options) { o.runID = "../elsewhere" },
|
||||
"filesystem traversal": func(o *options) { o.filesystemID = "../elsewhere" },
|
||||
"oversize ID": func(o *options) { o.runID = strings.Repeat("a", 33) },
|
||||
"duration small": func(o *options) { o.duration = time.Second },
|
||||
"duration large": func(o *options) { o.duration = 11 * time.Minute },
|
||||
"interval small": func(o *options) { o.interval = time.Millisecond },
|
||||
"interval large": func(o *options) { o.interval = 6 * time.Second },
|
||||
}
|
||||
for name, change := range changes {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
o := testOptions(t)
|
||||
original := o.directory
|
||||
change(&o)
|
||||
if _, err := prepareWriter(o); err == nil {
|
||||
t.Fatal("unsafe or unbounded input accepted")
|
||||
}
|
||||
entries, err := os.ReadDir(original)
|
||||
if err != nil || len(entries) != 0 {
|
||||
t.Fatal("invalid input created files")
|
||||
}
|
||||
})
|
||||
}
|
||||
for _, parent := range []bool{false, true} {
|
||||
t.Run("symlink", func(t *testing.T) {
|
||||
o := testOptions(t)
|
||||
real := t.TempDir()
|
||||
link := filepath.Join(o.directory, "link")
|
||||
if err := os.Symlink(real, link); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
o.directory = link
|
||||
if parent {
|
||||
if err := os.Mkdir(filepath.Join(real, "child"), 0700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
o.directory = filepath.Join(link, "child")
|
||||
}
|
||||
if _, err := prepareWriter(o); err == nil {
|
||||
t.Fatal("symlinked scratch path accepted")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestObserverCompletesWithOrderedWriteAndLivenessEvidence(t *testing.T) {
|
||||
o := testOptions(t)
|
||||
// Short timers exercise the event loop, not the CLI's native bounds.
|
||||
o.duration, o.interval = 250*time.Millisecond, 30*time.Millisecond
|
||||
var output bytes.Buffer
|
||||
if err := observe(context.Background(), o, &output, func(int) (string, error) { return strings.Repeat("a", 64), nil }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
seen := events(t, &output)
|
||||
if len(seen) < 4 || seen[0].Kind != "start" || seen[len(seen)-1].Result != "complete" {
|
||||
t.Fatalf("incomplete lifecycle: %#v", seen)
|
||||
}
|
||||
sequence := 0
|
||||
for i, event := range seen {
|
||||
if event.SchemaVersion != 1 || event.RunID != o.runID || event.FilesystemID != o.filesystemID || (i > 0 && event.ElapsedMS < seen[i-1].ElapsedMS) {
|
||||
t.Fatal("event identity or order")
|
||||
}
|
||||
if event.Kind == "write" {
|
||||
sequence++
|
||||
if event.Sequence != sequence || event.StartedUnixMS > event.UnixMS || event.SHA256 == "" {
|
||||
t.Fatal("write evidence incomplete")
|
||||
}
|
||||
}
|
||||
}
|
||||
if sequence == 0 {
|
||||
t.Fatal("no writes observed")
|
||||
}
|
||||
}
|
||||
|
||||
func TestObserverBlockedWriteDoesNotQueueMoreAndCannotPass(t *testing.T) {
|
||||
o := testOptions(t)
|
||||
o.duration, o.interval = 250*time.Millisecond, 30*time.Millisecond
|
||||
var output bytes.Buffer
|
||||
var calls atomic.Int32
|
||||
release := make(chan struct{})
|
||||
defer close(release)
|
||||
err := observe(context.Background(), o, &output, func(int) (string, error) {
|
||||
calls.Add(1)
|
||||
<-release
|
||||
return "", nil
|
||||
})
|
||||
if err == nil || calls.Load() != 1 {
|
||||
t.Fatal("blocked write passed or queued more work")
|
||||
}
|
||||
seen := events(t, &output)
|
||||
if len(seen) < 3 || seen[len(seen)-1].Result != "write-pending" {
|
||||
t.Fatal("blocked write erased liveness or deadline failure")
|
||||
}
|
||||
for _, event := range seen {
|
||||
if event.Kind == "write" {
|
||||
t.Fatal("uncompleted write reported successful")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestObserverFailureAndInterruptionRemainAdverse(t *testing.T) {
|
||||
o := testOptions(t)
|
||||
var output bytes.Buffer
|
||||
if err := observe(context.Background(), o, &output, func(int) (string, error) { return "", errors.New("private path") }); err == nil {
|
||||
t.Fatal("failed write passed")
|
||||
}
|
||||
if strings.Contains(output.String(), "private path") || events(t, &output)[1].Result != "write-failed" {
|
||||
t.Fatal("failure missing or private error disclosed")
|
||||
}
|
||||
output.Reset()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
called := false
|
||||
if err := observe(ctx, o, &output, func(int) (string, error) { called = true; return "", nil }); err == nil || called {
|
||||
t.Fatal("cancelled run started writes or passed")
|
||||
}
|
||||
if events(t, &output)[1].Result != "interrupted" {
|
||||
t.Fatal("interruption not retained")
|
||||
}
|
||||
}
|
||||
|
||||
type failedOutput struct{}
|
||||
|
||||
func (failedOutput) Write([]byte) (int, error) { return 0, errors.New("private output path") }
|
||||
|
||||
func TestObserverUncollectedOutputFailsBeforeStartingWrites(t *testing.T) {
|
||||
called := false
|
||||
err := observe(context.Background(), testOptions(t), failedOutput{}, func(int) (string, error) { called = true; return "", nil })
|
||||
if err == nil || called {
|
||||
t.Fatal("missing output passed or began writes")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,264 @@
|
||||
"""Synthetic checker controls; none is native PVE/QGA acceptance."""
|
||||
|
||||
import copy
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
SPEC = importlib.util.spec_from_file_location("backup_check", Path(__file__).with_name("check.py"))
|
||||
CHECK = importlib.util.module_from_spec(SPEC)
|
||||
SPEC.loader.exec_module(CHECK)
|
||||
SOURCE = "a" * 40
|
||||
HASH = "b" * 64
|
||||
|
||||
|
||||
def witness_events(filesystem="data"):
|
||||
events = []
|
||||
for kind, at, extra in [
|
||||
("start", 100000, {}),
|
||||
("write", 100020, {"sequence": 1, "started_unix_ms": 100010}),
|
||||
("heartbeat", 101000, {}),
|
||||
("write", 101020, {"sequence": 2, "started_unix_ms": 101010}),
|
||||
("heartbeat", 105000, {}),
|
||||
("write", 108020, {"sequence": 3, "started_unix_ms": 108010}),
|
||||
("write", 109020, {"sequence": 4, "started_unix_ms": 109010}),
|
||||
("stop", 120000, {"result": "complete"}),
|
||||
]:
|
||||
event = {"schema_version": 1, "kind": kind, "run_id": "run-01", "filesystem_id": filesystem,
|
||||
"unix_ms": at, "elapsed_ms": at - 100000, **extra}
|
||||
if kind == "write":
|
||||
payload = f"Pulse disposable backup witness\nrun-01\n{filesystem}\n{event['sequence']}\n".encode()
|
||||
payload += b"x" * (512 - len(payload))
|
||||
event["sha256"] = hashlib.sha256(payload).hexdigest()
|
||||
events.append(event)
|
||||
return events
|
||||
|
||||
|
||||
def record():
|
||||
baseline = {"phase": "before", "at_ms": 102000, "guest_identity_sha256": HASH, "guest_status": "available",
|
||||
"metadata_sha256": HASH, "memory_sha256": HASH, "disk_sha256": HASH,
|
||||
"memory_source": "guest-agent-meminfo", "disk_reason": "", "disk_history_latest_ms": 100020,
|
||||
"memory_history_latest_ms": 100040, "cpu_latest_ms": 100010}
|
||||
locked = {**baseline, "phase": "locked", "at_ms": 104500, "cpu_latest_ms": 104000,
|
||||
"guest_status": "deferred", "memory_source": "previous-snapshot", "disk_reason": "prev-vm-locked"}
|
||||
resumed = {**baseline, "phase": "resumed", "at_ms": 110000, "disk_history_latest_ms": 109500,
|
||||
"memory_history_latest_ms": 109500, "cpu_latest_ms": 109500}
|
||||
return {"schema_version": 1, "run_id": "run-01", "evidence_class": "native-disposable",
|
||||
"identity": {"pulse_source": SOURCE, "pulse_binary_sha256": HASH, "pve_packages_sha256": HASH,
|
||||
"qga_binary_sha256": HASH, "haos_image_sha256": HASH, "provenance_receipt_sha256": HASH},
|
||||
"preflight": {"pve_version": "9.2.21", "guest_type": "HAOS", "audit": True, "file_read": True,
|
||||
"tls_verified": True, "qga_enabled": True, "freeze_enabled": True,
|
||||
"single_pulse_process": True, "filesystem_map_sha256": HASH, "filesystem_ids": ["data"]},
|
||||
"window": {"start_ms": 100000, "end_ms": 120000, "clock_error_ms": 10},
|
||||
"backup": {"start_ms": 103000, "lock_start_ms": 103200, "freeze_ms": 104000, "thaw_request_ms": 106000,
|
||||
"lock_end_ms": 106200, "completed_ms": 107000, "status": "OK", "trace_sha256": HASH},
|
||||
"coverage": {"start_ms": 100000, "end_ms": 120000, "dropped_events": 0, "trace_sha256": HASH},
|
||||
"commands": [{"id": "request-01", "method": "get-fsinfo", "start_ms": 103100,
|
||||
"end_ms": 103500, "result": "ok"}],
|
||||
"snapshots": [baseline, locked, {**locked, "at_ms": 105500, "cpu_latest_ms": 105000}, resumed],
|
||||
"witnesses": []}
|
||||
|
||||
|
||||
class BackupRecordTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.root = Path(self.temp.name)
|
||||
self.record = record()
|
||||
self.add_witness()
|
||||
|
||||
def add_witness(self, filesystem="data", events=None):
|
||||
events = witness_events(filesystem) if events is None else events
|
||||
raw = b"".join(json.dumps(e).encode() + b"\n" for e in events)
|
||||
filename = filesystem + ".jsonl"
|
||||
(self.root / filename).write_bytes(raw)
|
||||
entry = {"filesystem_id": filesystem, "file": filename, "sha256": hashlib.sha256(raw).hexdigest(), "exit_code": 0}
|
||||
self.record["witnesses"] = [w for w in self.record["witnesses"] if w["filesystem_id"] != filesystem] + [entry]
|
||||
|
||||
def check(self):
|
||||
return CHECK.check_record(self.record, self.root, SOURCE)
|
||||
|
||||
def test_complete_record_checks_do_not_claim_native_acceptance(self):
|
||||
result = self.check()
|
||||
self.assertTrue(result["record_checks_passed"])
|
||||
self.assertFalse(result["native_acceptance_complete"])
|
||||
|
||||
def test_every_covered_filesystem_needs_an_independent_witness(self):
|
||||
self.record["preflight"]["filesystem_ids"].append("root")
|
||||
with self.assertRaisesRegex(CHECK.InvalidRecord, "witness missing"):
|
||||
self.check()
|
||||
self.add_witness("root")
|
||||
self.assertEqual(self.check()["filesystem_witnesses"], 2)
|
||||
|
||||
def test_ok_task_without_post_task_writes_is_not_thaw(self):
|
||||
self.add_witness(events=[e for e in witness_events() if e.get("sequence", 0) < 3])
|
||||
with self.assertRaisesRegex(CHECK.InvalidRecord, "post-task fsync"):
|
||||
self.check()
|
||||
|
||||
def test_outstanding_prefreeze_write_is_not_two_fresh_post_task_writes(self):
|
||||
events = witness_events()
|
||||
for event in events:
|
||||
if event.get("sequence") == 3:
|
||||
event["started_unix_ms"] = 102000
|
||||
self.add_witness(events=events)
|
||||
with self.assertRaisesRegex(CHECK.InvalidRecord, "post-task fsync"):
|
||||
self.check()
|
||||
|
||||
def test_wrong_id_timeout_and_incomplete_native_commands_stay_adverse(self):
|
||||
for result in ("wrong-command-id", "timeout", "response-incomplete", "permission-denied"):
|
||||
with self.subTest(result=result):
|
||||
self.record["commands"][0]["result"] = result
|
||||
with self.assertRaisesRegex(CHECK.InvalidRecord, "adverse"):
|
||||
self.check()
|
||||
|
||||
def test_lock_dispatch_concurrency_replay_and_absent_overlap_fail(self):
|
||||
original = copy.deepcopy(self.record)
|
||||
changes = [
|
||||
lambda r: r["commands"][0].update(start_ms=104000, end_ms=104500),
|
||||
lambda r: r["commands"][0].update(start_ms=103200),
|
||||
lambda r: r["commands"][0].update(end_ms=103150),
|
||||
lambda r: r["commands"].append({**r["commands"][0], "id": "another"}),
|
||||
lambda r: r["commands"].append(copy.deepcopy(r["commands"][0])),
|
||||
]
|
||||
for change in changes:
|
||||
self.record = copy.deepcopy(original)
|
||||
change(self.record)
|
||||
with self.subTest(commands=self.record["commands"]):
|
||||
with self.assertRaises(CHECK.InvalidRecord):
|
||||
self.check()
|
||||
|
||||
def test_cache_history_cpu_and_resume_controls(self):
|
||||
original = copy.deepcopy(self.record)
|
||||
changes = [
|
||||
lambda r: r["snapshots"][1].update(disk_history_latest_ms=104000),
|
||||
lambda r: r["snapshots"][1].update(memory_history_latest_ms=104000),
|
||||
lambda r: r["snapshots"][1].update(metadata_sha256="c" * 64),
|
||||
lambda r: r["snapshots"][1].update(memory_sha256="c" * 64),
|
||||
lambda r: r["snapshots"][1].update(disk_sha256="c" * 64),
|
||||
lambda r: r["snapshots"][1].update(cpu_latest_ms=100010),
|
||||
lambda r: r["snapshots"][1].update(guest_status="available"),
|
||||
lambda r: r["snapshots"][1].update(disk_reason=""),
|
||||
lambda r: r["snapshots"][1].update(memory_source="guest-agent-meminfo"),
|
||||
lambda r: r["snapshots"][-1].update(guest_identity_sha256="c" * 64),
|
||||
lambda r: r["snapshots"][-1].update(disk_history_latest_ms=100020),
|
||||
lambda r: r["snapshots"][-1].update(memory_history_latest_ms=100040),
|
||||
lambda r: r["snapshots"][-1].update(guest_status="deferred"),
|
||||
]
|
||||
for change in changes:
|
||||
self.record = copy.deepcopy(original)
|
||||
change(self.record)
|
||||
with self.subTest(snapshots=self.record["snapshots"]):
|
||||
with self.assertRaises(CHECK.InvalidRecord):
|
||||
self.check()
|
||||
|
||||
def test_missing_permissions_identity_trace_and_clock_are_not_acceptance(self):
|
||||
original = copy.deepcopy(self.record)
|
||||
changes = [
|
||||
lambda r: r["identity"].update(pulse_source="c" * 40),
|
||||
lambda r: r["identity"].update(haos_image_sha256="missing"),
|
||||
lambda r: r.update(evidence_class="source-control"),
|
||||
lambda r: r["preflight"].update(pve_version="9.0"),
|
||||
lambda r: r["preflight"].update(guest_type="ordinary-qga-subprocess"),
|
||||
lambda r: r["preflight"].update(audit=False),
|
||||
lambda r: r["preflight"].update(file_read=False),
|
||||
lambda r: r["preflight"].update(tls_verified=False),
|
||||
lambda r: r["preflight"].update(freeze_enabled=False),
|
||||
lambda r: r["preflight"].update(single_pulse_process=False),
|
||||
lambda r: r["coverage"].update(dropped_events=1),
|
||||
lambda r: r["coverage"].update(start_ms=103500),
|
||||
lambda r: r["coverage"].update(end_ms=107000),
|
||||
lambda r: r["window"].update(end_ms=800000),
|
||||
lambda r: r["window"].update(clock_error_ms=2000),
|
||||
lambda r: r["backup"].update(status="failed"),
|
||||
]
|
||||
for change in changes:
|
||||
self.record = copy.deepcopy(original)
|
||||
change(self.record)
|
||||
with self.subTest(record=self.record):
|
||||
with self.assertRaises(CHECK.InvalidRecord):
|
||||
self.check()
|
||||
|
||||
def test_witness_corruption_replay_missing_liveness_and_early_exit_fail(self):
|
||||
original = witness_events()
|
||||
changes = [
|
||||
lambda events: events[-1].update(result="write-pending"),
|
||||
lambda events: events[-1].update(result="interrupted"),
|
||||
lambda events: events[-1].update(unix_ms=110000, elapsed_ms=10000),
|
||||
lambda events: events[1].update(sha256=HASH),
|
||||
lambda events: events[5].update(sequence=1),
|
||||
lambda events: events[1].update(run_id="other-run"),
|
||||
lambda events: events[1].update(filesystem_id="other-fs"),
|
||||
lambda events: events[4].update(unix_ms=104000, elapsed_ms=4000),
|
||||
lambda events: events[4].update(elapsed_ms=4000),
|
||||
lambda events: events[1].update(kind=[]),
|
||||
lambda events: events[1].update(private_value="must not be accepted"),
|
||||
]
|
||||
for change in changes:
|
||||
events = copy.deepcopy(original)
|
||||
change(events)
|
||||
self.add_witness(events=events)
|
||||
with self.subTest(events=events):
|
||||
with self.assertRaises(CHECK.InvalidRecord):
|
||||
self.check()
|
||||
|
||||
def test_missing_symlinked_oversized_truncated_and_changed_witness_fail(self):
|
||||
path = self.root / "data.jsonl"
|
||||
original = path.read_bytes()
|
||||
for raw in (original[:-1], original + b"corrupted\n", b"x" * (CHECK.MAX_BYTES + 1)):
|
||||
path.write_bytes(raw)
|
||||
self.record["witnesses"][0]["sha256"] = hashlib.sha256(raw).hexdigest()
|
||||
with self.assertRaises(CHECK.InvalidRecord):
|
||||
self.check()
|
||||
self.add_witness()
|
||||
self.record["witnesses"][0]["sha256"] = HASH
|
||||
with self.assertRaisesRegex(CHECK.InvalidRecord, "mismatch"):
|
||||
self.check()
|
||||
self.add_witness()
|
||||
target = self.root / "retained.jsonl"
|
||||
path.rename(target)
|
||||
path.symlink_to(target.name)
|
||||
with self.assertRaisesRegex(CHECK.InvalidRecord, "regular"):
|
||||
self.check()
|
||||
path.unlink()
|
||||
with self.assertRaisesRegex(CHECK.InvalidRecord, "regular"):
|
||||
self.check()
|
||||
|
||||
def test_traversal_duplicate_filesystem_boolean_times_and_extra_fields_fail(self):
|
||||
original = copy.deepcopy(self.record)
|
||||
changes = [
|
||||
lambda r: r["witnesses"][0].update(file="../data.jsonl"),
|
||||
lambda r: r["witnesses"][0].update(exit_code=True),
|
||||
lambda r: r["witnesses"].append(copy.deepcopy(r["witnesses"][0])),
|
||||
lambda r: r["window"].update(start_ms=True),
|
||||
lambda r: r["snapshots"][0].update(at_ms=False),
|
||||
lambda r: r["commands"][0].update(method=[]),
|
||||
lambda r: r.update(private_names="do not accept private fields"),
|
||||
]
|
||||
for change in changes:
|
||||
self.record = copy.deepcopy(original)
|
||||
change(self.record)
|
||||
with self.assertRaises(CHECK.InvalidRecord):
|
||||
self.check()
|
||||
|
||||
def test_duplicate_json_keys_fail(self):
|
||||
with self.assertRaisesRegex(CHECK.InvalidRecord, "duplicate"):
|
||||
CHECK.decode(b'{"status":"OK","status":"failed"}')
|
||||
|
||||
def test_cli_returns_sanitised_failures_without_echoing_private_input(self):
|
||||
path = self.root / "record.json"
|
||||
self.record["snapshots"][0]["at_ms"] = "private-value-must-not-escape"
|
||||
path.write_text(json.dumps(self.record))
|
||||
result = subprocess.run([sys.executable, str(Path(__file__).with_name("check.py")),
|
||||
"--record", str(path), "--expected-source", SOURCE], capture_output=True, text=True)
|
||||
self.assertEqual(result.returncode, 1)
|
||||
self.assertNotIn("private-value-must-not-escape", result.stdout + result.stderr)
|
||||
self.assertEqual(json.loads(result.stderr)["failure"], "snapshot time")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user