AIOS Governed Continuation
abe81d7f64
Ported from authority branch agent/g7-security-critical-closure-20260914 (d5f5c9b, base=beabd053).
S1 auth: role/user from server token->identity mapping, ignore client X-AIOS-* headers.
S2 bootstrap credential: PBKDF2 random-salt, fail-closed SECRET, no plaintext fallback, session HMAC+expiry.
S3 path containment + control-plane file-read hardening.
S4 admin UI XSS + real governed approval + dashboard CSRF/branch validation.
S5 runtime delivery: registry.yaml conflict, openapi securityScheme, install.sh fail propagation, do_POST shadowing.
Tests: 5 negative suites (66 tests). Evidence pack in deliverables/g7-evidence/.
98 lines
3.7 KiB
Python
98 lines
3.7 KiB
Python
"""G7 security gate S5: runtime/delivery correctness tests."""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_registry_yaml_parses_without_conflict_markers() -> None:
|
|
p = ROOT / "agents" / "registry.yaml"
|
|
text = p.read_text(encoding="utf-8")
|
|
markers = [m for m in ("<<<<<<<", "=======", ">>>>>>>") if m in text]
|
|
assert not markers, f"registry.yaml still contains conflict markers: {markers}"
|
|
data = yaml.safe_load(text)
|
|
assert isinstance(data, dict)
|
|
# merged schema sanity: keys from both sides are present
|
|
assert "version" in data and "registry_version" in data
|
|
assert "commands" in data and "directories" in data
|
|
assert "lifecycle" in data and "audit" in data
|
|
assert isinstance(data["agents"], list)
|
|
|
|
|
|
def test_openapi_parses_and_security_refs_complete() -> None:
|
|
p = ROOT / "docs" / "gitops" / "openapi-chatgpt-actions.yaml"
|
|
spec = yaml.safe_load(p.read_text(encoding="utf-8"))
|
|
schemes = spec.get("components", {}).get("securitySchemes", {})
|
|
assert schemes, "openapi has no securitySchemes"
|
|
# walk every operation's security array and assert each scheme is defined
|
|
referenced: set[str] = set()
|
|
for path, item in spec.get("paths", {}).items():
|
|
for method, op in item.items():
|
|
if not isinstance(op, dict):
|
|
continue
|
|
for sec_ent in op.get("security", []) or []:
|
|
referenced.update(sec_ent.keys())
|
|
missing = referenced - set(schemes.keys())
|
|
assert not missing, f"security schemes referenced but undefined: {missing}"
|
|
# taskPrincipalAuth must now be defined
|
|
assert "taskPrincipalAuth" in schemes, "taskPrincipalAuth not defined"
|
|
|
|
|
|
def test_install_sh_propagates_service_failures() -> None:
|
|
p = ROOT / "deploy" / "install" / "install.sh"
|
|
text = p.read_text(encoding="utf-8")
|
|
# service enable/restart must not be masked by `|| true`
|
|
masked = re.findall(
|
|
r"systemctl (?:enable|restart) [a-z0-9.-]+\.service\s*\|\|\s*true",
|
|
text,
|
|
)
|
|
assert not masked, f"service start failures still masked: {masked}"
|
|
# sanity: set -e is on, a failed restart will now abort the script
|
|
assert "set -e" in text
|
|
|
|
|
|
def test_shell_scripts_have_no_undefined_variable_aborts() -> None:
|
|
candidates = [
|
|
ROOT / "delta-fix" / "apply_delta_fix_only.sh",
|
|
ROOT / "scripts" / "silent_install_v5.sh",
|
|
ROOT / "deploy" / "install" / "install.sh",
|
|
]
|
|
for p in candidates:
|
|
assert p.exists(), f"missing script: {p}"
|
|
# syntax check (prefer Git Bash; plain `bash` may resolve to WSL on Windows)
|
|
bash_candidates = [
|
|
r"C:\Program Files\Git\bin\bash.exe",
|
|
r"C:\Program Files (x86)\Git\bin\bash.exe",
|
|
"/usr/bin/bash",
|
|
"bash",
|
|
]
|
|
bash_bin = next((b for b in bash_candidates if Path(b).exists()), "bash")
|
|
r = subprocess.run(
|
|
[bash_bin, "-n", p.as_posix()], capture_output=True
|
|
)
|
|
err = r.stderr.decode("utf-8", errors="replace")
|
|
assert r.returncode == 0, f"bash -n failed for {p}:\n{err}"
|
|
text = p.read_text(encoding="utf-8")
|
|
# strict-mode scripts must not expand undefined variables
|
|
for bad in ("$PKG ", "$TARGET "):
|
|
assert bad not in text, f"{p.name} may expand undefined variable: {bad}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
|
|
failed = 0
|
|
for fn in fns:
|
|
try:
|
|
fn()
|
|
print(f"PASS {fn.__name__}")
|
|
except AssertionError as exc:
|
|
failed += 1
|
|
print(f"FAIL {fn.__name__}: {exc}")
|
|
sys.exit(1 if failed else 0)
|