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