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