132 lines
4.7 KiB
Python
132 lines
4.7 KiB
Python
"""Local, non-SSH tests for the bounded Phase2 transport and status contract."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import subprocess
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
|
|
SCRIPT = Path(__file__).parents[1] / "scripts" / "phase2_introspect_and_freeze.py"
|
|
SPEC = importlib.util.spec_from_file_location("phase2_introspect_and_freeze", SCRIPT)
|
|
assert SPEC and SPEC.loader
|
|
phase2 = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(phase2)
|
|
|
|
|
|
def test_transport_sends_large_payload_without_argv_growth():
|
|
payload = b"x" * (256 * 1024)
|
|
observed = {}
|
|
|
|
def runner(argv, **kwargs):
|
|
observed["argv"] = argv
|
|
observed["kwargs"] = kwargs
|
|
return SimpleNamespace(returncode=0, stdout=b'{"ok":true}')
|
|
|
|
assert phase2.run_remote_python(payload, runner=runner) == b'{"ok":true}'
|
|
assert observed["kwargs"]["input"] == payload
|
|
argv = observed["argv"]
|
|
assert argv == [
|
|
"ssh.exe",
|
|
"-T",
|
|
"-o",
|
|
"BatchMode=yes",
|
|
"-o",
|
|
"ConnectTimeout=12",
|
|
"hw",
|
|
'test "$(hostname)" = cn-network-mkt && sudo -n python3 -',
|
|
]
|
|
assert sum(len(part.encode("utf-8")) + 1 for part in argv) <= 512
|
|
assert all(payload not in part.encode("utf-8") for part in argv)
|
|
assert all("base64" not in part.lower() for part in argv)
|
|
|
|
|
|
def test_transport_reports_timeout_without_stderr():
|
|
def runner(argv, **kwargs):
|
|
raise subprocess.TimeoutExpired(argv, kwargs["timeout"], stderr=b"secret=do-not-leak")
|
|
|
|
with pytest.raises(phase2.RemoteTransportError, match="timed out") as error:
|
|
phase2.run_remote_python(b"payload", runner=runner, timeout=3)
|
|
assert "secret" not in str(error.value)
|
|
|
|
|
|
def test_transport_reports_nonzero_without_stderr():
|
|
def runner(argv, **kwargs):
|
|
return SimpleNamespace(returncode=23, stdout=b"", stderr=b"token=do-not-leak")
|
|
|
|
with pytest.raises(phase2.RemoteTransportError, match="status 23") as error:
|
|
phase2.run_remote_python(b"payload", runner=runner)
|
|
assert "token" not in str(error.value)
|
|
|
|
|
|
def test_status_flags_are_explicit_and_missing_public_symbol_fails_closed():
|
|
valid = phase2.validate_public_function_ast(
|
|
"def offline_migrate_task_principal_schema():\n pass\n"
|
|
"def supersede_p0_scope():\n pass\n"
|
|
)
|
|
assert valid["ok"] is True
|
|
|
|
missing = phase2.validate_public_function_ast(
|
|
"def offline_migrate_task_principal_schema():\n pass\n"
|
|
)
|
|
assert missing["ok"] is False
|
|
assert missing["missing"] == ["supersede_p0_scope"]
|
|
|
|
status = phase2.make_phase2_status(
|
|
"FAIL", errors=["PUBLIC_SYMBOL_MISSING:supersede_p0_scope"]
|
|
)
|
|
assert status["status"] == "FAIL"
|
|
assert status["fail_closed"] is True
|
|
assert all(status[flag] is False for flag in phase2.MUTATION_FLAGS)
|
|
|
|
|
|
def test_dynamic_reexport_is_accepted_only_with_explicit_globals_export_mechanism():
|
|
source = (
|
|
"for _name, _value in _legacy.__dict__.items():\n"
|
|
" globals()[_name] = _value\n"
|
|
)
|
|
result = phase2.validate_public_function_ast(source)
|
|
assert result["ok"] is True
|
|
assert result["dynamic_reexport"] is True
|
|
|
|
no_export = "x = 1\n"
|
|
result = phase2.validate_public_function_ast(no_export)
|
|
assert result["ok"] is False
|
|
assert result["missing"] == list(phase2.DEFAULT_REQUIRED_ADMISSION_FUNCTIONS)
|
|
|
|
|
|
def test_status_validator_rejects_mutation_claims():
|
|
status = phase2.make_phase2_status("PASS")
|
|
assert phase2.validate_phase2_status(status)["read_only"] is True
|
|
status["git_mutation"] = True
|
|
with pytest.raises(phase2.Phase2StatusError, match="git_mutation"):
|
|
phase2.validate_phase2_status(status)
|
|
|
|
|
|
def test_request_payload_contains_request_but_argv_does_not():
|
|
payload = phase2.build_remote_payload(
|
|
{
|
|
"issuer_pid": 123,
|
|
"expected_admission_sha256": "a" * 64,
|
|
}
|
|
)
|
|
assert b'"issuer_pid":123' in payload
|
|
assert b'"expected_admission_sha256":"' in payload
|
|
argv = phase2.build_ssh_argv()
|
|
assert all("123" not in part and "aaaa" not in part for part in argv)
|
|
|
|
|
|
def test_live_defaults_use_current_issuer_and_storage_migration_module():
|
|
assert phase2.DEFAULT_ISSUER_UNIT == "aios-task-principal-issuer.service"
|
|
assert phase2.DEFAULT_MIGRATION_MODULE == "aios_task_principal_storage_migration"
|
|
payload = phase2.build_remote_payload()
|
|
assert b"aios-task-principal-issuer.service" in payload
|
|
assert b"aios_task_principal_storage_migration" in payload
|
|
stale_issuer = "aios-task-principal-" + "admission.service"
|
|
stale_migration = "aios_task_principal_" + "migration"
|
|
assert stale_issuer.encode() not in payload
|
|
assert stale_migration.encode() not in payload
|