server.c

/*
 *  Copyright (C) 2016-2018 Red Hat, Inc.
 *  Copyright (C) 2005  Anthony Liguori <anthony@codemonkey.ws>
 *
 *  Network Block Device Server Side
 *
 *  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; under 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, see <http://www.gnu.org/licenses/>.
 */

#include "qemu/osdep.h"

#include "block/export.h"
#include "qapi/error.h"
#include "qemu/queue.h"
#include "trace.h"
#include "nbd-internal.h"
#include "qemu/units.h"

#define NBD_META_ID_BASE_ALLOCATION 0
#define NBD_META_ID_DIRTY_BITMAP 1

/*
 * NBD_MAX_BLOCK_STATUS_EXTENTS: 1 MiB of extents data. An empirical
 * constant. If an increase is needed, note that the NBD protocol
 * recommends no larger than 32 mb, so that the client won't consider
 * the reply as a denial of service attack.
 */
#define NBD_MAX_BLOCK_STATUS_EXTENTS (1 * MiB / 8)

static int system_errno_to_nbd_errno(int err)
{
    switch (err) {
    case 0:
        return NBD_SUCCESS;
    case EPERM:
    case EROFS:
        return NBD_EPERM;
    case EIO:
        return NBD_EIO;
    case ENOMEM:
        return NBD_ENOMEM;
#ifdef EDQUOT
    case EDQUOT:
#endif
    case EFBIG:
    case ENOSPC:
        return NBD_ENOSPC;
    case EOVERFLOW:
        return NBD_EOVERFLOW;
    case ENOTSUP:
#if ENOTSUP != EOPNOTSUPP
    case EOPNOTSUPP:
#endif
        return NBD_ENOTSUP;
    case ESHUTDOWN:
        return NBD_ESHUTDOWN;
    case EINVAL:
    default:
        return NBD_EINVAL;
    }
}

/* Definitions for opaque data types */

typedef struct NBDRequestData NBDRequestData;

struct NBDRequestData {
    QSIMPLEQ_ENTRY(NBDRequestData) entry;
    NBDClient *client;
    uint8_t *data;
    bool complete;
};

struct NBDExport {
    BlockExport common;

    char *name;
    char *description;
    uint64_t size;
    uint16_t nbdflags;
    QTAILQ_HEAD(, NBDClient) clients;
    QTAILQ_ENTRY(NBDExport) next;

    BlockBackend *eject_notifier_blk;
    Notifier eject_notifier;

    BdrvDirtyBitmap *export_bitmap;
    char *export_bitmap_context;
};

static QTAILQ_HEAD(, NBDExport) exports = QTAILQ_HEAD_INITIALIZER(exports);

/* NBDExportMetaContexts represents a list of contexts to be exported,
 * as selected by NBD_OPT_SET_META_CONTEXT. Also used for
 * NBD_OPT_LIST_META_CONTEXT. */
typedef struct NBDExportMetaContexts {
    NBDExport *exp;
    bool valid; /* means that negotiation of the option finished without
                   errors */
    bool base_allocation; /* export base:allocation context (block status) */
    bool bitmap; /* export qemu:dirty-bitmap:<export bitmap name> */
} NBDExportMetaContexts;

struct NBDClient {
    int refcount;
    void (*close_fn)(NBDClient *client, bool negotiated);

    NBDExport *exp;
    QCryptoTLSCreds *tlscreds;
    char *tlsauthz;
    QIOChannelSocket *sioc; /* The underlying data channel */
    QIOChannel *ioc; /* The current I/O channel which may differ (eg TLS) */

    Coroutine *recv_coroutine;

    CoMutex send_lock;
    Coroutine *send_coroutine;

    QTAILQ_ENTRY(NBDClient) next;
    int nb_requests;
    bool closing;

    uint32_t check_align; /* If non-zero, check for aligned client requests */

    bool structured_reply;
    NBDExportMetaContexts export_meta;

    uint32_t opt; /* Current option being negotiated */
    uint32_t optlen; /* remaining length of data in ioc for the option being
                        negotiated now */
};

static void nbd_client_receive_next_request(NBDClient *client);

