1 /*
2 * Copyright (c) 1999, 2014, Oracle and/or its affiliates. All rights reserved.
3 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4 *
5 * This code is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License version 2 only, as
7 * published by the Free Software Foundation. Oracle designates this
8 * particular file as subject to the "Classpath" exception as provided
9 * by Oracle in the LICENSE file that accompanied this code.
10 *
11 * This code is distributed in the hope that it will be useful, but WITHOUT
12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14 * version 2 for more details (a copy is included in the LICENSE file that
15 * accompanied this code).
16 *
17 * You should have received a copy of the GNU General Public License version
18 * 2 along with this work; if not, write to the Free Software Foundation,
19 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
20 *
21 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
22 * or visit www.oracle.com if you need additional information or have any
23 * questions.
24 */
25
26 package com.sun.jndi.ldap;
27
28 import java.io.BufferedInputStream;
29 import java.io.BufferedOutputStream;
30 import java.io.InterruptedIOException;
31 import java.io.IOException;
32 import java.io.OutputStream;
33 import java.io.InputStream;
34 import java.net.Socket;
35 import javax.net.ssl.SSLSocket;
36
37 import javax.naming.CommunicationException;
38 import javax.naming.ServiceUnavailableException;
39 import javax.naming.NamingException;
40 import javax.naming.InterruptedNamingException;
41
42 import javax.naming.ldap.Control;
43
44 import java.lang.reflect.Method;
45 import java.lang.reflect.Constructor;
46 import java.lang.reflect.InvocationTargetException;
47 import java.util.Arrays;
48 import sun.misc.IOUtils;
49 //import javax.net.SocketFactory;
50
51 /**
52 * A thread that creates a connection to an LDAP server.
53 * After the connection, the thread reads from the connection.
54 * A caller can invoke methods on the instance to read LDAP responses
55 * and to send LDAP requests.
56 * <p>
57 * There is a one-to-one correspondence between an LdapClient and
58 * a Connection. Access to Connection and its methods is only via
59 * LdapClient with two exceptions: SASL authentication and StartTLS.
60 * SASL needs to access Connection's socket IO streams (in order to do encryption
61 * of the security layer). StartTLS needs to do replace IO streams
62 * and close the IO streams on nonfatal close. The code for SASL
63 * authentication can be treated as being the same as from LdapClient
64 * because the SASL code is only ever called from LdapClient, from
65 * inside LdapClient's synchronized authenticate() method. StartTLS is called
66 * directly by the application but should only occur when the underlying
67 * connection is quiet.
68 * <p>
69 * In terms of synchronization, worry about data structures
70 * used by the Connection thread because that usage might contend
71 * with calls by the main threads (i.e., those that call LdapClient).
72 * Main threads need to worry about contention with each other.
73 * Fields that Connection thread uses:
74 * inStream - synced access and update; initialized in constructor;
75 * referenced outside class unsync'ed (by LdapSasl) only
76 * when connection is quiet
77 * traceFile, traceTagIn, traceTagOut - no sync; debugging only
78 * parent - no sync; initialized in constructor; no updates
79 * pendingRequests - sync
80 * pauseLock - per-instance lock;
81 * paused - sync via pauseLock (pauseReader())
82 * Members used by main threads (LdapClient):
83 * host, port - unsync; read-only access for StartTLS and debug messages
84 * setBound(), setV3() - no sync; called only by LdapClient.authenticate(),
85 * which is a sync method called only when connection is "quiet"
86 * getMsgId() - sync
87 * writeRequest(), removeRequest(),findRequest(), abandonOutstandingReqs() -
88 * access to shared pendingRequests is sync
89 * writeRequest(), abandonRequest(), ldapUnbind() - access to outStream sync
90 * cleanup() - sync
91 * readReply() - access to sock sync
92 * unpauseReader() - (indirectly via writeRequest) sync on pauseLock
93 * Members used by SASL auth (main thread):
94 * inStream, outStream - no sync; used to construct new stream; accessed
95 * only when conn is "quiet" and not shared
96 * replaceStreams() - sync method
97 * Members used by StartTLS:
98 * inStream, outStream - no sync; used to record the existing streams;
99 * accessed only when conn is "quiet" and not shared
100 * replaceStreams() - sync method
101 * <p>
102 * Handles anonymous, simple, and SASL bind for v3; anonymous and simple
103 * for v2.
104 * %%% made public for access by LdapSasl %%%
105 *
106 * @author Vincent Ryan
107 * @author Rosanna Lee
108 * @author Jagane Sundar
109 */
110 public final class Connection implements Runnable {
111
112 private static final boolean debug = false;
113 private static final int dump = 0; // > 0 r, > 1 rw
114 public static final long DEFAULT_READ_TIMEOUT_MILLIS = 15 * 1000; // 15 second timeout;
115
116
117 final private Thread worker; // Initialized in constructor
118
119 private boolean v3 = true; // Set in setV3()
120
121 final public String host; // used by LdapClient for generating exception messages
122 // used by StartTlsResponse when creating an SSL socket
123 final public int port; // used by LdapClient for generating exception messages
124 // used by StartTlsResponse when creating an SSL socket
125
126 private boolean bound = false; // Set in setBound()
127
128 // All three are initialized in constructor and read-only afterwards
129 private OutputStream traceFile = null;
130 private String traceTagIn = null;
131 private String traceTagOut = null;
132
133 // Initialized in constructor; read and used externally (LdapSasl);
134 // Updated in replaceStreams() during "quiet", unshared, period
135 public InputStream inStream; // must be public; used by LdapSasl
136
137 // Initialized in constructor; read and used externally (LdapSasl);
138 // Updated in replaceOutputStream() during "quiet", unshared, period
139 public OutputStream outStream; // must be public; used by LdapSasl
140
141 // Initialized in constructor; read and used externally (TLS) to
142 // get new IO streams; closed during cleanup
143 public Socket sock; // for TLS
144
145 // For processing "disconnect" unsolicited notification
146 // Initialized in constructor
147 final private LdapClient parent;
148
149 // Incremented and returned in sync getMsgId()
150 private int outMsgId = 0;
151
152 //
153 // The list of ldapRequests pending on this binding
154 //
155 // Accessed only within sync methods
156 private LdapRequest pendingRequests = null;
157
158 volatile IOException closureReason = null;
159 volatile boolean useable = true; // is Connection still useable
160
161 int readTimeout;
162 int connectTimeout;
163
164 // true means v3; false means v2
165 // Called in LdapClient.authenticate() (which is synchronized)
166 // when connection is "quiet" and not shared; no need to synchronize
167 void setV3(boolean v) {
168 v3 = v;
169 }
170
171 // A BIND request has been successfully made on this connection
172 // When cleaning up, remember to do an UNBIND
173 // Called in LdapClient.authenticate() (which is synchronized)
174 // when connection is "quiet" and not shared; no need to synchronize
175 void setBound() {
176 bound = true;
177 }
178
179 ////////////////////////////////////////////////////////////////////////////
180 //
181 // Create an LDAP Binding object and bind to a particular server
182 //
183 ////////////////////////////////////////////////////////////////////////////
184
185 Connection(LdapClient parent, String host, int port, String socketFactory,
186 int connectTimeout, int readTimeout, OutputStream trace) throws NamingException {
187
188 this.host = host;
189 this.port = port;
190 this.parent = parent;
191 this.readTimeout = readTimeout;
192 this.connectTimeout = connectTimeout;
193
194 if (trace != null) {
195 traceFile = trace;
196 traceTagIn = "<- " + host + ":" + port + "\n\n";
197 traceTagOut = "-> " + host + ":" + port + "\n\n";
198 }
199
200 //
201 // Connect to server
202 //
203 try {
204 sock = createSocket(host, port, socketFactory, connectTimeout);
205
206 if (debug) {
207 System.err.println("Connection: opening socket: " + host + "," + port);
208 }
209
210 inStream = new BufferedInputStream(sock.getInputStream());
211 outStream = new BufferedOutputStream(sock.getOutputStream());
212
213 } catch (InvocationTargetException e) {
214 Throwable realException = e.getTargetException();
215 // realException.printStackTrace();
216
217 CommunicationException ce =
218 new CommunicationException(host + ":" + port);
219 ce.setRootCause(realException);
220 throw ce;
221 } catch (Exception e) {
222 // Class.forName() seems to do more error checking
223 // and will throw IllegalArgumentException and such.
224 // That's why we need to have a catch all here and
225 // ignore generic exceptions.
226 // Also catches all IO errors generated by socket creation.
227 CommunicationException ce =
228 new CommunicationException(host + ":" + port);
229 ce.setRootCause(e);
230 throw ce;
231 }
232
233 worker = Obj.helper.createThread(this);
234 worker.setDaemon(true);
235 worker.start();
236 }
237
238 /*
239 * Create an InetSocketAddress using the specified hostname and port number.
240 */
241 private Object createInetSocketAddress(String host, int port)
242 throws NoSuchMethodException {
243
244 try {
245 Class<?> inetSocketAddressClass =
246 Class.forName("java.net.InetSocketAddress");
247
248 Constructor<?> inetSocketAddressCons =
249 inetSocketAddressClass.getConstructor(new Class<?>[]{
250 String.class, int.class});
251
252 return inetSocketAddressCons.newInstance(new Object[]{
253 host, new Integer(port)});
254
255 } catch (ClassNotFoundException |
256 InstantiationException |
257 InvocationTargetException |
258 IllegalAccessException e) {
259 throw new NoSuchMethodException();
260
261 }
262 }
263
264 /*
265 * Create a Socket object using the specified socket factory and time limit.
266 *
267 * If a timeout is supplied and unconnected sockets are supported then
268 * an unconnected socket is created and the timeout is applied when
269 * connecting the socket. If a timeout is supplied but unconnected sockets
270 * are not supported then the timeout is ignored and a connected socket
271 * is created.
272 */
273 private Socket createSocket(String host, int port, String socketFactory,
274 int connectTimeout) throws Exception {
275
276 Socket socket = null;
277
278 if (socketFactory != null) {
279
280 // create the factory
281
282 Class<?> socketFactoryClass = Obj.helper.loadClass(socketFactory);
283 Method getDefault =
284 socketFactoryClass.getMethod("getDefault", new Class<?>[]{});
285 Object factory = getDefault.invoke(null, new Object[]{});
286
287 // create the socket
288
289 Method createSocket = null;
290
291 if (connectTimeout > 0) {
292
293 try {
294 createSocket = socketFactoryClass.getMethod("createSocket",
295 new Class<?>[]{});
296
297 Method connect = Socket.class.getMethod("connect",
298 new Class<?>[]{Class.forName("java.net.SocketAddress"),
299 int.class});
300 Object endpoint = createInetSocketAddress(host, port);
301
302 // unconnected socket
303 socket =
304 (Socket)createSocket.invoke(factory, new Object[]{});
305
306 if (debug) {
307 System.err.println("Connection: creating socket with " +
308 "a timeout using supplied socket factory");
309 }
310
311 // connected socket
312 connect.invoke(socket, new Object[]{
313 endpoint, new Integer(connectTimeout)});
314
315 } catch (NoSuchMethodException e) {
316 // continue (but ignore connectTimeout)
317 }
318 }
319
320 if (socket == null) {
321 createSocket = socketFactoryClass.getMethod("createSocket",
322 new Class<?>[]{String.class, int.class});
323
324 if (debug) {
325 System.err.println("Connection: creating socket using " +
326 "supplied socket factory");
327 }
328 // connected socket
329 socket = (Socket) createSocket.invoke(factory,
330 new Object[]{host, new Integer(port)});
331 }
332 } else {
333
334 if (connectTimeout > 0) {
335
336 try {
337 Constructor<Socket> socketCons =
338 Socket.class.getConstructor(new Class<?>[]{});
339
340 Method connect = Socket.class.getMethod("connect",
341 new Class<?>[]{Class.forName("java.net.SocketAddress"),
342 int.class});
343 Object endpoint = createInetSocketAddress(host, port);
344
345 socket = socketCons.newInstance(new Object[]{});
346
347 if (debug) {
348 System.err.println("Connection: creating socket with " +
349 "a timeout");
350 }
351 connect.invoke(socket, new Object[]{
352 endpoint, new Integer(connectTimeout)});
353
354 } catch (NoSuchMethodException e) {
355 // continue (but ignore connectTimeout)
356 }
357 }
358
359 if (socket == null) {
360 if (debug) {
361 System.err.println("Connection: creating socket");
362 }
363 // connected socket
364 socket = new Socket(host, port);
365 }
366 }
367
368 // For LDAP connect timeouts on LDAP over SSL connections must treat
369 // the SSL handshake following socket connection as part of the timeout.
370 // So explicitly set a socket read timeout, trigger the SSL handshake,
371 // then reset the timeout.
372 if (connectTimeout > 0 && socket instanceof SSLSocket) {
373 SSLSocket sslSocket = (SSLSocket) socket;
374 int socketTimeout = sslSocket.getSoTimeout();
375
376 sslSocket.setSoTimeout(connectTimeout); // reuse full timeout value
377 sslSocket.startHandshake();
378 sslSocket.setSoTimeout(socketTimeout);
379 }
380
381 return socket;
382 }
383
384 ////////////////////////////////////////////////////////////////////////////
385 //
386 // Methods to IO to the LDAP server
387 //
388 ////////////////////////////////////////////////////////////////////////////
389
390 synchronized int getMsgId() {
391 return ++outMsgId;
392 }
393
394 LdapRequest writeRequest(BerEncoder ber, int msgId) throws IOException {
395 return writeRequest(ber, msgId, false /* pauseAfterReceipt */, -1);
396 }
397
398 LdapRequest writeRequest(BerEncoder ber, int msgId,
399 boolean pauseAfterReceipt) throws IOException {
400 return writeRequest(ber, msgId, pauseAfterReceipt, -1);
401 }
402
403 LdapRequest writeRequest(BerEncoder ber, int msgId,
404 boolean pauseAfterReceipt, int replyQueueCapacity) throws IOException {
405
406 LdapRequest req =
407 new LdapRequest(msgId, pauseAfterReceipt, replyQueueCapacity);
408 addRequest(req);
409
410 if (traceFile != null) {
411 Ber.dumpBER(traceFile, traceTagOut, ber.getBuf(), 0, ber.getDataLen());
412 }
413
414
415 // unpause reader so that it can get response
416 // NOTE: Must do this before writing request, otherwise might
417 // create a race condition where the writer unblocks its own response
418 unpauseReader();
419
420 if (debug) {
421 System.err.println("Writing request to: " + outStream);
422 }
423
424 try {
425 synchronized (this) {
426 outStream.write(ber.getBuf(), 0, ber.getDataLen());
427 outStream.flush();
428 }
429 } catch (IOException e) {
430 cleanup(null, true);
431 throw (closureReason = e); // rethrow
432 }
433
434 return req;
435 }
436
437 /**
438 * Reads a reply; waits until one is ready.
439 */
440 BerDecoder readReply(LdapRequest ldr)
441 throws IOException, NamingException {
442 BerDecoder rber;
443 boolean waited = false;
444
445 while (((rber = ldr.getReplyBer()) == null) && !waited) {
446 try {
447 // If socket closed, don't even try
448 synchronized (this) {
449 if (sock == null) {
450 throw new ServiceUnavailableException(host + ":" + port +
451 "; socket closed");
452 }
453 }
454 synchronized (ldr) {
455 // check if condition has changed since our last check
456 rber = ldr.getReplyBer();
457 if (rber == null) {
458 if (readTimeout > 0) { // Socket read timeout is specified
459
460 // will be woken up before readTimeout only if reply is
461 // available
462 ldr.wait(readTimeout);
463 } else {
464 ldr.wait(DEFAULT_READ_TIMEOUT_MILLIS);
465 }
466 waited = true;
467 } else {
468 break;
469 }
470 }
471 } catch (InterruptedException ex) {
472 throw new InterruptedNamingException(
473 "Interrupted during LDAP operation");
474 }
475 }
476
477 if ((rber == null) && waited) {
478 abandonRequest(ldr, null);
479 throw new NamingException("LDAP response read timed out, timeout used:"
480 + readTimeout + "ms." );
481
482 }
483 return rber;
484 }
485
486
487 ////////////////////////////////////////////////////////////////////////////
488 //
489 // Methods to add, find, delete, and abandon requests made to server
490 //
491 ////////////////////////////////////////////////////////////////////////////
492
493 private synchronized void addRequest(LdapRequest ldapRequest) {
494
495 LdapRequest ldr = pendingRequests;
496 if (ldr == null) {
497 pendingRequests = ldapRequest;
498 ldapRequest.next = null;
499 } else {
500 ldapRequest.next = pendingRequests;
501 pendingRequests = ldapRequest;
502 }
503 }
504
505 synchronized LdapRequest findRequest(int msgId) {
506
507 LdapRequest ldr = pendingRequests;
508 while (ldr != null) {
509 if (ldr.msgId == msgId) {
510 return ldr;
511 }
512 ldr = ldr.next;
513 }
514 return null;
515
516 }
517
518 synchronized void removeRequest(LdapRequest req) {
519 LdapRequest ldr = pendingRequests;
520 LdapRequest ldrprev = null;
521
522 while (ldr != null) {
523 if (ldr == req) {
524 ldr.cancel();
525
526 if (ldrprev != null) {
527 ldrprev.next = ldr.next;
528 } else {
529 pendingRequests = ldr.next;
530 }
531 ldr.next = null;
532 }
533 ldrprev = ldr;
534 ldr = ldr.next;
535 }
536 }
537
538 void abandonRequest(LdapRequest ldr, Control[] reqCtls) {
539 // Remove from queue
540 removeRequest(ldr);
541
542 BerEncoder ber = new BerEncoder(256);
543 int abandonMsgId = getMsgId();
544
545 //
546 // build the abandon request.
547 //
548 try {
549 ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
550 ber.encodeInt(abandonMsgId);
551 ber.encodeInt(ldr.msgId, LdapClient.LDAP_REQ_ABANDON);
552
553 if (v3) {
554 LdapClient.encodeControls(ber, reqCtls);
555 }
556 ber.endSeq();
557
558 if (traceFile != null) {
559 Ber.dumpBER(traceFile, traceTagOut, ber.getBuf(), 0,
560 ber.getDataLen());
561 }
562
563 synchronized (this) {
564 outStream.write(ber.getBuf(), 0, ber.getDataLen());
565 outStream.flush();
566 }
567
568 } catch (IOException ex) {
569 //System.err.println("ldap.abandon: " + ex);
570 }
571
572 // Don't expect any response for the abandon request.
573 }
574
575 synchronized void abandonOutstandingReqs(Control[] reqCtls) {
576 LdapRequest ldr = pendingRequests;
577
578 while (ldr != null) {
579 abandonRequest(ldr, reqCtls);
580 pendingRequests = ldr = ldr.next;
581 }
582 }
583
584 ////////////////////////////////////////////////////////////////////////////
585 //
586 // Methods to unbind from server and clear up resources when object is
587 // destroyed.
588 //
589 ////////////////////////////////////////////////////////////////////////////
590
591 private void ldapUnbind(Control[] reqCtls) {
592
593 BerEncoder ber = new BerEncoder(256);
594 int unbindMsgId = getMsgId();
595
596 //
597 // build the unbind request.
598 //
599
600 try {
601
602 ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
603 ber.encodeInt(unbindMsgId);
604 // IMPLICIT TAGS
605 ber.encodeByte(LdapClient.LDAP_REQ_UNBIND);
606 ber.encodeByte(0);
607
608 if (v3) {
609 LdapClient.encodeControls(ber, reqCtls);
610 }
611 ber.endSeq();
612
613 if (traceFile != null) {
614 Ber.dumpBER(traceFile, traceTagOut, ber.getBuf(),
615 0, ber.getDataLen());
616 }
617
618 synchronized (this) {
619 outStream.write(ber.getBuf(), 0, ber.getDataLen());
620 outStream.flush();
621 }
622
623 } catch (IOException ex) {
624 //System.err.println("ldap.unbind: " + ex);
625 }
626
627 // Don't expect any response for the unbind request.
628 }
629
630 /**
631 * @param reqCtls Possibly null request controls that accompanies the
632 * abandon and unbind LDAP request.
633 * @param notifyParent true means to call parent LdapClient back, notifying
634 * it that the connection has been closed; false means not to notify
635 * parent. If LdapClient invokes cleanup(), notifyParent should be set to
636 * false because LdapClient already knows that it is closing
637 * the connection. If Connection invokes cleanup(), notifyParent should be
638 * set to true because LdapClient needs to know about the closure.
639 */
640 void cleanup(Control[] reqCtls, boolean notifyParent) {
641 boolean nparent = false;
642
643 synchronized (this) {
644 useable = false;
645
646 if (sock != null) {
647 if (debug) {
648 System.err.println("Connection: closing socket: " + host + "," + port);
649 }
650 try {
651 if (!notifyParent) {
652 abandonOutstandingReqs(reqCtls);
653 }
654 if (bound) {
655 ldapUnbind(reqCtls);
656 }
657 } finally {
658 try {
659 outStream.flush();
660 sock.close();
661 unpauseReader();
662 } catch (IOException ie) {
663 if (debug)
664 System.err.println("Connection: problem closing socket: " + ie);
665 }
666 if (!notifyParent) {
667 LdapRequest ldr = pendingRequests;
668 while (ldr != null) {
669 ldr.cancel();
670 ldr = ldr.next;
671 }
672 }
673 sock = null;
674 }
675 nparent = notifyParent;
676 }
677 if (nparent) {
678 LdapRequest ldr = pendingRequests;
679 while (ldr != null) {
680
681 synchronized (ldr) {
682 ldr.notify();
683 ldr = ldr.next;
684 }
685 }
686 }
687 }
688 if (nparent) {
689 parent.processConnectionClosure();
690 }
691 }
692
693
694 // Assume everything is "quiet"
695 // "synchronize" might lead to deadlock so don't synchronize method
696 // Use streamLock instead for synchronizing update to stream
697
698 synchronized public void replaceStreams(InputStream newIn, OutputStream newOut) {
699 if (debug) {
700 System.err.println("Replacing " + inStream + " with: " + newIn);
701 System.err.println("Replacing " + outStream + " with: " + newOut);
702 }
703
704 inStream = newIn;
705
706 // Cleanup old stream
707 try {
708 outStream.flush();
709 } catch (IOException ie) {
710 if (debug)
711 System.err.println("Connection: cannot flush outstream: " + ie);
712 }
713
714 // Replace stream
715 outStream = newOut;
716 }
717
718 /**
719 * Used by Connection thread to read inStream into a local variable.
720 * This ensures that there is no contention between the main thread
721 * and the Connection thread when the main thread updates inStream.
722 */
723 synchronized private InputStream getInputStream() {
724 return inStream;
725 }
726
727
728 ////////////////////////////////////////////////////////////////////////////
729 //
730 // Code for pausing/unpausing the reader thread ('worker')
731 //
732 ////////////////////////////////////////////////////////////////////////////
733
734 /*
735 * The main idea is to mark requests that need the reader thread to
736 * pause after getting the response. When the reader thread gets the response,
737 * it waits on a lock instead of returning to the read(). The next time a
738 * request is sent, the reader is automatically unblocked if necessary.
739 * Note that the reader must be unblocked BEFORE the request is sent.
740 * Otherwise, there is a race condition where the request is sent and
741 * the reader thread might read the response and be unblocked
742 * by writeRequest().
743 *
744 * This pause gives the main thread (StartTLS or SASL) an opportunity to
745 * update the reader's state (e.g., its streams) if necessary.
746 * The assumption is that the connection will remain quiet during this pause
747 * (i.e., no intervening requests being sent).
748 *<p>
749 * For dealing with StartTLS close,
750 * when the read() exits either due to EOF or an exception,
751 * the reader thread checks whether there is a new stream to read from.
752 * If so, then it reattempts the read. Otherwise, the EOF or exception
753 * is processed and the reader thread terminates.
754 * In a StartTLS close, the client first replaces the SSL IO streams with
755 * plain ones and then closes the SSL socket.
756 * If the reader thread attempts to read, or was reading, from
757 * the SSL socket (that is, it got to the read BEFORE replaceStreams()),
758 * the SSL socket close will cause the reader thread to
759 * get an EOF/exception and reexamine the input stream.
760 * If the reader thread sees a new stream, it reattempts the read.
761 * If the underlying socket is still alive, then the new read will succeed.
762 * If the underlying socket has been closed also, then the new read will
763 * fail and the reader thread exits.
764 * If the reader thread attempts to read, or was reading, from the plain
765 * socket (that is, it got to the read AFTER replaceStreams()), the
766 * SSL socket close will have no effect on the reader thread.
767 *
768 * The check for new stream is made only
769 * in the first attempt at reading a BER buffer; the reader should
770 * never be in midst of reading a buffer when a nonfatal close occurs.
771 * If this occurs, then the connection is in an inconsistent state and
772 * the safest thing to do is to shut it down.
773 */
774
775 private Object pauseLock = new Object(); // lock for reader to wait on while paused
776 private boolean paused = false; // paused state of reader
777
778 /*
779 * Unpauses reader thread if it was paused
780 */
781 private void unpauseReader() throws IOException {
782 synchronized (pauseLock) {
783 if (paused) {
784 if (debug) {
785 System.err.println("Unpausing reader; read from: " +
786 inStream);
787 }
788 paused = false;
789 pauseLock.notify();
790 }
791 }
792 }
793
794 /*
795 * Pauses reader so that it stops reading from the input stream.
796 * Reader blocks on pauseLock instead of read().
797 * MUST be called from within synchronized (pauseLock) clause.
798 */
799 private void pauseReader() throws IOException {
800 if (debug) {
801 System.err.println("Pausing reader; was reading from: " +
802 inStream);
803 }
804 paused = true;
805 try {
806 while (paused) {
807 pauseLock.wait(); // notified by unpauseReader
808 }
809 } catch (InterruptedException e) {
810 throw new InterruptedIOException(
811 "Pause/unpause reader has problems.");
812 }
813 }
814
815
816 ////////////////////////////////////////////////////////////////////////////
817 //
818 // The LDAP Binding thread. It does the mux/demux of multiple requests
819 // on the same TCP connection.
820 //
821 ////////////////////////////////////////////////////////////////////////////
822
823
824 public void run() {
825 byte inbuf[]; // Buffer for reading incoming bytes
826 int inMsgId; // Message id of incoming response
827 int bytesread; // Number of bytes in inbuf
828 int br; // Temp; number of bytes read from stream
829 int offset; // Offset of where to store bytes in inbuf
830 int seqlen; // Length of ASN sequence
831 int seqlenlen; // Number of sequence length bytes
832 boolean eos; // End of stream
833 BerDecoder retBer; // Decoder for ASN.1 BER data from inbuf
834 InputStream in = null;
835
836 try {
837 while (true) {
838 try {
839 // type and length (at most 128 octets for long form)
840 inbuf = new byte[129];
841
842 offset = 0;
843 seqlen = 0;
844 seqlenlen = 0;
845
846 in = getInputStream();
847
848 // check that it is the beginning of a sequence
849 bytesread = in.read(inbuf, offset, 1);
850 if (bytesread < 0) {
851 if (in != getInputStream()) {
852 continue; // a new stream to try
853 } else {
854 break; // EOF
855 }
856 }
857
858 if (inbuf[offset++] != (Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR))
859 continue;
860
861 // get length of sequence
862 bytesread = in.read(inbuf, offset, 1);
863 if (bytesread < 0)
864 break; // EOF
865 seqlen = inbuf[offset++];
866
867 // if high bit is on, length is encoded in the
868 // subsequent length bytes and the number of length bytes
869 // is equal to & 0x80 (i.e. length byte with high bit off).
870 if ((seqlen & 0x80) == 0x80) {
871 seqlenlen = seqlen & 0x7f; // number of length bytes
872
873 bytesread = 0;
874 eos = false;
875
876 // Read all length bytes
877 while (bytesread < seqlenlen) {
878 br = in.read(inbuf, offset+bytesread,
879 seqlenlen-bytesread);
880 if (br < 0) {
881 eos = true;
882 break; // EOF
883 }
884 bytesread += br;
885 }
886
887 // end-of-stream reached before length bytes are read
888 if (eos)
889 break; // EOF
890
891 // Add contents of length bytes to determine length
892 seqlen = 0;
893 for( int i = 0; i < seqlenlen; i++) {
894 seqlen = (seqlen << 8) + (inbuf[offset+i] & 0xff);
895 }
896 offset += bytesread;
897 }
898
899 // read in seqlen bytes
900 byte[] left = IOUtils.readFully(in, seqlen, false);
901 inbuf = Arrays.copyOf(inbuf, offset + left.length);
902 System.arraycopy(left, 0, inbuf, offset, left.length);
903 offset += left.length;
904 /*
905 if (dump > 0) {
906 System.err.println("seqlen: " + seqlen);
907 System.err.println("bufsize: " + offset);
908 System.err.println("bytesleft: " + bytesleft);
909 System.err.println("bytesread: " + bytesread);
910 }
911 */
912
913
914 try {
915 retBer = new BerDecoder(inbuf, 0, offset);
916
917 if (traceFile != null) {
918 Ber.dumpBER(traceFile, traceTagIn, inbuf, 0, offset);
919 }
920
921 retBer.parseSeq(null);
922 inMsgId = retBer.parseInt();
923 retBer.reset(); // reset offset
924
925 boolean needPause = false;
926
927 if (inMsgId == 0) {
928 // Unsolicited Notification
929 parent.processUnsolicited(retBer);
930 } else {
931 LdapRequest ldr = findRequest(inMsgId);
932
933 if (ldr != null) {
934
935 /**
936 * Grab pauseLock before making reply available
937 * to ensure that reader goes into paused state
938 * before writer can attempt to unpause reader
939 */
940 synchronized (pauseLock) {
941 needPause = ldr.addReplyBer(retBer);
942 if (needPause) {
943 /*
944 * Go into paused state; release
945 * pauseLock
946 */
947 pauseReader();
948 }
949
950 // else release pauseLock
951 }
952 } else {
953 // System.err.println("Cannot find" +
954 // "LdapRequest for " + inMsgId);
955 }
956 }
957 } catch (Ber.DecodeException e) {
958 //System.err.println("Cannot parse Ber");
959 }
960 } catch (IOException ie) {
961 if (debug) {
962 System.err.println("Connection: Inside Caught " + ie);
963 ie.printStackTrace();
964 }
965
966 if (in != getInputStream()) {
967 // A new stream to try
968 // Go to top of loop and continue
969 } else {
970 if (debug) {
971 System.err.println("Connection: rethrowing " + ie);
972 }
973 throw ie; // rethrow exception
974 }
975 }
976 }
977
978 if (debug) {
979 System.err.println("Connection: end-of-stream detected: "
980 + in);
981 }
982 } catch (IOException ex) {
983 if (debug) {
984 System.err.println("Connection: Caught " + ex);
985 }
986 closureReason = ex;
987 } finally {
988 cleanup(null, true); // cleanup
989 }
990 if (debug) {
991 System.err.println("Connection: Thread Exiting");
992 }
993 }
994
995
996 // This code must be uncommented to run the LdapAbandonTest.
997 /*public void sendSearchReqs(String dn, int numReqs) {
998 int i;
999 String attrs[] = null;
1000 for(i = 1; i <= numReqs; i++) {
1001 BerEncoder ber = new BerEncoder(2048);
1002
1003 try {
1004 ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
1005 ber.encodeInt(i);
1006 ber.beginSeq(LdapClient.LDAP_REQ_SEARCH);
1007 ber.encodeString(dn == null ? "" : dn);
1008 ber.encodeInt(0, LdapClient.LBER_ENUMERATED);
1009 ber.encodeInt(3, LdapClient.LBER_ENUMERATED);
1010 ber.encodeInt(0);
1011 ber.encodeInt(0);
1012 ber.encodeBoolean(true);
1013 LdapClient.encodeFilter(ber, "");
1014 ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
1015 ber.encodeStringArray(attrs);
1016 ber.endSeq();
1017 ber.endSeq();
1018 ber.endSeq();
1019 writeRequest(ber, i);
1020 //System.err.println("wrote request " + i);
1021 } catch (Exception ex) {
1022 //System.err.println("ldap.search: Caught " + ex + " building req");
1023 }
1024
1025 }
1026 } */
1027 }