117 lines
4.3 KiB
Python
117 lines
4.3 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from aios_autonomous_release_candidate.errors import CredentialUnavailable
|
|
from aios_autonomous_release_candidate.evidence import EvidenceChain
|
|
from aios_autonomous_release_candidate.execution_manager import ExecutionManager
|
|
from aios_autonomous_release_candidate.models import CredentialRequest, ResourceBinding
|
|
from aios_autonomous_release_candidate.native_adapters import OpenClawNativeAdapter
|
|
from aios_autonomous_release_candidate.secret_manager import SecretManagerV2
|
|
from aios_autonomous_release_candidate.store import StateStore
|
|
from test_native_adapter_contracts import Provider
|
|
|
|
|
|
NOW = datetime(2026, 7, 26, tzinfo=timezone.utc)
|
|
|
|
|
|
class Transport:
|
|
def __init__(self) -> None:
|
|
self.calls = []
|
|
|
|
def exchange(self, operation, payload):
|
|
self.calls.append((operation, payload))
|
|
base = {
|
|
"task_id": payload["task_id"],
|
|
"credential_id": payload.get("credential_id", "credential-openclaw-1"),
|
|
"plaintext_secret_returned": False,
|
|
}
|
|
if operation == "issue":
|
|
return {
|
|
**base,
|
|
"credential_reference": "systemd-credential:openclaw-gateway-candidate",
|
|
"state": "ISSUED",
|
|
"expires_at": payload["expires_at"],
|
|
}
|
|
if operation == "audit":
|
|
return {**base, "events": [{"operation": "issue", "secrets_included": False}]}
|
|
if operation in {"project", "status"}:
|
|
return {**base, "state": "PROJECTED"}
|
|
if operation == "verify_revocation":
|
|
return {**base, "state": "REVOKED", "post_revocation_auth": "REJECTED"}
|
|
return {**base, "state": "REVOKED"}
|
|
|
|
|
|
def credential_request() -> CredentialRequest:
|
|
return CredentialRequest(
|
|
request_id="request-1",
|
|
task_id="task-execution",
|
|
session_id="session-1",
|
|
worker_id="worker-1",
|
|
audience="openclaw-candidate",
|
|
consumer_service="aios-autonomous-release-worker-candidate.service",
|
|
binding=ResourceBinding("aios/aios", "patch/execution", "a" * 40, ("services/test",)),
|
|
operations=("native_session.spawn",),
|
|
provider="openclaw",
|
|
runtime_user="aiman",
|
|
approval_plan_id="owner-stage-b-native-sessions-20260726",
|
|
not_before=NOW,
|
|
expires_at=NOW + timedelta(minutes=5),
|
|
nonce="nonce-execution-1",
|
|
ttl_seconds=300,
|
|
max_uses=1,
|
|
)
|
|
|
|
|
|
def test_secret_manager_uses_reference_only_protocol() -> None:
|
|
transport = Transport()
|
|
manager = SecretManagerV2(transport)
|
|
receipt = manager.issue(credential_request(), NOW)
|
|
assert receipt.reference == "systemd-credential:openclaw-gateway-candidate"
|
|
assert receipt.plaintext_secret_returned is False
|
|
manager.project(receipt)
|
|
manager.status(receipt)
|
|
manager.revoke(receipt)
|
|
manager.verify_revocation(receipt)
|
|
events = manager.audit(receipt)
|
|
assert events[0]["secrets_included"] is False
|
|
assert [operation for operation, _ in transport.calls] == [
|
|
"issue",
|
|
"project",
|
|
"status",
|
|
"revoke",
|
|
"verify_revocation",
|
|
"audit",
|
|
]
|
|
|
|
|
|
def test_secret_manager_rejects_plaintext_fields() -> None:
|
|
transport = Transport()
|
|
|
|
def unsafe(operation, payload):
|
|
return {"task_id": payload["task_id"], "credential_id": "credential-1", "api_key": "redacted"}
|
|
|
|
transport.exchange = unsafe
|
|
with pytest.raises(CredentialUnavailable, match="SECRET_MANAGER_PLAINTEXT_REJECTED"):
|
|
SecretManagerV2(transport).issue(credential_request(), NOW)
|
|
|
|
|
|
def test_execution_manager_records_simulation_without_native_claim(tmp_path: Path) -> None:
|
|
evidence = EvidenceChain(StateStore(tmp_path / "state.sqlite3"), "task-execution")
|
|
adapter = OpenClawNativeAdapter(
|
|
Provider("openclaw"), "systemd-credential:openclaw-gateway-candidate"
|
|
)
|
|
result = ExecutionManager(evidence).run_native_canary(
|
|
adapter,
|
|
task_id="task-execution",
|
|
parent_session_id="parent-1",
|
|
recipient_binding={"recipient_id": "weixin-owner"},
|
|
text="candidate canary",
|
|
)
|
|
assert result.broker_session_verified is True
|
|
assert result.openclaw_native_child_session_verified is False
|
|
assert result.simulation is True
|
|
assert result.session_closed is True
|
|
assert evidence.verify()
|