perf(delivery): reuse verified bundle reads during node startup

This commit is contained in:
2026-10-22 02:01:57 +08:00
parent cc8f4217db
commit 0cd94a20ec
8 changed files with 550 additions and 11 deletions
+19 -3
View File
@@ -429,9 +429,25 @@ and independent destination byte verification remain required.
Public bundle inventories read every payload byte with at most four readers in
flight. Each reader opens without following a final symlink and verifies the
file identity and metadata before and after hashing. The inventory checks those
identities again before returning. All readers are joined on success or failure;
separate processes still obtain their own complete byte proofs. Normal service
and user-program limits, startup deadlines and the recovery target are unchanged.
identities again before returning. All readers are joined on success or failure.
During one `start`, `resume` or `restart`, the fixed Root maintenance helper reads
the complete bundle once. Its Root startup checks and node-health helpers can
reuse that proof while the same transient maintenance invocation is running.
A Root-only systemd runtime directory holds a record of at most 4096 bytes,
binding the full metadata digest to the selected manifest, release, configuration,
boot, MainPID and InvocationID. Each consumer checks the live owner and every
file/directory identity and timestamp before and after use. An eligible malformed,
expired or changed record refuses startup; a missing record requires full bytes.
Inherited proofs cannot be re-exported or retained in a process-local scope.
The maintenance helper removes its exact record before success, and systemd
removes the runtime directory on unit exit, including a killed helper. A final
verification or cleanup failure closes admission. Outside a live maintenance
operation, standalone checks read every byte. Sandbox image verification and
execution isolation remain required
for every actual request. Each consuming process records one fixed journal event
with the manifest digest and maintenance invocation. Normal service and user-program
limits, startup deadlines and the recovery target are unchanged.
PG and MQ start together within the two-CPU node ceiling, then Root starts on its
own before the ordinary service pairs. Root keeps its original preparation limit.
Verified installer and stopped-node recovery inventories can keep at most four
+26 -7
View File
@@ -19,6 +19,7 @@ from . import (
factory_service,
install,
node_health,
operation_proof,
operator_units,
preflight,
proxy_install,
@@ -565,16 +566,31 @@ class Maintenance:
from .recovery_database import admission_guard
with admission_guard():
self.start_locked(resume=resume)
if not resume:
# Preserve the ordinary refusal of start on an active node,
# before publishing a proof or changing its admission state.
self.require_fresh_start()
try:
availability.write(self.selected, "draining")
with operation_proof.publish(self.selected, self.payload.parent):
self.start_locked(resume=resume)
except (OSError, ValueError, subprocess.SubprocessError):
# A final proof/cleanup failure can occur after readiness opened
# admission. Close it before the caller records this failure.
self.close_failed_admission()
raise
def require_fresh_start(self) -> None:
for name in APPS:
value = self.unit(name)
if value["ActiveState"] != "inactive" or value["MainPID"] != "0":
raise DeliveryError("fresh start requires stopped application units")
self.empty_cgroup(name)
self.root_stopped()
def start_locked(self, *, resume: bool) -> None:
if not resume:
for name in APPS:
value = self.unit(name)
if value["ActiveState"] != "inactive" or value["MainPID"] != "0":
raise DeliveryError("fresh start requires stopped application units")
self.empty_cgroup(name)
self.root_stopped()
self.require_fresh_start()
availability.write(self.selected, "draining")
for group in START_GROUPS:
self.start_group(group)
@@ -605,6 +621,8 @@ def run(action: str) -> dict:
"recovery_identity.py",
"restore.py",
"dependency_initialize.py",
"manifest.py",
"operation_proof.py",
):
shipped = payload / "delivery/scripts/release" / name
preflight.protected(shipped)
@@ -643,6 +661,7 @@ def run(action: str) -> dict:
maintenance.record["success"] = True
maintenance.save("complete")
except (OSError, ValueError, subprocess.SubprocessError) as error:
maintenance.record["success"] = False
maintenance.record["failure_code"] = "maintenance_unconfirmed"
# Publication errors are not hidden by a successful-looking receipt.
maintenance.save("failed")
+7
View File
@@ -372,7 +372,14 @@ def validate(data: dict) -> dict:
def verify(bundle: Path, expected_sha256: str) -> dict:
"""A bounded operation may reuse a complete byte proof only while unchanged."""
from .operation_proof import reuse
expected_sha256 = sha256(expected_sha256)
# Inherited proofs are revalidated at every call and never remembered in a
# same-PID scope. They cannot outlive or be exported beyond their owner.
inherited = reuse(bundle, expected_sha256)
if inherited is not None:
return inherited
scope = _READ_SCOPE.get()
if scope is None or scope.pid != os.getpid():
return _verify_contents(bundle, expected_sha256)
+2
View File
@@ -87,6 +87,8 @@ def binding() -> tuple[dict, dict, Path]:
"service_profiles.py",
"outbox_health.py",
"journal.py",
"manifest.py",
"operation_proof.py",
"proxy_service.py",
"proxy_install.py",
"proxy_guard.py",
+285
View File
@@ -0,0 +1,285 @@
"""Complete bundle byte proofs limited to one live Root maintenance invocation."""
import hashlib
import os
import re
import stat
import time
from contextlib import contextmanager
from pathlib import Path
from . import preflight
from .policy import DeliveryError, canonical, digest, fields, integer, read_json, sha256
DIRECTORY = Path("/run/kf-node-verification")
RECORD = DIRECTORY / "byte-proof.json"
PARENT = "kf-node-maintenance.service"
GROUP = "/sys/fs/cgroup/kfonlinejudge.slice/"
CONSUMERS = frozenset(
"kf-" + name + ".service"
for name in (
"postgres",
"rabbitmq",
"sandbox-factory",
"coordinator",
"api",
"outbox",
"problem-import",
"worker",
"proxy",
"node-health",
)
)
MAX_AGE_NS = 900 * 10**9
MAX_RECORD = 4096
_REPORTED: tuple[int, str] | None = None
def _root() -> bool:
return (os.getuid(), os.geteuid(), os.getgid(), os.getegid()) == (
0,
0,
0,
0,
) and set(os.getgroups()) <= {0}
def _now() -> int:
return time.clock_gettime_ns(time.CLOCK_BOOTTIME)
def _boot() -> str:
return Path("/proc/sys/kernel/random/boot_id").read_text().strip()
def _signature(value: dict) -> str:
return hashlib.sha256(canonical(value)).hexdigest()
def _manager(release: str, configuration: str) -> dict:
from . import service_install
expected = {
"LoadState": "loaded",
"ActiveState": "active",
"SubState": "running",
"Transient": "yes",
"ControlGroup": "/kfonlinejudge.slice/" + PARENT,
"Description": "KF fixed node-maintenance " + release + "." + configuration,
"User": "root",
"Group": "root",
"MemoryMax": "268435456",
"MemorySwapMax": "0",
"TasksMax": "32",
"CPUQuotaPerSecUSec": "500ms",
"RuntimeMaxUSec": "15min",
"NoNewPrivileges": "yes",
"CapabilityBoundingSet": "",
"AmbientCapabilities": "",
"Restart": "no",
"KillMode": "control-group",
"RuntimeDirectory": DIRECTORY.name,
"RuntimeDirectoryMode": "0700",
}
names = [*expected, "MainPID", "InvocationID"]
observed = dict(
line.split("=", 1)
for line in service_install.systemctl(
"show", PARENT, "--property=" + ",".join(names)
).splitlines()
if "=" in line
)
if (
set(observed) != set(names)
or any(observed[key] != value for key, value in expected.items())
or not re.fullmatch(r"[1-9][0-9]{0,9}", observed.get("MainPID", ""))
or not re.fullmatch(r"[0-9a-f]{32}", observed.get("InvocationID", ""))
or observed["InvocationID"] == "0" * 32
):
raise DeliveryError("bundle proof maintenance invocation unavailable")
return {"pid": int(observed["MainPID"]), "invocation": observed["InvocationID"]}
def _read() -> tuple[dict, tuple[int, ...]]:
preflight.protected(DIRECTORY, directory=True)
directory = DIRECTORY.lstat()
if directory.st_gid != 0 or stat.S_IMODE(directory.st_mode) != 0o700:
raise DeliveryError("private maintenance proof directory required")
preflight.protected(RECORD)
before = RECORD.lstat()
if before.st_gid != 0 or stat.S_IMODE(before.st_mode) != 0o400:
raise DeliveryError("private maintenance proof file required")
value = read_json(RECORD, MAX_RECORD)
after = RECORD.lstat()
if (before.st_dev, before.st_ino, before.st_ctime_ns) != (
after.st_dev,
after.st_ino,
after.st_ctime_ns,
):
raise DeliveryError("maintenance proof changed while reading")
validate(value)
return value, (
before.st_dev,
before.st_ino,
before.st_mode,
before.st_nlink,
before.st_uid,
before.st_gid,
before.st_size,
before.st_mtime_ns,
before.st_ctime_ns,
)
def validate(value: dict) -> dict:
fields(
value,
{
"schema_version",
"bundle",
"manifest_sha256",
"release_id",
"configuration_id",
"metadata_sha256",
"boot_id",
"issued_boottime_ns",
"pid",
"invocation",
},
"maintenance bundle proof",
)
integer(value["schema_version"], 1, 1, "maintenance proof schema")
integer(value["pid"], 1, 2**31 - 1, "maintenance proof PID")
integer(value["issued_boottime_ns"], 0, 2**63 - 1, "maintenance proof time")
for key, pattern in (
("release_id", r"sd-[0-9a-f]{12}"),
("configuration_id", r"cfg-[0-9a-f]{24}"),
("invocation", r"[0-9a-f]{32}"),
("boot_id", r"[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}"),
):
if not isinstance(value[key], str) or not re.fullmatch(pattern, value[key]):
raise DeliveryError("maintenance proof identity differs")
sha256(value["manifest_sha256"])
sha256(value["metadata_sha256"])
return value
def _live(value: dict) -> None:
if (
value["boot_id"] != _boot()
or not 0 <= _now() - value["issued_boottime_ns"] <= MAX_AGE_NS
or _manager(value["release_id"], value["configuration_id"])
!= {key: value[key] for key in ("pid", "invocation")}
):
raise DeliveryError("maintenance byte proof expired or owner changed")
def _observed(value: dict) -> None:
"""One fixed journal event per process/invocation, never proof authority."""
global _REPORTED
owner = (os.getpid(), value["invocation"])
if _REPORTED == owner:
return
message = (
"KF complete byte proof reused manifest="
+ value["manifest_sha256"]
+ " invocation="
+ value["invocation"]
+ "\n"
).encode()
if os.write(2, message) != len(message):
raise DeliveryError("maintenance byte proof observation unavailable")
_REPORTED = owner
def reuse(bundle: Path, expected: str) -> dict | None:
"""Missing/ineligible proofs use full bytes; eligible corrupt proofs refuse."""
if not _root() or str(preflight.current_cgroup()) not in {GROUP + unit for unit in CONSUMERS}:
return None
try:
RECORD.lstat()
except FileNotFoundError:
return None
value, identity = _read()
if value["bundle"] != str(bundle) or value["manifest_sha256"] != expected:
raise DeliveryError("maintenance byte proof bundle differs")
_live(value)
preflight.protected(bundle, directory=True)
from . import manifest
before = manifest._bundle_signature(bundle)
if _signature(before) != value["metadata_sha256"]:
raise DeliveryError("maintenance byte proof metadata changed")
metadata = bundle / "manifest.json"
if digest(metadata) != expected:
raise DeliveryError("bundle manifest fingerprint mismatch")
data = manifest.validate(read_json(metadata))
if data["release_id"] != value["release_id"]:
raise DeliveryError("maintenance byte proof release differs")
manifest._unchanged_bundle(bundle, before)
reread, current = _read()
if current != identity or reread != value:
raise DeliveryError("maintenance byte proof replaced during use")
_live(value)
_observed(value)
return data
@contextmanager
def publish(selected: dict, bundle: Path):
"""Read all bytes once; systemd removes the private directory at unit exit."""
from . import install, manifest, operator_units
operator_units.guarded("node-maintenance")
if not _root() or str(preflight.current_cgroup()) != GROUP + PARENT:
raise DeliveryError("maintenance byte proof requires its fixed owner")
manager = _manager(selected["release_id"], selected["configuration_id"])
if manager["pid"] != os.getpid():
raise DeliveryError("only maintenance MainPID may publish byte proof")
if install.selection(check_configuration=False) != selected:
raise DeliveryError("maintenance proof selection changed")
preflight.protected(DIRECTORY, directory=True)
if RECORD.exists() or RECORD.is_symlink():
raise DeliveryError("existing maintenance byte proof refused")
created = None
try:
with manifest.verification_scope():
preflight.protected(bundle, directory=True)
before = manifest._bundle_signature(bundle)
if any(stamp[4:6] != (0, 0) or stamp[2] & 0o022 for stamp in before.values()):
raise DeliveryError("maintenance byte proof requires a protected Root tree")
data = manifest.verify(bundle, selected["manifest_sha256"])
manifest._unchanged_bundle(bundle, before)
if data["release_id"] != selected["release_id"]:
raise DeliveryError("maintenance byte proof release differs")
value = {
"schema_version": 1,
"bundle": str(bundle),
"manifest_sha256": selected["manifest_sha256"],
"release_id": selected["release_id"],
"configuration_id": selected["configuration_id"],
"metadata_sha256": _signature(before),
"boot_id": _boot(),
"issued_boottime_ns": _now(),
**manager,
}
contents = canonical(value)
if len(contents) > MAX_RECORD:
raise DeliveryError("maintenance byte proof exceeds bound")
install.exclusive_file(RECORD, contents, 0o400)
actual, created = _read()
if actual != value:
raise DeliveryError("maintenance byte proof publication differs")
_live(value)
yield
if install.selection(check_configuration=False) != selected:
raise DeliveryError("maintenance proof selection changed")
_live(value)
manifest._unchanged_bundle(bundle, before)
finally:
if created is not None:
actual, identity = _read()
if identity != created or actual != value:
raise DeliveryError("foreign maintenance proof; refused cleanup")
RECORD.unlink()
install.sync_directory(DIRECTORY)
+3
View File
@@ -187,7 +187,10 @@ def run(
]
if name == "node-maintenance":
command += [
"--property=RuntimeDirectory=kf-node-verification",
"--property=RuntimeDirectoryMode=0700",
"--property=ReadWritePaths=/etc/kf-admission /var/lib/kf-onlinejudge/operations"
" /run/kf-node-verification",
]
if name in {"import-health", "outbox-health"}:
if notification_probe: