16905649657

Committed 12 Aug 2025 10:05AM UTC coverage: 91.018% (+0.02%) from 90.999%

Build # 16905649657

Build Type

push

github

Committed by

erayhanoglu

Commit Message

2.22.6

Coverage Stats

905 of 1138 branches covered (79.53%)

Branch coverage included in aggregate %.

5621 of 6032 relevant lines covered (93.19%)

200.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.63

/src/protocol/pg-socket.ts

import crypto from 'node:crypto';
import net from 'node:net';
import path from 'node:path';
import tls from 'node:tls';
import promisify from 'putil-promisify';
import { ConnectionState } from '../constants.js';
import type { ConnectionConfiguration } from '../interfaces/database-connection-params.js';
import { SafeEventEmitter } from '../safe-event-emitter.js';
import { Callback, Maybe } from '../types.js';
import { Backend } from './backend.js';
import { DatabaseError } from './database-error.js';
import { Frontend } from './frontend.js';
import { Protocol } from './protocol.js';
import { SASL } from './sasl.js';

const DEFAULT_PORT_NUMBER = 5432;
const COMMAND_RESULT_PATTERN = /^([^\d]+)(?: (\d+)(?: (\d+))?)?$/;

export type CaptureCallback = (
  code: Protocol.BackendMessageCode,
  msg: any,
  done: (err: Maybe<Error>, result?: any) => void,
) => void | Promise<void>;

export interface SocketError extends Error {
  code: string;
}

export class PgSocket extends SafeEventEmitter {
  private _state = ConnectionState.CLOSED;
  private _socket?: net.Socket;
  private _backend = new Backend();
  private _frontend: Frontend;
  private _sessionParameters: Record<string, string> = {};
  private _saslSession?: SASL.Session;
  private _processID?: number;
  private _secretKey?: number;

  constructor(public options: ConnectionConfiguration) {
    super();
    this._frontend = new Frontend({ buffer: options.buffer });
    this.setMaxListeners(99);
  }

  get state(): ConnectionState {
    if (!this._socket || this._socket.destroyed)
      this._state = ConnectionState.CLOSED;
    return this._state;
  }

  get processID(): Maybe<number> {
    return this._processID;
  }

  get secretKey(): Maybe<number> {
    return this._secretKey;
  }

  get sessionParameters(): Record<string, string> {
    return this._sessionParameters;
  }

  connect() {
    if (this._socket) return;
    this._state = ConnectionState.CONNECTING;
    const options = this.options;
    const socket = (this._socket = new net.Socket());

    const errorHandler = (err: Error) => {
      this._state = ConnectionState.CLOSED;
      this._removeListeners();
      this._reset();
      socket.destroy();
      this._socket = undefined;
      this.emit('error', err);
    };

    const connectHandler = () => {
      socket.setTimeout(0);
      if (this.options.keepAlive || this.options.keepAlive == null)
        socket.setKeepAlive(true);
      socket.write(this._frontend.getSSLRequestMessage());
      socket.once('data', x => {
        this._removeListeners();
        if (x.toString() === 'S') {
          const tslOptions = { ...options.ssl, socket };
          if (options.host && net.isIP(options.host) === 0)
            tslOptions.servername = options.host;
          const tlsSocket = (this._socket = tls.connect(tslOptions));
          tlsSocket.once('error', errorHandler);
          tlsSocket.once('secureConnect', () => {
            this._removeListeners();
            this._handleConnect();
          });
          return;
        }
        if (x.toString() === 'N') {
          if (options.requireSSL) {
            return errorHandler(
              new Error('Server does not support SSL connections'),
            );
          }
          this._removeListeners();
          this._handleConnect();
          return;
        }
        return errorHandler(
          new Error('There was an error establishing an SSL connection'),
        );
      });
    };

    socket.setNoDelay(true);
    socket.setTimeout(options.connectTimeoutMs || 30000, () =>
      errorHandler(new Error('Connection timed out')),
    );
    socket.once('error', errorHandler);
    socket.once('connect', connectHandler);

    this.emit('connecting');
    const port = options.port || DEFAULT_PORT_NUMBER;
    if (options.host && options.host.startsWith('/')) {
      socket.connect(path.join(options.host, '/.s.PGSQL.' + port));
    } else
      socket.connect(
        options.port || DEFAULT_PORT_NUMBER,
        options.host || 'localhost',
      );
  }

