From ea42bc6d20e59f03d511db4f7cce19de81f5a867 Mon Sep 17 00:00:00 2001 From: Lillem4n Date: Sat, 5 Sep 2026 21:00:52 +0200 Subject: [PATCH] Refuse a single unreadable PDU instead of the whole link (#79) * Regression tests: a PDU the codec cannot read costs that PDU, not the link * Refuse a single unreadable PDU instead of the whole link * smscsim interop asserts the first attempt gets every DLR * Record the smscsim sequence number defect as fixed * One framing rule and one response-command lookup, per the architecture review * Apply the stability review's nits: honest sessionError docs and one link-survives assertion --- AGENTS.md | 28 ++- README.md | 4 +- interop-tests/findings/01-smscsim.md | 5 + interop-tests/smscsim.test.ts | 76 +++---- src/outgoing-requests.ts | 5 + src/pdu-framer.ts | 7 +- src/pdu-transport.ts | 15 +- src/pdu.ts | 106 ++++++++-- src/session.ts | 34 ++- test/pdu.test.ts | 77 ++++++- test/session.test.ts | 295 ++++++++++++++++++++++----- 11 files changed, 511 insertions(+), 141 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index c7b6769..e28ba33 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -313,6 +313,20 @@ Grouped by what each one constrains. from `stat:` and been right. A transient state also carries `err:000`, since a message still on its way has not failed. +- **A refused PDU is answered from its header, and any 32-bit `sequence_number` is echoed as it + arrived.** Maintainer's call, 2026-09-05 via the interop plan. The header of a framed PDU always + parses, so it carries the answer SMPP 3.4 4.3 asks for, with the status 3.4 names for the part + that would not parse. Rejected: nacking a refused *response*, whose sequence number is one of + ours — the `generic_nack` would land in the peer's own numbering and nack a request of the peer's + we never saw, so a refused response is written back nothing and settles the request it names + instead. An unknown command id with the response bit set takes that branch too: a peer echoing a + sequence number of ours is answering something, and settling it reaches the undetermined outcome + `responseTimeout` would have reached anyway, sooner. Rejected: clamping a sequence number outside 4.7.1's 0x00000001–0x7FFFFFFF into range + before answering, which correlates with nothing at the peer — stacks write the field as a plain + uint32 (ukarim/smscsim signs every unprompted `deliver_sm` with a raw `rand.Int()`), so goal 3 + keeps that traffic and `PendingRequests.nextSeqNr()`, the only thing that invents one, is what + holds our own sends inside the spec. + - **`smsIdFormat` names a notation per place, and normalisation never reaches inside a `-` id.** An SMSC may answer `submit_sm_resp` in hex and write the receipt's `id:` in decimal, so one transform over both sides cannot make them equal. `submitResp` covers the `receipted_message_id` @@ -354,11 +368,15 @@ Grouped by what each one constrains. on arrival a fresh `minDelay` every cycle — one TCP connect and bind per second, forever. A drop after a healthy link still retries at `minDelay`. -- **A stream this library cannot read is a dead link, not a dead session.** Maintainer's call, - 2026-08-31: a framing or codec error tears the link down through `teardown()`, so the reconnect - loop retries it on a fresh socket with a fresh framer — which is what a desynced stream needs, and - the common cause. `sessionError` still carries every failure, so a peer that only ever sends - garbage is visible in the log rather than silent. +- **A stream this library cannot frame is a dead link; one PDU it cannot parse is not.** + Maintainer's call, 2026-08-31, narrowed 2026-09-05 via the interop plan: a `command_length` below + 16 or above `maxPduLength` leaves nothing that can say where the next PDU starts, so it tears the + link down through `teardown()` and the reconnect loop retries it on a fresh socket with a fresh + framer. Every other codec failure honoured `command_length`, so the stream is still in sync and + the next PDU starts where it says — tearing the link down there cost one peer half its receipts + and its MO to a reconnect loop (`interop-tests/findings/01-smscsim.md`), and left the peer waiting + for answers it was owed. `sessionError` carries every failure of either kind, never coalesced or + suppressed, so a peer that only ever sends garbage is visible in the log rather than silent. - **A deliberate shutdown drains; an unusable link and an abort do not.** `close()` and `unbind()` wait on the send window rather than the pending map — the map misses a segment still queued behind diff --git a/README.md b/README.md index 385ba6d..51700fb 100644 --- a/README.md +++ b/README.md @@ -78,7 +78,7 @@ Every one is optional. | `shutdownTimeout` | `5000` | How long `close()` and `unbind()` wait for the requests this end already sent and the messages the application has not answered. `0` waits forever for the requests, which end when the peer answers or `responseTimeout` expires — so setting both to `0` never ends. The messages fall back to `responseTimeout`, or to its default where that is `0` too, since nothing but the application ends that wait. | | `maxOutstanding` | `10` | Requests allowed on the wire at once; further sends queue. | | `smsIdFormat` | — | The notation the SMSC writes message ids in, per place it writes them: `{ receipt: 'decimal', submitResp: 'hex' }`. Only needed where the two disagree. | -| `reconnect` | on | Re-binds after a drop, an idle timeout, or a stream the library cannot read, backing off from `minDelay` 1 s to `maxDelay` 30 s and starting over at `minDelay` once a link has lasted `maxDelay`. `{ minDelay, maxDelay }` retunes it; `false` turns it off, so a drop ends the session. | +| `reconnect` | on | Re-binds after a drop, an idle timeout, or a stream the library cannot frame, backing off from `minDelay` 1 s to `maxDelay` 30 s and starting over at `minDelay` once a link has lasted `maxDelay`. `{ minDelay, maxDelay }` retunes it; `false` turns it off, so a drop ends the session. | | `log` | silent | Any object with `debug`, `error`, `info`, `verbose` and `warn` methods — see [Logging](#logging). | | `signal` | — | An `AbortSignal` that cancels connecting and tears the session down. | @@ -298,7 +298,7 @@ TypeScript users can import `SmppLog` to have the compiler check one. | `close` | The session is over, because nothing will bring the link back. Fires once, whether you closed it or the link failed for good. | | `disconnected` | The link dropped and the reconnect loop will retry it. Do not open a replacement client here — the session you hold comes back on its own, and `reconnected` says when. Fires again for each attempt that reconnects and then fails, so it is not one-to-one with `reconnected`. | | `reconnected` | The client re-bound after a drop. | -| `sessionError` | Something failed on a live session, including a hook or listener that threw or, if it was `async`, rejected. | +| `sessionError` | Something failed on a live session, including a hook or listener that threw or, if it was `async`, rejected. Fires for each PDU the codec refused, and the link carries on: a refused request is answered with the status SMPP names, and a refused response is answered with nothing and settles the request it named as `unanswered`. | | `data` | Raw bytes arrived on the socket. | | `incomingPdu` | A complete PDU arrived, as a buffer. | | `incomingPduObj` | The same PDU, parsed into an object. | diff --git a/interop-tests/findings/01-smscsim.md b/interop-tests/findings/01-smscsim.md index 37d043f..eea75a7 100644 --- a/interop-tests/findings/01-smscsim.md +++ b/interop-tests/findings/01-smscsim.md @@ -106,6 +106,11 @@ are silently lost and bounce the link. Against a spec-conforming peer (small inc numbers) it never fires, so it is plausibly why the suite's own dummy peers never caught it - which is the whole reason this experiment exists. +**Fixed** in PR #79: only a framing error tears the link down now, any 32-bit `sequence_number` is +read and echoed, and `smscsim.test.ts`'s retry crutch is gone. A rerun of `./interop-tests/run.py +smscsim` shows 54 frames, `deliver_sm: 6` answered by `deliver_sm_resp: 6`, `bind_transceiver: 5` +(no reconnects), `malformed: 0`, `expert errors: 0`, 8/8 tests passing on their first attempt. + ## Peer quirks - No PDU validation (documented): a bad `interface_version` or malformed PDU is never rejected. diff --git a/interop-tests/smscsim.test.ts b/interop-tests/smscsim.test.ts index 783f4c3..868f194 100644 --- a/interop-tests/smscsim.test.ts +++ b/interop-tests/smscsim.test.ts @@ -12,14 +12,8 @@ const PEER_WEB_PORT = Number(process.env.PEER_WEB_PORT ?? '12775'); const FAILING_PEER_HOST = process.env.FAILING_PEER_HOST ?? 'smscsim-failing'; const FAILING_PEER_PORT = Number(process.env.FAILING_PEER_PORT ?? '2775'); -// smscsim signs every deliver_sm it sends unprompted (a DLR, or an injected MO) with a -// `rand.Int()`-derived sequence_number, unconstrained to the SMPP 3.4 4.7.1 ceiling -// (0x7FFFFFFF) - about half land above it, and pdu.ts refuses the PDU outright, so the DLR or -// MO is silently lost (see findings/01-smscsim.md, "an out-of-range deliver_sm sequence -// number"). Retrying with a fresh send works around that peer+library interaction without -// hiding it: a scenario only fails here if it keeps missing well past what chance alone explains. -const DLR_RETRY_BUDGET_MS = 3000; -const DLR_MAX_ATTEMPTS = 20; +// smscsim keys its refusal on each submit_sm's own sequence number parity, so two sends minimum. +const PARITY_MAX_ATTEMPTS = 6; function delay(ms: number): Promise { return new Promise(resolve => { setTimeout(resolve, ms); }); @@ -38,22 +32,19 @@ async function waitFor(get: () => T | undefined, budget = 5000): Promise { - for (let attempt = 0; attempt < DLR_MAX_ATTEMPTS; attempt++) { - const sent = await session.sendSms({ dlr: true, from: '46701113311', message, to: '46709771337' }); +/** Sends once and waits for every segment of it to be matched by a `dlr` event. */ +async function sendAndAwaitDlrs(session: Session, dlrs: Dlr[], message: string): Promise { + const sent = await session.sendSms({ dlr: true, from: '46701113311', message, to: '46709771337' }); - assert.equal(sent.err, undefined); + assert.equal(sent.err, undefined); - const complete = await waitFor( - () => (sent.smsIds.every(id => dlrs.some(dlr => dlr.smsId === id)) ? true : undefined), - DLR_RETRY_BUDGET_MS, - ); + const complete = await waitFor(() => ( + sent.smsIds.every(id => dlrs.some(dlr => dlr.smsId === id)) ? true : undefined + )); - if (complete) return sent.smsIds; - } + assert.ok(complete, 'every segment of the first send should get a DLR'); - throw new Error(`no attempt got a DLR for every segment within ${String(DLR_MAX_ATTEMPTS)} tries`); + return sent.smsIds; } describe('smscsim - C1 bind, keepalive, unbind', () => { @@ -113,7 +104,7 @@ describe('smscsim - a single SMS', () => { session.on('dlr', dlr => { dlrs.push(dlr); }); - const smsIds = await sendUntilAllDlrsArrive(session, dlrs, 'hello world'); + const smsIds = await sendAndAwaitDlrs(session, dlrs, 'hello world'); assert.equal(smsIds.length, 1); @@ -142,7 +133,7 @@ describe('smscsim - multipart segments', () => { // 200 plain GSM chars: over the 160-char single-segment budget, under the 306-char // 2-segment one (153 septets each). - const smsIds = await sendUntilAllDlrsArrive(session, dlrs, 'a'.repeat(200)); + const smsIds = await sendAndAwaitDlrs(session, dlrs, 'a'.repeat(200)); assert.equal(smsIds.length, 2); }); @@ -160,7 +151,7 @@ describe('smscsim - multipart segments', () => { // 一 (2 bytes) + an emoji (a surrogate pair, 4 bytes) + 70 padding chars (2 bytes each): // 146 bytes, over the 140-byte single-segment budget, under the 268-byte 2-segment one. - const smsIds = await sendUntilAllDlrsArrive(session, dlrs, `一😀${'x'.repeat(70)}`); + const smsIds = await sendAndAwaitDlrs(session, dlrs, `一😀${'x'.repeat(70)}`); assert.equal(smsIds.length, 2); }); @@ -183,30 +174,23 @@ describe('smscsim - MO injection through the web UI', () => { session.on('sms', sms => { incoming.push(sms); }); - let sms: Sms | undefined; + const response = await fetch(`http://${PEER_HOST}:${String(PEER_WEB_PORT)}/`, { + body: new URLSearchParams({ + message: 'hello from the web UI', + recipient: '46709771337', + sender: '46701113311', + system_id: 'mo-inject', + }), + method: 'POST', + redirect: 'manual', + }); - for (let attempt = 0; attempt < DLR_MAX_ATTEMPTS && !sms; attempt++) { - const before = incoming.length; + assert.equal(response.status, 303); + assert.match(response.headers.get('location') ?? '', /message=/); - const response = await fetch(`http://${PEER_HOST}:${String(PEER_WEB_PORT)}/`, { - body: new URLSearchParams({ - message: 'hello from the web UI', - recipient: '46709771337', - sender: '46701113311', - system_id: 'mo-inject', - }), - method: 'POST', - redirect: 'manual', - }); + const sms = await waitFor(() => incoming[0]); - assert.equal(response.status, 303); - assert.match(response.headers.get('location') ?? '', /message=/); - - await waitFor(() => incoming[before], DLR_RETRY_BUDGET_MS); - sms = incoming[before]; - } - - assert.ok(sms, 'no MO message arrived across the attempts'); + assert.ok(sms, 'the injected MO should arrive on the first attempt'); assert.equal(sms.from, '46701113311'); assert.equal(sms.to, '46709771337'); assert.equal(sms.message, 'hello from the web UI'); @@ -230,7 +214,7 @@ describe('smscsim-failing - C12 refusals', () => { let refusedSeen = false; let acceptedConfirmed = false; - for (let attempt = 0; attempt < DLR_MAX_ATTEMPTS && !(refusedSeen && acceptedConfirmed); attempt++) { + for (let attempt = 0; attempt < PARITY_MAX_ATTEMPTS && !(refusedSeen && acceptedConfirmed); attempt++) { const before = dlrs.length; // Sequential: smscsim keys its refusal on each submit_sm's own sequence number parity. @@ -253,7 +237,7 @@ describe('smscsim-failing - C12 refusals', () => { assert.ok(smsId); - const matched = await waitFor(() => dlrs.slice(before).find(dlr => dlr.smsId === smsId), DLR_RETRY_BUDGET_MS); + const matched = await waitFor(() => dlrs.slice(before).find(dlr => dlr.smsId === smsId)); if (matched) { assert.equal(matched.statusMsg, 'UNDELIVERABLE'); diff --git a/src/outgoing-requests.ts b/src/outgoing-requests.ts index e3b4090..fe661e4 100644 --- a/src/outgoing-requests.ts +++ b/src/outgoing-requests.ts @@ -72,6 +72,11 @@ export class OutgoingRequests { return this.pending.deliver(pduObj); } + /** A response the codec refused settles its request instead of leaving it to time out. */ + settleRefused(seqNr: number, err: Error): void { + this.pending.settle(seqNr, { err }); + } + /** Sends a request and resolves with the peer's response. */ request(input: PduObjectInput, options: SendOptions): Promise> { // Ahead of the drain, so a misuse is named as one rather than blamed on the shutdown. diff --git a/src/pdu-framer.ts b/src/pdu-framer.ts index c4d49d0..8719f8f 100644 --- a/src/pdu-framer.ts +++ b/src/pdu-framer.ts @@ -1,5 +1,5 @@ import type { Result } from './result.ts'; -import { maxPduLength } from './pdu.ts'; +import { framingRefusal } from './pdu.ts'; /** * Cuts a byte stream into whole PDUs. @@ -31,10 +31,9 @@ export class PduFramer { while (this.length >= 16) { const cmdLength = this.join(16).readUInt32BE(0); + const unframable = framingRefusal(cmdLength); - if (cmdLength < 16 || cmdLength > maxPduLength) { - return { err: new Error(`Refusing a cmd_length of ${String(cmdLength)}`) }; - } + if (unframable) return { err: unframable }; if (this.length < cmdLength) break; diff --git a/src/pdu-transport.ts b/src/pdu-transport.ts index b1f74c0..bcf35f8 100644 --- a/src/pdu-transport.ts +++ b/src/pdu-transport.ts @@ -3,7 +3,7 @@ import type { SmppLog } from './log.ts'; import type { Socket } from 'node:net'; import type { VoidResult } from './result.ts'; import { PduFramer } from './pdu-framer.ts'; -import { pduToObj } from './pdu.ts'; +import { PduRefusedError, pduToObj } from './pdu.ts'; export type PduTransportOptions = { log: SmppLog; @@ -14,6 +14,8 @@ export type PduTransportOptions = { /** A complete PDU, before it is parsed. */ onFramed: (pdu: Buffer) => void; onPdu: (pduObj: PduObject) => void; + /** A framed PDU the codec could not read. The stream is still in sync, so the link is not lost. */ + onRefused: (refused: PduRefusedError) => void; /** Nothing further can be read off this stream, whatever the socket does next. */ onUnreadable: (err: Error) => void; }; @@ -79,6 +81,17 @@ export class PduTransport { const parsed = pduToObj(pdu); + if (parsed.err instanceof PduRefusedError) { + this.options.log.warn('transport - refusing a PDU it could not read', { + message: parsed.err.message, + reason: parsed.err.reason, + }); + this.options.onRefused(parsed.err); + + continue; + } + + // The framer applies framingRefusal() first, so only a caller that skips it lands here. if (parsed.err) { this.options.log.warn('transport - could not parse an incoming PDU', { message: parsed.err.message, diff --git a/src/pdu.ts b/src/pdu.ts index 5ff6339..cd05832 100644 --- a/src/pdu.ts +++ b/src/pdu.ts @@ -11,12 +11,24 @@ import { errorNameById, errors, isErrorName } from './defs/errors.ts'; import { paramNumber } from './defs/types.ts'; import { tlvDefault, tlvs, tlvsById } from './defs/tlvs.ts'; -/** Sequence numbers are a 31-bit field; 0x7fffffff is reserved. */ +/** The highest sequence number this library hands out; SMPP 3.4 4.7.1 reserves 0x7fffffff. */ export const maxSeqNr = 2147483646; +/** What the field holds. Read and echoed in full, because peers do write above the spec's range. */ +const maxWireSeqNr = 0xFFFFFFFF; + /** A hostile peer must not be able to make us allocate arbitrarily. */ export const maxPduLength = 1024 * 1024; +/** Why a command_length cannot frame a stream: past it nothing can say where the next PDU starts. */ +export function framingRefusal(cmdLength: number): Error | undefined { + if (cmdLength < 16 || cmdLength > maxPduLength) { + return new Error(`Refusing a cmd_length of ${String(cmdLength)}`); + } + + return undefined; +} + export type TlvInput = { /** Resolved from the record key; pass it for a tag the TLV table does not define. */ tagId?: number | undefined; @@ -46,6 +58,57 @@ export type PduObject = { tlvs: Record; }; +/** The 16 octets a framed PDU always has, whatever its body turns out to be. */ +export type PduHeader = { + cmdId: number; + cmdLength: number; + cmdName: CommandName | undefined; + cmdStatusId: number; + seqNr: number; +}; + +/** Which part of a PDU the codec could not read. */ +export type PduRefusalReason = 'body' | 'command' | 'tlvs'; + +/** A PDU refused with the stream still in sync, so only this one PDU is lost. */ +export class PduRefusedError extends Error { + readonly header: PduHeader; + readonly reason: PduRefusalReason; + + constructor(header: PduHeader, reason: PduRefusalReason, cause: Error) { + const named = header.cmdName ?? `command id ${String(header.cmdId)}`; + + super(`Refused ${named} with seqNr ${String(header.seqNr)}: ${cause.message}`, { cause }); + this.header = header; + this.name = 'PduRefusedError'; + this.reason = reason; + } +} + +// ESME_RINVTLVSTREAM is SMPP 5.0's name for 0xC0, which SMPP 3.4 spells ESME_RINVOPTPARSTREAM. +const refusalStatus = { + body: 'ESME_RINVCMDLEN', + command: 'ESME_RINVCMDID', + tlvs: 'ESME_RINVTLVSTREAM', +} as const satisfies Record; + +/** The response SMPP pairs with a request command, where it has one. */ +function respNameFor(cmdName: CommandName | undefined): CommandName | undefined { + if (cmdName === undefined) return undefined; + + const respName = `${cmdName}_resp`; + + return isCommandName(respName) ? respName : undefined; +} + +/** SMPP 3.4 4.3: a PDU whose command has no response of its own is refused with generic_nack. */ +export function refusalAnswer(refused: PduRefusedError): { cmdName: CommandName; cmdStatus: ErrorName } { + return { + cmdName: respNameFor(refused.header.cmdName) ?? 'generic_nack', + cmdStatus: refusalStatus[refused.reason], + }; +} + const respBit = 0x80000000; export function isResp(pduObj: Pick): boolean { @@ -209,7 +272,7 @@ function buildPdu( return { err: new Error(`Invalid cmdStatus: ${JSON.stringify(cmdStatus)}`) }; } - if (!Number.isInteger(seqNr) || seqNr < 0 || seqNr > maxSeqNr) { + if (!Number.isInteger(seqNr) || seqNr < 0 || seqNr > maxWireSeqNr) { return { err: new Error(`Invalid seqNr: ${JSON.stringify(seqNr)}`) }; } @@ -295,20 +358,24 @@ function readParams( return { offset, params }; } -function parseOnce(pdu: Buffer, trailingNull: boolean): Result<{ aligned: boolean; pduObj: PduObject }> { - const cmdLength = pdu.readUInt32BE(0); +function headerOf(pdu: Buffer): PduHeader { const cmdId = pdu.readUInt32BE(4); - const cmdName = commandNameById(cmdId); + + return { + cmdId, + cmdLength: pdu.readUInt32BE(0), + cmdName: commandNameById(cmdId), + cmdStatusId: pdu.readUInt32BE(8), + seqNr: pdu.readUInt32BE(12), + }; +} + +function parseOnce(pdu: Buffer, trailingNull: boolean): Result<{ aligned: boolean; pduObj: PduObject }> { + const header = headerOf(pdu); + const { cmdId, cmdLength, cmdName, cmdStatusId, seqNr } = header; if (!cmdName) { - return { err: new Error(`Unknown PDU command id: ${String(cmdId)}`) }; - } - - const cmdStatusId = pdu.readUInt32BE(8); - const seqNr = pdu.readUInt32BE(12); - - if (seqNr > maxSeqNr) { - return { err: new Error(`Invalid seqNr, exceeds ${String(maxSeqNr)}: ${String(seqNr)}`) }; + return { err: new PduRefusedError(header, 'command', new Error('Unknown PDU command id')) }; } // SMPP 3.4 4.4.2 and friends: a response with a non-zero status carries no body at all. @@ -316,11 +383,11 @@ function parseOnce(pdu: Buffer, trailingNull: boolean): Result<{ aligned: boolea ? { offset: 16, params: {} } : readParams(cmdName, pdu, trailingNull); - if (read.err) return { err: read.err }; + if (read.err) return { err: new PduRefusedError(header, 'body', read.err) }; const parsed = parseTlvs(pdu, read.offset, cmdLength); - if (parsed.err) return { err: parsed.err }; + if (parsed.err) return { err: new PduRefusedError(header, 'tlvs', parsed.err) }; const params = read.params; const message = params.short_message; @@ -352,10 +419,9 @@ function checkFraming(pdu: Buffer): VoidResult { } const cmdLength = pdu.readUInt32BE(0); + const unframable = framingRefusal(cmdLength); - if (cmdLength < 16 || cmdLength > maxPduLength) { - return { err: new Error(`Refusing a cmd_length of ${String(cmdLength)}`) }; - } + if (unframable) return { err: unframable }; if (cmdLength > pdu.length) { return { err: new Error(`cmd_length ${String(cmdLength)} exceeds the ${String(pdu.length)} octets given`) }; @@ -417,9 +483,9 @@ export function pduReturn( return parsed.err ? { err: parsed.err } : pduReturn(parsed.pduObj, status, params, tlvs); } - const respName = `${pdu.cmdName}_resp`; + const respName = respNameFor(pdu.cmdName); - if (!isCommandName(respName)) { + if (!respName) { return { err: new Error(`"${pdu.cmdName}" has no response command`) }; } diff --git a/src/session.ts b/src/session.ts index 4385722..203b281 100644 --- a/src/session.ts +++ b/src/session.ts @@ -1,7 +1,7 @@ import type { ErrorName } from './defs/errors.ts'; import type { MessageDlr } from './dlr-merger.ts'; import type { ParamValue } from './defs/types.ts'; -import type { PduObject, PduObjectInput, TlvInput } from './pdu.ts'; +import type { PduObject, PduObjectInput, PduRefusedError, TlvInput } from './pdu.ts'; import type { BindType, CloseOptions, ReconnectOptions, SendOptions, SessionEvents, SessionOptions } from './session-options.ts'; import type { Result, VoidResult } from './result.ts'; import type { SendSmsOptions, SendSmsResult } from './send-sms.ts'; @@ -18,7 +18,7 @@ import { leftOf } from './idle-waiters.ts'; import { errorFrom } from './error-from.ts'; import { optionalParamsMinVersion } from './defs/constants.ts'; import { bindCarries, bindCommands, defaultSystemId, defaults } from './session-options.ts'; -import { isResp, pduReturn } from './pdu.ts'; +import { isResp, objToPdu, pduReturn, refusalAnswer } from './pdu.ts'; import { guardedLog } from './log.ts'; import { submitSms, unsent } from './send-sms.ts'; import { ConcatReference } from './udh.ts'; @@ -173,20 +173,25 @@ export class Session extends EventEmitter { params: Record = {}, tlvs?: Record, ): Promise { - const built = pduReturn(pdu, status, params, tlvs); + return Promise.resolve( + this.answer(pduReturn(pdu, status, params, tlvs), pdu.cmdName, pdu.seqNr), + ); + } + + private answer(built: Result<{ buffer: Buffer }>, cmdName: string, seqNr: number): VoidResult { const sent = built.err ? { err: built.err } : this.transport.write(built.buffer); // A peer that unbinds and drops the link takes our response with it; that is not a failure. if (sent.err && !this.closed) { this.log.warn('session - could not answer a request', { - cmdName: pdu.cmdName, + cmdName, message: sent.err.message, - seqNr: pdu.seqNr, + seqNr, }); this.emit('sessionError', sent.err); } - return Promise.resolve(sent); + return sent; } async sendSms(sms: SendSmsOptions, options: SendOptions = {}): Promise { @@ -244,6 +249,7 @@ export class Session extends EventEmitter { onError: err => { this.emit('sessionError', err); }, onFramed: pdu => { this.emit('incomingPdu', pdu); }, onPdu: pduObj => { this.dispatch(pduObj); }, + onRefused: refused => { this.refuse(refused); }, onUnreadable: err => { this.emit('sessionError', err); this.teardown(); @@ -388,6 +394,22 @@ export class Session extends EventEmitter { }); } + /** A PDU the codec refused. Its header parsed, so the peer gets an answer and the link stays. */ + private refuse(refused: PduRefusedError): void { + const { cmdId, cmdName, seqNr } = refused.header; + + this.emit('sessionError', refused); + + // A response carries a sequence number of ours, so writing one back lands in the peer's space. + if (isResp(refused.header)) { + this.outgoing.settleRefused(seqNr, refused); + + return; + } + + this.answer(objToPdu({ ...refusalAnswer(refused), seqNr }), cmdName ?? String(cmdId), seqNr); + } + private resetTimers(): void { if (this.closed) return; diff --git a/test/pdu.test.ts b/test/pdu.test.ts index 8134712..8891c73 100644 --- a/test/pdu.test.ts +++ b/test/pdu.test.ts @@ -1,6 +1,6 @@ import assert from 'node:assert/strict'; import test, { describe } from 'node:test'; -import { isCommand, isResp, objToPdu, pduReturn, pduToObj } from '../src/pdu.ts'; +import { PduRefusedError, isCommand, isResp, objToPdu, pduReturn, pduToObj, refusalAnswer } from '../src/pdu.ts'; function encode(...args: Parameters): Buffer { const { buffer, err } = objToPdu(...args); @@ -44,8 +44,11 @@ describe('header', () => { assert.ok(isResp(pduObj)); }); - test('rejects an unknown command and an out-of-range sequence number', () => { - assert.ok(objToPdu({ cmdName: 'submit_sm', seqNr: 2147483647 }).err instanceof Error); + // SMPP 3.4 4.7.1 stops the range at 0x7FFFFFFF, but peers write the field as a plain uint32. + test('carries any 32-bit sequence number, and refuses one the field cannot hold', () => { + assert.equal(decode(encode({ cmdName: 'enquire_link', seqNr: 0x80000001 })).seqNr, 0x80000001); + assert.equal(decode(encode({ cmdName: 'enquire_link', seqNr: 0xFFFFFFFF })).seqNr, 0xFFFFFFFF); + assert.ok(objToPdu({ cmdName: 'submit_sm', seqNr: 0x100000000 }).err instanceof Error); }); }); @@ -437,6 +440,74 @@ describe('malformed input', () => { test('refuses an absurd command length instead of allocating for it', () => { assert.ok(pduToObj(Buffer.from('ffffffff0000000400000000000000ff', 'hex')).err instanceof Error); }); + + test('reads the header of a PDU it cannot parse, and names which part it choked on', () => { + const unknown = encode({ cmdName: 'enquire_link', seqNr: 9 }); + + unknown.writeUInt32BE(0x00010001, 4); + + const refusedCommand = pduToObj(unknown).err; + + assert.ok(refusedCommand instanceof PduRefusedError); + assert.equal(refusedCommand.reason, 'command'); + assert.equal(refusedCommand.header.cmdName, undefined); + assert.equal(refusedCommand.header.cmdId, 0x00010001); + assert.equal(refusedCommand.header.seqNr, 9); + assert.deepEqual(refusalAnswer(refusedCommand), { + cmdName: 'generic_nack', + cmdStatus: 'ESME_RINVCMDID', + }); + + const whole = encode({ + cmdName: 'deliver_sm', + params: { destination_addr: '46709771337', short_message: 'hello', source_addr: '46701113311' }, + seqNr: 10, + }); + // sm_length still declares five octets of short_message; three of them never arrived. + const short = whole.subarray(0, whole.length - 3); + + short.writeUInt32BE(short.length, 0); + + const refusedBody = pduToObj(short).err; + + assert.ok(refusedBody instanceof PduRefusedError); + assert.equal(refusedBody.reason, 'body'); + assert.equal(refusedBody.header.cmdName, 'deliver_sm'); + assert.deepEqual(refusalAnswer(refusedBody), { + cmdName: 'deliver_sm_resp', + cmdStatus: 'ESME_RINVCMDLEN', + }); + + // message_state, declaring four octets of value with one of them on the wire. + const truncatedTlv = Buffer.concat([ + encode({ cmdName: 'deliver_sm', params: { short_message: 'hello' }, seqNr: 11 }), + Buffer.from('0427000401', 'hex'), + ]); + + truncatedTlv.writeUInt32BE(truncatedTlv.length, 0); + + const refusedTlvs = pduToObj(truncatedTlv).err; + + assert.ok(refusedTlvs instanceof PduRefusedError); + assert.equal(refusedTlvs.reason, 'tlvs'); + assert.deepEqual(refusalAnswer(refusedTlvs), { + cmdName: 'deliver_sm_resp', + cmdStatus: 'ESME_RINVTLVSTREAM', + }); + }); + + test('answers a command with no response of its own with generic_nack', () => { + const outbind = encode({ cmdName: 'outbind', params: { system_id: 'smsc' }, seqNr: 12 }); + // Both C-Octet Strings lose their terminator, so system_id runs off the end of the PDU. + const short = outbind.subarray(0, outbind.length - 2); + + short.writeUInt32BE(short.length, 0); + + const refused = pduToObj(short).err; + + assert.ok(refused instanceof PduRefusedError); + assert.equal(refusalAnswer(refused).cmdName, 'generic_nack'); + }); }); describe('isCommand()', () => { diff --git a/test/session.test.ts b/test/session.test.ts index 034df1c..f9c9813 100644 --- a/test/session.test.ts +++ b/test/session.test.ts @@ -71,33 +71,66 @@ function raceWithin(ms: number, promise: Promise): Promise { return Promise.race([promise, delay(ms).then((): false => false)]); } -type Peer = { close: () => Promise; port: number }; +/** A socket read as a queue of parsed PDUs; `handled` takes the ones the peer answers itself. */ +function pduQueue( + sock: net.Socket, + handled: (pduObj: PduObject) => boolean = () => false, +): () => Promise { + const framer = new PduFramer(); + const queue: PduObject[] = []; + const waiting: ((pduObj: PduObject) => void)[] = []; -/** Answers binds and nothing else, which is what a link the peer has stopped serving looks like. */ -async function bindOnlyPeer(t: TestContext, options: { dropOn?: string } = {}): Promise { + sock.on('data', chunk => { + framer.push(chunk); + + for (const pdu of framer.next().pdus ?? []) { + const { pduObj } = pduToObj(pdu); + + if (!pduObj || handled(pduObj)) continue; + + const next = waiting.shift(); + + if (next) next(pduObj); + else queue.push(pduObj); + } + }); + + return () => { + const queued = queue.shift(); + + return queued ? Promise.resolve(queued) : once(resolve => waiting.push(resolve)); + }; +} + +type Peer = { + close: () => Promise; + /** The next PDU the ESME sent that the peer did not answer itself. */ + next: () => Promise; + port: number; + /** Raw octets, so a test can say what objToPdu would not build. */ + writeRaw: (bytes: Buffer) => void; +}; + +/** An SMSC answering binds and nothing else, which is what the tests write the other answers for. */ +async function smscPeer(t: TestContext, options: { dropOn?: string } = {}): Promise { const sockets: net.Socket[] = []; + let queued: (() => Promise) | undefined; const listener = net.createServer(sock => { - const framer = new PduFramer(); - sockets.push(sock); - sock.on('data', chunk => { - framer.push(chunk); + queued = pduQueue(sock, pduObj => { + if (pduObj.cmdName === options.dropOn) { + sock.destroy(); - for (const pdu of framer.next().pdus ?? []) { - const { pduObj } = pduToObj(pdu); - - if (pduObj && pduObj.cmdName === options.dropOn) { - sock.destroy(); - - return; - } - - if (!pduObj || !bindCommands.includes(pduObj.cmdName)) continue; - - const { buffer } = pduReturn(pduObj, 'ESME_ROK', { system_id: 'silent' }); - - if (buffer) sock.write(buffer); + return true; } + + if (!bindCommands.includes(pduObj.cmdName)) return false; + + const { buffer } = pduReturn(pduObj, 'ESME_ROK', { system_id: 'silent' }); + + if (buffer) sock.write(buffer); + + return true; }); }); @@ -112,7 +145,13 @@ async function bindOnlyPeer(t: TestContext, options: { dropOn?: string } = {}): await new Promise(resolve => { listener.close(() => { resolve(); }); }); }, + next: () => { + assert.ok(queued, 'nothing has connected to the peer yet'); + + return queued(); + }, port: typeof address === 'object' && address !== null ? address.port : 0, + writeRaw: (bytes: Buffer) => { sockets[sockets.length - 1]?.write(bytes); }, }; t.after(() => peer.close()); @@ -142,39 +181,14 @@ type RawPeer = { /** A peer driven PDU by PDU, which is the only way to say things the client never says. */ function rawPeer(t: TestContext, port: number): RawPeer { - const framer = new PduFramer(); - const queue: PduObject[] = []; - const waiting: ((pduObj: PduObject) => void)[] = []; const sock = net.connect({ port }); - - sock.on('data', chunk => { - framer.push(chunk); - - const { pdus } = framer.next(); - - for (const pdu of pdus ?? []) { - const { pduObj } = pduToObj(pdu); - - if (!pduObj) continue; - - const next = waiting.shift(); - - if (next) next(pduObj); - else queue.push(pduObj); - } - }); + const next = pduQueue(sock); t.after(() => { sock.destroy(); }); return { close: () => { sock.destroy(); }, - next: () => { - const queued = queue.shift(); - - return queued - ? Promise.resolve(queued) - : once(resolve => waiting.push(resolve)); - }, + next, write: input => { const { buffer } = objToPdu(input); @@ -218,7 +232,7 @@ describe('bind', () => { // Plenty of SMSCs drop the connection on unbind instead of answering it. test('takes a close that follows our unbind as a clean unbind', async t => { - const peer = await bindOnlyPeer(t, { dropOn: 'unbind' }); + const peer = await smscPeer(t, { dropOn: 'unbind' }); const { session } = await client({ port: peer.port, responseTimeout: 2000 }); assert.ok(session); @@ -227,7 +241,7 @@ describe('bind', () => { }); test('still reports a close that lands on another in-flight request', async t => { - const peer = await bindOnlyPeer(t, { dropOn: 'enquire_link' }); + const peer = await smscPeer(t, { dropOn: 'enquire_link' }); const { session } = await client({ port: peer.port, responseTimeout: 2000 }); assert.ok(session); @@ -240,7 +254,7 @@ describe('bind', () => { }); test('reports an unbind the peer left unanswered on a link that stays up', async t => { - const peer = await bindOnlyPeer(t); + const peer = await smscPeer(t); const { session } = await client({ port: peer.port, responseTimeout: 150 }); assert.ok(session); @@ -413,7 +427,7 @@ describe('bind', () => { // The spec: an absent sc_interface_version means the SMSC supports no optional parameters. test('takes an SMSC that declares no version as older than 3.4', async t => { - const peer = await bindOnlyPeer(t); + const peer = await smscPeer(t); const { session } = await client({ port: peer.port }); assert.ok(session); @@ -1189,6 +1203,179 @@ describe('robustness', () => { }); }); +describe('a PDU the codec cannot read', () => { + const receipt = { + destination_addr: '46709771337', + esm_class: consts.ESM_CLASS.MC_DELIVERY_RECEIPT, + short_message: 'id:0199e0e9-4a3e-7c62-9a4b-1f0c5d7e8a21 stat:DELIVRD err:000 text:', + source_addr: '46701113311', + }; + + /** The octets objToPdu built, wearing a sequence number it would refuse to write itself. */ + function withSeqNr(input: PduObjectInput, seqNr: number): Buffer { + const { buffer } = objToPdu(input); + + assert.ok(buffer); + buffer.writeUInt32BE(seqNr, 12); + + return buffer; + } + + /** command_length honoured, so the stream stays in sync, with the declared body cut short. */ + function shortened(input: PduObjectInput, octets: number): Buffer { + const { buffer } = objToPdu(input); + + assert.ok(buffer); + + const cut = buffer.subarray(0, buffer.length - octets); + + cut.writeUInt32BE(cut.length, 0); + + return cut; + } + + /** The same, with a message_state TLV declaring four octets of value and carrying one. */ + function truncatedTlv(input: PduObjectInput): Buffer { + const { buffer } = objToPdu(input); + + assert.ok(buffer); + + const appended = Buffer.concat([buffer, Buffer.from('0427000401', 'hex')]); + + appended.writeUInt32BE(appended.length, 0); + + return appended; + } + + /** The peer's next PDU within a budget: a dropped link must fail the test, not hang it. */ + async function answerTo(peer: Peer): Promise { + const pduObj = await raceWithin(2000, peer.next()); + + assert.ok(pduObj, 'the peer was never answered'); + + return pduObj; + } + + async function bound(t: TestContext, options: Parameters[0] = {}) { + const peer = await smscPeer(t); + const { session } = await client({ port: peer.port, ...options }); + + assert.ok(session); + closeAfter(t, session); + + return { peer, session }; + } + + // ukarim/smscsim signs every deliver_sm it sends unprompted with a raw uint32, so about half + // land above SMPP 3.4 4.7.1's ceiling; refusing them cost the link (findings/01-smscsim.md). + test('answers a deliver_sm whose sequence number is above the spec range, and keeps the link', async t => { + const { peer, session } = await bound(t); + const reported = once(resolve => { session.on('dlr', resolve); }); + + peer.writeRaw(withSeqNr({ cmdName: 'deliver_sm', params: receipt, seqNr: 1 }, 0x80000001)); + + const answered = await answerTo(peer); + + assert.equal(answered.cmdName, 'deliver_sm_resp'); + assert.equal(answered.cmdStatus, 'ESME_ROK'); + assert.equal(answered.seqNr, 0x80000001); + + const dlr = await raceWithin(2000, reported); + + assert.ok(dlr, 'the receipt is a report, not a reason to drop the link'); + assert.equal(dlr.statusMsg, 'DELIVERED'); + + peer.writeRaw(withSeqNr({ cmdName: 'enquire_link', seqNr: 1 }, 0xFFFFFFFF)); + + const pinged = await answerTo(peer); + + assert.equal(pinged.cmdName, 'enquire_link_resp'); + assert.equal(pinged.seqNr, 0xFFFFFFFF); + }); + + test('answers an unknown command id with generic_nack ESME_RINVCMDID', async t => { + const { peer, session } = await bound(t); + const failed = once(resolve => { session.on('sessionError', resolve); }); + const vendorSpecific = withSeqNr({ cmdName: 'enquire_link', seqNr: 1 }, 9); + + vendorSpecific.writeUInt32BE(0x00010001, 4); + peer.writeRaw(vendorSpecific); + + const answered = await answerTo(peer); + + assert.equal(answered.cmdName, 'generic_nack'); + assert.equal(answered.cmdStatus, 'ESME_RINVCMDID'); + assert.equal(answered.seqNr, 9); + assert.ok((await raceWithin(2000, failed)) instanceof Error, 'one sessionError per refused PDU'); + }); + + test('answers a deliver_sm with a truncated TLV stream with ESME_RINVTLVSTREAM', async t => { + const { peer, session } = await bound(t); + let reports = 0; + + session.on('dlr', () => { reports++; }); + peer.writeRaw(truncatedTlv({ cmdName: 'deliver_sm', params: receipt, seqNr: 5 })); + + const answered = await answerTo(peer); + + assert.equal(answered.cmdName, 'deliver_sm_resp'); + assert.equal(answered.cmdStatus, 'ESME_RINVTLVSTREAM'); + assert.equal(answered.seqNr, 5); + assert.equal(reports, 0, 'a refused PDU is not a report'); + + // The regression this fixes: the link, and the stream's sync, outlive the refused PDU. + peer.writeRaw(withSeqNr({ cmdName: 'enquire_link', seqNr: 1 }, 78)); + assert.equal((await answerTo(peer)).cmdName, 'enquire_link_resp'); + }); + + test('answers a deliver_sm whose body is shorter than it declares with ESME_RINVCMDLEN', async t => { + const { peer } = await bound(t); + + peer.writeRaw(shortened({ cmdName: 'deliver_sm', params: receipt, seqNr: 6 }, 3)); + + const answered = await answerTo(peer); + + assert.equal(answered.cmdName, 'deliver_sm_resp'); + assert.equal(answered.cmdStatus, 'ESME_RINVCMDLEN'); + assert.equal(answered.seqNr, 6); + }); + + test('settles the request a response it could not read was answering', async t => { + const { peer, session } = await bound(t, { responseTimeout: 60000 }); + const sending = session.sendSms({ from: '46701113311', message: 'hi', to: '46709771337' }); + const submitted = await answerTo(peer); + + assert.equal(submitted.cmdName, 'submit_sm'); + peer.writeRaw(truncatedTlv({ + cmdName: 'submit_sm_resp', + params: { message_id: '0199e0ea-1c88-7a41-b6d2-4e7f0a9c3b15' }, + seqNr: submitted.seqNr, + })); + + const sent = await raceWithin(2000, sending); + + assert.ok(sent, 'the request settles on the refusal rather than on the response timeout'); + assert.ok(sent.err instanceof Error); + assert.equal(sent.unanswered, 1); + + // Nothing goes back: a response carries a sequence number of ours, not one of the peer's. + peer.writeRaw(withSeqNr({ cmdName: 'enquire_link', seqNr: 1 }, 77)); + assert.equal((await answerTo(peer)).cmdName, 'enquire_link_resp'); + }); + + test('tears the link down when the stream itself cannot be framed', async t => { + const { peer, session } = await bound(t, { reconnect: false }); + const failed = once(resolve => { session.on('sessionError', resolve); }); + const closed = once(resolve => { session.on('close', () => { resolve(true); }); }); + + // A command_length below the 16 octet header leaves nothing that can find the next PDU. + peer.writeRaw(Buffer.from('00000004000000150000000000000001', 'hex')); + + assert.ok((await raceWithin(2000, failed)) instanceof Error); + assert.equal(await raceWithin(2000, closed), true); + }); +}); + describe('application hooks that throw or reject', () => { test('turns a throwing authenticate into a session error', async t => { const smpp = await startServer(t, { @@ -1493,7 +1680,7 @@ describe('application hooks that throw or reject', () => { describe('link timers', () => { test('closes a client link the peer has stopped answering', async t => { - const peer = await bindOnlyPeer(t); + const peer = await smscPeer(t); const { err, session } = await client({ enquireLinkInterval: 50, port: peer.port, @@ -1513,7 +1700,7 @@ describe('link timers', () => { }); test('reconnects a link that timed out', async t => { - const peer = await bindOnlyPeer(t); + const peer = await smscPeer(t); const { session } = await client({ enquireLinkInterval: 40, port: peer.port,