389-ds-base/ldap/servers/plugins/replication/repl5_connection.c

/** BEGIN COPYRIGHT BLOCK
 * This Program is free software; you can redistribute it and/or modify it under
 * the terms of the GNU General Public License as published by the Free Software
 * Foundation; version 2 of the License.
 * 
 * This Program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License along with
 * this Program; if not, write to the Free Software Foundation, Inc., 59 Temple
 * Place, Suite 330, Boston, MA 02111-1307 USA.
 * 
 * In addition, as a special exception, Red Hat, Inc. gives You the additional
 * right to link the code of this Program with code not covered under the GNU
 * General Public License ("Non-GPL Code") and to distribute linked combinations
 * including the two, subject to the limitations in this paragraph. Non-GPL Code
 * permitted under this exception must only link to the code of this Program
 * through those well defined interfaces identified in the file named EXCEPTION
 * found in the source code files (the "Approved Interfaces"). The files of
 * Non-GPL Code may instantiate templates or use macros or inline functions from
 * the Approved Interfaces without causing the resulting work to be covered by
 * the GNU General Public License. Only Red Hat, Inc. may make changes or
 * additions to the list of Approved Interfaces. You must obey the GNU General
 * Public License in all respects for all of the Program code and other code used
 * in conjunction with the Program except the Non-GPL Code covered by this
 * exception. If you modify this file, you may extend this exception to your
 * version of the file, but you are not obligated to do so. If you do not wish to
 * provide this exception without modification, you must delete this exception
 * statement from your version and license this file solely under the GPL without
 * exception. 
 * 
 * 
 * Copyright (C) 2001 Sun Microsystems, Inc. Used by permission.
 * Copyright (C) 2005 Red Hat, Inc.
 * All rights reserved.
 * END COPYRIGHT BLOCK **/

#ifdef HAVE_CONFIG_H
#  include <config.h>
#endif


/* repl5_connection.c */
/*

 The connection object manages a connection to a single replication
 consumer.

XXXggood what to do on timeout? If we close connection, then we won't leave a
replica locked. Seems like right thing to do.
*/

#include "repl5.h"
#include "ldappr.h"
#include "ldap-extension.h"

typedef struct repl_connection
{
	char *hostname;
	int port;
	char *binddn;
	int bindmethod;
	int state;
	int last_operation;
	int last_ldap_error;
	const char *status;
	char *last_ldap_errmsg;
	PRUint32 transport_flags;
	LDAP *ld;
	int supports_ldapv3; /* 1 if does, 0 if doesn't, -1 if not determined */
	int supports_ds50_repl; /* 1 if does, 0 if doesn't, -1 if not determined */
	int supports_ds40_repl; /* 1 if does, 0 if doesn't, -1 if not determined */
	int supports_ds71_repl; /* 1 if does, 0 if doesn't, -1 if not determined */
	int linger_time; /* time in seconds to leave an idle connection open */
	PRBool linger_active;
	Slapi_Eq_Context *linger_event;
	PRBool delete_after_linger;
	int refcnt;
	const Repl_Agmt *agmt;
	PRLock *lock;
	struct timeval timeout;
	int flag_agmt_changed;
	char *plain;
} repl_connection;

/* #define DEFAULT_LINGER_TIME (5 * 60) */ /* 5 minutes */
#define DEFAULT_LINGER_TIME (60) 

/* Controls we add on every outbound operation */

static LDAPControl manageDSAITControl = {LDAP_CONTROL_MANAGEDSAIT, {0, ""}, '\0'};
static int attribute_string_value_present(LDAP *ld, LDAPMessage *entry,
	const char *type, const char *value);
static int bind_and_check_pwp(Repl_Connection *conn, char * binddn, char *password);
static int do_simple_bind (Repl_Connection *conn, LDAP *ld, char * binddn, char *password);

static int s_debug_timeout = 0;
static int s_debug_level = 0;
static Slapi_Eq_Context repl5_start_debug_timeout(int *setlevel);
static void repl5_stop_debug_timeout(Slapi_Eq_Context eqctx, int *setlevel);
static void repl5_debug_timeout_callback(time_t when, void *arg);

#define STATE_CONNECTED 600
#define STATE_DISCONNECTED 601

#define STATUS_DISCONNECTED "disconnected"
#define STATUS_CONNECTED "connected"
#define STATUS_PROCESSING_ADD "processing add operation"
#define STATUS_PROCESSING_DELETE "processing delete operation"
#define STATUS_PROCESSING_MODIFY "processing modify operation"
#define STATUS_PROCESSING_RENAME "processing rename operation"
#define STATUS_PROCESSING_EXTENDED_OPERATION "processing extended operation"
#define STATUS_LINGERING "lingering"
#define STATUS_SHUTTING_DOWN "shutting down"
#define STATUS_BINDING "connecting and binding"
#define STATUS_SEARCHING "processing search operation"

#define CONN_NO_OPERATION 0
#define CONN_ADD 1
#define CONN_DELETE 2
#define CONN_MODIFY 3
#define CONN_RENAME 4
#define CONN_EXTENDED_OPERATION 5
#define CONN_BIND 6
#define CONN_INIT 7

/* These are errors returned from ldap operations which should cause us to disconnect and
   retry the connection later */
#define IS_DISCONNECT_ERROR(rc) (rc == LDAP_SERVER_DOWN || rc == LDAP_CONNECT_ERROR || rc == LDAP_INVALID_CREDENTIALS || rc == LDAP_INAPPROPRIATE_AUTH || rc == LDAP_LOCAL_ERROR)

/* Forward declarations */
static void close_connection_internal(Repl_Connection *conn);

/*
 * Create a new conenction object. Returns a pointer to the object, or
 * NULL if an error occurs.
 */
Repl_Connection *
conn_new(Repl_Agmt *agmt)
{
	Repl_Connection *rpc;

	rpc = (Repl_Connection *)slapi_ch_malloc(sizeof(repl_connection));
	if ((rpc->lock = PR_NewLock()) == NULL)
	{
		goto loser;
	}
	rpc->hostname = agmt_get_hostname(agmt);
	rpc->port = agmt_get_port(agmt);
	rpc->binddn = agmt_get_binddn(agmt);
	rpc->bindmethod = agmt_get_bindmethod(agmt);
	rpc->transport_flags = agmt_get_transport_flags(agmt);
	rpc->ld = NULL;
	rpc->state = STATE_DISCONNECTED;
	rpc->last_operation = CONN_NO_OPERATION;
	rpc->last_ldap_error = LDAP_SUCCESS;
	rpc->last_ldap_errmsg = NULL;
	rpc->supports_ldapv3 = -1;
	rpc->supports_ds40_repl = -1;
	rpc->supports_ds50_repl = -1;
	rpc->supports_ds71_repl = -1;

	rpc->linger_active = PR_FALSE;
	rpc->delete_after_linger = PR_FALSE;
	rpc->linger_event = NULL;
	rpc->linger_time = DEFAULT_LINGER_TIME;
	rpc->status = STATUS_DISCONNECTED;
	rpc->agmt = agmt;
	rpc->refcnt = 1;
	rpc->timeout.tv_sec = agmt_get_timeout(agmt);
	rpc->timeout.tv_usec = 0;
	rpc->flag_agmt_changed = 0;
	rpc->plain = NULL;
	return rpc;
loser:
	conn_delete(rpc);
	return NULL;
}


/*
 * Return PR_TRUE if the connection is in the connected state
 */
static PRBool
conn_connected(Repl_Connection *conn)
{
	PRBool return_value;
	PR_Lock(conn->lock);
	return_value = STATE_CONNECTED == conn->state;
	PR_Unlock(conn->lock);
	return return_value;
}


/*
 * Destroy a connection object.
 */
static void
conn_delete_internal(Repl_Connection *conn)
{
	PR_ASSERT(NULL != conn);
	close_connection_internal(conn);
	/* slapi_ch_free accepts NULL pointer */
	slapi_ch_free((void **)&conn->hostname);
	slapi_ch_free((void **)&conn->binddn);
	slapi_ch_free((void **)&conn->plain);
}

/*
 * Destroy a connection. It is an error to use the connection object
 * after conn_delete() has been called.
 */
void
conn_delete(Repl_Connection *conn)
{
	PRBool destroy_it = PR_FALSE;

	PR_ASSERT(NULL != conn);
	PR_Lock(conn->lock);
	if (conn->linger_active)
	{
		if (slapi_eq_cancel(conn->linger_event) == 1)
		{
			/* Event was found and cancelled. Destroy the connection object. */
			PR_Unlock(conn->lock);
			destroy_it = PR_TRUE;
		}
		else
		{
			/*
			 * The event wasn't found, but we think it's still active.
			 * That means an event is in the process of being fired
			 * off, so arrange for the event to destroy the object .
			 */
			conn->delete_after_linger = PR_TRUE;
			PR_Unlock(conn->lock);
		}
	}
	if (destroy_it)
	{
		conn_delete_internal(conn);
	}
}


/*
 * Return the last operation type processed by the connection
 * object, and the LDAP error encountered.
 */
void
conn_get_error(Repl_Connection *conn, int *operation, int *error)
{
	PR_Lock(conn->lock);
	*operation = conn->last_operation;
	*error = conn->last_ldap_error;
	PR_Unlock(conn->lock);
}

/*
 * Return the last operation type processed by the connection
 * object, and the LDAP error encountered.
 * Beware that the error string will only be in scope and valid
 * before the next operation result has been read from the connection
 * (so don't alias the pointer).
 */
void
conn_get_error_ex(Repl_Connection *conn, int *operation, int *error, char **error_string)
{
	PR_Lock(conn->lock);
	*operation = conn->last_operation;
	*error = conn->last_ldap_error;
	*error_string = conn->last_ldap_errmsg;
	PR_Unlock(conn->lock);
}

/* Returns the result (asyncronously) from an opertation and also returns that operations message ID */
/* The _ex version handles a bunch of parameters (retoidp et al) that were present in the original
 * sync operation functions, but were never actually used) */
ConnResult
conn_read_result_ex(Repl_Connection *conn, char **retoidp, struct berval **retdatap, LDAPControl ***returned_controls, int *message_id, int block)
{
			LDAPMessage *res = NULL;
			int setlevel = 0;
			int rc = 0;
			int return_value = 0;
			LDAPControl **loc_returned_controls = NULL;
			struct timeval local_timeout = {0};
			time_t time_now = 0;
			time_t start_time = time( NULL );
			int backoff_time = 1;
			Slapi_Eq_Context eqctx = repl5_start_debug_timeout(&setlevel);

			/* Here, we want to not block inside ldap_result().
			 * Reason is that blocking there will deadlock with a
			 * concurrent sender. We send concurrently, and hence
			 * blocking is not good : deadlock results.
			 * So, instead, we call ldap_result() with a zero timeout.
			 * This makes it do a non-blocking poll and return to us
			 * if there's no data to read.
			 * We can then handle our timeout here by sleeping and re-trying.
			 * In order that we do pickup results reasonably quickly,
			 * we implement a backoff algorithm for the sleep: if we
			 * keep getting results quickly then we won't spend much time sleeping.
			 */

			while (1) 
			{
				rc = ldap_result(conn->ld, LDAP_RES_ANY , 1, &local_timeout, &res);
				if (0 != rc)
				{
					/* Something other than a timeout happened */
					break;
				}
				if (block)
				{
					/* Did the connection's timeout expire ? */
					time_now = time( NULL );
					if (conn->timeout.tv_sec <= ( time_now - start_time ))
					{
						/* We timed out */
						rc = 0;
						break;
					}
					/* Otherwise we backoff */
					DS_Sleep(PR_MillisecondsToInterval(backoff_time));
					if (backoff_time < 1000) 
					{
						backoff_time <<= 1;
					}
				} else
				{
					rc = 0;
					break;
				}
			}

			repl5_stop_debug_timeout(eqctx, &setlevel);

			if (0 == rc)
			{
				/* Timeout */
				rc = ldap_get_lderrno(conn->ld, NULL, NULL);
				conn->last_ldap_error = LDAP_TIMEOUT;
				return_value = CONN_TIMEOUT;
			}
			else if (-1 == rc)
			{
				/* Error */
				char *s = NULL;
		
				rc = ldap_get_lderrno(conn->ld, NULL, &s);
				conn->last_ldap_errmsg = s;
				conn->last_ldap_error = rc;
				/* some errors will require a disconnect and retry the connection
				   later */
				if (IS_DISCONNECT_ERROR(rc))
				{
					conn_disconnect(conn);
					return_value = CONN_NOT_CONNECTED;
				}
				else
				{
					conn->status = STATUS_CONNECTED;
					return_value = CONN_OPERATION_FAILED;
				}
			}
			else
			{
				int err;
				char *errmsg = NULL;
				char **referrals = NULL;
				char *matched = NULL;

				if (message_id) 
				{
					*message_id = ldap_msgid(res);
				}

				rc = ldap_parse_result(conn->ld, res, &err, &matched,
									   &errmsg, &referrals, &loc_returned_controls,
									   0 /* Don't free the result */);
				if (IS_DISCONNECT_ERROR(rc))
				{
					conn->last_ldap_error = rc;
					conn_disconnect(conn);
					return_value = CONN_NOT_CONNECTED;
				}
				else if (IS_DISCONNECT_ERROR(err))
				{
					conn->last_ldap_error = err;
					conn_disconnect(conn);
					return_value = CONN_NOT_CONNECTED;
				}
				/* Got a result */
				if ((rc == LDAP_SUCCESS) && (err == LDAP_BUSY))
				    	return_value = CONN_BUSY;
				else if (retoidp)
				{
					if (!((rc == LDAP_SUCCESS) && (err == LDAP_BUSY)))
					{
						if (rc == LDAP_SUCCESS) {
							rc = ldap_parse_extended_result(conn->ld, res, retoidp,
								retdatap, 0 /* Don't Free it */);
						} 
						conn->last_ldap_error = rc;
						return_value = (LDAP_SUCCESS == conn->last_ldap_error ? 
										CONN_OPERATION_SUCCESS : CONN_OPERATION_FAILED);
					}
				}
				else /* regular operation, result returned */
				{
					if (NULL != returned_controls)
					{
						*returned_controls = loc_returned_controls;
					}
					if (LDAP_SUCCESS != rc)
					{
						conn->last_ldap_error = rc;
					}
					else
					{
						conn->last_ldap_error = err;
					}
					return_value = LDAP_SUCCESS == conn->last_ldap_error ? CONN_OPERATION_SUCCESS : CONN_OPERATION_FAILED;
				}
				/*
				 * XXXggood do I need to free matched, referrals,
				 * anything else? Or can I pass NULL for the args
				 * I'm not interested in?
				 */
				/* Good question! Meanwhile, as RTM aproaches, let's free them... */
				slapi_ch_free((void **) &errmsg);
				slapi_ch_free((void **) &matched);
				charray_free(referrals);
				conn->status = STATUS_CONNECTED;
			}
			if (res) ldap_msgfree(res);
			return return_value;
}

ConnResult
conn_read_result(Repl_Connection *conn, int *message_id)
{
	return conn_read_result_ex(conn,NULL,NULL,NULL,message_id,1);
}

/*
 * Common code to send an LDAPv3 operation and collect the result.
 * Return values:
 * CONN_OPERATION_SUCCESS - the operation succeeded
 * CONN_OPERATION_FAILED - the operation was sent to the consumer
 * and failed. Use conn_get_error() to determine the LDAP error
 * code.
 * CONN_NOT_CONNECTED - no connection is active. The caller should
 * use conn_connect() to connect to the replica and bind, then should
 * reacquire the replica (if needed).
 * CONN_BUSY - the server is busy with previous requests, must wait for a while 
 * before retrying
 * DBDB: also returns the operation's message ID, if it was successfully sent, now that
 * we're reading results async.
 */
static ConnResult
perform_operation(Repl_Connection *conn, int optype, const char *dn,
	LDAPMod **attrs, const char *newrdn, const char *newparent,
	int deleteoldrdn, LDAPControl *update_control,
	const char *extop_oid, struct berval *extop_payload, int *message_id)
{
	int rc;
	ConnResult return_value = CONN_OPERATION_FAILED;
	LDAPControl *server_controls[3];
	/* LDAPControl **loc_returned_controls; */
	const char *op_string = NULL;
	const char *extra_op_string = NULL;
	int msgid = 0;

	server_controls[0] = &manageDSAITControl;
	server_controls[1] = update_control;
	server_controls[2] = NULL;

	if (conn_connected(conn))
	{
		int setlevel = 0;

		Slapi_Eq_Context eqctx = repl5_start_debug_timeout(&setlevel);

		/* Because the SDK isn't really thread-safe (it can deadlock between
		 * a thread sending an operation and a thread trying to retrieve a response
		 * on the same connection), we need to _first_ verify that the connection
		 * is writable. If it isn't, we can deadlock if we proceed any further...
		 */
		{
				struct PRPollDesc	pr_pd;
				PRIntervalTime	timeout = PR_SecondsToInterval(conn->timeout.tv_sec);
				PRFileDesc *prfd = NULL;
				struct lextiof_socket_private *socketargp = NULL;
				PRLDAPSocketInfo    soi;

				if ( (rc = ldap_get_option(conn->ld, LDAP_X_OPT_SOCKETARG, &socketargp)) != LDAP_SUCCESS)
				{
					slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
					"%s: Failed call to ldap_get_option in perform_operation: LDAP error %d (%s)\n",
					agmt_get_long_name(conn->agmt),
					op_string ? op_string : "NULL", rc, ldap_err2string(rc));
					conn->last_ldap_error = rc;
     				return CONN_OPERATION_FAILED;
				}
				memset( &soi, 0, sizeof(soi));
				soi.soinfo_size = PRLDAP_SOCKETINFO_SIZE;
				if (LDAP_SUCCESS != (rc = prldap_get_socket_info(0, socketargp, &soi))) 
				{
					slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
					"%s: Failed call to prldap_get_socket_info in perform_operation: LDAP error %d (%s)\n",
					agmt_get_long_name(conn->agmt),
					op_string ? op_string : "NULL", rc, ldap_err2string(rc));
					conn->last_ldap_error = rc;
     				return CONN_OPERATION_FAILED;
				} 
				prfd = soi.soinfo_prfd;
				/* Before we connect, the prfd can be null */
				if (prfd) 
				{
					pr_pd.fd = prfd;
					pr_pd.in_flags = PR_POLL_WRITE;
					pr_pd.out_flags = 0;
					rc = PR_Poll(&pr_pd, 1, timeout);
					/* Did we time out ? */
					if (rc == 0)
					{
						return CONN_TIMEOUT;
					}
				}
		}

		conn->last_operation = optype;
		switch (optype)
		{
		case CONN_ADD:
			conn->status = STATUS_PROCESSING_ADD;
			op_string = "add";
			rc = ldap_add_ext(conn->ld, dn, attrs, server_controls,
				NULL /* clientctls */,  &msgid);
			break;
		case CONN_MODIFY:
			conn->status = STATUS_PROCESSING_MODIFY;
			op_string = "modify";
			rc = ldap_modify_ext(conn->ld, dn, attrs, server_controls,
				NULL /* clientctls */,  &msgid);
			break;
		case CONN_DELETE:
			conn->status = STATUS_PROCESSING_DELETE;
			op_string = "delete";
			rc = ldap_delete_ext(conn->ld, dn, server_controls,
				NULL /* clientctls */, &msgid);
			break;
		case CONN_RENAME:
			conn->status = STATUS_PROCESSING_RENAME;
			op_string = "rename";
			rc = ldap_rename(conn->ld, dn, newrdn, newparent, deleteoldrdn,
				server_controls, NULL /* clientctls */, &msgid);
			break;
		case CONN_EXTENDED_OPERATION:
			conn->status = STATUS_PROCESSING_EXTENDED_OPERATION;
			op_string = "extended";
			extra_op_string = extop_oid;
			rc = ldap_extended_operation(conn->ld, extop_oid, extop_payload,
				server_controls, NULL /* clientctls */, &msgid);
		}
		repl5_stop_debug_timeout(eqctx, &setlevel);
		if (LDAP_SUCCESS == rc)
		{
			/* DBDB: The code that used to be here has been moved for async operation 
			 * Results are now picked up in another thread. All we need to do here is
			 * queue the operation details in the outstanding operation list.
			 */
			return_value = CONN_OPERATION_SUCCESS;
		}
		else
		{
            slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
					"%s: Failed to send %s operation: LDAP error %d (%s)\n",
					agmt_get_long_name(conn->agmt),
					op_string ? op_string : "NULL", rc, ldap_err2string(rc));
			conn->last_ldap_error = rc;
			if (IS_DISCONNECT_ERROR(rc))
			{
				conn_disconnect(conn);
				return_value = CONN_NOT_CONNECTED;
			}
			else
			{
				conn->status = STATUS_CONNECTED;
				return_value = CONN_OPERATION_FAILED;
			}
		}
	}
	else
	{
		/* conn->last_ldap_error has been set to a more specific value
		 * in conn_connected()
		 * conn->last_ldap_error = LDAP_SERVER_DOWN;
		 */
		return_value = CONN_NOT_CONNECTED;
	}
	if (message_id)
	{
		*message_id = msgid;
	}
	return return_value;
}

/*
 * Send an LDAP add operation.
 */
ConnResult
conn_send_add(Repl_Connection *conn, const char *dn, LDAPMod **attrs,
	LDAPControl *update_control, int *message_id)
{
	return perform_operation(conn, CONN_ADD, dn, attrs, NULL /* newrdn */,
		NULL /* newparent */, 0 /* deleteoldrdn */, update_control,
		NULL /* extop OID */, NULL /* extop payload */, message_id);
}


/*
 * Send an LDAP delete operation.
 */
ConnResult
conn_send_delete(Repl_Connection *conn, const char *dn,
	LDAPControl *update_control, int *message_id)
{
	return perform_operation(conn, CONN_DELETE, dn, NULL /* attrs */,
		NULL /* newrdn */, NULL /* newparent */, 0 /* deleteoldrdn */,
		update_control, NULL /* extop OID */, NULL /* extop payload */, message_id);
}


/*
 * Send an LDAP modify operation.
 */
ConnResult
conn_send_modify(Repl_Connection *conn, const char *dn, LDAPMod **mods,
	LDAPControl *update_control, int *message_id)
{
	return perform_operation(conn, CONN_MODIFY, dn, mods, NULL /* newrdn */,
		NULL /* newparent */, 0 /* deleteoldrdn */, update_control,
		NULL /* extop OID */, NULL /* extop payload */, message_id);
}

/*
 * Send an LDAP moddn operation.
 */
ConnResult
conn_send_rename(Repl_Connection *conn, const char *dn,
	const char *newrdn, const char *newparent, int deleteoldrdn,
	LDAPControl *update_control, int *message_id)
{
	return perform_operation(conn, CONN_RENAME, dn, NULL /* attrs */,
		newrdn, newparent, deleteoldrdn, update_control,
		NULL /* extop OID */, NULL /* extop payload */, message_id);
}


/*
 * Send an LDAP extended operation.
 */
ConnResult
conn_send_extended_operation(Repl_Connection *conn, const char *extop_oid,
	struct berval *payload,
	LDAPControl *update_control, int *message_id)
{
	return perform_operation(conn, CONN_EXTENDED_OPERATION, NULL /* dn */, NULL /* attrs */,
		NULL /* newrdn */, NULL /* newparent */,  0 /* deleteoldrdn */,
		update_control, extop_oid, payload, message_id);
}


/*
 * Synchronously read an entry and return a specific attribute's values.
 * Returns CONN_OPERATION_SUCCESS if successful. Returns
 * CONN_OPERATION_FAILED if the operation was sent but an LDAP error
 * occurred (conn->last_ldap_error is set in this case), and
 * CONN_NOT_CONNECTED if no connection was active.
 *
 * The caller must free the returned_bvals.
 */
ConnResult
conn_read_entry_attribute(Repl_Connection *conn, const char *dn,
	char *type, struct berval ***returned_bvals)
{
	ConnResult return_value;
	int ldap_rc;
	LDAPControl *server_controls[2];
	LDAPMessage *res = NULL;
	char *attrs[2];

	PR_ASSERT(NULL != type);
	if (conn_connected(conn))
	{
		server_controls[0] = &manageDSAITControl;
		server_controls[1] = NULL;
		attrs[0] = type;
		attrs[1] = NULL;
		ldap_rc = ldap_search_ext_s(conn->ld, dn, LDAP_SCOPE_BASE,
			"(objectclass=*)", attrs, 0 /* attrsonly */,
			server_controls, NULL /* client controls */,
			&conn->timeout, 0 /* sizelimit */, &res);
		if (LDAP_SUCCESS == ldap_rc)
		{
			LDAPMessage *entry = ldap_first_entry(conn->ld, res);
			if (NULL != entry)
			{
				*returned_bvals = ldap_get_values_len(conn->ld, entry, type);
			}
			return_value = CONN_OPERATION_SUCCESS;
		}
		else if (IS_DISCONNECT_ERROR(ldap_rc))
		{
			conn_disconnect(conn);
			return_value = CONN_NOT_CONNECTED;
		}
		else
		{
			return_value = CONN_OPERATION_FAILED;
		}
		conn->last_ldap_error = ldap_rc;
		if (NULL != res)
		{
			ldap_msgfree(res);
			res = NULL;
		}
	}
	else
	{
		return_value = CONN_NOT_CONNECTED;
	}
	return return_value;
}


/*
 * Return an pointer to a string describing the connection's status.
*/

const char *
conn_get_status(Repl_Connection *conn)
{
	return conn->status;
}



/*
 * Cancel any outstanding linger timer. Should be called when
 * a replication session is beginning.
 */
void
conn_cancel_linger(Repl_Connection *conn)
{
	PR_ASSERT(NULL != conn);
	PR_Lock(conn->lock);
	if (conn->linger_active)
	{
		slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
			"%s: Cancelling linger on the connection\n",
			agmt_get_long_name(conn->agmt));
		conn->linger_active = PR_FALSE;
		if (slapi_eq_cancel(conn->linger_event) == 1)
		{
			conn->refcnt--;
		}
		conn->linger_event = NULL;
		conn->status = STATUS_CONNECTED;
	}
	else
	{
		slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
			"%s: No linger to cancel on the connection\n",
			agmt_get_long_name(conn->agmt));
	}
	PR_Unlock(conn->lock);
}


/*
 * Called when our linger timeout timer expires. This means
 * we should check to see if perhaps the connection's become
 * active again, in which case we do nothing. Otherwise,
 * we close the connection.
 */
static void
linger_timeout(time_t event_time, void *arg)
{
	PRBool delete_now;
	Repl_Connection *conn = (Repl_Connection *)arg;

	PR_ASSERT(NULL != conn);
	slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
		"%s: Linger timeout has expired on the connection\n",
		agmt_get_long_name(conn->agmt));
	PR_Lock(conn->lock);
	if (conn->linger_active)
	{
		conn->linger_active = PR_FALSE;
		conn->linger_event = NULL;
		close_connection_internal(conn);
	}
	delete_now = conn->delete_after_linger;
	PR_Unlock(conn->lock);
	if (delete_now)
	{
		conn_delete_internal(conn);
	}
}


/*
 * Indicate that a session is ending. The linger timer starts when
 * this function is called.
 */
void
conn_start_linger(Repl_Connection *conn)
{
	time_t now;

	PR_ASSERT(NULL != conn);
	slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
		"%s: Beginning linger on the connection\n",
		agmt_get_long_name(conn->agmt));
	if (!conn_connected(conn))
	{
		slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
			"%s: No linger on the closed conn\n",
			agmt_get_long_name(conn->agmt));
		return;
	}
	time(&now);
	PR_Lock(conn->lock);
	if (conn->linger_active)
	{
		slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
			"%s: Linger already active on the connection\n",
			agmt_get_long_name(conn->agmt));
	}
	else
	{
		conn->linger_active = PR_TRUE;
		conn->linger_event = slapi_eq_once(linger_timeout, conn, now + conn->linger_time);
		conn->status = STATUS_LINGERING;
	}
	PR_Unlock(conn->lock);
}



/*
 * If no connection is currently active, opens a connection and binds to
 * the remote server. If a connection is open (e.g. lingering) then
 * this is a no-op.
 *
 * Returns CONN_OPERATION_SUCCESS on success, or CONN_OPERATION_FAILED
 * on failure. Sets conn->last_ldap_error and conn->last_operation;
 */
