1 /* $NetBSD: upstream.c,v 1.2 2021/08/14 16:14:58 christos Exp $ */ 2 3 /* $OpenLDAP$ */ 4 /* This work is part of OpenLDAP Software <http://www.openldap.org/>. 5 * 6 * Copyright 1998-2021 The OpenLDAP Foundation. 7 * All rights reserved. 8 * 9 * Redistribution and use in source and binary forms, with or without 10 * modification, are permitted only as authorized by the OpenLDAP 11 * Public License. 12 * 13 * A copy of this license is available in the file LICENSE in the 14 * top-level directory of the distribution or, alternatively, at 15 * <http://www.OpenLDAP.org/license.html>. 16 */ 17 18 #include <sys/cdefs.h> 19 __RCSID("$NetBSD: upstream.c,v 1.2 2021/08/14 16:14:58 christos Exp $"); 20 21 #include "portable.h" 22 23 #include <ac/socket.h> 24 #include <ac/errno.h> 25 #include <ac/string.h> 26 #include <ac/time.h> 27 #include <ac/unistd.h> 28 29 #include "lload.h" 30 31 #include "lutil.h" 32 #include "lutil_ldap.h" 33 34 #ifdef HAVE_CYRUS_SASL 35 static const sasl_callback_t client_callbacks[] = { 36 #ifdef SASL_CB_GETREALM 37 { SASL_CB_GETREALM, NULL, NULL }, 38 #endif 39 { SASL_CB_USER, NULL, NULL }, 40 { SASL_CB_AUTHNAME, NULL, NULL }, 41 { SASL_CB_PASS, NULL, NULL }, 42 { SASL_CB_LIST_END, NULL, NULL } 43 }; 44 #endif /* HAVE_CYRUS_SASL */ 45 46 static void upstream_unlink( LloadConnection *upstream ); 47 48 int 49 forward_response( LloadConnection *client, LloadOperation *op, BerElement *ber ) 50 { 51 BerElement *output; 52 BerValue response, controls = BER_BVNULL; 53 ber_int_t msgid; 54 ber_tag_t tag, response_tag; 55 ber_len_t len; 56 57 CONNECTION_LOCK(client); 58 if ( op->o_client_msgid ) { 59 msgid = op->o_client_msgid; 60 } else { 61 assert( op->o_pin_id ); 62 msgid = op->o_saved_msgid; 63 op->o_saved_msgid = 0; 64 } 65 CONNECTION_UNLOCK(client); 66 67 response_tag = ber_skip_element( ber, &response ); 68 69 tag = ber_peek_tag( ber, &len ); 70 if ( tag == LDAP_TAG_CONTROLS ) { 71 ber_skip_element( ber, &controls ); 72 } 73 74 Debug( LDAP_DEBUG_TRACE, "forward_response: " 75 "%s to client connid=%lu request msgid=%d\n", 76 lload_msgtype2str( response_tag ), op->o_client_connid, msgid ); 77 78 checked_lock( &client->c_io_mutex ); 79 output = client->c_pendingber; 80 if ( output == NULL && (output = ber_alloc()) == NULL ) { 81 ber_free( ber, 1 ); 82 checked_unlock( &client->c_io_mutex ); 83 return -1; 84 } 85 client->c_pendingber = output; 86 87 ber_printf( output, "t{titOtO}", LDAP_TAG_MESSAGE, 88 LDAP_TAG_MSGID, msgid, 89 response_tag, &response, 90 LDAP_TAG_CONTROLS, BER_BV_OPTIONAL( &controls ) ); 91 92 checked_unlock( &client->c_io_mutex ); 93 94 ber_free( ber, 1 ); 95 connection_write_cb( -1, 0, client ); 96 return 0; 97 } 98 99 int 100 forward_final_response( 101 LloadConnection *client, 102 LloadOperation *op, 103 BerElement *ber ) 104 { 105 int rc; 106 107 Debug( LDAP_DEBUG_STATS, "forward_final_response: " 108 "connid=%lu msgid=%d finishing up with a request for " 109 "client connid=%lu\n", 110 op->o_upstream_connid, op->o_upstream_msgid, op->o_client_connid ); 111 112 rc = forward_response( client, op, ber ); 113 114 op->o_res = LLOAD_OP_COMPLETED; 115 if ( !op->o_pin_id ) { 116 operation_unlink( op ); 117 } 118 119 return rc; 120 } 121 122 static int 123 handle_unsolicited( LloadConnection *c, BerElement *ber ) 124 { 125 CONNECTION_ASSERT_LOCKED(c); 126 if ( c->c_state != LLOAD_C_PREPARING ) { 127 c->c_state = LLOAD_C_CLOSING; 128 } 129 130 Debug( LDAP_DEBUG_STATS, "handle_unsolicited: " 131 "teardown for upstream connection connid=%lu\n", 132 c->c_connid ); 133 134 CONNECTION_DESTROY(c); 135 ber_free( ber, 1 ); 136 137 return -1; 138 } 139 140 /* 141 * Pull c->c_currentber from the connection and try to look up the operation on 142 * the upstream. 143 * 144 * If it's a notice of disconnection, we won't find it and need to tear down 145 * the connection and tell the clients, if we can't find the operation, ignore 146 * the message (either client already disconnected/abandoned it or the upstream 147 * is pulling our leg). 148 * 149 * Some responses need special handling: 150 * - Bind response 151 * - VC response where the client requested a Bind (both need to update the 152 * client's bind status) 153 * - search entries/referrals and intermediate responses (will not trigger 154 * operation to be removed) 155 * 156 * If the worker pool is overloaded, we might be called directly from 157 * the read callback, at that point, the connection hasn't been muted. 158 * 159 * TODO: when the client already has data pending on write, we should mute the 160 * upstream. 161 * - should record the BerElement on the Op and the Op on the client 162 * 163 * The following hold on entering any of the handlers: 164 * - op->o_upstream_refcnt > 0 165 * - op->o_upstream->c_refcnt > 0 166 * - op->o_client->c_refcnt > 0 167 */ 168 static int 169 handle_one_response( LloadConnection *c ) 170 { 171 BerElement *ber; 172 LloadOperation *op = NULL, needle = { .o_upstream_connid = c->c_connid }; 173 LloadOperationHandler handler = NULL; 174 ber_tag_t tag; 175 ber_len_t len; 176 int rc = LDAP_SUCCESS; 177 178 ber = c->c_currentber; 179 c->c_currentber = NULL; 180 181 tag = ber_get_int( ber, &needle.o_upstream_msgid ); 182 if ( tag != LDAP_TAG_MSGID ) { 183 rc = -1; 184 ber_free( ber, 1 ); 185 goto fail; 186 } 187 188 CONNECTION_LOCK(c); 189 if ( needle.o_upstream_msgid == 0 ) { 190 return handle_unsolicited( c, ber ); 191 } else if ( !( op = ldap_tavl_find( 192 c->c_ops, &needle, operation_upstream_cmp ) ) ) { 193 /* Already abandoned, do nothing */ 194 CONNECTION_UNLOCK(c); 195 ber_free( ber, 1 ); 196 return rc; 197 /* 198 } else if ( op->o_response_pending ) { 199 c->c_pendingop = op; 200 event_del( c->c_read_event ); 201 */ 202 } else { 203 CONNECTION_UNLOCK(c); 204 /* 205 op->o_response_pending = ber; 206 */ 207 208 tag = ber_peek_tag( ber, &len ); 209 switch ( tag ) { 210 case LDAP_RES_SEARCH_ENTRY: 211 case LDAP_RES_SEARCH_REFERENCE: 212 case LDAP_RES_INTERMEDIATE: 213 handler = forward_response; 214 break; 215 case LDAP_RES_BIND: 216 handler = handle_bind_response; 217 break; 218 case LDAP_RES_EXTENDED: 219 if ( op->o_tag == LDAP_REQ_BIND ) { 220 #ifdef LDAP_API_FEATURE_VERIFY_CREDENTIALS 221 if ( lload_features & LLOAD_FEATURE_VC ) { 222 handler = handle_vc_bind_response; 223 } else 224 #endif /* LDAP_API_FEATURE_VERIFY_CREDENTIALS */ 225 { 226 handler = handle_whoami_response; 227 } 228 } 229 break; 230 } 231 if ( !handler ) { 232 handler = forward_final_response; 233 } 234 } 235 if ( op ) { 236 op->o_last_response = slap_get_time(); 237 Debug( LDAP_DEBUG_STATS2, "handle_one_response: " 238 "upstream connid=%lu, processing response for " 239 "client connid=%lu, msgid=%d\n", 240 c->c_connid, op->o_client_connid, op->o_client_msgid ); 241 } else { 242 tag = ber_peek_tag( ber, &len ); 243 Debug( LDAP_DEBUG_STATS2, "handle_one_response: " 244 "upstream connid=%lu, %s, msgid=%d not for a pending " 245 "operation\n", 246 c->c_connid, lload_msgtype2str( tag ), 247 needle.o_upstream_msgid ); 248 } 249 250 if ( handler ) { 251 LloadConnection *client; 252 253 checked_lock( &op->o_link_mutex ); 254 client = op->o_client; 255 checked_unlock( &op->o_link_mutex ); 256 if ( client && IS_ALIVE( client, c_live ) ) { 257 rc = handler( client, op, ber ); 258 } else { 259 ber_free( ber, 1 ); 260 } 261 } else { 262 assert(0); 263 ber_free( ber, 1 ); 264 } 265 266 fail: 267 if ( rc ) { 268 Debug( LDAP_DEBUG_STATS, "handle_one_response: " 269 "error on processing a response (%s) on upstream connection " 270 "connid=%ld, tag=%lx\n", 271 lload_msgtype2str( tag ), c->c_connid, tag ); 272 CONNECTION_LOCK_DESTROY(c); 273 } 274 return rc; 275 } 276 277 #ifdef HAVE_CYRUS_SASL 278 static int 279 sasl_bind_step( LloadConnection *c, BerValue *scred, BerValue *ccred ) 280 { 281 LloadBackend *b = c->c_backend; 282 sasl_conn_t *ctx = c->c_sasl_authctx; 283 sasl_interact_t *prompts = NULL; 284 unsigned credlen; 285 int rc = -1; 286 287 if ( !ctx ) { 288 const char *mech = NULL; 289 #ifdef HAVE_TLS 290 void *ssl; 291 #endif /* HAVE_TLS */ 292 293 if ( sasl_client_new( "ldap", b->b_host, NULL, NULL, client_callbacks, 294 0, &ctx ) != SASL_OK ) { 295 goto done; 296 } 297 c->c_sasl_authctx = ctx; 298 299 assert( c->c_sasl_defaults == NULL ); 300 c->c_sasl_defaults = 301 lutil_sasl_defaults( NULL, bindconf.sb_saslmech.bv_val, 302 bindconf.sb_realm.bv_val, bindconf.sb_authcId.bv_val, 303 bindconf.sb_cred.bv_val, bindconf.sb_authzId.bv_val ); 304 305 #ifdef HAVE_TLS 306 /* Check for TLS */ 307 ssl = ldap_pvt_tls_sb_ctx( c->c_sb ); 308 if ( ssl ) { 309 struct berval authid = BER_BVNULL; 310 ber_len_t ssf; 311 312 ssf = ldap_pvt_tls_get_strength( ssl ); 313 (void)ldap_pvt_tls_get_my_dn( ssl, &authid, NULL, 0 ); 314 315 sasl_setprop( ctx, SASL_SSF_EXTERNAL, &ssf ); 316 sasl_setprop( ctx, SASL_AUTH_EXTERNAL, authid.bv_val ); 317 ch_free( authid.bv_val ); 318 #ifdef SASL_CHANNEL_BINDING /* 2.1.25+ */ 319 { 320 char cbinding[64]; 321 struct berval cbv = { sizeof(cbinding), cbinding }; 322 if ( ldap_pvt_tls_get_unique( ssl, &cbv, 0 ) ) { 323 sasl_channel_binding_t *cb = 324 ch_malloc( sizeof(*cb) + cbv.bv_len ); 325 void *cb_data; 326 cb->name = "ldap"; 327 cb->critical = 0; 328 cb->len = cbv.bv_len; 329 cb->data = cb_data = cb + 1; 330 memcpy( cb_data, cbv.bv_val, cbv.bv_len ); 331 sasl_setprop( ctx, SASL_CHANNEL_BINDING, cb ); 332 c->c_sasl_cbinding = cb; 333 } 334 } 335 #endif 336 } 337 #endif 338 339 #if !defined(_WIN32) 340 /* Check for local */ 341 if ( b->b_proto == LDAP_PROTO_IPC ) { 342 char authid[sizeof( "gidNumber=4294967295+uidNumber=4294967295," 343 "cn=peercred,cn=external,cn=auth" )]; 344 int ssf = LDAP_PVT_SASL_LOCAL_SSF; 345 346 sprintf( authid, 347 "gidNumber=%u+uidNumber=%u," 348 "cn=peercred,cn=external,cn=auth", 349 getegid(), geteuid() ); 350 sasl_setprop( ctx, SASL_SSF_EXTERNAL, &ssf ); 351 sasl_setprop( ctx, SASL_AUTH_EXTERNAL, authid ); 352 } 353 #endif 354 355 do { 356 rc = sasl_client_start( ctx, bindconf.sb_saslmech.bv_val, 357 &prompts, 358 (const char **)&ccred->bv_val, &credlen, 359 &mech ); 360 361 if ( rc == SASL_INTERACT ) { 362 if ( lutil_sasl_interact( NULL, LDAP_SASL_QUIET, 363 c->c_sasl_defaults, prompts ) ) { 364 break; 365 } 366 } 367 } while ( rc == SASL_INTERACT ); 368 369 ber_str2bv( mech, 0, 0, &c->c_sasl_bind_mech ); 370 } else { 371 assert( c->c_sasl_defaults ); 372 373 do { 374 rc = sasl_client_step( ctx, 375 (scred == NULL) ? NULL : scred->bv_val, 376 (scred == NULL) ? 0 : scred->bv_len, 377 &prompts, 378 (const char **)&ccred->bv_val, &credlen); 379 380 if ( rc == SASL_INTERACT ) { 381 if ( lutil_sasl_interact( NULL, LDAP_SASL_QUIET, 382 c->c_sasl_defaults, prompts ) ) { 383 break; 384 } 385 } 386 } while ( rc == SASL_INTERACT ); 387 } 388 389 if ( rc == SASL_OK ) { 390 sasl_ssf_t *ssf; 391 rc = sasl_getprop( ctx, SASL_SSF, (const void **)(char *)&ssf ); 392 if ( rc == SASL_OK && ssf && *ssf ) { 393 Debug( LDAP_DEBUG_CONNS, "sasl_bind_step: " 394 "connid=%lu mech=%s setting up a new SASL security layer\n", 395 c->c_connid, c->c_sasl_bind_mech.bv_val ); 396 ldap_pvt_sasl_install( c->c_sb, ctx ); 397 } 398 } 399 ccred->bv_len = credlen; 400 401 done: 402 Debug( LDAP_DEBUG_TRACE, "sasl_bind_step: " 403 "connid=%lu next step for SASL bind mech=%s rc=%d\n", 404 c->c_connid, c->c_sasl_bind_mech.bv_val, rc ); 405 return rc; 406 } 407 #endif /* HAVE_CYRUS_SASL */ 408 409 int 410 upstream_bind_cb( LloadConnection *c ) 411 { 412 BerElement *ber = c->c_currentber; 413 LloadBackend *b = c->c_backend; 414 BerValue matcheddn, message; 415 ber_tag_t tag; 416 ber_int_t msgid, result; 417 418 c->c_currentber = NULL; 419 420 if ( ber_scanf( ber, "it", &msgid, &tag ) == LBER_ERROR ) { 421 Debug( LDAP_DEBUG_ANY, "upstream_bind_cb: " 422 "protocol violation from server\n" ); 423 goto fail; 424 } 425 426 if ( msgid != ( c->c_next_msgid - 1 ) || tag != LDAP_RES_BIND ) { 427 Debug( LDAP_DEBUG_ANY, "upstream_bind_cb: " 428 "unexpected %s from server, msgid=%d\n", 429 lload_msgtype2str( tag ), msgid ); 430 goto fail; 431 } 432 433 if ( ber_scanf( ber, "{emm" /* "}" */, &result, &matcheddn, &message ) == 434 LBER_ERROR ) { 435 Debug( LDAP_DEBUG_ANY, "upstream_bind_cb: " 436 "response does not conform with a bind response\n" ); 437 goto fail; 438 } 439 440 switch ( result ) { 441 case LDAP_SUCCESS: 442 #ifdef HAVE_CYRUS_SASL 443 case LDAP_SASL_BIND_IN_PROGRESS: 444 if ( !BER_BVISNULL( &c->c_sasl_bind_mech ) ) { 445 BerValue scred = BER_BVNULL, ccred; 446 ber_len_t len; 447 int rc; 448 449 if ( ber_peek_tag( ber, &len ) == LDAP_TAG_SASL_RES_CREDS && 450 ber_scanf( ber, "m", &scred ) == LBER_ERROR ) { 451 Debug( LDAP_DEBUG_ANY, "upstream_bind_cb: " 452 "sasl bind response malformed\n" ); 453 goto fail; 454 } 455 456 rc = sasl_bind_step( c, &scred, &ccred ); 457 if ( rc != SASL_OK && 458 ( rc != SASL_CONTINUE || result == LDAP_SUCCESS ) ) { 459 goto fail; 460 } 461 462 if ( result == LDAP_SASL_BIND_IN_PROGRESS ) { 463 BerElement *outber; 464 465 checked_lock( &c->c_io_mutex ); 466 outber = c->c_pendingber; 467 if ( outber == NULL && (outber = ber_alloc()) == NULL ) { 468 checked_unlock( &c->c_io_mutex ); 469 goto fail; 470 } 471 c->c_pendingber = outber; 472 473 msgid = c->c_next_msgid++; 474 ber_printf( outber, "{it{iOt{OON}N}}", 475 msgid, LDAP_REQ_BIND, LDAP_VERSION3, 476 &bindconf.sb_binddn, LDAP_AUTH_SASL, 477 &c->c_sasl_bind_mech, BER_BV_OPTIONAL( &ccred ) ); 478 checked_unlock( &c->c_io_mutex ); 479 480 connection_write_cb( -1, 0, c ); 481 482 if ( rc == SASL_OK ) { 483 BER_BVZERO( &c->c_sasl_bind_mech ); 484 } 485 break; 486 } 487 } 488 if ( result == LDAP_SASL_BIND_IN_PROGRESS ) { 489 goto fail; 490 } 491 #endif /* HAVE_CYRUS_SASL */ 492 CONNECTION_LOCK(c); 493 c->c_pdu_cb = handle_one_response; 494 c->c_state = LLOAD_C_READY; 495 c->c_type = LLOAD_C_OPEN; 496 c->c_read_timeout = NULL; 497 Debug( LDAP_DEBUG_CONNS, "upstream_bind_cb: " 498 "connid=%lu finished binding, now active\n", 499 c->c_connid ); 500 CONNECTION_UNLOCK(c); 501 checked_lock( &b->b_mutex ); 502 LDAP_CIRCLEQ_REMOVE( &b->b_preparing, c, c_next ); 503 b->b_active++; 504 b->b_opening--; 505 b->b_failed = 0; 506 if ( b->b_last_conn ) { 507 LDAP_CIRCLEQ_INSERT_AFTER( 508 &b->b_conns, b->b_last_conn, c, c_next ); 509 } else { 510 LDAP_CIRCLEQ_INSERT_HEAD( &b->b_conns, c, c_next ); 511 } 512 b->b_last_conn = c; 513 backend_retry( b ); 514 checked_unlock( &b->b_mutex ); 515 break; 516 default: 517 Debug( LDAP_DEBUG_ANY, "upstream_bind_cb: " 518 "upstream bind failed, rc=%d, message='%s'\n", 519 result, message.bv_val ); 520 goto fail; 521 } 522 523 checked_lock( &c->c_io_mutex ); 524 c->c_io_state &= ~LLOAD_C_READ_HANDOVER; 525 checked_unlock( &c->c_io_mutex ); 526 event_add( c->c_read_event, c->c_read_timeout ); 527 ber_free( ber, 1 ); 528 return -1; 529 530 fail: 531 CONNECTION_LOCK_DESTROY(c); 532 ber_free( ber, 1 ); 533 return -1; 534 } 535 536 void * 537 upstream_bind( void *ctx, void *arg ) 538 { 539 LloadConnection *c = arg; 540 BerElement *ber; 541 ber_int_t msgid; 542 543 /* A reference was passed on to us */ 544 assert( IS_ALIVE( c, c_refcnt ) ); 545 546 if ( !IS_ALIVE( c, c_live ) ) { 547 RELEASE_REF( c, c_refcnt, c->c_destroy ); 548 return NULL; 549 } 550 551 CONNECTION_LOCK(c); 552 assert( !event_pending( c->c_read_event, EV_READ, NULL ) ); 553 c->c_pdu_cb = upstream_bind_cb; 554 CONNECTION_UNLOCK(c); 555 556 checked_lock( &c->c_io_mutex ); 557 ber = c->c_pendingber; 558 if ( ber == NULL && (ber = ber_alloc()) == NULL ) { 559 goto fail; 560 } 561 c->c_pendingber = ber; 562 msgid = c->c_next_msgid++; 563 564 if ( bindconf.sb_method == LDAP_AUTH_SIMPLE ) { 565 /* simple bind */ 566 ber_printf( ber, "{it{iOtON}}", 567 msgid, LDAP_REQ_BIND, LDAP_VERSION3, 568 &bindconf.sb_binddn, LDAP_AUTH_SIMPLE, 569 &bindconf.sb_cred ); 570 571 #ifdef HAVE_CYRUS_SASL 572 } else { 573 BerValue cred; 574 int rc; 575 576 rc = sasl_bind_step( c, NULL, &cred ); 577 if ( rc != SASL_OK && rc != SASL_CONTINUE ) { 578 goto fail; 579 } 580 581 ber_printf( ber, "{it{iOt{OON}N}}", 582 msgid, LDAP_REQ_BIND, LDAP_VERSION3, 583 &bindconf.sb_binddn, LDAP_AUTH_SASL, 584 &c->c_sasl_bind_mech, BER_BV_OPTIONAL( &cred ) ); 585 586 if ( rc == SASL_OK ) { 587 BER_BVZERO( &c->c_sasl_bind_mech ); 588 } 589 #endif /* HAVE_CYRUS_SASL */ 590 } 591 /* TODO: can we be paused at this point? Then we'd have to move this line 592 * after connection_write_cb */ 593 c->c_io_state &= ~LLOAD_C_READ_HANDOVER; 594 checked_unlock( &c->c_io_mutex ); 595 596 connection_write_cb( -1, 0, c ); 597 598 CONNECTION_LOCK(c); 599 c->c_read_timeout = lload_timeout_net; 600 event_add( c->c_read_event, c->c_read_timeout ); 601 CONNECTION_UNLOCK(c); 602 603 RELEASE_REF( c, c_refcnt, c->c_destroy ); 604 return NULL; 605 606 fail: 607 checked_unlock( &c->c_io_mutex ); 608 CONNECTION_LOCK_DESTROY(c); 609 RELEASE_REF( c, c_refcnt, c->c_destroy ); 610 return NULL; 611 } 612 613 /* 614 * The backend is already locked when entering the function. 615 */ 616 static int 617 upstream_finish( LloadConnection *c ) 618 { 619 LloadBackend *b = c->c_backend; 620 int is_bindconn = 0; 621 622 assert_locked( &b->b_mutex ); 623 CONNECTION_ASSERT_LOCKED(c); 624 assert( c->c_live ); 625 c->c_pdu_cb = handle_one_response; 626 627 /* Unless we are configured to use the VC exop, consider allocating the 628 * connection into the bind conn pool. Start off by allocating one for 629 * general use, then one for binds, then we start filling up the general 630 * connection pool, finally the bind pool */ 631 if ( 632 #ifdef LDAP_API_FEATURE_VERIFY_CREDENTIALS 633 !(lload_features & LLOAD_FEATURE_VC) && 634 #endif /* LDAP_API_FEATURE_VERIFY_CREDENTIALS */ 635 b->b_active && b->b_numbindconns ) { 636 if ( !b->b_bindavail ) { 637 is_bindconn = 1; 638 } else if ( b->b_active >= b->b_numconns && 639 b->b_bindavail < b->b_numbindconns ) { 640 is_bindconn = 1; 641 } 642 } 643 644 if ( is_bindconn ) { 645 LDAP_CIRCLEQ_REMOVE( &b->b_preparing, c, c_next ); 646 c->c_state = LLOAD_C_READY; 647 c->c_type = LLOAD_C_BIND; 648 b->b_bindavail++; 649 b->b_opening--; 650 b->b_failed = 0; 651 if ( b->b_last_bindconn ) { 652 LDAP_CIRCLEQ_INSERT_AFTER( 653 &b->b_bindconns, b->b_last_bindconn, c, c_next ); 654 } else { 655 LDAP_CIRCLEQ_INSERT_HEAD( &b->b_bindconns, c, c_next ); 656 } 657 b->b_last_bindconn = c; 658 } else if ( bindconf.sb_method == LDAP_AUTH_NONE ) { 659 LDAP_CIRCLEQ_REMOVE( &b->b_preparing, c, c_next ); 660 c->c_state = LLOAD_C_READY; 661 c->c_type = LLOAD_C_OPEN; 662 b->b_active++; 663 b->b_opening--; 664 b->b_failed = 0; 665 if ( b->b_last_conn ) { 666 LDAP_CIRCLEQ_INSERT_AFTER( &b->b_conns, b->b_last_conn, c, c_next ); 667 } else { 668 LDAP_CIRCLEQ_INSERT_HEAD( &b->b_conns, c, c_next ); 669 } 670 b->b_last_conn = c; 671 } else { 672 if ( ldap_pvt_thread_pool_submit( 673 &connection_pool, upstream_bind, c ) ) { 674 Debug( LDAP_DEBUG_ANY, "upstream_finish: " 675 "failed to set up a bind callback for connid=%lu\n", 676 c->c_connid ); 677 return -1; 678 } 679 /* keep a reference for upstream_bind */ 680 acquire_ref( &c->c_refcnt ); 681 682 Debug( LDAP_DEBUG_CONNS, "upstream_finish: " 683 "scheduled a bind callback for connid=%lu\n", 684 c->c_connid ); 685 return LDAP_SUCCESS; 686 } 687 event_add( c->c_read_event, c->c_read_timeout ); 688 689 Debug( LDAP_DEBUG_CONNS, "upstream_finish: " 690 "%sconnection connid=%lu for backend server '%s' is ready for " 691 "use\n", 692 is_bindconn ? "bind " : "", c->c_connid, b->b_name.bv_val ); 693 694 backend_retry( b ); 695 return LDAP_SUCCESS; 696 } 697 698 #ifdef HAVE_TLS 699 static void 700 upstream_tls_handshake_cb( evutil_socket_t s, short what, void *arg ) 701 { 702 LloadConnection *c = arg; 703 LloadBackend *b; 704 epoch_t epoch; 705 int rc = LDAP_SUCCESS; 706 707 CONNECTION_LOCK(c); 708 if ( what & EV_TIMEOUT ) { 709 Debug( LDAP_DEBUG_CONNS, "upstream_tls_handshake_cb: " 710 "connid=%lu, timeout reached, destroying\n", 711 c->c_connid ); 712 goto fail; 713 } 714 b = c->c_backend; 715 716 rc = ldap_pvt_tls_connect( lload_tls_backend_ld, c->c_sb, b->b_host ); 717 if ( rc < 0 ) { 718 goto fail; 719 } 720 721 if ( rc == 0 ) { 722 struct event_base *base = event_get_base( c->c_read_event ); 723 724 /* 725 * We're finished, replace the callbacks 726 * 727 * This is deadlock-safe, since both share the same base - the one 728 * that's just running us. 729 */ 730 event_del( c->c_read_event ); 731 event_del( c->c_write_event ); 732 733 c->c_read_timeout = NULL; 734 event_assign( c->c_read_event, base, c->c_fd, EV_READ|EV_PERSIST, 735 connection_read_cb, c ); 736 event_assign( c->c_write_event, base, c->c_fd, EV_WRITE, 737 connection_write_cb, c ); 738 Debug( LDAP_DEBUG_CONNS, "upstream_tls_handshake_cb: " 739 "connid=%lu finished\n", 740 c->c_connid ); 741 c->c_is_tls = LLOAD_TLS_ESTABLISHED; 742 743 CONNECTION_UNLOCK(c); 744 checked_lock( &b->b_mutex ); 745 CONNECTION_LOCK(c); 746 747 rc = upstream_finish( c ); 748 checked_unlock( &b->b_mutex ); 749 750 if ( rc ) { 751 goto fail; 752 } 753 } else if ( ber_sockbuf_ctrl( c->c_sb, LBER_SB_OPT_NEEDS_WRITE, NULL ) ) { 754 event_add( c->c_write_event, lload_write_timeout ); 755 Debug( LDAP_DEBUG_CONNS, "upstream_tls_handshake_cb: " 756 "connid=%lu need write rc=%d\n", 757 c->c_connid, rc ); 758 } 759 CONNECTION_UNLOCK(c); 760 return; 761 762 fail: 763 Debug( LDAP_DEBUG_CONNS, "upstream_tls_handshake_cb: " 764 "connid=%lu failed rc=%d\n", 765 c->c_connid, rc ); 766 767 assert( c->c_ops == NULL ); 768 epoch = epoch_join(); 769 CONNECTION_DESTROY(c); 770 epoch_leave( epoch ); 771 } 772 773 static int 774 upstream_starttls( LloadConnection *c ) 775 { 776 BerValue matcheddn, message, responseOid, 777 startTLSOid = BER_BVC(LDAP_EXOP_START_TLS); 778 BerElement *ber = c->c_currentber; 779 struct event_base *base; 780 ber_int_t msgid, result; 781 ber_tag_t tag; 782 783 c->c_currentber = NULL; 784 CONNECTION_LOCK(c); 785 786 if ( ber_scanf( ber, "it", &msgid, &tag ) == LBER_ERROR ) { 787 Debug( LDAP_DEBUG_ANY, "upstream_starttls: " 788 "protocol violation from server\n" ); 789 goto fail; 790 } 791 792 if ( msgid != ( c->c_next_msgid - 1 ) || tag != LDAP_RES_EXTENDED ) { 793 Debug( LDAP_DEBUG_ANY, "upstream_starttls: " 794 "unexpected %s from server, msgid=%d\n", 795 lload_msgtype2str( tag ), msgid ); 796 goto fail; 797 } 798 799 if ( ber_scanf( ber, "{emm}", &result, &matcheddn, &message ) == 800 LBER_ERROR ) { 801 Debug( LDAP_DEBUG_ANY, "upstream_starttls: " 802 "protocol violation on StartTLS response\n" ); 803 goto fail; 804 } 805 806 if ( (tag = ber_get_tag( ber )) != LBER_DEFAULT ) { 807 if ( tag != LDAP_TAG_EXOP_RES_OID || 808 ber_scanf( ber, "{m}", &responseOid ) == LBER_DEFAULT ) { 809 Debug( LDAP_DEBUG_ANY, "upstream_starttls: " 810 "protocol violation on StartTLS response\n" ); 811 goto fail; 812 } 813 814 if ( ber_bvcmp( &responseOid, &startTLSOid ) ) { 815 Debug( LDAP_DEBUG_ANY, "upstream_starttls: " 816 "oid=%s not a StartTLS response\n", 817 responseOid.bv_val ); 818 goto fail; 819 } 820 } 821 822 if ( result != LDAP_SUCCESS ) { 823 LloadBackend *b = c->c_backend; 824 int rc; 825 826 Debug( LDAP_DEBUG_STATS, "upstream_starttls: " 827 "server doesn't support StartTLS rc=%d message='%s'%s\n", 828 result, message.bv_val, 829 (c->c_is_tls == LLOAD_STARTTLS_OPTIONAL) ? ", ignored" : "" ); 830 if ( c->c_is_tls != LLOAD_STARTTLS_OPTIONAL ) { 831 goto fail; 832 } 833 c->c_is_tls = LLOAD_CLEARTEXT; 834 835 CONNECTION_UNLOCK(c); 836 checked_lock( &b->b_mutex ); 837 CONNECTION_LOCK(c); 838 839 rc = upstream_finish( c ); 840 checked_unlock( &b->b_mutex ); 841 842 if ( rc ) { 843 goto fail; 844 } 845 846 ber_free( ber, 1 ); 847 CONNECTION_UNLOCK(c); 848 849 checked_lock( &c->c_io_mutex ); 850 c->c_io_state &= ~LLOAD_C_READ_HANDOVER; 851 checked_unlock( &c->c_io_mutex ); 852 853 /* Do not keep handle_pdus running, we have adjusted c_read_event as we 854 * need it. */ 855 return -1; 856 } 857 858 base = event_get_base( c->c_read_event ); 859 860 c->c_io_state &= ~LLOAD_C_READ_HANDOVER; 861 event_del( c->c_read_event ); 862 event_del( c->c_write_event ); 863 864 c->c_read_timeout = lload_timeout_net; 865 event_assign( c->c_read_event, base, c->c_fd, EV_READ|EV_PERSIST, 866 upstream_tls_handshake_cb, c ); 867 event_assign( c->c_write_event, base, c->c_fd, EV_WRITE, 868 upstream_tls_handshake_cb, c ); 869 870 event_add( c->c_read_event, c->c_read_timeout ); 871 event_add( c->c_write_event, lload_write_timeout ); 872 873 CONNECTION_UNLOCK(c); 874 875 ber_free( ber, 1 ); 876 return -1; 877 878 fail: 879 ber_free( ber, 1 ); 880 CONNECTION_DESTROY(c); 881 return -1; 882 } 883 #endif /* HAVE_TLS */ 884 885 /* 886 * We must already hold b->b_mutex when called. 887 */ 888 LloadConnection * 889 upstream_init( ber_socket_t s, LloadBackend *b ) 890 { 891 LloadConnection *c; 892 struct event_base *base = lload_get_base( s ); 893 struct event *event; 894 int flags; 895 896 assert( b != NULL ); 897 898 flags = (b->b_proto == LDAP_PROTO_IPC) ? CONN_IS_IPC : 0; 899 if ( (c = lload_connection_init( s, b->b_host, flags )) == NULL ) { 900 return NULL; 901 } 902 903 CONNECTION_LOCK(c); 904 c->c_backend = b; 905 #ifdef HAVE_TLS 906 c->c_is_tls = b->b_tls; 907 #endif 908 c->c_pdu_cb = handle_one_response; 909 910 LDAP_CIRCLEQ_INSERT_HEAD( &b->b_preparing, c, c_next ); 911 c->c_type = LLOAD_C_PREPARING; 912 913 { 914 ber_len_t max = sockbuf_max_incoming_upstream; 915 ber_sockbuf_ctrl( c->c_sb, LBER_SB_OPT_SET_MAX_INCOMING, &max ); 916 } 917 918 event = event_new( base, s, EV_READ|EV_PERSIST, connection_read_cb, c ); 919 if ( !event ) { 920 Debug( LDAP_DEBUG_ANY, "upstream_init: " 921 "Read event could not be allocated\n" ); 922 goto fail; 923 } 924 c->c_read_event = event; 925 926 event = event_new( base, s, EV_WRITE, connection_write_cb, c ); 927 if ( !event ) { 928 Debug( LDAP_DEBUG_ANY, "upstream_init: " 929 "Write event could not be allocated\n" ); 930 goto fail; 931 } 932 /* We only add the write event when we have data pending */ 933 c->c_write_event = event; 934 935 c->c_destroy = upstream_destroy; 936 c->c_unlink = upstream_unlink; 937 938 #ifdef HAVE_TLS 939 if ( c->c_is_tls == LLOAD_CLEARTEXT ) { 940 #endif /* HAVE_TLS */ 941 if ( upstream_finish( c ) ) { 942 goto fail; 943 } 944 #ifdef HAVE_TLS 945 } else if ( c->c_is_tls == LLOAD_LDAPS ) { 946 event_assign( c->c_read_event, base, s, EV_READ|EV_PERSIST, 947 upstream_tls_handshake_cb, c ); 948 event_add( c->c_read_event, c->c_read_timeout ); 949 event_assign( c->c_write_event, base, s, EV_WRITE, 950 upstream_tls_handshake_cb, c ); 951 event_add( c->c_write_event, lload_write_timeout ); 952 } else if ( c->c_is_tls == LLOAD_STARTTLS || 953 c->c_is_tls == LLOAD_STARTTLS_OPTIONAL ) { 954 BerElement *output; 955 956 checked_lock( &c->c_io_mutex ); 957 if ( (output = c->c_pendingber = ber_alloc()) == NULL ) { 958 checked_unlock( &c->c_io_mutex ); 959 goto fail; 960 } 961 ber_printf( output, "t{tit{ts}}", LDAP_TAG_MESSAGE, 962 LDAP_TAG_MSGID, c->c_next_msgid++, 963 LDAP_REQ_EXTENDED, 964 LDAP_TAG_EXOP_REQ_OID, LDAP_EXOP_START_TLS ); 965 checked_unlock( &c->c_io_mutex ); 966 967 c->c_pdu_cb = upstream_starttls; 968 CONNECTION_UNLOCK(c); 969 connection_write_cb( s, 0, c ); 970 CONNECTION_LOCK(c); 971 if ( IS_ALIVE( c, c_live ) ) { 972 event_add( c->c_read_event, c->c_read_timeout ); 973 } 974 } 975 #endif /* HAVE_TLS */ 976 CONNECTION_UNLOCK(c); 977 978 return c; 979 980 fail: 981 if ( c->c_write_event ) { 982 event_del( c->c_write_event ); 983 event_free( c->c_write_event ); 984 } 985 if ( c->c_read_event ) { 986 event_del( c->c_read_event ); 987 event_free( c->c_read_event ); 988 } 989 990 c->c_state = LLOAD_C_INVALID; 991 c->c_live--; 992 c->c_refcnt--; 993 connection_destroy( c ); 994 995 return NULL; 996 } 997 998 static void 999 upstream_unlink( LloadConnection *c ) 1000 { 1001 LloadBackend *b = c->c_backend; 1002 struct event *read_event, *write_event; 1003 TAvlnode *root; 1004 long freed, executing; 1005 1006 Debug( LDAP_DEBUG_CONNS, "upstream_unlink: " 1007 "removing upstream connid=%lu\n", 1008 c->c_connid ); 1009 CONNECTION_ASSERT_LOCKED(c); 1010 1011 assert( c->c_state != LLOAD_C_INVALID ); 1012 assert( c->c_state != LLOAD_C_DYING ); 1013 1014 c->c_state = LLOAD_C_DYING; 1015 1016 read_event = c->c_read_event; 1017 write_event = c->c_write_event; 1018 1019 root = c->c_ops; 1020 c->c_ops = NULL; 1021 executing = c->c_n_ops_executing; 1022 c->c_n_ops_executing = 0; 1023 1024 CONNECTION_UNLOCK(c); 1025 1026 freed = ldap_tavl_free( root, (AVL_FREE)operation_lost_upstream ); 1027 assert( freed == executing ); 1028 1029 /* 1030 * Avoid a deadlock: 1031 * event_del will block if the event is currently executing its callback, 1032 * that callback might be waiting to lock c->c_mutex 1033 */ 1034 if ( read_event ) { 1035 event_del( read_event ); 1036 } 1037 1038 if ( write_event ) { 1039 event_del( write_event ); 1040 } 1041 1042 checked_lock( &b->b_mutex ); 1043 if ( c->c_type == LLOAD_C_PREPARING ) { 1044 LDAP_CIRCLEQ_REMOVE( &b->b_preparing, c, c_next ); 1045 b->b_opening--; 1046 b->b_failed++; 1047 } else if ( c->c_type == LLOAD_C_BIND ) { 1048 if ( c == b->b_last_bindconn ) { 1049 LloadConnection *prev = 1050 LDAP_CIRCLEQ_LOOP_PREV( &b->b_bindconns, c, c_next ); 1051 if ( prev == c ) { 1052 b->b_last_bindconn = NULL; 1053 } else { 1054 b->b_last_bindconn = prev; 1055 } 1056 } 1057 LDAP_CIRCLEQ_REMOVE( &b->b_bindconns, c, c_next ); 1058 b->b_bindavail--; 1059 } else { 1060 if ( c == b->b_last_conn ) { 1061 LloadConnection *prev = 1062 LDAP_CIRCLEQ_LOOP_PREV( &b->b_conns, c, c_next ); 1063 if ( prev == c ) { 1064 b->b_last_conn = NULL; 1065 } else { 1066 b->b_last_conn = prev; 1067 } 1068 } 1069 LDAP_CIRCLEQ_REMOVE( &b->b_conns, c, c_next ); 1070 b->b_active--; 1071 } 1072 b->b_n_ops_executing -= executing; 1073 backend_retry( b ); 1074 checked_unlock( &b->b_mutex ); 1075 1076 CONNECTION_LOCK(c); 1077 CONNECTION_ASSERT_LOCKED(c); 1078 } 1079 1080 void 1081 upstream_destroy( LloadConnection *c ) 1082 { 1083 Debug( LDAP_DEBUG_CONNS, "upstream_destroy: " 1084 "freeing connection connid=%lu\n", 1085 c->c_connid ); 1086 1087 CONNECTION_LOCK(c); 1088 assert( c->c_state == LLOAD_C_DYING ); 1089 c->c_state = LLOAD_C_INVALID; 1090 1091 assert( c->c_ops == NULL ); 1092 1093 if ( c->c_read_event ) { 1094 event_free( c->c_read_event ); 1095 c->c_read_event = NULL; 1096 } 1097 1098 if ( c->c_write_event ) { 1099 event_free( c->c_write_event ); 1100 c->c_write_event = NULL; 1101 } 1102 1103 if ( c->c_type != LLOAD_C_BIND ) { 1104 BER_BVZERO( &c->c_sasl_bind_mech ); 1105 } 1106 connection_destroy( c ); 1107 } 1108