/* Basic flow for negotiation

   Server         Client
   Negotiate

   or

   Server         Client
   Negotiate #1
                  Option
   Negotiate #2

   ----

   followed by

   Server         Client
                  Request
   Response
                  Request
   Response
                  ...
   ...
                  Request (type == 2)

*/

static inline void set_be_option_rep(NBDOptionReply *rep, uint32_t option,
                                     uint32_t type, uint32_t length)
{
    stq_be_p(&rep->magic, NBD_REP_MAGIC);
    stl_be_p(&rep->option, option);
    stl_be_p(&rep->type, type);
    stl_be_p(&rep->length, length);
}

/* Send a reply header, including length, but no payload.
 * Return -errno on error, 0 on success. */
static int nbd_negotiate_send_rep_len(NBDClient *client, uint32_t type,
                                      uint32_t len, Error **errp)
{
    NBDOptionReply rep;

    trace_nbd_negotiate_send_rep_len(client->opt, nbd_opt_lookup(client->opt),
                                     type, nbd_rep_lookup(type), len);

    assert(len < NBD_MAX_BUFFER_SIZE);

    set_be_option_rep(&rep, client->opt, type, len);
    return nbd_write(client->ioc, &rep, sizeof(rep), errp);
}

/* Send a reply header with default 0 length.
 * Return -errno on error, 0 on success. */
static int nbd_negotiate_send_rep(NBDClient *client, uint32_t type,
                                  Error **errp)
{
    return nbd_negotiate_send_rep_len(client, type, 0, errp);
}

/* Send an error reply.
 * Return -errno on error, 0 on success. */
static int GCC_FMT_ATTR(4, 0)
nbd_negotiate_send_rep_verr(NBDClient *client, uint32_t type,
                            Error **errp, const char *fmt, va_list va)
{
    ERRP_GUARD();
    g_autofree char *msg = NULL;
    int ret;
    size_t len;

    msg = g_strdup_vprintf(fmt, va);
    len = strlen(msg);
    assert(len < NBD_MAX_STRING_SIZE);
    trace_nbd_negotiate_send_rep_err(msg);
    ret = nbd_negotiate_send_rep_len(client, type, len, errp);
    if (ret < 0) {
        return ret;
    }
    if (nbd_write(client->ioc, msg, len, errp) < 0) {
        error_prepend(errp, "write failed (error message): ");
        return -EIO;
    }

    return 0;
}

/*
 * Return a malloc'd copy of @name suitable for use in an error reply.
 */
static char *
nbd_sanitize_name(const char *name)
{
    if (strnlen(name, 80) < 80) {
        return g_strdup(name);
    }
    /* XXX Should we also try to sanitize any control characters? */
    return g_strdup_printf("%.80s...", name);
}

/* Send an error reply.
 * Return -errno on error, 0 on success. */
static int GCC_FMT_ATTR(4, 5)
nbd_negotiate_send_rep_err(NBDClient *client, uint32_t type,
                           Error **errp, const char *fmt, ...)
{
    va_list va;
    int ret;

    va_start(va, fmt);
    ret = nbd_negotiate_send_rep_verr(client, type, errp, fmt, va);
    va_end(va);
    return ret;
}

/* Drop remainder of the current option, and send a reply with the
 * given error type and message. Return -errno on read or write
 * failure; or 0 if connection is still live. */
static int GCC_FMT_ATTR(4, 0)
nbd_opt_vdrop(NBDClient *client, uint32_t type, Error **errp,
              const char *fmt, va_list va)
{
    int ret = nbd_drop(client->ioc, client->optlen, errp);

    client->optlen = 0;
    if (!ret) {
        ret = nbd_negotiate_send_rep_verr(client, type, errp, fmt, va);
    }
    return ret;
}

static int GCC_FMT_ATTR(4, 5)
nbd_opt_drop(NBDClient *client, uint32_t type, Error **errp,
             const char *fmt, ...)
{
    int ret;
    va_list va;

    va_start(va, fmt);
    ret = nbd_opt_vdrop(client, type, errp, fmt, va);
    va_end(va);

    return ret;
}

static int GCC_FMT_ATTR(3, 4)
nbd_opt_invalid(NBDClient *client, Error **errp, const char *fmt, ...)
{
    int ret;
    va_list va;

    va_start(va, fmt);
    ret = nbd_opt_vdrop(client, NBD_REP_ERR_INVALID, errp, fmt, va);
    va_end(va);

    return ret;
}

/* Read size bytes from the unparsed payload of the current option.
 * Return -errno on I/O error, 0 if option was completely handled by
 * sending a reply about inconsistent lengths, or 1 on success. */
static int nbd_opt_read(NBDClient *client, void *buffer, size_t size,
                        Error **errp)
{
    if (size > client->optlen) {
        return nbd_opt_invalid(client, errp,
                               "Inconsistent lengths in option %s",
                               nbd_opt_lookup(client->opt));
    }
    client->optlen -= size;
    return qio_channel_read_all(client->ioc, buffer, size, errp) < 0 ? -EIO : 1;
}

/* Drop size bytes from the unparsed payload of the current option.
 * Return -errno on I/O error, 0 if option was completely handled by
 * sending a reply about inconsistent lengths, or 1 on success. */
static int nbd_opt_skip(NBDClient *client, size_t size, Error **errp)
{
    if (size > client->optlen) {
        return nbd_opt_invalid(client, errp,
                               "Inconsistent lengths in option %s",
                               nbd_opt_lookup(client->opt));
    }
    client->optlen -= size;
    return nbd_drop(client->ioc, size, errp) < 0 ? -EIO : 1;
}

/* nbd_opt_read_name
 *
 * Read a string with the format:
 *   uint32_t len     (<= NBD_MAX_STRING_SIZE)
 *   len bytes string (not 0-terminated)
 *
 * On success, @name will be allocated.
 * If @length is non-null, it will be set to the actual string length.
 *
 * Return -errno on I/O error, 0 if option was completely handled by
 * sending a reply about inconsistent lengths, or 1 on success.
 */
static int nbd_opt_read_name(NBDClient *client, char **name, uint32_t *length,
                             Error **errp)
{
    int ret;
    uint32_t len;
    g_autofree char *local_name = NULL;

    *name = NULL;
    ret = nbd_opt_read(client, &len, sizeof(len), errp);
    if (ret <= 0) {
        return ret;
    }
    len = cpu_to_be32(len);

    if (len > NBD_MAX_STRING_SIZE) {
        return nbd_opt_invalid(client, errp,
                               "Invalid name length: %" PRIu32, len);
    }

    local_name = g_malloc(len + 1);
    ret = nbd_opt_read(client, local_name, len, errp);
    if (ret <= 0) {
        return ret;
    }
    local_name[len] = '\0';

    if (length) {
        *length = len;
    }
    *name = g_steal_pointer(&local_name);

    return 1;
}

/* Send a single NBD_REP_SERVER reply to NBD_OPT_LIST, including payload.
 * Return -errno on error, 0 on success. */
static int nbd_negotiate_send_rep_list(NBDClient *client, NBDExport *exp,
                                       Error **errp)
{
    ERRP_GUARD();
    size_t name_len, desc_len;
    uint32_t len;
    const char *name = exp->name ? exp->name : "";
    const char *desc = exp->description ? exp->description : "";
    QIOChannel *ioc = client->ioc;
    int ret;

    trace_nbd_negotiate_send_rep_list(name, desc);
    name_len = strlen(name);
    desc_len = strlen(desc);
    assert(name_len <= NBD_MAX_STRING_SIZE && desc_len <= NBD_MAX_STRING_SIZE);
    len = name_len + desc_len + sizeof(len);
    ret = nbd_negotiate_send_rep_len(client, NBD_REP_SERVER, len, errp);
    if (ret < 0) {
        return ret;
    }

    len = cpu_to_be32(name_len);
    if (nbd_write(ioc, &len, sizeof(len), errp) < 0) {
        error_prepend(errp, "write failed (name length): ");
        return -EINVAL;
    }

    if (nbd_write(ioc, name, name_len, errp) < 0) {
        error_prepend(errp, "write failed (name buffer): ");
        return -EINVAL;
    }

    if (nbd_write(ioc, desc, desc_len, errp) < 0) {
        error_prepend(errp, "write failed (description buffer): ");
        return -EINVAL;
    }

    return 0;
}