ConnResult
conn_connect(Repl_Connection *conn)
{
	int ldap_rc;
	int optdata;
	int secure = 0;
	char* binddn = NULL;
	struct berval *creds;
	ConnResult return_value = CONN_OPERATION_SUCCESS;
	int pw_ret = 1;

	/** Connection already open just return SUCCESS **/
	if(conn->state == STATE_CONNECTED) return return_value;

	PR_Lock(conn->lock);
	if (conn->flag_agmt_changed) {
		/* So far we cannot change Hostname and Port */
		/* slapi_ch_free((void **)&conn->hostname); */
		/* conn->hostname = agmt_get_hostname(conn->agmt); */
		/* conn->port = agmt_get_port(conn->agmt); */
		slapi_ch_free((void **)&conn->binddn);
		conn->binddn = agmt_get_binddn(conn->agmt);
		conn->bindmethod = agmt_get_bindmethod(conn->agmt);
		conn->transport_flags = agmt_get_transport_flags(conn->agmt);
		conn->timeout.tv_sec = agmt_get_timeout(conn->agmt);
		conn->flag_agmt_changed = 0;
		slapi_ch_free((void **)&conn->plain);
	}
	PR_Unlock(conn->lock);

	creds = agmt_get_credentials(conn->agmt);

	if (conn->plain == NULL) {

		char *plain = NULL;
		
		/* kexcoff: for reversible encryption */
		/* We need to test the return code of pw_rever_decode in order to decide
		 * if a free for plain will be needed (pw_ret == 0) or not (pw_ret != 0) */
		pw_ret = pw_rever_decode(creds->bv_val, &plain, type_nsds5ReplicaCredentials);
		/* Pb occured in decryption: stop now, binding will fail */
		if ( pw_ret == -1 )
		{
			slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
				"%s: Decoding of the credentials failed.\n",
				agmt_get_long_name(conn->agmt));
		
			return_value = CONN_OPERATION_FAILED;
			conn->last_ldap_error = LDAP_INVALID_CREDENTIALS;
			conn->state = STATE_DISCONNECTED;
			return (return_value);
		} /* Else, does not mean that the plain is correct, only means the we had no internal
		   decoding pb */
		conn->plain = slapi_ch_strdup (plain);
		if (!pw_ret) slapi_ch_free((void**)&plain);
	}
	

	/* ugaston: if SSL has been selected in the replication agreement, SSL client
	 * initialisation should be done before ever trying to open any connection at all.
	 */
	if (conn->transport_flags == TRANSPORT_FLAG_TLS)
	{
		slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
				"%s: Replication secured by StartTLS not currently supported\n",
				agmt_get_long_name(conn->agmt));
		
		return_value = CONN_OPERATION_FAILED;
		conn->last_ldap_error = LDAP_STRONG_AUTH_NOT_SUPPORTED;
		conn->state = STATE_DISCONNECTED;
	} else if(conn->transport_flags == TRANSPORT_FLAG_SSL)
	{

		/** Make sure the SSL Library has been initialized before anything else **/
		if(slapd_security_library_is_initialized() != 1)
		{
			slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
				"%s: SSL Not Initialized, Replication over SSL FAILED\n",
				agmt_get_long_name(conn->agmt));
			conn->last_ldap_error = LDAP_INAPPROPRIATE_AUTH;
			conn->last_operation = CONN_INIT;
			ber_bvfree(creds);
			creds = NULL;
			return CONN_SSL_NOT_ENABLED;
		} else
		{
			secure = 1;
		}
	}
 
	if (return_value == CONN_OPERATION_SUCCESS) {
		int io_timeout_ms;
		/* Now we initialize the LDAP Structure and set options */
		
		slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
			"%s: Trying %s slapi_ldap_init\n",
			agmt_get_long_name(conn->agmt),
			secure ? "secure" : "non-secure");
		/* shared = 1 because we will read results from a second thread */
		conn->ld = slapi_ldap_init(conn->hostname, conn->port, secure, 1);
		if (NULL == conn->ld)
		{
			return_value = CONN_OPERATION_FAILED;
			conn->state = STATE_DISCONNECTED;
			conn->last_operation = CONN_INIT;
			conn->last_ldap_error = LDAP_LOCAL_ERROR;
			slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
				"%s: Failed to establish %sconnection to the consumer\n",
				agmt_get_long_name(conn->agmt),
				secure ? "secure " : "");
			ber_bvfree(creds);
			creds = NULL;
			return return_value;
		}
		
		/* slapi_ch_strdup is OK with NULL strings */
		binddn = slapi_ch_strdup(conn->binddn);

		slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
			"%s: binddn = %s,  passwd = %s\n",
			agmt_get_long_name(conn->agmt),
			binddn?binddn:"NULL", creds->bv_val?creds->bv_val:"NULL");
		
		/* Set some options for the connection. */
		optdata = LDAP_DEREF_NEVER; /* Don't dereference aliases */
		ldap_set_option(conn->ld, LDAP_OPT_DEREF, &optdata);
		
		optdata = LDAP_VERSION3; /* We need LDAP version 3 */
		ldap_set_option(conn->ld, LDAP_OPT_PROTOCOL_VERSION, &optdata);
		
		/* Don't chase any referrals (although we shouldn't get any) */
		ldap_set_option(conn->ld, LDAP_OPT_REFERRALS, LDAP_OPT_OFF);

		/* override the default timeout with the specified timeout */
		io_timeout_ms = conn->timeout.tv_sec * 1000 + conn->timeout.tv_usec / 1000;
		prldap_set_session_option(conn->ld, NULL, PRLDAP_OPT_IO_MAX_TIMEOUT,
								  io_timeout_ms);

		/* We've got an ld. Now bind to the server. */
		conn->last_operation = CONN_BIND;
		
	}

	if ( bind_and_check_pwp(conn, binddn, conn->plain) == CONN_OPERATION_FAILED )
	{
		conn->last_ldap_error = ldap_get_lderrno (conn->ld, NULL, NULL);
		conn->state = STATE_DISCONNECTED;
		return_value = CONN_OPERATION_FAILED;
	}
	else
	{
		conn->last_ldap_error = ldap_rc = LDAP_SUCCESS;
		conn->state = STATE_CONNECTED;
		return_value = CONN_OPERATION_SUCCESS;
	}
		
	
	ber_bvfree(creds);
	creds = NULL;

	slapi_ch_free((void**)&binddn);

	if(return_value == CONN_OPERATION_FAILED)
	{
		close_connection_internal(conn);
	} else
	{
		conn->last_ldap_error = ldap_rc = LDAP_SUCCESS;
		conn->state = STATE_CONNECTED;
	}

	return return_value;
}


static void
close_connection_internal(Repl_Connection *conn)
{
	if (NULL != conn->ld)
	{
		/* Since we call slapi_ldap_init, 
		   we must call slapi_ldap_unbind */
		slapi_ldap_unbind(conn->ld);
	}
	conn->ld = NULL;
	conn->state = STATE_DISCONNECTED;
	conn->status = STATUS_DISCONNECTED;
	conn->supports_ds50_repl = -1;
	conn->supports_ds71_repl = -1;
	slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
		"%s: Disconnected from the consumer\n", agmt_get_long_name(conn->agmt));
}

void
conn_disconnect(Repl_Connection *conn)
{
	PR_ASSERT(NULL != conn);
	PR_Lock(conn->lock);
	close_connection_internal(conn);
	PR_Unlock(conn->lock);
}


/*
 * Determine if the remote replica supports DS 5.0 replication.
 * Return codes:
 * CONN_SUPPORTS_DS5_REPL - the remote replica suport DS5 replication
 * CONN_DOES_NOT_SUPPORT_DS5_REPL - the remote replica does not
 * support DS5 replication.
 * CONN_OPERATION_FAILED - it could not be determined if the remote
 * replica supports DS5 replication.
 * CONN_NOT_CONNECTED - no connection was active.
 */
ConnResult
conn_replica_supports_ds5_repl(Repl_Connection *conn)
{
	ConnResult return_value;
	int ldap_rc;

	if (conn_connected(conn))
	{
		if (conn->supports_ds50_repl == -1) {
			LDAPMessage *res = NULL;
			LDAPMessage *entry = NULL;
			char *attrs[] = {"supportedcontrol", "supportedextension", NULL};
			
			conn->status = STATUS_SEARCHING;
			ldap_rc = ldap_search_ext_s(conn->ld, "", LDAP_SCOPE_BASE,
										"(objectclass=*)", attrs, 0 /* attrsonly */,
										NULL /* server controls */, NULL /* client controls */,
										&conn->timeout, LDAP_NO_LIMIT, &res);
			if (LDAP_SUCCESS == ldap_rc)
			{
				conn->supports_ds50_repl = 0;
				entry = ldap_first_entry(conn->ld, res);
				if (!attribute_string_value_present(conn->ld, entry, "supportedcontrol", REPL_NSDS50_UPDATE_INFO_CONTROL_OID))
				{
					return_value = CONN_DOES_NOT_SUPPORT_DS5_REPL;
				}
				else if (!attribute_string_value_present(conn->ld, entry, "supportedextension", REPL_START_NSDS50_REPLICATION_REQUEST_OID))
				{
					return_value = CONN_DOES_NOT_SUPPORT_DS5_REPL;
				}
				else if (!attribute_string_value_present(conn->ld, entry, "supportedextension", REPL_END_NSDS50_REPLICATION_REQUEST_OID))
				{
					return_value = CONN_DOES_NOT_SUPPORT_DS5_REPL;
				}
				else if (!attribute_string_value_present(conn->ld, entry, "supportedextension", REPL_NSDS50_REPLICATION_ENTRY_REQUEST_OID))
				{
					return_value = CONN_DOES_NOT_SUPPORT_DS5_REPL;
				}
				else if (!attribute_string_value_present(conn->ld, entry, "supportedextension", REPL_NSDS50_REPLICATION_RESPONSE_OID))
				{
					return_value = CONN_DOES_NOT_SUPPORT_DS5_REPL;
				}
				else
				{
					conn->supports_ds50_repl = 1;
					return_value = CONN_SUPPORTS_DS5_REPL;
				}
			}
			else
			{
				if (IS_DISCONNECT_ERROR(ldap_rc))
				{
					conn->last_ldap_error = ldap_rc;	/* specific reason */
					conn_disconnect(conn);
					return_value = CONN_NOT_CONNECTED;
				}
				else
				{
					return_value = CONN_OPERATION_FAILED;
				}
			}
			if (NULL != res)
				ldap_msgfree(res);
		}
		else {
			return_value = conn->supports_ds50_repl ? CONN_SUPPORTS_DS5_REPL : CONN_DOES_NOT_SUPPORT_DS5_REPL;
		}
	}
	else
	{
		/* Not connected */
		return_value = CONN_NOT_CONNECTED;
	}
	return return_value;
}


/*
 * Determine if the remote replica supports DS 5.0 replication.
 * Return codes:
 * CONN_SUPPORTS_DS71_REPL - the remote replica suport DS5 replication
 * CONN_DOES_NOT_SUPPORT_DS71_REPL - the remote replica does not
 * support DS5 replication.
 * CONN_OPERATION_FAILED - it could not be determined if the remote
 * replica supports DS5 replication.
 * CONN_NOT_CONNECTED - no connection was active.
 */
ConnResult
conn_replica_supports_ds71_repl(Repl_Connection *conn)
{
	ConnResult return_value;
	int ldap_rc;

	if (conn_connected(conn))
	{
		if (conn->supports_ds71_repl == -1) {
			LDAPMessage *res = NULL;
			LDAPMessage *entry = NULL;
			char *attrs[] = {"supportedcontrol", "supportedextension", NULL};
			
			conn->status = STATUS_SEARCHING;
			ldap_rc = ldap_search_ext_s(conn->ld, "", LDAP_SCOPE_BASE,
										"(objectclass=*)", attrs, 0 /* attrsonly */,
										NULL /* server controls */, NULL /* client controls */,
										&conn->timeout, LDAP_NO_LIMIT, &res);
			if (LDAP_SUCCESS == ldap_rc)
			{
				conn->supports_ds71_repl = 0;
				entry = ldap_first_entry(conn->ld, res);
				if (!attribute_string_value_present(conn->ld, entry, "supportedextension", REPL_NSDS71_REPLICATION_ENTRY_REQUEST_OID))
				{
					return_value = CONN_DOES_NOT_SUPPORT_DS71_REPL;
				}
				else
				{
					conn->supports_ds71_repl = 1;
					return_value = CONN_SUPPORTS_DS71_REPL;
				}
			}
			else
			{
				if (IS_DISCONNECT_ERROR(ldap_rc))
				{
					conn->last_ldap_error = ldap_rc;	/* specific reason */
					conn_disconnect(conn);
					return_value = CONN_NOT_CONNECTED;
				}
				else
				{
					return_value = CONN_OPERATION_FAILED;
				}
			}
			if (NULL != res)
				ldap_msgfree(res);
		}
		else {
			return_value = conn->supports_ds71_repl ? CONN_SUPPORTS_DS71_REPL : CONN_DOES_NOT_SUPPORT_DS71_REPL;
		}
	}
	else
	{
		/* Not connected */
		return_value = CONN_NOT_CONNECTED;
	}
	return return_value;
}

/* Determine if the replica is read-only */
ConnResult
conn_replica_is_readonly(Repl_Connection *conn)
{
	ReplicaId rid = agmt_get_consumer_rid( (Repl_Agmt *) conn->agmt, conn );
	if (rid == READ_ONLY_REPLICA_ID)
	{
		return CONN_IS_READONLY;
	} else 
	{
		return CONN_IS_NOT_READONLY;
	}
}


/*
 * Return 1 if "value" is a value of attribute type "type" in entry "entry".
 * Otherwise, return 0.
 */
static int
attribute_string_value_present(LDAP *ld, LDAPMessage *entry, const char *type,
	const char *value)
{
	int return_value = 0;

	if (NULL != entry)
	{
		char *atype = NULL;
		BerElement *ber = NULL;

		atype = ldap_first_attribute(ld, entry, &ber);
		while (NULL != atype && 0 == return_value)
		{
			if (strcasecmp(atype, type) == 0)
			{
				char **strvals = ldap_get_values(ld, entry, atype);
				int i;
				for (i = 0; return_value == 0 && NULL != strvals && NULL != strvals[i]; i++)
				{
					if (strcmp(strvals[i], value) == 0)
					{
						return_value = 1;
					}
				}
				if (NULL != strvals)
				{
					ldap_value_free(strvals);
				}
			}
			ldap_memfree(atype);
			atype = ldap_next_attribute(ld, entry, ber);
		}
		if (NULL != ber)
			ldap_ber_free(ber, 0);
		/* The last atype has not been freed yet */
		if (NULL != atype)
			ldap_memfree(atype);
	}
	return return_value;
}




/*
 * Read the remote server's schema entry, then read the local schema entry,
 * and compare the nsschemacsn attribute. If the local csn is newer, or
 * the remote csn is absent, push the schema down to the consumer.
 * Return codes:
 * CONN_SCHEMA_UPDATED if the schema was pushed successfully
 * CONN_SCHEMA_NO_UPDATE_NEEDED if the schema was as new or newer than
 * the local server's schema
 * CONN_OPERATION_FAILED if an error occurred
 * CONN_NOT_CONNECTED if no connection was active
 * NOTE: Should only be called when a replication session has been
 * established by sending a startReplication extended operation.
 */
ConnResult
conn_push_schema(Repl_Connection *conn, CSN **remotecsn)
{
	ConnResult return_value = CONN_OPERATION_SUCCESS;
	char *nsschemacsn = "nsschemacsn";
	Slapi_Entry **entries = NULL;
	Slapi_Entry *schema_entry = NULL;
	CSN *localcsn = NULL;
	Slapi_PBlock *spb = NULL;
	char localcsnstr[CSN_STRSIZE + 1] = {0};

	if (!conn_connected(conn))
	{
		return_value = CONN_NOT_CONNECTED;
		slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
			"%s: Schema replication update failed: not connected to consumer\n",
			agmt_get_long_name(conn->agmt));
	}
	else
	{
		localcsn = dup_global_schema_csn();
		if (NULL == localcsn)
		{
			/* Local server has epoch CSN, so don't push schema */
			return_value = CONN_SCHEMA_NO_UPDATE_NEEDED;
		}
		else if ( remotecsn && *remotecsn && csn_compare(localcsn, *remotecsn) <= 0 )
		{
			/* Local server schema is not newer than the remote one */
			return_value = CONN_SCHEMA_NO_UPDATE_NEEDED;
		}
		else
		{
			struct berval **remote_schema_csn_bervals = NULL;
			/* Get remote server's schema */
			return_value = conn_read_entry_attribute(conn, "cn=schema", nsschemacsn,
				&remote_schema_csn_bervals);
			if (CONN_OPERATION_SUCCESS == return_value)
			{
				if (NULL != remote_schema_csn_bervals && NULL != remote_schema_csn_bervals[0])
				{
					char remotecsnstr[CSN_STRSIZE + 1] = {0};
					memcpy(remotecsnstr, remote_schema_csn_bervals[0]->bv_val,
						remote_schema_csn_bervals[0]->bv_len);
					remotecsnstr[remote_schema_csn_bervals[0]->bv_len] = '\0';
					*remotecsn = csn_new_by_string(remotecsnstr);
					if (NULL != remotecsn && (csn_compare(localcsn, *remotecsn) <= 0))
					{
						return_value = CONN_SCHEMA_NO_UPDATE_NEEDED;
					}
					/* Need to free the remote_schema_csn_bervals */
					ber_bvecfree(remote_schema_csn_bervals);
				}
			}
		}
	}
	if (CONN_OPERATION_SUCCESS == return_value)
	{
		/* We know we need to push the schema out. */
		LDAPMod ocmod = {0};
		LDAPMod atmod = {0};
		LDAPMod csnmod = {0};
		LDAPMod *attrs[4] = {0};
		int numvalues = 0;
		Slapi_Attr *attr = NULL;
		char *csnvalues[2];

		ocmod.mod_type = "objectclasses";
		ocmod.mod_op = LDAP_MOD_REPLACE | LDAP_MOD_BVALUES;
		ocmod.mod_bvalues = NULL;
		atmod.mod_type = "attributetypes";
		atmod.mod_op = LDAP_MOD_REPLACE | LDAP_MOD_BVALUES;
		atmod.mod_bvalues = NULL;
		csnmod.mod_type = nsschemacsn;
		csnmod.mod_op = LDAP_MOD_REPLACE;
		csn_as_string (localcsn, PR_FALSE, localcsnstr); 
		csnvalues[0] = localcsnstr;
		csnvalues[1] = NULL;
		csnmod.mod_values = csnvalues;
		attrs[0] = &ocmod;
		attrs[1] = &atmod;
		attrs[2] = &csnmod;
		attrs[3] = NULL;

		return_value = CONN_OPERATION_FAILED; /* assume failure */

		/* Get local schema */
		spb = slapi_search_internal("cn=schema", LDAP_SCOPE_BASE, "(objectclass=*)",
			NULL /* controls */, NULL /* schema_csn_attrs */, 0 /* attrsonly */);
		slapi_pblock_get(spb, SLAPI_PLUGIN_INTOP_SEARCH_ENTRIES, &entries);
		if (NULL == entries || NULL == entries[0])
		{
			/* Whoops - couldn't read our own schema! */
			slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
				"%s: Error: unable to read local schema definitions.\n",
				agmt_get_long_name(conn->agmt));
			return_value = CONN_OPERATION_FAILED;
		}
		else
		{
			schema_entry = entries[0];
			if (slapi_entry_attr_find(schema_entry, "objectclasses", &attr) != -1)
			{
				int i, ind;
				Slapi_Value *value;
				slapi_attr_get_numvalues(attr, &numvalues);
				ocmod.mod_bvalues = (struct berval **)slapi_ch_malloc((numvalues + 1) *
					sizeof(struct berval *));
				for (i = 0, ind = slapi_attr_first_value(attr, &value);
					ind != -1; ind = slapi_attr_next_value(attr, ind, &value), i++)
				{
					/* XXXggood had to cast away const below */
					ocmod.mod_bvalues[i] = (struct berval *)slapi_value_get_berval(value);
				}
				ocmod.mod_bvalues[numvalues] = NULL;
				if (slapi_entry_attr_find(schema_entry, "attributetypes", &attr) != -1)
				{
					ConnResult result;
					slapi_attr_get_numvalues(attr, &numvalues);
					atmod.mod_bvalues = (struct berval **)slapi_ch_malloc((numvalues + 1) *
						sizeof(struct berval *));
					for (i = 0, ind = slapi_attr_first_value(attr, &value);
						ind != -1; ind = slapi_attr_next_value(attr, ind, &value), i++)
					{
						/* XXXggood had to cast away const below */
						atmod.mod_bvalues[i] = (struct berval *)slapi_value_get_berval(value);
					}
					atmod.mod_bvalues[numvalues] = NULL;

					result = conn_send_modify(conn, "cn=schema", attrs, NULL, NULL); /* DBDB: this needs to be fixed to use async */
					result = conn_read_result(conn,NULL);
					switch (result)
					{
					case CONN_OPERATION_FAILED:
						{
						int ldaperr = -1, optype = -1;
						conn_get_error(conn, &optype, &ldaperr);
						slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
							"%s: Schema replication update failed: %s\n",
							agmt_get_long_name(conn->agmt),
							ldaperr == -1 ? "Unknown Error" : ldap_err2string(ldaperr));
						}
					case CONN_NOT_CONNECTED:
						return_value = CONN_NOT_CONNECTED;
						break;
					case CONN_OPERATION_SUCCESS:
						return_value = CONN_SCHEMA_UPDATED;
						break;
					default:
						break;
					}
				}
			}
			else
			{
				slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
					"%s: Schema replication update failed: "
					"unable to prepare schema entry for transmission.\n",
					agmt_get_long_name(conn->agmt));
			}
		}
		/* slapi_ch_free accepts NULL pointer */
		slapi_ch_free((void **)&ocmod.mod_bvalues);
		slapi_ch_free((void **)&atmod.mod_bvalues);
	}
	if (NULL != spb)
	{
		slapi_free_search_results_internal(spb);
		slapi_pblock_destroy(spb);
		spb = NULL;
	}
	if (NULL != localcsn)
	{
		csn_free(&localcsn);
	}
	return return_value;
}

void
conn_set_timeout(Repl_Connection *conn, long timeout)
{
	PR_ASSERT(NULL != conn);
	PR_ASSERT(timeout >= 0);
	PR_Lock(conn->lock);
	conn->timeout.tv_sec = timeout;
	PR_Unlock(conn->lock);
}

long
conn_get_timeout(Repl_Connection *conn)
{
	long retval = 0;
	PR_ASSERT(NULL != conn);
	retval = conn->timeout.tv_sec;
	return retval;
}

void conn_set_agmt_changed(Repl_Connection *conn)
{
	PR_ASSERT(NULL != conn);
	PR_Lock(conn->lock);
	if (NULL != conn->agmt)
		conn->flag_agmt_changed = 1;
	PR_Unlock(conn->lock);
}

/*
 * Check the result of an ldap_simple_bind operation to see we it
 * contains the expiration controls
 * return: -1 error, not bound
 *          0, OK bind has succeeded
 */