  close(): void {
    if (!this._socket || this._socket.destroyed) {
      this._state = ConnectionState.CLOSED;
      this._socket = undefined;
      this._reset();
      return;
    }
    if (this._state === ConnectionState.CLOSING) return;
    const socket = this._socket;
    this._state = ConnectionState.CLOSING;
    this._removeListeners();
    socket.once('close', () => this._handleClose());
    socket.destroy();
  }

  sendParseMessage(args: Frontend.ParseMessageArgs, cb?: Callback): void {
    if (this.listenerCount('debug'))
      this.emit('debug', { location: 'PgSocket.sendParseMessage', args });

    this._send(this._frontend.getParseMessage(args), cb);
  }

  sendBindMessage(args: Frontend.BindMessageArgs, cb?: Callback): void {
    if (this.listenerCount('debug'))
      this.emit('debug', { location: 'PgSocket.sendBindMessage', args });

    this._send(this._frontend.getBindMessage(args), cb);
  }

  sendDescribeMessage(args: Frontend.DescribeMessageArgs, cb?: Callback): void {
    if (this.listenerCount('debug'))
      this.emit('debug', { location: 'PgSocket.sendDescribeMessage', args });

    this._send(this._frontend.getDescribeMessage(args), cb);
  }

  sendExecuteMessage(args: Frontend.ExecuteMessageArgs, cb?: Callback): void {
    if (this.listenerCount('debug'))
      this.emit('debug', { location: 'PgSocket.sendDescribeMessage', args });

    this._send(this._frontend.getExecuteMessage(args), cb);
  }

  sendCloseMessage(args: Frontend.CloseMessageArgs, cb?: Callback): void {
    if (this.listenerCount('debug'))
      this.emit('debug', { location: 'PgSocket.sendCloseMessage', args });

    this._send(this._frontend.getCloseMessage(args), cb);
  }

  sendQueryMessage(sql: string, cb?: Callback): void {
    if (this.listenerCount('debug'))
      this.emit('debug', { location: 'PgSocket.sendQueryMessage', sql });

    this._send(this._frontend.getQueryMessage(sql), cb);
  }

  sendFlushMessage(cb?: Callback): void {
    if (this.listenerCount('debug'))
      this.emit('debug', { location: 'PgSocket.sendFlushMessage' });

    this._send(this._frontend.getFlushMessage(), cb);
  }

  sendTerminateMessage(cb?: Callback): void {
    if (this.listenerCount('debug'))
      this.emit('debug', { location: 'PgSocket.sendTerminateMessage' });

    this._send(this._frontend.getTerminateMessage(), cb);
  }

  sendSyncMessage(): void {
    if (this.listenerCount('debug'))
      this.emit('debug', { location: 'PgSocket.sendSyncMessage' });

    this._send(this._frontend.getSyncMessage());
  }

  capture(callback: CaptureCallback): Promise<any> {
    if (
      this._state === ConnectionState.CLOSING ||
      this._state === ConnectionState.CLOSED
    ) {
      return Promise.reject(new Error('Connection closed'));
    }
    if (this._state !== ConnectionState.READY)
      return Promise.reject(new Error('Connection is not ready'));
    return new Promise((resolve, reject) => {
      const done = (err?: Error, result?: any) => {
        this.removeListener('close', closeHandler);
        this.removeListener('error', errorHandler);
        this.removeListener('message', msgHandler);
        if (err) reject(err);
        else resolve(result);
      };
      const errorHandler = (err: Error) => {
        this.removeListener('close', closeHandler);
        this.removeListener('message', msgHandler);
        reject(err);
      };
      const closeHandler = () => {
        this.removeListener('error', errorHandler);
        this.removeListener('message', msgHandler);
        reject(new Error('Connection closed'));
      };
      const msgHandler = (code: Protocol.BackendMessageCode, msg: any) => {
        const x = callback(code, msg, done);
        if (promisify.isPromise(x))
          (x as Promise<void>).catch(err => done(err));
      };
      this.once('close', closeHandler);
      this.once('error', errorHandler);
      this.on('message', msgHandler);
    });
  }

  protected _removeListeners(): void {
    if (!this._socket) return;
    this._socket.removeAllListeners('error');
    this._socket.removeAllListeners('connect');
    this._socket.removeAllListeners('data');
    this._socket.removeAllListeners('close');
  }

  protected _reset(): void {
    this._backend.reset();
    this._sessionParameters = {};
    this._processID = undefined;
    this._secretKey = undefined;
    this._saslSession = undefined;
  }

  protected _handleConnect(): void {
    const socket = this._socket;
    if (!socket) return;
    this._state = ConnectionState.AUTHORIZING;
    this._reset();
    socket.on('data', (data: Buffer) => this._handleData(data));
    socket.on('error', (err: SocketError) => this._handleError(err));
    socket.on('close', () => this._handleClose());
    this._send(
      this._frontend.getStartupMessage({
        user: this.options.user || 'postgres',
        database: this.options.database || '',
        application_name: this.options.applicationName || '',
      }),
    );
  }

  protected _handleClose(): void {
    this._reset();
    this._socket = undefined;
    this._state = ConnectionState.CLOSED;
    this.emit('close');
  }

  protected _handleError(err: unknown): void {
    if (this._state !== ConnectionState.READY) {
      this._socket?.end();
    }
    this.emit('error', err);
  }

  protected _handleData(data: Buffer): void {
    this._backend.parse(
      data,
      (code: Protocol.BackendMessageCode, payload?: any) => {
        try {
          switch (code) {
            case Protocol.BackendMessageCode.Authentication:
              this._handleAuthenticationMessage(payload);
              break;
            case Protocol.BackendMessageCode.ErrorResponse:
              this.emit('error', new DatabaseError(payload));
              break;
            case Protocol.BackendMessageCode.NoticeResponse:
              this.emit('notice', payload);
              break;
            case Protocol.BackendMessageCode.NotificationResponse:
              this.emit('notification', payload);
              break;
            case Protocol.BackendMessageCode.ParameterStatus:
              this._handleParameterStatus(
                payload as Protocol.ParameterStatusMessage,
              );
              break;
            case Protocol.BackendMessageCode.BackendKeyData:
              this._handleBackendKeyData(
                payload as Protocol.BackendKeyDataMessage,
              );
              break;
            case Protocol.BackendMessageCode.ReadyForQuery:
              if (this._state !== ConnectionState.READY) {
                this._state = ConnectionState.READY;
                this.emit('ready');
              } else this.emit('message', code, payload);
              break;
            case Protocol.BackendMessageCode.CommandComplete: {
              const msg = this._handleCommandComplete(payload);
              this.emit('message', code, msg);
              break;
            }
            default:
              this.emit('message', code, payload);
          }
        } catch (e) {
          this._handleError(e);
        }
      },
    );
  }

  protected _resolvePassword(cb: (password: string) => void): void {
    (async (): Promise<void> => {
      const pass =
        typeof this.options.password === 'function'
          ? await this.options.password()
          : this.options.password;
      cb(pass || '');
    })().catch(err => this._handleError(err));
  }

  protected _handleAuthenticationMessage(msg?: any): void {
    if (!msg) {
      this.emit('authenticate');
      return;
    }

    switch (msg.kind) {
      case Protocol.AuthenticationMessageKind.CleartextPassword:
        this._resolvePassword(password => {
          this._send(this._frontend.getPasswordMessage(password));
        });
        break;
      case Protocol.AuthenticationMessageKind.MD5Password:
        this._resolvePassword(password => {
          const md5 = (x: any) =>
            crypto.createHash('md5').update(x, 'utf8').digest('hex');
          const l = md5(password + this.options.user);
          const r = md5(Buffer.concat([Buffer.from(l), msg.salt]));
          const pass = 'md5' + r;
          this._send(this._frontend.getPasswordMessage(pass));
        });
        break;
      case Protocol.AuthenticationMessageKind.SASL: {
        if (!msg.mechanisms.includes('SCRAM-SHA-256')) {
          throw new Error(
            'SASL: Only mechanism SCRAM-SHA-256 is currently supported',
          );
        }
        const saslSession = (this._saslSession = SASL.createSession(
          this.options.user || '',
          'SCRAM-SHA-256',
        ));
        this._send(this._frontend.getSASLMessage(saslSession));
        break;
      }
      case Protocol.AuthenticationMessageKind.SASLContinue: {
        const saslSession = this._saslSession;
        if (!saslSession) throw new Error('SASL: Session not started yet');
        this._resolvePassword(password => {
          SASL.continueSession(saslSession, password, msg.data);
          const buf = this._frontend.getSASLFinalMessage(saslSession);
          this._send(buf);
        });
        break;
      }
      case Protocol.AuthenticationMessageKind.SASLFinal: {
        const session = this._saslSession;
        if (!session) throw new Error('SASL: Session not started yet');
        SASL.finalizeSession(session, msg.data);
        this._saslSession = undefined;
        break;
      }
      default:
        break;
    }
  }

  protected _handleParameterStatus(msg: Protocol.ParameterStatusMessage): void {
    this._sessionParameters[msg.name] = msg.value;
  }

  protected _handleBackendKeyData(msg: Protocol.BackendKeyDataMessage): void {
    this._processID = msg.processID;
    this._secretKey = msg.secretKey;
  }

  protected _handleCommandComplete(msg: any): Protocol.CommandCompleteMessage {
    const m = msg.command && msg.command.match(COMMAND_RESULT_PATTERN);
    const result: Protocol.CommandCompleteMessage = {
      command: m[1],
    };
    if (m[3] != null) {
      result.oid = parseInt(m[2], 10);
      result.rowCount = parseInt(m[3], 10);
    } else if (m[2]) result.rowCount = parseInt(m[2], 10);
    return result;
  }

  protected _send(data: Buffer, cb?: Callback): void {
    if (this._socket && this._socket.writable) {
      this._socket.write(data, cb);
    }
  }
}

1	import crypto from 'node:crypto';	1✔
2	import net from 'node:net';	1✔
3	import path from 'node:path';	1✔
4	import tls from 'node:tls';	1✔
5	import promisify from 'putil-promisify';	1✔
6	import { ConnectionState } from '../constants.js';	1✔
7	import type { ConnectionConfiguration } from '../interfaces/database-connection-params.js';	1✔
8	import { SafeEventEmitter } from '../safe-event-emitter.js';	1✔
9	import { Callback, Maybe } from '../types.js';	1✔
10	import { Backend } from './backend.js';	1✔
11	import { DatabaseError } from './database-error.js';	1✔
12	import { Frontend } from './frontend.js';	1✔
13	import { Protocol } from './protocol.js';	1✔
14	import { SASL } from './sasl.js';	1✔
15		1✔
16	const DEFAULT_PORT_NUMBER = 5432;	1✔
17	const COMMAND_RESULT_PATTERN = /^([^\d]+)(?: (\d+)(?: (\d+))?)?$/;	1✔
18		1✔
19	export type CaptureCallback = (	1✔
20	code: Protocol.BackendMessageCode,	1✔
21	msg: any,	1✔
22	done: (err: Maybe<Error>, result?: any) => void,	1✔
23	) => void \| Promise<void>;	1✔
24		1✔
25	export interface SocketError extends Error {	1✔
26	code: string;	1✔
27	}	1✔
28		1✔
29	export class PgSocket extends SafeEventEmitter {	1✔
30	private _state = ConnectionState.CLOSED;	1✔
31	private _socket?: net.Socket;	1✔
32	private _backend = new Backend();	1✔
33	private _frontend: Frontend;	1✔
34	private _sessionParameters: Record<string, string> = {};	1✔
35	private _saslSession?: SASL.Session;	1✔
36	private _processID?: number;	1✔
37	private _secretKey?: number;	1✔
38		1✔
39	constructor(public options: ConnectionConfiguration) {	1✔
40	super();	62✔
41	this._frontend = new Frontend({ buffer: options.buffer });	62✔
42	this.setMaxListeners(99);	62✔
43	}	62✔
44		1✔
45	get state(): ConnectionState {	1✔
46	if (!this._socket \|\| this._socket.destroyed)	1,485✔
47	this._state = ConnectionState.CLOSED;	1,485✔
48	return this._state;	1,485✔
49	}	1,485✔
50		1✔
51	get processID(): Maybe<number> {	1✔
52	return this._processID;	1✔
53	}	1✔
54		1✔
55	get secretKey(): Maybe<number> {	1✔
56	return this._secretKey;	1✔
57	}	1✔
58		1✔
59	get sessionParameters(): Record<string, string> {	1✔
60	return this._sessionParameters;	×
61	}	×
62		1✔
63	connect() {	1✔
64	if (this._socket) return;	62!
65	this._state = ConnectionState.CONNECTING;	62✔
66	const options = this.options;	62✔
67	const socket = (this._socket = new net.Socket());	62✔
68		62✔
69	const errorHandler = (err: Error) => {	62✔
70	this._state = ConnectionState.CLOSED;	×
71	this._removeListeners();	×
72	this._reset();	×
73	socket.destroy();	×
74	this._socket = undefined;	×
75	this.emit('error', err);	×
76	};	×
77		62✔
78	const connectHandler = () => {	62✔
79	socket.setTimeout(0);	62✔
80	if (this.options.keepAlive \|\| this.options.keepAlive == null)	62✔
81	socket.setKeepAlive(true);	62✔
82	socket.write(this._frontend.getSSLRequestMessage());	62✔
83	socket.once('data', x => {	62✔
84	this._removeListeners();	62✔
85	if (x.toString() === 'S') {	62✔
86	const tslOptions = { ...options.ssl, socket };	61✔
87	if (options.host && net.isIP(options.host) === 0)	61✔
88	tslOptions.servername = options.host;	61✔
89	const tlsSocket = (this._socket = tls.connect(tslOptions));	61✔
90	tlsSocket.once('error', errorHandler);	61✔
91	tlsSocket.once('secureConnect', () => {	61✔
92	this._removeListeners();	61✔
93	this._handleConnect();	61✔
94	});	61✔
95	return;	61✔
96	}	61✔
97	if (x.toString() === 'N') {	1✔
98	if (options.requireSSL) {	1!
99	return errorHandler(	×
100	new Error('Server does not support SSL connections'),	×
101	);	×
102	}	×
103	this._removeListeners();	1✔
104	this._handleConnect();	1✔
105	return;	1✔
106	}	1!
107	return errorHandler(	×
108	new Error('There was an error establishing an SSL connection'),	×
109	);	×
110	});	×
111	};	62✔
112		62✔
113	socket.setNoDelay(true);	62✔
114	socket.setTimeout(options.connectTimeoutMs \|\| 30000, () =>	62✔
115	errorHandler(new Error('Connection timed out')),	×
116	);	62✔
117	socket.once('error', errorHandler);	62✔
118	socket.once('connect', connectHandler);	62✔
119		62✔
120	this.emit('connecting');	62✔
121	const port = options.port \|\| DEFAULT_PORT_NUMBER;	62✔
122	if (options.host && options.host.startsWith('/')) {	62✔
123	socket.connect(path.join(options.host, '/.s.PGSQL.' + port));	1✔
124	} else	1✔
125	socket.connect(	61✔
126	options.port \|\| DEFAULT_PORT_NUMBER,	61✔
127	options.host \|\| 'localhost',	61!
128	);	62✔
129	}	62✔
130		1✔
131	close(): void {	1✔
132	if (!this._socket \|\| this._socket.destroyed) {	62!
133	this._state = ConnectionState.CLOSED;	×
134	this._socket = undefined;	×
135	this._reset();	×
136	return;	×
137	}	×
138	if (this._state === ConnectionState.CLOSING) return;	62!
139	const socket = this._socket;	62✔
140	this._state = ConnectionState.CLOSING;	62✔
141	this._removeListeners();	62✔
142	socket.once('close', () => this._handleClose());	62✔
143	socket.destroy();	62✔
144	}	62✔
145		1✔
146	sendParseMessage(args: Frontend.ParseMessageArgs, cb?: Callback): void {	1✔
147	if (this.listenerCount('debug'))	230✔
148	this.emit('debug', { location: 'PgSocket.sendParseMessage', args });	230!
149		230✔
150	this._send(this._frontend.getParseMessage(args), cb);	230✔
151	}	230✔
152		1✔
153	sendBindMessage(args: Frontend.BindMessageArgs, cb?: Callback): void {	1✔
154	if (this.listenerCount('debug'))	229✔
155	this.emit('debug', { location: 'PgSocket.sendBindMessage', args });	229!
156		229✔
157	this._send(this._frontend.getBindMessage(args), cb);	229✔
158	}	229✔
159		1✔
160	sendDescribeMessage(args: Frontend.DescribeMessageArgs, cb?: Callback): void {	1✔
161	if (this.listenerCount('debug'))	229✔
162	this.emit('debug', { location: 'PgSocket.sendDescribeMessage', args });	229!
163		229✔
164	this._send(this._frontend.getDescribeMessage(args), cb);	229✔
165	}	229✔
166		1✔
167	sendExecuteMessage(args: Frontend.ExecuteMessageArgs, cb?: Callback): void {	1✔
168	if (this.listenerCount('debug'))	230✔
169	this.emit('debug', { location: 'PgSocket.sendDescribeMessage', args });	230!
170		230✔
171	this._send(this._frontend.getExecuteMessage(args), cb);	230✔
172	}	230✔
173		1✔
174	sendCloseMessage(args: Frontend.CloseMessageArgs, cb?: Callback): void {	1✔
175	if (this.listenerCount('debug'))	456✔
176	this.emit('debug', { location: 'PgSocket.sendCloseMessage', args });	456!
177		456✔
178	this._send(this._frontend.getCloseMessage(args), cb);	456✔
179	}	456✔
180		1✔
181	sendQueryMessage(sql: string, cb?: Callback): void {	1✔
182	if (this.listenerCount('debug'))	123✔
183	this.emit('debug', { location: 'PgSocket.sendQueryMessage', sql });	123!
184		123✔
185	this._send(this._frontend.getQueryMessage(sql), cb);	123✔
186	}	123✔
187		1✔
188	sendFlushMessage(cb?: Callback): void {	1✔
189	if (this.listenerCount('debug'))	916✔
190	this.emit('debug', { location: 'PgSocket.sendFlushMessage' });	916!
191		916✔
192	this._send(this._frontend.getFlushMessage(), cb);	916✔
193	}	916✔
194		1✔
195	sendTerminateMessage(cb?: Callback): void {	1✔
196	if (this.listenerCount('debug'))	62✔
197	this.emit('debug', { location: 'PgSocket.sendTerminateMessage' });	62!
198		62✔
199	this._send(this._frontend.getTerminateMessage(), cb);	62✔
200	}	62✔
201		1✔
202	sendSyncMessage(): void {	1✔
203	if (this.listenerCount('debug'))	686✔
204	this.emit('debug', { location: 'PgSocket.sendSyncMessage' });	686!
205		686✔
206	this._send(this._frontend.getSyncMessage());	686✔
207	}	686✔
208		1✔
209	capture(callback: CaptureCallback): Promise<any> {	1✔
210	if (	1,725✔
211	this._state === ConnectionState.CLOSING \|\|	1,725✔
212	this._state === ConnectionState.CLOSED	1,725✔
213	) {	1,725!
214	return Promise.reject(new Error('Connection closed'));	×
215	}	×
216	if (this._state !== ConnectionState.READY)	1,725✔
217	return Promise.reject(new Error('Connection is not ready'));	1,725!
218	return new Promise((resolve, reject) => {	1,725✔
219	const done = (err?: Error, result?: any) => {	1,725✔
220	this.removeListener('close', closeHandler);	1,723✔
221	this.removeListener('error', errorHandler);	1,723✔
222	this.removeListener('message', msgHandler);	1,723✔
223	if (err) reject(err);	1,723!
224	else resolve(result);	1,723✔
225	};	1,723✔
226	const errorHandler = (err: Error) => {	1,725✔
227	this.removeListener('close', closeHandler);	2✔
228	this.removeListener('message', msgHandler);	2✔
229	reject(err);	2✔
230	};	2✔
231	const closeHandler = () => {	1,725✔
232	this.removeListener('error', errorHandler);	×
233	this.removeListener('message', msgHandler);	×
234	reject(new Error('Connection closed'));	×
235	};	×
236	const msgHandler = (code: Protocol.BackendMessageCode, msg: any) => {	1,725✔
237	const x = callback(code, msg, done);	3,082✔
238	if (promisify.isPromise(x))	3,082✔
239	(x as Promise<void>).catch(err => done(err));	3,082✔
240	};	3,082✔
241	this.once('close', closeHandler);	1,725✔
242	this.once('error', errorHandler);	1,725✔
243	this.on('message', msgHandler);	1,725✔
244	});	1,725✔
245	}	1,725✔
246		1✔
247	protected _removeListeners(): void {	1✔
248	if (!this._socket) return;	186!
249	this._socket.removeAllListeners('error');	186✔
250	this._socket.removeAllListeners('connect');	186✔
251	this._socket.removeAllListeners('data');	186✔
252	this._socket.removeAllListeners('close');	186✔
253	}	186✔
254		1✔
255	protected _reset(): void {	1✔
256	this._backend.reset();	124✔
257	this._sessionParameters = {};	124✔
258	this._processID = undefined;	124✔
259	this._secretKey = undefined;	124✔
260	this._saslSession = undefined;	124✔
261	}	124✔
262		1✔
263	protected _handleConnect(): void {	1✔
264	const socket = this._socket;	62✔
265	if (!socket) return;	62!
266	this._state = ConnectionState.AUTHORIZING;	62✔
267	this._reset();	62✔
268	socket.on('data', (data: Buffer) => this._handleData(data));	62✔
269	socket.on('error', (err: SocketError) => this._handleError(err));	62✔
270	socket.on('close', () => this._handleClose());	62✔
271	this._send(	62✔
272	this._frontend.getStartupMessage({	62✔
273	user: this.options.user \|\| 'postgres',	62!
274	database: this.options.database \|\| '',	62!
275	application_name: this.options.applicationName \|\| '',	62✔
276	}),	62✔
277	);	62✔
278	}	62✔
279		1✔
280	protected _handleClose(): void {	1✔
281	this._reset();	62✔
282	this._socket = undefined;	62✔
283	this._state = ConnectionState.CLOSED;	62✔
284	this.emit('close');	62✔
285	}	62✔
286		1✔
287	protected _handleError(err: unknown): void {	1✔
288	if (this._state !== ConnectionState.READY) {	×
289	this._socket?.end();	×
290	}	×
291	this.emit('error', err);	×
292	}	×
293		1✔
294	protected _handleData(data: Buffer): void {	1✔
295	this._backend.parse(	1,802✔
296	data,	1,802✔
297	(code: Protocol.BackendMessageCode, payload?: any) => {	1,802✔
298	try {	4,176✔
299	switch (code) {	4,176✔
300	case Protocol.BackendMessageCode.Authentication:	4,176✔
301	this._handleAuthenticationMessage(payload);	66✔
302	break;	66✔
303	case Protocol.BackendMessageCode.ErrorResponse:	4,176✔
304	this.emit('error', new DatabaseError(payload));	2✔
305	break;	2✔
306	case Protocol.BackendMessageCode.NoticeResponse:	4,176✔
307	this.emit('notice', payload);	8✔
308	break;	8✔
309	case Protocol.BackendMessageCode.NotificationResponse:	4,176✔
310	this.emit('notification', payload);	6✔
311	break;	6✔
312	case Protocol.BackendMessageCode.ParameterStatus:	4,176✔
313	this._handleParameterStatus(	888✔
314	payload as Protocol.ParameterStatusMessage,	888✔
315	);	888✔
316	break;	888✔
317	case Protocol.BackendMessageCode.BackendKeyData:	4,176✔
318	this._handleBackendKeyData(	62✔
319	payload as Protocol.BackendKeyDataMessage,	62✔
320	);	62✔
321	break;	62✔
322	case Protocol.BackendMessageCode.ReadyForQuery:	4,176✔
323	if (this._state !== ConnectionState.READY) {	871✔
324	this._state = ConnectionState.READY;	62✔
325	this.emit('ready');	62✔
326	} else this.emit('message', code, payload);	871✔
327	break;	871✔
328	case Protocol.BackendMessageCode.CommandComplete: {	4,176✔
329	const msg = this._handleCommandComplete(payload);	372✔
330	this.emit('message', code, msg);	372✔
331	break;	372✔
332	}	372✔
333	default:	4,176✔
334	this.emit('message', code, payload);	1,901✔
335	}	4,176✔
336	} catch (e) {	4,176!
337	this._handleError(e);	×
338	}	×
339	},	4,176✔
340	);	1,802✔
341	}	1,802✔
342		1✔
343	protected _resolvePassword(cb: (password: string) => void): void {	1✔
344	(async (): Promise<void> => {	2✔
345	const pass =	2✔
346	typeof this.options.password === 'function'	2!
347	? await this.options.password()	×
348	: this.options.password;	2✔
349	cb(pass \|\| '');	2!
350	})().catch(err => this._handleError(err));	2✔
351	}	2✔
352		1✔
353	protected _handleAuthenticationMessage(msg?: any): void {	1✔
354	if (!msg) {	66✔
355	this.emit('authenticate');	62✔
356	return;	62✔
357	}	62✔
358		4✔
359	switch (msg.kind) {	4✔
360	case Protocol.AuthenticationMessageKind.CleartextPassword:	66!
361	this._resolvePassword(password => {	×
362	this._send(this._frontend.getPasswordMessage(password));	×
363	});	×
364	break;	×
365	case Protocol.AuthenticationMessageKind.MD5Password:	66✔
366	this._resolvePassword(password => {	1✔
367	const md5 = (x: any) =>	1✔
368	crypto.createHash('md5').update(x, 'utf8').digest('hex');	2✔
369	const l = md5(password + this.options.user);	1✔
370	const r = md5(Buffer.concat([Buffer.from(l), msg.salt]));	1✔
371	const pass = 'md5' + r;	1✔
372	this._send(this._frontend.getPasswordMessage(pass));	1✔
373	});	1✔
374	break;	1✔
375	case Protocol.AuthenticationMessageKind.SASL: {	66✔
376	if (!msg.mechanisms.includes('SCRAM-SHA-256')) {	1!
377	throw new Error(	×
378	'SASL: Only mechanism SCRAM-SHA-256 is currently supported',	×
379	);	×
380	}	×
381	const saslSession = (this._saslSession = SASL.createSession(	1✔
382	this.options.user \|\| '',	1!
383	'SCRAM-SHA-256',	1✔
384	));	1✔
385	this._send(this._frontend.getSASLMessage(saslSession));	1✔
386	break;	1✔
387	}	1✔
388	case Protocol.AuthenticationMessageKind.SASLContinue: {	66✔
389	const saslSession = this._saslSession;	1✔
390	if (!saslSession) throw new Error('SASL: Session not started yet');	1!
391	this._resolvePassword(password => {	1✔
392	SASL.continueSession(saslSession, password, msg.data);	1✔
393	const buf = this._frontend.getSASLFinalMessage(saslSession);	1✔
394	this._send(buf);	1✔
395	});	1✔
396	break;	1✔
397	}	1✔
398	case Protocol.AuthenticationMessageKind.SASLFinal: {	66✔
399	const session = this._saslSession;	1✔
400	if (!session) throw new Error('SASL: Session not started yet');	1!
401	SASL.finalizeSession(session, msg.data);	1✔
402	this._saslSession = undefined;	1✔
403	break;	1✔
404	}	1✔
405	default:	66!
406	break;	×
407	}	66✔
408	}	66✔
409		1✔
410	protected _handleParameterStatus(msg: Protocol.ParameterStatusMessage): void {	1✔
411	this._sessionParameters[msg.name] = msg.value;	888✔
412	}	888✔
413		1✔
414	protected _handleBackendKeyData(msg: Protocol.BackendKeyDataMessage): void {	1✔
415	this._processID = msg.processID;	62✔
416	this._secretKey = msg.secretKey;	62✔
417	}	62✔
418		1✔
419	protected _handleCommandComplete(msg: any): Protocol.CommandCompleteMessage {	1✔
420	const m = msg.command && msg.command.match(COMMAND_RESULT_PATTERN);	372✔
421	const result: Protocol.CommandCompleteMessage = {	372✔
422	command: m[1],	372✔
423	};	372✔
424	if (m[3] != null) {	372✔
425	result.oid = parseInt(m[2], 10);	5✔
426	result.rowCount = parseInt(m[3], 10);	5✔
427	} else if (m[2]) result.rowCount = parseInt(m[2], 10);	372✔
428	return result;	372✔
429	}	372✔
430		1✔
431	protected _send(data: Buffer, cb?: Callback): void {	1✔
432	if (this._socket && this._socket.writable) {	3,224✔
433	this._socket.write(data, cb);	3,224✔
434	}	3,224✔
435	}	3,224✔
436	}	1✔

panates / postgrejs / 16905649657

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous