216 lines
8.1 KiB
Python
216 lines
8.1 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
APP = ROOT / "active-runtime" / "app"
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
"aios_durable_owner_authorization",
|
|
APP / "aios_durable_owner_authorization.py",
|
|
)
|
|
BROKER_MODULE = importlib.util.module_from_spec(SPEC)
|
|
import sys
|
|
sys.modules[SPEC.name] = BROKER_MODULE
|
|
assert SPEC.loader is not None
|
|
SPEC.loader.exec_module(BROKER_MODULE)
|
|
|
|
AuthorizationError = BROKER_MODULE.AuthorizationError
|
|
Broker = BROKER_MODULE.Broker
|
|
Constraints = BROKER_MODULE.Constraints
|
|
DurablePolicy = BROKER_MODULE.DurablePolicy
|
|
ReceiptStore = BROKER_MODULE.ReceiptStore
|
|
Upstream = BROKER_MODULE.Upstream
|
|
|
|
CALLER = "chatgpt-governed-implementer-singlehost"
|
|
SCOPE = "release:candidate"
|
|
GOVERNED_TASK_SCOPE = "governed-task:project"
|
|
SOURCE = "257c7a8df433f75b364bcaf69cbbc32e8c2e5d38"
|
|
|
|
|
|
def policy(*, expires_at=None, capabilities=None):
|
|
return DurablePolicy(
|
|
grant_id="grant-caller-projection-v1",
|
|
owner_identity="owner-durable-policy",
|
|
account=CALLER,
|
|
capabilities=frozenset(
|
|
capabilities or {GOVERNED_TASK_SCOPE, SCOPE}
|
|
),
|
|
issued_at=datetime.now(timezone.utc) - timedelta(minutes=1),
|
|
not_before=datetime.now(timezone.utc) - timedelta(minutes=1),
|
|
persistent_until_revoked=True,
|
|
constraints=Constraints(
|
|
repository="aios/aios",
|
|
account=CALLER,
|
|
allowed_target_prefix="agent/",
|
|
release_version_prefix="V2R2C",
|
|
max_task_risk="engineering",
|
|
),
|
|
policy_sha256="a" * 64,
|
|
expires_at=expires_at,
|
|
)
|
|
|
|
|
|
def make_broker(root):
|
|
return Broker(
|
|
policy_path=root / "policy.json",
|
|
revocation_path=root / "revocations.json",
|
|
store=ReceiptStore(root / "receipts.sqlite3"),
|
|
upstream=Upstream("http://127.0.0.1:1", "not-used"),
|
|
expected_account=CALLER,
|
|
expected_peer_user="test-peer",
|
|
allowed_peer_users=frozenset({"test-peer"}),
|
|
)
|
|
|
|
|
|
class CallerProjectionTests(unittest.TestCase):
|
|
def project(self, broker, request=None):
|
|
return broker.project_caller(
|
|
CALLER,
|
|
SCOPE,
|
|
request or {"source_commit": SOURCE, "version": "V2R2C2"},
|
|
idempotency_key="caller-projection-test-v1",
|
|
)
|
|
|
|
def test_positive_projection_ack_and_effective_scope(self):
|
|
broker = make_broker(Path(tempfile.mkdtemp()))
|
|
with patch.object(broker, "policy", return_value=policy()):
|
|
value = self.project(broker)
|
|
credential = value.pop("_credential")
|
|
self.assertGreaterEqual(len(credential), 32)
|
|
self.assertEqual("ACKED", value["projection_state"])
|
|
self.assertEqual("REHYDRATED", value["credential_state"])
|
|
self.assertEqual(
|
|
[GOVERNED_TASK_SCOPE, SCOPE],
|
|
value["effective_scopes"],
|
|
)
|
|
self.assertFalse(value["secret_exposed"])
|
|
|
|
def test_governed_task_projection_uses_durable_capability(self):
|
|
broker = make_broker(Path(tempfile.mkdtemp()))
|
|
with patch.object(broker, "policy", return_value=policy()):
|
|
value = broker.project_caller(
|
|
CALLER,
|
|
GOVERNED_TASK_SCOPE,
|
|
{},
|
|
idempotency_key="caller-projection-governed-task-v1",
|
|
)
|
|
value.pop("_credential")
|
|
self.assertEqual(GOVERNED_TASK_SCOPE, value["scope"])
|
|
self.assertEqual(
|
|
[GOVERNED_TASK_SCOPE, SCOPE],
|
|
value["effective_scopes"],
|
|
)
|
|
self.assertFalse(value["secret_exposed"])
|
|
|
|
def test_secret_material_is_not_durable(self):
|
|
root = Path(tempfile.mkdtemp())
|
|
broker = make_broker(root)
|
|
with patch.object(broker, "policy", return_value=policy()):
|
|
value = self.project(broker)
|
|
credential = value.pop("_credential")
|
|
durable = (root / "receipts.sqlite3").read_bytes()
|
|
self.assertNotIn(credential.encode(), durable)
|
|
self.assertNotIn("credential_material", durable.decode("utf-8", "ignore"))
|
|
self.assertFalse(value["secret_exposed"])
|
|
|
|
def test_caller_mismatch_is_denied(self):
|
|
broker = make_broker(Path(tempfile.mkdtemp()))
|
|
with self.assertRaises(AuthorizationError) as caught:
|
|
broker.project_caller(
|
|
"other-caller",
|
|
SCOPE,
|
|
{},
|
|
idempotency_key="caller-projection-test-v1",
|
|
)
|
|
self.assertEqual("CALLER_MISMATCH", caught.exception.code)
|
|
|
|
def test_scope_mismatch_is_denied(self):
|
|
broker = make_broker(Path(tempfile.mkdtemp()))
|
|
with self.assertRaises(AuthorizationError) as caught:
|
|
broker.project_caller(
|
|
CALLER,
|
|
"release:deploy",
|
|
{},
|
|
idempotency_key="caller-projection-test-v1",
|
|
)
|
|
self.assertEqual("SCOPE_MISMATCH", caught.exception.code)
|
|
|
|
def test_policy_missing_revoked_and_expired_fail_closed(self):
|
|
broker = make_broker(Path(tempfile.mkdtemp()))
|
|
for code in ("POLICY_FILE_INVALID", "OWNER_GRANT_REVOKED", "POLICY_EXPIRED"):
|
|
with self.subTest(code=code):
|
|
with patch.object(
|
|
broker,
|
|
"policy",
|
|
side_effect=AuthorizationError(code, 403),
|
|
):
|
|
with self.assertRaises(AuthorizationError) as caught:
|
|
self.project(broker)
|
|
self.assertEqual(code, caught.exception.code)
|
|
|
|
def test_invalid_projection_state_fails_closed(self):
|
|
with self.assertRaises(AuthorizationError) as caught:
|
|
Broker.validate_caller_projection(
|
|
{
|
|
"caller": CALLER,
|
|
"scope": SCOPE,
|
|
"projection_state": "CORRUPT",
|
|
"credential_state": "REHYDRATED",
|
|
"rehydration_state": "RECOVERED",
|
|
"effective_scopes": [SCOPE],
|
|
"secret_exposed": False,
|
|
}
|
|
)
|
|
self.assertEqual("PROJECTION_STATE_INVALID", caught.exception.code)
|
|
|
|
def test_restart_rehydrates_from_durable_metadata(self):
|
|
root = Path(tempfile.mkdtemp())
|
|
first = make_broker(root)
|
|
with patch.object(first, "policy", return_value=policy()):
|
|
initial = self.project(first)
|
|
first_credential = initial.pop("_credential")
|
|
second = make_broker(root)
|
|
with patch.object(second, "policy", return_value=policy()):
|
|
recovered = self.project(second)
|
|
second_credential = recovered.pop("_credential")
|
|
self.assertNotEqual(first_credential, second_credential)
|
|
self.assertEqual("RECOVERED", recovered["rehydration_state"])
|
|
self.assertEqual("ACKED", recovered["projection_state"])
|
|
|
|
def test_expired_policy_loader_rejects(self):
|
|
root = Path(tempfile.mkdtemp())
|
|
policy_path = root / "policy.json"
|
|
payload = {
|
|
"schema": "aios.owner-durable-authorization.v1",
|
|
"grant_id": "grant-caller-projection-v1",
|
|
"owner_identity": "owner-durable-policy",
|
|
"account": CALLER,
|
|
"capabilities": [SCOPE],
|
|
"issued_at": "2026-08-22T00:00:00Z",
|
|
"not_before": "2026-08-22T00:00:00Z",
|
|
"expires_at": "2026-08-22T01:00:00Z",
|
|
"persistent_until_revoked": True,
|
|
"constraints": {
|
|
"repository": "aios/aios",
|
|
"allowed_target_prefix": "agent/",
|
|
"release_version_prefix": "V2R2C",
|
|
"max_task_risk": "engineering",
|
|
},
|
|
}
|
|
policy_path.write_text(json.dumps(payload), encoding="utf-8")
|
|
with patch.object(BROKER_MODULE, "validate_root_file", lambda path, **kwargs: path):
|
|
with self.assertRaises(AuthorizationError) as caught:
|
|
DurablePolicy.load(policy_path)
|
|
self.assertEqual("POLICY_EXPIRED", caught.exception.code)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|