static int
bind_and_check_pwp(Repl_Connection *conn, char * binddn, char *password)
{

	LDAPControl	**ctrls = NULL;
	LDAPMessage	*res = NULL;
	char		*errmsg = NULL;
	LDAP		*ld = conn->ld;
	int 		msgid;
	int 		*msgidAdr = &msgid;
	int			rc;

	char * optype; /* ldap_simple_bind or slapd_SSL_client_bind */

	if ( conn->transport_flags == TRANSPORT_FLAG_SSL )
	{
		char *auth;
		optype = "ldap_sasl_bind";

		if ( conn->bindmethod == BINDMETHOD_SSL_CLIENTAUTH )
		{
			rc = slapd_sasl_ext_client_bind(conn->ld, &msgidAdr);	
			auth = "SSL client authentication";

			if ( rc == LDAP_SUCCESS )
			{
				if (conn->last_ldap_error != rc)
				{
					conn->last_ldap_error = rc;
					slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
					"%s: Replication bind with %s resumed\n",
					agmt_get_long_name(conn->agmt), auth);
				}
			}
			else
			{
				/* Do not report the same error over and over again */
				if (conn->last_ldap_error != rc)
				{
					conn->last_ldap_error = rc;
					slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
					"%s: Replication bind with %s failed: LDAP error %d (%s)\n",
					agmt_get_long_name(conn->agmt), auth, rc,
					ldap_err2string(rc));
				}

				return (CONN_OPERATION_FAILED);
			}
		}
		else
		{
			if( ( msgid = do_simple_bind( conn, ld, binddn, password ) ) == -1 )
			{
				return (CONN_OPERATION_FAILED);
			}
		}
	}
	else
	{
		optype = "ldap_simple_bind";
		if( ( msgid = do_simple_bind( conn, ld, binddn, password ) ) == -1 ) 
		{
			return (CONN_OPERATION_FAILED);
		}
	}

	/* Wait for the result */
	if ( ldap_result( ld, msgid, LDAP_MSG_ALL, NULL, &res ) == -1 ) 
	{
		slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, 
			"%s: Received error from consumer for %s operation\n", 
			agmt_get_long_name(conn->agmt), optype);

		return (CONN_OPERATION_FAILED);
	}
	/* Don't check ldap_result against 0 because, no timeout is specified */

	/* Free res as we won't use it any longer */
	if ( ldap_parse_result( ld, res, &rc, NULL, NULL, NULL, &ctrls, 1 /* Free res */) 
		!= LDAP_SUCCESS ) 
	{
		slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, 
			"%s: Received error from consumer for %s operation\n",
			agmt_get_long_name(conn->agmt), optype);

		return (CONN_OPERATION_FAILED);
	}

	if ( rc == LDAP_SUCCESS ) 
	{
		if ( ctrls ) 
		{
			int i;
			for( i = 0; ctrls[ i ] != NULL; ++i ) 
			{
				if ( !(strcmp( ctrls[ i ]->ldctl_oid, LDAP_CONTROL_PWEXPIRED)) )
				{
					/* Bind is successfull but password has expired */
					slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, 
						"%s: Succesfully bound %s to consumer, "
						"but password has expired on consumer.\n",
						agmt_get_long_name(conn->agmt), binddn);
				}					
				else if ( !(strcmp( ctrls[ i ]->ldctl_oid, LDAP_CONTROL_PWEXPIRING)) )
				{
					/* The password is expiring in n seconds */
					if ( (ctrls[ i ]->ldctl_value.bv_val != NULL) &&
						 (ctrls[ i ]->ldctl_value.bv_len > 0) )
					{
						int password_expiring = atoi( ctrls[ i ]->ldctl_value.bv_val );
						slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, 
							"%s: Succesfully bound %s to consumer, "
							"but password is expiring on consumer in %d seconds.\n",
							agmt_get_long_name(conn->agmt), binddn, password_expiring);
					}
				}
			}	
			ldap_controls_free( ctrls );
		}

		return (CONN_OPERATION_SUCCESS);
	}
	else 
	{
		/* errmsg is a pointer directly into the ld structure - do not free */
		rc = ldap_get_lderrno( ld, NULL, &errmsg );
		slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
			"%s: Replication bind to %s on consumer failed: %d (%s)\n",
			agmt_get_long_name(conn->agmt), binddn, rc, errmsg);

		conn->last_ldap_error = rc;	/* specific error */
		return (CONN_OPERATION_FAILED);
	}
}

static int
do_simple_bind (Repl_Connection *conn, LDAP *ld, char * binddn, char *password)
{
	int msgid;

	if( ( msgid = ldap_simple_bind( ld, binddn, password ) ) == -1 ) 
	{
		char *ldaperrtext = NULL;
		int ldaperr;
		int prerr = PR_GetError();

		ldaperr = ldap_get_lderrno( ld, NULL, &ldaperrtext );
		/* Do not report the same error over and over again */
		if (conn->last_ldap_error != ldaperr)
		{
			conn->last_ldap_error = ldaperr;
			slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, 
				"%s: Simple bind failed, " 
				SLAPI_COMPONENT_NAME_LDAPSDK " error %d (%s), "
				SLAPI_COMPONENT_NAME_NSPR " error %d (%s)\n",
				agmt_get_long_name(conn->agmt),
				ldaperr, ldaperrtext ? ldaperrtext : ldap_err2string(ldaperr),
				prerr, slapd_pr_strerror(prerr));
		}
	}
	else if (conn->last_ldap_error != LDAP_SUCCESS)
	{
		conn->last_ldap_error = LDAP_SUCCESS;
		slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
			"%s: Simple bind resumed\n",
			agmt_get_long_name(conn->agmt));
	}
	return msgid;
}

void
repl5_set_debug_timeout(const char *val)
{
	/* val looks like this: seconds[:debuglevel] */
	/* seconds is the number of seconds to wait until turning on the debug level */
	/* this should be less than the ldap connection timeout (default 10 minutes) */
	/* the optional debug level is the error log debugging level to use (default repl) */
	if (val) {
		const char *p = strchr(val, ':');
		s_debug_timeout = atoi(val);
		if (p) {
			s_debug_level = atoi(p+1);
		} else {
			s_debug_level = 8192;
		}
	}
}

static time_t 
PRTime2time_t (PRTime tm)
{
    PRInt64 rt;

    PR_ASSERT (tm);
    
    LL_DIV(rt, tm, PR_USEC_PER_SEC);

    return (time_t)rt;
}

static Slapi_Eq_Context
repl5_start_debug_timeout(int *setlevel)
{
	Slapi_Eq_Context eqctx = 0;
	if (s_debug_timeout && s_debug_level) {
		time_t now = time(NULL);
		eqctx = slapi_eq_once(repl5_debug_timeout_callback, setlevel,
							  s_debug_timeout + now);
	}
	return eqctx;
}

static void
repl5_stop_debug_timeout(Slapi_Eq_Context eqctx, int *setlevel)
{
	char buf[20];
	char msg[SLAPI_DSE_RETURNTEXT_SIZE];

	if (eqctx && !*setlevel) {
		int found = slapi_eq_cancel(eqctx);
	}

	if (s_debug_timeout && s_debug_level && *setlevel) {
		void config_set_errorlog_level(const char *type, char *buf, char *msg, int apply);
		sprintf(buf, "%d", 0);
		config_set_errorlog_level("nsslapd-errorlog-level", buf, msg, 1);
	}
}

static void
repl5_debug_timeout_callback(time_t when, void *arg)
{
	int *setlevel = (int *)arg;
	void config_set_errorlog_level(const char *type, char *buf, char *msg, int apply);
	char buf[20];
	char msg[SLAPI_DSE_RETURNTEXT_SIZE];

	*setlevel = 1;
	sprintf(buf, "%d", s_debug_level);
	config_set_errorlog_level("nsslapd-errorlog-level", buf, msg, 1);

	slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, 
		"repl5_debug_timeout_callback: set debug level to %d at %d\n",
		s_debug_level, when);
}