Stop a thrown listener escaping, bound what a peer can pin, and drop the body from failure responses

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<Result<{ session: Session }>> {
-	const log = options.log ?? silentLog;
+async function connect(options: ClientOptions, log: LogInt): Promise<Result<{ sock: Socket }>> {
+	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<Result<{ sess
 		return { err: opened.err };
 	}
 
+	return { sock: opened.sock };
+}
+
+/** Connects to an SMSC and binds. */
+export async function client(options: ClientOptions = {}): Promise<Result<{ session: Session }>> {
+	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;
 

src/defs/types.ts

@@ -215,13 +215,19 @@ export const string: WireType<string> = {
 	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<string> = {
 	},
 };
 
+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<string> = {
 	default: '',
@@ -350,7 +362,11 @@ export const dest_address_array: WireType<DestAddress[]> = {
 		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<DestAddress[]> = {
 
 				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<UnsuccessSme[]> = {
 
 			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);

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 `<base>-<n>` 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<MessageState, number> = {
+	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<string, number | undefined> = severity;
+
+	return ranked[dlr.statusMsg] ?? severity.UNKNOWN;
+}
+
 export class DlrMerger {
 	private readonly groups: ExpiringGroups<Group>;
 	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 };
 	}

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';

src/pdu.ts

@@ -45,8 +45,10 @@ export type PduObject = {
 	tlvs: Record<string, Tlv>;
 };
 
+const respBit = 0x80000000;
+
 export function isResp(pduObj: Pick<PduObject, 'cmdId'>): boolean {
-	return pduObj.cmdId >= 0x80000000;
+	return pduObj.cmdId >= respBit;
 }
 
 /**
@@ -169,6 +171,30 @@ function writeTlvs(tlvs: Record<string, TlvInput> | 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<string, ParamValue | undefined>,
+	tlvs: Record<string, TlvInput> | 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 };
 

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;

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),

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<ServerEvents> {
 	readonly sessions = new Set<Session>();
 
+	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<ServerEvents> {
 		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<K extends keyof ServerEvents>(
+		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<void> {
 		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<Result<{ server: Sm
 	if (created.err) return Promise.resolve({ err: created.err });
 
 	const listener = created.listener;
-	const smpp = new SmppServer(listener);
+	const smpp = new SmppServer(listener, log);
 
 	listener.on(created.useTls ? 'secureConnection' : 'connection', (sock: Socket) => {
 		onConnection(sock, options, smpp);

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;
+};

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<SessionEvents> {
 	/** Replaced on reconnect, so hold the session rather than this. */
@@ -70,6 +58,25 @@ export class Session extends EventEmitter<SessionEvents> {
 	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<K extends keyof SessionEvents>(
+		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<SessionEvents> {
 		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<SessionEvents> {
 		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,