mptyl/ThothII
1
0
Fork 0
Code Issues 25 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

test: add deterministic runtime config chain race regression

Browse Source
This commit is contained in:
marcopan
2026-08-11 12:16:25 +02:00
parent 1bca663ace
commit cf88e2df86
2 changed files with 103 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
harness/tht/config.py
+54 -10
View File
@@ -5,6 +5,7 @@ import re
import stat
import sys
import warnings
from collections.abc import Callable
from ipaddress import ip_address
from pathlib import Path
from typing import Annotated, Any, Literal
@@ -694,8 +695,25 @@ def _runtime_directory_identities(paths: list[Path]) -> list[dict[str, str]]:
return out
def _verify_runtime_directory_identities(config_path: Path, manifest_path: Path, expected: object) -> None:
current = _runtime_directory_identities([config_path.parent, manifest_path.parent])
def _verify_runtime_directory_identities(
config_path: Path,
manifest_path: Path,
expected: object,
*,
between_config_and_manifest_traversal: Callable[[], None] | None = None,
) -> None:
# Traverse the two destination branches separately. The callback is a narrow,
# package-private seam for deterministic direct tests of a replacement between
# those traversals; production always passes None.
current = _runtime_directory_identities([config_path.parent])
if between_config_and_manifest_traversal is not None:
between_config_and_manifest_traversal()
for identity in _runtime_directory_identities([manifest_path.parent]):
previous = next((item for item in current if item["path"] == identity["path"]), None)
if previous is None:
current.append(identity)
elif previous != identity:
raise ConfigError("Destinazione config runtime modificata")
if current != expected:
raise ConfigError("Destinazione config runtime modificata")
@@ -746,11 +764,20 @@ def _runtime_manifest_path(config_path: Path) -> Path:
raise OSError("runtime config path is not canonical")
return config_path.parent.parent / "runtime-config-manifests" / f"{config_path.stem}.json"
def load_config(path: Path) -> Config:
def _read_runtime_config_source(
path: Path,
*,
between_config_and_manifest_traversal: Callable[[], None] | None = None,
) -> tuple[str | None, dict[str, object] | None]:
"""Read a trusted runtime config, with a scoped direct-test race seam.
The callback is deliberately available only on this package-private helper.
Production callers go through :func:`load_config`, which always passes ``None``;
no environment variable or process-global hook can alter this traversal.
"""
# Registry leases authenticate the canonical pathname through a durable manifest
# digest. The config and manifest are opened component-by-component; FD 3/4 are
# reserved for the maintenance writer/root ABI.
runtime_manifest: dict[str, object] | None = None
expected_manifest = os.environ.get("THT_RUNTIME_CONFIG_MANIFEST_SHA256")
runtime_fd = os.environ.get("THT_CONFIG_FD")
manifest_fd = os.environ.get("THT_CONFIG_MANIFEST_FD")
@@ -767,7 +794,12 @@ def load_config(path: Path) -> Config:
if hashlib.sha256(manifest_bytes).hexdigest() != expected_manifest:
raise ConfigError("Manifest runtime modificato")
runtime_manifest = _strict_runtime_manifest(json.loads(manifest_bytes.decode("utf-8")))
_verify_runtime_directory_identities(path, _runtime_manifest_path(path), runtime_manifest["directory_identities"])
_verify_runtime_directory_identities(
path,
_runtime_manifest_path(path),
runtime_manifest["directory_identities"],
between_config_and_manifest_traversal=between_config_and_manifest_traversal,
)
if (runtime_manifest["config_sha256"] != hashlib.sha256(config_bytes).hexdigest()
or int(runtime_manifest["config_dev"]) != config_info.st_dev
or int(runtime_manifest["config_ino"]) != config_info.st_ino
@@ -776,7 +808,7 @@ def load_config(path: Path) -> Config:
or int(runtime_manifest["config_nlink"]) != config_info.st_nlink
or config_info.st_dev == manifest_info.st_dev and config_info.st_ino == manifest_info.st_ino):
raise ConfigError("Identità config runtime non valida")
source_text = config_bytes.decode("utf-8")
return config_bytes.decode("utf-8"), runtime_manifest
except (OSError, UnicodeError, ValueError, json.JSONDecodeError) as exc:
if isinstance(exc, ConfigError):
raise
@@ -786,7 +818,7 @@ def load_config(path: Path) -> Config:
os.close(config_fd)
if manifest_fd_local is not None:
os.close(manifest_fd_local)
elif runtime_fd is not None or manifest_fd is not None or legacy_expected is not None:
if runtime_fd is not None or manifest_fd is not None or legacy_expected is not None:
# Compatibility for direct /dev/fd callers. New backend leases never use it.
if runtime_fd is None or manifest_fd is None or legacy_expected is None or not re.fullmatch(r"[0-9a-f]{64}", legacy_expected):
raise ConfigError("Handoff runtime incompleto")
@@ -796,7 +828,12 @@ def load_config(path: Path) -> Config:
if hashlib.sha256(manifest_bytes).hexdigest() != legacy_expected:
raise ConfigError("Manifest runtime modificato")
runtime_manifest = _strict_runtime_manifest(json.loads(manifest_bytes.decode("utf-8")))
_verify_runtime_directory_identities(path, _runtime_manifest_path(path), runtime_manifest["directory_identities"])
_verify_runtime_directory_identities(
path,
_runtime_manifest_path(path),
runtime_manifest["directory_identities"],
between_config_and_manifest_traversal=between_config_and_manifest_traversal,
)
if (runtime_manifest["config_sha256"] != hashlib.sha256(config_bytes).hexdigest()
or int(runtime_manifest["config_dev"]) != config_info.st_dev
or int(runtime_manifest["config_ino"]) != config_info.st_ino
@@ -805,12 +842,19 @@ def load_config(path: Path) -> Config:
or int(runtime_manifest["config_nlink"]) != config_info.st_nlink
or config_info.st_dev == manifest_info.st_dev and config_info.st_ino == manifest_info.st_ino):
raise ConfigError("Identità config runtime non valida")
source_text = config_bytes.decode("utf-8")
return config_bytes.decode("utf-8"), runtime_manifest
except (OSError, UnicodeError, ValueError, json.JSONDecodeError) as exc:
if isinstance(exc, ConfigError):
raise
raise ConfigError("File di configurazione runtime non attendibile") from exc
else:
return None, None
def load_config(path: Path) -> Config:
source_text, runtime_manifest = _read_runtime_config_source(
path, between_config_and_manifest_traversal=None
)
if source_text is None:
if not path.exists():
raise ConfigError(f"File di configurazione non trovato: {path}")
try: