UNPKG

37.7 kBJavaScriptView Raw
1'use strict'
2
3const requestQueueFactory = require('./request-queue')
4const messageTrackerFactory = require('./message-tracker')
5const { MAX_MSGID } = require('./constants')
6
7const EventEmitter = require('events').EventEmitter
8const net = require('net')
9const tls = require('tls')
10const util = require('util')
11
12const once = require('once')
13const backoff = require('backoff')
14const vasync = require('vasync')
15const assert = require('assert-plus')
16const VError = require('verror').VError
17
18const Attribute = require('../attribute')
19const Change = require('../change')
20const Control = require('../controls/index').Control
21const SearchPager = require('./search_pager')
22const Protocol = require('../protocol')
23const dn = require('../dn')
24const errors = require('../errors')
25const filters = require('../filters')
26const messages = require('../messages')
27const url = require('../url')
28const CorkedEmitter = require('../corked_emitter')
29
30/// --- Globals
31
32const AbandonRequest = messages.AbandonRequest
33const AddRequest = messages.AddRequest
34const BindRequest = messages.BindRequest
35const CompareRequest = messages.CompareRequest
36const DeleteRequest = messages.DeleteRequest
37const ExtendedRequest = messages.ExtendedRequest
38const ModifyRequest = messages.ModifyRequest
39const ModifyDNRequest = messages.ModifyDNRequest
40const SearchRequest = messages.SearchRequest
41const UnbindRequest = messages.UnbindRequest
42const UnbindResponse = messages.UnbindResponse
43
44const LDAPResult = messages.LDAPResult
45const SearchEntry = messages.SearchEntry
46const SearchReference = messages.SearchReference
47// var SearchResponse = messages.SearchResponse
48const Parser = messages.Parser
49
50const PresenceFilter = filters.PresenceFilter
51
52const ConnectionError = errors.ConnectionError
53
54const CMP_EXPECT = [errors.LDAP_COMPARE_TRUE, errors.LDAP_COMPARE_FALSE]
55
56// node 0.6 got rid of FDs, so make up a client id for logging
57let CLIENT_ID = 0
58
59/// --- Internal Helpers
60
61function nextClientId () {
62 if (++CLIENT_ID === MAX_MSGID) { return 1 }
63
64 return CLIENT_ID
65}
66
67function validateControls (controls) {
68 if (Array.isArray(controls)) {
69 controls.forEach(function (c) {
70 if (!(c instanceof Control)) { throw new TypeError('controls must be [Control]') }
71 })
72 } else if (controls instanceof Control) {
73 controls = [controls]
74 } else {
75 throw new TypeError('controls must be [Control]')
76 }
77
78 return controls
79}
80
81function ensureDN (input, strict) {
82 if (dn.DN.isDN(input)) {
83 return dn
84 } else if (strict) {
85 return dn.parse(input)
86 } else if (typeof (input) === 'string') {
87 return input
88 } else {
89 throw new Error('invalid DN')
90 }
91}
92
93/// --- API
94
95/**
96 * Constructs a new client.
97 *
98 * The options object is required, and must contain either a URL (string) or
99 * a socketPath (string); the socketPath is only if you want to talk to an LDAP
100 * server over a Unix Domain Socket. Additionally, you can pass in a bunyan
101 * option that is the result of `new Logger()`, presumably after you've
102 * configured it.
103 *
104 * @param {Object} options must have either url or socketPath.
105 * @throws {TypeError} on bad input.
106 */
107function Client (options) {
108 assert.ok(options)
109
110 EventEmitter.call(this, options)
111
112 const self = this
113 this.urls = options.url ? [].concat(options.url).map(url.parse) : []
114 this._nextServer = 0
115 // updated in connectSocket() after each connect
116 this.host = undefined
117 this.port = undefined
118 this.secure = undefined
119 this.url = undefined
120 this.tlsOptions = options.tlsOptions
121 this.socketPath = options.socketPath || false
122
123 this.log = options.log.child({ clazz: 'Client' }, true)
124
125 this.timeout = parseInt((options.timeout || 0), 10)
126 this.connectTimeout = parseInt((options.connectTimeout || 0), 10)
127 this.idleTimeout = parseInt((options.idleTimeout || 0), 10)
128 if (options.reconnect) {
129 // Fall back to defaults if options.reconnect === true
130 const rOpts = (typeof (options.reconnect) === 'object')
131 ? options.reconnect
132 : {}
133 this.reconnect = {
134 initialDelay: parseInt(rOpts.initialDelay || 100, 10),
135 maxDelay: parseInt(rOpts.maxDelay || 10000, 10),
136 failAfter: parseInt(rOpts.failAfter, 10) || Infinity
137 }
138 }
139 this.strictDN = (options.strictDN !== undefined) ? options.strictDN : true
140
141 this.queue = requestQueueFactory({
142 size: parseInt((options.queueSize || 0), 10),
143 timeout: parseInt((options.queueTimeout || 0), 10)
144 })
145 if (options.queueDisable) {
146 this.queue.freeze()
147 }
148
149 // Implicitly configure setup action to bind the client if bindDN and
150 // bindCredentials are passed in. This will more closely mimic PooledClient
151 // auto-login behavior.
152 if (options.bindDN !== undefined &&
153 options.bindCredentials !== undefined) {
154 this.on('setup', function (clt, cb) {
155 clt.bind(options.bindDN, options.bindCredentials, function (err) {
156 if (err) {
157 if (self._socket) {
158 self._socket.destroy()
159 }
160 self.emit('error', err)
161 }
162 cb(err)
163 })
164 })
165 }
166
167 this._socket = null
168 this.connected = false
169 this.connect()
170}
171util.inherits(Client, EventEmitter)
172module.exports = Client
173
174/**
175 * Sends an abandon request to the LDAP server.
176 *
177 * The callback will be invoked as soon as the data is flushed out to the
178 * network, as there is never a response from abandon.
179 *
180 * @param {Number} messageID the messageID to abandon.
181 * @param {Control} controls (optional) either a Control or [Control].
182 * @param {Function} callback of the form f(err).
183 * @throws {TypeError} on invalid input.
184 */
185Client.prototype.abandon = function abandon (messageID, controls, callback) {
186 assert.number(messageID, 'messageID')
187 if (typeof (controls) === 'function') {
188 callback = controls
189 controls = []
190 } else {
191 controls = validateControls(controls)
192 }
193 assert.func(callback, 'callback')
194
195 const req = new AbandonRequest({
196 abandonID: messageID,
197 controls: controls
198 })
199
200 return this._send(req, 'abandon', null, callback)
201}
202
203/**
204 * Adds an entry to the LDAP server.
205 *
206 * Entry can be either [Attribute] or a plain JS object where the
207 * values are either a plain value or an array of values. Any value (that's
208 * not an array) will get converted to a string, so keep that in mind.
209 *
210 * @param {String} name the DN of the entry to add.
211 * @param {Object} entry an array of Attributes to be added or a JS object.
212 * @param {Control} controls (optional) either a Control or [Control].
213 * @param {Function} callback of the form f(err, res).
214 * @throws {TypeError} on invalid input.
215 */
216Client.prototype.add = function add (name, entry, controls, callback) {
217 assert.ok(name !== undefined, 'name')
218 assert.object(entry, 'entry')
219 if (typeof (controls) === 'function') {
220 callback = controls
221 controls = []
222 } else {
223 controls = validateControls(controls)
224 }
225 assert.func(callback, 'callback')
226
227 if (Array.isArray(entry)) {
228 entry.forEach(function (a) {
229 if (!Attribute.isAttribute(a)) { throw new TypeError('entry must be an Array of Attributes') }
230 })
231 } else {
232 const save = entry
233
234 entry = []
235 Object.keys(save).forEach(function (k) {
236 const attr = new Attribute({ type: k })
237 if (Array.isArray(save[k])) {
238 save[k].forEach(function (v) {
239 attr.addValue(v.toString())
240 })
241 } else {
242 attr.addValue(save[k].toString())
243 }
244 entry.push(attr)
245 })
246 }
247
248 const req = new AddRequest({
249 entry: ensureDN(name, this.strictDN),
250 attributes: entry,
251 controls: controls
252 })
253
254 return this._send(req, [errors.LDAP_SUCCESS], null, callback)
255}
256
257/**
258 * Performs a simple authentication against the server.
259 *
260 * @param {String} name the DN to bind as.
261 * @param {String} credentials the userPassword associated with name.
262 * @param {Control} controls (optional) either a Control or [Control].
263 * @param {Function} callback of the form f(err, res).
264 * @throws {TypeError} on invalid input.
265 */
266Client.prototype.bind = function bind (name,
267 credentials,
268 controls,
269 callback,
270 _bypass) {
271 if (typeof (name) !== 'string' && !(name instanceof dn.DN)) { throw new TypeError('name (string) required') }
272 assert.optionalString(credentials, 'credentials')
273 if (typeof (controls) === 'function') {
274 callback = controls
275 controls = []
276 } else {
277 controls = validateControls(controls)
278 }
279 assert.func(callback, 'callback')
280
281 const req = new BindRequest({
282 name: name || '',
283 authentication: 'Simple',
284 credentials: credentials || '',
285 controls: controls
286 })
287
288 // Connection errors will be reported to the bind callback too (useful when the LDAP server is not available)
289 const self = this
290 function callbackWrapper (err, ret) {
291 self.removeListener('connectError', callbackWrapper)
292 callback(err, ret)
293 }
294 this.addListener('connectError', callbackWrapper)
295
296 return this._send(req, [errors.LDAP_SUCCESS], null, callbackWrapper, _bypass)
297}
298
299/**
300 * Compares an attribute/value pair with an entry on the LDAP server.
301 *
302 * @param {String} name the DN of the entry to compare attributes with.
303 * @param {String} attr name of an attribute to check.
304 * @param {String} value value of an attribute to check.
305 * @param {Control} controls (optional) either a Control or [Control].
306 * @param {Function} callback of the form f(err, boolean, res).
307 * @throws {TypeError} on invalid input.
308 */
309Client.prototype.compare = function compare (name,
310 attr,
311 value,
312 controls,
313 callback) {
314 assert.ok(name !== undefined, 'name')
315 assert.string(attr, 'attr')
316 assert.string(value, 'value')
317 if (typeof (controls) === 'function') {
318 callback = controls
319 controls = []
320 } else {
321 controls = validateControls(controls)
322 }
323 assert.func(callback, 'callback')
324
325 const req = new CompareRequest({
326 entry: ensureDN(name, this.strictDN),
327 attribute: attr,
328 value: value,
329 controls: controls
330 })
331
332 return this._send(req, CMP_EXPECT, null, function (err, res) {
333 if (err) { return callback(err) }
334
335 return callback(null, (res.status === errors.LDAP_COMPARE_TRUE), res)
336 })
337}
338
339/**
340 * Deletes an entry from the LDAP server.
341 *
342 * @param {String} name the DN of the entry to delete.
343 * @param {Control} controls (optional) either a Control or [Control].
344 * @param {Function} callback of the form f(err, res).
345 * @throws {TypeError} on invalid input.
346 */
347Client.prototype.del = function del (name, controls, callback) {
348 assert.ok(name !== undefined, 'name')
349 if (typeof (controls) === 'function') {
350 callback = controls
351 controls = []
352 } else {
353 controls = validateControls(controls)
354 }
355 assert.func(callback, 'callback')
356
357 const req = new DeleteRequest({
358 entry: ensureDN(name, this.strictDN),
359 controls: controls
360 })
361
362 return this._send(req, [errors.LDAP_SUCCESS], null, callback)
363}
364
365/**
366 * Performs an extended operation on the LDAP server.
367 *
368 * Pretty much none of the LDAP extended operations return an OID
369 * (responseName), so I just don't bother giving it back in the callback.
370 * It's on the third param in `res` if you need it.
371 *
372 * @param {String} name the OID of the extended operation to perform.
373 * @param {String} value value to pass in for this operation.
374 * @param {Control} controls (optional) either a Control or [Control].
375 * @param {Function} callback of the form f(err, value, res).
376 * @throws {TypeError} on invalid input.
377 */
378Client.prototype.exop = function exop (name, value, controls, callback) {
379 assert.string(name, 'name')
380 if (typeof (value) === 'function') {
381 callback = value
382 controls = []
383 value = undefined
384 }
385 if (typeof (controls) === 'function') {
386 callback = controls
387 controls = []
388 } else {
389 controls = validateControls(controls)
390 }
391 assert.func(callback, 'callback')
392
393 const req = new ExtendedRequest({
394 requestName: name,
395 requestValue: value,
396 controls: controls
397 })
398
399 return this._send(req, [errors.LDAP_SUCCESS], null, function (err, res) {
400 if (err) { return callback(err) }
401
402 return callback(null, res.responseValue || '', res)
403 })
404}
405
406/**
407 * Performs an LDAP modify against the server.
408 *
409 * @param {String} name the DN of the entry to modify.
410 * @param {Change} change update to perform (can be [Change]).
411 * @param {Control} controls (optional) either a Control or [Control].
412 * @param {Function} callback of the form f(err, res).
413 * @throws {TypeError} on invalid input.
414 */
415Client.prototype.modify = function modify (name, change, controls, callback) {
416 assert.ok(name !== undefined, 'name')
417 assert.object(change, 'change')
418
419 const changes = []
420
421 function changeFromObject (change) {
422 if (!change.operation && !change.type) { throw new Error('change.operation required') }
423 if (typeof (change.modification) !== 'object') { throw new Error('change.modification (object) required') }
424
425 if (Object.keys(change.modification).length === 2 &&
426 typeof (change.modification.type) === 'string' &&
427 Array.isArray(change.modification.vals)) {
428 // Use modification directly if it's already normalized:
429 changes.push(new Change({
430 operation: change.operation || change.type,
431 modification: change.modification
432 }))
433 } else {
434 // Normalize the modification object
435 Object.keys(change.modification).forEach(function (k) {
436 const mod = {}
437 mod[k] = change.modification[k]
438 changes.push(new Change({
439 operation: change.operation || change.type,
440 modification: mod
441 }))
442 })
443 }
444 }
445
446 if (Change.isChange(change)) {
447 changes.push(change)
448 } else if (Array.isArray(change)) {
449 change.forEach(function (c) {
450 if (Change.isChange(c)) {
451 changes.push(c)
452 } else {
453 changeFromObject(c)
454 }
455 })
456 } else {
457 changeFromObject(change)
458 }
459
460 if (typeof (controls) === 'function') {
461 callback = controls
462 controls = []
463 } else {
464 controls = validateControls(controls)
465 }
466 assert.func(callback, 'callback')
467
468 const req = new ModifyRequest({
469 object: ensureDN(name, this.strictDN),
470 changes: changes,
471 controls: controls
472 })
473
474 return this._send(req, [errors.LDAP_SUCCESS], null, callback)
475}
476
477/**
478 * Performs an LDAP modifyDN against the server.
479 *
480 * This does not allow you to keep the old DN, as while the LDAP protocol
481 * has a facility for that, it's stupid. Just Search/Add.
482 *
483 * This will automatically deal with "new superior" logic.
484 *
485 * @param {String} name the DN of the entry to modify.
486 * @param {String} newName the new DN to move this entry to.
487 * @param {Control} controls (optional) either a Control or [Control].
488 * @param {Function} callback of the form f(err, res).
489 * @throws {TypeError} on invalid input.
490 */
491Client.prototype.modifyDN = function modifyDN (name,
492 newName,
493 controls,
494 callback) {
495 assert.ok(name !== undefined, 'name')
496 assert.string(newName, 'newName')
497 if (typeof (controls) === 'function') {
498 callback = controls
499 controls = []
500 } else {
501 controls = validateControls(controls)
502 }
503 assert.func(callback)
504
505 const DN = ensureDN(name)
506 // TODO: is non-strict handling desired here?
507 const newDN = dn.parse(newName)
508
509 const req = new ModifyDNRequest({
510 entry: DN,
511 deleteOldRdn: true,
512 controls: controls
513 })
514
515 if (newDN.length !== 1) {
516 req.newRdn = dn.parse(newDN.rdns.shift().toString())
517 req.newSuperior = newDN
518 } else {
519 req.newRdn = newDN
520 }
521
522 return this._send(req, [errors.LDAP_SUCCESS], null, callback)
523}
524
525/**
526 * Performs an LDAP search against the server.
527 *
528 * Note that the defaults for options are a 'base' search, if that's what
529 * you want you can just pass in a string for options and it will be treated
530 * as the search filter. Also, you can either pass in programatic Filter
531 * objects or a filter string as the filter option.
532 *
533 * Note that this method is 'special' in that the callback 'res' param will
534 * have two important events on it, namely 'entry' and 'end' that you can hook
535 * to. The former will emit a SearchEntry object for each record that comes
536 * back, and the latter will emit a normal LDAPResult object.
537 *
538 * @param {String} base the DN in the tree to start searching at.
539 * @param {Object} options parameters:
540 * - {String} scope default of 'base'.
541 * - {String} filter default of '(objectclass=*)'.
542 * - {Array} attributes [string] to return.
543 * - {Boolean} attrsOnly whether to return values.
544 * @param {Control} controls (optional) either a Control or [Control].
545 * @param {Function} callback of the form f(err, res).
546 * @throws {TypeError} on invalid input.
547 */
548Client.prototype.search = function search (base,
549 options,
550 controls,
551 callback,
552 _bypass) {
553 assert.ok(base !== undefined, 'search base')
554 if (Array.isArray(options) || (options instanceof Control)) {
555 controls = options
556 options = {}
557 } else if (typeof (options) === 'function') {
558 callback = options
559 controls = []
560 options = {
561 filter: new PresenceFilter({ attribute: 'objectclass' })
562 }
563 } else if (typeof (options) === 'string') {
564 options = { filter: filters.parseString(options) }
565 } else if (typeof (options) !== 'object') {
566 throw new TypeError('options (object) required')
567 }
568 if (typeof (options.filter) === 'string') {
569 options.filter = filters.parseString(options.filter)
570 } else if (!options.filter) {
571 options.filter = new PresenceFilter({ attribute: 'objectclass' })
572 } else if (!filters.isFilter(options.filter)) {
573 throw new TypeError('options.filter (Filter) required')
574 }
575 if (typeof (controls) === 'function') {
576 callback = controls
577 controls = []
578 } else {
579 controls = validateControls(controls)
580 }
581 assert.func(callback, 'callback')
582
583 if (options.attributes) {
584 if (!Array.isArray(options.attributes)) {
585 if (typeof (options.attributes) === 'string') {
586 options.attributes = [options.attributes]
587 } else {
588 throw new TypeError('options.attributes must be an Array of Strings')
589 }
590 }
591 }
592
593 const self = this
594 const baseDN = ensureDN(base, this.strictDN)
595
596 function sendRequest (ctrls, emitter, cb) {
597 const req = new SearchRequest({
598 baseObject: baseDN,
599 scope: options.scope || 'base',
600 filter: options.filter,
601 derefAliases: options.derefAliases || Protocol.NEVER_DEREF_ALIASES,
602 sizeLimit: options.sizeLimit || 0,
603 timeLimit: options.timeLimit || 10,
604 typesOnly: options.typesOnly || false,
605 attributes: options.attributes || [],
606 controls: ctrls
607 })
608
609 return self._send(req,
610 [errors.LDAP_SUCCESS],
611 emitter,
612 cb,
613 _bypass)
614 }
615
616 if (options.paged) {
617 // Perform automated search paging
618 const pageOpts = typeof (options.paged) === 'object' ? options.paged : {}
619 let size = 100 // Default page size
620 if (pageOpts.pageSize > 0) {
621 size = pageOpts.pageSize
622 } else if (options.sizeLimit > 1) {
623 // According to the RFC, servers should ignore the paging control if
624 // pageSize >= sizelimit. Some might still send results, but it's safer
625 // to stay under that figure when assigning a default value.
626 size = options.sizeLimit - 1
627 }
628
629 const pager = new SearchPager({
630 callback: callback,
631 controls: controls,
632 pageSize: size,
633 pagePause: pageOpts.pagePause
634 })
635 pager.on('search', sendRequest)
636 pager.begin()
637 } else {
638 sendRequest(controls, new CorkedEmitter(), callback)
639 }
640}
641
642/**
643 * Unbinds this client from the LDAP server.
644 *
645 * Note that unbind does not have a response, so this callback is actually
646 * optional; either way, the client is disconnected.
647 *
648 * @param {Function} callback of the form f(err).
649 * @throws {TypeError} if you pass in callback as not a function.
650 */
651Client.prototype.unbind = function unbind (callback) {
652 if (!callback) { callback = function () {} }
653
654 if (typeof (callback) !== 'function') { throw new TypeError('callback must be a function') }
655
656 // When the socket closes, it is useful to know whether it was due to a
657 // user-initiated unbind or something else.
658 this.unbound = true
659
660 if (!this._socket) { return callback() }
661
662 const req = new UnbindRequest()
663 return this._send(req, 'unbind', null, callback)
664}
665
666/**
667 * Attempt to secure connection with StartTLS.
668 */
669Client.prototype.starttls = function starttls (options,
670 controls,
671 callback,
672 _bypass) {
673 assert.optionalObject(options)
674 options = options || {}
675 callback = once(callback)
676 const self = this
677
678 if (this._starttls) {
679 return callback(new Error('STARTTLS already in progress or active'))
680 }
681
682 function onSend (err, emitter) {
683 if (err) {
684 callback(err)
685 return
686 }
687 /*
688 * Now that the request has been sent, block all outgoing messages
689 * until an error is received or we successfully complete the setup.
690 */
691 // TODO: block traffic
692 self._starttls = {
693 started: true
694 }
695
696 emitter.on('error', function (err) {
697 self._starttls = null
698 callback(err)
699 })
700 emitter.on('end', function (res) {
701 const sock = self._socket
702 /*
703 * Unplumb socket data during SSL negotiation.
704 * This will prevent the LDAP parser from stumbling over the TLS
705 * handshake and raising a ruckus.
706 */
707 sock.removeAllListeners('data')
708
709 options.socket = sock
710 const secure = tls.connect(options)
711 secure.once('secureConnect', function () {
712 /*
713 * Wire up 'data' and 'error' handlers like the normal socket.
714 * Handling 'end' events isn't necessary since the underlying socket
715 * will handle those.
716 */
717 secure.removeAllListeners('error')
718 secure.on('data', function onData (data) {
719 self.log.trace('data event: %s', util.inspect(data))
720
721 self._tracker.parser.write(data)
722 })
723 secure.on('error', function (err) {
724 self.log.trace({ err: err }, 'error event: %s', new Error().stack)
725
726 self.emit('error', err)
727 sock.destroy()
728 })
729 callback(null)
730 })
731 secure.once('error', function (err) {
732 // If the SSL negotiation failed, to back to plain mode.
733 self._starttls = null
734 secure.removeAllListeners()
735 callback(err)
736 })
737 self._starttls.success = true
738 self._socket = secure
739 })
740 }
741
742 const req = new ExtendedRequest({
743 requestName: '1.3.6.1.4.1.1466.20037',
744 requestValue: null,
745 controls: controls
746 })
747
748 return this._send(req,
749 [errors.LDAP_SUCCESS],
750 new EventEmitter(),
751 onSend,
752 _bypass)
753}
754
755/**
756 * Disconnect from the LDAP server and do not allow reconnection.
757 *
758 * If the client is instantiated with proper reconnection options, it's
759 * possible to initiate new requests after a call to unbind since the client
760 * will attempt to reconnect in order to fulfill the request.
761 *
762 * Calling destroy will prevent any further reconnection from occurring.
763 *
764 * @param {Object} err (Optional) error that was cause of client destruction
765 */
766Client.prototype.destroy = function destroy (err) {
767 this.destroyed = true
768 this.queue.freeze()
769 // Purge any queued requests which are now meaningless
770 this.queue.flush(function (msg, expect, emitter, cb) {
771 if (typeof (cb) === 'function') {
772 cb(new Error('client destroyed'))
773 }
774 })
775 if (this.connected) {
776 this.unbind()
777 } else if (this._socket) {
778 this._socket.destroy()
779 }
780 this.emit('destroy', err)
781}
782
783/**
784 * Initiate LDAP connection.
785 */
786Client.prototype.connect = function connect () {
787 if (this.connecting || this.connected) {
788 return
789 }
790 const self = this
791 const log = this.log
792 let socket
793 let tracker
794
795 // Establish basic socket connection
796 function connectSocket (cb) {
797 const server = self.urls[self._nextServer]
798 self._nextServer = (self._nextServer + 1) % self.urls.length
799
800 cb = once(cb)
801
802 function onResult (err, res) {
803 if (err) {
804 if (self.connectTimer) {
805 clearTimeout(self.connectTimer)
806 self.connectTimer = null
807 }
808 self.emit('connectError', err)
809 }
810 cb(err, res)
811 }
812 function onConnect () {
813 if (self.connectTimer) {
814 clearTimeout(self.connectTimer)
815 self.connectTimer = null
816 }
817 socket.removeAllListeners('error')
818 .removeAllListeners('connect')
819 .removeAllListeners('secureConnect')
820
821 tracker.id = nextClientId() + '__' + tracker.id
822 self.log = self.log.child({ ldap_id: tracker.id }, true)
823
824 // Move on to client setup
825 setupClient(cb)
826 }
827
828 const port = (server && server.port) || self.socketPath
829 const host = server && server.hostname
830 if (server && server.secure) {
831 socket = tls.connect(port, host, self.tlsOptions)
832 socket.once('secureConnect', onConnect)
833 } else {
834 socket = net.connect(port, host)
835 socket.once('connect', onConnect)
836 }
837 socket.once('error', onResult)
838 initSocket(server)
839
840 // Setup connection timeout handling, if desired
841 if (self.connectTimeout) {
842 self.connectTimer = setTimeout(function onConnectTimeout () {
843 if (!socket || !socket.readable || !socket.writeable) {
844 socket.destroy()
845 self._socket = null
846 onResult(new ConnectionError('connection timeout'))
847 }
848 }, self.connectTimeout)
849 }
850 }
851
852 // Initialize socket events and LDAP parser.
853 function initSocket (url) {
854 tracker = messageTrackerFactory({
855 id: url ? url.href : self.socketPath,
856 parser: new Parser({ log: log })
857 })
858
859 // This won't be set on TLS. So. Very. Annoying.
860 if (typeof (socket.setKeepAlive) !== 'function') {
861 socket.setKeepAlive = function setKeepAlive (enable, delay) {
862 return socket.socket
863 ? socket.socket.setKeepAlive(enable, delay)
864 : false
865 }
866 }
867
868 socket.on('data', function onData (data) {
869 log.trace('data event: %s', util.inspect(data))
870
871 tracker.parser.write(data)
872 })
873
874 // The "router"
875 tracker.parser.on('message', function onMessage (message) {
876 message.connection = self._socket
877 const callback = tracker.fetch(message.messageID)
878
879 if (!callback) {
880 log.error({ message: message.json }, 'unsolicited message')
881 return false
882 }
883
884 return callback(message)
885 })
886
887 tracker.parser.on('error', function onParseError (err) {
888 self.emit('error', new VError(err, 'Parser error for %s',
889 tracker.id))
890 self.connected = false
891 socket.end()
892 })
893 }
894
895 // After connect, register socket event handlers and run any setup actions
896 function setupClient (cb) {
897 cb = once(cb)
898
899 // Indicate failure if anything goes awry during setup
900 function bail (err) {
901 socket.destroy()
902 cb(err || new Error('client error during setup'))
903 }
904 // Work around lack of close event on tls.socket in node < 0.11
905 ((socket.socket) ? socket.socket : socket).once('close', bail)
906 socket.once('error', bail)
907 socket.once('end', bail)
908 socket.once('timeout', bail)
909 socket.once('cleanupSetupListeners', function onCleanup () {
910 socket.removeListener('error', bail)
911 .removeListener('close', bail)
912 .removeListener('end', bail)
913 .removeListener('timeout', bail)
914 })
915
916 self._socket = socket
917 self._tracker = tracker
918
919 // Run any requested setup (such as automatically performing a bind) on
920 // socket before signalling successful connection.
921 // This setup needs to bypass the request queue since all other activity is
922 // blocked until the connection is considered fully established post-setup.
923 // Only allow bind/search/starttls for now.
924 const basicClient = {
925 bind: function bindBypass (name, credentials, controls, callback) {
926 return self.bind(name, credentials, controls, callback, true)
927 },
928 search: function searchBypass (base, options, controls, callback) {
929 return self.search(base, options, controls, callback, true)
930 },
931 starttls: function starttlsBypass (options, controls, callback) {
932 return self.starttls(options, controls, callback, true)
933 },
934 unbind: self.unbind.bind(self)
935 }
936 vasync.forEachPipeline({
937 func: function (f, callback) {
938 f(basicClient, callback)
939 },
940 inputs: self.listeners('setup')
941 }, function (err, res) {
942 if (err) {
943 self.emit('setupError', err)
944 }
945 cb(err)
946 })
947 }
948
949 // Wire up "official" event handlers after successful connect/setup
950 function postSetup () {
951 // cleanup the listeners we attached in setup phrase.
952 socket.emit('cleanupSetupListeners');
953
954 // Work around lack of close event on tls.socket in node < 0.11
955 ((socket.socket) ? socket.socket : socket).once('close',
956 self._onClose.bind(self))
957 socket.on('end', function onEnd () {
958 log.trace('end event')
959
960 self.emit('end')
961 socket.end()
962 })
963 socket.on('error', function onSocketError (err) {
964 log.trace({ err: err }, 'error event: %s', new Error().stack)
965
966 self.emit('error', err)
967 socket.destroy()
968 })
969 socket.on('timeout', function onTimeout () {
970 log.trace('timeout event')
971
972 self.emit('socketTimeout')
973 socket.end()
974 })
975
976 const server = self.urls[self._nextServer]
977 if (server) {
978 self.host = server.hostname
979 self.port = server.port
980 self.secure = server.secure
981 }
982 }
983
984 let retry
985 let failAfter
986 if (this.reconnect) {
987 retry = backoff.exponential({
988 initialDelay: this.reconnect.initialDelay,
989 maxDelay: this.reconnect.maxDelay
990 })
991 failAfter = this.reconnect.failAfter
992 if (this.urls.length > 1 && failAfter) {
993 failAfter *= this.urls.length
994 }
995 } else {
996 retry = backoff.exponential({
997 initialDelay: 1,
998 maxDelay: 2
999 })
1000 failAfter = this.urls.length || 1
1001 }
1002 retry.failAfter(failAfter)
1003
1004 retry.on('ready', function (num, delay) {
1005 if (self.destroyed) {
1006 // Cease connection attempts if destroyed
1007 return
1008 }
1009 connectSocket(function (err) {
1010 if (!err) {
1011 postSetup()
1012 self.connecting = false
1013 self.connected = true
1014 self.emit('connect', socket)
1015 self.log.debug('connected after %d attempt(s)', num + 1)
1016 // Flush any queued requests
1017 self._flushQueue()
1018 self._connectRetry = null
1019 } else {
1020 retry.backoff(err)
1021 }
1022 })
1023 })
1024 retry.on('fail', function (err) {
1025 if (self.destroyed) {
1026 // Silence any connect/setup errors if destroyed
1027 return
1028 }
1029 self.log.debug('failed to connect after %d attempts', failAfter)
1030 // Communicate the last-encountered error
1031 if (err instanceof ConnectionError) {
1032 self.emit('connectTimeout', err)
1033 } else if (err.code === 'ECONNREFUSED') {
1034 self.emit('connectRefused', err)
1035 } else {
1036 self.emit('error', err)
1037 }
1038 })
1039
1040 this._connectRetry = retry
1041 this.connecting = true
1042 retry.backoff()
1043}
1044
1045/// --- Private API
1046
1047/**
1048 * Flush queued requests out to the socket.
1049 */
1050Client.prototype._flushQueue = function _flushQueue () {
1051 // Pull items we're about to process out of the queue.
1052 this.queue.flush(this._send.bind(this))
1053}
1054
1055/**
1056 * Clean up socket/parser resources after socket close.
1057 */
1058Client.prototype._onClose = function _onClose (closeError) {
1059 const socket = this._socket
1060 const tracker = this._tracker
1061 socket.removeAllListeners('connect')
1062 .removeAllListeners('data')
1063 .removeAllListeners('drain')
1064 .removeAllListeners('end')
1065 .removeAllListeners('error')
1066 .removeAllListeners('timeout')
1067 this._socket = null
1068 this.connected = false;
1069
1070 ((socket.socket) ? socket.socket : socket).removeAllListeners('close')
1071
1072 this.log.trace('close event had_err=%s', closeError ? 'yes' : 'no')
1073
1074 this.emit('close', closeError)
1075 // On close we have to walk the outstanding messages and go invoke their
1076 // callback with an error.
1077 tracker.purge(function (msgid, cb) {
1078 if (socket.unbindMessageID !== msgid) {
1079 return cb(new ConnectionError(tracker.id + ' closed'))
1080 } else {
1081 // Unbinds will be communicated as a success since we're closed
1082 const unbind = new UnbindResponse({ messageID: msgid })
1083 unbind.status = 'unbind'
1084 return cb(unbind)
1085 }
1086 })
1087
1088 // Trash any parser or starttls state
1089 this._tracker = null
1090 delete this._starttls
1091
1092 // Automatically fire reconnect logic if the socket was closed for any reason
1093 // other than a user-initiated unbind.
1094 if (this.reconnect && !this.unbound) {
1095 this.connect()
1096 }
1097 this.unbound = false
1098 return false
1099}
1100
1101/**
1102 * Maintain idle timer for client.
1103 *
1104 * Will start timer to fire 'idle' event if conditions are satisfied. If
1105 * conditions are not met and a timer is running, it will be cleared.
1106 *
1107 * @param {Boolean} override explicitly disable timer.
1108 */
1109Client.prototype._updateIdle = function _updateIdle (override) {
1110 if (this.idleTimeout === 0) {
1111 return
1112 }
1113 // Client must be connected but not waiting on any request data
1114 const self = this
1115 function isIdle (disable) {
1116 return ((disable !== true) &&
1117 (self._socket && self.connected) &&
1118 (self._tracker.pending === 0))
1119 }
1120 if (isIdle(override)) {
1121 if (!this._idleTimer) {
1122 this._idleTimer = setTimeout(function () {
1123 // Double-check idleness in case socket was torn down
1124 if (isIdle()) {
1125 self.emit('idle')
1126 }
1127 }, this.idleTimeout)
1128 }
1129 } else {
1130 if (this._idleTimer) {
1131 clearTimeout(this._idleTimer)
1132 this._idleTimer = null
1133 }
1134 }
1135}
1136
1137/**
1138 * Attempt to send an LDAP request.
1139 */
1140Client.prototype._send = function _send (message,
1141 expect,
1142 emitter,
1143 callback,
1144 _bypass) {
1145 assert.ok(message)
1146 assert.ok(expect)
1147 assert.optionalObject(emitter)
1148 assert.ok(callback)
1149
1150 // Allow connect setup traffic to bypass checks
1151 if (_bypass && this._socket && this._socket.writable) {
1152 return this._sendSocket(message, expect, emitter, callback)
1153 }
1154 if (!this._socket || !this.connected) {
1155 if (!this.queue.enqueue(message, expect, emitter, callback)) {
1156 callback(new ConnectionError('connection unavailable'))
1157 }
1158 // Initiate reconnect if needed
1159 if (this.reconnect) {
1160 this.connect()
1161 }
1162 return false
1163 } else {
1164 this._flushQueue()
1165 return this._sendSocket(message, expect, emitter, callback)
1166 }
1167}
1168
1169Client.prototype._sendSocket = function _sendSocket (message,
1170 expect,
1171 emitter,
1172 callback) {
1173 const conn = this._socket
1174 const tracker = this._tracker
1175 const log = this.log
1176 const self = this
1177 let timer = false
1178 let sentEmitter = false
1179
1180 function sendResult (event, obj) {
1181 if (event === 'error') {
1182 self.emit('resultError', obj)
1183 }
1184 if (emitter) {
1185 if (event === 'error') {
1186 // Error will go unhandled if emitter hasn't been sent via callback.
1187 // Execute callback with the error instead.
1188 if (!sentEmitter) { return callback(obj) }
1189 }
1190 return emitter.emit(event, obj)
1191 }
1192
1193 if (event === 'error') { return callback(obj) }
1194
1195 return callback(null, obj)
1196 }
1197
1198 function messageCallback (msg) {
1199 if (timer) { clearTimeout(timer) }
1200
1201 log.trace({ msg: msg ? msg.json : null }, 'response received')
1202
1203 if (expect === 'abandon') { return sendResult('end', null) }
1204
1205 if (msg instanceof SearchEntry || msg instanceof SearchReference) {
1206 let event = msg.constructor.name
1207 event = event[0].toLowerCase() + event.slice(1)
1208 return sendResult(event, msg)
1209 } else {
1210 tracker.remove(message.messageID)
1211 // Potentially mark client as idle
1212 self._updateIdle()
1213
1214 if (msg instanceof LDAPResult) {
1215 if (expect.indexOf(msg.status) === -1) {
1216 return sendResult('error', errors.getError(msg))
1217 }
1218 return sendResult('end', msg)
1219 } else if (msg instanceof Error) {
1220 return sendResult('error', msg)
1221 } else {
1222 return sendResult('error', new errors.ProtocolError(msg.type))
1223 }
1224 }
1225 }
1226
1227 function onRequestTimeout () {
1228 self.emit('timeout', message)
1229 const cb = tracker.fetch(message.messageID)
1230 if (cb) {
1231 // FIXME: the timed-out request should be abandoned
1232 cb(new errors.TimeoutError('request timeout (client interrupt)'))
1233 }
1234 }
1235
1236 function writeCallback () {
1237 if (expect === 'abandon') {
1238 // Mark the messageID specified as abandoned
1239 tracker.abandon(message.abandonID)
1240 // No need to track the abandon request itself
1241 tracker.remove(message.id)
1242 return callback(null)
1243 } else if (expect === 'unbind') {
1244 conn.unbindMessageID = message.id
1245 // Mark client as disconnected once unbind clears the socket
1246 self.connected = false
1247 // Some servers will RST the connection after receiving an unbind.
1248 // Socket errors are blackholed since the connection is being closed.
1249 conn.removeAllListeners('error')
1250 conn.on('error', function () {})
1251 conn.end()
1252 } else if (emitter) {
1253 sentEmitter = true
1254 return callback(null, emitter)
1255 }
1256 return false
1257 }
1258
1259 // Start actually doing something...
1260 tracker.track(message, messageCallback)
1261 // Mark client as active
1262 this._updateIdle(true)
1263
1264 if (self.timeout) {
1265 log.trace('Setting timeout to %d', self.timeout)
1266 timer = setTimeout(onRequestTimeout, self.timeout)
1267 }
1268
1269 log.trace('sending request %j', message.json)
1270
1271 try {
1272 return conn.write(message.toBer(), writeCallback)
1273 } catch (e) {
1274 if (timer) { clearTimeout(timer) }
1275
1276 log.trace({ err: e }, 'Error writing message to socket')
1277 return callback(e)
1278 }
1279}