ext/libcouchbase/src/mcserver/negotiate.cc in libcouchbase-0.3.3 vs ext/libcouchbase/src/mcserver/negotiate.cc in libcouchbase-1.0.0
- old
+ new
@@ -14,28 +14,34 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <algorithm>
+#include <string>
+#include <sstream>
+#include <vector>
+
#include "packetutils.h"
#include "mcserver.h"
#include "logging.h"
#include "settings.h"
#include <lcbio/lcbio.h>
#include <lcbio/timer-ng.h>
#include <lcbio/ssl.h>
#include <cbsasl/cbsasl.h>
#include "negotiate.h"
#include "ctx-log-inl.h"
+#include "auth-priv.h"
using namespace lcb;
#define LOGARGS(ctx, lvl) ctx->settings, "negotiation", LCB_LOG_##lvl, __FILE__, __LINE__
static void cleanup_negotiated(SessionInfo* info);
static void handle_ioerr(lcbio_CTX *ctx, lcb_error_t err);
#define SESSREQ_LOGFMT "<%s:%s> (SASLREQ=%p) "
+
static void timeout_handler(void *arg);
#define SESSREQ_LOGID(s) get_ctx_host(s->ctx), get_ctx_port(s->ctx), (void*)s
static void
@@ -58,18 +64,22 @@
}
bool setup(const lcbio_NAMEINFO& nistrs, const lcb_host_t& host,
const lcb::Authenticator& auth);
void start(lcbio_SOCKET *sock);
+ void send_list_mechs();
bool send_hello();
bool send_step(const lcb::MemcachedResponse& packet);
bool read_hello(const lcb::MemcachedResponse& packet);
void send_auth(const char *sasl_data, unsigned ndata);
void handle_read(lcbio_CTX *ioctx);
+ bool maybe_select_bucket();
enum MechStatus { MECH_UNAVAILABLE, MECH_NOT_NEEDED, MECH_OK };
MechStatus set_chosen_mech(std::string& mechlist, const char **data, unsigned int *ndata);
+ bool request_errmap();
+ bool update_errmap(const lcb::MemcachedResponse& packet);
SessionRequestImpl(lcbio_CONNDONE_cb callback, void *data, uint32_t timeout, lcbio_TABLE *iot, lcb_settings* settings_)
: ctx(NULL), cb(callback), cbdata(data),
timer(lcbio_timer_new(iot, this, timeout_handler)),
last_err(LCB_SUCCESS), info(NULL),
@@ -116,12 +126,33 @@
lcbio_unref(s);
delete this;
}
- void set_error(lcb_error_t error, const char *msg = "") {
- lcb_log(LOGARGS(this, ERR), SESSREQ_LOGFMT "Error: 0x%x, %s", SESSREQ_LOGID(this), error, msg);
+ void set_error(lcb_error_t error, const char *msg, const lcb::MemcachedResponse *packet = NULL) {
+ char *err_ref = NULL, *err_ctx = NULL;
+ if (packet) {
+ MemcachedResponse::parse_enhanced_error(packet->value(), packet->vallen(), &err_ref, &err_ctx);
+ }
+ if (err_ref || err_ctx) {
+ std::stringstream emsg;
+ if (err_ref) {
+ emsg << "ref: \"" << err_ref << "\"";
+ }
+ if (err_ctx) {
+ if (err_ref) {
+ emsg << ", ";
+ }
+ emsg << "context: \"" << err_ctx << "\"";
+ }
+ lcb_log(LOGARGS(this, ERR), SESSREQ_LOGFMT "Error: 0x%x, %s (%s)",
+ SESSREQ_LOGID(this), error, msg, emsg.str().c_str());
+ free(err_ref);
+ free(err_ctx);
+ } else {
+ lcb_log(LOGARGS(this, ERR), SESSREQ_LOGFMT "Error: 0x%x, %s", SESSREQ_LOGID(this), error, msg);
+ }
if (last_err == LCB_SUCCESS) {
last_err = error;
}
}
@@ -131,10 +162,11 @@
union {
cbsasl_secret_t secret;
char buffer[256];
} u_auth;
+ std::string username;
lcbio_CTX *ctx;
lcbio_CONNDONE_cb cb;
void *cbdata;
lcbio_pTIMER timer;
@@ -150,20 +182,16 @@
static int
sasl_get_username(void *context, int id, const char **result, unsigned int *len)
{
SessionRequestImpl *ctx = SessionRequestImpl::get(context);
- const char *u = NULL, *p = NULL;
if (!context || !result || (id != CBSASL_CB_USER && id != CBSASL_CB_AUTHNAME)) {
return SASL_BADPARAM;
}
- lcbauth_get_upass(ctx->settings->auth, &u, &p);
- *result = u;
- if (len) {
- *len = (unsigned int)strlen(*result);
- }
+ *result = ctx->username.c_str();
+ *len = ctx->username.size();
return SASL_OK;
}
static int
@@ -205,26 +233,25 @@
for (size_t ii = 0; ii < 3; ii++) {
sasl_callbacks[ii].context = this;
}
- const char *pass = NULL, *user = NULL;
- lcbauth_get_upass(&auth, &user, &pass);
+ // Get the credentials
+ username = auth.username_for(settings->bucket);
+ const std::string& pass = auth.password_for(settings->bucket);
- if (pass) {
- unsigned long pwlen = (unsigned long)strlen(pass);
+ if (!pass.empty()) {
size_t maxlen = sizeof(u_auth.buffer) - offsetof(cbsasl_secret_t, data);
- u_auth.secret.len = pwlen;
+ u_auth.secret.len = pass.size();
- if (pwlen < maxlen) {
- memcpy(u_auth.secret.data, pass, pwlen);
+ if (pass.size() < maxlen) {
+ memcpy(u_auth.secret.data, pass.c_str(), pass.size());
} else {
return false;
}
}
-
cbsasl_error_t saslerr = cbsasl_client_new(
"couchbase", host.host, nistrs.local, nistrs.remote,
sasl_callbacks, 0, &sasl_client);
return saslerr == SASL_OK;
}
@@ -321,10 +348,15 @@
{
lcb_U16 features[MEMCACHED_TOTAL_HELLO_FEATURES];
unsigned nfeatures = 0;
features[nfeatures++] = PROTOCOL_BINARY_FEATURE_TLS;
+ features[nfeatures++] = PROTOCOL_BINARY_FEATURE_XATTR;
+ features[nfeatures++] = PROTOCOL_BINARY_FEATURE_SELECT_BUCKET;
+ if (settings->use_errmap) {
+ features[nfeatures++] = PROTOCOL_BINARY_FEATURE_XERROR;
+ }
if (settings->tcp_nodelay) {
features[nfeatures++] = PROTOCOL_BINARY_FEATURE_TCPNODELAY;
}
#ifndef LCB_NO_SNAPPY
@@ -357,14 +389,23 @@
lcbio_ctx_put(ctx, clistr, nclistr);
for (size_t ii = 0; ii < nfeatures; ii++) {
lcb_U16 tmp = htons(features[ii]);
lcbio_ctx_put(ctx, &tmp, sizeof tmp);
}
+
lcbio_ctx_rwant(ctx, 24);
return true;
}
+void
+SessionRequestImpl::send_list_mechs()
+{
+ lcb::MemcachedRequest req(PROTOCOL_BINARY_CMD_SASL_LIST_MECHS);
+ lcbio_ctx_put(ctx, req.data(), req.size());
+ LCBIO_CTX_RSCHEDULE(ctx, 24);
+}
+
bool
SessionRequestImpl::read_hello(const lcb::MemcachedResponse& resp)
{
/* some caps */
const char *cur;
@@ -372,33 +413,89 @@
const char *limit = payload + resp.bodylen();
for (cur = payload; cur < limit; cur += 2) {
lcb_U16 tmp;
memcpy(&tmp, cur, sizeof(tmp));
tmp = ntohs(tmp);
- lcb_log(LOGARGS(this, DEBUG), SESSREQ_LOGFMT "Found feature 0x%x (%s)", SESSREQ_LOGID(this), tmp, protocol_feature_2_text(tmp));
+ lcb_log(LOGARGS(this, DEBUG), SESSREQ_LOGFMT "Server supports feature: 0x%x (%s)", SESSREQ_LOGID(this), tmp, protocol_feature_2_text(tmp));
info->server_features.push_back(tmp);
}
return true;
}
-typedef enum {
- SREQ_S_WAIT,
- SREQ_S_AUTHDONE,
- SREQ_S_HELLODONE,
- SREQ_S_ERROR
-} sreq_STATE;
+bool
+SessionRequestImpl::request_errmap() {
+ lcb::MemcachedRequest hdr(PROTOCOL_BINARY_CMD_GET_ERROR_MAP);
+ uint16_t version = htons(1);
+ hdr.sizes(0, 0, 2);
+ const char *p = reinterpret_cast<const char *>(&version);
+ lcbio_ctx_put(ctx, hdr.data(), hdr.size());
+ lcbio_ctx_put(ctx, p, 2);
+ lcbio_ctx_rwant(ctx, 24);
+ return true;
+}
+
+bool
+SessionRequestImpl::update_errmap(const lcb::MemcachedResponse& resp)
+{
+ // Get the error map object
+ using lcb::errmap::ErrorMap;
+
+ std::string errmsg;
+ ErrorMap& mm = *settings->errmap;
+ ErrorMap::ParseStatus status = mm.parse(
+ resp.body<const char*>(), resp.bodylen(), errmsg);
+
+ if (status != ErrorMap::UPDATED && status != ErrorMap::NOT_UPDATED) {
+ errmsg = "Couldn't update error map: " + errmsg;
+ set_error(LCB_PROTOCOL_ERROR, errmsg.c_str());
+ return false;
+ }
+
+ return true;
+}
+
+// Returns true if sending the SELECT_BUCKET command, false otherwise.
+bool
+SessionRequestImpl::maybe_select_bucket() {
+
+ // Only send a SELECT_BUCKET if we have the SELECT_BUCKET bit enabled.
+ if (!info->has_feature(PROTOCOL_BINARY_FEATURE_SELECT_BUCKET)) {
+ return false;
+ }
+
+ if (!settings->select_bucket) {
+ lcb_log(LOGARGS(this, WARN), SESSREQ_LOGFMT "SELECT_BUCKET Disabled by application", SESSREQ_LOGID(this));
+ return false;
+ }
+
+ // send the SELECT_BUCKET command:
+ lcb_log(LOGARGS(this, INFO), SESSREQ_LOGFMT "Sending SELECT_BUCKET", SESSREQ_LOGID(this));
+ lcb::MemcachedRequest req(PROTOCOL_BINARY_CMD_SELECT_BUCKET);
+ req.sizes(0, strlen(settings->bucket), 0);
+ lcbio_ctx_put(ctx, req.data(), req.size());
+ lcbio_ctx_put(ctx, settings->bucket, strlen(settings->bucket));
+ LCBIO_CTX_RSCHEDULE(ctx, 24);
+ return true;
+}
+
+static bool isUnsupported(uint16_t status) {
+ return status == PROTOCOL_BINARY_RESPONSE_NOT_SUPPORTED ||
+ status == PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND ||
+ status == PROTOCOL_BINARY_RESPONSE_EACCESS;
+}
+
/**
* It's assumed the server buffers will be reset upon close(), so we must make
* sure to _not_ release the ringbuffer if that happens.
*/
void
SessionRequestImpl::handle_read(lcbio_CTX *ioctx)
{
lcb::MemcachedResponse resp;
unsigned required;
- sreq_STATE state = SREQ_S_WAIT;
+ bool completed = false;
GT_NEXT_PACKET:
if (!resp.load(ioctx, &required)) {
LCBIO_CTX_RSCHEDULE(ioctx, required);
@@ -413,72 +510,96 @@
std::string mechs(resp.body<const char*>(), resp.bodylen());
MechStatus mechrc = set_chosen_mech(mechs, &mechlist_data, &nmechlist_data);
if (mechrc == MECH_OK) {
send_auth(mechlist_data, nmechlist_data);
- state = SREQ_S_WAIT;
} else if (mechrc == MECH_UNAVAILABLE) {
- state = SREQ_S_ERROR;
+ // Do nothing - error already set
} else {
- state = SREQ_S_HELLODONE;
+ completed = !maybe_select_bucket();
}
break;
}
case PROTOCOL_BINARY_CMD_SASL_AUTH: {
if (status == PROTOCOL_BINARY_RESPONSE_SUCCESS) {
- send_hello();
- state = SREQ_S_AUTHDONE;
+ completed = !maybe_select_bucket();
break;
- }
-
- if (status != PROTOCOL_BINARY_RESPONSE_AUTH_CONTINUE) {
- set_error(LCB_AUTH_ERROR, "SASL AUTH failed");
- state = SREQ_S_ERROR;
+ } else if (status == PROTOCOL_BINARY_RESPONSE_AUTH_CONTINUE) {
+ send_step(resp);
+ } else {
+ set_error(LCB_AUTH_ERROR, "SASL AUTH failed", &resp);
break;
}
- if (send_step(resp) && send_hello()) {
- state = SREQ_S_WAIT;
- } else {
- state = SREQ_S_ERROR;
- }
break;
}
case PROTOCOL_BINARY_CMD_SASL_STEP: {
- if (status != PROTOCOL_BINARY_RESPONSE_SUCCESS) {
- lcb_log(LOGARGS(this, WARN), SESSREQ_LOGFMT "SASL auth failed with STATUS=0x%x", SESSREQ_LOGID(this), status);
- set_error(LCB_AUTH_ERROR, "SASL Step Failed");
- state = SREQ_S_ERROR;
+ if (status == PROTOCOL_BINARY_RESPONSE_SUCCESS) {
+ completed = !maybe_select_bucket();
} else {
- /* Wait for pipelined HELLO response */
- state = SREQ_S_AUTHDONE;
+ lcb_log(LOGARGS(this, WARN), SESSREQ_LOGFMT "SASL auth failed with STATUS=0x%x", SESSREQ_LOGID(this), status);
+ set_error(LCB_AUTH_ERROR, "SASL Step failed", &resp);
}
break;
}
case PROTOCOL_BINARY_CMD_HELLO: {
- state = SREQ_S_HELLODONE;
if (status == PROTOCOL_BINARY_RESPONSE_SUCCESS) {
if (!read_hello(resp)) {
set_error(LCB_PROTOCOL_ERROR, "Couldn't parse HELLO");
+ break;
}
- } else if (status == PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND ||
- status == PROTOCOL_BINARY_RESPONSE_NOT_SUPPORTED) {
+ } else if (isUnsupported(status)) {
lcb_log(LOGARGS(this, DEBUG), SESSREQ_LOGFMT "Server does not support HELLO", SESSREQ_LOGID(this));
- /* nothing */
} else {
- set_error(LCB_PROTOCOL_ERROR, "Hello response unexpected");
- state = SREQ_S_ERROR;
+ lcb_log(LOGARGS(this, ERROR), SESSREQ_LOGFMT "Unexpected status 0x%x received for HELLO", SESSREQ_LOGID(this), status);
+ set_error(LCB_PROTOCOL_ERROR, "Hello response unexpected", &resp);
+ break;
}
+
+ if (info->has_feature(PROTOCOL_BINARY_FEATURE_XERROR)) {
+ request_errmap();
+ } else {
+ lcb_log(LOGARGS(this, TRACE), SESSREQ_LOGFMT "GET_ERRORMAP unsupported/disabled", SESSREQ_LOGID(this));
+ }
+
+ // In any event, it's also time to send the LIST_MECHS request
+ send_list_mechs();
break;
}
+ case PROTOCOL_BINARY_CMD_GET_ERROR_MAP: {
+ if (status == PROTOCOL_BINARY_RESPONSE_SUCCESS) {
+ if (!update_errmap(resp)) {
+ }
+ } else if (isUnsupported(status)) {
+ lcb_log(LOGARGS(this, DEBUG), SESSREQ_LOGFMT "Server does not support GET_ERRMAP (0x%x)", SESSREQ_LOGID(this), status);
+ } else {
+ lcb_log(LOGARGS(this, ERROR), SESSREQ_LOGFMT "Unexpected status 0x%x received for GET_ERRMAP", SESSREQ_LOGID(this), status);
+ set_error(LCB_PROTOCOL_ERROR, "GET_ERRMAP response unexpected", &resp);
+ }
+ // Note, there is no explicit state transition here. LIST_MECHS is
+ // pipelined after this request.
+ break;
+ }
+
+ case PROTOCOL_BINARY_CMD_SELECT_BUCKET: {
+ if (status == PROTOCOL_BINARY_RESPONSE_SUCCESS) {
+ completed = true;
+ } else if (status == PROTOCOL_BINARY_RESPONSE_EACCESS) {
+ set_error(LCB_AUTH_ERROR, "Provided credentials not allowed for bucket", &resp);
+ } else {
+ lcb_log(LOGARGS(this, ERROR), SESSREQ_LOGFMT "Unexpected status 0x%x received for SELECT_BUCKET", SESSREQ_LOGID(this), status);
+ set_error(LCB_PROTOCOL_ERROR, "Other auth error", &resp);
+ }
+ break;
+ }
+
default: {
- state = SREQ_S_ERROR;
lcb_log(LOGARGS(this, ERROR), SESSREQ_LOGFMT "Received unknown response. OP=0x%x. RC=0x%x", SESSREQ_LOGID(this), resp.opcode(), resp.status());
- set_error(LCB_NOT_SUPPORTED, "Received unknown response");
+ set_error(LCB_NOT_SUPPORTED, "Received unknown response", &resp);
break;
}
}
// We need to release the packet's buffers before actually destroying the
@@ -487,13 +608,11 @@
// Once there is no more any dependencies on the buffers, we can succeed
// or fail the request, potentially destroying the underlying connection
if (has_error()) {
fail();
- } else if (state == SREQ_S_ERROR) {
- fail(LCB_ERROR, "FIXME: Error code set without description");
- } else if (state == SREQ_S_HELLODONE) {
+ } else if (completed) {
success();
} else {
goto GT_NEXT_PACKET;
}
}
@@ -534,12 +653,16 @@
set_error(LCB_EINTERNAL, "Couldn't start SASL client");
lcbio_async_signal(timer);
return;
}
- lcb::MemcachedRequest hdr(PROTOCOL_BINARY_CMD_SASL_LIST_MECHS);
- lcbio_ctx_put(ctx, hdr.data(), hdr.size());
+ if (settings->send_hello) {
+ send_hello();
+ } else {
+ lcb_log(LOGARGS(this, INFO), SESSREQ_LOGFMT "HELLO negotiation disabled by user", SESSREQ_LOGID(this));
+ send_list_mechs();
+ }
LCBIO_CTX_RSCHEDULE(ctx, 24);
}
SessionRequestImpl::~SessionRequestImpl()
{
@@ -553,13 +676,9 @@
lcbio_ctx_close(ctx, NULL, NULL);
}
if (sasl_client) {
cbsasl_dispose(&sasl_client);
}
-}
-
-void lcb::sessreq_cancel(SessionRequest *sreq) {
- sreq->cancel();
}
SessionRequest *
SessionRequest::start(lcbio_SOCKET *sock, lcb_settings_st *settings,
uint32_t tmo, lcbio_CONNDONE_cb callback, void *data)