287 lines
13 KiB
Python
287 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
APP = ROOT / "active-runtime" / "app"
|
|
sys.path.insert(0, str(APP))
|
|
|
|
from signer_contract_compatibility import ( # noqa: E402
|
|
SignerContractError,
|
|
_prepare,
|
|
canonical_json_bytes,
|
|
deployed_signer_help,
|
|
run_compatible_signer,
|
|
resolve_ref_oid,
|
|
run_compatible_validator,
|
|
sha256_json,
|
|
)
|
|
|
|
|
|
SYNTHETIC_BASE_OID = "1" * 40
|
|
BASE_OID = SYNTHETIC_BASE_OID
|
|
REPOSITORY = "aios/aios"
|
|
|
|
|
|
class SignerContractCompatibilityTests(unittest.TestCase):
|
|
def test_canonical_json_is_stable(self) -> None:
|
|
value = {"z": 1, "nested": {"b": 2, "a": 1}, "a": [3, 2, 1]}
|
|
self.assertEqual(
|
|
canonical_json_bytes(value),
|
|
canonical_json_bytes({"a": [3, 2, 1], "z": 1, "nested": {"a": 1, "b": 2}}),
|
|
)
|
|
self.assertEqual(len(sha256_json(value)), 64)
|
|
|
|
def test_nested_request_normalizes_to_deployed_flat_contract(self) -> None:
|
|
now = datetime.now(timezone.utc)
|
|
request = {
|
|
"schema": "aios.task-scoped-writer-grant-request.v1",
|
|
"task_id": "V2R1C6_SIGNER_COMPATIBILITY_UNIT",
|
|
"repository": REPOSITORY,
|
|
"base": {"ref": "refs/heads/main", "oid": BASE_OID},
|
|
"target": {"branch": "agent/v2r1c6-signer-contract-unit-20260806"},
|
|
"execution_contract": {"writer_route": "POST /api/commit-file"},
|
|
"owner_authorization": {
|
|
"identity": "aiman",
|
|
"reference": "owner-authorization:unit:20260806",
|
|
"authenticated_role": "owner",
|
|
},
|
|
"allowed_operations": [
|
|
"create_worktree",
|
|
"create_target_branch_if_absent",
|
|
"commit_to_target_branch",
|
|
"push_target_branch",
|
|
],
|
|
"denied_operations": [
|
|
"direct_main_push",
|
|
"force_push",
|
|
"tag_create",
|
|
"tag_update",
|
|
"tag_delete",
|
|
"branch_delete",
|
|
"repository_admin",
|
|
"baseline_write",
|
|
"production_deploy",
|
|
"production_marker_write",
|
|
"history_rewrite",
|
|
],
|
|
"one_time_use": True,
|
|
"authorization": {
|
|
"production_change_authorized": False,
|
|
"tag_change_authorized": False,
|
|
"baseline_change_authorized": False,
|
|
"main_write_authorized": False,
|
|
},
|
|
"issued_at": now.isoformat().replace("+00:00", "Z"),
|
|
"not_before": (now - timedelta(seconds=2)).isoformat().replace("+00:00", "Z"),
|
|
"expires_at": (now + timedelta(seconds=300)).isoformat().replace("+00:00", "Z"),
|
|
"nonce": "unit-nonce",
|
|
}
|
|
plan = {
|
|
"schema": "aios.task-scoped-writer-approval-plan.v1",
|
|
"approval_plan_id": "sha256:" + sha256_json(request),
|
|
"grant_request_sha256": sha256_json(request),
|
|
"owner_authorization_reference": "owner-authorization:unit:20260806",
|
|
"owner_identity": "aiman",
|
|
}
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
grant_dir = Path(temporary)
|
|
(grant_dir / "approval-plan.json").write_text(json.dumps(plan), encoding="utf-8")
|
|
(grant_dir / "grant-request.json").write_text(json.dumps(request), encoding="utf-8")
|
|
prepared = _prepare(grant_dir)
|
|
flat = json.loads((grant_dir / "credential-request.json").read_text(encoding="utf-8"))
|
|
grant = json.loads((grant_dir / "grant-payload.json").read_text(encoding="utf-8"))
|
|
self.assertEqual(flat["repository"], REPOSITORY)
|
|
self.assertEqual(flat["base_oid"], BASE_OID)
|
|
self.assertEqual(flat["target_branch"], request["target"]["branch"])
|
|
self.assertEqual(flat["owner_authorization_reference"], request["owner_authorization"]["reference"])
|
|
self.assertEqual(grant["credential_secret_ref"], "task-writer:" + prepared["grant_id"])
|
|
self.assertEqual(grant["grant_id"], prepared["grant_id"])
|
|
|
|
def test_real_signer_help_and_self_test_are_available(self) -> None:
|
|
signer = "/usr/local/libexec/aios-grant-signer-adapter"
|
|
if not Path(signer).is_file():
|
|
self.skipTest("deployed signer is not present")
|
|
if os.geteuid() != 0:
|
|
self.skipTest("deployed signer self-test requires root")
|
|
help_result = subprocess.run([signer, "sign", "--help"], capture_output=True, text=True, check=False)
|
|
self.assertEqual(help_result.returncode, 0)
|
|
self.assertTrue(deployed_signer_help(help_result.stdout, help_result.stderr))
|
|
self_test = subprocess.run([signer, "self-test"], capture_output=True, text=True, check=False)
|
|
self.assertEqual(self_test.returncode, 0)
|
|
self.assertIn("SIGNER_SELF_TEST=PASS", self_test.stdout)
|
|
|
|
def test_resolve_ref_oid_uses_full_synthetic_commit_and_fails_closed(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
repo = Path(temporary) / "synthetic.git"
|
|
subprocess.run(["git", "init", "--bare", str(repo)], check=True, capture_output=True)
|
|
tree = subprocess.check_output(
|
|
["git", "--git-dir", str(repo), "mktree"],
|
|
input="",
|
|
text=True,
|
|
).strip()
|
|
env = os.environ.copy()
|
|
env.update({
|
|
"GIT_AUTHOR_NAME": "synthetic",
|
|
"GIT_AUTHOR_EMAIL": "synthetic@example.invalid",
|
|
"GIT_COMMITTER_NAME": "synthetic",
|
|
"GIT_COMMITTER_EMAIL": "synthetic@example.invalid",
|
|
})
|
|
commit = subprocess.check_output(
|
|
["git", "--git-dir", str(repo), "commit-tree", tree],
|
|
input="synthetic commit\n",
|
|
text=True,
|
|
env=env,
|
|
).strip()
|
|
subprocess.run(
|
|
["git", "--git-dir", str(repo), "update-ref", "refs/heads/main", commit],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
self.assertEqual(resolve_ref_oid(repo, "refs/heads/main"), commit)
|
|
with self.assertRaisesRegex(SignerContractError, "authoritative_ref_unavailable"):
|
|
resolve_ref_oid(repo, "refs/heads/missing")
|
|
with self.assertRaisesRegex(SignerContractError, "authoritative_ref_invalid"):
|
|
resolve_ref_oid(repo, "main")
|
|
with patch(
|
|
"signer_contract_compatibility.subprocess.run",
|
|
return_value=subprocess.CompletedProcess(
|
|
args=[],
|
|
returncode=0,
|
|
stdout="deadbeef\\n",
|
|
stderr="",
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(SignerContractError, "authoritative_ref_invalid_oid"):
|
|
resolve_ref_oid(repo, "refs/heads/main")
|
|
|
|
def test_source_does_not_pin_the_pre_merge_main_oid(self) -> None:
|
|
source = Path(__file__).read_text(encoding="utf-8")
|
|
obsolete = "".join((
|
|
"875d41c3", "10109314", "94142920", "67fd700d", "c214d4d3",
|
|
))
|
|
self.assertEqual(len(obsolete), 40)
|
|
self.assertNotIn(obsolete, source)
|
|
|
|
def test_unsigned_legacy_result_is_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
signer = root / "legacy-signer"
|
|
signer.write_text(
|
|
"#!/usr/bin/python3\n"
|
|
"import json,sys\n"
|
|
"from pathlib import Path\n"
|
|
"if '--help' in sys.argv: print('--grant-dir --result-json --self-test'); raise SystemExit(0)\n"
|
|
"args=dict(zip(sys.argv[1::2],sys.argv[2::2]))\n"
|
|
"Path(args['--result-json']).write_text(json.dumps({'status':'SIGNED','grant_id':'g','grant_digest_sha256':'d','signing_key_identity':'k','signature_present':False}))\n",
|
|
encoding="utf-8",
|
|
)
|
|
signer.chmod(0o700)
|
|
grant_dir = root / "grant"
|
|
grant_dir.mkdir()
|
|
(grant_dir / "approval-plan.json").write_text("{}", encoding="utf-8")
|
|
(grant_dir / "grant-request.json").write_text("{}", encoding="utf-8")
|
|
with self.assertRaises(SignerContractError) as captured:
|
|
run_compatible_signer([str(signer)], grant_dir, root / "result.json")
|
|
self.assertEqual(str(captured.exception), "legacy_signature_missing")
|
|
|
|
@unittest.skipUnless(
|
|
os.environ.get("AIOS_RUN_REAL_SIGNER_TESTS") == "1",
|
|
"live signer integration is opt-in",
|
|
)
|
|
def test_real_signer_and_validator_round_trip(self) -> None:
|
|
now = datetime.now(timezone.utc)
|
|
branch = "agent/v2r1c6-signer-contract-unit-live-20260806"
|
|
live_base_oid = resolve_ref_oid(
|
|
os.getenv("AIOS_GITOPS_AUTHORITATIVE_GIT_DIR", "/data/aios-gitops/repo.git"),
|
|
"refs/heads/main",
|
|
)
|
|
request = {
|
|
"schema": "aios.task-scoped-writer-grant-request.v1",
|
|
"task_id": "V2R1C6_SIGNER_COMPATIBILITY_LIVE_UNIT",
|
|
"repository": REPOSITORY,
|
|
"base": {"ref": "refs/heads/main", "oid": live_base_oid},
|
|
"target": {"branch": branch},
|
|
"execution_contract": {"writer_route": "POST /api/commit-file"},
|
|
"owner_authorization": {
|
|
"identity": "aiman",
|
|
"reference": "owner-authorization:live-unit:20260806",
|
|
"authenticated_role": "owner",
|
|
},
|
|
"allowed_operations": [
|
|
"create_worktree",
|
|
"create_target_branch_if_absent",
|
|
"commit_to_target_branch",
|
|
"push_target_branch",
|
|
],
|
|
"denied_operations": [
|
|
"direct_main_push",
|
|
"force_push",
|
|
"tag_create",
|
|
"tag_update",
|
|
"tag_delete",
|
|
"branch_delete",
|
|
"repository_admin",
|
|
"baseline_write",
|
|
"production_deploy",
|
|
"production_marker_write",
|
|
"history_rewrite",
|
|
],
|
|
"one_time_use": True,
|
|
"authorization": {
|
|
"production_change_authorized": False,
|
|
"tag_change_authorized": False,
|
|
"baseline_change_authorized": False,
|
|
"main_write_authorized": False,
|
|
},
|
|
"issued_at": now.isoformat().replace("+00:00", "Z"),
|
|
"not_before": (now - timedelta(seconds=2)).isoformat().replace("+00:00", "Z"),
|
|
"expires_at": (now + timedelta(seconds=300)).isoformat().replace("+00:00", "Z"),
|
|
"nonce": "live-unit-nonce",
|
|
}
|
|
plan = {
|
|
"schema": "aios.task-scoped-writer-approval-plan.v1",
|
|
"approval_plan_id": "sha256:" + sha256_json(request),
|
|
"grant_request_sha256": sha256_json(request),
|
|
"owner_authorization_reference": request["owner_authorization"]["reference"],
|
|
"owner_identity": "aiman",
|
|
}
|
|
old_repo = os.environ.get("AIOS_GITOPS_VALIDATOR_REPO_PATH")
|
|
os.environ["AIOS_GITOPS_VALIDATOR_REPO_PATH"] = os.getenv("AIOS_GITOPS_AUTHORITATIVE_GIT_DIR", "/data/aios-gitops/repo.git")
|
|
try:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
grant_dir = root / "grant"
|
|
grant_dir.mkdir(mode=0o700)
|
|
(grant_dir / "approval-plan.json").write_text(json.dumps(plan), encoding="utf-8")
|
|
(grant_dir / "grant-request.json").write_text(json.dumps(request), encoding="utf-8")
|
|
signer_result = root / "signer-result.json"
|
|
validator_result = root / "validator-result.json"
|
|
run_compatible_signer(["/usr/local/libexec/aios-grant-signer-adapter"], grant_dir, signer_result)
|
|
prepared = json.loads((grant_dir / "credential-request.json").read_text(encoding="utf-8"))
|
|
self.assertEqual(prepared["base_oid"], live_base_oid)
|
|
run_compatible_validator(["/usr/local/libexec/aios-grant-validator"], grant_dir, validator_result)
|
|
signed = json.loads(signer_result.read_text(encoding="utf-8"))
|
|
validated = json.loads(validator_result.read_text(encoding="utf-8"))
|
|
self.assertEqual(signed["status"], "SIGNED")
|
|
self.assertTrue(signed["signature_present"])
|
|
self.assertEqual(validated["status"], "VALID")
|
|
self.assertTrue(all(validated["checks"].values()))
|
|
finally:
|
|
if old_repo is None:
|
|
os.environ.pop("AIOS_GITOPS_VALIDATOR_REPO_PATH", None)
|
|
else:
|
|
os.environ["AIOS_GITOPS_VALIDATOR_REPO_PATH"] = old_repo
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|