fix: harden runtime config lease identity publication

harness/tht/config.py

@@ -599,10 +599,14 @@ def _strict_runtime_manifest(raw: object) -> dict[str, object]:
         "version", "workspace_id", "workspace_revision", "descriptor_git_blob",
         "descriptor_sha256", "descriptor_dev", "descriptor_ino", "config_sha256",
         "config_dwh_binding", "config_dev", "config_ino", "config_size", "config_mode",
-        "config_uid", "config_nlink",
+        "config_uid", "config_nlink", "directory_identities",
     }
     if not isinstance(raw, dict) or set(raw) != required or raw.get("version") != 1:
         raise ConfigError("Manifest runtime non valido")
+    if (not isinstance(raw.get("workspace_id"), str) or not re.fullmatch(r"[a-z][a-z0-9-]{2,62}", raw["workspace_id"])
+            or not isinstance(raw.get("workspace_revision"), str) or not re.fullmatch(r"[0-9a-f]{40}", raw["workspace_revision"])
+            or not isinstance(raw.get("descriptor_git_blob"), str) or not re.fullmatch(r"[0-9a-f]{40}", raw["descriptor_git_blob"])):
+        raise ConfigError("Manifest runtime non valido")
     if not isinstance(raw.get("config_dwh_binding"), dict):
         raise ConfigError("Manifest runtime non valido")
     binding = raw["config_dwh_binding"]
@@ -614,12 +618,88 @@ def _strict_runtime_manifest(raw: object) -> dict[str, object]:
     for key in ("descriptor_dev", "descriptor_ino", "config_dev", "config_ino", "config_size", "config_uid", "config_nlink"):
         if not isinstance(raw[key], str) or not raw[key].isdigit():
             raise ConfigError("Manifest runtime non valido")
+    identities = raw.get("directory_identities")
+    if (not isinstance(identities, list) or not identities or
+            any(not isinstance(item, dict) or set(item) != {"path", "dev", "ino", "mode", "uid"}
+                or not isinstance(item["path"], str) or not item["path"].startswith("/")
+                or any(not isinstance(item[key], str) or not item[key].isdigit()
+                       for key in ("dev", "ino", "mode", "uid"))
+                or item["mode"] == "0"
+                for item in identities)):
+        raise ConfigError("Manifest runtime non valido")
+    if len({item["path"] for item in identities}) != len(identities):
+        raise ConfigError("Manifest runtime non valido")
     if raw["config_mode"] != "400":
         raise ConfigError("Manifest runtime non valido")
     return raw
 
 
 
+def _canonical_runtime_path(path: Path) -> Path:
+    value = str(path)
+    if value == "/tmp" or value.startswith("/tmp/"):
+        return Path("/private" + value)
+    if value == "/var" or value.startswith("/var/"):
+        return Path("/private" + value)
+    return path
+
+
+def _runtime_directory_identities(paths: list[Path]) -> list[dict[str, str]]:
+    """Open/stat every directory component and return its current identity chain."""
+    out: list[dict[str, str]] = []
+    seen: set[str] = set()
+    for path in paths:
+        canonical = _canonical_runtime_path(path)
+        parts = list(canonical.parts)
+        if not parts or parts[0] != "/":
+            raise OSError("runtime config path is invalid")
+        current = "/"
+        components = ["/"] + parts[1:]
+        for component in components:
+            if component != "/":
+                current = current.rstrip("/") + "/" + component
+            # Re-open shared prefixes for each destination branch too: a
+            # replacement between config-parent and manifest-parent traversal
+            # must not be hidden by de-duplication.
+            # lstat before and fstat after open closes the stat/open replacement
+            # window for each component, including canonical destination parents.
+            parent = os.open("/", os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+            try:
+                for name in [part for part in Path(current).parts[1:-1]]:
+                    nxt = _open_runtime_component(parent, name)
+                    os.close(parent); parent = nxt
+                if current == "/":
+                    fd = os.dup(parent)
+                else:
+                    name = Path(current).name
+                    fd = _open_runtime_component(parent, name)
+                try:
+                    info = os.fstat(fd)
+                    if not stat.S_ISDIR(info.st_mode) or info.st_nlink < 1:
+                        raise OSError("unsafe runtime directory")
+                    identity = {"path": current, "dev": str(info.st_dev), "ino": str(info.st_ino),
+                                "mode": format(stat.S_IMODE(info.st_mode), "o"),
+                                "uid": str(info.st_uid)}
+                    if current in seen:
+                        previous = next(item for item in out if item["path"] == current)
+                        if identity != previous:
+                            raise OSError("runtime config directory changed")
+                    else:
+                        out.append(identity)
+                        seen.add(current)
+                finally:
+                    os.close(fd)
+            finally:
+                os.close(parent)
+    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])
+    if current != expected:
+        raise ConfigError("Destinazione config runtime modificata")
+
+
 def _open_runtime_component(parent: int, name: str) -> int:
     before = os.stat(name, dir_fd=parent, follow_symlinks=False)
     if stat.S_ISLNK(before.st_mode):
@@ -687,6 +767,7 @@ 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"])
             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
@@ -715,6 +796,7 @@ 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"])
             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