authority client with delegation tests

Author: rcholic

examples/authority_client_local_policy.yaml

@@ -0,0 +1,9 @@
+rules:
+  - name: allow-orders-create
+    effect: allow
+    principals:
+      - agent:checkout
+    actions:
+      - http.post
+    resources:
+      - https://api.vendor.com/orders

examples/authority_client_local_yaml.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+
+def _ensure_repo_root_on_syspath() -> None:
+    repo_root = Path(__file__).resolve().parents[1]
+    root = str(repo_root)
+    if root not in sys.path:
+        sys.path.insert(0, root)
+
+
+def _build_request() -> object:
+    _ensure_repo_root_on_syspath()
+    from predicate_contracts import (  # pylint: disable=import-error
+        ActionRequest,
+        ActionSpec,
+        PrincipalRef,
+        StateEvidence,
+        VerificationEvidence,
+    )
+
+    return ActionRequest(
+        principal=PrincipalRef(principal_id="agent:checkout"),
+        action_spec=ActionSpec(
+            action="http.post",
+            resource="https://api.vendor.com/orders",
+            intent="submit customer order",
+        ),
+        state_evidence=StateEvidence(source="sdk-python", state_hash="sha256:example"),
+        verification_evidence=VerificationEvidence(),
+    )
+
+
+def run(policy_file: str, secret_key: str) -> dict[str, object]:
+    _ensure_repo_root_on_syspath()
+    from predicate_authority import AuthorityClient  # pylint: disable=import-error
+
+    context = AuthorityClient.from_policy_file(
+        policy_file=policy_file,
+        secret_key=secret_key,
+        ttl_seconds=120,
+    )
+    client = context.client
+    decision = client.authorize(_build_request())
+    token_verified = False
+    if decision.mandate is not None:
+        token_verified = client.verify_token(decision.mandate.token) is not None
+    return {
+        "policy_file": policy_file,
+        "allowed": decision.allowed,
+        "reason": decision.reason.value,
+        "token_issued": decision.mandate is not None,
+        "token_verified": token_verified,
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Local AuthorityClient example using YAML policy.")
+    parser.add_argument(
+        "--policy-file",
+        default="examples/authority_client_local_policy.yaml",
+        help="Path to local YAML policy file.",
+    )
+    parser.add_argument(
+        "--secret-key",
+        default="dev-secret",
+        help="Signing key used for local mandates.",
+    )
+    args = parser.parse_args()
+    payload = run(policy_file=args.policy_file, secret_key=args.secret_key)
+    print(json.dumps(payload, indent=2, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

examples/delegation/delegate.py

@@ -0,0 +1,155 @@
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import json
+import sys
+from collections.abc import Callable
+from pathlib import Path
+from typing import Any, cast
+
+
+def _ensure_repo_root_on_syspath() -> None:
+    repo_root = Path(__file__).resolve().parents[2]
+    root = str(repo_root)
+    if root not in sys.path:
+        sys.path.insert(0, root)
+
+
+def _build_request() -> object:
+    _ensure_repo_root_on_syspath()
+    from predicate_contracts import (  # pylint: disable=import-error
+        ActionRequest,
+        ActionSpec,
+        PrincipalRef,
+        StateEvidence,
+        VerificationEvidence,
+    )
+
+    return ActionRequest(
+        principal=PrincipalRef(principal_id="agent:root"),
+        action_spec=ActionSpec(
+            action="task.delegate",
+            resource="worker:queue/main",
+            intent="delegate processing to worker agent",
+        ),
+        state_evidence=StateEvidence(source="delegate.py", state_hash="sha256:delegate"),
+        verification_evidence=VerificationEvidence(),
+    )
+
+
+def _run_worker(
+    worker_script: str,
+    token: str,
+    secret_key: str,
+    revocation_file: str,
+    policy_file: str,
+) -> dict[str, object]:
+    worker_run = _load_worker_runner(worker_script)
+    payload = worker_run(
+        token=token,
+        secret_key=secret_key,
+        revocation_file=revocation_file,
+        policy_file=policy_file,
+    )
+    if not isinstance(payload, dict):
+        raise RuntimeError("worker payload must be an object")
+    return cast(dict[str, object], payload)
+
+
+def _load_worker_runner(worker_script: str) -> Callable[..., Any]:
+    worker_path = Path(worker_script).resolve()
+    spec = importlib.util.spec_from_file_location("delegation_worker_runtime", worker_path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Unable to load worker module from path: {worker_script}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    run_callable = getattr(module, "run", None)
+    if not callable(run_callable):
+        raise RuntimeError("Worker module must expose callable run(...) function.")
+    return cast(Callable[..., Any], run_callable)
+
+
+def run(
+    policy_file: str,
+    worker_script: str,
+    revocation_file: str,
+    secret_key: str,
+) -> dict[str, object]:
+    _ensure_repo_root_on_syspath()
+    from predicate_authority import AuthorityClient  # pylint: disable=import-error
+
+    context = AuthorityClient.from_policy_file(
+        policy_file=policy_file,
+        secret_key=secret_key,
+        ttl_seconds=120,
+    )
+    client = context.client
+
+    decision = client.authorize(_build_request())
+    if not decision.allowed or decision.mandate is None:
+        return {
+            "root_allowed": False,
+            "worker_allowed_before_revoke": False,
+            "worker_allowed_after_revoke": False,
+        }
+
+    token = decision.mandate.token
+    Path(revocation_file).write_text(
+        json.dumps({"revoked_principal_ids": []}, indent=2),
+        encoding="utf-8",
+    )
+    before = _run_worker(worker_script, token, secret_key, revocation_file, policy_file)
+
+    client.revoke_principal("agent:root")
+    Path(revocation_file).write_text(
+        json.dumps({"revoked_principal_ids": ["agent:root"]}, indent=2),
+        encoding="utf-8",
+    )
+    after = _run_worker(worker_script, token, secret_key, revocation_file, policy_file)
+
+    return {
+        "root_allowed": True,
+        "root_delegation_depth": decision.mandate.claims.delegation_depth,
+        "root_chain_hash": decision.mandate.claims.delegation_chain_hash,
+        "worker_allowed_before_revoke": bool(before.get("allowed", False)),
+        "worker_allowed_after_revoke": bool(after.get("allowed", False)),
+        "worker_delegation_depth_before_revoke": before.get("delegation_depth"),
+        "worker_chain_verified_before_revoke": bool(before.get("chain_verified", False)),
+        "before_reason": before.get("reason"),
+        "after_reason": after.get("reason"),
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Delegation simulation for local authority runtime."
+    )
+    parser.add_argument(
+        "--policy-file",
+        default="examples/delegation/policy.yaml",
+        help="Path to policy file for the root delegating agent.",
+    )
+    parser.add_argument(
+        "--worker-script",
+        default="examples/delegation/worker.py",
+        help="Path to worker.py.",
+    )
+    parser.add_argument(
+        "--revocation-file",
+        default="examples/delegation/revocations.json",
+        help="Path to revocation state shared with worker.",
+    )
+    parser.add_argument("--secret-key", default="dev-secret")
+    args = parser.parse_args()
+    payload = run(
+        policy_file=args.policy_file,
+        worker_script=args.worker_script,
+        revocation_file=args.revocation_file,
+        secret_key=args.secret_key,
+    )
+    print(json.dumps(payload, indent=2, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

examples/delegation/policy.yaml

@@ -0,0 +1,19 @@
+rules:
+  - name: allow-delegate-task
+    effect: allow
+    principals:
+      - agent:root
+    actions:
+      - task.delegate
+    resources:
+      - worker:queue/*
+    max_delegation_depth: 1
+  - name: allow-worker-execute
+    effect: allow
+    principals:
+      - agent:worker
+    actions:
+      - job.execute
+    resources:
+      - queue://jobs/*
+    max_delegation_depth: 1

examples/delegation/worker.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+
+def _ensure_repo_root_on_syspath() -> None:
+    repo_root = Path(__file__).resolve().parents[2]
+    root = str(repo_root)
+    if root not in sys.path:
+        sys.path.insert(0, root)
+
+
+def _load_revocations(path: str) -> list[str]:
+    file_path = Path(path)
+    if not file_path.exists():
+        return []
+    payload = json.loads(file_path.read_text(encoding="utf-8"))
+    if not isinstance(payload, dict):
+        return []
+    revoked = payload.get("revoked_principal_ids", [])
+    if not isinstance(revoked, list):
+        return []
+    return [str(item) for item in revoked]
+
+
+def _build_worker_request() -> object:
+    _ensure_repo_root_on_syspath()
+    from predicate_contracts import (  # pylint: disable=import-error
+        ActionRequest,
+        ActionSpec,
+        PrincipalRef,
+        StateEvidence,
+        VerificationEvidence,
+    )
+
+    return ActionRequest(
+        principal=PrincipalRef(principal_id="agent:worker"),
+        action_spec=ActionSpec(
+            action="job.execute",
+            resource="queue://jobs/high-priority",
+            intent="execute delegated job",
+        ),
+        state_evidence=StateEvidence(source="worker.py", state_hash="sha256:worker"),
+        verification_evidence=VerificationEvidence(),
+    )
+
+
+def run(
+    token: str,
+    secret_key: str,
+    revocation_file: str,
+    policy_file: str,
+) -> dict[str, object]:
+    _ensure_repo_root_on_syspath()
+    from predicate_authority import AuthorityClient  # pylint: disable=import-error
+
+    context = AuthorityClient.from_policy_file(
+        policy_file=policy_file,
+        secret_key=secret_key,
+        ttl_seconds=120,
+    )
+    client = context.client
+    revoked_principal_ids = _load_revocations(revocation_file)
+    for principal_id in revoked_principal_ids:
+        client.revoke_principal(principal_id)
+
+    parent_mandate = client.verify_token(token)
+    if parent_mandate is None:
+        if "agent:root" in revoked_principal_ids:
+            return {"allowed": False, "reason": "revoked_root_token"}
+        return {"allowed": False, "reason": "invalid_or_expired_token"}
+
+    decision = client.authorize(
+        _build_worker_request(),
+        parent_mandate=parent_mandate,
+    )
+    if not decision.allowed or decision.mandate is None:
+        denied_reason = (
+            "revoked_root_token"
+            if parent_mandate.claims.principal_id in revoked_principal_ids
+            else "denied"
+        )
+        return {"allowed": False, "reason": denied_reason}
+
+    chain_ok = client.verify_delegation_chain(
+        token=decision.mandate.token,
+        parent_token=token,
+    )
+    return {
+        "allowed": True,
+        "reason": "ok",
+        "principal_id": decision.mandate.claims.principal_id,
+        "delegated_by": decision.mandate.claims.delegated_by,
+        "delegation_depth": decision.mandate.claims.delegation_depth,
+        "chain_hash": decision.mandate.claims.delegation_chain_hash,
+        "chain_verified": chain_ok,
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Worker process for delegation simulation.")
+    parser.add_argument("--token", required=True)
+    parser.add_argument("--secret-key", default="dev-secret")
+    parser.add_argument("--revocation-file", required=True)
+    parser.add_argument("--policy-file", required=True)
+    args = parser.parse_args()
+    payload = run(
+        token=args.token,
+        secret_key=args.secret_key,
+        revocation_file=args.revocation_file,
+        policy_file=args.policy_file,
+    )
+    print(json.dumps(payload, indent=2, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()