/* Process the NBD_OPT_LIST command, with a potential series of replies.
 * Return -errno on error, 0 on success. */
static int nbd_negotiate_handle_list(NBDClient *client, Error **errp)
{
    NBDExport *exp;
    assert(client->opt == NBD_OPT_LIST);

    /* For each export, send a NBD_REP_SERVER reply. */
    QTAILQ_FOREACH(exp, &exports, next) {
        if (nbd_negotiate_send_rep_list(client, exp, errp)) {
            return -EINVAL;
        }
    }
    /* Finish with a NBD_REP_ACK. */
    return nbd_negotiate_send_rep(client, NBD_REP_ACK, errp);
}

static void nbd_check_meta_export(NBDClient *client)
{
    client->export_meta.valid &= client->exp == client->export_meta.exp;
}

/* Send a reply to NBD_OPT_EXPORT_NAME.
 * Return -errno on error, 0 on success. */
static int nbd_negotiate_handle_export_name(NBDClient *client, bool no_zeroes,
                                            Error **errp)
{
    ERRP_GUARD();
    g_autofree char *name = NULL;
    char buf[NBD_REPLY_EXPORT_NAME_SIZE] = "";
    size_t len;
    int ret;
    uint16_t myflags;

    /* Client sends:
        [20 ..  xx]   export name (length bytes)
       Server replies:
        [ 0 ..   7]   size
        [ 8 ..   9]   export flags
        [10 .. 133]   reserved     (0) [unless no_zeroes]
     */
    trace_nbd_negotiate_handle_export_name();
    if (client->optlen > NBD_MAX_STRING_SIZE) {
        error_setg(errp, "Bad length received");
        return -EINVAL;
    }
    name = g_malloc(client->optlen + 1);
    if (nbd_read(client->ioc, name, client->optlen, "export name", errp) < 0) {
        return -EIO;
    }
    name[client->optlen] = '\0';
    client->optlen = 0;

    trace_nbd_negotiate_handle_export_name_request(name);

    client->exp = nbd_export_find(name);
    if (!client->exp) {
        error_setg(errp, "export not found");
        return -EINVAL;
    }

    myflags = client->exp->nbdflags;
    if (client->structured_reply) {
        myflags |= NBD_FLAG_SEND_DF;
    }
    trace_nbd_negotiate_new_style_size_flags(client->exp->size, myflags);
    stq_be_p(buf, client->exp->size);
    stw_be_p(buf + 8, myflags);
    len = no_zeroes ? 10 : sizeof(buf);
    ret = nbd_write(client->ioc, buf, len, errp);
    if (ret < 0) {
        error_prepend(errp, "write failed: ");
        return ret;
    }

    QTAILQ_INSERT_TAIL(&client->exp->clients, client, next);
    blk_exp_ref(&client->exp->common);
    nbd_check_meta_export(client);

    return 0;
}

/* Send a single NBD_REP_INFO, with a buffer @buf of @length bytes.
 * The buffer does NOT include the info type prefix.
 * Return -errno on error, 0 if ready to send more. */
static int nbd_negotiate_send_info(NBDClient *client,
                                   uint16_t info, uint32_t length, void *buf,
                                   Error **errp)
{
    int rc;

    trace_nbd_negotiate_send_info(info, nbd_info_lookup(info), length);
    rc = nbd_negotiate_send_rep_len(client, NBD_REP_INFO,
                                    sizeof(info) + length, errp);
    if (rc < 0) {
        return rc;
    }
    info = cpu_to_be16(info);
    if (nbd_write(client->ioc, &info, sizeof(info), errp) < 0) {
        return -EIO;
    }
    if (nbd_write(client->ioc, buf, length, errp) < 0) {
        return -EIO;
    }
    return 0;
}

/* nbd_reject_length: Handle any unexpected payload.
 * @fatal requests that we quit talking to the client, even if we are able
 * to successfully send an error reply.
 * Return:
 * -errno  transmission error occurred or @fatal was requested, errp is set
 * 0       error message successfully sent to client, errp is not set
 */
static int nbd_reject_length(NBDClient *client, bool fatal, Error **errp)
{
    int ret;

    assert(client->optlen);
    ret = nbd_opt_invalid(client, errp, "option '%s' has unexpected length",
                          nbd_opt_lookup(client->opt));
    if (fatal && !ret) {
        error_setg(errp, "option '%s' has unexpected length",
                   nbd_opt_lookup(client->opt));
        return -EINVAL;
    }
    return ret;
}

/* Handle NBD_OPT_INFO and NBD_OPT_GO.
 * Return -errno on error, 0 if ready for next option, and 1 to move
 * into transmission phase.  */
static int nbd_negotiate_handle_info(NBDClient *client, Error **errp)
{
    int rc;
    g_autofree char *name = NULL;
    NBDExport *exp;
    uint16_t requests;
    uint16_t request;
    uint32_t namelen;
    bool sendname = false;
    bool blocksize = false;
    uint32_t sizes[3];
    char buf[sizeof(uint64_t) + sizeof(uint16_t)];
    uint32_t check_align = 0;
    uint16_t myflags;

    /* Client sends:
        4 bytes: L, name length (can be 0)
        L bytes: export name
        2 bytes: N, number of requests (can be 0)
        N * 2 bytes: N requests
    */
    rc = nbd_opt_read_name(client, &name, &namelen, errp);
    if (rc <= 0) {
        return rc;
    }
    trace_nbd_negotiate_handle_export_name_request(name);

    rc = nbd_opt_read(client, &requests, sizeof(requests), errp);
    if (rc <= 0) {
        return rc;
    }
    requests = be16_to_cpu(requests);
    trace_nbd_negotiate_handle_info_requests(requests);
    while (requests--) {
        rc = nbd_opt_read(client, &request, sizeof(request), errp);
        if (rc <= 0) {
            return rc;
        }
        request = be16_to_cpu(request);
        trace_nbd_negotiate_handle_info_request(request,
                                                nbd_info_lookup(request));
        /* We care about NBD_INFO_NAME and NBD_INFO_BLOCK_SIZE;
         * everything else is either a request we don't know or
         * something we send regardless of request */
        switch (request) {
        case NBD_INFO_NAME:
            sendname = true;
            break;
        case NBD_INFO_BLOCK_SIZE:
            blocksize = true;
            break;
        }
    }
    if (client->optlen) {
        return nbd_reject_length(client, false, errp);
    }

    exp = nbd_export_find(name);
    if (!exp) {
        g_autofree char *sane_name = nbd_sanitize_name(name);

        return nbd_negotiate_send_rep_err(client, NBD_REP_ERR_UNKNOWN,
                                          errp, "export '%s' not present",
                                          sane_name);
    }

    /* Don't bother sending NBD_INFO_NAME unless client requested it */
    if (sendname) {
        rc = nbd_negotiate_send_info(client, NBD_INFO_NAME, namelen, name,
                                     errp);
        if (rc < 0) {
            return rc;
        }
    }

    /* Send NBD_INFO_DESCRIPTION only if available, regardless of
     * client request */
    if (exp->description) {
        size_t len = strlen(exp->description);

        assert(len <= NBD_MAX_STRING_SIZE);
        rc = nbd_negotiate_send_info(client, NBD_INFO_DESCRIPTION,
                                     len, exp->description, errp);
        if (rc < 0) {
            return rc;
        }
    }

    /* Send NBD_INFO_BLOCK_SIZE always, but tweak the minimum size
     * according to whether the client requested it, and according to
     * whether this is OPT_INFO or OPT_GO. */
    /* minimum - 1 for back-compat, or actual if client will obey it. */
    if (client->opt == NBD_OPT_INFO || blocksize) {
        check_align = sizes[0] = blk_get_request_alignment(exp->common.blk);
    } else {
        sizes[0] = 1;
    }
    assert(sizes[0] <= NBD_MAX_BUFFER_SIZE);
    /* preferred - Hard-code to 4096 for now.
     * TODO: is blk_bs(blk)->bl.opt_transfer appropriate? */
    sizes[1] = MAX(4096, sizes[0]);
    /* maximum - At most 32M, but smaller as appropriate. */
    sizes[2] = MIN(blk_get_max_transfer(exp->common.blk), NBD_MAX_BUFFER_SIZE);
    trace_nbd_negotiate_handle_info_block_size(sizes[0], sizes[1], sizes[2]);
    sizes[0] = cpu_to_be32(sizes[0]);
    sizes[1] = cpu_to_be32(sizes[1]);
    sizes[2] = cpu_to_be32(sizes[2]);
    rc = nbd_negotiate_send_info(client, NBD_INFO_BLOCK_SIZE,
                                 sizeof(sizes), sizes, errp);
    if (rc < 0) {
        return rc;
    }

    /* Send NBD_INFO_EXPORT always */
    myflags = exp->nbdflags;
    if (client->structured_reply) {
        myflags |= NBD_FLAG_SEND_DF;
    }
    trace_nbd_negotiate_new_style_size_flags(exp->size, myflags);
    stq_be_p(buf, exp->size);
    stw_be_p(buf + 8, myflags);
    rc = nbd_negotiate_send_info(client, NBD_INFO_EXPORT,
                                 sizeof(buf), buf, errp);
    if (rc < 0) {
        return rc;
    }

    /*
     * If the client is just asking for NBD_OPT_INFO, but forgot to
     * request block sizes in a situation that would impact
     * performance, then return an error. But for NBD_OPT_GO, we
     * tolerate all clients, regardless of alignments.
     */
    if (client->opt == NBD_OPT_INFO && !blocksize &&
        blk_get_request_alignment(exp->common.blk) > 1) {
        return nbd_negotiate_send_rep_err(client,
                                          NBD_REP_ERR_BLOCK_SIZE_REQD,
                                          errp,
                                          "request NBD_INFO_BLOCK_SIZE to "
                                          "use this export");
    }

    /* Final reply */
    rc = nbd_negotiate_send_rep(client, NBD_REP_ACK, errp);
    if (rc < 0) {
        return rc;
    }

    if (client->opt == NBD_OPT_GO) {
        client->exp = exp;
        client->check_align = check_align;
        QTAILQ_INSERT_TAIL(&client->exp->clients, client, next);
        blk_exp_ref(&client->exp->common);
        nbd_check_meta_export(client);
        rc = 1;
    }
    return rc;
}


/* Handle NBD_OPT_STARTTLS. Return NULL to drop connection, or else the
 * new channel for all further (now-encrypted) communication. */
static QIOChannel *nbd_negotiate_handle_starttls(NBDClient *client,
                                                 Error **errp)
{
    QIOChannel *ioc;
    QIOChannelTLS *tioc;
    struct NBDTLSHandshakeData data = { 0 };

    assert(client->opt == NBD_OPT_STARTTLS);

    trace_nbd_negotiate_handle_starttls();
    ioc = client->ioc;

    if (nbd_negotiate_send_rep(client, NBD_REP_ACK, errp) < 0) {
        return NULL;
    }

    tioc = qio_channel_tls_new_server(ioc,
                                      client->tlscreds,
                                      client->tlsauthz,
                                      errp);
    if (!tioc) {
        return NULL;
    }

    qio_channel_set_name(QIO_CHANNEL(tioc), "nbd-server-tls");
    trace_nbd_negotiate_handle_starttls_handshake();
    data.loop = g_main_loop_new(g_main_context_default(), FALSE);
    qio_channel_tls_handshake(tioc,
                              nbd_tls_handshake,
                              &data,
                              NULL,
                              NULL);

    if (!data.complete) {
        g_main_loop_run(data.loop);
    }
    g_main_loop_unref(data.loop);
    if (data.error) {
        object_unref(OBJECT(tioc));
        error_propagate(errp, data.error);
        return NULL;
    }

    return QIO_CHANNEL(tioc);
}

/* nbd_negotiate_send_meta_context
 *
 * Send one chunk of reply to NBD_OPT_{LIST,SET}_META_CONTEXT
 *
 * For NBD_OPT_LIST_META_CONTEXT @context_id is ignored, 0 is used instead.
 */
