404 lines
15 KiB
Python
404 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import secrets
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
APP = ROOT / "active-runtime" / "app"
|
|
sys.path.insert(0, str(APP))
|
|
|
|
from aios_task_identity_overlay_api import OverlayServer # noqa: E402
|
|
from authorized_task_writer_provisioning import ProvisioningError # noqa: E402
|
|
from owner_authorization_broker import ACTION # noqa: E402
|
|
from task_identity_overlay import ( # noqa: E402
|
|
IdentityOverlayConfig,
|
|
IdentityOverlayError,
|
|
OVERLAY_SCOPE,
|
|
TaskBinding,
|
|
TaskIdentityOverlayService,
|
|
TaskIdentityOverlayStore,
|
|
)
|
|
|
|
|
|
TASK_ID = "V2R1C6_TASK_SCOPED_WRITER_PROVISIONING"
|
|
BRANCH = "agent/v2r1c6-mainline-reconciliation-20260805"
|
|
REPOSITORY = "aios/aios"
|
|
BASE_OID = "875d41c3101093149414292067fd700dc214d4d3"
|
|
OWNER = "aiman"
|
|
|
|
|
|
class FakeProvisioner:
|
|
def __init__(self) -> None:
|
|
self.calls = []
|
|
|
|
def provision(self, raw, owner):
|
|
self.calls.append((dict(raw), owner))
|
|
return {
|
|
"status": "READY_FOR_CODE_AUTHORING",
|
|
"approval_plan_id": "sha256:" + "1" * 64,
|
|
"grant_id": "grant-overlay-test",
|
|
"grant_digest_sha256": "2" * 64,
|
|
"branch": BRANCH,
|
|
"branch_oid": BASE_OID,
|
|
"base_oid": BASE_OID,
|
|
"grant_consumed": True,
|
|
"writer_scope": {
|
|
"workspace_role": "task-scoped-isolated-writer",
|
|
"writes_enabled": True,
|
|
"allowed_branch": BRANCH,
|
|
"allowed_base_oid": BASE_OID,
|
|
"allowed_routes": ["POST /api/commit-file"],
|
|
"allow_main_write": False,
|
|
"allow_tag_write": False,
|
|
"allow_baseline_write": False,
|
|
"allow_force_push": False,
|
|
},
|
|
}
|
|
|
|
|
|
class BlockingProvisioner:
|
|
def provision(self, raw, owner):
|
|
raise ProvisioningError("AUTHORIZED_BASE_OID_DRIFT", 409)
|
|
|
|
|
|
class IdentityOverlayCase(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name)
|
|
self.config = IdentityOverlayConfig(
|
|
state_root=self.root / "state",
|
|
secret_root=self.root / "run",
|
|
broker_state_root=self.root / "broker",
|
|
binding=TaskBinding(
|
|
task_id=TASK_ID,
|
|
repository=REPOSITORY,
|
|
base_ref="refs/heads/main",
|
|
base_oid=BASE_OID,
|
|
target_branch=BRANCH,
|
|
),
|
|
owner_identity=OWNER,
|
|
max_ttl_seconds=900,
|
|
require_root_issue=False,
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
self.temp.cleanup()
|
|
|
|
def request(self, **updates):
|
|
value = {
|
|
"action": ACTION,
|
|
"task_id": TASK_ID,
|
|
"repository": REPOSITORY,
|
|
"base_ref": "refs/heads/main",
|
|
"base_oid": BASE_OID,
|
|
"target_branch": BRANCH,
|
|
"one_time_use": True,
|
|
"expires_in_seconds": 900,
|
|
"production_change_authorized": False,
|
|
"tag_change_authorized": False,
|
|
"baseline_change_authorized": False,
|
|
"main_write_authorized": False,
|
|
"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",
|
|
],
|
|
}
|
|
value.update(updates)
|
|
return value
|
|
|
|
def issue(self, store=None):
|
|
return (store or TaskIdentityOverlayStore(self.config)).issue(
|
|
ttl_seconds=900,
|
|
authorization_assertion_ref="owner-session:V2R1C6:approved",
|
|
operator_identity="infra-agent",
|
|
)
|
|
|
|
def token(self, result):
|
|
return Path(result["secret_ref"]).read_text().strip()
|
|
|
|
def test_issue_creates_root_only_secret_and_hashed_metadata(self):
|
|
result = self.issue()
|
|
secret = Path(result["secret_ref"])
|
|
self.assertTrue(secret.is_file())
|
|
self.assertEqual(0, secret.stat().st_mode & 0o077)
|
|
metadata = json.loads(
|
|
(
|
|
self.config.state_root
|
|
/ result["overlay_id"]
|
|
/ "metadata.json"
|
|
).read_text()
|
|
)
|
|
self.assertEqual("ACTIVE", metadata["status"])
|
|
self.assertNotIn(self.token(result), json.dumps(metadata))
|
|
self.assertEqual(OVERLAY_SCOPE, metadata["scope"])
|
|
self.assertFalse(metadata["wildcard_scope"])
|
|
|
|
def test_issue_is_idempotent_for_same_active_binding(self):
|
|
first = self.issue()
|
|
second = self.issue()
|
|
self.assertTrue(second["idempotent"])
|
|
self.assertEqual(first["overlay_id"], second["overlay_id"])
|
|
self.assertEqual(first["secret_ref"], second["secret_ref"])
|
|
|
|
def test_root_is_required_when_configured(self):
|
|
config = IdentityOverlayConfig(
|
|
**{
|
|
**self.config.__dict__,
|
|
"require_root_issue": True,
|
|
}
|
|
)
|
|
with mock.patch("task_identity_overlay.os.geteuid", return_value=1000):
|
|
with self.assertRaises(IdentityOverlayError) as caught:
|
|
self.issue(TaskIdentityOverlayStore(config))
|
|
self.assertEqual("OVERLAY_ROOT_ISSUER_REQUIRED", caught.exception.code)
|
|
|
|
def test_collision_with_existing_secret_ref_is_rejected(self):
|
|
existing = self.root / "existing.token"
|
|
existing.write_text("fixed-token\n")
|
|
config = IdentityOverlayConfig(
|
|
**{
|
|
**self.config.__dict__,
|
|
"collision_secret_refs": (existing,),
|
|
}
|
|
)
|
|
with mock.patch(
|
|
"task_identity_overlay.secrets.token_urlsafe",
|
|
return_value="fixed-token",
|
|
):
|
|
with self.assertRaises(IdentityOverlayError) as caught:
|
|
self.issue(TaskIdentityOverlayStore(config))
|
|
self.assertEqual(
|
|
"OVERLAY_SECRET_IDENTITY_COLLISION", caught.exception.code
|
|
)
|
|
|
|
def test_exact_request_authenticates_and_locks_overlay(self):
|
|
store = TaskIdentityOverlayStore(self.config)
|
|
issued = self.issue(store)
|
|
session = store.authenticate_and_lock(
|
|
self.token(issued), self.request()
|
|
)
|
|
self.assertEqual(issued["overlay_id"], session.overlay_id)
|
|
status = store.status(session.overlay_id)
|
|
self.assertEqual("LOCKED", status["status"])
|
|
self.assertEqual(1, status["attempt_count"])
|
|
|
|
def test_wrong_branch_is_rejected_without_consuming_overlay(self):
|
|
store = TaskIdentityOverlayStore(self.config)
|
|
issued = self.issue(store)
|
|
with self.assertRaises(IdentityOverlayError) as caught:
|
|
store.authenticate_and_lock(
|
|
self.token(issued), self.request(target_branch="agent/other")
|
|
)
|
|
self.assertEqual("OVERLAY_BINDING_MISMATCH", caught.exception.code)
|
|
self.assertEqual("ACTIVE", store.status(issued["overlay_id"])["status"])
|
|
self.assertTrue(Path(issued["secret_ref"]).exists())
|
|
|
|
def test_wrong_token_is_rejected(self):
|
|
store = TaskIdentityOverlayStore(self.config)
|
|
self.issue(store)
|
|
with self.assertRaises(IdentityOverlayError) as caught:
|
|
store.authenticate_and_lock("not-the-token", self.request())
|
|
self.assertEqual("OVERLAY_CREDENTIAL_INVALID", caught.exception.code)
|
|
|
|
def test_same_token_cannot_authorize_different_request(self):
|
|
store = TaskIdentityOverlayStore(self.config)
|
|
issued = self.issue(store)
|
|
token = self.token(issued)
|
|
store.authenticate_and_lock(token, self.request())
|
|
changed = self.request(expires_in_seconds=600)
|
|
with self.assertRaises(IdentityOverlayError) as caught:
|
|
store.authenticate_and_lock(token, changed)
|
|
self.assertEqual(
|
|
"OVERLAY_REPLAY_REQUEST_MISMATCH", caught.exception.code
|
|
)
|
|
|
|
def test_complete_consumes_and_removes_secret(self):
|
|
store = TaskIdentityOverlayStore(self.config)
|
|
issued = self.issue(store)
|
|
session = store.authenticate_and_lock(
|
|
self.token(issued), self.request()
|
|
)
|
|
metadata = store.complete(session, {"status": "READY_FOR_CODE_AUTHORING"})
|
|
self.assertEqual("CONSUMED", metadata["status"])
|
|
self.assertFalse(Path(issued["secret_ref"]).exists())
|
|
|
|
def test_fail_revokes_secret_and_records_error(self):
|
|
store = TaskIdentityOverlayStore(self.config)
|
|
issued = self.issue(store)
|
|
session = store.authenticate_and_lock(
|
|
self.token(issued), self.request()
|
|
)
|
|
metadata = store.fail(
|
|
session, error_code="TEST_BLOCKED", error_status=409
|
|
)
|
|
self.assertEqual("FAILED", metadata["status"])
|
|
self.assertEqual("TEST_BLOCKED", metadata["error_code"])
|
|
self.assertFalse(Path(issued["secret_ref"]).exists())
|
|
|
|
def test_revoke_removes_active_secret(self):
|
|
store = TaskIdentityOverlayStore(self.config)
|
|
issued = self.issue(store)
|
|
metadata = store.revoke(
|
|
issued["overlay_id"], reason="task-cancelled"
|
|
)
|
|
self.assertEqual("REVOKED", metadata["status"])
|
|
self.assertFalse(Path(issued["secret_ref"]).exists())
|
|
|
|
def test_service_calls_existing_broker_and_consumes_overlay(self):
|
|
fake = FakeProvisioner()
|
|
store = TaskIdentityOverlayStore(self.config)
|
|
issued = self.issue(store)
|
|
result = TaskIdentityOverlayService(
|
|
self.config, provisioner=fake
|
|
).provision(self.request(), self.token(issued))
|
|
self.assertEqual("READY_FOR_CODE_AUTHORING", result["status"])
|
|
self.assertTrue(result["identity_overlay_consumed"])
|
|
self.assertTrue(result["identity_overlay_secret_removed"])
|
|
self.assertEqual(1, len(fake.calls))
|
|
_, owner = fake.calls[0]
|
|
self.assertEqual(OWNER, owner.identity)
|
|
self.assertTrue(owner.authorization_reference.startswith("owner-auth:"))
|
|
|
|
def test_service_failure_marks_overlay_failed(self):
|
|
store = TaskIdentityOverlayStore(self.config)
|
|
issued = self.issue(store)
|
|
with self.assertRaises(ProvisioningError):
|
|
TaskIdentityOverlayService(
|
|
self.config, provisioner=BlockingProvisioner()
|
|
).provision(self.request(), self.token(issued))
|
|
status = store.status(issued["overlay_id"])
|
|
self.assertEqual("FAILED", status["status"])
|
|
self.assertEqual("AUTHORIZED_BASE_OID_DRIFT", status["error_code"])
|
|
self.assertFalse(Path(issued["secret_ref"]).exists())
|
|
|
|
def test_wildcard_and_production_capabilities_are_rejected(self):
|
|
store = TaskIdentityOverlayStore(self.config)
|
|
issued = self.issue(store)
|
|
with self.assertRaises(IdentityOverlayError) as caught:
|
|
store.authenticate_and_lock(
|
|
self.token(issued),
|
|
self.request(production_change_authorized=True),
|
|
)
|
|
self.assertEqual(
|
|
"OVERLAY_PRODUCTION_CAPABILITY_REJECTED", caught.exception.code
|
|
)
|
|
|
|
|
|
def _http_server(self, provisioner=None):
|
|
os.environ["AIOS_GITOPS_IDENTITY_OVERLAY_ENABLED"] = "true"
|
|
server = OverlayServer(
|
|
("127.0.0.1", 0),
|
|
TaskIdentityOverlayService(
|
|
self.config, provisioner=provisioner or FakeProvisioner()
|
|
),
|
|
)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
self.addCleanup(lambda: thread.join(timeout=5))
|
|
self.addCleanup(server.server_close)
|
|
self.addCleanup(server.shutdown)
|
|
return server
|
|
|
|
def _http_post(self, server, body, token=None):
|
|
host, port = server.server_address
|
|
headers = {"Content-Type": "application/json"}
|
|
if token is not None:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
request = urllib.request.Request(
|
|
f"http://{host}:{port}/api/governance/owner-authorizations",
|
|
data=json.dumps(body).encode("utf-8"),
|
|
method="POST",
|
|
headers=headers,
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=5) as response:
|
|
return response.status, json.loads(response.read())
|
|
except urllib.error.HTTPError as exc:
|
|
return exc.code, json.loads(exc.read())
|
|
|
|
def test_http_health_is_loopback_overlay(self):
|
|
server = self._http_server()
|
|
host, port = server.server_address
|
|
with urllib.request.urlopen(
|
|
f"http://{host}:{port}/healthz", timeout=5
|
|
) as response:
|
|
body = json.loads(response.read())
|
|
self.assertEqual(200, response.status)
|
|
self.assertEqual("active", body["status"])
|
|
self.assertFalse(body["global_token_registry_used"])
|
|
self.assertTrue(body["loopback_only"])
|
|
|
|
def test_http_unauthenticated_request_is_401(self):
|
|
server = self._http_server()
|
|
status, body = self._http_post(server, self.request())
|
|
self.assertEqual(401, status)
|
|
self.assertEqual("OVERLAY_AUTHENTICATION_REQUIRED", body["error"])
|
|
|
|
def test_http_wrong_token_is_401(self):
|
|
server = self._http_server()
|
|
self.issue()
|
|
status, body = self._http_post(
|
|
server, self.request(), token="wrong-token"
|
|
)
|
|
self.assertEqual(401, status)
|
|
self.assertEqual("OVERLAY_CREDENTIAL_INVALID", body["error"])
|
|
|
|
|
|
def test_http_wrong_token_cannot_probe_policy(self):
|
|
server = self._http_server()
|
|
self.issue()
|
|
status, body = self._http_post(
|
|
server,
|
|
self.request(production_change_authorized=True),
|
|
token="wrong-token",
|
|
)
|
|
self.assertEqual(401, status)
|
|
self.assertEqual("OVERLAY_CREDENTIAL_INVALID", body["error"])
|
|
|
|
def test_http_production_capability_is_rejected_without_consuming(self):
|
|
server = self._http_server()
|
|
issued = self.issue()
|
|
status, body = self._http_post(
|
|
server,
|
|
self.request(production_change_authorized=True),
|
|
token=self.token(issued),
|
|
)
|
|
self.assertEqual(403, status)
|
|
self.assertEqual(
|
|
"OVERLAY_PRODUCTION_CAPABILITY_REJECTED", body["error"]
|
|
)
|
|
store = TaskIdentityOverlayStore(self.config)
|
|
self.assertEqual(
|
|
"ACTIVE", store.status(issued["overlay_id"])["status"]
|
|
)
|
|
self.assertTrue(Path(issued["secret_ref"]).exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|