diff --git a/PRIORITY.md b/PRIORITY.md index 6b18bc8..68522d2 100644 --- a/PRIORITY.md +++ b/PRIORITY.md @@ -1,4 +1,4 @@ -# Numeric transport priority — 0.7.2-SNAPSHOT +# Numeric transport priority — protocol 2 ```kotlin import net.sergeych.kiloparsec.call @@ -8,13 +8,15 @@ remote.call(command, arguments, priority = 100) remote.push(notification, arguments, priority = 0) ``` -Larger signed `Int` values go first. Equal priorities preserve FIFO. Existing +Larger signed `Int` values go first within the allowed nonce range. Equal +priorities use nonce order for encrypted packets and FIFO for unnumbered packets. Existing call/push entry points remain unchanged and use zero unless invoked inside an explicit priority call context. The overloads are extension functions, preserving existing RemoteInterface implementers. A coroutine context carries the selected value through the encrypted RPC wrappers; it becomes serialized outer-block data. -The adapter owns one outgoing writer. Waiting for a response does not occupy the +The transport owns a nonce-aware outgoing writer and submits one block at a time +to the adapter, preserving the selected order through the carrier. Waiting for a response does not occupy the writer: calls retain independent request IDs and response waiters. Both transport responses and errors copy the request's priority, including encrypted RPC errors returned inside an outer response. Pushes have no normal response. @@ -30,10 +32,8 @@ existing length prefix, WebSocket messages and UDP data blocks provide that boundary. The old decoder consumes the legacy body and ignores trailing bytes. Do not use this EOF-terminated extension to concatenate unframed Block objects or embed them directly in a positional collection; wrap each block in its own bytes. -The handshake remains version 1 because the original wire format is retained. - -Old peers interoperate but do not honor/inherit priority. New peers can decode -both formats. BiPack @Extendable is deliberately not added: it would insert a new +The priority trailer itself retains its legacy encoding. The secure handshake +requires protocol 2; this does not enable communication with protocol-1 peers. BiPack @Extendable is deliberately not added: it would insert a new field-count prefix before existing fields and break old readers. ## Adapter behavior @@ -45,7 +45,8 @@ it does not deserialize/copy/decrypt the payload or guess from the last four byt or legacy devices. Built-in ProxyDevice-based adapters override it to enter their writer directly. The old output channel remains available, bridged into that writer; new transports should call send so pending senders are visible to priority ordering. -Custom adapters retain their existing FIFO behavior until they implement priority. +Transport callers now receive nonce-aware scheduling even with a custom FIFO +adapter. Direct adapter users retain the adapter-level scheduling behavior. Queue admission is bounded to 128 waiting blocks and 32 MiB (plus the block currently being written). Overflow fails explicitly rather than allocating an @@ -56,7 +57,43 @@ batch-collection delay. WebSocket writes flush after each block to avoid excessi look-ahead in the engine's FIFO. Socket/TLS buffers and network loss can still delay control behind bytes already submitted; priority is not a preemption guarantee. -## Verification +## Nonce-aware admission and fairness + +`OutQueue.push(message, nonce, priority)` records scheduling metadata locally; no +new wire fields are added. `pop()` finds the smallest queued nonce and selects the +highest-priority message no more than 2730 positions ahead of it. This is two +thirds of the 4096-position receive replay window. Gaps count as nonce distance, +not as transmitted packets. The bound prevents an old encrypted message from +being starved until it becomes stale. It leaves headroom for external reordering, +but does not promise acceptance after arbitrary network delay or reordering. + +A transport's preparation mutex covers nonce allocation, encryption, final block +encoding and queue admission. Encryption remains in the secure-layer callbacks. +All encrypted paths (identity/finished, calls, pushes and replies) use this path. +The mutex is released before waiting for transmission. Full queues reject +admission immediately; there is no capacity wait while holding this mutex. +Cancellation may leave a nonce gap but never reuses a nonce. + +The transport writer serializes carrier submissions, so a downstream adapter +cannot reorder these submissions again. Public direct writes to an adapter's +output channel bypass this transport contract. + +Large-message preparation is not preemptible by priority. A local JVM probe with +concurrent 1 MiB and 1 KiB payloads measured preparation including mutex waiting +at medians of about 87 ms for both sizes (maximum about 118 ms). This includes +encoding and allocation, not just cipher time; it is a local observation, not a +throughput or latency guarantee. + +## Current verification + +Focused tests cover replay-window boundaries and bitmap wraparound, concurrent +nonce admission in both directions, duplicate/stale delivery, local send failure +recovery, cancellation/capacity/close, and an encrypted 4098-message backlog whose +old low-priority message remains acceptable. The saturated encrypted-link test +continues to pass. Kotlin JVM tests, JS/Wasm compilation, TypeScript tests and +TypeScript-to-JVM interoperability were checked for this change. + +## Earlier priority-release verification Focused JVM tests cover signed numeric ordering, stable ties, queued cancellation, bounded admission, close, absent/partial trailers, unchanged legacy bodies, and diff --git a/README.md b/README.md index 481fbed..f0c8a6b 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # Kiloparsec The Kotlin implementation and the native [TypeScript + WASM client](typescript/README.md) -are maintained here together. Current source version **0.8.0-SNAPSHOT** uses +are maintained here together. Current source version **0.8.0** uses **protocol 2**, which authenticates RPC correlation metadata inside encrypted packets. Both peers must be upgraded together; protocol 1 peers are rejected with no automatic downgrade. See [the security analysis](typescript/docs/SECURITY.md). diff --git a/RELEASE-0.8.0.md b/RELEASE-0.8.0.md new file mode 100644 index 0000000..d6749c2 --- /dev/null +++ b/RELEASE-0.8.0.md @@ -0,0 +1,15 @@ +# Kiloparsec 0.8.0 + +Stable release of authenticated protocol 2 and the TypeScript/WASM client source. + +- Authenticate RPC kind, correlation and priority; confirm accepted client identity with an encrypted server-finished response. +- Discard duplicate and stale authenticated packets without closing healthy sessions or consuming pending response waiters. +- Bound priority overtaking with the nonce-aware outgoing queue. Serialize encryption and queue admission, retaining progress for older low-priority messages. +- Preserve sessions after local send failures and use a bounded bitmap for replay tracking. +- Honor application initialization rejection on both peers, including subclasses of security and connection-break exceptions. + +Protocol 2 requires coordinated peer upgrades from protocol 1. The fixes preserve the protocol-2 wire encoding. The TypeScript client retains synchronous FIFO WebSocket sending; its private npm package is distributed as source in this release. + +Validation: 64 JVM tests, 47 Kotlin/JS Node tests, 38 TypeScript unit tests and live TypeScript/JVM interoperability passed. Real loopback TCP/UDP tests cover bidirectional calls and UDP duplication/reordering. All 13 Maven publications built; a separate consumer using staged release artifacts passed interoperability without source substitution. Browser, Android and Apple runtimes were not qualified in this release run. + +See [the review](REVIEW-237d404.md) for issue resolutions and validation limits, and [priority semantics](PRIORITY.md) for scheduling guarantees. diff --git a/REVIEW-237d404.md b/REVIEW-237d404.md new file mode 100644 index 0000000..a34de7c --- /dev/null +++ b/REVIEW-237d404.md @@ -0,0 +1,207 @@ +# Review of authenticated protocol 2 + +Reviewed commit: `237d404bcd4d52e015b5e0d7cd4feaec4bd848ff` — Add authenticated protocol 2 and TypeScript WASM client. + +Review date: 2026-10-21. Baseline: parent commit `88ceeb2`. + +This document records findings for separate discussion and repair. Issue numbers are stable. Implementation changes and their validation are recorded under closed issues. Proposed remedies for open issues are discussion options, not an agreed redesign. + +## Assessment + +The reviewed commit fixed real authentication gaps, particularly unprotected RPC response correlation, but introduced availability regressions. The findings below preserve the original analysis. I1–I5 are now fixed for release 0.8.0; I6–I7 were closed as non-defects. No review issues remain open. + +The nonce itself was already present in protocol 1. Receiver-side replay tracking, the fixed window, and counter-exhaustion checks are new. These changes should be evaluated separately from authenticated RPC metadata and the server-finished acknowledgement. + +## Reason for the original change + +The committed assessment says the TypeScript port exposed two protocol-1 weaknesses. Inspection of the parent code supports both mechanisms: + +1. Encrypted results had inner `forId = 0`; their actual request correlation was only in the plaintext outer transport ID. An intermediary could swap two outer IDs and associate authentic answers with the wrong concurrent calls when result codecs were compatible. Protocol 2 authenticates kind, correlation and priority inside encryption. +2. `L0ClientId` returned plaintext `Unit`. An intermediary could forge that completion without the server accepting the identity message. Protocol 2 returns an encrypted server-finished record echoing the exact accepted identity bytes. + +Other changes include separate client/server signature domains, fresh server exchange keys per connection, repeated-handshake rejection, nonce-state preservation during identity completion, and counter-exhaustion checks. These are distinct design decisions and need not be discussed as one indivisible feature. + +## What server-finished means + +It is the response to the existing `L0ClientId` RPC, not an additional network round trip or an acknowledgement of every UDP packet. + +Previously: client sends encrypted identity; server replies with plaintext Unit. Now: client sends encrypted identity; server replies with encrypted `ProtectedMessage(SERVER_FINISHED, 0, 0, identityBytes)`. + +The client decrypts it and checks the kind, zero correlation/priority, and exact identity payload. With an honest authenticated server, this establishes that the session-key holder processed and accepted that identity message before the client announces connection completion. It does not establish application authorization, authenticate an otherwise untrusted server key, or acknowledge subsequent RPC execution. + +The server completes its local established state before sending this response. Losing the response can therefore leave the server established while the client is still waiting. This is an acknowledgement, not an atomic agreement or a delivery-recovery mechanism. + +## Existing nonces versus new replay checks + +Protocol 1 already contained `Package(nonce: ULong, encryptedMessage)`, `private var nonce = 0UL`, and `val n = protectedOp.invoke { nonce++ }`. It derived the cipher nonce from the session token and that counter. Each sending direction has its own encryption key/counter context. + +The encryption nonce selects a unique encryption instance under a key. It is not secret, is not the RPC ID, and does not require packets to arrive in order. Reusing a nonce for distinct plaintexts under the same key is unsafe. Retransmitting an already-produced ciphertext is a different question from encrypting new plaintext with a reused nonce. + +Protocol 1 decrypted using the counter supplied with each packet, without remembering which counters it had already seen. Consequently, a captured valid encrypted call could be replayed and reach application execution again. + +Protocol 2 adds receiver memory: highest authenticated nonce plus the set of seen nonces in a 4,096-position window. This recognizes repeats and bounds memory, at the cost of rejecting sufficiently old unseen packets. Only successfully authenticated packets advance the window; a forged large counter cannot move it forward. + +The tasks are distinct: nonce uniqueness protects encryption; replay tracking prevents repeated acceptance; authenticated RPC correlation binds an answer to its request. Neither correlation nor server-finished intrinsically requires this particular replay-window size or session-closing policy. + +## Repair validation (2026-10-21) + +- 42 focused JVM tests passed, including existing security, priority and transport tests and the new replay/queue/admission regressions. +- TypeScript: 38 passed; the normally skipped interop fixture was run separately and passed against this checkout's JVM server. +- Release validation: all 64 JVM and 47 Kotlin/JS Node tests passed. Real loopback UDP exercised reordered and duplicated packets and concurrent bidirectional RPCs; real TCP exercised concurrent bidirectional RPCs and priority. The new carrier test cancels its collector before closing the listener. +- All 13 Maven publications and Dokka built successfully. A standalone consumer resolved staged 0.8.0 artifacts without source substitution and passed the TypeScript interoperability suite (authentication, duplex, pushes, errors, concurrent calls and reconnect). +- Kotlin Wasm compilation passed. Browser, Android and Apple runtime qualification was not performed. Apple KLIB publication does not qualify final Apple application linking or runtime. +- Protocol-2 wire encoding remains unchanged; older protocol-2 peers still retain their own previous replay-disconnect behavior until upgraded. +- The encrypted backlog test validates all 4098 messages, including an old low-priority message, decrypt successfully after scheduling. Concurrent real JVM preparation exercises both directions and calls, replies, pushes and handshake messages. +- A temporary preparation benchmark (source retained at `/tmp/PreparationBenchmarkTest.kt`) ran concurrent 1 MiB and 1 KiB payloads under the preparation mutex. Measured medians including lock waiting were 86.86 ms and 86.60 ms, with maxima 117.40 ms and 117.65 ms. This is a local JVM observation including serialization and allocation; large-message preparation can delay small messages before queue scheduling. The benchmark source was removed from the checkout. +- Historical probe/test counts below describe the original review, not the final repair run. + +## Original review validation and limits + +- 14 focused JVM tests passed: 7 existing authenticated-envelope tests, 3 security-boundary tests, and 4 temporary review probes. +- The review probes checked replay-window boundaries, adjacent reordering survival, deep reordering session closure, and local send failure session closure. +- Follow-up I5 check: 3 additional focused JVM probes passed, reproducing ignored client/server initialization rejection and confirming rejection through the application `execute` gate. Probe source saved at `/tmp/ReviewI5Test.kt` and removed from the checkout. +- TypeScript tests: 35 passed, 1 interoperability test skipped. +- No live UDP impairment, browser qualification, full cryptographic audit, or sustained-load benchmark was performed. +- Temporary probe source was saved at `/tmp/Review237d404Test.kt` and removed from the checkout. This path is ephemeral. +- Existing `.idea/misc.xml` changes and untracked `kotlin-js-store/` were left untouched. + +All review issues are closed: I1–I5 are included in release 0.8.0; I6–I7 require no remediation. + +## Closed issues + +### Issue 1 — Duplicate or stale packets terminate the session + +**Resolution (2026-10-21):** Done — duplicate/stale encrypted packets are silently discarded on Kotlin and TypeScript. Incoming calls/pushes are not executed again; response authentication/replay checks run before consuming the pending waiter. A valid later reply can still complete it. Authentication/routing failures remain fatal. No automatic retry was added. + +**Original finding (before repair):** + +**Priority:** High for UDP. **Status:** Confirmed regression; reproduced. + +`KiloParams.decrypt` rejects a previously received nonce, or an unseen nonce at least 4,096 positions behind the highest authenticated nonce. `KiloL0Interface.executeTransportBlock` turns rejection into `SecurityException`, a subclass of `BreakConnectionException`, and the transport closes. + +Observed behavior: + +| Delivery | Result | +| --- | --- | +| Adjacent packets reversed | Both accepted | +| Unseen nonce 4,095 positions behind highest | Accepted | +| Unseen nonce 4,096 positions behind highest | Rejected | +| Duplicate incoming call or push | Rejected; session closes | + +Duplicate responses are different: the outer transport normally ignores responses whose pending request was already removed. A stale response for a still-pending call reaches authentication and can close the session. + +A controlled test held an encrypted push, delivered 4,096 newer pushes, then delivered the original. The valid original packet caused session closure. An adjacent-reordering test survived. The existing duplicate-call regression also explicitly expects connection termination. + +**Impact:** ordinary datagram duplication or sufficiently deep reordering becomes a session-wide failure, affecting unrelated calls. This is an availability issue, not evidence of broken encryption. + +**Discussion direction:** distinguish duplicate/stale packet disposition from fatal errors; discard without repeating application execution. If reliable delivery is desired, separately specify timeouts, retransmission and response caching. Discarding duplicates alone does not implement reliable RPC. + +References: [KiloParams.kt](src/commonMain/kotlin/net/sergeych/kiloparsec/KiloParams.kt), [KiloL0Interface.kt](src/commonMain/kotlin/net/sergeych/kiloparsec/KiloL0Interface.kt), [Transport.kt](src/commonMain/kotlin/net/sergeych/kiloparsec/Transport.kt), [AuthenticatedEnvelopeTest.kt](src/commonTest/kotlin/AuthenticatedEnvelopeTest.kt). + +### Issue 2 — Priority scheduling can exceed the replay window locally + +**Resolution (2026-10-21):** Done — `OutQueue` enforces the 2730-position allowance; one preparation mutex covers nonce allocation, sealing, encoding and admission. The transport owns the nonce-aware writer and submits serially to the existing adapter, preserving its selected order. Encryption stays in secure-layer callbacks. Full queues reject immediately rather than wait under the mutex. + +**Original finding (before repair):** + +**Priority:** High under sustained priority traffic. **Status:** Code-established interaction; actual scheduler saturation not reproduced. + +Encryption allocates a nonce before `PriorityBlockWriter` selects transmission order. A low-priority encrypted packet can remain queued while newer high-priority packets overtake it. There is no fairness or maximum-overtaking guarantee. The 128-item queue bound limits simultaneous occupancy, not the total number of packets that can overtake one waiting packet. + +If overtaking spans 4,096 nonce positions, eventual delivery triggers issue 1. This can occur even with ordered TCP/WebSocket delivery: the reordering happens before the carrier sends the bytes. Calls, replies and pushes share one nonce sequence per sending direction. + +**Agreed design (2026-10-21):** introduce a specific `OutQueue` with `push(message, nonce, priority)` and `pop()`. Preserve encryption in its existing layer. Pass nonce metadata explicitly to the queue rather than decoding encrypted-package structures in the scheduler. + +For receiver window `W = 4096`, use a local overtaking allowance `B = floor(2 * W / 3) = 2730`: + +```text +pop(): + oldest = minimum queued nonce + eligible = messages with nonce - oldest <= B + select highest priority among eligible + break priority ties by smallest nonce +``` + +Use subtraction after establishing the minimum to avoid addition overflow. The allowance is based on nonce distance, not the number of successful sends. This accommodates gaps from failed or cancelled messages and prevents starvation by indefinitely newer high-priority traffic. + +One mutex per sending direction covers nonce allocation, encryption/encoding, and queue insertion as one ordered preparation/admission operation. Queue insertion and selection must themselves be synchronized. This prevents an older allocated nonce from remaining outside the queue while newer packets are transmitted. All encrypted message paths sharing that sending key must participate, including calls, replies, pushes, and handshake ciphertexts. + +Wait for transmission completion outside the mutex. Queue-capacity waiting must not prevent the writer from draining the queue; coordinate capacity reservation/admission accordingly. Failed preparation may consume a nonce, which must never be reused. Cancellation and close must release reservations and completion waiters. + +The queue and writer must preserve the selected transmission order. This bounds scheduler-induced reordering while leaving roughly one third of the receiver window for subsequent network/receive-processing reordering. It does not guarantee acceptance after arbitrarily deep external reordering; issue 1 remains a separate fix. + +**Validation targets:** priority selection inside the permitted range; oldest-message progress under sustained higher-priority traffic; exact window boundary; nonce gaps; concurrent preparation/admission; cancellation, capacity and close behavior; and an integration test showing a delayed low-priority encrypted message arrives without becoming stale. Measure preparation-lock contention with large messages rather than assuming encryption cost is negligible. + +**Implementation status:** implemented and verified. A 4098-message encrypted backlog, concurrent duplex admission, queue capacity/cancellation/close and existing saturated-link responsiveness tests pass. Network reordering beyond the receiver window remains discardable, without closing the session. + +References: [KiloParams.kt](src/commonMain/kotlin/net/sergeych/kiloparsec/KiloParams.kt), [PriorityBlockWriter.kt](src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/PriorityBlockWriter.kt). + +### Issue 3 — Local send failures are misclassified as authentication failures + +**Resolution (2026-10-21):** Done — plaintext remote errors are explicitly distinguished at the receive boundary. Local preparation/admission/write failures retain their original errors and do not abort the healthy session. A local reply-send failure is logged without fabricating a plaintext remote error; the caller retains its timeout/cancellation policy. + +**Original finding (before repair):** + +**Priority:** Medium. **Status:** Confirmed regression; reproduced. + +`KiloRemoteInterface.call` catches almost every exception from `transport.callWithId`, closes the transport, and throws `SecurityException("unauthenticated outer error")`. That operation includes local encryption and sending, not only receipt of a plaintext remote error. + +The priority writer throws `RemoteInterface.Exception("Outgoing block queue is full")` on admission failure. Injecting that same local send exception reproduced an authentication error and termination of both peers. + +**Impact:** local congestion closes healthy sessions and unrelated pending calls; diagnostics incorrectly suggest a remote security failure. + +**Discussion direction:** explicitly distinguish plaintext remote errors at the receive boundary from local encoding, admission and send failures. + +References: [KiloRemoteInterface.kt](src/commonMain/kotlin/net/sergeych/kiloparsec/KiloRemoteInterface.kt), [PriorityBlockWriter.kt](src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/PriorityBlockWriter.kt). + +### Issue 4 — Replay tracking adds a full-window scan per advancing packet + +**Resolution (2026-10-21):** Done — Kotlin and TypeScript use 4096-bit circular replay windows. Normal in-order advancement clears one bit, large jumps clear a fixed bitmap, and only successfully authenticated packets update history. + +**Original finding (before repair):** + +**Priority:** Medium performance concern. **Status:** Confirmed algorithmic cost; throughput impact not benchmarked. + +Kotlin uses `received.removeAll` when the high watermark advances; TypeScript scans its received set. In steady in-order traffic with a populated window, this is approximately 4,096 age checks per packet. Kotlin performs this under the lock also used for nonce allocation and decryption. + +**Impact:** bounded memory, but avoidable CPU work and contention for a high-throughput transport. + +**Discussion direction:** if retaining windowed replay protection, use a circular bitmap or another bounded representation with inexpensive advancement. + +References: [KiloParams.kt](src/commonMain/kotlin/net/sergeych/kiloparsec/KiloParams.kt), [client.ts](typescript/src/client.ts). Algorithm background: [RFC 6479](https://www.rfc-editor.org/rfc/rfc6479.html). + +### Issue 5 — Connection-init policy rejection is swallowed + +**Status:** Done on 2026-10-21 — initialization rejection now propagates on both ends. + +**Design decision:** preserve Kotlin's code-based remote-party authorization through the local interface and initialization handlers. Both peers receive verified identity and session data through `KiloScope`. No trusted-key-list constructor parameter is required; the earlier proposal to add one is withdrawn. + +**Original defect:** the shared `invokeAll` helper discarded every handler exception using `runCatching`, including deliberate security/connection rejection. This defect predates commit `237d404`. Follow-up probes reproduced successful connection and application execution despite rejection on either peer. + +**Fix:** replace `runCatching` with explicit `try/catch`. Rethrow `LocalInterface.BreakConnectionException` and all subclasses, including `RemoteInterface.SecurityException`; continue ignoring other handler failures as before. Rejection stops subsequent handlers. The shared helper also propagates this exception family when invoked for disconnect handlers. + +**Validation:** `ConnectionInitializationTest` covers client and server initialization rejection using a custom `SecurityException` subclass and a custom `BreakConnectionException` subclass. Each case verifies that the policy sees the verified remote identity, both connection runs finish, later init handlers do not run, connected state is not announced, and a subsequent client RPC fails as closed. A fifth test verifies that ordinary handler failures still allow initialization and a successful RPC. The four rejection cases failed before the fix and passed after it. All 15 focused JVM tests passed, including `AuthenticatedEnvelopeTest` and `SecurityBoundaryTest`. + +**Scope:** this completes the requested exception-propagation repair. Parameter publication order and initialization-time RPC semantics were not changed; these tests do not establish a general concurrent-dispatch barrier before initialization completes. + +References: [KiloClientConnection.kt](src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClientConnection.kt), [ConnectionInitializationTest.kt](src/commonTest/kotlin/ConnectionInitializationTest.kt). + +### Issue 6 — Protocol 2 requires coordinated peer upgrades + +**Status:** Closed — intentional compatibility decision, not a defect; no remediation required. + +Both peers require protocol 2, with no automatic v1 fallback. The new envelope and identity acknowledgement deliberately break wire compatibility with v1. Coordinated peer upgrades were an agreed design decision, not an outstanding issue. + +The original issue number is retained for reference. + +References: [KiloProtocol.kt](src/commonMain/kotlin/net/sergeych/kiloparsec/KiloProtocol.kt), [SECURITY.md](typescript/docs/SECURITY.md). + +### Issue 7 — Replay protection is not business-operation deduplication + +**Status:** Closed on 2026-10-21 — not a protocol defect; excluded from remediation. + +Kiloparsec protects against replayed encrypted packets and forged or altered protocol messages. A new valid RPC issued by application code, on the same connection or after reconnecting, is a new request. The protocol cannot determine whether repeating that operation is intentional or an application mistake. + +Business-operation deduplication requires application-defined semantics, such as a durable operation identifier and an explicit idempotency contract. It belongs to business logic and should not have been classified as a Kiloparsec issue. + +**Resolution:** no protocol change required. The original issue number is retained for reference. diff --git a/build.gradle.kts b/build.gradle.kts index b3a5163..456b7c2 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -19,7 +19,7 @@ plugins { } group = "net.sergeych" -version = "0.8.0-SNAPSHOT" +version = "0.8.0" repositories { google() @@ -35,7 +35,10 @@ kotlin { jvm() js { browser() - nodejs() + nodejs { + // Replay-window tests deliberately process more than 4096 messages. + testTask { useMocha { timeout = "30s" } } + } } linuxX64() linuxArm64() diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClientConnection.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClientConnection.kt index 6b499ef..9f94ebf 100644 --- a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClientConnection.kt +++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClientConnection.kt @@ -87,7 +87,8 @@ class KiloClientConnection( val identity = pack(ClientIdentity(secretIdKey?.verifyingKey, secretIdKey?.sign(clientAuthentication(params.token)))) - val finished = params.decryptMessage(transport.call(L0ClientId, params.encrypt(identity))) + val finished = unpack(transport.callWithId(L0ClientId, + readResponse = params::decrypt) { params.encode(identity) }) if (finished.kind != ProtectedKind.SERVER_FINISHED || finished.correlation != 0uL || finished.priority != 0 || !finished.payload.contentEquals(identity)) throw RemoteInterface.SecurityException("invalid server finished") @@ -145,4 +146,13 @@ class KiloClientConnection( } internal fun Collection>.invokeAll(scope: KiloScope) = - forEach { runCatching { scope.it() } } + forEach { handler -> + try { + scope.handler() + } catch (e: LocalInterface.BreakConnectionException) { + // Includes SecurityException and application-defined rejection subclasses. + throw e + } catch (_: Throwable) { + // Preserve best-effort notification for other handler failures. + } + } diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloL0Interface.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloL0Interface.kt index 17c3f2c..41b947b 100644 --- a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloL0Interface.kt +++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloL0Interface.kt @@ -20,12 +20,13 @@ internal class KiloL0Interface( // Plaintext transport errors must never use application exception constructors. override fun getErrorBuilder(code: String): ((String, UByteArray?) -> Throwable)? = null - override suspend fun executeTransportBlock(scope: Unit, block: Transport.Block.Call): UByteArray { + override suspend fun executeTransportBlock(scope: Unit, block: Transport.Block.Call): PreparedPayload { if (block.name != L0Call.name) return super.executeTransportBlock(scope, block) if (!allowHandshakeCallbacks && !deferredParams.isCompleted) throw RemoteInterface.SecurityException("application call before authentication") val params = deferredParams.await() val frame: ProtectedMessage = try { params.decryptMessage(unpack(block.packedArgs)) } + catch (e: DiscardPacket) { throw e } catch (e: Throwable) { throw RemoteInterface.SecurityException("invalid authenticated call") } val expected = if (block.id == 0u) ProtectedKind.PUSH else ProtectedKind.CALL if (frame.kind != expected || frame.correlation != block.id.toULong() || frame.priority != block.priority) @@ -40,8 +41,12 @@ internal class KiloL0Interface( catch (e: CancellationException) { throw e } catch (e: Throwable) { clientInterface.encodeError(0u, e) } // Push has no reply; its authenticated kind cannot be changed by rewriting id=0. - if (frame.kind == ProtectedKind.PUSH) return pack(Unit) + if (frame.kind == ProtectedKind.PUSH) return PreparedPayload { OutgoingData(pack(Unit)) } val kind = if (result is Transport.Block.Error) ProtectedKind.ERROR else ProtectedKind.RESPONSE - return pack(params.encryptMessage(ProtectedMessage(kind, frame.correlation, frame.priority, pack(result)))) + val payload = pack(result) + return PreparedPayload { + val encoded = params.encodeMessage(ProtectedMessage(kind, frame.correlation, frame.priority, payload)) + OutgoingData(pack(encoded.bytes), encoded.nonce) + } } } diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloParams.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloParams.kt index 862d39a..0ac550c 100644 --- a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloParams.kt +++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloParams.kt @@ -45,8 +45,7 @@ data class KiloParams( ) private var nonce = 0UL - private var highestReceived: ULong? = null - private val received = mutableSetOf() + private var received = ReplayWindow() val scope: KiloScope by lazy { object : KiloScope { @@ -68,7 +67,9 @@ data class KiloParams( /** * Encrypt using send keys and proper nonce */ - fun encrypt(message: UByteArray, fillFactor: Float = 0f): UByteArray { + fun encrypt(message: UByteArray, fillFactor: Float = 0f): UByteArray = encode(message, fillFactor).bytes + + internal fun encode(message: UByteArray, fillFactor: Float = 0f): OutgoingData { val fill = if (fillFactor > 0f) 0..(message.size * fillFactor).roundToInt() else @@ -79,24 +80,20 @@ data class KiloParams( nonce++ } - return pack( + return OutgoingData(pack( Package(n, sessionKey.encryptWithNonce(message, numericNonce.withULong(n), fill)) - ) + ), n) } /** Authenticate and consume one incoming nonce, allowing bounded out-of-order delivery. */ - fun decrypt(encryptedMessage: UByteArray): UByteArray = protectDecryption { - val p: Package = BipackDecoder.decode(encryptedMessage.toDataSource()) - protectedOp.invoke { - val high = highestReceived - if (p.nonce in received || (high != null && high >= 4096uL && p.nonce <= high - 4096uL)) - throw RemoteInterface.SecurityException("replayed or stale encrypted packet") - val clear = sessionKey.decryptWithNonce(p.encryptedMessage, numericNonce.withULong(p.nonce)) - received.add(p.nonce) - if (high == null || p.nonce > high) { - highestReceived = p.nonce - if (p.nonce >= 4096uL) received.removeAll { it <= p.nonce - 4096uL } + fun decrypt(encryptedMessage: UByteArray): UByteArray { + val p = protectDecryption { BipackDecoder.decode(encryptedMessage.toDataSource()) } + return protectedOp.invoke { + if (received.containsOrStale(p.nonce)) throw DiscardPacket() + val clear = protectDecryption { + sessionKey.decryptWithNonce(p.encryptedMessage, numericNonce.withULong(p.nonce)) } + received.record(p.nonce) clear } } @@ -105,11 +102,12 @@ data class KiloParams( internal fun authenticated(identity: VerifyingPublicKey): KiloParams = protectedOp.invoke { copy(remoteIdentity = identity).also { it.nonce = nonce - it.highestReceived = highestReceived - it.received.addAll(received) + it.received = received.copy() } } + internal fun encodeMessage(message: ProtectedMessage): OutgoingData = encode(pack(message)) + internal fun encryptMessage(message: ProtectedMessage): UByteArray = encrypt(pack(message)) internal fun decryptMessage(message: UByteArray): ProtectedMessage = net.sergeych.utools.unpack(decrypt(message)) diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloRemoteInterface.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloRemoteInterface.kt index a8521e5..65fc1ea 100644 --- a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloRemoteInterface.kt +++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloRemoteInterface.kt @@ -4,12 +4,8 @@ package net.sergeych.kiloparsec import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CancellationException import kotlinx.coroutines.currentCoroutineContext -import kotlinx.serialization.builtins.serializer -import net.sergeych.utools.pack import net.sergeych.utools.unpack -private val l0Notification = Command(L0Call.name, L0Call.argsSerializer, Unit.serializer()) - /** Full-duplex remote interface with authenticated request/result correlation. */ class KiloRemoteInterface( private val deferredParams: CompletableDeferred>, @@ -22,21 +18,20 @@ class KiloRemoteInterface( var requestId = 0u val call = cmd.packCall(args) val encoded = try { - transport.callWithId(L0Call) { id -> + transport.callWithId(L0Call, readResponse = params::decrypt) { id -> requestId = id - params.encryptMessage(ProtectedMessage(ProtectedKind.CALL, id.toULong(), priority, call)) + params.encodeMessage(ProtectedMessage(ProtectedKind.CALL, id.toULong(), priority, call)) } } catch (e: CancellationException) { throw e } catch (e: RemoteInterface.ClosedException) { throw e } - catch (e: Throwable) { - // An outer plaintext error is not an authenticated business result. + catch (e: RemoteInterface.SecurityException) { transport.abortSecurity() - throw RemoteInterface.SecurityException("unauthenticated outer error") + throw e } val frame: ProtectedMessage val block: Transport.Block try { - frame = params.decryptMessage(encoded) + frame = unpack(encoded) if (frame.correlation != requestId.toULong() || frame.priority != priority) throw RemoteInterface.SecurityException("authenticated response routing mismatch") block = unpack(frame.payload) @@ -60,7 +55,9 @@ class KiloRemoteInterface( override suspend fun push(cmd: Command, args: A) { val params = deferredParams.await() val priority = currentCoroutineContext()[CallPriority]?.value ?: 0 - params.transport.push(l0Notification, params.encryptMessage( - ProtectedMessage(ProtectedKind.PUSH, 0uL, priority, cmd.packCall(args)))) + val payload = cmd.packCall(args) + (params.transport as Transport<*>).pushPrepared(L0Call.name, priority) { + params.encodeMessage(ProtectedMessage(ProtectedKind.PUSH, 0uL, priority, payload)) + } } } diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloServerConnection.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloServerConnection.kt index 61e7c3b..e592f75 100644 --- a/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloServerConnection.kt +++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/KiloServerConnection.kt @@ -71,10 +71,10 @@ class KiloServerConnection( serverSigningKey?.seal(serverAuthentication(params!!.token))) } } - on(L0ClientId) { encoded -> handshakeLock.withLock { - if (established) throw RemoteInterface.SecurityException("repeated client identity") + onPrepared(L0ClientId) { encoded -> handshakeLock.withLock { var p = params ?: throw RemoteInterface.SecurityException("wrong handshake sequence") val identity = p.decrypt(encoded) + if (established) throw RemoteInterface.SecurityException("repeated client identity") val ci = unpack(identity) if ((ci.clientIdKey == null) != (ci.signature == null)) throw RemoteInterface.SecurityException("incomplete client identity") @@ -89,7 +89,10 @@ class KiloServerConnection( kiloRemoteInterface.complete(KiloRemoteInterface(deferredParams, clientInterface)) connectedScope = p.scope clientInterface.onConnectHandlers.invokeAll(connectedScope) - p.encryptMessage(ProtectedMessage(ProtectedKind.SERVER_FINISHED, 0uL, 0, identity)) + PreparedPayload { + val result = p.encodeMessage(ProtectedMessage(ProtectedKind.SERVER_FINISHED, 0uL, 0, identity)) + OutgoingData(net.sergeych.utools.pack(result.bytes), result.nonce) + } } } } diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/LocalInterface.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/LocalInterface.kt index 11f09ac..3bce97c 100644 --- a/src/commonMain/kotlin/net/sergeych/kiloparsec/LocalInterface.kt +++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/LocalInterface.kt @@ -16,6 +16,7 @@ import net.sergeych.mp_logger.exception import net.sergeych.mp_logger.info import net.sergeych.tools.AtomicCounter import net.sergeych.utools.firstNonNull +import net.sergeych.utools.unpack import kotlin.reflect.KClass private typealias RawCommandHandler = suspend (C, UByteArray) -> UByteArray @@ -82,9 +83,19 @@ open class LocalInterface : Loggable by LogTag("LocalInterface${idCounter.inc .invoke(scope, packedArgs) - /** Transport hook keeps outer routing metadata available to the authenticated layer. */ - internal open suspend fun executeTransportBlock(scope: S, block: Transport.Block.Call): UByteArray = - execute(scope, block.name, block.packedArgs) + private val preparedHandlers = mutableMapOf PreparedPayload>() + + /** Secure handshake replies can defer sealing until atomic outgoing admission. */ + internal fun onPrepared(command: Command, handler: suspend S.(A) -> PreparedPayload) { + preparedHandlers[command.name] = { scope, args -> scope.handler(unpack(command.argsSerializer, args)) } + } + + /** Dispatch now; encode/seal the result later under the outgoing admission mutex. */ + internal open suspend fun executeTransportBlock(scope: S, block: Transport.Block.Call): PreparedPayload { + preparedHandlers[block.name]?.let { return it(scope, block.packedArgs) } + val result = execute(scope, block.name, block.packedArgs) + return PreparedPayload { OutgoingData(result) } + } private val errorByClass = mutableMapOf, String>() private val errorBuilder = mutableMapOf Throwable>() diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/ReplayWindow.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/ReplayWindow.kt new file mode 100644 index 0000000..aee6990 --- /dev/null +++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/ReplayWindow.kt @@ -0,0 +1,47 @@ +package net.sergeych.kiloparsec + +internal const val REPLAY_WINDOW_SIZE = 4096 +internal const val MAX_NONCE_OVERTAKING = REPLAY_WINDOW_SIZE * 2 / 3 + +/** A packet disposition, not a session failure. Never dispatched or replied to. */ +internal class DiscardPacket : RuntimeException("Duplicate or stale encrypted packet") + +/** Fixed circular bitmap. Callers serialize check/authenticate/record as one operation. */ +internal class ReplayWindow { + private val bits = LongArray(REPLAY_WINDOW_SIZE / 64) + private var highest: ULong? = null + fun copy(): ReplayWindow = ReplayWindow().also { + bits.copyInto(it.bits) + it.highest = highest + } + private fun index(n: ULong) = (n % REPLAY_WINDOW_SIZE.toULong()).toInt() + fun containsOrStale(n: ULong): Boolean { + val h = highest ?: return false + if (n > h) return false + if (h - n >= REPLAY_WINDOW_SIZE.toULong()) return true + val i = index(n) + return bits[i / 64] and (1L shl (i % 64)) != 0L + } + fun record(n: ULong) { + val h = highest + if (h == null || n > h) { + if (h == null || n - h >= REPLAY_WINDOW_SIZE.toULong()) bits.fill(0) + else { + // Clear only newly entered positions, not every remembered nonce. + var next = h + repeat((n - h).toInt()) { + val i = index(++next) + bits[i / 64] = bits[i / 64] and (1L shl (i % 64)).inv() + } + } + highest = n + } + val i = index(n) + bits[i / 64] = bits[i / 64] or (1L shl (i % 64)) + } +} + +/** Local-only metadata; does not change wire encoding. */ +internal data class OutgoingData(val bytes: UByteArray, val nonce: ULong? = null) +internal fun interface PreparedPayload { fun encode(): OutgoingData } +internal class UnauthenticatedOuterError : RemoteInterface.SecurityException("Unauthenticated outer error") diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/Transport.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/Transport.kt index 2658ad3..4f23966 100644 --- a/src/commonMain/kotlin/net/sergeych/kiloparsec/Transport.kt +++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/Transport.kt @@ -26,7 +26,7 @@ import kotlinx.serialization.encoding.Decoder import kotlinx.serialization.encoding.Encoder import kotlinx.serialization.serializer import net.sergeych.bipack.Unsigned -import net.sergeych.crypto2.toDump +import net.sergeych.kiloparsec.adapter.PriorityBlockWriter import net.sergeych.mp_logger.* import net.sergeych.utools.pack import net.sergeych.utools.unpack @@ -115,6 +115,9 @@ class Transport( } } + private val outgoing = Mutex() + private val writerReady = CompletableDeferred() + private val responseReaders = mutableMapOf UByteArray>() private val access = Mutex() private var lastId = 0u private val calls = mutableMapOf>() @@ -127,13 +130,18 @@ class Transport( * @throws RemoteInterface.ClosedException */ private suspend fun sendCallBlock(name: String, packedArgs: UByteArray): UByteArray = - sendBoundCallBlock(name) { packedArgs } + sendBoundCallBlock(name) { OutgoingData(packedArgs) } /** Allocate the outer ID before sealing its authenticated inner counterpart. */ - private suspend fun sendBoundCallBlock(name: String, packedArgs: (UInt) -> UByteArray): UByteArray { + private suspend fun sendBoundCallBlock( + name: String, + readResponse: ((UByteArray) -> UByteArray)? = null, + packedArgs: (UInt) -> OutgoingData, + ): UByteArray { if (isClosed) throw RemoteInterface.ClosedException() - val b: Block.Call + val id: UInt + val priority = currentCoroutineContext()[CallPriority]?.value ?: 0 val deferred = CompletableDeferred() // We need to shield calls and lastID with mutex, but nothing more: @@ -141,20 +149,23 @@ class Transport( if (isClosed) throw RemoteInterface.ClosedException() // the order is important: first id in use MUST BE >= 1, not zero: if (lastId == UInt.MAX_VALUE) throw RemoteInterface.ClosedException("RPC ID space exhausted") - val id = ++lastId - b = Block.Call(id, name, packedArgs(id)).also { it.priority = currentCoroutineContext()[CallPriority]?.value ?: 0 } - calls[b.id] = deferred + id = ++lastId + calls[id] = deferred + if (readResponse != null) responseReaders[id] = readResponse } // Suspend on actual channel backpressure; never poll or delay a successful send. try { - device.send(pack(b)) + sendPrepared { + val payload = packedArgs(id) + OutgoingData(pack(Block.Call(id, name, payload.bytes).also { it.priority = priority }), payload.nonce) + } return deferred.await() } catch (_: ClosedSendChannelException) { throw RemoteInterface.ClosedException() } finally { // Cancellation must not retain a response waiter for the connection lifetime. - withContext(NonCancellable) { access.withLock { calls.remove(b.id) } } + withContext(NonCancellable) { access.withLock { calls.remove(id); responseReaders.remove(id) } } } } @@ -169,7 +180,7 @@ class Transport( // All push blocks have the same id == 0: val b = Block.Call(0u, name, packedArgs).also { it.priority = currentCoroutineContext()[CallPriority]?.value ?: 0 } - val r = runCatching { device.send(pack(b).also { debug { ">>$\n${it.toDump()}" } }) } + val r = runCatching { send(b) } when(val e = r.exceptionOrNull()) { null -> return is RemoteInterface.ClosedException, is CancellationException, is RemoteInterface.RemoteException @@ -180,9 +191,34 @@ class Transport( internal suspend fun callWithId( cmd: Command, - build: (UInt) -> UByteArray, - ): UByteArray = unpack(cmd.resultSerializer, - sendBoundCallBlock(cmd.name) { id -> pack(cmd.argsSerializer, build(id)) }) + readResponse: (UByteArray) -> UByteArray = { it }, + build: (UInt) -> OutgoingData, + ): UByteArray = sendBoundCallBlock(cmd.name, { raw -> + try { readResponse(unpack(cmd.resultSerializer, raw)) } + catch (e: DiscardPacket) { throw e } + catch (e: Throwable) { throw RemoteInterface.SecurityException("Invalid authenticated response") } + }) { id -> + val payload = build(id) + OutgoingData(pack(cmd.argsSerializer, payload.bytes), payload.nonce) + } + + internal suspend fun pushPrepared(name: String, priority: Int, prepare: () -> OutgoingData) { + sendPrepared { + val payload = prepare() + OutgoingData(pack(Block.Call(0u, name, pack(payload.bytes)).also { it.priority = priority }), payload.nonce) + } + } + + /** Atomic preparation/admission, with carrier completion awaited outside the mutex. */ + private suspend fun sendPrepared(prepare: () -> OutgoingData) { + val writer = writerReady.await() + val awaitSent = outgoing.withLock { + if (isClosed) throw RemoteInterface.ClosedException() + val frame = prepare() + writer.enqueue(frame.bytes, frame.nonce) + } + awaitSent() + } internal suspend fun abortSecurity() { isClosed = true @@ -210,6 +246,10 @@ class Transport( */ suspend fun run() { coroutineScope { + // One ordered submission to the carrier at a time: adapter queues cannot + // undo the nonce-aware scheduling performed here. + val writer = PriorityBlockWriter(this, continueOnWriteFailure = true) { device.send(it) } + writerReady.complete(writer) debug { "awaiting incoming blocks" } // todo: rewrite it to close the job with no exceptions at all, always while (isActive && !isClosed) { @@ -218,35 +258,56 @@ class Transport( val b = unpack(packed) when (b) { is Block.Error -> access.withLock { - val error = localInterface.decodeError(b) + val error = if (responseReaders.containsKey(b.forId)) UnauthenticatedOuterError() else localInterface.decodeError(b) warning { "decoded error: ${error::class.simpleName}: $error" } - calls.remove(b.forId)?.completeExceptionally(localInterface.decodeError(b)) + responseReaders.remove(b.forId) + calls.remove(b.forId)?.completeExceptionally(error) ?: warning { "error handler not found for ${b.forId}" } info { "error processed" } } is Block.Response -> access.withLock { - calls.remove(b.forId)?.complete(b.packedResult) - ?: warning { "wait handle not found for ${b.forId}" } + val pending = calls[b.forId] + if (pending != null) { + try { + val result = responseReaders[b.forId]?.invoke(b.packedResult) ?: b.packedResult + calls.remove(b.forId) + responseReaders.remove(b.forId) + pending.complete(result) + } catch (_: DiscardPacket) { + // Keep the waiter: a valid response may still arrive. + } catch (failure: Throwable) { + calls.remove(b.forId) + responseReaders.remove(b.forId) + pending.completeExceptionally(failure) + } + } } is Block.Call -> launch { try { - if (b.id == 0u) - // Command does not waits return - localInterface.executeTransportBlock(commandContext, b) - else - send( - Block.Response( - b.id, - localInterface.executeTransportBlock(commandContext, b) - ).also { it.priority = b.priority } - ) + val result = localInterface.executeTransportBlock(commandContext, b) + if (b.id != 0u) { + try { + sendPrepared { + val encoded = result.encode() + OutgoingData(pack(Block.Response(b.id, encoded.bytes).also { it.priority = b.priority }), encoded.nonce) + } + } catch (failure: CancellationException) { throw failure } + catch (failure: LocalInterface.BreakConnectionException) { throw failure } + catch (failure: Throwable) { + // Local reply admission/I/O failure is not a plaintext + // remote business error. Leave peer timeout policy intact. + warning { "Reply send failed: ${failure.message}" } + } + } + } catch (_: DiscardPacket) { + // Duplicate/stale input is silently discarded, including pushes. } catch (x: LocalInterface.BreakConnectionException) { // handler forced close warning { "handler requested closing of the connection (${x.flushSendQueue}" } isClosed = true - if (x.flushSendQueue) device.flush() + if (x.flushSendQueue) { writer.flush(); device.flush() } device.close() } catch (x: CancellationException) { throw x @@ -290,6 +351,8 @@ class Transport( for (c in calls.values) c.completeExceptionally(RemoteInterface.ClosedException()) calls.clear() + responseReaders.clear() + writer.close() debug { "calls clear has been called" } // Closing a device also releases dispatch children waiting on unfinished handshakes. currentCoroutineContext().cancelChildren() @@ -301,7 +364,7 @@ class Transport( private suspend fun send(block: Block) { try { - device.send(pack(block)) + sendPrepared { OutgoingData(pack(block)) } } catch (_: ClosedSendChannelException) { throw RemoteInterface.ClosedException() } diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/OutQueue.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/OutQueue.kt new file mode 100644 index 0000000..bf7b648 --- /dev/null +++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/OutQueue.kt @@ -0,0 +1,25 @@ +package net.sergeych.kiloparsec.adapter + +import net.sergeych.kiloparsec.MAX_NONCE_OVERTAKING + +/** Externally synchronized. Nonces belong to one sending-key stream. */ +internal class OutQueue(private val allowance: ULong = MAX_NONCE_OVERTAKING.toULong()) { + private data class Entry(val value: T, val nonce: ULong?, val priority: Int) + private val entries = mutableListOf>() + val values get() = entries.map { it.value } + fun push(value: T, nonce: ULong?, priority: Int) { entries.add(Entry(value, nonce, priority)) } + fun pop(): T? { + val oldest = entries.mapNotNull { it.nonce }.minOrNull() + var best: Entry? = null + for (entry in entries) { + if (oldest != null && entry.nonce != null && entry.nonce - oldest > allowance) continue + val previous = best + if (previous == null || entry.priority > previous.priority || + (entry.priority == previous.priority && entry.nonce != null && previous.nonce != null && entry.nonce < previous.nonce)) best = entry + } + return best?.also { entries.remove(it) }?.value + } + fun remove(value: T) { entries.removeAll { it.value === value } } + fun removeAll(predicate: (T) -> Boolean) { entries.removeAll { predicate(it.value) } } + fun clear() { entries.clear() } +} diff --git a/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/PriorityBlockWriter.kt b/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/PriorityBlockWriter.kt index 040cd3f..c8f675e 100644 --- a/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/PriorityBlockWriter.kt +++ b/src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/PriorityBlockWriter.kt @@ -43,19 +43,20 @@ internal fun blockPriority(bytes: UByteArray): Int { /** * One adapter writer. Send completes after its write callback, not after an RPC reply. - * Queue ordering is numeric priority then FIFO. Cancellation removes queued work; + * Queue ordering is nonce-bounded numeric priority; unnumbered ties remain FIFO. Cancellation removes queued work; * bytes already being written cannot be recalled. No rate limit or idle batching delay. */ internal class PriorityBlockWriter( scope: CoroutineScope, private val maxQueuedItems: Int = 128, private val maxQueuedBytes: Int = 32 * 1024 * 1024, + private val continueOnWriteFailure: Boolean = false, private val write: suspend (UByteArray) -> Unit, ) { private class Pending(val bytes: UByteArray, val priority: Int, val done: CompletableDeferred) private val mutex = Mutex() private val wake = Channel(Channel.CONFLATED) - private val pending = mutableListOf() + private val pending = OutQueue() private var closed = false private var writing: Pending? = null private val job = scope.launch { @@ -64,13 +65,17 @@ internal class PriorityBlockWriter( while (isActive) { current = mutex.withLock { pending.removeAll { !it.done.isActive } - // maxByOrNull retains the first equal value: stable FIFO ties. - pending.maxByOrNull { it.priority }?.also { pending.remove(it) }.also { writing = it } + pending.pop().also { writing = it } } val entry = current if (entry == null) { wake.receive(); continue } - if (entry.done.isActive) write(entry.bytes) - entry.done.complete(Unit) + try { + if (entry.done.isActive) write(entry.bytes) + entry.done.complete(Unit) + } catch (failure: Throwable) { + entry.done.completeExceptionally(failure) + if (!continueOnWriteFailure || failure is CancellationException) throw failure + } mutex.withLock { writing = null } current = null } @@ -81,32 +86,36 @@ internal class PriorityBlockWriter( withContext(NonCancellable) { mutex.withLock { closed = true - pending.forEach { it.done.completeExceptionally(RemoteInterface.ClosedException()) } + pending.values.forEach { it.done.completeExceptionally(RemoteInterface.ClosedException()) } pending.clear() } } } } - suspend fun send(bytes: UByteArray) { + /** Admission never waits for the carrier. Caller may hold its preparation mutex here. */ + suspend fun enqueue(bytes: UByteArray, nonce: ULong? = null): suspend () -> Unit { currentCoroutineContext().ensureActive() val entry = Pending(bytes, blockPriority(bytes), CompletableDeferred()) mutex.withLock { if (closed || !job.isActive) throw RemoteInterface.ClosedException() pending.removeAll { !it.done.isActive } - if (pending.size >= maxQueuedItems || bytes.size > maxQueuedBytes - pending.sumOf { it.bytes.size }) + if (pending.values.size >= maxQueuedItems || bytes.size > maxQueuedBytes - pending.values.sumOf { it.bytes.size }) throw RemoteInterface.Exception("Outgoing block queue is full") - pending.add(entry) + pending.push(entry, nonce, entry.priority) } wake.trySend(Unit) - try { entry.done.await() } - finally { - entry.done.cancel() - withContext(NonCancellable) { mutex.withLock { pending.remove(entry) } } + return { + try { entry.done.await() } + finally { + entry.done.cancel() + withContext(NonCancellable) { mutex.withLock { pending.remove(entry) } } + } } } + suspend fun send(bytes: UByteArray) { enqueue(bytes).invoke() } suspend fun flush() { - val queued = mutex.withLock { (pending + listOfNotNull(writing)).map { it.done } } + val queued = mutex.withLock { (pending.values + listOfNotNull(writing)).map { it.done } } queued.forEach { it.await() } } fun close() { job.cancel() } diff --git a/src/commonTest/kotlin/AuthenticatedEnvelopeTest.kt b/src/commonTest/kotlin/AuthenticatedEnvelopeTest.kt index dff089d..86ccb60 100644 --- a/src/commonTest/kotlin/AuthenticatedEnvelopeTest.kt +++ b/src/commonTest/kotlin/AuthenticatedEnvelopeTest.kt @@ -127,10 +127,14 @@ class AuthenticatedEnvelopeTest { val clientJob = launch { runCatching { client.run { if (it) ready.complete(Unit) } } } try { ready.await() - withTimeout(2000) { runCatching { client.call(mutate, 1) } } - withTimeout(2000) { serverJob.join() } + assertEquals(1, withTimeout(2000) { client.call(mutate, 1) }) + runCurrent() assertTrue(attacked) - assertTrue(executions <= 1, "The replay must never execute the application operation again") + assertEquals(1, executions, "The replay must never execute the application operation again") + assertTrue(serverJob.isActive) + assertEquals(2, client.call(mutate, 2)) + runCurrent() + assertEquals(2, executions) } finally { listOf(serverJob, clientJob).forEach { it.cancelAndJoin() } } } diff --git a/src/commonTest/kotlin/ConnectionInitializationTest.kt b/src/commonTest/kotlin/ConnectionInitializationTest.kt new file mode 100644 index 0000000..e322fce --- /dev/null +++ b/src/commonTest/kotlin/ConnectionInitializationTest.kt @@ -0,0 +1,88 @@ +@file:OptIn(ExperimentalUnsignedTypes::class) + +import kotlinx.coroutines.* +import kotlinx.coroutines.test.runTest +import net.sergeych.crypto2.SigningSecretKey +import net.sergeych.crypto2.initCrypto +import net.sergeych.kiloparsec.* +import kotlin.test.* +import kotlin.time.Duration.Companion.seconds + +class ConnectionInitializationTest { + private class PolicyRejected : RemoteInterface.SecurityException("Peer rejected by policy") + private class InitializationAborted : LocalInterface.BreakConnectionException("Initialization aborted") + + @Test fun serverSecurityRejectionClosesConnection() = rejectsInitialization(true, PolicyRejected()) + @Test fun clientSecurityRejectionClosesConnection() = rejectsInitialization(false, PolicyRejected()) + @Test fun serverBreakSubclassClosesConnection() = rejectsInitialization(true, InitializationAborted()) + @Test fun clientBreakSubclassClosesConnection() = rejectsInitialization(false, InitializationAborted()) + + private fun rejectsInitialization(rejectOnServer: Boolean, rejection: Throwable) = runTest(timeout = 10.seconds) { + initCrypto() + val (a, b) = createTestDevice() + val serverKey = SigningSecretKey.new() + val clientKey = SigningSecretKey.new() + val clientInterface = KiloInterface() + val serverInterface = KiloInterface() + var policyCalled = false + var identityMatched = false + var laterHandlerCalled = false + var clientConnected = false + (if (rejectOnServer) serverInterface else clientInterface).apply { + onConnected { + identityMatched = remoteIdentity == if (rejectOnServer) clientKey.verifyingKey else serverKey.verifyingKey + policyCalled = true + throw rejection + } + onConnected { laterHandlerCalled = true } + } + val server = KiloServerConnection(serverInterface, b, Unit, serverKey) + val client = KiloClientConnection(clientInterface, a, Unit, clientKey) + val serverRun = async { runCatching { server.run() } } + val clientRun = async { runCatching { client.run { if (it) clientConnected = true } } } + try { + withTimeout(2000) { serverRun.await(); clientRun.await() } + assertTrue(policyCalled, "Initialization policy must run") + assertTrue(identityMatched, "Policy must receive the verified remote identity") + assertFalse(laterHandlerCalled, "Rejection must stop subsequent initialization handlers") + assertFalse(clientConnected, "Rejected initialization must not announce a connected client") + if (!rejectOnServer) assertSame(rejection, clientRun.await().exceptionOrNull()) + val echo by command() + assertIs(runCatching { client.call(echo, "after rejection") }.exceptionOrNull()) + } finally { + serverRun.cancelAndJoin() + clientRun.cancelAndJoin() + } + } + + @Test fun ordinaryHandlerFailuresStillAllowInitialization() = runTest(timeout = 10.seconds) { + initCrypto() + val (a, b) = createTestDevice() + val echo by command() + var serverInitialized = false + var clientInitialized = false + val serverInterface = KiloInterface().apply { + onConnected { throw IllegalStateException("Nonfatal handler failure") } + onConnected { serverInitialized = true } + on(echo) { it } + } + val clientInterface = KiloInterface().apply { + onConnected { throw IllegalStateException("Nonfatal handler failure") } + onConnected { clientInitialized = true } + } + val ready = CompletableDeferred() + val server = KiloServerConnection(serverInterface, b, Unit) + val client = KiloClientConnection(clientInterface, a, Unit) + val serverJob = launch { server.run() } + val clientJob = launch { client.run { if (it) ready.complete(Unit) } } + try { + ready.await() + assertTrue(serverInitialized) + assertTrue(clientInitialized) + assertEquals("ok", client.call(echo, "ok")) + } finally { + serverJob.cancelAndJoin() + clientJob.cancelAndJoin() + } + } +} diff --git a/src/commonTest/kotlin/OutgoingSecurityTest.kt b/src/commonTest/kotlin/OutgoingSecurityTest.kt new file mode 100644 index 0000000..01a4ca1 --- /dev/null +++ b/src/commonTest/kotlin/OutgoingSecurityTest.kt @@ -0,0 +1,114 @@ +@file:OptIn(ExperimentalUnsignedTypes::class, kotlinx.coroutines.ExperimentalCoroutinesApi::class) + +import kotlinx.coroutines.* +import kotlinx.coroutines.test.* +import net.sergeych.crypto2.initCrypto +import net.sergeych.kiloparsec.* +import net.sergeych.utools.pack +import net.sergeych.utools.unpack +import kotlin.test.* +import kotlin.time.Duration.Companion.seconds + +class OutgoingSecurityTest { + @Test fun replayedReplyLeavesWaiterForValidReply() = runTest(timeout = 10.seconds) { + initCrypto() + val echo by command() + val (a, b) = createTestDevice() + var armed = false + var saved: UByteArray? = null + val adversary = object : Transport.Device by b { + override suspend fun send(block: UByteArray) { + val bytes = block + val decoded = unpack(bytes) + if (armed && decoded is Transport.Block.Response) { + saved?.let { b.send(pack(decoded.copy(packedResult = it))) } + saved = decoded.packedResult + } + b.send(bytes) + } + } + val ready = CompletableDeferred() + val server = KiloServerConnection(KiloInterface().apply { on(echo) { it } }, adversary, Unit) + val client = KiloClientConnection(KiloInterface(), a, Unit) + val sj = launch { server.run() } + val cj = launch { client.run { if (it) ready.complete(Unit) } } + try { + ready.await(); armed = true + assertEquals("first", client.call(echo, "first")) + assertEquals("second", client.call(echo, "second")) + assertTrue(sj.isActive); assertTrue(cj.isActive) + } finally { sj.cancelAndJoin(); cj.cancelAndJoin() } + } + + @Test fun localReplySendFailureDoesNotForgePlaintextErrorOrClose() = runTest(timeout = 10.seconds) { + initCrypto() + val echo by command() + val (a, b) = createTestDevice() + var fail = false + var plaintextErrors = 0 + val device = object : Transport.Device by b { + override suspend fun send(block: UByteArray) { + val bytes = block + val decoded = unpack(bytes) + if (decoded is Transport.Block.Error) plaintextErrors++ + if (fail && decoded is Transport.Block.Response) throw RemoteInterface.Exception("Local write failure") + b.send(bytes) + } + } + val ready = CompletableDeferred() + val server = KiloServerConnection(KiloInterface().apply { on(echo) { it } }, device, Unit) + val client = KiloClientConnection(KiloInterface(), a, Unit) + val sj = launch { server.run() } + val cj = launch { client.run { if (it) ready.complete(Unit) } } + try { + ready.await(); fail = true + assertNull(withTimeoutOrNull(100) { client.call(echo, 1) }) + assertEquals(0, plaintextErrors) + fail = false + assertEquals(2, client.call(echo, 2)) + } finally { sj.cancelAndJoin(); cj.cancelAndJoin() } + } + + @Test fun allSecurePathsShareOrderedNonceAdmission() = runTest(timeout = 20.seconds) { + initCrypto() + val echo by command() + val notify by command() + val (a, b) = createTestDevice() + fun observing(device: Transport.Device, nonces: MutableList, skipHandshakeReply: Boolean = false) = object : Transport.Device by device { + override suspend fun send(block: UByteArray) { + val bytes = block + when (val decoded = unpack(bytes)) { + is Transport.Block.Call -> if (decoded.name != "L0Request") + nonces += unpack(unpack(decoded.packedArgs)).nonce + is Transport.Block.Response -> if (!skipHandshakeReply || decoded.forId != 1u) + nonces += unpack(unpack(decoded.packedResult)).nonce + else -> Unit + } + device.send(bytes) + } + } + val an = mutableListOf(); val bn = mutableListOf() + val api = KiloInterface().apply { on(echo) { it }; on(notify) {} } + val ready = CompletableDeferred() + val server = KiloServerConnection(api, observing(b, bn, true), Unit) + val client = KiloClientConnection(api, observing(a, an), Unit) + val sj = launch { server.run() } + val cj = launch { client.run { if (it) ready.complete(Unit) } } + try { + ready.await() + // Real JVM parallel preparation, requests in both directions, replies and pushes. + withContext(Dispatchers.Default) { + (0 until 32).map { n -> async { + val remote: RemoteInterface = if (n % 2 == 0) client else server + remote.push(notify, n) + assertEquals(n, remote.call(echo, n)) + } }.awaitAll() + } + for (nonces in listOf(an, bn)) { + assertEquals(0uL, nonces.first()) + assertTrue(nonces.zipWithNext().all { (x, y) -> y > x }, "Equal-priority admission must follow nonce allocation: $nonces") + assertTrue(nonces.size >= 49) + } + } finally { sj.cancelAndJoin(); cj.cancelAndJoin() } + } +} diff --git a/src/commonTest/kotlin/ReplayDeliveryTest.kt b/src/commonTest/kotlin/ReplayDeliveryTest.kt new file mode 100644 index 0000000..677b872 --- /dev/null +++ b/src/commonTest/kotlin/ReplayDeliveryTest.kt @@ -0,0 +1,97 @@ +@file:OptIn(ExperimentalUnsignedTypes::class) +import kotlinx.coroutines.* +import kotlinx.coroutines.test.* +import net.sergeych.crypto2.* +import net.sergeych.kiloparsec.* +import net.sergeych.utools.pack +import net.sergeych.utools.unpack +import kotlin.test.* +import kotlin.time.Duration.Companion.seconds + +class ReplayDeliveryTest { + @Test fun replayWindowBoundaries() = runTest { + initCrypto() + val a = SafeKeyExchange(); val b = SafeKeyExchange() + val (da, db) = createTestDevice() + val ta = Transport(da, LocalInterface(), Unit) + val tb = Transport(db, LocalInterface(), Unit) + val sender = KiloParams(false, ta, a.clientSessionKey(b.publicKey), Unit, null, ta) + val receiver = KiloParams(true, tb, b.serverSessionKey(a.publicKey), Unit, null, tb) + val packets = (0..4096).map { sender.encrypt(pack(it)) } + assertEquals(4096, unpack(receiver.decrypt(packets[4096]))) + assertEquals(2, unpack(receiver.decrypt(packets[2]))) + assertEquals(1, unpack(receiver.decrypt(packets[1]))) + assertFails { receiver.decrypt(packets[0]) } + assertFails { receiver.decrypt(packets[1]) } + da.close(); db.close() + } + @Test fun localSendFailurePreservesHealthyConnection() = runTest(timeout = 10.seconds) { + initCrypto() + val echo by command() + val (a,b) = createTestDevice() + val ready = CompletableDeferred() + var fail = true + val localFailure = object: Transport.Device by a { + override suspend fun send(block: UByteArray) { + val decoded = unpack(block) + if(fail && decoded is Transport.Block.Call && decoded.name == "L0Call") + throw RemoteInterface.Exception("Outgoing block queue is full") + a.send(block) + } + } + val server=KiloServerConnection(KiloInterface().apply { on(echo) { it } }, b, Unit) + val client=KiloClientConnection(KiloInterface(),localFailure,Unit) + val sj=launch { runCatching { server.run() } } + val cj=launch { runCatching { client.run { if(it) ready.complete(Unit) } } } + try { + ready.await() + val result=runCatching { client.call(echo,"x") } + assertIs(result.exceptionOrNull()) + assertFalse(result.exceptionOrNull() is RemoteInterface.SecurityException) + assertTrue(cj.isActive); assertTrue(sj.isActive) + fail = false + assertEquals("recovered", client.call(echo,"recovered")) + } finally { sj.cancelAndJoin(); cj.cancelAndJoin() } + } + + @Test fun adjacentReorderingSurvives() = reorderedPushes(1, false) + @Test fun deepReorderingDiscardsWithoutClosing() = reorderedPushes(4096, true) + private fun reorderedPushes(overtakers: Int, shouldDiscard: Boolean) = runTest(timeout = 20.seconds) { + initCrypto() + val notify by command() + val (a,b)=createTestDevice() + val ready=CompletableDeferred() + val newerReceived=CompletableDeferred() + val heldReceived=CompletableDeferred() + var held: UByteArray?=null + val reorder=object: Transport.Device by a { + override suspend fun send(block: UByteArray) { + val decoded=unpack(block) + if(decoded is Transport.Block.Call && decoded.name=="L0Call" && held==null) held=block + else a.send(block) + } + } + val server=KiloServerConnection(KiloInterface().apply { on(notify) { + if(it==overtakers) newerReceived.complete(Unit) + if(it==0) heldReceived.complete(Unit) + } }, b, Unit) + val client=KiloClientConnection(KiloInterface(),reorder,Unit) + val sj=launch { runCatching { server.run() } } + val cj=launch { runCatching { client.run { if(it) ready.complete(Unit) } } } + try { + ready.await() + for(i in 0..overtakers) client.push(notify,i) + newerReceived.await() + a.send(held!!) + if(shouldDiscard) { + runCurrent() + assertTrue(sj.isActive); assertTrue(cj.isActive) + assertFalse(heldReceived.isCompleted) + client.ping() + } else { + withTimeout(2000) { heldReceived.await() } + assertTrue(sj.isActive); assertTrue(cj.isActive) + } + } finally { sj.cancelAndJoin(); cj.cancelAndJoin() } + } +} diff --git a/src/commonTest/kotlin/ReplayWindowTest.kt b/src/commonTest/kotlin/ReplayWindowTest.kt new file mode 100644 index 0000000..9d6005e --- /dev/null +++ b/src/commonTest/kotlin/ReplayWindowTest.kt @@ -0,0 +1,28 @@ +@file:OptIn(ExperimentalUnsignedTypes::class) +package net.sergeych.kiloparsec + +import kotlin.random.Random +import kotlin.test.* + +class ReplayWindowTest { + @Test fun boundariesWraparoundAndLargeJumpsMatchReference() { + val window = ReplayWindow() + val seen = mutableSetOf() + var highest: ULong? = null + val random = Random(41) + val inputs = mutableListOf(4096uL, 1uL, 2uL, 0uL, 1uL, 8191uL, 4096uL, 8192uL) + repeat(15000) { inputs += random.nextInt(0, 30000).toULong() } + inputs += listOf(ULong.MAX_VALUE - 1u, ULong.MAX_VALUE, ULong.MAX_VALUE - 4095u, ULong.MAX_VALUE - 4096u) + for (n in inputs) { + val h = highest + val reject = n in seen || (h != null && n <= h && h - n >= 4096u) + assertEquals(reject, window.containsOrStale(n), "nonce=$n highest=$h") + if (!reject) { + window.record(n) + seen += n + if (h == null || n > h) highest = n + assertTrue(window.containsOrStale(n)) + } + } + } +} diff --git a/src/commonTest/kotlin/net/sergeych/kiloparsec/adapter/OutQueueTest.kt b/src/commonTest/kotlin/net/sergeych/kiloparsec/adapter/OutQueueTest.kt new file mode 100644 index 0000000..67ae5cf --- /dev/null +++ b/src/commonTest/kotlin/net/sergeych/kiloparsec/adapter/OutQueueTest.kt @@ -0,0 +1,123 @@ +@file:OptIn(ExperimentalUnsignedTypes::class, kotlinx.coroutines.ExperimentalCoroutinesApi::class) +package net.sergeych.kiloparsec.adapter + +import createTestDevice +import kotlinx.coroutines.* +import kotlinx.coroutines.test.* +import net.sergeych.kiloparsec.* +import net.sergeych.utools.pack +import net.sergeych.utools.unpack +import kotlin.test.* + +class OutQueueTest { + @Test fun priorityWithinWindowAndOldestProgress() { + val queue = OutQueue() + queue.push(100u, 100u, -100) + for (n in 101uL..5000uL) queue.push(n, n, 100) + val sent = mutableListOf() + while (true) sent += queue.pop() ?: break + assertEquals((101uL..2830uL).toList(), sent.take(2730)) + assertEquals(100uL, sent[2730]) + assertEquals(4901, sent.toSet().size) + } + @Test fun gapsAndMaximumNonceDoNotOverflow() { + val queue = OutQueue() + queue.push(ULong.MAX_VALUE, ULong.MAX_VALUE, 100) + queue.push(ULong.MAX_VALUE - 2731u, ULong.MAX_VALUE - 2731u, -100) + assertEquals(ULong.MAX_VALUE - 2731u, queue.pop()) + assertEquals(ULong.MAX_VALUE, queue.pop()) + assertNull(queue.pop()) + } + @Test fun writerHandlesCancellationCapacityAndContinuesAfterLocalFailure() = runTest { + val gate = CompletableDeferred() + val entered = CompletableDeferred() + val observed = mutableListOf() + val writer = PriorityBlockWriter(backgroundScope, maxQueuedItems = 2, continueOnWriteFailure = true) { + val n = it[0].toInt() + if (n == 0) { entered.complete(Unit); gate.await() } + if (n == 3) throw IllegalStateException("Local carrier rejection") + observed += n + } + val first = async { writer.enqueue(ubyteArrayOf(0u), 0u).invoke() } + entered.await() + val cancelled = async { writer.enqueue(ubyteArrayOf(1u), 1u).invoke() } + runCurrent() + val second = async { writer.enqueue(ubyteArrayOf(2u), 2u).invoke() } + runCurrent() + assertFailsWith { writer.enqueue(ubyteArrayOf(9u), 9u) } + cancelled.cancelAndJoin() + gate.complete(Unit) + first.await(); second.await() + assertFailsWith { writer.enqueue(ubyteArrayOf(3u), 3u).invoke() } + writer.enqueue(ubyteArrayOf(4u), 4u).invoke() + assertEquals(listOf(0, 2, 4), observed) + writer.close(); runCurrent() + assertFailsWith { writer.enqueue(ubyteArrayOf(5u), 5u) } + } + @Test fun writerUsesNonceBoundBeforePriority() = runTest { + val gate = CompletableDeferred() + val entered = CompletableDeferred() + val order = mutableListOf() + val writer = PriorityBlockWriter(backgroundScope) { bytes -> + val n = (unpack(bytes) as Transport.Block.Call).id.toInt() + if (n == 0) { entered.complete(Unit); gate.await() } + order += n + } + fun frame(n: Int, priority: Int) = pack(Transport.Block.Call(n.toUInt(), "x", ubyteArrayOf()).also { it.priority = priority }) + val first = async { writer.enqueue(frame(0, 0), 0u).invoke() } + entered.await() + val old = writer.enqueue(frame(1, 0), 1u) + val newer = writer.enqueue(frame(2, 100), 2732u) + gate.complete(Unit); first.await(); old(); newer() + assertEquals(3, order.size) + assertEquals(listOf(0, 1, 2), order) + writer.close() + } + + @Test fun encryptedBacklogCrossingWindowStillDeliversEveryMessage() = runTest { + net.sergeych.crypto2.initCrypto() + val a = net.sergeych.crypto2.SafeKeyExchange() + val b = net.sergeych.crypto2.SafeKeyExchange() + val (da, db) = createTestDevice() + val ta = Transport(da, LocalInterface(), Unit) + val tb = Transport(db, LocalInterface(), Unit) + val sender = KiloParams(false, ta, a.clientSessionKey(b.publicKey), Unit, null, ta) + val receiver = KiloParams(true, tb, b.serverSessionKey(a.publicKey), Unit, null, tb) + val gate = CompletableDeferred() + val entered = CompletableDeferred() + val order = mutableListOf() + // Larger test capacity reaches the boundary in one batch; production retains 128. + val writer = PriorityBlockWriter(backgroundScope, maxQueuedItems = 5000) { bytes -> + val frame = unpack(bytes) as Transport.Block.Call + if (frame.id == 0u) { entered.complete(Unit); gate.await() } + order += unpack(receiver.decrypt(unpack(frame.packedArgs))) + } + suspend fun admit(n: Int, priority: Int): suspend () -> Unit { + val encoded = sender.encode(pack(n)) + val frame = Transport.Block.Call(n.toUInt(), "L0Call", pack(encoded.bytes)).also { it.priority = priority } + return writer.enqueue(pack(frame), encoded.nonce) + } + try { + val first = admit(0, 0) + entered.await() + val completions = mutableListOf(admit(1, -100)) + for (n in 2..4097) completions += admit(n, 100) + gate.complete(Unit); first() + completions.forEach { it() } + assertEquals(4098, order.size) + assertEquals((0..4097).toSet(), order.toSet()) + assertEquals(2731, order.indexOf(1)) + } finally { writer.close(); da.close(); db.close() } + } + + @Test fun closingWriterReleasesCurrentAndQueuedSenders() = runTest { + val entered = CompletableDeferred() + val writer = PriorityBlockWriter(backgroundScope) { entered.complete(Unit); awaitCancellation() } + val first = async { runCatching { writer.send(ubyteArrayOf(0u)) } } + entered.await() + val second = async { runCatching { writer.enqueue(ubyteArrayOf(1u), 1u).invoke() } } + runCurrent(); writer.close() + assertTrue(first.await().isFailure) + assertTrue(second.await().isFailure) + } +} diff --git a/src/jvmTest/kotlin/net/sergeych/kiloparsec/adapter/CarrierReleaseTest.kt b/src/jvmTest/kotlin/net/sergeych/kiloparsec/adapter/CarrierReleaseTest.kt new file mode 100644 index 0000000..ef08d00 --- /dev/null +++ b/src/jvmTest/kotlin/net/sergeych/kiloparsec/adapter/CarrierReleaseTest.kt @@ -0,0 +1,91 @@ +@file:OptIn(ExperimentalUnsignedTypes::class) +package net.sergeych.kiloparsec.adapter + +import kotlinx.coroutines.* +import kotlinx.coroutines.channels.Channel +import net.sergeych.crypto2.SigningSecretKey +import net.sergeych.crypto2.initCrypto +import net.sergeych.kiloparsec.* +import net.sergeych.utools.unpack +import kotlin.test.* + +/** Real loopback carriers; impairment occurs before writing actual datagrams. */ +class CarrierReleaseTest { + @Test fun udpDuplicatesReorderingAndConcurrentDuplex() = carrier(udp = true) + @Test fun tcpConcurrentDuplexAndPriority() = carrier(udp = false) + + private fun carrier(udp: Boolean) = runBlocking { + withTimeout(20_000) { + initCrypto() + val tcp = if (!udp) TcpServer(0, "127.0.0.1") else null + val datagrams = if (udp) UdpServer(0, "127.0.0.1") else null + val received = Channel(Channel.UNLIMITED) + val echo by command() + val notify by command() + val api = KiloInterface().apply { + on(echo) { it } + on(notify) { received.send(it) } + } + val serverReady = CompletableDeferred>() + val serverKey = SigningSecretKey.new() + val collector = launch { + (tcp?.transportFlow ?: datagrams!!.transportFlow).collect { device -> + val server = KiloServerConnection(api, device, Unit, serverKey) + serverReady.complete(server) + launch { server.run() } + } + } + val port = tcp?.boundPort() ?: datagrams!!.boundPort() + val raw = if (udp) connectUdpDevice(NetworkAddress("127.0.0.1", port)) + else connectTcpDevice(NetworkAddress("127.0.0.1", port)) + var held: UByteArray? = null + var impaired = false + val device = object : Transport.Device by raw { + override suspend fun send(block: UByteArray) { + val decoded = unpack(block) + if (udp && !impaired && decoded is Transport.Block.Call && decoded.name == "L0Call" && decoded.id == 0u) { + val first = held + if (first == null) held = block + else { + raw.send(block) + raw.send(first) + raw.send(first) // exact duplicate over the real UDP socket + impaired = true + } + } else raw.send(block) + } + } + val ready = CompletableDeferred() + val clientApi = KiloInterface().apply { + onConnected { if (remoteIdentity != serverKey.verifyingKey) throw RemoteInterface.SecurityException() } + on(echo) { it } + } + val client = KiloClientConnection(clientApi, device, Unit) + val clientJob = launch { client.run { if (it) ready.complete(Unit) } } + try { + ready.await() + val server = serverReady.await() + client.push(notify, 1) + client.push(notify, 2) + val notices = setOf(received.receive(), received.receive()) + assertEquals(setOf(1, 2), notices) + coroutineScope { + (0 until 32).map { n -> async { + val peer: RemoteInterface = if (n % 2 == 0) client else server + assertEquals(n, peer.call(echo, n, priority = if (n % 3 == 0) 100 else 0)) + } }.awaitAll() + } + client.ping() + assertNull(withTimeoutOrNull(100) { received.receive() }, "Duplicate must not execute again") + assertTrue(clientJob.isActive) + if (udp) assertTrue(impaired) + } finally { + clientJob.cancelAndJoin() + raw.close() + collector.cancelAndJoin() + tcp?.close(); datagrams?.close() + received.close() + } + } + } +} diff --git a/typescript/README.md b/typescript/README.md index ed6b24a..eed5b45 100644 --- a/typescript/README.md +++ b/typescript/README.md @@ -90,7 +90,7 @@ Long.MIN_VALUE correctly in this baseline; TS rejects it explicitly. Packet sizes, concurrent incoming work, pending calls and send buffers are bounded. Receive replay tracking permits out-of-order counters within 4096 positions and -fails closed outside the window. Authenticated envelopes additionally bind packet kind, RPC ID and priority +silently discards duplicates and packets outside the window without closing the session. Authenticated envelopes additionally bind packet kind, RPC ID and priority to the encrypted payload. Priority trailers are preserved, but this first client uses WebSocket FIFO rather than a priority scheduler. No TCP/UDP adapters, carrier cycling, automatic reconnection, or code diff --git a/typescript/docs/SECURITY.md b/typescript/docs/SECURITY.md index d0a554e..499602a 100644 --- a/typescript/docs/SECURITY.md +++ b/typescript/docs/SECURITY.md @@ -32,7 +32,11 @@ is not accepted as an authenticated application result. Session nonces are replay-checked on both peers, including pushes. A bounded 4096-position window supports responses completed out of order and rejects replay -or excessively stale traffic. Send nonces and transport IDs do not wrap. Nonce +or excessively stale traffic by silently discarding those packets, without closing +the session. A discarded response leaves the pending call waiting for a valid reply +or its timeout/cancellation. Replay state uses a circular bitmap; unauthenticated +packets never advance it. Kotlin bounds local priority overtaking to 2730 nonce +positions and serializes nonce allocation, sealing and queue admission. Send nonces and transport IDs do not wrap. Nonce state used during identity exchange is preserved when the identity is completed. The routing envelope still allows denial of service by an active carrier: packets @@ -79,7 +83,8 @@ silently invent a key provisioning or trust-on-first-use policy. - `test/security.test.mjs`: swapped responses reject; wrong pin, version/role/token signature substitution, forged/mismatched finished, expired/tampered server Seal, - damaged/replayed packets, altered call IDs/kinds/priorities fail closed. + damaged packets and altered call IDs/kinds/priorities fail closed; duplicates + and stale packets are discarded without executing again or closing the session. - Kotlin `AuthenticatedEnvelopeTest`: normal signed duplex and session identities, response-ID swap, call-to-push rewrite, priority rewrite, replay, forged Unit acknowledgement and protocol downgrade. diff --git a/typescript/evaluation/baseline/build.gradle.kts b/typescript/evaluation/baseline/build.gradle.kts index c7e35c3..a330454 100644 --- a/typescript/evaluation/baseline/build.gradle.kts +++ b/typescript/evaluation/baseline/build.gradle.kts @@ -1,5 +1,5 @@ plugins { kotlin("multiplatform"); kotlin("plugin.serialization") } kotlin { js { nodejs(); binaries.executable(); generateTypeScriptDefinitions() } - sourceSets { jsMain.dependencies { implementation("net.sergeych:kiloparsec:0.8.0-SNAPSHOT") } } + sourceSets { jsMain.dependencies { implementation("net.sergeych:kiloparsec:0.8.0") } } } diff --git a/typescript/evaluation/server/build.gradle.kts b/typescript/evaluation/server/build.gradle.kts index 306b8f7..da7ae95 100644 --- a/typescript/evaluation/server/build.gradle.kts +++ b/typescript/evaluation/server/build.gradle.kts @@ -1,7 +1,7 @@ plugins { kotlin("jvm"); kotlin("plugin.serialization"); application } kotlin { jvmToolchain(21); compilerOptions { optIn.add("kotlin.ExperimentalUnsignedTypes") } } dependencies { - implementation("net.sergeych:kiloparsec:0.8.0-SNAPSHOT") + implementation("net.sergeych:kiloparsec:0.8.0") implementation("io.ktor:ktor-server-netty:3.1.1") implementation("org.slf4j:slf4j-simple:2.0.17") } diff --git a/typescript/evaluation/wasm-baseline/build.gradle.kts b/typescript/evaluation/wasm-baseline/build.gradle.kts index 0a743cb..8ac81c6 100644 --- a/typescript/evaluation/wasm-baseline/build.gradle.kts +++ b/typescript/evaluation/wasm-baseline/build.gradle.kts @@ -4,5 +4,5 @@ plugins { kotlin("multiplatform"); kotlin("plugin.serialization") } kotlin { @OptIn(ExperimentalWasmDsl::class) wasmJs { nodejs(); binaries.executable() } - sourceSets { wasmJsMain.dependencies { implementation("net.sergeych:kiloparsec:0.8.0-SNAPSHOT") } } + sourceSets { wasmJsMain.dependencies { implementation("net.sergeych:kiloparsec:0.8.0") } } } diff --git a/typescript/src/client.ts b/typescript/src/client.ts index 4524fb3..1bfabb8 100644 --- a/typescript/src/client.ts +++ b/typescript/src/client.ts @@ -1,3 +1,4 @@ +import { ReplayWindow, DiscardPacket } from './replay.js'; import { bytes, int64, type Codec } from './bipack.js'; import { blockCodec, handshakeCodec, clientIdentityCodec, encryptedPackageCodec, commandCallCodec, sealedDataCodec, protectedMessageCodec, authenticationMessage, type Block } from './wire.js'; import { createKeyExchange, verifySignature, type SessionCrypto } from './crypto.js'; @@ -51,7 +52,7 @@ export interface ConnectOptions { /** For a host-provided WebSocket implementation. Must implement the standard binary WebSocket API. */ readonly socketFactory?: (url: string) => WebSocket; } -type Pending = { resolve(value: Uint8Array): void; reject(error: unknown): void; cleanup(): void }; +type Pending = { resolve(value: Uint8Array): void; reject(error: unknown): void; cleanup(): void; readResponse?: (raw: Uint8Array) => Uint8Array }; const pingCommand = command('@kiloparsec/ping', int64, int64); const empty = new Uint8Array(); function bound(value: number, label: string, min = 1): number { @@ -74,8 +75,7 @@ export class Client { #token: Uint8Array = empty; #remoteIdentity: Uint8Array = empty; #nonce = 0n; - #highestReceived = -1n; - #received = new Set(); + #received = new ReplayWindow(); #nextId = 1; #pending = new Map(); #incoming = new Set(); @@ -99,7 +99,7 @@ export class Client { if (!(event.data instanceof ArrayBuffer)) throw new ConnectionError('protocol', 'Expected binary WebSocket message'); if (event.data.byteLength > this.maxPacketBytes) throw new ConnectionError('protocol', 'Packet too large'); this.#receive(blockCodec.decode(new Uint8Array(event.data))); - } catch { this.#fail(new ConnectionError('protocol', 'Invalid peer packet')); } + } catch (error) { if (!(error instanceof DiscardPacket)) this.#fail(new ConnectionError('protocol', 'Invalid peer packet')); } }; #onClose = () => this.#fail(new ConnectionError('closed')); #onError = () => this.#fail(new ConnectionError('closed', 'WebSocket failed')); @@ -163,8 +163,8 @@ export class Client { // The trusted server may call us from its onConnected handler before replying. client.#ready = true; const identityBytes = clientIdentityCodec.encode({ clientIdKey: publicKey, signature }); - const result = await client.#rpc('L0ClientId', bytes.encode(client.#encrypt(identityBytes)), { signal: options.signal }); - const finished = protectedMessageCodec.decode(client.#decrypt(bytes.decode(result))); + const result = await client.#rpc('L0ClientId', bytes.encode(client.#encrypt(identityBytes)), { signal: options.signal }, raw => client.#decrypt(bytes.decode(raw))); + const finished = protectedMessageCodec.decode(result); if (finished.kind !== 'finished' || finished.correlation !== 0n || finished.priority !== 0 || !same(finished.payload, identityBytes)) { throw new ConnectionError('authentication', 'Invalid server identity confirmation'); } @@ -198,7 +198,7 @@ export class Client { if (packet.length > this.maxPacketBytes || this.#socket.bufferedAmount + packet.length > this.maxPacketBytes * 2) throw new ConnectionError('overloaded', 'Send buffer limit'); this.#socket.send(packet); } - #rpc(name: string, packedArgs: Uint8Array | ((id: number) => Uint8Array), options: CallOptions = {}): Promise { + #rpc(name: string, packedArgs: Uint8Array | ((id: number) => Uint8Array), options: CallOptions = {}, readResponse?: (raw: Uint8Array) => Uint8Array): Promise { if (this.#closed) return Promise.reject(new ConnectionError('closed')); if (options.signal?.aborted) return Promise.reject(new ConnectionError('aborted')); if (this.#pending.size >= this.maxPendingCalls) return Promise.reject(new ConnectionError('overloaded')); @@ -210,7 +210,7 @@ export class Client { const abort = () => finish(new ConnectionError('aborted')); const timer = setTimeout(() => finish(new ConnectionError('timeout')), timeout); const cleanup = () => { clearTimeout(timer); options.signal?.removeEventListener('abort', abort); }; - this.#pending.set(id, { resolve, reject, cleanup }); + this.#pending.set(id, { resolve, reject, cleanup, readResponse }); options.signal?.addEventListener('abort', abort, { once: true }); try { this.#send({ kind: 'call', id, name, packedArgs: typeof packedArgs === 'function' ? packedArgs(id) : packedArgs, priority: options.priority ?? 0 }); } catch (error) { finish(error); } @@ -221,20 +221,25 @@ export class Client { if (block.kind !== 'call') { const pending = this.#pending.get(block.forId); if (!pending) return; // Late response after timeout/abort; no implicit retry. - this.#pending.delete(block.forId); pending.cleanup(); - if (block.kind === 'response') pending.resolve(block.packedResult); + // Decode/authenticate before consuming the waiter: duplicates and stale replies + // must leave it available for a later valid response (or timeout/abort). + if (block.kind === 'response') { + const result = pending.readResponse ? pending.readResponse(block.packedResult) : block.packedResult; + this.#pending.delete(block.forId); pending.cleanup(); pending.resolve(result); + } else { + this.#pending.delete(block.forId); pending.cleanup(); const error = new ConnectionError('protocol', 'Unencrypted outer RPC error'); pending.reject(error); this.#fail(error); } return; } if (block.name !== 'L0Call' || !this.#crypto || !this.#remoteIdentity.length) throw new ConnectionError('protocol'); - if (this.#incomingCount >= this.maxIncomingCalls || (block.id !== 0 && this.#incoming.has(block.id))) throw new ConnectionError('overloaded'); const message = protectedMessageCodec.decode(this.#decrypt(bytes.decode(block.packedArgs))); if (message.kind !== (block.id === 0 ? 'push' : 'call') || message.correlation !== BigInt(block.id) || message.priority !== (block.priority ?? 0)) { throw new ConnectionError('protocol', 'Unauthenticated call metadata'); } + if (this.#incomingCount >= this.maxIncomingCalls || (block.id !== 0 && this.#incoming.has(block.id))) throw new ConnectionError('overloaded'); const call = commandCallCodec.decode(message.payload); this.#incomingCount++; if (block.id) this.#incoming.add(block.id); @@ -265,14 +270,9 @@ export class Client { #decrypt(payload: Uint8Array): Uint8Array { if (!this.#crypto || this.#closed) throw new ConnectionError('closed'); const pkg = encryptedPackageCodec.decode(payload); - // Responses may finish out of order. Bounded replay window fails closed when too old. - if (this.#received.has(pkg.nonce) || pkg.nonce <= this.#highestReceived - 4096n) throw new ConnectionError('protocol', 'Replayed or stale encrypted packet'); + if (this.#received.containsOrStale(pkg.nonce)) throw new DiscardPacket(); const clear = this.#crypto.decrypt(pkg.nonce, pkg.encryptedMessage); - this.#received.add(pkg.nonce); - if (pkg.nonce > this.#highestReceived) { - this.#highestReceived = pkg.nonce; - for (const n of this.#received) if (n <= pkg.nonce - 4096n) this.#received.delete(n); - } + this.#received.record(pkg.nonce); return clear; } /** Execute a typed command. Timeout/abort stops waiting, never cancels remote work. */ @@ -285,10 +285,10 @@ export class Client { requestId = id; return bytes.encode(this.#encrypt(protectedMessageCodec.encode({kind: 'call', correlation: BigInt(id), priority, payload: commandCallCodec.encode({ name: cmd.name, serializedArgs: cmd.args.encode(args) }) }))); - }, options); + }, options, raw => this.#decrypt(bytes.decode(raw))); let reply: Block; try { - const message = protectedMessageCodec.decode(this.#decrypt(bytes.decode(raw))); + const message = protectedMessageCodec.decode(raw); if ((message.kind !== 'response' && message.kind !== 'error') || message.correlation !== BigInt(requestId) || message.priority !== priority) { throw new ConnectionError('protocol', 'Response metadata does not match request'); } diff --git a/typescript/src/replay.ts b/typescript/src/replay.ts new file mode 100644 index 0000000..e185593 --- /dev/null +++ b/typescript/src/replay.ts @@ -0,0 +1,25 @@ +/** Bounded circular bitmap; only authenticated messages may call record(). */ +export class ReplayWindow { + #highest = -1n; + #bits = new Uint32Array(128); + containsOrStale(n: bigint): boolean { + if (n > this.#highest) return false; + if (this.#highest - n >= 4096n) return true; + const i = Number(n % 4096n); + return (this.#bits[i >>> 5]! & (1 << (i & 31))) !== 0; + } + record(n: bigint): void { + if (n > this.#highest) { + if (this.#highest < 0n || n - this.#highest >= 4096n) this.#bits.fill(0); + else for (let next = this.#highest + 1n; next <= n; next++) { + const i = Number(next % 4096n); + this.#bits[i >>> 5] = this.#bits[i >>> 5]! & ~(1 << (i & 31)); + } + this.#highest = n; + } + const i = Number(n % 4096n); + this.#bits[i >>> 5] = this.#bits[i >>> 5]! | (1 << (i & 31)); + } + clear(): void { this.#highest = -1n; this.#bits.fill(0); } +} +export class DiscardPacket extends Error {} diff --git a/typescript/test/replay.test.mjs b/typescript/test/replay.test.mjs new file mode 100644 index 0000000..d7d5778 --- /dev/null +++ b/typescript/test/replay.test.mjs @@ -0,0 +1,17 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {ReplayWindow} from '../dist/replay.js'; + +test('circular replay bitmap agrees with a reference across wraparound and jumps', () => { + const window = new ReplayWindow(), seen = new Set(); + let high = -1n, seed = 17; + const inputs = [4096n, 1n, 2n, 0n, 1n, 8191n, 4096n, 8192n]; + for (let i=0;i<15000;i++) { seed = (Math.imul(seed,1664525)+1013904223)>>>0; inputs.push(BigInt(seed % 30000)); } + inputs.push((1n<<64n)-1n, (1n<<64n)-4096n, (1n<<64n)-4097n); + for (const n of inputs) { + const reject = seen.has(n) || (n <= high && high-n >= 4096n); + assert.equal(window.containsOrStale(n), reject); + if (!reject) { window.record(n); seen.add(n); if(n>high) high=n; assert.equal(window.containsOrStale(n),true); } + } + window.clear(); assert.equal(window.containsOrStale(0n),false); +}); diff --git a/typescript/test/security.test.mjs b/typescript/test/security.test.mjs index 9c20be2..f58b718 100644 --- a/typescript/test/security.test.mjs +++ b/typescript/test/security.test.mjs @@ -25,6 +25,7 @@ async function peer(mode = 'normal') { return bytes.encode(encryptedPackageCodec.encode({nonce:n, encryptedMessage:session.encrypt(n,value)})); }; async function receive(block) { + if (block.kind !== 'call') return; if (block.name === 'L0Request') { const request = handshakeCodec.decode(block.packedArgs); session = exchange.serverSession(request.publicKey); @@ -47,6 +48,10 @@ async function peer(mode = 'normal') { const call = commandCallCodec.decode(message.payload); let packedResult = encrypted(protectedMessageCodec.encode({kind:'response',correlation:message.correlation,priority:message.priority,payload:blockCodec.encode({kind:'response',forId:0,packedResult:call.serializedArgs})})); if (mode === 'damage') packedResult[packedResult.length - 1] ^= 1; + if (mode === 'replay-then-valid') { + if (previous) emit({kind:'response',forId:block.id,packedResult:previous}); + previous = packedResult; + } if (mode === 'replay') { const saved = packedResult; packedResult = previous ?? saved; previous = saved; } if (mode === 'swap') { calls.push({id:block.id, packedResult}); @@ -58,8 +63,10 @@ async function peer(mode = 'normal') { } } return {key, socket, injectCall({id=90, signedId=90n, priority=0, signedPriority=0, kind='call'}={}) { - emit({kind:'call', id, priority, name:'L0Call', packedArgs:encrypted(protectedMessageCodec.encode({kind,correlation:signedId,priority:signedPriority,payload:commandCallCodec.encode({name:'callback',serializedArgs:string.encode('payload')})}))}); - }, connect: (pin, extra={}) => Client.connect('ws://test.invalid/', {expectedServerKey:pin ?? key.publicKey,session:null, + const packet = {kind:'call', id, priority, name:'L0Call', packedArgs:encrypted(protectedMessageCodec.encode({kind,correlation:signedId,priority:signedPriority,payload:commandCallCodec.encode({name:'callback',serializedArgs:string.encode('payload')})}))}; + emit(packet); + return () => emit(packet); + }, skipNonces(count) { nonce += BigInt(count); }, connect: (pin, extra={}) => Client.connect('ws://test.invalid/', {expectedServerKey:pin ?? key.publicKey,session:null, heartbeatMs:0,timeoutMs:1000,socketFactory:() => socket,...extra}), cleanup() {socket.close();session?.destroy();key.destroy();exchange.destroy(); if(peerError) throw peerError;} }; } const echo = command('echo',string,string); @@ -86,11 +93,12 @@ test('damaged encrypted response closes connection', async () => { await assert.rejects(client.call(echo,'again'),e=>e.code==='closed'); } finally {p.cleanup();} }); -test('replayed encrypted response closes connection', async () => { +test('replayed encrypted response is discarded without closing connection', async () => { const p=await peer('replay'); try { const client=await p.connect(); assert.equal(await client.call(echo,'first'),'first'); - await assert.rejects(client.call(echo,'second'),e=>e.code==='protocol'); await client.closed; + await assert.rejects(client.call(echo,'second',{timeoutMs:20}),e=>e.code==='timeout'); + assert.equal(p.socket.readyState,1); } finally {p.cleanup();} }); test('outer forId substitution fails closed without delivering swapped responses', async () => { @@ -130,3 +138,31 @@ for (const [name, metadata] of [ } finally {p.cleanup();} }); } + + +test('a replayed response does not consume the waiter for the valid reply', async () => { + const p = await peer('replay-then-valid'); + try { + const client = await p.connect(); + assert.equal(await client.call(echo, 'first'), 'first'); + assert.equal(await client.call(echo, 'second'), 'second'); + assert.equal(p.socket.readyState, 1); + } finally { p.cleanup(); } +}); + +test('duplicate and stale incoming pushes are discarded without closing', async () => { + const p = await peer(); + let calls = 0; + const local = new LocalInterface().on(command('callback', string, string), () => { calls++; return 'ok'; }); + try { + const client = await p.connect(undefined, {local}); + const replay = p.injectCall({id:0, signedId:0n, kind:'push'}); + replay(); + assert.equal(calls, 1); + p.skipNonces(4096); + p.injectCall({id:0, signedId:0n, kind:'push'}); + replay(); + assert.equal(calls, 2); + assert.equal(await client.call(echo, 'still alive'), 'still alive'); + } finally { p.cleanup(); } +});