6 files changed, 203 insertions, 51 deletions

diff --git a/net/ceph/auth_x.c b/net/ceph/auth_x.c
index 7e38b729696a..15845814a0f2 100644
--- a/net/ceph/auth_x.c
+++ b/net/ceph/auth_x.c
@@ -8,6 +8,7 @@
 
 #include <linux/ceph/decode.h>
 #include <linux/ceph/auth.h>
+#include <linux/ceph/messenger.h>
 
 #include "crypto.h"
 #include "auth_x.h"
@@ -293,6 +294,11 @@ static int ceph_x_build_authorizer(struct ceph_auth_client *ac,
         dout("build_authorizer for %s %p\n",
              ceph_entity_type_name(th->service), au);
 
+        ceph_crypto_key_destroy(&au->session_key);
+        ret = ceph_crypto_key_clone(&au->session_key, &th->session_key);
+        if (ret)
+                return ret;
+
         maxlen = sizeof(*msg_a) + sizeof(msg_b) +
                 ceph_x_encrypt_buflen(ticket_blob_len);
         dout("  need len %d\n", maxlen);
@@ -302,8 +308,10 @@ static int ceph_x_build_authorizer(struct ceph_auth_client *ac,
         }
         if (!au->buf) {
                 au->buf = ceph_buffer_new(maxlen, GFP_NOFS);
-                if (!au->buf)
+                if (!au->buf) {
+                        ceph_crypto_key_destroy(&au->session_key);
                         return -ENOMEM;
+                }
         }
         au->service = th->service;
         au->secret_id = th->secret_id;
@@ -329,7 +337,7 @@ static int ceph_x_build_authorizer(struct ceph_auth_client *ac,
         get_random_bytes(&au->nonce, sizeof(au->nonce));
         msg_b.struct_v = 1;
         msg_b.nonce = cpu_to_le64(au->nonce);
-        ret = ceph_x_encrypt(&th->session_key, &msg_b, sizeof(msg_b),
+        ret = ceph_x_encrypt(&au->session_key, &msg_b, sizeof(msg_b),
                              p, end - p);
         if (ret < 0)
                 goto out_buf;
@@ -560,6 +568,8 @@ static int ceph_x_create_authorizer(
         auth->authorizer_buf_len = au->buf->vec.iov_len;
         auth->authorizer_reply_buf = au->reply_buf;
         auth->authorizer_reply_buf_len = sizeof (au->reply_buf);
+        auth->sign_message = ac->ops->sign_message;
+        auth->check_message_signature = ac->ops->check_message_signature;
 
         return 0;
 }
@@ -588,17 +598,13 @@ static int ceph_x_verify_authorizer_reply(struct ceph_auth_client *ac,
                                           struct ceph_authorizer *a, size_t len)
 {
         struct ceph_x_authorizer *au = (void *)a;
-        struct ceph_x_ticket_handler *th;
         int ret = 0;
         struct ceph_x_authorize_reply reply;
         void *preply = &reply;
         void *p = au->reply_buf;
         void *end = p + sizeof(au->reply_buf);
 
-        th = get_ticket_handler(ac, au->service);
-        if (IS_ERR(th))
-                return PTR_ERR(th);
-        ret = ceph_x_decrypt(&th->session_key, &p, end, &preply, sizeof(reply));
+        ret = ceph_x_decrypt(&au->session_key, &p, end, &preply, sizeof(reply));
         if (ret < 0)
                 return ret;
         if (ret != sizeof(reply))
@@ -618,6 +624,7 @@ static void ceph_x_destroy_authorizer(struct ceph_auth_client *ac,
 {
         struct ceph_x_authorizer *au = (void *)a;
 
+        ceph_crypto_key_destroy(&au->session_key);
         ceph_buffer_put(au->buf);
         kfree(au);
 }
@@ -663,6 +670,59 @@ static void ceph_x_invalidate_authorizer(struct ceph_auth_client *ac,
                 memset(&th->validity, 0, sizeof(th->validity));
 }
 
+static int calcu_signature(struct ceph_x_authorizer *au,
+                           struct ceph_msg *msg, __le64 *sig)
+{
+        int ret;
+        char tmp_enc[40];
+        __le32 tmp[5] = {
+                16u, msg->hdr.crc, msg->footer.front_crc,
+                msg->footer.middle_crc, msg->footer.data_crc,
+        };
+        ret = ceph_x_encrypt(&au->session_key, &tmp, sizeof(tmp),
+                             tmp_enc, sizeof(tmp_enc));
+        if (ret < 0)
+                return ret;
+        *sig = *(__le64*)(tmp_enc + 4);
+        return 0;
+}
+
+static int ceph_x_sign_message(struct ceph_auth_handshake *auth,
+                               struct ceph_msg *msg)
+{
+        int ret;
+        if (!auth->authorizer)
+                return 0;
+        ret = calcu_signature((struct ceph_x_authorizer *)auth->authorizer,
+                              msg, &msg->footer.sig);
+        if (ret < 0)
+                return ret;
+        msg->footer.flags |= CEPH_MSG_FOOTER_SIGNED;
+        return 0;
+}
+
+static int ceph_x_check_message_signature(struct ceph_auth_handshake *auth,
+                                          struct ceph_msg *msg)
+{
+        __le64 sig_check;
+        int ret;
+
+        if (!auth->authorizer)
+                return 0;
+        ret = calcu_signature((struct ceph_x_authorizer *)auth->authorizer,
+                              msg, &sig_check);
+        if (ret < 0)
+                return ret;
+        if (sig_check == msg->footer.sig)
+                return 0;
+        if (msg->footer.flags & CEPH_MSG_FOOTER_SIGNED)
+                dout("ceph_x_check_message_signature %p has signature %llx "
+                     "expect %llx\n", msg, msg->footer.sig, sig_check);
+        else
+                dout("ceph_x_check_message_signature %p sender did not set "
+                     "CEPH_MSG_FOOTER_SIGNED\n", msg);
+        return -EBADMSG;
+}
 
 static const struct ceph_auth_client_ops ceph_x_ops = {
         .name = "x",
@@ -677,6 +737,8 @@ static const struct ceph_auth_client_ops ceph_x_ops = {
         .invalidate_authorizer = ceph_x_invalidate_authorizer,
         .reset =  ceph_x_reset,
         .destroy = ceph_x_destroy,
+        .sign_message = ceph_x_sign_message,
+        .check_message_signature = ceph_x_check_message_signature,
 };
 
 
diff --git a/net/ceph/auth_x.h b/net/ceph/auth_x.h
index 65ee72082d99..e8b7c6917d47 100644
--- a/net/ceph/auth_x.h
+++ b/net/ceph/auth_x.h
@@ -26,6 +26,7 @@ struct ceph_x_ticket_handler {
 
 
 struct ceph_x_authorizer {
+        struct ceph_crypto_key session_key;
         struct ceph_buffer *buf;
         unsigned int service;
         u64 nonce;
diff --git a/net/ceph/buffer.c b/net/ceph/buffer.c
index 621b5f65407f..add5f921a0ff 100644
--- a/net/ceph/buffer.c
+++ b/net/ceph/buffer.c
@@ -6,7 +6,7 @@
 
 #include <linux/ceph/buffer.h>
 #include <linux/ceph/decode.h>
-#include <linux/ceph/libceph.h> /* for ceph_kv{malloc,free} */
+#include <linux/ceph/libceph.h> /* for ceph_kvmalloc */
 
 struct ceph_buffer *ceph_buffer_new(size_t len, gfp_t gfp)
 {
@@ -35,7 +35,7 @@ void ceph_buffer_release(struct kref *kref)
         struct ceph_buffer *b = container_of(kref, struct ceph_buffer, kref);
 
         dout("buffer_release %p\n", b);
-        ceph_kvfree(b->vec.iov_base);
+        kvfree(b->vec.iov_base);
         kfree(b);
 }
 EXPORT_SYMBOL(ceph_buffer_release);
diff --git a/net/ceph/ceph_common.c b/net/ceph/ceph_common.c
index 58fbfe134f93..5d5ab67f516d 100644
--- a/net/ceph/ceph_common.c
+++ b/net/ceph/ceph_common.c
@@ -184,14 +184,6 @@ void *ceph_kvmalloc(size_t size, gfp_t flags)
         return __vmalloc(size, flags | __GFP_HIGHMEM, PAGE_KERNEL);
 }
 
-void ceph_kvfree(const void *ptr)
-{
-        if (is_vmalloc_addr(ptr))
-                vfree(ptr);
-        else
-                kfree(ptr);
-}
-
 
 static int parse_fsid(const char *str, struct ceph_fsid *fsid)
 {
@@ -245,6 +237,8 @@ enum {
         Opt_noshare,
         Opt_crc,
         Opt_nocrc,
+        Opt_cephx_require_signatures,
+        Opt_nocephx_require_signatures,
 };
 
 static match_table_t opt_tokens = {
@@ -263,6 +257,8 @@ static match_table_t opt_tokens = {
         {Opt_noshare, "noshare"},
         {Opt_crc, "crc"},
         {Opt_nocrc, "nocrc"},
+        {Opt_cephx_require_signatures, "cephx_require_signatures"},
+        {Opt_nocephx_require_signatures, "nocephx_require_signatures"},
         {-1, NULL}
 };
 
@@ -461,6 +457,12 @@ ceph_parse_options(char *options, const char *dev_name,
                 case Opt_nocrc:
                         opt->flags |= CEPH_OPT_NOCRC;
                         break;
+                case Opt_cephx_require_signatures:
+                        opt->flags &= ~CEPH_OPT_NOMSGAUTH;
+                        break;
+                case Opt_nocephx_require_signatures:
+                        opt->flags |= CEPH_OPT_NOMSGAUTH;
+                        break;
 
                 default:
                         BUG_ON(token);
@@ -504,6 +506,9 @@ struct ceph_client *ceph_create_client(struct ceph_options *opt, void *private,
         init_waitqueue_head(&client->auth_wq);
         client->auth_err = 0;
 
+        if (!ceph_test_opt(client, NOMSGAUTH))
+                required_features |= CEPH_FEATURE_MSG_AUTH;
+
         client->extra_mon_dispatch = NULL;
         client->supported_features = CEPH_FEATURES_SUPPORTED_DEFAULT |
                 supported_features;
diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
index 8d1653caffdb..33a2f201e460 100644
--- a/net/ceph/messenger.c
+++ b/net/ceph/messenger.c
@@ -1196,8 +1196,18 @@ static void prepare_write_message_footer(struct ceph_connection *con)
         dout("prepare_write_message_footer %p\n", con);
         con->out_kvec_is_msg = true;
         con->out_kvec[v].iov_base = &m->footer;
-        con->out_kvec[v].iov_len = sizeof(m->footer);
-        con->out_kvec_bytes += sizeof(m->footer);
+        if (con->peer_features & CEPH_FEATURE_MSG_AUTH) {
+                if (con->ops->sign_message)
+                        con->ops->sign_message(con, m);
+                else
+                        m->footer.sig = 0;
+                con->out_kvec[v].iov_len = sizeof(m->footer);
+                con->out_kvec_bytes += sizeof(m->footer);
+        } else {
+                m->old_footer.flags = m->footer.flags;
+                con->out_kvec[v].iov_len = sizeof(m->old_footer);
+                con->out_kvec_bytes += sizeof(m->old_footer);
+        }
         con->out_kvec_left++;
         con->out_more = m->more_to_follow;
         con->out_msg_done = true;
@@ -2249,6 +2259,7 @@ static int read_partial_message(struct ceph_connection *con)
         int ret;
         unsigned int front_len, middle_len, data_len;
         bool do_datacrc = !con->msgr->nocrc;
+        bool need_sign = (con->peer_features & CEPH_FEATURE_MSG_AUTH);
         u64 seq;
         u32 crc;
 
@@ -2361,12 +2372,21 @@ static int read_partial_message(struct ceph_connection *con)
         }
 
         /* footer */
-        size = sizeof (m->footer);
+        if (need_sign)
+                size = sizeof(m->footer);
+        else
+                size = sizeof(m->old_footer);
+
         end += size;
         ret = read_partial(con, end, size, &m->footer);
         if (ret <= 0)
                 return ret;
 
+        if (!need_sign) {
+                m->footer.flags = m->old_footer.flags;
+                m->footer.sig = 0;
+        }
+
         dout("read_partial_message got msg %p %d (%u) + %d (%u) + %d (%u)\n",
              m, front_len, m->footer.front_crc, middle_len,
              m->footer.middle_crc, data_len, m->footer.data_crc);
@@ -2390,6 +2410,12 @@ static int read_partial_message(struct ceph_connection *con)
                 return -EBADMSG;
         }
 
+        if (need_sign && con->ops->check_message_signature &&
+            con->ops->check_message_signature(con, m)) {
+                pr_err("read_partial_message %p signature check failed\n", m);
+                return -EBADMSG;
+        }
+
         return 1; /* done! */
 }
 
@@ -3288,7 +3314,7 @@ static int ceph_con_in_msg_alloc(struct ceph_connection *con, int *skip)
 static void ceph_msg_free(struct ceph_msg *m)
 {
         dout("%s %p\n", __func__, m);
-        ceph_kvfree(m->front.iov_base);
+        kvfree(m->front.iov_base);
         kmem_cache_free(ceph_msg_cache, m);
 }
 
diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index 6f164289bde8..53299c7b0ca4 100644
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -292,6 +292,10 @@ static void osd_req_op_data_release(struct ceph_osd_request *osd_req,
                 ceph_osd_data_release(&op->cls.request_data);
                 ceph_osd_data_release(&op->cls.response_data);
                 break;
+        case CEPH_OSD_OP_SETXATTR:
+        case CEPH_OSD_OP_CMPXATTR:
+                ceph_osd_data_release(&op->xattr.osd_data);
+                break;
         default:
                 break;
         }
@@ -476,8 +480,7 @@ void osd_req_op_extent_init(struct ceph_osd_request *osd_req,
         size_t payload_len = 0;
 
         BUG_ON(opcode != CEPH_OSD_OP_READ && opcode != CEPH_OSD_OP_WRITE &&
-               opcode != CEPH_OSD_OP_DELETE && opcode != CEPH_OSD_OP_ZERO &&
-               opcode != CEPH_OSD_OP_TRUNCATE);
+               opcode != CEPH_OSD_OP_ZERO && opcode != CEPH_OSD_OP_TRUNCATE);
 
         op->extent.offset = offset;
         op->extent.length = length;
@@ -545,6 +548,39 @@ void osd_req_op_cls_init(struct ceph_osd_request *osd_req, unsigned int which,
 }
 EXPORT_SYMBOL(osd_req_op_cls_init);
 
+int osd_req_op_xattr_init(struct ceph_osd_request *osd_req, unsigned int which,
+                          u16 opcode, const char *name, const void *value,
+                          size_t size, u8 cmp_op, u8 cmp_mode)
+{
+        struct ceph_osd_req_op *op = _osd_req_op_init(osd_req, which, opcode);
+        struct ceph_pagelist *pagelist;
+        size_t payload_len;
+
+        BUG_ON(opcode != CEPH_OSD_OP_SETXATTR && opcode != CEPH_OSD_OP_CMPXATTR);
+
+        pagelist = kmalloc(sizeof(*pagelist), GFP_NOFS);
+        if (!pagelist)
+                return -ENOMEM;
+
+        ceph_pagelist_init(pagelist);
+
+        payload_len = strlen(name);
+        op->xattr.name_len = payload_len;
+        ceph_pagelist_append(pagelist, name, payload_len);
+
+        op->xattr.value_len = size;
+        ceph_pagelist_append(pagelist, value, size);
+        payload_len += size;
+
+        op->xattr.cmp_op = cmp_op;
+        op->xattr.cmp_mode = cmp_mode;
+
+        ceph_osd_data_pagelist_init(&op->xattr.osd_data, pagelist);
+        op->payload_len = payload_len;
+        return 0;
+}
+EXPORT_SYMBOL(osd_req_op_xattr_init);
+
 void osd_req_op_watch_init(struct ceph_osd_request *osd_req,
                                 unsigned int which, u16 opcode,
                                 u64 cookie, u64 version, int flag)
@@ -626,7 +662,6 @@ static u64 osd_req_encode_op(struct ceph_osd_request *req,
         case CEPH_OSD_OP_READ:
         case CEPH_OSD_OP_WRITE:
         case CEPH_OSD_OP_ZERO:
-        case CEPH_OSD_OP_DELETE:
         case CEPH_OSD_OP_TRUNCATE:
                 if (src->op == CEPH_OSD_OP_WRITE)
                         request_data_len = src->extent.length;
@@ -676,6 +711,19 @@ static u64 osd_req_encode_op(struct ceph_osd_request *req,
                 dst->alloc_hint.expected_write_size =
                     cpu_to_le64(src->alloc_hint.expected_write_size);
                 break;
+        case CEPH_OSD_OP_SETXATTR:
+        case CEPH_OSD_OP_CMPXATTR:
+                dst->xattr.name_len = cpu_to_le32(src->xattr.name_len);
+                dst->xattr.value_len = cpu_to_le32(src->xattr.value_len);
+                dst->xattr.cmp_op = src->xattr.cmp_op;
+                dst->xattr.cmp_mode = src->xattr.cmp_mode;
+                osd_data = &src->xattr.osd_data;
+                ceph_osdc_msg_data_add(req->r_request, osd_data);
+                request_data_len = osd_data->pagelist->length;
+                break;
+        case CEPH_OSD_OP_CREATE:
+        case CEPH_OSD_OP_DELETE:
+                break;
         default:
                 pr_err("unsupported osd opcode %s\n",
                         ceph_osd_op_name(src->op));
@@ -705,7 +753,8 @@ static u64 osd_req_encode_op(struct ceph_osd_request *req,
 struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
                                                struct ceph_file_layout *layout,
                                                struct ceph_vino vino,
-                                               u64 off, u64 *plen, int num_ops,
+                                               u64 off, u64 *plen,
+                                               unsigned int which, int num_ops,
                                                int opcode, int flags,
                                                struct ceph_snap_context *snapc,
                                                u32 truncate_seq,
@@ -716,13 +765,11 @@ struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
         u64 objnum = 0;
         u64 objoff = 0;
         u64 objlen = 0;
-        u32 object_size;
-        u64 object_base;
         int r;
 
         BUG_ON(opcode != CEPH_OSD_OP_READ && opcode != CEPH_OSD_OP_WRITE &&
-               opcode != CEPH_OSD_OP_DELETE && opcode != CEPH_OSD_OP_ZERO &&
-               opcode != CEPH_OSD_OP_TRUNCATE);
+               opcode != CEPH_OSD_OP_ZERO && opcode != CEPH_OSD_OP_TRUNCATE &&
+               opcode != CEPH_OSD_OP_CREATE && opcode != CEPH_OSD_OP_DELETE);
 
         req = ceph_osdc_alloc_request(osdc, snapc, num_ops, use_mempool,
                                         GFP_NOFS);
@@ -738,29 +785,24 @@ struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
                 return ERR_PTR(r);
         }
 
-        object_size = le32_to_cpu(layout->fl_object_size);
-        object_base = off - objoff;
-        if (!(truncate_seq == 1 && truncate_size == -1ULL)) {
-                if (truncate_size <= object_base) {
-                        truncate_size = 0;
-                } else {
-                        truncate_size -= object_base;
-                        if (truncate_size > object_size)
-                                truncate_size = object_size;
+        if (opcode == CEPH_OSD_OP_CREATE || opcode == CEPH_OSD_OP_DELETE) {
+                osd_req_op_init(req, which, opcode);
+        } else {
+                u32 object_size = le32_to_cpu(layout->fl_object_size);
+                u32 object_base = off - objoff;
+                if (!(truncate_seq == 1 && truncate_size == -1ULL)) {
+                        if (truncate_size <= object_base) {
+                                truncate_size = 0;
+                        } else {
+                                truncate_size -= object_base;
+                                if (truncate_size > object_size)
+                                        truncate_size = object_size;
+                        }
                 }
+                osd_req_op_extent_init(req, which, opcode, objoff, objlen,
+                                       truncate_size, truncate_seq);
         }
 
-        osd_req_op_extent_init(req, 0, opcode, objoff, objlen,
-                                truncate_size, truncate_seq);
-
-        /*
-         * A second op in the ops array means the caller wants to
-         * also issue a include a 'startsync' command so that the
-         * osd will flush data quickly.
-         */
-        if (num_ops > 1)
-                osd_req_op_init(req, 1, CEPH_OSD_OP_STARTSYNC);
-
         req->r_base_oloc.pool = ceph_file_layout_pg_pool(*layout);
 
         snprintf(req->r_base_oid.name, sizeof(req->r_base_oid.name),
@@ -2626,7 +2668,7 @@ int ceph_osdc_readpages(struct ceph_osd_client *osdc,
 
         dout("readpages on ino %llx.%llx on %llu~%llu\n", vino.ino,
              vino.snap, off, *plen);
-        req = ceph_osdc_new_request(osdc, layout, vino, off, plen, 1,
+        req = ceph_osdc_new_request(osdc, layout, vino, off, plen, 0, 1,
                                     CEPH_OSD_OP_READ, CEPH_OSD_FLAG_READ,
                                     NULL, truncate_seq, truncate_size,
                                     false);
@@ -2669,7 +2711,7 @@ int ceph_osdc_writepages(struct ceph_osd_client *osdc, struct ceph_vino vino,
         int page_align = off & ~PAGE_MASK;
 
         BUG_ON(vino.snap != CEPH_NOSNAP);       /* snapshots aren't writeable */
-        req = ceph_osdc_new_request(osdc, layout, vino, off, &len, 1,
+        req = ceph_osdc_new_request(osdc, layout, vino, off, &len, 0, 1,
                                     CEPH_OSD_OP_WRITE,
                                     CEPH_OSD_FLAG_ONDISK | CEPH_OSD_FLAG_WRITE,
                                     snapc, truncate_seq, truncate_size,
@@ -2920,6 +2962,20 @@ static int invalidate_authorizer(struct ceph_connection *con)
         return ceph_monc_validate_auth(&osdc->client->monc);
 }
 
+static int sign_message(struct ceph_connection *con, struct ceph_msg *msg)
+{
+        struct ceph_osd *o = con->private;
+        struct ceph_auth_handshake *auth = &o->o_auth;
+        return ceph_auth_sign_message(auth, msg);
+}
+
+static int check_message_signature(struct ceph_connection *con, struct ceph_msg *msg)
+{
+        struct ceph_osd *o = con->private;
+        struct ceph_auth_handshake *auth = &o->o_auth;
+        return ceph_auth_check_message_signature(auth, msg);
+}
+
 static const struct ceph_connection_operations osd_con_ops = {
         .get = get_osd_con,
         .put = put_osd_con,
@@ -2928,5 +2984,7 @@ static const struct ceph_connection_operations osd_con_ops = {
         .verify_authorizer_reply = verify_authorizer_reply,
         .invalidate_authorizer = invalidate_authorizer,
         .alloc_msg = alloc_msg,
+        .sign_message = sign_message,
+        .check_message_signature = check_message_signature,
         .fault = osd_reset,
 };