248 lines
12 KiB
Python
248 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Evaluate a bounded, semantic, loopback-only activation health window."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
|
|
SCHEMA = "com.cezen.nexus-one.activation-health-evidence.v1"
|
|
MAX_RESPONSE_BYTES = 65536
|
|
UNIT_NAME = re.compile(r"^[A-Za-z0-9_.@-]+\.service$")
|
|
|
|
|
|
class HealthConfigError(ValueError):
|
|
pass
|
|
|
|
|
|
def canonical(value: object) -> bytes:
|
|
return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode()
|
|
|
|
|
|
def validate_config(config: dict) -> list[dict]:
|
|
if set(config) != {"schema", "components"} or config["schema"] != \
|
|
"com.cezen.nexus-one.activation-health-profile.v1":
|
|
raise HealthConfigError("HEALTH_PROFILE_INVALID")
|
|
components = config["components"]
|
|
if not isinstance(components, list) or not components:
|
|
raise HealthConfigError("HEALTH_COMPONENTS_INVALID")
|
|
names = set()
|
|
result = []
|
|
for component in components:
|
|
if not isinstance(component, dict) or set(component) != {
|
|
"name", "url", "required", "expected_service", "systemd_unit", "max_restarts"}:
|
|
raise HealthConfigError("HEALTH_COMPONENT_INVALID")
|
|
name = component["name"]
|
|
if not isinstance(name, str) or not name or name in names:
|
|
raise HealthConfigError("HEALTH_COMPONENT_NAME_INVALID")
|
|
names.add(name)
|
|
if not isinstance(component["required"], bool):
|
|
raise HealthConfigError("HEALTH_COMPONENT_REQUIRED_INVALID")
|
|
expected = component["expected_service"]
|
|
if not isinstance(expected, str) or not expected:
|
|
raise HealthConfigError("HEALTH_COMPONENT_EXPECTED_SERVICE_INVALID")
|
|
unit = component["systemd_unit"]
|
|
if unit is not None and (not isinstance(unit, str) or not UNIT_NAME.fullmatch(unit)):
|
|
raise HealthConfigError("HEALTH_COMPONENT_SYSTEMD_UNIT_INVALID")
|
|
maximum = component["max_restarts"]
|
|
if not isinstance(maximum, int) or isinstance(maximum, bool) or maximum < 0 or maximum > 1000:
|
|
raise HealthConfigError("HEALTH_COMPONENT_MAX_RESTARTS_INVALID")
|
|
parsed = urllib.parse.urlparse(component["url"])
|
|
if parsed.scheme != "http" or parsed.username or parsed.password or parsed.query \
|
|
or parsed.fragment or not parsed.port or parsed.path == "":
|
|
raise HealthConfigError("HEALTH_URL_INVALID")
|
|
try:
|
|
address = ipaddress.ip_address(parsed.hostname or "")
|
|
except ValueError as error:
|
|
raise HealthConfigError("HEALTH_URL_NOT_LITERAL_LOOPBACK") from error
|
|
if not address.is_loopback:
|
|
raise HealthConfigError("HEALTH_URL_NOT_LOOPBACK")
|
|
result.append(dict(component))
|
|
return result
|
|
|
|
|
|
def fetch(component: dict, timeout_seconds: float) -> dict:
|
|
request = urllib.request.Request(component["url"], headers={"Accept": "application/json"})
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
|
|
if response.status != 200:
|
|
return {"ok": False, "code": f"HTTP_{response.status}"}
|
|
content_type = response.headers.get_content_type()
|
|
if content_type != "application/json":
|
|
return {"ok": False, "code": "CONTENT_TYPE_INVALID"}
|
|
payload = response.read(MAX_RESPONSE_BYTES + 1)
|
|
if len(payload) > MAX_RESPONSE_BYTES:
|
|
return {"ok": False, "code": "RESPONSE_TOO_LARGE"}
|
|
except urllib.error.HTTPError as error:
|
|
return {"ok": False, "code": f"HTTP_{error.code}"}
|
|
except TimeoutError:
|
|
return {"ok": False, "code": "TIMEOUT"}
|
|
except (OSError, urllib.error.URLError):
|
|
return {"ok": False, "code": "UNREACHABLE"}
|
|
try:
|
|
decoded = json.loads(payload)
|
|
except (UnicodeDecodeError, json.JSONDecodeError):
|
|
return {"ok": False, "code": "MALFORMED_JSON"}
|
|
if not isinstance(decoded, dict) or set(decoded) != {"service", "status"}:
|
|
return {"ok": False, "code": "RESPONSE_SHAPE_INVALID"}
|
|
if decoded["service"] != component["expected_service"]:
|
|
return {"ok": False, "code": "SERVICE_IDENTITY_MISMATCH"}
|
|
if decoded["status"] not in {"ready", "degraded", "failed"}:
|
|
return {"ok": False, "code": "STATUS_INVALID"}
|
|
return {"ok": decoded["status"] == "ready", "code": decoded["status"].upper(),
|
|
"status": decoded["status"]}
|
|
|
|
|
|
def restart_state(component: dict) -> dict:
|
|
unit = component["systemd_unit"]
|
|
if unit is None:
|
|
return {"ok": True, "code": "RESTART_CHECK_NOT_APPLICABLE", "restarts": 0}
|
|
try:
|
|
process = subprocess.run(
|
|
["systemctl", "show", unit, "--property=NRestarts", "--property=ActiveState",
|
|
"--property=SubState", "--no-pager"], check=True, capture_output=True, text=True,
|
|
timeout=5,
|
|
)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return {"ok": False, "code": "UNIT_STATE_UNAVAILABLE", "restarts": None}
|
|
values = {}
|
|
for line in process.stdout.splitlines():
|
|
key, separator, value = line.partition("=")
|
|
if not separator or key in values:
|
|
return {"ok": False, "code": "UNIT_STATE_INVALID", "restarts": None}
|
|
values[key] = value
|
|
if set(values) != {"NRestarts", "ActiveState", "SubState"} or not values["NRestarts"].isdigit():
|
|
return {"ok": False, "code": "UNIT_STATE_INVALID", "restarts": None}
|
|
restarts = int(values["NRestarts"])
|
|
if restarts > component["max_restarts"]:
|
|
return {"ok": False, "code": "RESTART_STORM", "restarts": restarts}
|
|
if values["ActiveState"] != "active":
|
|
return {"ok": False, "code": "UNIT_NOT_ACTIVE", "restarts": restarts}
|
|
return {"ok": True, "code": "RESTARTS_OK", "restarts": restarts}
|
|
|
|
|
|
def atomic_evidence(path: Path, evidence: dict) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
|
|
try:
|
|
os.fchmod(fd, 0o600)
|
|
with os.fdopen(fd, "wb") as stream:
|
|
stream.write(canonical(evidence))
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(temporary, path)
|
|
directory_fd = os.open(path.parent, os.O_RDONLY)
|
|
try:
|
|
os.fsync(directory_fd)
|
|
finally:
|
|
os.close(directory_fd)
|
|
finally:
|
|
if os.path.exists(temporary):
|
|
os.unlink(temporary)
|
|
|
|
|
|
def evaluate(config: dict, operation_id: str, samples: int, interval_seconds: float,
|
|
timeout_seconds: float, window_timeout_seconds: float, allow_degraded: bool,
|
|
fetcher: Callable = fetch, restart_checker: Callable = restart_state,
|
|
sleeper: Callable = time.sleep, clock: Callable = time.monotonic) -> dict:
|
|
components = validate_config(config)
|
|
if not operation_id or any(character not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_." for character in operation_id):
|
|
raise HealthConfigError("OPERATION_ID_INVALID")
|
|
if samples < 1 or samples > 100 or interval_seconds < 0 or timeout_seconds <= 0 \
|
|
or window_timeout_seconds <= 0 or timeout_seconds > window_timeout_seconds:
|
|
raise HealthConfigError("HEALTH_WINDOW_BOUNDS_INVALID")
|
|
records = []
|
|
outcome = "ready"
|
|
deadline = clock() + window_timeout_seconds
|
|
for sample in range(1, samples + 1):
|
|
for component in components:
|
|
remaining = deadline - clock()
|
|
if remaining <= 0:
|
|
result = {"ok": False, "code": "WINDOW_TIMEOUT"}
|
|
restart = {"code": "NOT_CHECKED", "restarts": None}
|
|
else:
|
|
restart = restart_checker(component)
|
|
if restart.get("ok") is not True:
|
|
result = restart
|
|
else:
|
|
remaining = deadline - clock()
|
|
result = {"ok": False, "code": "WINDOW_TIMEOUT"} if remaining <= 0 else \
|
|
fetcher(component, min(timeout_seconds, remaining))
|
|
if clock() > deadline:
|
|
result = {"ok": False, "code": "WINDOW_TIMEOUT"}
|
|
status = result.get("status")
|
|
accepted_degraded = status == "degraded" and not component["required"] and allow_degraded
|
|
accepted = result.get("ok") is True or accepted_degraded
|
|
if accepted_degraded:
|
|
outcome = "degraded"
|
|
records.append({"sample": sample, "component": component["name"],
|
|
"required": component["required"], "accepted": accepted,
|
|
"code": result.get("code", "PROBE_RESULT_INVALID"),
|
|
"restart_code": restart.get("code"),
|
|
"restart_count": restart.get("restarts")})
|
|
if not accepted:
|
|
outcome = "rejected"
|
|
return {"schema": SCHEMA, "operation_id": operation_id,
|
|
"profile_sha256": hashlib.sha256(canonical(config)).hexdigest(),
|
|
"samples_required": samples, "samples_completed": sample,
|
|
"allow_degraded": allow_degraded, "outcome": outcome, "checks": records}
|
|
if sample != samples:
|
|
if clock() + interval_seconds > deadline:
|
|
records.append({"sample": sample, "component": "window", "required": True,
|
|
"accepted": False, "code": "WINDOW_TIMEOUT",
|
|
"restart_code": "NOT_APPLICABLE", "restart_count": None})
|
|
return {"schema": SCHEMA, "operation_id": operation_id,
|
|
"profile_sha256": hashlib.sha256(canonical(config)).hexdigest(),
|
|
"samples_required": samples, "samples_completed": sample,
|
|
"allow_degraded": allow_degraded, "outcome": "rejected",
|
|
"checks": records}
|
|
sleeper(interval_seconds)
|
|
return {"schema": SCHEMA, "operation_id": operation_id,
|
|
"profile_sha256": hashlib.sha256(canonical(config)).hexdigest(),
|
|
"samples_required": samples, "samples_completed": samples,
|
|
"allow_degraded": allow_degraded, "outcome": outcome, "checks": records}
|
|
|
|
|
|
def main(argv: list[str] | None = None, fetcher: Callable = fetch,
|
|
restart_checker: Callable = restart_state, sleeper: Callable = time.sleep,
|
|
clock: Callable = time.monotonic) -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--operation-id", required=True)
|
|
parser.add_argument("--profile-config", type=Path, required=True)
|
|
parser.add_argument("--samples", type=int, required=True)
|
|
parser.add_argument("--interval-seconds", type=float, required=True)
|
|
parser.add_argument("--timeout-seconds", type=float, required=True)
|
|
parser.add_argument("--window-timeout-seconds", type=float, required=True)
|
|
parser.add_argument("--evidence", type=Path, required=True)
|
|
parser.add_argument("--allow-degraded", action="store_true")
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
config = json.loads(args.profile_config.read_text(encoding="utf-8"))
|
|
evidence = evaluate(config, args.operation_id, args.samples, args.interval_seconds,
|
|
args.timeout_seconds, args.window_timeout_seconds, args.allow_degraded,
|
|
fetcher, restart_checker, sleeper, clock)
|
|
atomic_evidence(args.evidence, evidence)
|
|
except (HealthConfigError, OSError, json.JSONDecodeError) as error:
|
|
print(json.dumps({"accepted": False, "code": str(error)}, sort_keys=True))
|
|
return 2
|
|
print(json.dumps({"accepted": evidence["outcome"] in {"ready", "degraded"},
|
|
"outcome": evidence["outcome"]}, sort_keys=True))
|
|
return 0 if evidence["outcome"] in {"ready", "degraded"} else 23
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|