From 884afdb87b35498faeee93685b221e2cae108587 Mon Sep 17 00:00:00 2001 From: Lilleman auf Larv Date: Thu, 27 Aug 2026 10:55:34 +0200 Subject: [PATCH] Stop a thrown listener escaping, bound what a peer can pin, and drop the body from failure responses --- AGENTS.md | 4 ++ README.md | 20 +++++--- src/client.ts | 23 +++++++-- src/defs/types.ts | 30 ++++++++++-- src/dlr-merger.ts | 26 +++++++++- src/index.ts | 3 +- src/pdu.ts | 43 +++++++++++++---- src/reassembly.ts | 11 ++++- src/send-sms.ts | 1 - src/server.ts | 84 +++++++++++++++++++++++---------- src/session-options.ts | 52 ++++++++++++++++++++ src/session.ts | 57 ++++++++++++---------- test/pdu.test.ts | 14 ++++-- test/session-extras.test.ts | 35 +++++++++++++- test/session.test.ts | 94 +++++++++++++++++++++++++++++++++++-- test/types.test.ts | 8 ++++ todo.md | 2 +- 17 files changed, 420 insertions(+), 87 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index fafafe9..f6ce0a9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -156,6 +156,10 @@ exactly 140. ## Decisions +- **A close arriving after our own `unbind` is a clean unbind, not an error.** Maintainer's call, + 2026-08-26: most SMSCs drop the socket instead of answering, so the documented shutdown would + otherwise always report a failure. It does mask a socket that died mid-unbind for an unrelated + reason, which is accepted — the peer sees the same TCP close either way. - **The published surface is frozen at what `src/index.ts` exports today.** `Session` is exported and publicly constructible, which is why `SessionOptions` and `ReconnectOptions` are public too — that is correct, not a leak, and it has been raised twice. The collaborators `session.ts` delegates to diff --git a/README.md b/README.md index 499a232..1286ddd 100644 --- a/README.md +++ b/README.md @@ -82,7 +82,7 @@ Every one is optional. | `tls` | `false` | `true` for defaults, or a `tls.ConnectionOptions` object for a private CA or a client certificate. | | `enquireLinkInterval` | `20000` | How often to send `enquire_link` on a quiet link. | | `idleTimeout` | `2 × enquireLinkInterval` | Give up on a link the peer has stopped answering; with `reconnect` set, it re-binds. | -| `responseTimeout` | `30000` | How long to wait for a response before giving up on it. | +| `responseTimeout` | `30000` | How long to wait for a response before giving up on it; `0` waits forever. | | `maxOutstanding` | `10` | Requests allowed on the wire at once; further sends queue. | | `reconnect` | off | `{ minDelay, maxDelay }` to re-bind automatically after a drop or an idle timeout, with exponential backoff. | | `log` | silent | A `@larvit/log` instance. | @@ -111,9 +111,10 @@ const { err, pduObjs, smsIds } = await session.sendSms({ from, message, to }); ``` `err` is set when the SMSC refuses a segment, and it names the status it refused with. Because every -segment goes on the wire together, `smsIds` then holds the ids of the segments the SMSC did accept — -retry only what is missing from it. A message needing more than 255 segments is refused before -anything is sent, since the concatenation header numbers segments in a single octet. +segment goes on the wire together, `pduObjs` and `smsIds` then hold what the SMSC did accept — enough +to reconcile against a later receipt, not enough to resend the rest, so treat a partial failure as a +failed message. A message needing more than 255 segments is refused before anything is sent, since +the concatenation header numbers segments in a single octet. ## Server @@ -178,6 +179,7 @@ await smpp.close(); // stop listening and close every live session | `tls` | `false` | A `tls.TlsOptions` object with your certificate and key. | | `idleTimeout` | `40000` | Drop a peer that has been silent this long. | | `maxReassembly` | `1000` | Incomplete multipart messages held per session. | +| `maxOctets` | `67108864` | Bytes of incomplete multipart messages held per session. | | `reassemblyTimeout` | `300000` | How long a late segment can still join an incomplete message. | | `responseTimeout`, `maxOutstanding`, `log`, `signal` | as for the client | | @@ -205,7 +207,7 @@ is exactly what this library promises not to do. | --- | --- | | `sms` | An SMS arrives, reassembled if it was multipart. Carries `sendResp()`, `sendDlr()` and the `smsId` it was answered with. | | `dlr` | A delivery report arrives, one per segment. | -| `messageDlr` | Every segment of a multipart message sent with `dlr: true` has been reported on. | +| `messageDlr` | Every segment of a multipart message sent with `dlr: true` has been reported on, carrying the worst status of the segments. | | `close` | The connection closed. | | `reconnected` | The client re-bound after a drop (only with `reconnect` configured). | | `sessionError` | Something failed on a live session, including a hook or listener that threw. | @@ -255,8 +257,9 @@ The spec tables are exported both individually (`cmds`, `consts`, `encodings`, ` resolving to a result object with an optional `err`. Nothing rejects. - **`server()` resolves once, when it is listening**, and gives you a handle with `close()`, `port` and a `session` event. It no longer calls your callback once per incoming connection. -- **The id a message is answered with goes to `sendResp({ smsId })`**, and `sms.smsId` is read-only. - Assigning it no longer works, so set the id where the response is sent rather than before it. +- **The id a message is answered with goes to `sendResp({ smsId })`**, and `sms.smsId` is read-only: + it reports what the response actually carried. Delete any `sms.smsId = …` line — assigning to it + throws a `TypeError`, since modules are always strict mode — and pass the id to `sendResp()`. - **`checkuserpass` is now `authenticate`**, takes `{ password, session, systemId, systemType }` and returns `false` or `{ userData }`. - **Renamed options:** `enqLinkTiming` → `enquireLinkInterval`, server `timeout` → `idleTimeout`. @@ -299,6 +302,9 @@ have worked around any of these, remove the workaround: - Binds now declare `interface_version` 0x34. 0.4.0 declared 0x00, which tells the SMSC the ESME speaks SMPP 3.3 or earlier — and a spec-following SMSC then withholds every optional parameter, including the TLVs delivery receipts are carried in. +- A response reporting a failure now carries no body, which is what the spec defines and what other + implementations send. 0.4.0 filled the body with empty defaults, so a refused `submit_sm_resp` went + out with an empty `message_id` a caller could mistake for a real one. - `submit_multi` was missing its `sm_length` field, so its `short_message` never round-tripped. The corrected framing is cross-checked against [node-smpp](https://github.com/farhadi/node-smpp), an diff --git a/src/client.ts b/src/client.ts index f418726..7f49d94 100644 --- a/src/client.ts +++ b/src/client.ts @@ -3,6 +3,7 @@ import type { LogInt } from '@larvit/log'; import type { Result, VoidResult } from './result.ts'; import type { Socket } from 'node:net'; import { Session } from './session.ts'; +import { checkSessionOptions } from './session-options.ts'; import { connect as netConnect } from 'node:net'; import { connect as tlsConnect } from 'node:tls'; import { defaultInterfaceVersion } from './defs/constants.ts'; @@ -152,9 +153,15 @@ function createSession(options: ClientOptions, log: LogInt, sock: Socket): Sessi }); } -/** Connects to an SMSC and binds. */ -export async function client(options: ClientOptions = {}): Promise> { - const log = options.log ?? silentLog; +async function connect(options: ClientOptions, log: LogInt): Promise> { + const checked = checkSessionOptions(options); + + if (checked.err) { + log.warn('client - option out of range', { message: checked.err.message }); + + return { err: checked.err }; + } + const opened = await openSocket(options); if (opened.err) { @@ -167,6 +174,16 @@ export async function client(options: ClientOptions = {}): Promise> { + const log = options.log ?? silentLog; + const opened = await connect(options, log); + + if (opened.err) return { err: opened.err }; + const session = createSession(options, log, opened.sock); const signal = options.signal; diff --git a/src/defs/types.ts b/src/defs/types.ts index 4617da4..3b6f4eb 100644 --- a/src/defs/types.ts +++ b/src/defs/types.ts @@ -215,13 +215,19 @@ export const string: WireType = { size(value) { const { err, text } = wantText(value); - return err ? { err } : { size: text.length + 1 }; + if (err) return { err }; + + return tooLongForLengthOctet(text) ?? { size: text.length + 1 }; }, write(value, buffer, offset) { const { err, text } = wantText(value); if (err) return { err }; + const lengthErr = tooLongForLengthOctet(text); + + if (lengthErr) return lengthErr; + const rangeErr = outOfRange(buffer, offset, text.length + 1); if (rangeErr) return { err: rangeErr }; @@ -233,6 +239,12 @@ export const string: WireType = { }, }; +function tooLongForLengthOctet(text: string): { err: Error } | undefined { + if (text.length <= 0xFF) return undefined; + + return { err: new Error(`Octet String is ${String(text.length)} octets, the length octet holds 255`) }; +} + /** C-Octet String: NULL-terminated. */ export const cstring: WireType = { default: '', @@ -350,7 +362,11 @@ export const dest_address_array: WireType = { for (const dest of addresses) { if ('dl_name' in dest) { buf.writeUInt8(2, offset++); - writeCstring(dest.dl_name, buf, offset); + + const name = writeCstring(dest.dl_name, buf, offset); + + if (name.err) return { err: name.err }; + offset += dest.dl_name.length + 1; } else { buf.writeUInt8(1, offset++); @@ -363,7 +379,10 @@ export const dest_address_array: WireType = { if (npi.err) return { err: npi.err }; - writeCstring(dest.destination_addr, buf, offset); + const addr = writeCstring(dest.destination_addr, buf, offset); + + if (addr.err) return { err: addr.err }; + offset += dest.destination_addr.length + 1; } } @@ -448,7 +467,10 @@ export const unsuccess_sme_array: WireType = { if (npi.err) return { err: npi.err }; - writeCstring(sme.destination_addr, buf, offset); + const addr = writeCstring(sme.destination_addr, buf, offset); + + if (addr.err) return { err: addr.err }; + offset += sme.destination_addr.length + 1; const status = writeInt32(sme.error_status_code, buf, offset); diff --git a/src/dlr-merger.ts b/src/dlr-merger.ts index 4a848e8..411bb36 100644 --- a/src/dlr-merger.ts +++ b/src/dlr-merger.ts @@ -1,4 +1,5 @@ import type { Dlr } from './dlr.ts'; +import type { MessageState } from './defs/constants.ts'; import type { LogInt } from '@larvit/log'; import { ExpiringGroups } from './expiring-groups.ts'; @@ -24,6 +25,29 @@ const numbered = /^(.*)-(\d+)$/; * numbered its ids `-` off one base — the convention this library's own server follows. An * SMSC that hands out unrelated ids per segment cannot be merged, so nothing is reported for it. */ +/** + * MESSAGE_STATE is a flat enum, not a ranking — ACCEPTED is 6 where UNDELIVERABLE is 5 — so reducing + * on the wire value reports a part-failed message as delivered. Rank it deliberately instead. + */ +const severity: Record = { + DELIVERED: 0, + ACCEPTED: 1, + ENROUTE: 2, + SCHEDULED: 3, + SKIPPED: 4, + UNKNOWN: 5, + EXPIRED: 6, + DELETED: 7, + REJECTED: 8, + UNDELIVERABLE: 9, +}; + +function severityOf(dlr: Dlr): number { + const ranked: Record = severity; + + return ranked[dlr.statusMsg] ?? severity.UNKNOWN; +} + export class DlrMerger { private readonly groups: ExpiringGroups; private readonly log: LogInt; @@ -86,7 +110,7 @@ export class DlrMerger { this.groups.delete(base); const segments = [...group.parts.entries()].sort(([a], [b]) => a - b).map(([, one]) => one); - const worst = segments.reduce((carry, one) => (one.statusId > carry.statusId ? one : carry)); + const worst = segments.reduce((carry, one) => (severityOf(one) > severityOf(carry) ? one : carry)); return { ...worst, segments, smsId: base }; } diff --git a/src/index.ts b/src/index.ts index 13b4246..55faad2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -35,7 +35,7 @@ export { uuidv7 } from './uuid.ts'; export type { BindType, ClientOptions } from './client.ts'; export type { Dlr, Receipt } from './dlr.ts'; -export type { Sms, SmsInput } from './sms.ts'; +export type { SendRespOptions, Sms, SmsInput } from './sms.ts'; export type { ConcatInfo } from './udh.ts'; export type { Result, VoidResult } from './result.ts'; export type { @@ -49,6 +49,7 @@ export type { ReconnectOptions, SendOptions, SendSmsOptions, + SendSmsResult, SessionEvents, SessionOptions, } from './session.ts'; diff --git a/src/pdu.ts b/src/pdu.ts index f9c6927..d2b04e5 100644 --- a/src/pdu.ts +++ b/src/pdu.ts @@ -45,8 +45,10 @@ export type PduObject = { tlvs: Record; }; +const respBit = 0x80000000; + export function isResp(pduObj: Pick): boolean { - return pduObj.cmdId >= 0x80000000; + return pduObj.cmdId >= respBit; } /** @@ -169,6 +171,30 @@ function writeTlvs(tlvs: Record | undefined): Result<{ chunks: return { chunks }; } +/** + * SMPP 3.4: a response reporting a failure carries no body, so its fields are not "unused but + * present" — they are absent, and a peer that reads them anyway reads past the end of the PDU. + */ +function buildBody( + definition: CommandDefinition, + cmdName: CommandName, + cmdStatus: ErrorName, + params: Record, + tlvs: Record | undefined, +): Result<{ body: Buffer }> { + if (errors[cmdStatus] !== 0 && definition.id >= respBit) return { body: Buffer.alloc(0) }; + + const written = writeParams(definition, resolveShortMessage(params), cmdName); + + if (written.err) return { err: written.err }; + + const writtenTlvs = writeTlvs(tlvs); + + if (writtenTlvs.err) return { err: writtenTlvs.err }; + + return { body: Buffer.concat([...written.chunks, ...writtenTlvs.chunks]) }; +} + function buildPdu( cmdName: CommandName, cmdStatus: ErrorName, @@ -190,15 +216,11 @@ function buildPdu( return { err: new Error(`Invalid seqNr: ${JSON.stringify(seqNr)}`) }; } - const written = writeParams(definition, resolveShortMessage(params), cmdName); + const built = buildBody(definition, cmdName, cmdStatus, params, tlvs); - if (written.err) return { err: written.err }; + if (built.err) return { err: built.err }; - const writtenTlvs = writeTlvs(tlvs); - - if (writtenTlvs.err) return { err: writtenTlvs.err }; - - const body = Buffer.concat([...written.chunks, ...writtenTlvs.chunks]); + const body = built.body; const header = Buffer.alloc(16); header.writeUInt32BE(body.length + 16, 0); @@ -292,7 +314,10 @@ function parseOnce(pdu: Buffer, trailingNull: boolean): Result<{ aligned: boolea return { err: new Error(`Invalid seqNr, exceeds ${String(maxSeqNr)}: ${String(seqNr)}`) }; } - const read = readParams(cmdName, pdu, trailingNull); + // SMPP 3.4 4.4.2 and friends: a response with a non-zero status carries no body at all. + const read = cmdStatusId !== 0 && cmdLength === 16 + ? { offset: 16, params: {} } + : readParams(cmdName, pdu, trailingNull); if (read.err) return { err: read.err }; diff --git a/src/reassembly.ts b/src/reassembly.ts index 7906f16..47f59e7 100644 --- a/src/reassembly.ts +++ b/src/reassembly.ts @@ -42,15 +42,22 @@ function detach(pduObj: PduObject): PduObject { return { ...pduObj, params, tlvs }; } +// A cstring param arrives as a string, and source_addr alone can carry most of a 1 MiB PDU. +function sizeOf(value: unknown): number { + if (Buffer.isBuffer(value)) return value.length; + + return typeof value === 'string' ? value.length : 0; +} + function octetsOf(pduObj: PduObject): number { let octets = 0; for (const value of Object.values(pduObj.params)) { - if (Buffer.isBuffer(value)) octets += value.length; + octets += sizeOf(value); } for (const tlv of Object.values(pduObj.tlvs)) { - if (Buffer.isBuffer(tlv.tagValue)) octets += tlv.tagValue.length; + octets += sizeOf(tlv.tagValue); } return octets; diff --git a/src/send-sms.ts b/src/send-sms.ts index 57f7fc0..11e403e 100644 --- a/src/send-sms.ts +++ b/src/send-sms.ts @@ -61,7 +61,6 @@ export function submitSmParams( dest_addr_npi: sms.destinationAddrNpi ?? 0, dest_addr_ton: sms.destinationAddrTon ?? addressTon(sms.to), short_message: segment, - sm_length: segment.length, source_addr: sms.from, source_addr_npi: sms.sourceAddrNpi ?? 0, source_addr_ton: sms.sourceAddrTon ?? addressTon(sms.from), diff --git a/src/server.ts b/src/server.ts index bd451db..134adb8 100644 --- a/src/server.ts +++ b/src/server.ts @@ -1,10 +1,11 @@ import type { LogInt } from '@larvit/log'; import type { PduObject, TlvInput } from './pdu.ts'; -import type { Result } from './result.ts'; +import type { Result, VoidResult } from './result.ts'; import type { Server as NetServer, Socket } from 'node:net'; import type { Server as TlsServer, TlsOptions } from 'node:tls'; import { EventEmitter } from 'node:events'; import { Session, bindCommands, defaultSystemId } from './session.ts'; +import { checkSessionOptions } from './session-options.ts'; import { createServer as createNetServer } from 'node:net'; import { createServer as createTlsServer } from 'node:tls'; import { defaultInterfaceVersion } from './defs/constants.ts'; @@ -27,6 +28,7 @@ export type ServerOptions = { interfaceVersion?: number; log?: LogInt; maxOutstanding?: number; + maxOctets?: number; maxReassembly?: number; port?: number; reassemblyTimeout?: number; @@ -52,10 +54,12 @@ const defaults = { export class SmppServer extends EventEmitter { readonly sessions = new Set(); + private readonly log: LogInt; private readonly server: NetServer; - constructor(server: NetServer) { + constructor(server: NetServer, log: LogInt) { super(); + this.log = log; this.server = server; } @@ -66,11 +70,34 @@ export class SmppServer extends EventEmitter { return typeof address === 'object' && address !== null ? address.port : 0; } + /** A listener that throws is the application's bug; it must not become ours. Hard rule 1. */ + override emit( + event: K, + ...args: K extends keyof ServerEvents ? ServerEvents[K] : never + ): boolean { + try { + return super.emit(event, ...args); + } catch (thrown: unknown) { + const err = thrown instanceof Error ? thrown : new Error(String(thrown)); + + this.log.error('server - a listener threw', { event, message: err.message }); + + // Guarded against the listener that throws being the one listening for this. + if (event !== 'serverError') this.emit('serverError', err); + + return false; + } + } + /** Stops listening and closes every live session. */ close(): Promise { return new Promise(resolve => { for (const session of this.sessions) { - session.close(); + try { + session.close(); + } catch (thrown: unknown) { + this.emit('serverError', thrown instanceof Error ? thrown : new Error(String(thrown))); + } } this.sessions.clear(); @@ -173,6 +200,7 @@ function onConnection(sock: Socket, options: ServerOptions, server: SmppServer): idleTimeout: options.idleTimeout ?? defaults.idleTimeout, log, maxOutstanding: options.maxOutstanding, + maxOctets: options.maxOctets, maxReassembly: options.maxReassembly, onRequest: (bound, pduObj) => onRequest(bound, pduObj, options), reassemblyTimeout: options.reassemblyTimeout, @@ -203,6 +231,29 @@ function createSecureListener(tlsOptions: TlsOptions, log: LogInt): TlsServer { return listener; } +function checkOptions(options: ServerOptions, log: LogInt, port: number): VoidResult { + const checked = checkSessionOptions(options); + + if (checked.err) return { err: checked.err }; + + if (options.tls === true) { + log.warn('server - tls without a certificate', { port }); + + return { err: new Error('Listening over TLS needs tls: { cert, key }') }; + } + + // An int8 TLV on every bind response: out of range here means no ESME can ever bind. + const version = options.interfaceVersion ?? defaults.interfaceVersion; + + if (!Number.isInteger(version) || version < 0 || version > 0xFF) { + log.warn('server - interface version out of range', { interfaceVersion: version }); + + return { err: new Error(`interfaceVersion must be 0-255, got ${String(version)}`) }; + } + + return {}; +} + function createListener( options: ServerOptions, log: LogInt, @@ -210,20 +261,9 @@ function createListener( ): Result<{ listener: NetServer | TlsServer; useTls: boolean }> { const useTls = options.tls !== undefined && options.tls !== false; const tlsOptions = typeof options.tls === 'object' ? options.tls : undefined; - const version = options.interfaceVersion ?? defaults.interfaceVersion; + const checked = checkOptions(options, log, port); - if (useTls && !tlsOptions) { - log.warn('server - tls without a certificate', { port }); - - return { err: new Error('Listening over TLS needs tls: { cert, key }') }; - } - - // An int8 TLV on every bind response: out of range here means no ESME can ever bind. - if (!Number.isInteger(version) || version < 0 || version > 0xFF) { - log.warn('server - interface version out of range', { interfaceVersion: version }); - - return { err: new Error(`interfaceVersion must be 0-255, got ${String(version)}`) }; - } + if (checked.err) return { err: checked.err }; return { listener: tlsOptions ? createSecureListener(tlsOptions, log) : createNetServer(), @@ -242,15 +282,7 @@ function onListening(listener: NetServer, smpp: SmppServer, options: ServerOptio log.info('server - listening', { host: options.host ?? '*', port: smpp.port }); - // close() runs the application's own 'close' listeners, so a throw from one lands here. - options.signal?.addEventListener('abort', () => { - void smpp.close().catch((thrown: unknown) => { - const err = thrown instanceof Error ? thrown : new Error(String(thrown)); - - log.warn('server - could not close on abort', { message: err.message }); - smpp.emit('serverError', err); - }); - }, { once: true }); + options.signal?.addEventListener('abort', () => { void smpp.close(); }, { once: true }); } /** Starts listening for SMPP connections. Resolves once the socket is bound. */ @@ -262,7 +294,7 @@ export function server(options: ServerOptions = {}): Promise { onConnection(sock, options, smpp); diff --git a/src/session-options.ts b/src/session-options.ts index a8682e2..f2e810b 100644 --- a/src/session-options.ts +++ b/src/session-options.ts @@ -1,9 +1,30 @@ +import type { Dlr } from './dlr.ts'; import type { LogInt } from '@larvit/log'; +import type { MessageDlr } from './dlr-merger.ts'; import type { PduObject } from './pdu.ts'; import type { Result, VoidResult } from './result.ts'; import type { Session } from './session.ts'; +import type { Sms } from './sms.ts'; import type { Socket } from 'node:net'; +export type SessionEvents = { + close: []; + data: [Buffer]; + dlr: [Dlr, PduObject]; + incomingPdu: [Buffer]; + incomingPduObj: [PduObject]; + messageDlr: [MessageDlr]; + reconnected: []; + sessionError: [Error]; + sms: [Sms]; +}; + +export const bindCommands: readonly string[] = [ + 'bind_receiver', + 'bind_transceiver', + 'bind_transmitter', +]; + export type SendOptions = { signal?: AbortSignal | undefined }; /** @@ -21,6 +42,7 @@ export type SessionOptions = { enquireLinkInterval?: number | undefined; idleTimeout?: number | undefined; log?: LogInt | undefined; + maxOctets?: number | undefined; maxOutstanding?: number | undefined; maxReassembly?: number | undefined; /** @@ -51,3 +73,33 @@ export const defaults = { responseTimeout: 30_000, systemId: defaultSystemId, }; + +/** + * A count below 1 does not fail loudly anywhere downstream: `maxOutstanding: 0` leaves every send + * queued behind a slot that is never freed, so the call never settles at all. + */ +export function checkSessionOptions(options: SessionCounts): VoidResult { + const limits: [string, number, number][] = [ + ['idleTimeout', options.idleTimeout ?? 0, 0], + ['maxOutstanding', options.maxOutstanding ?? defaults.maxOutstanding, 1], + ['maxReassembly', options.maxReassembly ?? defaults.maxReassembly, 1], + ['reassemblyTimeout', options.reassemblyTimeout ?? defaults.reassemblyTimeout, 0], + ['responseTimeout', options.responseTimeout ?? defaults.responseTimeout, 0], + ]; + + for (const [name, value, min] of limits) { + if (!Number.isInteger(value) || value < min) { + return { err: new Error(`${name} must be ${String(min)} or more, got ${String(value)}`) }; + } + } + + return {}; +} + +export type SessionCounts = { + idleTimeout?: number | undefined; + maxOutstanding?: number | undefined; + maxReassembly?: number | undefined; + reassemblyTimeout?: number | undefined; + responseTimeout?: number | undefined; +}; diff --git a/src/session.ts b/src/session.ts index 699de6d..4024eb6 100644 --- a/src/session.ts +++ b/src/session.ts @@ -1,13 +1,11 @@ -import type { Dlr } from './dlr.ts'; import type { ErrorName } from './defs/errors.ts'; import type { LogInt } from '@larvit/log'; import type { MessageDlr } from './dlr-merger.ts'; import type { ParamValue } from './defs/types.ts'; import type { PduObject, PduObjectInput, TlvInput } from './pdu.ts'; -import type { ReconnectOptions, SendOptions, SessionOptions } from './session-options.ts'; +import type { ReconnectOptions, SendOptions, SessionEvents, SessionOptions } from './session-options.ts'; import type { Result, VoidResult } from './result.ts'; import type { SendSmsOptions, SendSmsResult } from './send-sms.ts'; -import type { Sms } from './sms.ts'; import type { Socket } from 'node:net'; import { DlrMerger } from './dlr-merger.ts'; import { EventEmitter } from 'node:events'; @@ -20,33 +18,23 @@ import { SendWindow } from './send-window.ts'; import { concatInfo } from './udh.ts'; import { consts, optionalParamsMinVersion } from './defs/constants.ts'; import { createSms } from './sms.ts'; -import { defaultSystemId, defaults } from './session-options.ts'; +import { bindCommands, defaultSystemId, defaults } from './session-options.ts'; import { dlrFromPdu } from './dlr.ts'; import { isResp, objToPdu, pduReturn, pduToObj } from './pdu.ts'; import { paramText } from './defs/types.ts'; import { silentLog } from './log.ts'; import { submitSms } from './send-sms.ts'; -export type { MessageDlr, ReconnectOptions, SendOptions, SendSmsOptions, SessionOptions }; -export { defaultSystemId }; - -export type SessionEvents = { - close: []; - data: [Buffer]; - dlr: [Dlr, PduObject]; - incomingPdu: [Buffer]; - incomingPduObj: [PduObject]; - messageDlr: [MessageDlr]; - reconnected: []; - sessionError: [Error]; - sms: [Sms]; +export type { + MessageDlr, + ReconnectOptions, + SendOptions, + SendSmsOptions, + SendSmsResult, + SessionEvents, + SessionOptions, }; - -export const bindCommands: readonly string[] = [ - 'bind_receiver', - 'bind_transceiver', - 'bind_transmitter', -]; +export { bindCommands, defaultSystemId }; export class Session extends EventEmitter { /** Replaced on reconnect, so hold the session rather than this. */ @@ -70,6 +58,25 @@ export class Session extends EventEmitter { private concatReference = 0; private framer = new PduFramer(); + /** A listener that throws is the application's bug; it must not become ours. Hard rule 1. */ + override emit( + event: K, + ...args: K extends keyof SessionEvents ? SessionEvents[K] : never + ): boolean { + try { + return super.emit(event, ...args); + } catch (thrown: unknown) { + const err = thrown instanceof Error ? thrown : new Error(String(thrown)); + + this.log.error('session - a listener threw', { event, message: err.message }); + + // Guarded against the listener that throws being the one listening for this. + if (event !== 'sessionError') this.emit('sessionError', err); + + return false; + } + } + constructor(options: SessionOptions) { super(); @@ -84,6 +91,7 @@ export class Session extends EventEmitter { this.reassembler = new Reassembler({ log: this.log, max: options.maxReassembly ?? defaults.maxReassembly, + maxOctets: options.maxOctets, timeout: options.reassemblyTimeout ?? defaults.reassemblyTimeout, }); this.reconnectLoop = this.loopFor(options.reconnect); @@ -138,7 +146,8 @@ export class Session extends EventEmitter { const built = pduReturn(pdu, status, params, tlvs); const sent = built.err ? { err: built.err } : this.write(built.buffer); - if (sent.err) { + // 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, message: sent.err.message, diff --git a/test/pdu.test.ts b/test/pdu.test.ts index f0222bb..8134712 100644 --- a/test/pdu.test.ts +++ b/test/pdu.test.ts @@ -24,7 +24,8 @@ describe('header', () => { test('writes command length, id, status and sequence number', () => { const pdu = encode({ cmdName: 'bind_transceiver_resp', cmdStatus: 'ESME_RALYBND', seqNr: 1 }); - assert.equal(pdu.readUInt32BE(0), 17); + // A failure response is header-only, so 16 rather than 17 with an empty system_id. + assert.equal(pdu.readUInt32BE(0), 16); assert.equal(pdu.readUInt32BE(4).toString(16), '80000009'); assert.equal(pdu.readUInt32BE(8), 5); assert.equal(pdu.readUInt32BE(12), 1); @@ -397,7 +398,7 @@ describe('pduReturn()', () => { params: { destination_addr: '46709771337', short_message: 'hi', source_addr: 'foo' }, seqNr: 9, })); - const { buffer, err } = pduReturn(request, 'ESME_RINVDSTADR', { message_id: 'abc123' }); + const { buffer, err } = pduReturn(request, 'ESME_ROK', { message_id: 'abc123' }); assert.equal(err, undefined); assert.ok(buffer); @@ -405,9 +406,16 @@ describe('pduReturn()', () => { const pduObj = decode(buffer); assert.equal(pduObj.cmdName, 'submit_sm_resp'); - assert.equal(pduObj.cmdStatus, 'ESME_RINVDSTADR'); + assert.equal(pduObj.cmdStatus, 'ESME_ROK'); assert.equal(pduObj.params.message_id, 'abc123'); assert.equal(pduObj.seqNr, 9); + + // The spec drops the body of a failure response, so the id a caller passes is not sent. + const refused = pduReturn(request, 'ESME_RINVDSTADR', { message_id: 'abc123' }); + + assert.ok(refused.buffer); + assert.equal(refused.buffer.length, 16); + assert.equal(decode(refused.buffer).params.message_id, undefined); }); test('refuses a command that has no response', () => { diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index ad6f99d..5ee11d5 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -1,7 +1,9 @@ import assert from 'node:assert/strict'; import net from 'node:net'; import test, { describe } from 'node:test'; +import type { Dlr } from '../src/dlr.ts'; import type { ErrorName } from '../src/defs/errors.ts'; +import type { MessageState } from '../src/defs/constants.ts'; import type { MessageDlr } from '../src/session.ts'; import type { PduObject, PduObjectInput } from '../src/pdu.ts'; import type { Result } from '../src/result.ts'; @@ -9,7 +11,9 @@ import type { SendSmsResult } from '../src/send-sms.ts'; import type { Sms } from '../src/sms.ts'; import type { SmppServer } from '../src/server.ts'; import { Reassembler, decodeSegments } from '../src/reassembly.ts'; +import { DlrMerger } from '../src/dlr-merger.ts'; import { client } from '../src/client.ts'; +import { consts } from '../src/defs/constants.ts'; import { errors } from '../src/defs/errors.ts'; import { server } from '../src/server.ts'; import { silentLog } from '../src/log.ts'; @@ -108,6 +112,34 @@ describe('merged delivery reports', () => { }); }); +describe('merging segment statuses', () => { + function receipt(smsId: string, statusMsg: MessageState): Dlr { + return { + doneDate: undefined, + errorCode: undefined, + receipt: undefined, + smsId, + statusId: consts.MESSAGE_STATE[statusMsg], + statusMsg, + }; + } + + // MESSAGE_STATE is a flat enum: ACCEPTED is 6 where UNDELIVERABLE is 5, so reducing on the + // wire value called a part-failed message delivered. + test('reports the worse of two states the wire numbers the other way round', () => { + const merger = new DlrMerger({ log: silentLog, max: 10, now: () => 0, timeout: 60_000 }); + + merger.expect(['msg-1', 'msg-2']); + + assert.equal(merger.collect(receipt('msg-1', 'UNDELIVERABLE')), undefined); + + const merged = merger.collect(receipt('msg-2', 'ACCEPTED')); + + assert.ok(merged); + assert.equal(merged.statusMsg, 'UNDELIVERABLE'); + }); +}); + describe('sendSms()', () => { function submitResp(seqNr: number, messageId: string, status: ErrorName = 'ESME_ROK'): PduObject { return { @@ -332,7 +364,8 @@ describe('reassembly bounds', () => { const reassembler = new Reassembler({ log: silentLog, max: 10, - maxOctets: 30, + // One segment is 36 octets: 14 of short_message plus the two 11-octet addresses. + maxOctets: 80, now: () => 0, timeout: 60_000, }); diff --git a/test/session.test.ts b/test/session.test.ts index 9bcf0be..6f87c38 100644 --- a/test/session.test.ts +++ b/test/session.test.ts @@ -353,12 +353,14 @@ describe('bind', () => { await named.close(); }); - test('answers a refused bind with its own system_id too', async () => { + // The echo leak cannot reach a refusal at all: the spec gives a failure response no body. + test('answers a refused bind with no body to leak', async () => { const smpp = await startServer({ authenticate: () => false, systemId: 'the-smsc' }); const refused = await bindRaw(smpp, 0x34); assert.equal(refused.cmdStatus, 'ESME_RBINDFAIL'); - assert.equal(refused.params.system_id, 'the-smsc'); + assert.equal(refused.cmdLength, 16); + assert.deepEqual(refused.params, {}); await smpp.close(); }); @@ -373,7 +375,7 @@ describe('bind', () => { await smpp.close(); }); - test('answers a second bind with ESME_RALYBND and its own system_id', async () => { + test('answers a second bind with ESME_RALYBND and no body', async () => { const smpp = await startServer({ systemId: 'the-smsc' }); const peer = rawPeer(smpp.port); @@ -384,7 +386,7 @@ describe('bind', () => { const again = await peer.next(); assert.equal(again.cmdStatus, 'ESME_RALYBND'); - assert.equal(again.params.system_id, 'the-smsc'); + assert.deepEqual(again.params, {}); peer.close(); await smpp.close(); @@ -905,6 +907,90 @@ describe('application hooks that throw', () => { await smpp.close(); }); + // The guard for a throwing sms listener used to emit sessionError from inside its own catch. + test('survives a sessionError listener that throws as well', async () => { + const smpp = await startServer(); + + smpp.on('session', session => { + session.on('sessionError', () => { throw new Error('the reporter exploded too'); }); + session.on('sms', () => { throw new Error('listener exploded'); }); + }); + + const { session } = await connect(smpp, { responseTimeout: 200 }); + + assert.ok(session); + + const sent = await session.sendSms({ + from: '46701113311', + message: 'blows up both listeners', + to: '46709771337', + }); + + assert.ok(sent.err instanceof Error); + + session.close(); + await smpp.close(); + }); + + test('closes even when an application close listener throws', async () => { + const smpp = await startServer(); + + smpp.on('session', session => { + session.on('close', () => { throw new Error('close listener exploded'); }); + }); + + const { session } = await connect(smpp); + + assert.ok(session); + await smpp.close(); + assert.equal(smpp.sessions.size, 0); + + session.close(); + }); + + test('refuses a send window that can never free a slot', async () => { + const smpp = await startServer(); + const { err, session } = await connect(smpp, { maxOutstanding: 0 }); + + assert.ok(err instanceof Error); + assert.match(err.message, /maxOutstanding/); + assert.equal(session, undefined); + + await smpp.close(); + }); + + test('keeps the message id off a submit_sm_resp that refuses the message', async () => { + const smpp = await startServer(); + + smpp.on('session', bound => { + bound.on('sms', sms => { void sms.sendResp({ status: 'ESME_RMSGQFUL' }); }); + }); + + const peer = rawPeer(smpp.port); + + peer.write(bindOf(0x34)); + await peer.next(); + peer.write({ + cmdName: 'submit_sm', + params: { + destination_addr: '46709771337', + short_message: 'full queue', + source_addr: '46701113311', + }, + seqNr: 2, + }); + + const refused = await peer.next(); + + assert.equal(refused.cmdName, 'submit_sm_resp'); + assert.equal(refused.cmdStatus, 'ESME_RMSGQFUL'); + assert.equal(refused.cmdLength, 16); + assert.deepEqual(refused.params, {}); + + peer.close(); + await smpp.close(); + }); + test('keeps the reconnect loop alive when connect throws', async () => { let attempts = 0; const loop = new ReconnectLoop({ diff --git a/test/types.test.ts b/test/types.test.ts index b53f2f3..d7b503b 100644 --- a/test/types.test.ts +++ b/test/types.test.ts @@ -49,6 +49,14 @@ describe('string (Octet String)', () => { assert.deepEqual(types.string.read(encoded, 0), { bytesRead: 9, value: expected }); }); + // The length is one octet, so a longer value has nowhere to say how long it is. + test('refuses a value longer than the length octet can count', () => { + const tooLong = 'x'.repeat(256); + + assert.ok(types.string.size(tooLong).err instanceof Error); + assert.ok(types.string.write(tooLong, Buffer.alloc(300), 0).err instanceof Error); + }); + test('sizes as the string plus its length octet', () => { assert.deepEqual(types.string.size(expected), { size: 9 }); }); diff --git a/todo.md b/todo.md index 3d7b15c..c399830 100644 --- a/todo.md +++ b/todo.md @@ -5,7 +5,7 @@ rules there constrain every item below. ## Status -The rewrite is **feature complete and green**: 184 tests, lint and typecheck clean, verified on Node +The rewrite is **feature complete and green**: 190 tests, lint and typecheck clean, verified on Node 18, 20, 22 and 24. What is left is release work and a few things worth adding before or after 1.0.0. ```bash