782 lines
30 KiB
Python
782 lines
30 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import importlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import stat
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from types import ModuleType
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
MODULE_PATH = ROOT / "tools/aios_spark_runner/runnerctl.py"
|
|
DEPLOY = ROOT / "deploy/spark-runner"
|
|
SOURCE_COMMIT = subprocess.check_output(
|
|
["git", "rev-parse", "HEAD"], cwd=ROOT, text=True
|
|
).strip()
|
|
SOURCE_URI = "gitea://aios/aios.git#tools/aios_spark_runner"
|
|
|
|
|
|
def runnerctl() -> ModuleType:
|
|
assert MODULE_PATH.is_file(), "runner implementation is missing"
|
|
spec = importlib.util.spec_from_file_location("aios_spark_runner_runnerctl_test", MODULE_PATH)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def write_config(
|
|
path: Path,
|
|
*,
|
|
artifact_sha: str = "b" * 64,
|
|
secret_ref: str | None = "secret://runtime/aios-spark-runner/gateway/v1",
|
|
**changes: object,
|
|
) -> dict[str, object]:
|
|
data: dict[str, object] = {
|
|
"runner_artifact_uri_or_repo_path": SOURCE_URI,
|
|
"runner_artifact_version_or_digest": SOURCE_COMMIT,
|
|
"runner_artifact_sha256_or_signature": f"sha256:{artifact_sha}",
|
|
"runner_secret_ref": secret_ref,
|
|
"runner_run_user": "aios-spark-runner",
|
|
"runner_run_group": "aios-spark-runner",
|
|
"runner_workdir": "/var/lib/aios-spark-runner",
|
|
"runner_heartbeat_path_or_endpoint": "/run/aios-spark-runner/heartbeat.json",
|
|
"service_name_or_container_name": "aios-spark-runner.service",
|
|
}
|
|
data.update(changes)
|
|
path.write_text(json.dumps(data), encoding="utf-8")
|
|
path.chmod(0o600)
|
|
return data
|
|
|
|
|
|
def signing_keys(tmp_path: Path) -> tuple[Path, Path]:
|
|
private = tmp_path / "artifact-signing-key.pem"
|
|
public = tmp_path / "artifact-signing-key.pub.pem"
|
|
subprocess.run(
|
|
["openssl", "genpkey", "-algorithm", "ED25519", "-out", str(private)],
|
|
check=True,
|
|
)
|
|
subprocess.run(
|
|
["openssl", "pkey", "-in", str(private), "-pubout", "-out", str(public)],
|
|
check=True,
|
|
)
|
|
private.chmod(0o600)
|
|
public.chmod(0o600)
|
|
return private, public
|
|
|
|
|
|
def build_artifact(
|
|
tmp_path: Path, *, commit: str = SOURCE_COMMIT
|
|
) -> tuple[Path, Path, Path, Path]:
|
|
module = runnerctl()
|
|
tmp_path.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
output = tmp_path / "dist"
|
|
private, public = signing_keys(tmp_path)
|
|
result = module.build_artifact(
|
|
commit, output, signing_key=private, source_repo=ROOT
|
|
)
|
|
return (
|
|
Path(result["artifact"]),
|
|
Path(result["config"]),
|
|
Path(result["manifest"]),
|
|
public,
|
|
)
|
|
|
|
|
|
def test_required_delivery_paths_exist() -> None:
|
|
expected = [
|
|
"tools/aios_spark_runner/__init__.py",
|
|
"tools/aios_spark_runner/runnerctl.py",
|
|
"tools/aios_spark_runner/broker.py",
|
|
"tools/aios_spark_runner/provenance.py",
|
|
"deploy/spark-runner/build-artifact.sh",
|
|
"deploy/spark-runner/provision.sh",
|
|
"deploy/spark-runner/rollback.sh",
|
|
"deploy/spark-runner/uninstall.sh",
|
|
"deploy/spark-runner/acceptance.sh",
|
|
"deploy/spark-runner/runner-config.schema.json",
|
|
"deploy/spark-runner/artifact-manifest.schema.json",
|
|
"deploy/spark-runner/runner-config.template.json",
|
|
"deploy/spark-runner/README.md",
|
|
"deploy/systemd/aios-spark-runner.service",
|
|
"deploy/sysusers.d/aios-spark-runner.conf",
|
|
"deploy/tmpfiles.d/aios-spark-runner.conf",
|
|
]
|
|
missing = [path for path in expected if not (ROOT / path).is_file()]
|
|
assert not missing, missing
|
|
|
|
|
|
def test_template_locks_nonsecret_defaults_and_blocks_secret() -> None:
|
|
path = DEPLOY / "runner-config.template.json"
|
|
assert path.is_file()
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
assert data == {
|
|
"runner_artifact_uri_or_repo_path": SOURCE_URI,
|
|
"runner_artifact_version_or_digest": None,
|
|
"runner_artifact_sha256_or_signature": None,
|
|
"runner_secret_ref": None,
|
|
"runner_run_user": "aios-spark-runner",
|
|
"runner_run_group": "aios-spark-runner",
|
|
"runner_workdir": "/var/lib/aios-spark-runner",
|
|
"runner_heartbeat_path_or_endpoint": "/run/aios-spark-runner/heartbeat.json",
|
|
"service_name_or_container_name": "aios-spark-runner.service",
|
|
}
|
|
assert "REQUIRED" not in path.read_text(encoding="utf-8")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("changes", "match"),
|
|
[
|
|
({"runner_artifact_uri_or_repo_path": "https://example.invalid/latest"}, "artifact URI"),
|
|
({"runner_artifact_version_or_digest": "latest"}, "source commit"),
|
|
({"runner_artifact_version_or_digest": "A" * 40}, "source commit"),
|
|
({"runner_artifact_sha256_or_signature": "sha256:1234"}, "SHA-256"),
|
|
({"runner_artifact_sha256_or_signature": "signature:fixture"}, "signature"),
|
|
({"runner_run_user": "root"}, "run user"),
|
|
({"runner_run_group": "root"}, "run group"),
|
|
({"runner_workdir": "relative"}, "workdir"),
|
|
({"runner_heartbeat_path_or_endpoint": "/tmp/heartbeat"}, "heartbeat"),
|
|
({"service_name_or_container_name": "runner.service"}, "service"),
|
|
({"extra": "value"}, "unknown"),
|
|
],
|
|
)
|
|
def test_config_rejects_mutable_placeholder_or_unknown_values(
|
|
tmp_path: Path, changes: dict[str, object], match: str
|
|
) -> None:
|
|
module = runnerctl()
|
|
path = tmp_path / "runner.json"
|
|
write_config(path, **changes)
|
|
with pytest.raises(module.ConfigError, match=match):
|
|
module.load_config(path, expected_uid=os.getuid())
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value",
|
|
[
|
|
None,
|
|
"",
|
|
"REQUIRED",
|
|
"secret://",
|
|
"env://RUNNER_TOKEN",
|
|
"secret://runtime/aios-spark-runner/gateway/../admin",
|
|
"secret://other/aios-spark-runner/gateway/v1",
|
|
],
|
|
)
|
|
def test_missing_or_invalid_secret_ref_fails_closed(tmp_path: Path, value: str | None) -> None:
|
|
module = runnerctl()
|
|
path = tmp_path / "runner.json"
|
|
write_config(path, secret_ref=value)
|
|
with pytest.raises(module.ConfigError, match="BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE"):
|
|
module.load_config(path, expected_uid=os.getuid())
|
|
|
|
def test_json_schema_locks_exact_gateway_secret_scope() -> None:
|
|
schema = json.loads(
|
|
(DEPLOY / "runner-config.schema.json").read_text(encoding="utf-8")
|
|
)
|
|
assert schema["additionalProperties"] is False
|
|
assert set(schema["required"]) == set(schema["properties"])
|
|
pattern = schema["properties"]["runner_secret_ref"]["pattern"]
|
|
assert re.fullmatch(
|
|
pattern, "secret://runtime/aios-spark-runner/gateway/v1"
|
|
)
|
|
for rejected in [
|
|
"secret://runtime/aios-spark-runner/gateway/../admin",
|
|
"secret://runtime/aios-spark-runner/gateway/",
|
|
"secret://other/aios-spark-runner/gateway/v1",
|
|
]:
|
|
assert re.fullmatch(pattern, rejected) is None
|
|
|
|
|
|
def test_offline_config_can_validate_without_resolving_secret(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
path = tmp_path / "runner.json"
|
|
write_config(path, secret_ref=None)
|
|
config = module.load_config(path, require_secret_ref=False, expected_uid=os.getuid())
|
|
assert config.runner_secret_ref is None
|
|
assert "secret" not in repr(config).lower()
|
|
|
|
|
|
def test_inline_secret_fields_are_rejected(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
path = tmp_path / "runner.json"
|
|
data = write_config(path)
|
|
data["token"] = "must-not-be-accepted"
|
|
path.write_text(json.dumps(data), encoding="utf-8")
|
|
path.chmod(0o600)
|
|
with pytest.raises(module.ConfigError, match="unknown"):
|
|
module.load_config(path, expected_uid=os.getuid())
|
|
|
|
|
|
def test_config_leaf_symlink_is_rejected(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
real = tmp_path / "real.json"
|
|
link = tmp_path / "link.json"
|
|
write_config(real)
|
|
link.symlink_to(real)
|
|
with pytest.raises(module.ConfigError, match="symlink"):
|
|
module.load_config(link, expected_uid=os.getuid())
|
|
|
|
|
|
def test_config_ancestor_symlink_is_rejected(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
real_dir = tmp_path / "real"
|
|
real_dir.mkdir()
|
|
link_dir = tmp_path / "link"
|
|
link_dir.symlink_to(real_dir, target_is_directory=True)
|
|
path = real_dir / "runner.json"
|
|
write_config(path)
|
|
with pytest.raises(module.ConfigError, match="symlink"):
|
|
module.load_config(link_dir / "runner.json", expected_uid=os.getuid())
|
|
|
|
|
|
def test_permissive_config_mode_is_rejected(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
path = tmp_path / "runner.json"
|
|
write_config(path)
|
|
path.chmod(0o666)
|
|
with pytest.raises(module.ConfigError, match="writable"):
|
|
module.load_config(path, expected_uid=os.getuid())
|
|
|
|
|
|
def test_runtime_config_mode_is_exactly_enforced(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
path = tmp_path / "runner.json"
|
|
write_config(path)
|
|
path.chmod(0o644)
|
|
with pytest.raises(module.ConfigError, match="mode"):
|
|
module.load_config(path, expected_uid=os.getuid())
|
|
|
|
|
|
def test_wrong_config_owner_is_rejected(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
path = tmp_path / "runner.json"
|
|
write_config(path)
|
|
with pytest.raises(module.ConfigError, match="owner"):
|
|
module.load_config(path, expected_uid=os.getuid() + 1)
|
|
|
|
|
|
def test_artifact_hash_is_verified(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
artifact = tmp_path / "artifact.tar.gz"
|
|
artifact.write_bytes(b"runner")
|
|
artifact.chmod(0o600)
|
|
digest = module.sha256_file(artifact)
|
|
assert module.validate_artifact(artifact, f"sha256:{digest}", expected_uid=os.getuid()) == digest
|
|
|
|
|
|
def test_artifact_hash_mismatch_is_rejected(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
artifact = tmp_path / "artifact.tar.gz"
|
|
artifact.write_bytes(b"runner")
|
|
artifact.chmod(0o600)
|
|
with pytest.raises(module.ConfigError, match="digest mismatch"):
|
|
module.validate_artifact(artifact, f"sha256:{'0' * 64}", expected_uid=os.getuid())
|
|
|
|
|
|
def test_artifact_symlink_is_rejected(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
artifact = tmp_path / "artifact.tar.gz"
|
|
link = tmp_path / "link.tar.gz"
|
|
artifact.write_bytes(b"runner")
|
|
artifact.chmod(0o600)
|
|
link.symlink_to(artifact)
|
|
with pytest.raises(module.ConfigError, match="symlink"):
|
|
module.validate_artifact(link, f"sha256:{'0' * 64}", expected_uid=os.getuid())
|
|
|
|
def test_artifact_ancestor_symlink_is_rejected(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
real = tmp_path / "real"
|
|
real.mkdir()
|
|
artifact = real / "artifact.tar.gz"
|
|
artifact.write_bytes(b"runner")
|
|
artifact.chmod(0o600)
|
|
link = tmp_path / "linked"
|
|
link.symlink_to(real, target_is_directory=True)
|
|
with pytest.raises(module.ConfigError, match="symlink"):
|
|
module.validate_artifact(
|
|
link / artifact.name, f"sha256:{'0' * 64}", expected_uid=os.getuid()
|
|
)
|
|
|
|
|
|
def test_heartbeat_output_ancestor_symlink_is_rejected(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
config_path = tmp_path / "runner.json"
|
|
write_config(config_path)
|
|
config = module.load_config(config_path, expected_uid=os.getuid())
|
|
real = tmp_path / "real-run"
|
|
real.mkdir(mode=0o700)
|
|
link = tmp_path / "linked-run"
|
|
link.symlink_to(real, target_is_directory=True)
|
|
with pytest.raises(module.ConfigError, match="symlink"):
|
|
module.write_heartbeat(config, link / "heartbeat.json", os.getpid(), module.utc_now())
|
|
assert not (real / "heartbeat.json").exists()
|
|
|
|
|
|
def test_heartbeat_is_atomic_schema_bound_and_private(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
config_path = tmp_path / "runner.json"
|
|
write_config(config_path)
|
|
config = module.load_config(config_path, expected_uid=os.getuid())
|
|
heartbeat = tmp_path / "run" / "heartbeat.json"
|
|
heartbeat.parent.mkdir(mode=0o700)
|
|
started = module.utc_now()
|
|
module.write_heartbeat(config, heartbeat, os.getpid(), started)
|
|
data = json.loads(heartbeat.read_text(encoding="utf-8"))
|
|
assert data == {
|
|
"schema_version": 1,
|
|
"service": "aios-spark-runner",
|
|
"pid": os.getpid(),
|
|
"started_at": started,
|
|
"heartbeat_at": data["heartbeat_at"],
|
|
"artifact_digest": "b" * 64,
|
|
"source_commit": SOURCE_COMMIT,
|
|
"status": "healthy",
|
|
}
|
|
assert stat.S_IMODE(heartbeat.stat().st_mode) == 0o600
|
|
assert not list(heartbeat.parent.glob(".heartbeat.json.*"))
|
|
|
|
|
|
def test_runtime_config_write_fsyncs_file_and_directory(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
module = runnerctl()
|
|
config_path = tmp_path / "runner.json"
|
|
write_config(config_path)
|
|
config = module.load_config(config_path, expected_uid=os.getuid())
|
|
heartbeat = tmp_path / "run" / "heartbeat.json"
|
|
heartbeat.parent.mkdir(mode=0o700)
|
|
securefs = importlib.import_module("tools.aios_spark_runner.securefs")
|
|
original = securefs.os.fsync
|
|
calls: list[int] = []
|
|
|
|
def recording_fsync(fd: int) -> None:
|
|
calls.append(fd)
|
|
original(fd)
|
|
|
|
monkeypatch.setattr(securefs.os, "fsync", recording_fsync)
|
|
module.write_heartbeat(config, heartbeat, os.getpid(), module.utc_now())
|
|
assert len(calls) >= 2
|
|
|
|
|
|
def test_heartbeat_freshness_and_pid_are_verified(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
config_path = tmp_path / "runner.json"
|
|
write_config(config_path)
|
|
config = module.load_config(config_path, expected_uid=os.getuid())
|
|
heartbeat = tmp_path / "heartbeat.json"
|
|
module.write_heartbeat(config, heartbeat, os.getpid(), module.utc_now())
|
|
result = module.check_heartbeat(
|
|
config, heartbeat, max_age_seconds=30, expected_uid=os.getuid()
|
|
)
|
|
assert result["status"] == "healthy"
|
|
data = json.loads(heartbeat.read_text(encoding="utf-8"))
|
|
data["heartbeat_at"] = "2000-01-01T00:00:00Z"
|
|
heartbeat.write_text(json.dumps(data), encoding="utf-8")
|
|
heartbeat.chmod(0o600)
|
|
with pytest.raises(module.ConfigError, match="stale"):
|
|
module.check_heartbeat(config, heartbeat, max_age_seconds=30, expected_uid=os.getuid())
|
|
|
|
|
|
def test_heartbeat_wrong_pid_or_digest_is_rejected(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
config_path = tmp_path / "runner.json"
|
|
write_config(config_path)
|
|
config = module.load_config(config_path, expected_uid=os.getuid())
|
|
heartbeat = tmp_path / "heartbeat.json"
|
|
module.write_heartbeat(config, heartbeat, os.getpid(), module.utc_now())
|
|
data = json.loads(heartbeat.read_text(encoding="utf-8"))
|
|
data["pid"] = 999999999
|
|
heartbeat.write_text(json.dumps(data), encoding="utf-8")
|
|
heartbeat.chmod(0o600)
|
|
with pytest.raises(module.ConfigError, match="PID"):
|
|
module.check_heartbeat(config, heartbeat, max_age_seconds=30, expected_uid=os.getuid())
|
|
data["pid"] = os.getpid()
|
|
data["artifact_digest"] = "0" * 64
|
|
heartbeat.write_text(json.dumps(data), encoding="utf-8")
|
|
heartbeat.chmod(0o600)
|
|
with pytest.raises(module.ConfigError, match="artifact digest"):
|
|
module.check_heartbeat(config, heartbeat, max_age_seconds=30, expected_uid=os.getuid())
|
|
|
|
|
|
def test_worktree_probe_creates_reads_and_cleans_branch(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
source = tmp_path / "source"
|
|
bare = tmp_path / "repo.git"
|
|
source.mkdir()
|
|
subprocess.run(["git", "init", "-q", "-b", "main"], cwd=source, check=True)
|
|
subprocess.run(["git", "config", "user.name", "Test"], cwd=source, check=True)
|
|
subprocess.run(["git", "config", "user.email", "test@example.invalid"], cwd=source, check=True)
|
|
(source / "README").write_text("fixture\n", encoding="utf-8")
|
|
subprocess.run(["git", "add", "README"], cwd=source, check=True)
|
|
subprocess.run(["git", "commit", "-qm", "fixture"], cwd=source, check=True)
|
|
commit = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=source, text=True).strip()
|
|
subprocess.run(["git", "clone", "-q", "--bare", str(source), str(bare)], check=True)
|
|
before = subprocess.check_output(
|
|
["git", f"--git-dir={bare}", "rev-parse", "refs/heads/main"], text=True
|
|
).strip()
|
|
result = module.probe_worktree(bare, commit, tmp_path / "probes")
|
|
after = subprocess.check_output(
|
|
["git", f"--git-dir={bare}", "rev-parse", "refs/heads/main"], text=True
|
|
).strip()
|
|
assert result == {"repo_read": "PASS", "temp_worktree": "PASS", "cleanup": "PASS"}
|
|
assert before == after
|
|
assert not (tmp_path / "probes").exists()
|
|
refs = subprocess.check_output(
|
|
["git", f"--git-dir={bare}", "for-each-ref", "--format=%(refname)", "refs/heads/probe-"],
|
|
text=True,
|
|
)
|
|
assert refs == ""
|
|
|
|
|
|
def test_build_artifact_is_deterministic_and_bound(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
private, public = signing_keys(tmp_path)
|
|
first = module.build_artifact(
|
|
SOURCE_COMMIT, tmp_path / "one", signing_key=private, source_repo=ROOT
|
|
)
|
|
second = module.build_artifact(
|
|
SOURCE_COMMIT, tmp_path / "two", signing_key=private, source_repo=ROOT
|
|
)
|
|
assert Path(first["artifact"]).read_bytes() == Path(second["artifact"]).read_bytes()
|
|
assert first["sha256"] == second["sha256"]
|
|
checksum = Path(first["sha256_file"])
|
|
assert checksum.read_text(encoding="ascii") == (
|
|
f"{first['sha256']} {Path(first['artifact']).name}\n"
|
|
)
|
|
assert stat.S_IMODE(checksum.stat().st_mode) == 0o600
|
|
assert Path(first["sha256_file"]).name.endswith(".tar.gz.sha256")
|
|
manifest = json.loads(Path(first["manifest"]).read_text(encoding="utf-8"))
|
|
assert manifest["source_commit"] == SOURCE_COMMIT
|
|
assert set(manifest) == {
|
|
"source_commit", "artifact_digest", "sha256", "signature", "created_at"
|
|
}
|
|
assert manifest["artifact_digest"] == f"sha256:{first['sha256']}"
|
|
assert manifest["sha256"] == first["sha256"]
|
|
verified = module.verify_manifest(
|
|
Path(first["artifact"]),
|
|
Path(first["manifest"]),
|
|
public,
|
|
ROOT,
|
|
expected_uid=os.getuid(),
|
|
)
|
|
assert verified == manifest
|
|
config = json.loads(Path(first["config"]).read_text(encoding="utf-8"))
|
|
assert config["runner_artifact_sha256_or_signature"] == f"sha256:{first['sha256']}"
|
|
assert config["runner_secret_ref"] is None
|
|
|
|
|
|
def test_provision_missing_secret_fails_without_output(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
artifact, config, manifest, public = build_artifact(tmp_path)
|
|
root = tmp_path / "root"
|
|
with pytest.raises(module.ConfigError, match="BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE"):
|
|
module.provision(
|
|
artifact, manifest, public, ROOT, config, root,
|
|
test_mode=True, expected_uid=os.getuid()
|
|
)
|
|
assert not (root / "opt/aios-spark-runner/current").exists()
|
|
|
|
def test_production_provision_requires_broker_availability_probe(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
artifact, config_path, manifest, public = build_artifact(tmp_path)
|
|
data = json.loads(config_path.read_text(encoding="utf-8"))
|
|
data["runner_secret_ref"] = "secret://runtime/aios-spark-runner/gateway/v1"
|
|
config_path.write_text(json.dumps(data), encoding="utf-8")
|
|
config_path.chmod(0o600)
|
|
root = tmp_path / "must-not-exist"
|
|
with pytest.raises(
|
|
module.ConfigError, match="BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE"
|
|
):
|
|
module.provision(
|
|
artifact,
|
|
manifest,
|
|
public,
|
|
ROOT,
|
|
config_path,
|
|
root,
|
|
test_mode=False,
|
|
expected_uid=os.getuid(),
|
|
)
|
|
assert not root.exists()
|
|
|
|
|
|
def test_gateway_secret_ref_probe_is_value_free_and_exact(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
probe_path = tmp_path / "secret-ref-probe"
|
|
probe_path.write_text(
|
|
"#!/bin/sh\n"
|
|
"[ \"$1\" = check ] || exit 2\n"
|
|
"IFS= read -r ref\n"
|
|
"[ \"$ref\" = secret://runtime/aios-spark-runner/gateway/v1 ] || exit 3\n"
|
|
"printf 'AVAILABLE\\n'\n",
|
|
encoding="utf-8",
|
|
)
|
|
probe_path.chmod(0o700)
|
|
probe = module.make_secret_ref_probe(
|
|
probe_path, expected_uid=os.getuid(), timeout_seconds=2
|
|
)
|
|
assert probe("secret://runtime/aios-spark-runner/gateway/v1") is True
|
|
assert probe("secret://runtime/aios-spark-runner/gateway/v2") is False
|
|
|
|
|
|
@pytest.mark.parametrize("unsafe", ["symlink", "writable", "relative"])
|
|
def test_gateway_probe_executable_boundary_is_fail_closed(
|
|
tmp_path: Path, unsafe: str, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
module = runnerctl()
|
|
executable = tmp_path / "probe"
|
|
executable.write_text("#!/bin/sh\nprintf 'AVAILABLE\\n'\n", encoding="utf-8")
|
|
executable.chmod(0o700)
|
|
candidate = executable
|
|
if unsafe == "symlink":
|
|
candidate = tmp_path / "probe-link"
|
|
candidate.symlink_to(executable)
|
|
elif unsafe == "writable":
|
|
executable.chmod(0o722)
|
|
else:
|
|
monkeypatch.chdir(tmp_path)
|
|
candidate = Path("probe")
|
|
with pytest.raises(module.ConfigError, match="BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE"):
|
|
module.make_secret_ref_probe(candidate, expected_uid=os.getuid())
|
|
|
|
|
|
def test_forged_provenance_is_rejected(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
artifact, _, manifest, public = build_artifact(tmp_path)
|
|
original = json.loads(manifest.read_text(encoding="utf-8"))
|
|
for field, value, match in [
|
|
("sha256", "0" * 64, "SHA-256"),
|
|
("artifact_digest", f"sha256:{'0' * 64}", "digest"),
|
|
("signature", "ed25519:" + "A" * 88, "signature"),
|
|
("source_commit", "f" * 40, "actual Git commit"),
|
|
]:
|
|
forged = dict(original)
|
|
forged[field] = value
|
|
manifest.write_text(json.dumps(forged), encoding="utf-8")
|
|
manifest.chmod(0o600)
|
|
with pytest.raises(module.ConfigError, match=match):
|
|
module.verify_manifest(
|
|
artifact, manifest, public, ROOT, expected_uid=os.getuid()
|
|
)
|
|
manifest.write_text(json.dumps(original), encoding="utf-8")
|
|
manifest.chmod(0o600)
|
|
|
|
|
|
def test_artifact_manifest_schema_requires_full_provenance() -> None:
|
|
schema = json.loads(
|
|
(DEPLOY / "artifact-manifest.schema.json").read_text(encoding="utf-8")
|
|
)
|
|
required = {
|
|
"source_commit", "artifact_digest", "sha256", "signature", "created_at"
|
|
}
|
|
assert set(schema["required"]) == required
|
|
assert set(schema["properties"]) == required
|
|
assert schema["additionalProperties"] is False
|
|
|
|
def test_provision_preserves_existing_parent_permissions_and_normalizes_config(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
module = runnerctl()
|
|
artifact, config_path, manifest, public = build_artifact(tmp_path)
|
|
data = json.loads(config_path.read_text(encoding="utf-8"))
|
|
data["runner_secret_ref"] = "secret://runtime/aios-spark-runner/gateway/v1"
|
|
config_path.write_text(json.dumps(data, indent=2), encoding="utf-8")
|
|
config_path.chmod(0o600)
|
|
root = tmp_path / "root"
|
|
root.mkdir(mode=0o700)
|
|
opt = root / "opt"
|
|
opt.mkdir(mode=0o755)
|
|
result = module.provision(
|
|
artifact, manifest, public, ROOT, config_path, root,
|
|
test_mode=True, expected_uid=os.getuid()
|
|
)
|
|
assert stat.S_IMODE(opt.stat().st_mode) == 0o755
|
|
installed = json.loads(
|
|
(root / "etc/aios-spark-runner/runner.json").read_text(encoding="utf-8")
|
|
)
|
|
assert set(installed) == set(data)
|
|
assert installed["runner_secret_ref"] == data["runner_secret_ref"]
|
|
module.rollback(root, Path(result["receipt"]), test_mode=True)
|
|
|
|
|
|
def test_provision_is_idempotent_and_rollback_removes_first_install(tmp_path: Path) -> None:
|
|
module = runnerctl()
|
|
artifact, config_path, manifest, public = build_artifact(tmp_path)
|
|
data = json.loads(config_path.read_text(encoding="utf-8"))
|
|
data["runner_secret_ref"] = "secret://runtime/aios-spark-runner/gateway/v1"
|
|
config_path.write_text(json.dumps(data), encoding="utf-8")
|
|
config_path.chmod(0o600)
|
|
root = tmp_path / "root"
|
|
first = module.provision(
|
|
artifact, manifest, public, ROOT, config_path, root,
|
|
test_mode=True, expected_uid=os.getuid()
|
|
)
|
|
current = root / "opt/aios-spark-runner/current"
|
|
assert current.is_dir() and not current.is_symlink()
|
|
marker = current / "manifest.json"
|
|
before = marker.read_bytes()
|
|
second = module.provision(
|
|
artifact, manifest, public, ROOT, config_path, root,
|
|
test_mode=True, expected_uid=os.getuid()
|
|
)
|
|
assert marker.read_bytes() == before
|
|
assert first["artifact_sha256"] == second["artifact_sha256"]
|
|
receipt = Path(second["receipt"])
|
|
first_rollback = module.rollback(root, receipt, test_mode=True)
|
|
second_rollback = module.rollback(root, receipt, test_mode=True)
|
|
assert first_rollback["rollback_verified"] == "PASS"
|
|
assert second_rollback == first_rollback
|
|
assert not current.exists()
|
|
transaction = json.loads(receipt.read_text(encoding="utf-8"))
|
|
assert transaction["install_transaction_id"]
|
|
assert transaction["previous_artifact"] is None
|
|
assert transaction["rollback_target"] is None
|
|
assert transaction["rollback_verified"] is True
|
|
assert not list((root / "var/lib/aios-spark-runner").glob("rollback-*"))
|
|
|
|
|
|
def test_failed_upgrade_automatically_restores_previous_artifact(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
module = runnerctl()
|
|
artifact, config_path, manifest, public = build_artifact(tmp_path / "build")
|
|
data = json.loads(config_path.read_text(encoding="utf-8"))
|
|
data["runner_secret_ref"] = "secret://runtime/aios-spark-runner/gateway/v1"
|
|
config_path.write_text(json.dumps(data), encoding="utf-8")
|
|
config_path.chmod(0o600)
|
|
root = tmp_path / "root"
|
|
first = module.provision(
|
|
artifact, manifest, public, ROOT, config_path, root,
|
|
test_mode=True, expected_uid=os.getuid()
|
|
)
|
|
receipt_before = Path(first["receipt"]).read_bytes()
|
|
marker = root / "opt/aios-spark-runner/current/manifest.json"
|
|
marker_data = json.loads(marker.read_text(encoding="utf-8"))
|
|
marker_data["artifact_digest"] = "0" * 64
|
|
marker.write_text(json.dumps(marker_data), encoding="utf-8")
|
|
marker.chmod(0o600)
|
|
installed_before = marker.read_bytes()
|
|
installer_globals = module.provision.__globals__
|
|
original_write = installer_globals["atomic_write"]
|
|
failed = False
|
|
|
|
def fail_once(path: Path, payload: bytes, *, mode: int = 0o600) -> None:
|
|
nonlocal failed
|
|
if not failed and str(path).endswith("etc/aios-spark-runner/runner.json"):
|
|
failed = True
|
|
raise RuntimeError("injected install failure")
|
|
original_write(path, payload, mode=mode)
|
|
|
|
monkeypatch.setitem(installer_globals, "atomic_write", fail_once)
|
|
with pytest.raises(RuntimeError, match="injected install failure"):
|
|
module.provision(
|
|
artifact, manifest, public, ROOT, config_path, root,
|
|
test_mode=True, expected_uid=os.getuid()
|
|
)
|
|
assert marker.read_bytes() == installed_before
|
|
assert Path(first["receipt"]).read_bytes() == receipt_before
|
|
|
|
def test_systemd_unit_has_mandatory_sandbox_and_identity() -> None:
|
|
unit = ROOT / "deploy/systemd/aios-spark-runner.service"
|
|
assert unit.is_file()
|
|
text = unit.read_text(encoding="utf-8")
|
|
for directive in [
|
|
"User=aios-spark-runner",
|
|
"Group=aios-spark-runner",
|
|
"WorkingDirectory=/var/lib/aios-spark-runner",
|
|
"UMask=0077",
|
|
"NoNewPrivileges=true",
|
|
"PrivateTmp=true",
|
|
"ProtectSystem=strict",
|
|
"MemoryDenyWriteExecute=true",
|
|
"ProtectHome=true",
|
|
"ProtectKernelTunables=true",
|
|
"ProtectKernelModules=true",
|
|
"ProtectControlGroups=true",
|
|
"RestrictSUIDSGID=true",
|
|
"LockPersonality=true",
|
|
"RestrictRealtime=true",
|
|
"CapabilityBoundingSet=",
|
|
"AmbientCapabilities=",
|
|
"RuntimeDirectory=aios-spark-runner",
|
|
"SystemCallFilter=@system-service",
|
|
"ReadOnlyPaths=/data/aios-gitops/repo.git",
|
|
"StateDirectory=aios-spark-runner",
|
|
"ReadWritePaths=/var/lib/aios-spark-runner /run/aios-spark-runner",
|
|
"InaccessiblePaths=/data/aios-gitops/worktrees/controller",
|
|
]:
|
|
assert directive in text
|
|
assert "User=root" not in text
|
|
assert "Environment=" not in text
|
|
|
|
|
|
def test_sysusers_tmpfiles_and_unit_identity_are_consistent() -> None:
|
|
sysusers = (ROOT / "deploy/sysusers.d/aios-spark-runner.conf").read_text(encoding="utf-8")
|
|
tmpfiles = (ROOT / "deploy/tmpfiles.d/aios-spark-runner.conf").read_text(encoding="utf-8")
|
|
assert "aios-spark-runner" in sysusers
|
|
assert "/usr/sbin/nologin" in sysusers
|
|
assert "d /var/lib/aios-spark-runner 0700 aios-spark-runner aios-spark-runner" in tmpfiles
|
|
assert "d /run/aios-spark-runner 0700 aios-spark-runner aios-spark-runner" in tmpfiles
|
|
assert "d /etc/aios-spark-runner 0750 root aios-spark-runner" in tmpfiles
|
|
|
|
|
|
def test_shell_scripts_are_fail_closed_and_do_not_download_or_daemonize() -> None:
|
|
for name in ["build-artifact.sh", "provision.sh", "rollback.sh", "uninstall.sh", "acceptance.sh"]:
|
|
path = DEPLOY / name
|
|
assert path.is_file()
|
|
text = path.read_text(encoding="utf-8")
|
|
assert "set -euo pipefail" in text
|
|
assert "curl | sh" not in text
|
|
assert "nohup" not in text
|
|
assert "tmux" not in text
|
|
assert "git reset" not in text
|
|
assert "git rebase" not in text
|
|
assert "git commit --amend" not in text
|
|
assert "push main" not in text
|
|
|
|
def test_build_artifact_shell_entrypoint_executes_real_cli(tmp_path: Path) -> None:
|
|
output = tmp_path / "shell-dist"
|
|
private, _ = signing_keys(tmp_path)
|
|
result = subprocess.run(
|
|
[
|
|
str(DEPLOY / "build-artifact.sh"),
|
|
"--commit",
|
|
SOURCE_COMMIT,
|
|
"--signing-key",
|
|
str(private),
|
|
"--source-repo",
|
|
str(ROOT),
|
|
"--output",
|
|
str(output),
|
|
],
|
|
cwd=ROOT,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
assert result.returncode == 0, result.stderr
|
|
response = json.loads(result.stdout)
|
|
assert Path(response["artifact"]).is_file()
|
|
assert Path(response["sha256_file"]).is_file()
|
|
|
|
|
|
def test_readme_keeps_install_blocked_without_secret_ref() -> None:
|
|
path = DEPLOY / "README.md"
|
|
assert path.is_file()
|
|
text = path.read_text(encoding="utf-8")
|
|
assert "BLOCKED_GATEWAY_SECRET_REF_UNAVAILABLE" in text
|
|
assert "secret://" in text
|
|
assert "curl | sh" not in text
|
|
assert "tmux" not in text
|
|
assert "nohup" not in text
|