static int nbd_negotiate_send_meta_context(NBDClient *client,
                                           const char *context,
                                           uint32_t context_id,
                                           Error **errp)
{
    NBDOptionReplyMetaContext opt;
    struct iovec iov[] = {
        {.iov_base = &opt, .iov_len = sizeof(opt)},
        {.iov_base = (void *)context, .iov_len = strlen(context)}
    };

    assert(iov[1].iov_len <= NBD_MAX_STRING_SIZE);
    if (client->opt == NBD_OPT_LIST_META_CONTEXT) {
        context_id = 0;
    }

    trace_nbd_negotiate_meta_query_reply(context, context_id);
    set_be_option_rep(&opt.h, client->opt, NBD_REP_META_CONTEXT,
                      sizeof(opt) - sizeof(opt.h) + iov[1].iov_len);
    stl_be_p(&opt.context_id, context_id);

    return qio_channel_writev_all(client->ioc, iov, 2, errp) < 0 ? -EIO : 0;
}

/* Read strlen(@pattern) bytes, and set @match to true if they match @pattern.
 * @match is never set to false.
 *
 * Return -errno on I/O error, 0 if option was completely handled by
 * sending a reply about inconsistent lengths, or 1 on success.
 *
 * Note: return code = 1 doesn't mean that we've read exactly @pattern.
 * It only means that there are no errors.
 */
static int nbd_meta_pattern(NBDClient *client, const char *pattern, bool *match,
                            Error **errp)
{
    int ret;
    char *query;
    size_t len = strlen(pattern);

    assert(len);

    query = g_malloc(len);
    ret = nbd_opt_read(client, query, len, errp);
    if (ret <= 0) {
        g_free(query);
        return ret;
    }

    if (strncmp(query, pattern, len) == 0) {
        trace_nbd_negotiate_meta_query_parse(pattern);
        *match = true;
    } else {
        trace_nbd_negotiate_meta_query_skip("pattern not matched");
    }
    g_free(query);

    return 1;
}

/*
 * Read @len bytes, and set @match to true if they match @pattern, or if @len
 * is 0 and the client is performing _LIST_. @match is never set to false.
 *
 * Return -errno on I/O error, 0 if option was completely handled by
 * sending a reply about inconsistent lengths, or 1 on success.
 *
 * Note: return code = 1 doesn't mean that we've read exactly @pattern.
 * It only means that there are no errors.
 */
static int nbd_meta_empty_or_pattern(NBDClient *client, const char *pattern,
                                     uint32_t len, bool *match, Error **errp)
{
    if (len == 0) {
        if (client->opt == NBD_OPT_LIST_META_CONTEXT) {
            *match = true;
        }
        trace_nbd_negotiate_meta_query_parse("empty");
        return 1;
    }

    if (len != strlen(pattern)) {
        trace_nbd_negotiate_meta_query_skip("different lengths");
        return nbd_opt_skip(client, len, errp);
    }

    return nbd_meta_pattern(client, pattern, match, errp);
}

/* nbd_meta_base_query
 *
 * Handle queries to 'base' namespace. For now, only the base:allocation
 * context is available.  'len' is the amount of text remaining to be read from
 * the current name, after the 'base:' portion has been stripped.
 *
 * Return -errno on I/O error, 0 if option was completely handled by
 * sending a reply about inconsistent lengths, or 1 on success.
 */
static int nbd_meta_base_query(NBDClient *client, NBDExportMetaContexts *meta,
                               uint32_t len, Error **errp)
{
    return nbd_meta_empty_or_pattern(client, "allocation", len,
                                     &meta->base_allocation, errp);
}

/* nbd_meta_bitmap_query
 *
 * Handle query to 'qemu:' namespace.
 * @len is the amount of text remaining to be read from the current name, after
 * the 'qemu:' portion has been stripped.
 *
 * Return -errno on I/O error, 0 if option was completely handled by
 * sending a reply about inconsistent lengths, or 1 on success. */
