test_secure_peer_durable_approval_isolated.py
Rama main · réplica de lectura del repositorio original.
"""Negotiated durable approvals: real signed requests/store, memory transport.
Run under public_chat_share_safe_tests.py; no monolith, listener or network.
"""
from contextlib import closing
import json
from pathlib import Path
import unittest
from unittest import mock
import uuid
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from agentsdock_team_hub.secure_peer import (
PAIRING_ATTEMPT_RETENTION_SECONDS, PAIRING_TTL_SECONDS, SecurePeerClient,
SecurePeerError, SecurePeerStore, build_pairing_request,
)
import tests.test_secure_peer_auto_join as legacy
class DurableApprovalTests(unittest.TestCase):
setUp = legacy.SecurePeerAutoJoinTests.setUp
restart = legacy.SecurePeerAutoJoinTests.restart
begin = legacy.SecurePeerAutoJoinTests.begin
approve = legacy.SecurePeerAutoJoinTests.approve
approved = legacy.SecurePeerAutoJoinTests.approved
activate = legacy.SecurePeerAutoJoinTests.activate
advertisement = True
def exchange(self, host, port, method, path, **kwargs):
if (method, path) == ("GET", "/v1/health"):
from agentsdock_team_hub.security import canonical_json
health = self.store.public_health()
if self.advertisement is None:
health.pop("durable_pairing_approval_v1", None)
else:
health["durable_pairing_approval_v1"] = self.advertisement
return 200, [("Content-Type", "application/json")], canonical_json(health), self.leaf
return legacy.SecurePeerAutoJoinTests.exchange(self, host, port, method, path, **kwargs)
def request(self, pending):
return json.loads(self.client._connection_row(pending["connection_id"])["pairing_request_json"])
def test_host_advertises_and_builder_default_remains_legacy(self):
self.assertIs(self.store.public_health()["durable_pairing_approval_v1"], True)
key = Ed25519PrivateKey.generate()
request = build_pairing_request(key, server_identity="legacy-peer", display_name="Legacy peer",
host_ca_fingerprint=self.store.ca_fingerprint, created_at=self.now, requested_scopes=["teamspace.read"])
self.assertNotIn("durable_pairing_approval", request["capabilities"])
pending = self.store.submit_pairing(request)
self.assertEqual(pending["expires_at"], self.now + PAIRING_TTL_SECONDS)
self.now += PAIRING_TTL_SECONDS + 1
self.assertEqual(self.store.poll_pairing(pending["pairing_id"], pending["poll_token"])["status"], "expired")
def test_only_literal_true_negotiates_durable_requests(self):
for advertised in (None, False, 1, "true", {}, []):
with self.subTest(advertised=advertised):
self.advertisement = advertised
pending = self.begin(complete_on_approval=True)
self.assertNotIn("durable_pairing_approval", self.request(pending)["capabilities"])
self.assertEqual(pending["pairing_expires_at"], self.now + PAIRING_TTL_SECONDS)
self.advertisement = True
pending = self.begin(complete_on_approval=True)
request = self.request(pending)
self.assertIn("durable_pairing_approval", request["capabilities"])
# Host parsing verifies the actual request signature covering opt-in.
self.assertEqual(self.store._normalize_pairing_payload(request)[0], request)
self.assertEqual(pending["pairing_expires_at"], 0)
def test_wait_days_restart_approve_poll_and_activate_without_new_consent(self):
pending = self.begin(complete_on_approval=True)
request = self.request(pending)
self.now += 8 * 24 * 60 * 60
self.store = SecurePeerStore(self.root / "host", "isolated-host-001", "isolated-hub-001", clock=lambda: self.now)
self.client = self.restart()
self.assertEqual(self.client.expire_pending_pairings(), 0)
incoming = self.store.list_pairings(team_id="isolated-team-001", status="pending")
self.assertEqual([row["pairing_id"] for row in incoming], [pending["pairing_id"]])
snapshot = self.client.auto_completion_snapshot(pending["connection_id"])
self.assertEqual((snapshot["state"], snapshot["deadline"]), ("pending", None))
self.assertTrue(snapshot["connection"]["complete_on_approval"])
self.assertIsNone(self.client.list_auto_completion_candidates()[0]["auto_completion_deadline"])
self.assertEqual(self.client.auto_completion_info(pending["connection_id"]), {"state": "pending", "deadline": None})
self.approve(pending)
approved = self.client.poll_pairing(pending["connection_id"])
self.assertEqual(approved["status"], "approved")
self.assertEqual(self.request(approved), request)
active = self.activate(approved)
self.assertTrue(active["active"])
self.assertEqual(self.client.auto_completion_state(active["connection_id"]), "completed")
self.assertEqual(self.activate(approved)["connection_id"], active["connection_id"])
def test_lost_committed_post_retries_exact_signed_request_after_legacy_retention(self):
captured = {}
def lose(response):
captured.update(response)
self.post_hook = None
raise SecurePeerError("transport_failed", "Isolated lost acknowledgment", 502)
self.post_hook = lose
with self.assertRaises(SecurePeerError):
self.begin(complete_on_approval=True)
with closing(self.client._connect()) as db:
attempt = dict(db.execute("SELECT * FROM client_pairing_attempts").fetchone())
self.assertEqual(db.execute("SELECT expires_at FROM client_join_intents").fetchone()[0], 0)
self.now += PAIRING_ATTEMPT_RETENTION_SECONDS + 1
self.client = self.restart()
recovered = self.client.recover_pairing_attempts()
self.assertEqual(recovered["recovered"], [attempt["connection_id"]])
self.assertEqual((recovered["retired"], recovered["remaining"]), (0, 0))
current = self.client.get_connection(attempt["connection_id"])
self.assertEqual(current["pairing_id"], captured["pairing_id"])
self.assertEqual(self.request(current), json.loads(attempt["request_json"]))
self.assertEqual(len(self.store.list_pairings()), 1)
self.assertEqual(current["pairing_expires_at"], 0)
self.assertTrue(current["complete_on_approval"])
def test_cancel_after_days_is_terminal_across_restart_and_host_replay(self):
pending = self.begin(complete_on_approval=True)
request = self.request(pending)
self.now += 3 * 24 * 60 * 60
result = self.client.cancel_pairing(pending["connection_id"], str(uuid.uuid4()))
self.assertEqual(result["status"], "cancelled")
self.client = self.restart()
self.assertEqual(self.client.list_auto_completion_candidates(), [])
self.assertEqual(self.client.auto_completion_state(pending["connection_id"]), "cancelled")
self.assertEqual(self.store.submit_pairing(request)["status"], "cancelled")
self.assertEqual(list(self.client.keys_dir.iterdir()), [])
with self.assertRaises(SecurePeerError):
self.approve(pending)
def test_reject_after_days_stays_terminal(self):
pending = self.begin(complete_on_approval=True)
self.now += 3 * 24 * 60 * 60
self.store.reject_pairing(pending["pairing_id"], "isolated-owner-001", "Explicit rejection",
expected_peer_server_identity=self.client.server_identity,
expected_transcript_hash=pending["transcript_hash"], idempotency_key=str(uuid.uuid4()))
rejected = self.client.poll_pairing(pending["connection_id"])
self.assertEqual(rejected["status"], "rejected")
self.assertFalse(rejected["complete_on_approval"])
self.assertEqual(self.client.list_auto_completion_candidates(), [])
def test_legacy_and_durable_pending_coexist_without_reviving_expired(self):
self.advertisement = None
old = self.begin(complete_on_approval=True)
self.now += PAIRING_TTL_SECONDS + 1
self.assertEqual(self.client.get_connection(old["connection_id"])["status"], "expired")
self.advertisement = True
new = self.begin(complete_on_approval=True)
self.now += 3 * 24 * 60 * 60
self.assertEqual(self.client.get_connection(old["connection_id"])["status"], "expired")
self.assertEqual(self.client.get_connection(new["connection_id"])["status"], "pending")
self.assertEqual([row["pairing_id"] for row in self.store.list_pairings(status="pending")], [new["pairing_id"]])
def test_unsigned_zero_expiry_from_legacy_host_is_rejected_initially_and_on_poll(self):
self.advertisement = None
self.post_hook = lambda response: response.update(expires_at=0)
with self.assertRaises(SecurePeerError):
self.begin(complete_on_approval=True)
self.post_hook = None
pending = self.begin(complete_on_approval=True)
self.poll_hook = lambda response: response.update(expires_at=0)
with self.assertRaises(SecurePeerError) as caught:
self.client.poll_pairing(pending["connection_id"])
self.assertEqual(caught.exception.code, "remote_invalid")
self.assertGreater(self.client.get_connection(pending["connection_id"])["pairing_expires_at"], 0)
def test_durable_flag_does_not_bypass_configured_scope_and_capability_policy(self):
pending = self.begin()
request = self.request(pending)
self.assertTrue(self.client._pairing_request_matches_configured_policy(request))
restricted = SecurePeerClient(self.root / "restricted", "restricted-guest", "Restricted",
clock=lambda: self.now, pairing_capabilities=("cert_renewal", "teamspace"))
self.assertFalse(restricted._pairing_request_matches_configured_policy(request))
narrower = {**request, "capabilities": ["cert_renewal", "durable_pairing_approval", "teamspace"]}
self.assertTrue(restricted._pairing_request_matches_configured_policy(narrower))
self.assertFalse(restricted._pairing_request_matches_configured_policy({
**narrower, "requested_scopes": ["teamspace.read", "cross_chat.request_reply"]}))
self.assertFalse(restricted._pairing_request_matches_configured_policy({
**narrower, "capabilities": narrower["capabilities"] + ["unknown"]}))
def test_new_join_and_host_role_pause_cancel_original_durable_consent(self):
old = self.begin(complete_on_approval=True)
current = self.begin(complete_on_approval=True)
self.assertEqual(self.client.auto_completion_state(old["connection_id"]), "cancelled")
self.approve(old)
old_approved = self.client.poll_pairing(old["connection_id"])
with self.assertRaises(SecurePeerError):
self.activate(old_approved)
self.client.cancel_auto_completions()
self.approve(current)
approved = self.client.poll_pairing(current["connection_id"])
with self.assertRaises(SecurePeerError):
self.activate(approved)
self.assertEqual(self.client.list_auto_completion_candidates(), [])
# Re-run the existing exact safety races under negotiated durable consent,
# in addition to the unchanged legacy variants in the original module.
test_cancel_during_activation_health_is_terminal_and_retryable = legacy.SecurePeerAutoJoinTests.test_cancel_during_activation_health_is_terminal_and_retryable
test_cancel_fences_consent_before_waiting_for_stalled_poll = legacy.SecurePeerAutoJoinTests.test_cancel_fences_consent_before_waiting_for_stalled_poll
test_host_pause_without_active_member_and_new_join_supersede_consent = legacy.SecurePeerAutoJoinTests.test_host_pause_without_active_member_and_new_join_supersede_consent
test_manual_binding_change_and_aba_invalidate_pending_join = legacy.SecurePeerAutoJoinTests.test_manual_binding_change_and_aba_invalidate_pending_join
test_revocation_and_forgetting_keep_terminal_intent_unusable = legacy.SecurePeerAutoJoinTests.test_revocation_and_forgetting_keep_terminal_intent_unusable
test_exact_identity_mismatch_rejects_activation_before_health = legacy.SecurePeerAutoJoinTests.test_exact_identity_mismatch_rejects_activation_before_health
def test_durable_optin_is_signed_and_cannot_be_added_to_legacy_request(self):
request = build_pairing_request(Ed25519PrivateKey.generate(), server_identity="signed-peer",
display_name="Signed peer", host_ca_fingerprint=self.store.ca_fingerprint,
created_at=self.now, requested_scopes=["teamspace.read"])
request["capabilities"] = sorted([*request["capabilities"], "durable_pairing_approval"])
with self.assertRaises(SecurePeerError) as caught:
self.store.submit_pairing(request)
self.assertEqual(caught.exception.code, "signature_invalid")
self.assertEqual(self.store.list_pairings(), [])
def test_negotiation_does_not_relax_fresh_request_timestamp_validation(self):
request = build_pairing_request(Ed25519PrivateKey.generate(), server_identity="late-peer",
display_name="Late peer", host_ca_fingerprint=self.store.ca_fingerprint,
created_at=self.now, capabilities=("cert_renewal", "durable_pairing_approval", "teamspace"),
requested_scopes=["teamspace.read"])
self.now += 8 * 24 * 60 * 60
with self.assertRaises(SecurePeerError) as caught:
self.store.submit_pairing(request)
self.assertEqual(caught.exception.code, "pairing_expired")
self.assertEqual(self.store.list_pairings(), [])
def test_shared_nat_can_submit_more_than_sixteen_pending_without_losing_global_bound(self):
for index in range(17):
request = build_pairing_request(Ed25519PrivateKey.generate(),
server_identity=f"nat-peer-{index}", display_name=f"NAT peer {index}",
host_ca_fingerprint=self.store.ca_fingerprint, created_at=self.now,
capabilities=("cert_renewal", "durable_pairing_approval", "teamspace"),
requested_scopes=["teamspace.read"])
self.store.submit_pairing(request, source_ip="192.0.2.55", source_port=40000 + index)
self.assertEqual(len(self.store.list_pairings(status="pending")), 17)
with mock.patch("agentsdock_team_hub.secure_peer.PAIRING_STATUS_LIMIT", 17):
request = build_pairing_request(Ed25519PrivateKey.generate(), server_identity="one-too-many",
display_name="Capacity", host_ca_fingerprint=self.store.ca_fingerprint,
created_at=self.now, requested_scopes=["teamspace.read"])
with self.assertRaises(SecurePeerError) as caught:
self.store.submit_pairing(request, source_ip="192.0.2.56", source_port=41000)
self.assertEqual(caught.exception.code, "pairing_capacity")
if __name__ == "__main__":
unittest.main()