335 lines
13 KiB
Python
335 lines
13 KiB
Python
"""Безопасный запуск уже распакованного bootstrap-пакета RX-агента.
|
||
|
||
Утилита принадлежит MAIN и не входит в подписанный bootstrap. Она проверяет
|
||
HMAC и точный состав распакованного пакета до запуска, сохраняет stdout,
|
||
stderr и снимки дерева только во внешнем каталоге, а после завершения
|
||
подтверждает, что пакет не изменился.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
from dataclasses import asdict, dataclass
|
||
from datetime import datetime, timezone
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import os
|
||
from pathlib import Path, PurePosixPath
|
||
import re
|
||
import subprocess
|
||
from typing import Any, Sequence
|
||
|
||
|
||
MANIFEST_NAME = "BOOTSTRAP_MANIFEST.json"
|
||
SIGNATURE_NAME = "BOOTSTRAP_MANIFEST.hmac"
|
||
INSTALLER_NAME = "INSTALL_RX_AGENT.cmd"
|
||
SIGNATURE_DOMAIN = b"RF001-UPDATE-MANIFEST-V1\n"
|
||
SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
|
||
|
||
|
||
class BootstrapInstallError(RuntimeError):
|
||
"""Базовая ошибка безопасного запуска bootstrap."""
|
||
|
||
|
||
class PackageValidationError(BootstrapInstallError):
|
||
"""Пакет не прошёл HMAC, manifest или точную проверку состава."""
|
||
|
||
|
||
class EvidencePathError(BootstrapInstallError):
|
||
"""Каталог доказательств небезопасен или уже содержит журналы запуска."""
|
||
|
||
|
||
class PackageMutationError(BootstrapInstallError):
|
||
"""Дерево bootstrap изменилось во время запуска установщика."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TreeEntry:
|
||
path: str
|
||
bytes: int
|
||
sha256: str
|
||
modified_time_ns: int
|
||
|
||
|
||
def utc_now() -> str:
|
||
return datetime.now(timezone.utc).isoformat()
|
||
|
||
|
||
def sha256_file(path: Path) -> str:
|
||
digest = hashlib.sha256()
|
||
with path.open("rb") as stream:
|
||
while block := stream.read(1024 * 1024):
|
||
digest.update(block)
|
||
return digest.hexdigest()
|
||
|
||
|
||
def snapshot_tree(root: Path) -> list[TreeEntry]:
|
||
entries: list[TreeEntry] = []
|
||
for path in sorted(
|
||
(candidate for candidate in root.rglob("*") if candidate.is_file()),
|
||
key=lambda candidate: candidate.as_posix().casefold(),
|
||
):
|
||
stat = path.stat()
|
||
entries.append(
|
||
TreeEntry(
|
||
path=path.relative_to(root).as_posix(),
|
||
bytes=stat.st_size,
|
||
sha256=sha256_file(path),
|
||
modified_time_ns=stat.st_mtime_ns,
|
||
)
|
||
)
|
||
return entries
|
||
|
||
|
||
def _content_identity(entries: Sequence[TreeEntry]) -> list[tuple[str, int, str]]:
|
||
return [(entry.path, entry.bytes, entry.sha256) for entry in entries]
|
||
|
||
|
||
def _write_json(path: Path, value: Any) -> None:
|
||
path.write_text(
|
||
json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
|
||
def _relative_manifest_path(value: object) -> str:
|
||
if not isinstance(value, str) or not value:
|
||
raise PackageValidationError("bootstrap manifest contains an empty path")
|
||
if "\\" in value or ":" in value:
|
||
raise PackageValidationError(f"unsafe bootstrap path: {value}")
|
||
relative = PurePosixPath(value)
|
||
if relative.is_absolute() or ".." in relative.parts or "." in relative.parts:
|
||
raise PackageValidationError(f"unsafe bootstrap path: {value}")
|
||
return relative.as_posix()
|
||
|
||
|
||
def validate_pristine_package(package_root: Path, token_file: Path) -> dict[str, Any]:
|
||
package_root = package_root.resolve()
|
||
token_file = token_file.resolve()
|
||
if not package_root.is_dir():
|
||
raise PackageValidationError("bootstrap package directory is absent")
|
||
manifest_path = package_root / MANIFEST_NAME
|
||
signature_path = package_root / SIGNATURE_NAME
|
||
installer_path = package_root / INSTALLER_NAME
|
||
for required in (manifest_path, signature_path, installer_path, token_file):
|
||
if not required.is_file():
|
||
raise PackageValidationError(f"required file is absent: {required}")
|
||
|
||
manifest_bytes = manifest_path.read_bytes()
|
||
token = token_file.read_text(encoding="utf-8").strip().encode("utf-8")
|
||
if len(token) < 32:
|
||
raise PackageValidationError("installed signing token is invalid")
|
||
supplied_signature = signature_path.read_text(encoding="ascii").strip().lower()
|
||
expected_signature = hmac.new(
|
||
token,
|
||
SIGNATURE_DOMAIN + manifest_bytes,
|
||
hashlib.sha256,
|
||
).hexdigest()
|
||
if not SHA256_PATTERN.fullmatch(supplied_signature) or not hmac.compare_digest(
|
||
supplied_signature,
|
||
expected_signature,
|
||
):
|
||
raise PackageValidationError("bootstrap manifest signature mismatch")
|
||
|
||
try:
|
||
manifest = json.loads(manifest_bytes.decode("utf-8"))
|
||
except (UnicodeDecodeError, json.JSONDecodeError) as error:
|
||
raise PackageValidationError("bootstrap manifest is invalid JSON") from error
|
||
if not isinstance(manifest, dict):
|
||
raise PackageValidationError("bootstrap manifest root must be an object")
|
||
if manifest.get("schema_version") != 1 or manifest.get("package_kind") != "BOOTSTRAP":
|
||
raise PackageValidationError("unsupported bootstrap manifest")
|
||
rows = manifest.get("files")
|
||
if not isinstance(rows, list) or not rows:
|
||
raise PackageValidationError("bootstrap manifest file list is invalid")
|
||
|
||
expected: dict[str, tuple[int, str]] = {}
|
||
for row in rows:
|
||
if not isinstance(row, dict):
|
||
raise PackageValidationError("bootstrap manifest file row is invalid")
|
||
relative = _relative_manifest_path(row.get("path"))
|
||
folded = relative.casefold()
|
||
if folded in expected:
|
||
raise PackageValidationError(f"duplicate bootstrap path: {relative}")
|
||
byte_count = row.get("bytes")
|
||
digest = row.get("sha256")
|
||
if type(byte_count) is not int or byte_count < 0:
|
||
raise PackageValidationError(f"invalid bootstrap size: {relative}")
|
||
if not isinstance(digest, str) or not SHA256_PATTERN.fullmatch(digest.lower()):
|
||
raise PackageValidationError(f"invalid bootstrap SHA256: {relative}")
|
||
expected[folded] = (byte_count, digest.lower())
|
||
|
||
source = package_root.joinpath(*PurePosixPath(relative).parts)
|
||
if not source.is_file() or source.is_symlink():
|
||
raise PackageValidationError(f"bootstrap file is absent or unsafe: {relative}")
|
||
if source.stat().st_size != byte_count:
|
||
raise PackageValidationError(f"bootstrap file size mismatch: {relative}")
|
||
if sha256_file(source) != digest.lower():
|
||
raise PackageValidationError(f"bootstrap file SHA256 mismatch: {relative}")
|
||
|
||
actual_entries = snapshot_tree(package_root)
|
||
actual = {entry.path.casefold() for entry in actual_entries}
|
||
allowed = set(expected) | {MANIFEST_NAME.casefold(), SIGNATURE_NAME.casefold()}
|
||
unexpected = sorted(actual - allowed)
|
||
missing = sorted(allowed - actual)
|
||
if unexpected or missing:
|
||
raise PackageValidationError(
|
||
"bootstrap package allowlist mismatch; "
|
||
f"unexpected={unexpected}; missing={missing}"
|
||
)
|
||
return {
|
||
"bootstrap_id": manifest.get("bootstrap_id"),
|
||
"agent_release_id": manifest.get("agent_release_id"),
|
||
"agent_version": manifest.get("agent_version"),
|
||
"hmac": "PASS",
|
||
"manifest_file_count": len(expected),
|
||
"package_file_count": len(actual_entries),
|
||
"unexpected": 0,
|
||
"missing": 0,
|
||
}
|
||
|
||
|
||
def _assert_external_evidence(package_root: Path, evidence_root: Path) -> None:
|
||
package_root = package_root.resolve()
|
||
evidence_root = evidence_root.resolve()
|
||
if evidence_root == package_root or package_root in evidence_root.parents:
|
||
raise EvidencePathError("evidence directory must be outside the package tree")
|
||
|
||
|
||
def run_installer(
|
||
package_root: Path,
|
||
evidence_root: Path,
|
||
token_file: Path,
|
||
*,
|
||
command: Sequence[str] | None = None,
|
||
timeout_seconds: float | None = None,
|
||
) -> dict[str, Any]:
|
||
package_root = package_root.resolve()
|
||
evidence_root = evidence_root.resolve()
|
||
_assert_external_evidence(package_root, evidence_root)
|
||
preflight = validate_pristine_package(package_root, token_file)
|
||
|
||
evidence_root.mkdir(parents=True, exist_ok=True)
|
||
paths = {
|
||
"stdout": evidence_root / "install.stdout.txt",
|
||
"stderr": evidence_root / "install.stderr.txt",
|
||
"runtime": evidence_root / "install_runtime.json",
|
||
"before": evidence_root / "package_tree_before.json",
|
||
"immediate": evidence_root / "package_tree_immediate.json",
|
||
"after": evidence_root / "package_tree_after.json",
|
||
"preflight": evidence_root / "package_preflight.json",
|
||
}
|
||
occupied = [str(path) for path in paths.values() if path.exists()]
|
||
if occupied:
|
||
raise EvidencePathError(f"runtime evidence paths already exist: {occupied}")
|
||
|
||
before = snapshot_tree(package_root)
|
||
_write_json(paths["preflight"], preflight)
|
||
_write_json(paths["before"], [asdict(entry) for entry in before])
|
||
invocation = list(command) if command is not None else [
|
||
os.environ.get("COMSPEC", "cmd.exe"),
|
||
"/d",
|
||
"/c",
|
||
"call",
|
||
str(package_root / INSTALLER_NAME),
|
||
]
|
||
metadata: dict[str, Any] = {
|
||
"started_at_utc": utc_now(),
|
||
"package_root": str(package_root),
|
||
"evidence_root": str(evidence_root),
|
||
"working_directory": str(package_root),
|
||
"stdout_path": str(paths["stdout"]),
|
||
"stderr_path": str(paths["stderr"]),
|
||
"rf_rx": 0,
|
||
"rf_tx": 0,
|
||
}
|
||
creation_flags = getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
||
with paths["stdout"].open("wb") as stdout_stream, paths["stderr"].open(
|
||
"wb"
|
||
) as stderr_stream:
|
||
process = subprocess.Popen(
|
||
invocation,
|
||
cwd=package_root,
|
||
stdin=subprocess.DEVNULL,
|
||
stdout=stdout_stream,
|
||
stderr=stderr_stream,
|
||
creationflags=creation_flags,
|
||
)
|
||
metadata["installer_process_id"] = process.pid
|
||
immediate = snapshot_tree(package_root)
|
||
_write_json(paths["immediate"], [asdict(entry) for entry in immediate])
|
||
metadata["package_content_unchanged_immediate"] = (
|
||
_content_identity(immediate) == _content_identity(before)
|
||
)
|
||
metadata["package_metadata_unchanged_immediate"] = immediate == before
|
||
try:
|
||
exit_code = process.wait(timeout=timeout_seconds)
|
||
except subprocess.TimeoutExpired as error:
|
||
metadata["timed_out"] = True
|
||
metadata["completed_at_utc"] = utc_now()
|
||
_write_json(paths["runtime"], metadata)
|
||
raise BootstrapInstallError(
|
||
"installer timeout; the process was not restarted or killed automatically"
|
||
) from error
|
||
|
||
after = snapshot_tree(package_root)
|
||
_write_json(paths["after"], [asdict(entry) for entry in after])
|
||
metadata.update(
|
||
{
|
||
"completed_at_utc": utc_now(),
|
||
"exit_code": exit_code,
|
||
"timed_out": False,
|
||
"package_content_unchanged_after": (
|
||
_content_identity(after) == _content_identity(before)
|
||
),
|
||
"package_metadata_unchanged_after": after == before,
|
||
"stdout_bytes": paths["stdout"].stat().st_size,
|
||
"stderr_bytes": paths["stderr"].stat().st_size,
|
||
}
|
||
)
|
||
_write_json(paths["runtime"], metadata)
|
||
if not metadata["package_content_unchanged_immediate"] or not metadata[
|
||
"package_content_unchanged_after"
|
||
]:
|
||
raise PackageMutationError("bootstrap package changed during installer run")
|
||
if exit_code != 0:
|
||
raise BootstrapInstallError(f"installer exited with code {exit_code}")
|
||
return metadata
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--package-root", required=True, type=Path)
|
||
parser.add_argument("--evidence-root", required=True, type=Path)
|
||
parser.add_argument("--token-file", required=True, type=Path)
|
||
parser.add_argument("--timeout-seconds", type=float, default=None)
|
||
return parser.parse_args()
|
||
|
||
|
||
def main() -> int:
|
||
arguments = parse_args()
|
||
try:
|
||
result = run_installer(
|
||
arguments.package_root,
|
||
arguments.evidence_root,
|
||
arguments.token_file,
|
||
timeout_seconds=arguments.timeout_seconds,
|
||
)
|
||
except BootstrapInstallError as error:
|
||
print(f"BOOTSTRAP INSTALL = FAILED: {error}")
|
||
print("RF RX = 0")
|
||
print("RF TX = 0")
|
||
return 1
|
||
print("BOOTSTRAP INSTALL = COMPLETE")
|
||
print(f"INSTALL EXIT CODE = {result['exit_code']}")
|
||
print(f"EVIDENCE ROOT = {result['evidence_root']}")
|
||
print("PACKAGE TREE UNCHANGED = YES")
|
||
print("RF RX = 0")
|
||
print("RF TX = 0")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|