static int nbd_meta_qemu_query(NBDClient *client, NBDExportMetaContexts *meta,
                               uint32_t len, Error **errp)
{
    bool dirty_bitmap = false;
    size_t dirty_bitmap_len = strlen("dirty-bitmap:");
    int ret;

    if (!meta->exp->export_bitmap) {
        trace_nbd_negotiate_meta_query_skip("no dirty-bitmap exported");
        return nbd_opt_skip(client, len, errp);
    }

    if (len == 0) {
        if (client->opt == NBD_OPT_LIST_META_CONTEXT) {
            meta->bitmap = true;
        }
        trace_nbd_negotiate_meta_query_parse("empty");
        return 1;
    }

    if (len < dirty_bitmap_len) {
        trace_nbd_negotiate_meta_query_skip("not dirty-bitmap:");
        return nbd_opt_skip(client, len, errp);
    }

    len -= dirty_bitmap_len;
    ret = nbd_meta_pattern(client, "dirty-bitmap:", &dirty_bitmap, errp);
    if (ret <= 0) {
        return ret;
    }
    if (!dirty_bitmap) {
        trace_nbd_negotiate_meta_query_skip("not dirty-bitmap:");
        return nbd_opt_skip(client, len, errp);
    }

    trace_nbd_negotiate_meta_query_parse("dirty-bitmap:");

    return nbd_meta_empty_or_pattern(
            client, meta->exp->export_bitmap_context +
            strlen("qemu:dirty_bitmap:"), len, &meta->bitmap, errp);
}

/* nbd_negotiate_meta_query
 *
 * Parse namespace name and call corresponding function to parse body of the
 * query.
 *
 * The only supported namespaces are 'base' and 'qemu'.
 *
 * The function aims not wasting time and memory to read long unknown namespace
 * names.
 *
 * Return -errno on I/O error, 0 if option was completely handled by
 * sending a reply about inconsistent lengths, or 1 on success. */
static int nbd_negotiate_meta_query(NBDClient *client,
                                    NBDExportMetaContexts *meta, Error **errp)
{
    /*
     * Both 'qemu' and 'base' namespaces have length = 5 including a
     * colon. If another length namespace is later introduced, this
     * should certainly be refactored.
     */
    int ret;
    size_t ns_len = 5;
    char ns[5];
    uint32_t len;

    ret = nbd_opt_read(client, &len, sizeof(len), errp);
    if (ret <= 0) {
        return ret;
    }
    len = cpu_to_be32(len);

    if (len > NBD_MAX_STRING_SIZE) {
        trace_nbd_negotiate_meta_query_skip("length too long");
        return nbd_opt_skip(client, len, errp);
    }
    if (len < ns_len) {
        trace_nbd_negotiate_meta_query_skip("length too short");
        return nbd_opt_skip(client, len, errp);
    }

    len -= ns_len;
    ret = nbd_opt_read(client, ns, ns_len, errp);
    if (ret <= 0) {
        return ret;
    }

    if (!strncmp(ns, "base:", ns_len)) {
        trace_nbd_negotiate_meta_query_parse("base:");
        return nbd_meta_base_query(client, meta, len, errp);
    } else if (!strncmp(ns, "qemu:", ns_len)) {
        trace_nbd_negotiate_meta_query_parse("qemu:");
        return nbd_meta_qemu_query(client, meta, len, errp);
    }

    trace_nbd_negotiate_meta_query_skip("unknown namespace");
    return nbd_opt_skip(client, len, errp);
}

/* nbd_negotiate_meta_queries
 * Handle NBD_OPT_LIST_META_CONTEXT and NBD_OPT_SET_META_CONTEXT
 *
 * Return -errno on I/O error, or 0 if option was completely handled. */
static int nbd_negotiate_meta_queries(NBDClient *client,
                                      NBDExportMetaContexts *meta, Error **errp)
{
    int ret;
    g_autofree char *export_name = NULL;
    NBDExportMetaContexts local_meta;
    uint32_t nb_queries;
    int i;

    if (!client->structured_reply) {
        return nbd_opt_invalid(client, errp,
                               "request option '%s' when structured reply "
                               "is not negotiated",
                               nbd_opt_lookup(client->opt));
    }

    if (client->opt == NBD_OPT_LIST_META_CONTEXT) {
        /* Only change the caller's meta on SET. */