* There is a one-to-one correspondence between an LdapClient and * a Connection. Access to Connection and its methods is only via * LdapClient with two exceptions: SASL authentication and StartTLS. * SASL needs to access Connection's socket IO streams (in order to do encryption * of the security layer). StartTLS needs to do replace IO streams * and close the IO streams on nonfatal close. The code for SASL * authentication can be treated as being the same as from LdapClient * because the SASL code is only ever called from LdapClient, from * inside LdapClient's synchronized authenticate() method. StartTLS is called * directly by the application but should only occur when the underlying * connection is quiet. *
* In terms of synchronization, worry about data structures * used by the Connection thread because that usage might contend * with calls by the main threads (i.e., those that call LdapClient). * Main threads need to worry about contention with each other. * Fields that Connection thread uses: * inStream - synced access and update; initialized in constructor; * referenced outside class unsync'ed (by LdapSasl) only * when connection is quiet * traceFile, traceTagIn, traceTagOut - no sync; debugging only * parent - no sync; initialized in constructor; no updates * pendingRequests - sync * pauseLock - per-instance lock; * paused - sync via pauseLock (pauseReader()) * Members used by main threads (LdapClient): * host, port - unsync; read-only access for StartTLS and debug messages * setBound(), setV3() - no sync; called only by LdapClient.authenticate(), * which is a sync method called only when connection is "quiet" * getMsgId() - sync * writeRequest(), removeRequest(),findRequest(), abandonOutstandingReqs() - * access to shared pendingRequests is sync * writeRequest(), abandonRequest(), ldapUnbind() - access to outStream sync * cleanup() - sync * readReply() - access to sock sync * unpauseReader() - (indirectly via writeRequest) sync on pauseLock * Members used by SASL auth (main thread): * inStream, outStream - no sync; used to construct new stream; accessed * only when conn is "quiet" and not shared * replaceStreams() - sync method * Members used by StartTLS: * inStream, outStream - no sync; used to record the existing streams; * accessed only when conn is "quiet" and not shared * replaceStreams() - sync method *
* Handles anonymous, simple, and SASL bind for v3; anonymous and simple
* for v2.
* %%% made public for access by LdapSasl %%%
*
* @author Vincent Ryan
* @author Rosanna Lee
* @author Jagane Sundar
*/
public final class Connection implements Runnable {
private static final boolean debug = false;
private static final int dump = 0; // > 0 r, > 1 rw
final private Thread worker; // Initialized in constructor
private boolean v3 = true; // Set in setV3()
final public String host; // used by LdapClient for generating exception messages
// used by StartTlsResponse when creating an SSL socket
final public int port; // used by LdapClient for generating exception messages
// used by StartTlsResponse when creating an SSL socket
private boolean bound = false; // Set in setBound()
// All three are initialized in constructor and read-only afterwards
private OutputStream traceFile = null;
private String traceTagIn = null;
private String traceTagOut = null;
// Initialized in constructor; read and used externally (LdapSasl);
// Updated in replaceStreams() during "quiet", unshared, period
public InputStream inStream; // must be public; used by LdapSasl
// Initialized in constructor; read and used externally (LdapSasl);
// Updated in replaceOutputStream() during "quiet", unshared, period
public OutputStream outStream; // must be public; used by LdapSasl
// Initialized in constructor; read and used externally (TLS) to
// get new IO streams; closed during cleanup
public Socket sock; // for TLS
// For processing "disconnect" unsolicited notification
// Initialized in constructor
final private LdapClient parent;
// Incremented and returned in sync getMsgId()
private int outMsgId = 0;
//
// The list of ldapRequests pending on this binding
//
// Accessed only within sync methods
private LdapRequest pendingRequests = null;
volatile IOException closureReason = null;
volatile boolean useable = true; // is Connection still useable
int readTimeout;
int connectTimeout;
private static final boolean IS_HOSTNAME_VERIFICATION_DISABLED
= hostnameVerificationDisabledValue();
private static boolean hostnameVerificationDisabledValue() {
PrivilegedAction
* For dealing with StartTLS close,
* when the read() exits either due to EOF or an exception,
* the reader thread checks whether there is a new stream to read from.
* If so, then it reattempts the read. Otherwise, the EOF or exception
* is processed and the reader thread terminates.
* In a StartTLS close, the client first replaces the SSL IO streams with
* plain ones and then closes the SSL socket.
* If the reader thread attempts to read, or was reading, from
* the SSL socket (that is, it got to the read BEFORE replaceStreams()),
* the SSL socket close will cause the reader thread to
* get an EOF/exception and reexamine the input stream.
* If the reader thread sees a new stream, it reattempts the read.
* If the underlying socket is still alive, then the new read will succeed.
* If the underlying socket has been closed also, then the new read will
* fail and the reader thread exits.
* If the reader thread attempts to read, or was reading, from the plain
* socket (that is, it got to the read AFTER replaceStreams()), the
* SSL socket close will have no effect on the reader thread.
*
* The check for new stream is made only
* in the first attempt at reading a BER buffer; the reader should
* never be in midst of reading a buffer when a nonfatal close occurs.
* If this occurs, then the connection is in an inconsistent state and
* the safest thing to do is to shut it down.
*/
private final Object pauseLock = new Object(); // lock for reader to wait on while paused
private boolean paused = false; // paused state of reader
/*
* Unpauses reader thread if it was paused
*/
private void unpauseReader() throws IOException {
synchronized (pauseLock) {
if (paused) {
if (debug) {
System.err.println("Unpausing reader; read from: " +
inStream);
}
paused = false;
pauseLock.notify();
}
}
}
/*
* Pauses reader so that it stops reading from the input stream.
* Reader blocks on pauseLock instead of read().
* MUST be called from within synchronized (pauseLock) clause.
*/
private void pauseReader() throws IOException {
if (debug) {
System.err.println("Pausing reader; was reading from: " +
inStream);
}
paused = true;
try {
while (paused) {
pauseLock.wait(); // notified by unpauseReader
}
} catch (InterruptedException e) {
throw new InterruptedIOException(
"Pause/unpause reader has problems.");
}
}
////////////////////////////////////////////////////////////////////////////
//
// The LDAP Binding thread. It does the mux/demux of multiple requests
// on the same TCP connection.
//
////////////////////////////////////////////////////////////////////////////
public void run() {
byte inbuf[]; // Buffer for reading incoming bytes
int inMsgId; // Message id of incoming response
int bytesread; // Number of bytes in inbuf
int br; // Temp; number of bytes read from stream
int offset; // Offset of where to store bytes in inbuf
int seqlen; // Length of ASN sequence
int seqlenlen; // Number of sequence length bytes
boolean eos; // End of stream
BerDecoder retBer; // Decoder for ASN.1 BER data from inbuf
InputStream in = null;
try {
while (true) {
try {
// type and length (at most 128 octets for long form)
inbuf = new byte[129];
offset = 0;
seqlen = 0;
seqlenlen = 0;
in = getInputStream();
// check that it is the beginning of a sequence
bytesread = in.read(inbuf, offset, 1);
if (bytesread < 0) {
if (in != getInputStream()) {
continue; // a new stream to try
} else {
break; // EOF
}
}
if (inbuf[offset++] != (Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR))
continue;
// get length of sequence
bytesread = in.read(inbuf, offset, 1);
if (bytesread < 0)
break; // EOF
seqlen = inbuf[offset++];
// if high bit is on, length is encoded in the
// subsequent length bytes and the number of length bytes
// is equal to & 0x80 (i.e. length byte with high bit off).
if ((seqlen & 0x80) == 0x80) {
seqlenlen = seqlen & 0x7f; // number of length bytes
bytesread = 0;
eos = false;
// Read all length bytes
while (bytesread < seqlenlen) {
br = in.read(inbuf, offset+bytesread,
seqlenlen-bytesread);
if (br < 0) {
eos = true;
break; // EOF
}
bytesread += br;
}
// end-of-stream reached before length bytes are read
if (eos)
break; // EOF
// Add contents of length bytes to determine length
seqlen = 0;
for( int i = 0; i < seqlenlen; i++) {
seqlen = (seqlen << 8) + (inbuf[offset+i] & 0xff);
}
offset += bytesread;
}
// read in seqlen bytes
byte[] left = readFully(in, seqlen);
inbuf = Arrays.copyOf(inbuf, offset + left.length);
System.arraycopy(left, 0, inbuf, offset, left.length);
offset += left.length;
/*
if (dump > 0) {
System.err.println("seqlen: " + seqlen);
System.err.println("bufsize: " + offset);
System.err.println("bytesleft: " + bytesleft);
System.err.println("bytesread: " + bytesread);
}
*/
try {
retBer = new BerDecoder(inbuf, 0, offset);
if (traceFile != null) {
Ber.dumpBER(traceFile, traceTagIn, inbuf, 0, offset);
}
retBer.parseSeq(null);
inMsgId = retBer.parseInt();
retBer.reset(); // reset offset
boolean needPause = false;
if (inMsgId == 0) {
// Unsolicited Notification
parent.processUnsolicited(retBer);
} else {
LdapRequest ldr = findRequest(inMsgId);
if (ldr != null) {
/**
* Grab pauseLock before making reply available
* to ensure that reader goes into paused state
* before writer can attempt to unpause reader
*/
synchronized (pauseLock) {
needPause = ldr.addReplyBer(retBer);
if (needPause) {
/*
* Go into paused state; release
* pauseLock
*/
pauseReader();
}
// else release pauseLock
}
} else {
// System.err.println("Cannot find" +
// "LdapRequest for " + inMsgId);
}
}
} catch (Ber.DecodeException e) {
//System.err.println("Cannot parse Ber");
}
} catch (IOException ie) {
if (debug) {
System.err.println("Connection: Inside Caught " + ie);
ie.printStackTrace();
}
if (in != getInputStream()) {
// A new stream to try
// Go to top of loop and continue
} else {
if (debug) {
System.err.println("Connection: rethrowing " + ie);
}
throw ie; // rethrow exception
}
}
}
if (debug) {
System.err.println("Connection: end-of-stream detected: "
+ in);
}
} catch (IOException ex) {
if (debug) {
System.err.println("Connection: Caught " + ex);
}
closureReason = ex;
} finally {
cleanup(null, true); // cleanup
}
if (debug) {
System.err.println("Connection: Thread Exiting");
}
}
private static byte[] readFully(InputStream is, int length)
throws IOException
{
byte[] buf = new byte[Math.min(length, 8192)];
int nread = 0;
while (nread < length) {
int bytesToRead;
if (nread >= buf.length) { // need to allocate a larger buffer
bytesToRead = Math.min(length - nread, buf.length + 8192);
if (buf.length < nread + bytesToRead) {
buf = Arrays.copyOf(buf, nread + bytesToRead);
}
} else {
bytesToRead = buf.length - nread;
}
int count = is.read(buf, nread, bytesToRead);
if (count < 0) {
if (buf.length != nread)
buf = Arrays.copyOf(buf, nread);
break;
}
nread += count;
}
return buf;
}
// This code must be uncommented to run the LdapAbandonTest.
/*public void sendSearchReqs(String dn, int numReqs) {
int i;
String attrs[] = null;
for(i = 1; i <= numReqs; i++) {
BerEncoder ber = new BerEncoder(2048);
try {
ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
ber.encodeInt(i);
ber.beginSeq(LdapClient.LDAP_REQ_SEARCH);
ber.encodeString(dn == null ? "" : dn);
ber.encodeInt(0, LdapClient.LBER_ENUMERATED);
ber.encodeInt(3, LdapClient.LBER_ENUMERATED);
ber.encodeInt(0);
ber.encodeInt(0);
ber.encodeBoolean(true);
LdapClient.encodeFilter(ber, "");
ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
ber.encodeStringArray(attrs);
ber.endSeq();
ber.endSeq();
ber.endSeq();
writeRequest(ber, i);
//System.err.println("wrote request " + i);
} catch (Exception ex) {
//System.err.println("ldap.search: Caught " + ex + " building req");
}
}
} */
}