diff --git a/docs/network-foundation.md b/docs/network-foundation.md index 6a95cb71..5f20b5c3 100644 --- a/docs/network-foundation.md +++ b/docs/network-foundation.md @@ -33,6 +33,10 @@ versions are separate namespaces. This protocol prefers version `2`, with a version `1` fallback for diagnostic peers without node-profile support. Connecting to a project does not adopt that project's release channel or publisher authority. +For an explicit bounded list of alternative addresses for the same trusted peer, +see [the candidate-connection API](peer-candidates.md). It adds no discovery or +relay service and never retries negotiation failures or application operations. + ## Configure identities and trust explicitly Each side needs an operator-provided PEM certificate, corresponding unencrypted diff --git a/docs/peer-candidates.md b/docs/peer-candidates.md new file mode 100644 index 00000000..3d7be4ca --- /dev/null +++ b/docs/peer-candidates.md @@ -0,0 +1,85 @@ +# Explicit trusted peer candidates + +`connect_peer_candidates` is a bounded connection convenience for **one already +trusted peer** at alternative operator-supplied endpoints. It is a first local +NET-1 contract slice, not discovery, NAT traversal, a relay, public deployment or +proof of two-machine reachability. Existing `PeerClient.connect` remains available +for a single explicit endpoint. + +```python +from aethermesh_core.network import connect_peer_candidates + +connection = await connect_peer_candidates( + existing_identity, + (primary_endpoint, alternate_endpoint), + project_id="selected-project", + timeout=10.0, +) +async with connection: + status = await connection.client.request("status", {}, timeout=2.0) +``` + +The caller provides an existing `TLSIdentity`, a tuple of one to eight distinct +`PeerEndpoint` values, one project, and a finite total timeout in `(0, 3600]` seconds. +Every candidate must use the same exact certificate fingerprint and expected node +ID (or all omit that ID). Host, port and expected SAN name may differ only as +explicitly configured. Each endpoint still undergoes TLS 1.3 mutual authentication, +CA/SAN/pin checks, optional node-ID checking and project/protocol negotiation. +The helper never learns trust or endpoints from a peer, persists configuration, +creates keys, discovers addresses or starts a listener. + +Candidates are tried once in caller order, using a new client for each attempt. +One monotonic deadline covers the entire invocation; it is never reset between +attempts. Existing `Limits.handshake_timeout` may bound an individual attempt more +narrowly. Async task cancellation propagates, closes the owned attempt and does +not advance to another candidate. Cleanup uses the existing bounded socket-close +mechanism and may finish after cancellation/deadline delivery; synchronous local +certificate loading is not preemptible. This is not a hard wall-clock guarantee. + +Only two typed failures permit another attempt: + +- `EndpointUnavailable`, a `ConnectionClosed` subtype: transport establishment + failed before a TLS transport was handed to the protocol negotiator. +- `EndpointTimeout`, a `RequestTimeout` subtype: establishment timed out before + protocol negotiation began. + +Once transport establishment succeeds, **every** negotiation failure is terminal, +including EOF or timeout. This distinction matters because a server can reject a +project by closing without an explicit protocol error. Authentication failures, +wrong pins/SANs/node IDs, malformed/incompatible protocol, configuration errors and +unexpected exceptions never trigger fallback. Cleanup failures stop attempts; when +cleaning up a terminal failure, its original type (including cancellation) is +preserved and an ordinary cleanup exception is attached as its cause. A new +cancellation during cleanup remains terminal. Existing handlers for the base +`ConnectionClosed` and `RequestTimeout` classes continue to work. Exhausting all +candidates raises static `ConnectionClosed`; exhausting the total deadline raises +`RequestTimeout`. Only expiry of the helper’s own timer is translated to this +shared-deadline diagnostic; unrelated local/cleanup timeouts remain terminal +without being relabeled. No peer-supplied error strings or TLS paths enter new diagnostics. + +`CandidateConnection` returns the selected endpoint, attempt count and caller-owned +client. Close it explicitly or use its async context manager. Successful connection +is an observation, not a guarantee the peer remains live. No requests are issued +beyond normal connection negotiation and any explicitly supplied profile announcement. +There is no operation callback, application retry, automatic reconnect or operation +replay. After handoff, request failures and uncertain outcomes remain the caller's +responsibility under the existing SDK contracts. + +Tests use mocks for deterministic deadline boundaries and disposable TLS loopback +fixtures for refusal/fallback, identity/project rejection, transport versus +negotiation timeout, cancellation and an explicitly requested echo. Installed-wheel +acceptance verifies an unavailable address followed by an authenticated connection +and a caller-issued status request. These are not independent-machine or Internet +acceptance. Full NET-1 still needs separately reviewed discovery/relay interfaces, +operator-authorized infrastructure and two-machine evidence. + +This connection utility does not grant publisher, release, serving, retention, +repair-execution or deletion authority. Retention's trusted nondecreasing-time or +durable observation-high-water gate, fresh ownership/byte checks and separate +authority remain mandatory before future serving or execution. GitHub remains +operational; project publishers, channels and transport identities remain separate. + +Endpoints are frozen caller-owned configuration values. Normal field assignment +is rejected, and replacement values are validated before connection. This API is +not a sandbox against code in the same process deliberately bypassing Python +immutability or changing trusted identity files. diff --git a/scripts/peer_install_smoke.py b/scripts/peer_install_smoke.py index 88c569a1..85dbe65f 100644 --- a/scripts/peer_install_smoke.py +++ b/scripts/peer_install_smoke.py @@ -320,6 +320,28 @@ async def download_api_smoke(peer: str, pki: Any, pins: list[str], cwd: Path) -> assert server.stdout is not None ready = json.loads(await asyncio.wait_for(server.stdout.readline(), 10)) endpoint = PeerEndpoint("127.0.0.1", ready["port"], "localhost", pins[0]) + from aethermesh_core.network import Limits, connect_peer_candidates + + with socket.socket() as unavailable: + unavailable.bind(("127.0.0.1", 0)) + candidate = PeerEndpoint( + "127.0.0.1", unavailable.getsockname()[1], "localhost", pins[0] + ) + connected = await connect_peer_candidates( + pki.client.identity(), + (candidate, endpoint), + project_id="api-download", + timeout=5, + limits=Limits(handshake_timeout=0.5), + ) + async with connected: + assert connected.attempted_count == 2 and connected.endpoint == endpoint + await connected.client.request("status", {}, timeout=2) + assert not connected.client.connected + print( + "Installed explicit peer candidate fallback passed; no discovery or application replay." + ) + staging, output = cwd / "api-stage", cwd / "api-output" # End one installed SDK process after cooperative cancellation, then diff --git a/src/aethermesh_core/network/__init__.py b/src/aethermesh_core/network/__init__.py index 39e8fe24..59d77481 100644 --- a/src/aethermesh_core/network/__init__.py +++ b/src/aethermesh_core/network/__init__.py @@ -6,6 +6,7 @@ from .bundle import BundleInfo, inspect_bundle from .bundle_export import export_bundle +from .candidates import CandidateConnection, connect_peer_candidates from .client import PeerClient from .config import Limits, PeerEndpoint, TLSIdentity, certificate_fingerprint from .content import CONTENT_OPERATION, MAX_CONTENT_BYTES, ContentRef, ContentStore @@ -13,6 +14,8 @@ from .errors import ( AuthenticationError, ConnectionClosed, + EndpointTimeout, + EndpointUnavailable, PeerError, ProtocolError, RemoteError, @@ -44,12 +47,15 @@ "SUPPORTED_PROTOCOL_VERSIONS", "AuthenticationError", "BundleInfo", + "CandidateConnection", "ConnectionClosed", "ContentRef", "ContentStore", "DownloadError", "DownloadProgress", "DownloadResult", + "EndpointTimeout", + "EndpointUnavailable", "EvaluationBudget", "HardwareProfile", "Limits", @@ -67,6 +73,7 @@ "TransferManifest", "TransferStore", "certificate_fingerprint", + "connect_peer_candidates", "download_transfer", "evaluation_task", "export_bundle", diff --git a/src/aethermesh_core/network/candidates.py b/src/aethermesh_core/network/candidates.py new file mode 100644 index 00000000..6a8729ef --- /dev/null +++ b/src/aethermesh_core/network/candidates.py @@ -0,0 +1,122 @@ +"""Explicit bounded alternatives for one trusted peer, never discovery or replay.""" + +from __future__ import annotations + +import asyncio +import math +from dataclasses import dataclass +from typing import Self + +from .client import PeerClient +from .config import Limits, PeerEndpoint, TLSIdentity +from .errors import ( + ConnectionClosed, + EndpointTimeout, + EndpointUnavailable, + RequestTimeout, +) +from .profile import NodeProfile + + +@dataclass(frozen=True) +class CandidateConnection: + """Caller-owned session; close explicitly or use an async context manager.""" + + client: PeerClient + endpoint: PeerEndpoint + attempted_count: int + + async def close(self) -> None: + await self.client.close() + + async def __aenter__(self) -> Self: + return self + + async def __aexit__(self, *_exc: object) -> None: + await self.close() + + +def _validate( + identity: TLSIdentity, + endpoints: tuple[PeerEndpoint, ...], + timeout: float, + limits: Limits | None, +) -> None: + if not isinstance(identity, TLSIdentity) or ( + limits is not None and not isinstance(limits, Limits) + ): + raise TypeError("explicit TLS identity and typed limits required") + if ( + isinstance(timeout, bool) + or not isinstance(timeout, (int, float)) + or not math.isfinite(timeout) + or not 0 < timeout <= 3600 + ): + raise ValueError("candidate timeout must be finite and within (0, 3600]") + if ( + not isinstance(endpoints, tuple) + or not 1 <= len(endpoints) <= 8 + or any(not isinstance(row, PeerEndpoint) for row in endpoints) + or len(set(endpoints)) != len(endpoints) + ): + raise ValueError("select one to eight distinct explicit peer endpoints") + if len({(row.fingerprint, row.expected_node_id) for row in endpoints}) != 1: + raise ValueError("all candidates must pin the same peer identity") + + +async def connect_peer_candidates( + identity: TLSIdentity, + endpoints: tuple[PeerEndpoint, ...], + *, + project_id: str, + timeout: float, + limits: Limits | None = None, + node_profile: NodeProfile | None = None, +) -> CandidateConnection: + """Try each trusted endpoint once, in caller order, under one total deadline. + + Only pre-negotiation transport unavailability or timeout permits fallback. + Authentication/protocol/configuration failures and cancellation are terminal. + No requests are sent beyond normal peer negotiation/profile announcement. + This does not discover, trust, reconnect, relay or replay application work. + The deadline bounds attempts; cooperative cancellation awaits existing bounded + socket cleanup, so it is not a hard wall-clock limit on synchronous local I/O. + """ + _validate(identity, endpoints, timeout, limits) + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + budget = asyncio.timeout_at(deadline) + try: + async with budget: + for count, endpoint in enumerate(endpoints, 1): + if loop.time() >= deadline: + raise RequestTimeout("Candidate connection deadline expired.") + client = PeerClient( + identity, + project_id=project_id, + limits=limits, + node_profile=node_profile, + ) + try: + await client.connect(endpoint) + if loop.time() >= deadline: + raise RequestTimeout("Candidate connection deadline expired.") + except (EndpointUnavailable, EndpointTimeout): + await client.close() + if loop.time() >= deadline: + raise RequestTimeout( + "Candidate connection deadline expired." + ) from None + except BaseException as failure: + try: + await client.close() + except Exception as cleanup: + raise failure from cleanup + raise + else: + return CandidateConnection(client, endpoint, count) + except TimeoutError as exc: + if not budget.expired(): + raise + raise RequestTimeout("Candidate connection deadline expired.") from exc + raise ConnectionClosed("No selected endpoint established the peer session.") diff --git a/src/aethermesh_core/network/client.py b/src/aethermesh_core/network/client.py index 86f4773d..8013a748 100644 --- a/src/aethermesh_core/network/client.py +++ b/src/aethermesh_core/network/client.py @@ -20,6 +20,8 @@ from .errors import ( AuthenticationError, ConnectionClosed, + EndpointTimeout, + EndpointUnavailable, PeerError, ProtocolError, RemoteError, @@ -124,46 +126,57 @@ async def connect(self, endpoint: PeerEndpoint) -> None: await self._close_session() writer: asyncio.StreamWriter | None = None try: - async with asyncio.timeout(self._limits.handshake_timeout): - reader, writer = await asyncio.open_connection( - endpoint.host, - endpoint.port, - ssl=self._identity.client_context(), - server_hostname=endpoint.server_name, - ssl_handshake_timeout=self._limits.handshake_timeout, - ssl_shutdown_timeout=1.0, - ) - if peer_fingerprint(writer) != endpoint.fingerprint: - raise AuthenticationError( - "The peer certificate pin did not match." + try: + async with asyncio.timeout(self._limits.handshake_timeout): + reader, writer = await asyncio.open_connection( + endpoint.host, + endpoint.port, + ssl=self._identity.client_context(), + server_hostname=endpoint.server_name, + ssl_handshake_timeout=self._limits.handshake_timeout, + ssl_shutdown_timeout=1.0, ) - await write_frame( - writer, - { - "type": "hello", - "versions": list(SUPPORTED_PROTOCOL_VERSIONS), - "project": self._project, - }, - ) - welcome = await read_frame(reader) - capabilities = self._welcome(welcome) - info = self._peer(welcome, endpoint) - await self._announce_identity(reader, writer, info.protocol_version) - self._writer = writer - self._capabilities = capabilities - self._peer_info = info - self._next_id = 1 - self._receiver = asyncio.create_task(self._receive(reader, writer)) - writer = None - except ssl.SSLError as exc: - raise AuthenticationError("Peer TLS authentication failed.") from exc - except TimeoutError as exc: - raise RequestTimeout("Peer connection timed out.") from exc - except (OSError, asyncio.IncompleteReadError) as exc: - raise ConnectionClosed("Could not establish the peer session.") from exc - finally: + if peer_fingerprint(writer) != endpoint.fingerprint: + raise AuthenticationError( + "The peer certificate pin did not match." + ) + await write_frame( + writer, + { + "type": "hello", + "versions": list(SUPPORTED_PROTOCOL_VERSIONS), + "project": self._project, + }, + ) + welcome = await read_frame(reader) + capabilities = self._welcome(welcome) + info = self._peer(welcome, endpoint) + await self._announce_identity( + reader, writer, info.protocol_version + ) + self._writer = writer + self._capabilities = capabilities + self._peer_info = info + self._next_id = 1 + self._receiver = asyncio.create_task(self._receive(reader, writer)) + writer = None + except ssl.SSLError as exc: + raise AuthenticationError( + "Peer TLS authentication failed." + ) from exc + except TimeoutError as exc: + failure = EndpointTimeout if writer is None else RequestTimeout + raise failure("Peer connection timed out.") from exc + except (OSError, asyncio.IncompleteReadError) as exc: + closed = EndpointUnavailable if writer is None else ConnectionClosed + raise closed("Could not establish the peer session.") from exc + except BaseException as failure: if writer is not None: - await close_writer(writer) + try: + await close_writer(writer) + except Exception as cleanup: + raise failure from cleanup + raise async def _announce_identity( self, @@ -488,11 +501,13 @@ async def _close_session(self) -> None: self._capabilities = () self._peer_info = None self._fail_pending(ConnectionClosed("The peer session was closed.")) - if receiver is not None: - receiver.cancel() - await asyncio.gather(receiver, return_exceptions=True) - if writer is not None: - await close_writer(writer) + try: + if receiver is not None: + receiver.cancel() + await asyncio.gather(receiver, return_exceptions=True) + finally: + if writer is not None: + await close_writer(writer) async def close(self) -> None: """Cancel pending work, close the socket and await owned resources.""" diff --git a/src/aethermesh_core/network/errors.py b/src/aethermesh_core/network/errors.py index 788da990..00479314 100644 --- a/src/aethermesh_core/network/errors.py +++ b/src/aethermesh_core/network/errors.py @@ -21,6 +21,14 @@ class RequestTimeout(PeerError): """A connection or operation exceeded its configured deadline.""" +class EndpointUnavailable(ConnectionClosed): + """Transport could not be established; no protocol negotiation was started.""" + + +class EndpointTimeout(RequestTimeout): + """Transport establishment timed out before protocol negotiation started.""" + + class RemoteError(PeerError): """An authenticated peer rejected a request with a stable error code.""" diff --git a/tests/test_network_candidates.py b/tests/test_network_candidates.py new file mode 100644 index 00000000..c54513fb --- /dev/null +++ b/tests/test_network_candidates.py @@ -0,0 +1,443 @@ +"""Explicit connection fallback preserves trust and never replays application work.""" + +import asyncio +import socket +import tempfile +import unittest +from dataclasses import replace +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +from aethermesh_core.network import ( + AuthenticationError, + ConnectionClosed, + EndpointTimeout, + EndpointUnavailable, + Limits, + PeerClient, + PeerEndpoint, + PeerService, + ProtocolError, + RequestTimeout, + TLSIdentity, + candidates, + certificate_fingerprint, + connect_peer_candidates, +) +from tests.network_test_support import TestPKI + +IDENTITY = TLSIdentity(Path("certificate"), Path("key"), Path("trust")) +ENDPOINTS = ( + PeerEndpoint("127.0.0.1", 12345, "localhost", "a" * 64), + PeerEndpoint("127.0.0.1", 12346, "localhost", "a" * 64), +) + + +class CandidateContractTests(unittest.IsolatedAsyncioTestCase): + async def call(self, **kwargs): + return await connect_peer_candidates( + kwargs.pop("identity", IDENTITY), + kwargs.pop("endpoints", ENDPOINTS), + **({"project_id": "test", "timeout": 1} | kwargs), + ) + + async def test_validation_precedes_client_creation(self): + invalid = [ + {"identity": None}, + {"limits": object()}, + *( + {"timeout": value} + for value in (True, 0, -1, float("nan"), float("inf"), 3601, "1") + ), + *( + {"endpoints": value} + for value in ( + (), + list(ENDPOINTS), + (None,), + ENDPOINTS * 5, + (ENDPOINTS[0],) * 2, + ) + ), + {"endpoints": (ENDPOINTS[0], replace(ENDPOINTS[1], fingerprint="b" * 64))}, + { + "endpoints": ( + ENDPOINTS[0], + replace(ENDPOINTS[1], expected_node_id="b" * 64), + ) + }, + ] + with patch.object(candidates, "PeerClient") as create: + for kwargs in invalid: + with ( + self.subTest(kwargs=kwargs), + self.assertRaises((ValueError, TypeError)), + ): + await self.call(**kwargs) + create.assert_not_called() + with self.assertRaises(ValueError): + await self.call(project_id="invalid project") + + async def test_transport_fallback_hands_off_only_successful_session(self): + for failure in ( + EndpointUnavailable("closed"), + EndpointTimeout("attempt timeout"), + ): + with self.subTest(failure=type(failure).__name__): + failed = MagicMock( + connect=AsyncMock(side_effect=failure), close=AsyncMock() + ) + good = MagicMock(connect=AsyncMock(), close=AsyncMock()) + with patch.object( + candidates, "PeerClient", side_effect=(failed, good) + ) as create: + connected = await self.call() + self.assertIs(connected.client, good) + self.assertEqual(connected.endpoint, ENDPOINTS[1]) + self.assertEqual(connected.attempted_count, 2) + failed.close.assert_awaited_once() + good.close.assert_not_called() + self.assertEqual(create.call_count, 2) + self.assertTrue( + all( + call.kwargs["project_id"] == "test" + for call in create.call_args_list + ) + ) + async with connected as same: + self.assertIs(same, connected) + good.close.assert_awaited_once() + good.request.assert_not_called() + + async def test_security_protocol_and_configuration_failures_are_terminal(self): + for failure in ( + AuthenticationError("bad pin"), + ProtocolError("wrong project"), + ConnectionClosed("closed during negotiation"), + RequestTimeout("negotiation timeout"), + ValueError("config"), + RuntimeError("unexpected"), + ): + with self.subTest(failure=type(failure).__name__): + client = MagicMock( + connect=AsyncMock(side_effect=failure), close=AsyncMock() + ) + with ( + patch.object( + candidates, "PeerClient", return_value=client + ) as create, + self.assertRaises(type(failure)), + ): + await self.call() + create.assert_called_once() + client.close.assert_awaited_once() + + async def test_exhaustion_is_bounded_and_static(self): + clients = [ + MagicMock( + connect=AsyncMock( + side_effect=EndpointUnavailable("private endpoint detail") + ), + close=AsyncMock(), + ) + for _ in ENDPOINTS + ] + with ( + patch.object(candidates, "PeerClient", side_effect=clients) as create, + self.assertRaises(ConnectionClosed) as caught, + ): + await self.call() + self.assertNotIn("private", str(caught.exception)) + self.assertEqual(create.call_count, 2) + for client in clients: + client.close.assert_awaited_once() + + async def test_one_total_deadline_stops_stalled_attempt(self): + started = asyncio.Event() + + async def stall(endpoint): + started.set() + await asyncio.Event().wait() + + client = MagicMock(connect=AsyncMock(side_effect=stall), close=AsyncMock()) + with ( + patch.object(candidates, "PeerClient", return_value=client) as create, + self.assertRaises(RequestTimeout), + ): + await self.call(timeout=0.02) + self.assertTrue(started.is_set()) + create.assert_called_once() + client.close.assert_awaited_once() + + async def test_cancellation_propagates_and_closes_owned_client(self): + started = asyncio.Event() + + async def stall(endpoint): + started.set() + await asyncio.Event().wait() + + client = MagicMock(connect=AsyncMock(side_effect=stall), close=AsyncMock()) + with patch.object(candidates, "PeerClient", return_value=client) as create: + task = asyncio.create_task(self.call()) + await started.wait() + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + create.assert_called_once() + client.close.assert_awaited_once() + + async def test_expired_before_attempt_or_after_connect_never_hands_off(self): + now = asyncio.get_running_loop().time() + for clock, attempts in (((now, now + 2), 0), ((now, now, now + 2, now + 2), 1)): + with self.subTest(attempts=attempts): + fake_loop = MagicMock(time=MagicMock(side_effect=clock)) + client = MagicMock(connect=AsyncMock(), close=AsyncMock()) + with ( + patch.object( + candidates.asyncio, "get_running_loop", return_value=fake_loop + ), + patch.object( + candidates, "PeerClient", return_value=client + ) as create, + self.assertRaises(RequestTimeout), + ): + await self.call() + self.assertEqual(create.call_count, attempts) + self.assertEqual(client.close.await_count, attempts) + + async def test_deadline_is_not_reset_between_attempts(self): + now = asyncio.get_running_loop().time() + fake_loop = MagicMock( + time=MagicMock(side_effect=(now, now, now + 0.5, now + 1.1)) + ) + failed = MagicMock( + connect=AsyncMock(side_effect=EndpointUnavailable("closed")), + close=AsyncMock(), + ) + with ( + patch.object( + candidates.asyncio, "get_running_loop", return_value=fake_loop + ), + patch.object(candidates, "PeerClient", return_value=failed) as create, + self.assertRaises(RequestTimeout), + ): + await self.call() + create.assert_called_once() + failed.close.assert_awaited_once() + + async def test_deadline_exhausted_by_failed_attempt_is_terminal(self): + now = asyncio.get_running_loop().time() + clock = MagicMock(time=MagicMock(side_effect=(now, now, now + 2))) + client = MagicMock( + connect=AsyncMock(side_effect=EndpointTimeout("attempt")), close=AsyncMock() + ) + with ( + patch.object(candidates.asyncio, "get_running_loop", return_value=clock), + patch.object(candidates, "PeerClient", return_value=client) as create, + self.assertRaises(RequestTimeout), + ): + await self.call() + create.assert_called_once() + client.close.assert_awaited_once() + + async def test_cancellation_during_receiver_cleanup_closes_transport(self): + from tests.test_network_client import writer_mock + + client = PeerClient(IDENTITY, project_id="test") + writer = writer_mock() + cancelling = asyncio.Event() + + async def receiver(): + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + cancelling.set() + await asyncio.Event().wait() + + task = asyncio.create_task(receiver()) + await asyncio.sleep(0) + client._writer, client._receiver = writer, task + closing = asyncio.create_task(client.close()) + await asyncio.wait_for(cancelling.wait(), 1) + closing.cancel() + with self.assertRaises(asyncio.CancelledError): + await closing + self.assertTrue(task.done()) + writer.close.assert_called_once() + self.assertFalse(client.connected) + + +class CandidateLoopbackTests(unittest.IsolatedAsyncioTestCase): + @classmethod + def setUpClass(cls): + cls.temp = tempfile.TemporaryDirectory() + cls.pki = TestPKI.create(Path(cls.temp.name)) + + @classmethod + def tearDownClass(cls): + cls.temp.cleanup() + + async def asyncSetUp(self): + self.service = PeerService( + self.pki.server.identity(), + project_id="candidates", + allowed_peers=frozenset( + {certificate_fingerprint(self.pki.client.certificate)} + ), + capabilities=frozenset({"echo"}), + ) + await self.service.start() + self.addAsyncCleanup(self.service.close) + self.endpoint = PeerEndpoint( + *self.service.address, + "localhost", + certificate_fingerprint(self.pki.server.certificate), + ) + + async def test_closed_address_then_real_tls_and_explicit_single_request(self): + reserved = socket.socket() + self.addCleanup(reserved.close) + reserved.bind(("127.0.0.1", 0)) + port = reserved.getsockname()[1] + result = await connect_peer_candidates( + self.pki.client.identity(), + (replace(self.endpoint, port=port), self.endpoint), + project_id="candidates", + timeout=3, + limits=Limits(handshake_timeout=0.5), + ) + async with result: + self.assertEqual(result.attempted_count, 2) + self.assertEqual( + await result.client.request("echo", {"text": "once"}, timeout=1), + {"text": "once"}, + ) + self.assertFalse(result.client.connected) + self.assertIsNone(result.client._receiver) + + async def test_real_identity_and_project_errors_never_try_second_address(self): + for project, pin, error in ( + ("candidates", "0" * 64, AuthenticationError), + ("wrong", self.endpoint.fingerprint, ConnectionClosed), + ): + with self.subTest(project=project): + original = PeerClient.connect + attempts = [] + + async def connect( + client, endpoint, attempts=attempts, original=original + ): + attempts.append(endpoint) + return await original(client, endpoint) + + first = replace(self.endpoint, fingerprint=pin) + with ( + patch.object(PeerClient, "connect", connect), + self.assertRaises(error), + ): + await connect_peer_candidates( + self.pki.client.identity(), + (first, replace(first, host="localhost")), + project_id=project, + timeout=3, + ) + self.assertEqual(attempts, [first]) + + async def stalled_endpoint(self, *, tls): + from aethermesh_core.network.protocol import close_writer, read_frame + + entered = asyncio.Event() + tasks = set() + + async def handler(reader, writer): + task = asyncio.current_task() + tasks.add(task) + try: + if tls: + await read_frame(reader) + entered.set() + await reader.read() + finally: + await close_writer(writer) + tasks.discard(task) + + server = await asyncio.start_server( + handler, + "127.0.0.1", + 0, + ssl=self.pki.server.identity().server_context() if tls else None, + ) + + async def cleanup(): + server.close() + await server.wait_closed() + pending = tuple(tasks) + for task in pending: + task.cancel() + await asyncio.gather(*pending, return_exceptions=True) + + self.addAsyncCleanup(cleanup) + return replace(self.endpoint, port=server.sockets[0].getsockname()[1]), entered + + async def test_real_transport_timeout_can_fall_back_but_negotiation_timeout_cannot( + self, + ): + raw, started = await self.stalled_endpoint(tls=False) + result = await connect_peer_candidates( + self.pki.client.identity(), + (raw, self.endpoint), + project_id="candidates", + timeout=3, + limits=Limits(handshake_timeout=0.5), + ) + self.assertTrue(started.is_set()) + self.assertEqual(result.attempted_count, 2) + await result.close() + negotiated, started = await self.stalled_endpoint(tls=True) + original = PeerClient.connect + attempts = [] + + async def connect(client, endpoint): + attempts.append(endpoint) + return await original(client, endpoint) + + with ( + patch.object(PeerClient, "connect", connect), + self.assertRaises(RequestTimeout) as caught, + ): + await connect_peer_candidates( + self.pki.client.identity(), + (negotiated, self.endpoint), + project_id="candidates", + timeout=3, + limits=Limits(handshake_timeout=0.5), + ) + self.assertNotIsInstance(caught.exception, EndpointTimeout) + self.assertTrue(started.is_set()) + self.assertEqual(attempts, [negotiated]) + + async def test_real_cancellation_stops_without_fallback(self): + endpoint, started = await self.stalled_endpoint(tls=True) + original = PeerClient.connect + clients = [] + + async def connect(client, selected): + clients.append(client) + return await original(client, selected) + + with patch.object(PeerClient, "connect", connect): + task = asyncio.create_task( + connect_peer_candidates( + self.pki.client.identity(), + (endpoint, self.endpoint), + project_id="candidates", + timeout=3, + ) + ) + await asyncio.wait_for(started.wait(), 2) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + self.assertEqual(len(clients), 1) + self.assertFalse(clients[0].connected) + self.assertIsNone(clients[0]._receiver) diff --git a/tests/test_network_candidates_adversarial.py b/tests/test_network_candidates_adversarial.py new file mode 100644 index 00000000..6aaf56cc --- /dev/null +++ b/tests/test_network_candidates_adversarial.py @@ -0,0 +1,238 @@ +"""Adversarial local connector verification; no live-network acceptance claim.""" + +import asyncio +import unittest +from dataclasses import FrozenInstanceError, replace +from unittest.mock import AsyncMock, MagicMock, patch + +from aethermesh_core.network import ( + AuthenticationError, + ConnectionClosed, + EndpointTimeout, + EndpointUnavailable, + ProtocolError, + RequestTimeout, + candidates, +) +from tests.test_network_candidates import ENDPOINTS, IDENTITY + + +class AdversarialCandidateTests(unittest.IsolatedAsyncioTestCase): + async def call(self): + return await candidates.connect_peer_candidates( + IDENTITY, + ENDPOINTS + (replace(ENDPOINTS[1], port=12347),), + project_id="test", + timeout=1, + ) + + async def test_order_preserved_and_terminal_middle_failure_blocks_last_candidate( + self, + ): + for failure in ( + AuthenticationError("identity"), + ProtocolError("protocol"), + ConnectionClosed("negotiation EOF"), + RequestTimeout("negotiation deadline"), + ): + with self.subTest(failure=type(failure).__name__): + clients = [ + MagicMock( + connect=AsyncMock(side_effect=EndpointUnavailable("first")), + close=AsyncMock(), + ), + MagicMock( + connect=AsyncMock(side_effect=failure), close=AsyncMock() + ), + MagicMock(connect=AsyncMock(), close=AsyncMock()), + ] + with ( + patch.object( + candidates, "PeerClient", side_effect=clients + ) as create, + self.assertRaises(type(failure)), + ): + await self.call() + self.assertEqual(create.call_count, 2) + self.assertEqual(clients[0].connect.await_args.args, (ENDPOINTS[0],)) + self.assertEqual(clients[1].connect.await_args.args, (ENDPOINTS[1],)) + for client in clients[:2]: + client.close.assert_awaited_once() + client.request.assert_not_called() + clients[2].connect.assert_not_called() + + async def test_cleanup_failure_is_terminal_even_if_its_type_is_retryable(self): + for cleanup in ( + RuntimeError("cleanup"), + TimeoutError("cleanup timeout before shared deadline"), + EndpointUnavailable("cleanup unavailable"), + EndpointTimeout("cleanup timeout"), + ): + with self.subTest(cleanup=type(cleanup).__name__): + client = MagicMock( + connect=AsyncMock(side_effect=EndpointUnavailable("connect")), + close=AsyncMock(side_effect=cleanup), + ) + with ( + patch.object( + candidates, "PeerClient", return_value=client + ) as create, + self.assertRaises(type(cleanup)), + ): + await self.call() + create.assert_called_once() + client.close.assert_awaited_once() + + async def test_cleanup_consumes_original_deadline(self): + now = asyncio.get_running_loop().time() + clock = [now] + + async def cleanup(): + clock[0] += 2 + + client = MagicMock( + connect=AsyncMock(side_effect=EndpointTimeout("first")), + close=AsyncMock(side_effect=cleanup), + ) + loop = MagicMock(time=lambda: clock[0]) + with ( + patch.object(candidates.asyncio, "get_running_loop", return_value=loop), + patch.object(candidates, "PeerClient", return_value=client) as create, + self.assertRaises(RequestTimeout), + ): + await self.call() + create.assert_called_once() + + async def test_cancellation_during_retry_cleanup_stops_ordering(self): + entered = asyncio.Event() + + async def cleanup(): + entered.set() + await asyncio.Event().wait() + + client = MagicMock( + connect=AsyncMock(side_effect=EndpointUnavailable("first")), + close=AsyncMock(side_effect=cleanup), + ) + with patch.object(candidates, "PeerClient", return_value=client) as create: + task = asyncio.create_task(self.call()) + await asyncio.wait_for(entered.wait(), 1) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + self.assertTrue(task.cancelled()) + create.assert_called_once() + + async def test_cleanup_exception_does_not_replace_primary_cancellation_or_authentication( + self, + ): + for failure in (asyncio.CancelledError(), AuthenticationError("pin mismatch")): + with self.subTest(failure=type(failure).__name__): + cleanup = RuntimeError("cleanup failed") + client = MagicMock( + connect=AsyncMock(side_effect=failure), + close=AsyncMock(side_effect=cleanup), + ) + with ( + patch.object( + candidates, "PeerClient", return_value=client + ) as create, + self.assertRaises(type(failure)) as caught, + ): + await self.call() + self.assertIs(caught.exception, failure) + self.assertIs(caught.exception.__cause__, cleanup) + create.assert_called_once() + + async def test_second_cancellation_during_terminal_cleanup_never_falls_back(self): + connecting, cleaning = asyncio.Event(), asyncio.Event() + + async def connect(endpoint): + connecting.set() + await asyncio.Event().wait() + + async def cleanup(): + cleaning.set() + await asyncio.Event().wait() + + client = MagicMock( + connect=AsyncMock(side_effect=connect), close=AsyncMock(side_effect=cleanup) + ) + with patch.object(candidates, "PeerClient", return_value=client) as create: + task = asyncio.create_task(self.call()) + await asyncio.wait_for(connecting.wait(), 1) + task.cancel() + await asyncio.wait_for(cleaning.wait(), 1) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + create.assert_called_once() + self.assertTrue(task.cancelled()) + + async def test_normal_endpoint_mutation_is_rejected_and_replacement_is_revalidated( + self, + ): + endpoint = replace(ENDPOINTS[0]) + with self.assertRaises(FrozenInstanceError): + endpoint.fingerprint = "b" * 64 + with self.assertRaises(TypeError): + ENDPOINTS[0] = endpoint + with ( + patch.object(candidates, "PeerClient") as create, + self.assertRaises(ValueError), + ): + await candidates.connect_peer_candidates( + IDENTITY, + (endpoint, replace(ENDPOINTS[1], fingerprint="b" * 64)), + project_id="test", + timeout=1, + ) + create.assert_not_called() + + async def test_actual_task_cancellation_and_total_timeout_survive_cleanup_failure( + self, + ): + for external in (True, False): + with self.subTest(external=external): + entered = asyncio.Event() + + async def stalled(endpoint, entered=entered): + entered.set() + await asyncio.Event().wait() + + client = MagicMock( + connect=AsyncMock(side_effect=stalled), + close=AsyncMock(side_effect=RuntimeError("cleanup")), + ) + with patch.object( + candidates, "PeerClient", return_value=client + ) as create: + task = asyncio.create_task( + candidates.connect_peer_candidates( + IDENTITY, + ENDPOINTS, + project_id="test", + timeout=1 if external else 0.02, + ) + ) + await asyncio.wait_for(entered.wait(), 1) + if external: + task.cancel() + with self.assertRaises( + asyncio.CancelledError if external else RequestTimeout + ): + await task + self.assertEqual(task.cancelled(), external) + create.assert_called_once() + + async def test_terminal_nonbudget_timeout_is_not_relabeled(self): + failure = TimeoutError("unexpected local timeout") + client = MagicMock(connect=AsyncMock(side_effect=failure), close=AsyncMock()) + with ( + patch.object(candidates, "PeerClient", return_value=client) as create, + self.assertRaises(TimeoutError) as caught, + ): + await self.call() + self.assertIs(caught.exception, failure) + create.assert_called_once() + client.close.assert_awaited_once() diff --git a/tests/test_network_client.py b/tests/test_network_client.py index c4b38dbe..7e5cd27c 100644 --- a/tests/test_network_client.py +++ b/tests/test_network_client.py @@ -126,6 +126,64 @@ async def asyncTearDown(self): class ClientLifecycleTests(ClientStreamTestCase): + async def test_connect_cleanup_preserves_negotiation_failure(self): + for failure in ( + AuthenticationError("identity"), + ProtocolError("protocol"), + asyncio.CancelledError("cancelled"), + ): + with self.subTest(failure=type(failure).__name__): + cleanup = ValueError("inner writer cleanup") + with ( + patch( + "aethermesh_core.network.client.read_frame", + AsyncMock(side_effect=failure), + ), + patch( + "aethermesh_core.network.client.close_writer", + AsyncMock(side_effect=cleanup), + ) as close, + self.assertRaises(type(failure)) as caught, + ): + await self.client.connect(ENDPOINT) + self.assertIs(caught.exception, failure) + self.assertIs(caught.exception.__cause__, cleanup) + close.assert_awaited_once_with(self.writer) + self.assertFalse(self.client.connected) + self.assertFalse(self.client._lifecycle_lock.locked()) + + async def test_connect_task_cancellation_survives_inner_cleanup_failure(self): + entered = asyncio.Event() + failures = [] + + async def blocked_read(_reader): + entered.set() + try: + await asyncio.Future() + except asyncio.CancelledError as failure: + failures.append(failure) + raise + + cleanup = RuntimeError("inner cleanup") + with ( + patch("aethermesh_core.network.client.read_frame", blocked_read), + patch( + "aethermesh_core.network.client.close_writer", + AsyncMock(side_effect=cleanup), + ) as close, + ): + task = asyncio.create_task(self.client.connect(ENDPOINT)) + await entered.wait() + task.cancel() + with self.assertRaises(asyncio.CancelledError) as caught: + await task + self.assertIs(caught.exception, failures[0]) + self.assertIs(caught.exception.__cause__, cleanup) + self.assertTrue(task.cancelled()) + close.assert_awaited_once_with(self.writer) + self.assertFalse(self.client.connected) + self.assertFalse(self.client._lifecycle_lock.locked()) + async def test_connect_context_status_reply_and_close(self): self.assertFalse(self.client.connected) self.assertEqual(self.client.capabilities, ())