Adding upstream version 18.2.2.upstream/18.2.2

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

50 files changed, 7335 insertions, 0 deletions

diff --git a/src/auth/Auth.h b/src/auth/Auth.h
new file mode 100644
index 000000000..5521c8d3f
--- /dev/null
+++ b/src/auth/Auth.h
@@ -0,0 +1,319 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_AUTHTYPES_H
+#define CEPH_AUTHTYPES_H
+
+#include "Crypto.h"
+#include "common/entity_name.h"
+
+// The _MAX values are a bit wonky here because we are overloading the first
+// byte of the auth payload to identify both the type of authentication to be
+// used *and* the encoding version for the authenticator.  So, we define a
+// range.
+enum {
+  AUTH_MODE_NONE = 0,
+  AUTH_MODE_AUTHORIZER = 1,
+  AUTH_MODE_AUTHORIZER_MAX = 9,
+  AUTH_MODE_MON = 10,
+  AUTH_MODE_MON_MAX = 19,
+};
+
+
+struct EntityAuth {
+  CryptoKey key;
+  std::map<std::string, ceph::buffer::list> caps;
+  CryptoKey pending_key; ///< new but uncommitted key
+
+  void encode(ceph::buffer::list& bl) const {
+    __u8 struct_v = 3;
+    using ceph::encode;
+    encode(struct_v, bl);
+    encode((uint64_t)CEPH_AUTH_UID_DEFAULT, bl);
+    encode(key, bl);
+    encode(caps, bl);
+    encode(pending_key, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    __u8 struct_v;
+    decode(struct_v, bl);
+    if (struct_v >= 2) {
+      uint64_t old_auid;
+      decode(old_auid, bl);
+    }
+    decode(key, bl);
+    decode(caps, bl);
+    if (struct_v >= 3) {
+      decode(pending_key, bl);
+    }
+  }
+};
+WRITE_CLASS_ENCODER(EntityAuth)
+
+inline std::ostream& operator<<(std::ostream& out, const EntityAuth& a)
+{
+  out << "auth(key=" << a.key;
+  if (!a.pending_key.empty()) {
+    out << " pending_key=" << a.pending_key;
+  }
+  out << ")";
+  return out;
+}
+
+struct AuthCapsInfo {
+  bool allow_all;
+  ceph::buffer::list caps;
+
+  AuthCapsInfo() : allow_all(false) {}
+
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    __u8 struct_v = 1;
+    encode(struct_v, bl);
+    __u8 a = (__u8)allow_all;
+    encode(a, bl);
+    encode(caps, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    __u8 struct_v;
+    decode(struct_v, bl);
+    __u8 a;
+    decode(a, bl);
+    allow_all = (bool)a;
+    decode(caps, bl);
+  }
+};
+WRITE_CLASS_ENCODER(AuthCapsInfo)
+
+/*
+ * The ticket (if properly validated) authorizes the principal use
+ * services as described by 'caps' during the specified validity
+ * period.
+ */
+struct AuthTicket {
+  EntityName name;
+  uint64_t global_id; /* global instance id */
+  utime_t created, renew_after, expires;
+  AuthCapsInfo caps;
+  __u32 flags;
+
+  AuthTicket() : global_id(0), flags(0){}
+
+  void init_timestamps(utime_t now, double ttl) {
+    created = now;
+    expires = now;
+    expires += ttl;
+    renew_after = now;
+    renew_after += ttl / 2.0;
+  }
+
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    __u8 struct_v = 2;
+    encode(struct_v, bl);
+    encode(name, bl);
+    encode(global_id, bl);
+    encode((uint64_t)CEPH_AUTH_UID_DEFAULT, bl);
+    encode(created, bl);
+    encode(expires, bl);
+    encode(caps, bl);
+    encode(flags, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    __u8 struct_v;
+    decode(struct_v, bl);
+    decode(name, bl);
+    decode(global_id, bl);
+    if (struct_v >= 2) {
+      uint64_t old_auid;
+      decode(old_auid, bl);
+    }
+    decode(created, bl);
+    decode(expires, bl);
+    decode(caps, bl);
+    decode(flags, bl);
+  }
+};
+WRITE_CLASS_ENCODER(AuthTicket)
+
+
+/*
+ * abstract authorizer class
+ */
+struct AuthAuthorizer {
+  __u32 protocol;
+  ceph::buffer::list bl;
+  CryptoKey session_key;
+
+  explicit AuthAuthorizer(__u32 p) : protocol(p) {}
+  virtual ~AuthAuthorizer() {}
+  virtual bool verify_reply(ceph::buffer::list::const_iterator& reply,
+			    std::string *connection_secret) = 0;
+  virtual bool add_challenge(CephContext *cct,
+			     const ceph::buffer::list& challenge) = 0;
+};
+
+struct AuthAuthorizerChallenge {
+  virtual ~AuthAuthorizerChallenge() {}
+};
+
+struct AuthConnectionMeta {
+  uint32_t auth_method = CEPH_AUTH_UNKNOWN;  //< CEPH_AUTH_*
+
+  /// client: initial empty, but populated if server said bad method
+  std::vector<uint32_t> allowed_methods;
+
+  int auth_mode = AUTH_MODE_NONE;  ///< AUTH_MODE_*
+
+  int con_mode = 0;  ///< negotiated mode
+
+  bool is_mode_crc() const {
+    return con_mode == CEPH_CON_MODE_CRC;
+  }
+  bool is_mode_secure() const {
+    return con_mode == CEPH_CON_MODE_SECURE;
+  }
+
+  CryptoKey session_key;           ///< per-ticket key
+
+  size_t get_connection_secret_length() const {
+    switch (con_mode) {
+    case CEPH_CON_MODE_CRC:
+      return 0;
+    case CEPH_CON_MODE_SECURE:
+      return 16 * 4;
+    }
+    return 0;
+  }
+  std::string connection_secret;   ///< per-connection key
+
+  std::unique_ptr<AuthAuthorizer> authorizer;
+  std::unique_ptr<AuthAuthorizerChallenge> authorizer_challenge;
+
+  ///< set if msgr1 peer doesn't support CEPHX_V2
+  bool skip_authorizer_challenge = false;
+};
+
+/*
+ * Key management
+ */ 
+#define KEY_ROTATE_NUM 3   /* prev, current, next */
+
+struct ExpiringCryptoKey {
+  CryptoKey key;
+  utime_t expiration;
+
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    __u8 struct_v = 1;
+    encode(struct_v, bl);
+    encode(key, bl);
+    encode(expiration, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    __u8 struct_v;
+    decode(struct_v, bl);
+    decode(key, bl);
+    decode(expiration, bl);
+  }
+};
+WRITE_CLASS_ENCODER(ExpiringCryptoKey)
+
+inline std::ostream& operator<<(std::ostream& out, const ExpiringCryptoKey& c)
+{
+  return out << c.key << " expires " << c.expiration;
+}
+
+struct RotatingSecrets {
+  std::map<uint64_t, ExpiringCryptoKey> secrets;
+  version_t max_ver;
+
+  RotatingSecrets() : max_ver(0) {}
+
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    __u8 struct_v = 1;
+    encode(struct_v, bl);
+    encode(secrets, bl);
+    encode(max_ver, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    __u8 struct_v;
+    decode(struct_v, bl);
+    decode(secrets, bl);
+    decode(max_ver, bl);
+  }
+  
+  uint64_t add(ExpiringCryptoKey& key) {
+    secrets[++max_ver] = key;
+    while (secrets.size() > KEY_ROTATE_NUM)
+      secrets.erase(secrets.begin());
+    return max_ver;
+  }
+  
+  bool need_new_secrets() const {
+    return secrets.size() < KEY_ROTATE_NUM;
+  }
+  bool need_new_secrets(const utime_t& now) const {
+    return secrets.size() < KEY_ROTATE_NUM || current().expiration <= now;
+  }
+
+  ExpiringCryptoKey& previous() {
+    return secrets.begin()->second;
+  }
+  ExpiringCryptoKey& current() {
+    auto p = secrets.begin();
+    ++p;
+    return p->second;
+  }
+  const ExpiringCryptoKey& current() const {
+    auto p = secrets.begin();
+    ++p;
+    return p->second;
+  }
+  ExpiringCryptoKey& next() {
+    return secrets.rbegin()->second;
+  }
+  bool empty() {
+    return secrets.empty();
+  }
+
+  void dump();
+};
+WRITE_CLASS_ENCODER(RotatingSecrets)
+
+
+
+class KeyStore {
+public:
+  virtual ~KeyStore() {}
+  virtual bool get_secret(const EntityName& name, CryptoKey& secret) const = 0;
+  virtual bool get_service_secret(uint32_t service_id, uint64_t secret_id,
+				  CryptoKey& secret) const = 0;
+};
+
+inline bool auth_principal_needs_rotating_keys(EntityName& name)
+{
+  uint32_t ty(name.get_type());
+  return ((ty == CEPH_ENTITY_TYPE_OSD)
+      || (ty == CEPH_ENTITY_TYPE_MDS)
+      || (ty == CEPH_ENTITY_TYPE_MGR));
+}
+
+#endif
diff --git a/src/auth/AuthAuthorizeHandler.h b/src/auth/AuthAuthorizeHandler.h
new file mode 100644
index 000000000..fc4a87886
--- /dev/null
+++ b/src/auth/AuthAuthorizeHandler.h
@@ -0,0 +1,47 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_AUTHAUTHORIZEHANDLER_H
+#define CEPH_AUTHAUTHORIZEHANDLER_H
+
+#include "Auth.h"
+#include "include/common_fwd.h"
+#include "include/types.h"
+#include "common/ceph_mutex.h"
+// Different classes of session crypto handling
+
+#define SESSION_CRYPTO_NONE 0
+#define SESSION_SYMMETRIC_AUTHENTICATE 1
+#define SESSION_SYMMETRIC_ENCRYPT 2
+
+class KeyRing;
+
+struct AuthAuthorizeHandler {
+  virtual ~AuthAuthorizeHandler() {}
+  virtual bool verify_authorizer(
+    CephContext *cct,
+    const KeyStore& keys,
+    const ceph::buffer::list& authorizer_data,
+    size_t connection_secret_required_len,
+    ceph::buffer::list *authorizer_reply,
+    EntityName *entity_name,
+    uint64_t *global_id,
+    AuthCapsInfo *caps_info,
+    CryptoKey *session_key,
+    std::string *connection_secret,
+    std::unique_ptr<AuthAuthorizerChallenge> *challenge) = 0;
+  virtual int authorizer_session_crypto() = 0;
+};
+
+#endif
diff --git a/src/auth/AuthClient.h b/src/auth/AuthClient.h
new file mode 100644
index 000000000..c4ca01992
--- /dev/null
+++ b/src/auth/AuthClient.h
@@ -0,0 +1,51 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#pragma once
+
+#include <cstdint>
+#include <vector>
+#include "include/buffer_fwd.h"
+
+class AuthConnectionMeta;
+class Connection;
+class CryptoKey;
+
+class AuthClient {
+public:
+  virtual ~AuthClient() {}
+
+  /// Build an authentication request to begin the handshake
+  virtual int get_auth_request(
+    Connection *con,
+    AuthConnectionMeta *auth_meta,
+    uint32_t *method,
+    std::vector<uint32_t> *preferred_modes,
+    ceph::buffer::list *out) = 0;
+
+  /// Handle server's request to continue the handshake
+  virtual int handle_auth_reply_more(
+    Connection *con,
+    AuthConnectionMeta *auth_meta,
+    const ceph::buffer::list& bl,
+    ceph::buffer::list *reply) = 0;
+
+  /// Handle server's indication that authentication succeeded
+  virtual int handle_auth_done(
+    Connection *con,
+    AuthConnectionMeta *auth_meta,
+    uint64_t global_id,
+    uint32_t con_mode,
+    const ceph::buffer::list& bl,
+    CryptoKey *session_key,
+    std::string *connection_secret) = 0;
+
+  /// Handle server's indication that the previous auth attempt failed
+  virtual int handle_auth_bad_method(
+    Connection *con,
+    AuthConnectionMeta *auth_meta,
+    uint32_t old_auth_method,
+    int result,
+    const std::vector<uint32_t>& allowed_methods,
+    const std::vector<uint32_t>& allowed_modes) = 0;
+};
diff --git a/src/auth/AuthClientHandler.cc b/src/auth/AuthClientHandler.cc
new file mode 100644
index 000000000..5e6521dfd
--- /dev/null
+++ b/src/auth/AuthClientHandler.cc
@@ -0,0 +1,42 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+
+#include <errno.h>
+
+#include "AuthClientHandler.h"
+#include "cephx/CephxClientHandler.h"
+#ifdef HAVE_GSSAPI
+#include "krb/KrbClientHandler.hpp"
+#endif
+#include "none/AuthNoneClientHandler.h"
+
+
+AuthClientHandler*
+AuthClientHandler::create(CephContext* cct, int proto,
+			  RotatingKeyRing* rkeys)
+{
+  switch (proto) {
+  case CEPH_AUTH_CEPHX:
+    return new CephxClientHandler(cct, rkeys);
+  case CEPH_AUTH_NONE:
+    return new AuthNoneClientHandler{cct};
+#ifdef HAVE_GSSAPI
+  case CEPH_AUTH_GSS: 
+    return new KrbClientHandler(cct);
+#endif
+  default:
+    return NULL;
+  }
+}
diff --git a/src/auth/AuthClientHandler.h b/src/auth/AuthClientHandler.h
new file mode 100644
index 000000000..aba21b415
--- /dev/null
+++ b/src/auth/AuthClientHandler.h
@@ -0,0 +1,73 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_AUTHCLIENTHANDLER_H
+#define CEPH_AUTHCLIENTHANDLER_H
+
+
+#include "auth/Auth.h"
+#include "include/common_fwd.h"
+
+class RotatingKeyRing;
+
+class AuthClientHandler {
+protected:
+  CephContext *cct;
+  EntityName name;
+  uint64_t global_id;
+  uint32_t want;
+  uint32_t have;
+  uint32_t need;
+
+public:
+  explicit AuthClientHandler(CephContext *cct_)
+    : cct(cct_), global_id(0), want(CEPH_ENTITY_TYPE_AUTH), have(0), need(0)
+  {}
+  virtual ~AuthClientHandler() {}
+
+  virtual AuthClientHandler* clone() const = 0;
+
+  void init(const EntityName& n) { name = n; }
+  
+  void set_want_keys(__u32 keys) {
+    want = keys | CEPH_ENTITY_TYPE_AUTH;
+    validate_tickets();
+  }
+
+  virtual int get_protocol() const = 0;
+
+  virtual void reset() = 0;
+  virtual void prepare_build_request() = 0;
+  virtual void build_initial_request(ceph::buffer::list *bl) const {
+    // this is empty for methods cephx and none.
+  }
+  virtual int build_request(ceph::buffer::list& bl) const = 0;
+  virtual int handle_response(int ret, ceph::buffer::list::const_iterator& iter,
+			      CryptoKey *session_key,
+			      std::string *connection_secret) = 0;
+  virtual bool build_rotating_request(ceph::buffer::list& bl) const = 0;
+
+  virtual AuthAuthorizer *build_authorizer(uint32_t service_id) const = 0;
+
+  virtual bool need_tickets() = 0;
+
+  virtual void set_global_id(uint64_t id) = 0;
+
+  static AuthClientHandler* create(CephContext* cct, int proto, RotatingKeyRing* rkeys);
+protected:
+  virtual void validate_tickets() = 0;
+};
+
+#endif
+
diff --git a/src/auth/AuthMethodList.cc b/src/auth/AuthMethodList.cc
new file mode 100644
index 000000000..6efcea4e9
--- /dev/null
+++ b/src/auth/AuthMethodList.cc
@@ -0,0 +1,71 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#include <algorithm>
+#include "common/debug.h"
+#include "include/str_list.h"
+
+#include "AuthMethodList.h"
+
+const static int dout_subsys = ceph_subsys_auth;
+
+
+AuthMethodList::AuthMethodList(CephContext *cct, std::string str)
+{
+  std::list<std::string> sup_list;
+  get_str_list(str, sup_list);
+  if (sup_list.empty()) {
+    lderr(cct) << "WARNING: empty auth protocol list" << dendl;
+  }
+  for (auto iter = sup_list.begin(); iter != sup_list.end(); ++iter) {
+    ldout(cct, 5) << "adding auth protocol: " << *iter << dendl;
+    if (iter->compare("cephx") == 0) {
+      auth_supported.push_back(CEPH_AUTH_CEPHX);
+    } else if (iter->compare("none") == 0) {
+      auth_supported.push_back(CEPH_AUTH_NONE);
+    } else if (iter->compare("gss") == 0) {
+      auth_supported.push_back(CEPH_AUTH_GSS);
+    } else {
+      auth_supported.push_back(CEPH_AUTH_UNKNOWN);
+      lderr(cct) << "WARNING: unknown auth protocol defined: " << *iter << dendl;
+    }
+  }
+  if (auth_supported.empty()) {
+    lderr(cct) << "WARNING: no auth protocol defined, use 'cephx' by default" << dendl;
+    auth_supported.push_back(CEPH_AUTH_CEPHX);
+  }
+}
+
+bool AuthMethodList::is_supported_auth(int auth_type)
+{
+  return std::find(auth_supported.begin(), auth_supported.end(), auth_type) != auth_supported.end();
+}
+
+int AuthMethodList::pick(const std::set<__u32>& supported)
+{
+  for (auto p = supported.rbegin(); p != supported.rend(); ++p)
+    if (is_supported_auth(*p))
+      return *p;
+  return CEPH_AUTH_UNKNOWN;
+}
+
+void AuthMethodList::remove_supported_auth(int auth_type)
+{
+  for (auto p = auth_supported.begin(); p != auth_supported.end(); ) {
+    if (*p == (__u32)auth_type)
+      auth_supported.erase(p++);
+    else 
+      ++p;
+  }
+}
diff --git a/src/auth/AuthMethodList.h b/src/auth/AuthMethodList.h
new file mode 100644
index 000000000..e139218d3
--- /dev/null
+++ b/src/auth/AuthMethodList.h
@@ -0,0 +1,41 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_AUTHMETHODLIST_H
+#define CEPH_AUTHMETHODLIST_H
+
+#include "include/common_fwd.h"
+#include "include/int_types.h"
+
+#include <list>
+#include <set>
+#include <string>
+
+class AuthMethodList {
+  std::list<__u32> auth_supported;
+public:
+  AuthMethodList(CephContext *cct, std::string str);
+
+  bool is_supported_auth(int auth_type);
+  int pick(const std::set<__u32>& supported);
+
+  const std::list<__u32>& get_supported_set() const {
+    return auth_supported;
+  }
+
+  void remove_supported_auth(int auth_type);
+};
+
+
+#endif
diff --git a/src/auth/AuthRegistry.cc b/src/auth/AuthRegistry.cc
new file mode 100644
index 000000000..0043567fb
--- /dev/null
+++ b/src/auth/AuthRegistry.cc
@@ -0,0 +1,373 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "AuthRegistry.h"
+
+#include "cephx/CephxAuthorizeHandler.h"
+#ifdef HAVE_GSSAPI
+#include "krb/KrbAuthorizeHandler.hpp"
+#endif
+#include "none/AuthNoneAuthorizeHandler.h"
+#include "common/ceph_context.h"
+#include "common/debug.h"
+#include "auth/KeyRing.h"
+
+#define dout_subsys ceph_subsys_auth
+#undef dout_prefix
+#define dout_prefix *_dout << "AuthRegistry(" << this << ") "
+
+using std::string;
+
+AuthRegistry::AuthRegistry(CephContext *cct)
+  : cct(cct)
+{
+  cct->_conf.add_observer(this);
+}
+
+AuthRegistry::~AuthRegistry()
+{
+  cct->_conf.remove_observer(this);
+  for (auto i : authorize_handlers) {
+    delete i.second;
+  }
+}
+
+const char** AuthRegistry::get_tracked_conf_keys() const
+{
+  static const char *keys[] = {
+    "auth_supported",
+    "auth_client_required",
+    "auth_cluster_required",
+    "auth_service_required",
+    "ms_mon_cluster_mode",
+    "ms_mon_service_mode",
+    "ms_mon_client_mode",
+    "ms_cluster_mode",
+    "ms_service_mode",
+    "ms_client_mode",
+    "keyring",
+    NULL
+  };
+  return keys;
+}
+
+void AuthRegistry::handle_conf_change(
+  const ConfigProxy& conf,
+  const std::set<std::string>& changed)
+{
+  std::scoped_lock l(lock);
+  _refresh_config();
+}
+
+
+void AuthRegistry::_parse_method_list(const string& s,
+				      std::vector<uint32_t> *v)
+{
+  std::list<std::string> sup_list;
+  get_str_list(s, sup_list);
+  if (sup_list.empty()) {
+    lderr(cct) << "WARNING: empty auth protocol list" << dendl;
+  }
+  v->clear();
+  for (auto& i : sup_list) {
+    ldout(cct, 5) << "adding auth protocol: " << i << dendl;
+    if (i == "cephx") {
+      v->push_back(CEPH_AUTH_CEPHX);
+    } else if (i == "none") {
+      v->push_back(CEPH_AUTH_NONE);
+    } else if (i == "gss") {
+      v->push_back(CEPH_AUTH_GSS);
+    } else {
+      lderr(cct) << "WARNING: unknown auth protocol defined: " << i << dendl;
+    }
+  }
+  if (v->empty()) {
+    lderr(cct) << "WARNING: no auth protocol defined" << dendl;
+  }
+  ldout(cct,20) << __func__ << " " << s << " -> " << *v << dendl;
+}
+
+void AuthRegistry::_parse_mode_list(const string& s,
+				    std::vector<uint32_t> *v)
+{
+  std::list<std::string> sup_list;
+  get_str_list(s, sup_list);
+  if (sup_list.empty()) {
+    lderr(cct) << "WARNING: empty auth protocol list" << dendl;
+  }
+  v->clear();
+  for (auto& i : sup_list) {
+    ldout(cct, 5) << "adding con mode: " << i << dendl;
+    if (i == "crc") {
+      v->push_back(CEPH_CON_MODE_CRC);
+    } else if (i == "secure") {
+      v->push_back(CEPH_CON_MODE_SECURE);
+    } else {
+      lderr(cct) << "WARNING: unknown connection mode " << i << dendl;
+    }
+  }
+  if (v->empty()) {
+    lderr(cct) << "WARNING: no connection modes defined" << dendl;
+  }
+  ldout(cct,20) << __func__ << " " << s << " -> " << *v << dendl;
+}
+
+void AuthRegistry::_refresh_config()
+{
+  if (cct->_conf->auth_supported.size()) {
+    _parse_method_list(cct->_conf->auth_supported, &cluster_methods);
+    _parse_method_list(cct->_conf->auth_supported, &service_methods);
+    _parse_method_list(cct->_conf->auth_supported, &client_methods);
+  } else {
+    _parse_method_list(cct->_conf->auth_cluster_required, &cluster_methods);
+    _parse_method_list(cct->_conf->auth_service_required, &service_methods);
+    _parse_method_list(cct->_conf->auth_client_required, &client_methods);
+  }
+  _parse_mode_list(cct->_conf.get_val<string>("ms_mon_cluster_mode"),
+		   &mon_cluster_modes);
+  _parse_mode_list(cct->_conf.get_val<string>("ms_mon_service_mode"),
+		   &mon_service_modes);
+  _parse_mode_list(cct->_conf.get_val<string>("ms_mon_client_mode"),
+		   &mon_client_modes);
+  _parse_mode_list(cct->_conf.get_val<string>("ms_cluster_mode"),
+		   &cluster_modes);
+  _parse_mode_list(cct->_conf.get_val<string>("ms_service_mode"),
+		   &service_modes);
+  _parse_mode_list(cct->_conf.get_val<string>("ms_client_mode"),
+		   &client_modes);
+
+  ldout(cct,10) << __func__ << " cluster_methods " << cluster_methods
+		<< " service_methods " << service_methods
+		<< " client_methods " << client_methods
+		<< dendl;
+  ldout(cct,10) << __func__ << " mon_cluster_modes " << mon_cluster_modes
+		<< " mon_service_modes " << mon_service_modes
+		<< " mon_client_modes " << mon_client_modes
+		<< "; cluster_modes " << cluster_modes
+		<< " service_modes " << service_modes
+		<< " client_modes " << client_modes
+		<< dendl;
+
+  // if we have no keyring, filter out cephx
+  _no_keyring_disabled_cephx = false;
+  bool any_cephx = false;
+  for (auto *p : {&cluster_methods, &service_methods, &client_methods}) {
+    auto q = std::find(p->begin(), p->end(), CEPH_AUTH_CEPHX);
+    if (q != p->end()) {
+      any_cephx = true;
+      break;
+    }
+  }
+  if (any_cephx) {
+    KeyRing k;
+    int r = k.from_ceph_context(cct);
+    if (r == -ENOENT) {
+      for (auto *p : {&cluster_methods, &service_methods, &client_methods}) {
+	auto q = std::find(p->begin(), p->end(), CEPH_AUTH_CEPHX);
+	if (q != p->end()) {
+	  p->erase(q);
+	  _no_keyring_disabled_cephx = true;
+	}
+      }
+    }
+    if (_no_keyring_disabled_cephx) {
+      lderr(cct) << "no keyring found at " << cct->_conf->keyring
+	       << ", disabling cephx" << dendl;
+    }
+  }
+}
+
+void AuthRegistry::get_supported_methods(
+  int peer_type,
+  std::vector<uint32_t> *methods,
+  std::vector<uint32_t> *modes) const
+{
+  if (methods) {
+    methods->clear();
+  }
+  if (modes) {
+    modes->clear();
+  }
+  std::scoped_lock l(lock);
+  switch (cct->get_module_type()) {
+  case CEPH_ENTITY_TYPE_CLIENT:
+    // i am client
+    if (methods) {
+      *methods = client_methods;
+    }
+    if (modes) {
+      switch (peer_type) {
+      case CEPH_ENTITY_TYPE_MON:
+      case CEPH_ENTITY_TYPE_MGR:
+	*modes = mon_client_modes;
+	break;
+      default:
+	*modes = client_modes;
+      }
+    }
+    return;
+  case CEPH_ENTITY_TYPE_MON:
+  case CEPH_ENTITY_TYPE_MGR:
+    // i am mon/mgr
+    switch (peer_type) {
+    case CEPH_ENTITY_TYPE_MON:
+    case CEPH_ENTITY_TYPE_MGR:
+      // they are mon/mgr
+      if (methods) {
+	*methods = cluster_methods;
+      }
+      if (modes) {
+	*modes = mon_cluster_modes;
+      }
+      break;
+    default:
+      // they are anything but mons
+      if (methods) {
+	*methods = service_methods;
+      }
+      if (modes) {
+	*modes = mon_service_modes;
+      }
+    }
+    return;
+  default:
+    // i am a non-mon daemon
+    switch (peer_type) {
+    case CEPH_ENTITY_TYPE_MON:
+    case CEPH_ENTITY_TYPE_MGR:
+      // they are a mon daemon
+      if (methods) {
+	*methods = cluster_methods;
+      }
+      if (modes) {
+	*modes = mon_cluster_modes;
+      }
+      break;
+    case CEPH_ENTITY_TYPE_MDS:
+    case CEPH_ENTITY_TYPE_OSD:
+      // they are another daemon
+      if (methods) {
+	*methods = cluster_methods;
+      }
+      if (modes) {
+	*modes = cluster_modes;
+      }
+      break;
+    default:
+      // they are a client
+      if (methods) {
+	*methods = service_methods;
+      }
+      if (modes) {
+	*modes = service_modes;
+      }
+      break;
+    }
+  }
+}
+
+bool AuthRegistry::is_supported_method(int peer_type, int method) const
+{
+  std::vector<uint32_t> s;
+  get_supported_methods(peer_type, &s);
+  return std::find(s.begin(), s.end(), method) != s.end();
+}
+
+bool AuthRegistry::any_supported_methods(int peer_type) const
+{
+  std::vector<uint32_t> s;
+  get_supported_methods(peer_type, &s);
+  return !s.empty();
+}
+
+void AuthRegistry::get_supported_modes(
+  int peer_type,
+  uint32_t auth_method,
+  std::vector<uint32_t> *modes) const
+{
+  std::vector<uint32_t> s;
+  get_supported_methods(peer_type, nullptr, &s);
+  if (auth_method == CEPH_AUTH_NONE) {
+    // filter out all but crc for AUTH_NONE
+    modes->clear();
+    for (auto mode : s) {
+      if (mode == CEPH_CON_MODE_CRC) {
+	modes->push_back(mode);
+      }
+    }
+  } else {
+    *modes = s;
+  }
+}
+
+uint32_t AuthRegistry::pick_mode(
+  int peer_type,
+  uint32_t auth_method,
+  const std::vector<uint32_t>& preferred_modes)
+{
+  std::vector<uint32_t> allowed_modes;
+  get_supported_modes(peer_type, auth_method, &allowed_modes);
+  for (auto mode : preferred_modes) {
+    if (std::find(allowed_modes.begin(), allowed_modes.end(), mode)
+	!= allowed_modes.end()) {
+      return mode;
+    }
+  }
+  ldout(cct,1) << "failed to pick con mode from client's " << preferred_modes
+	       << " and our " << allowed_modes << dendl;
+  return CEPH_CON_MODE_UNKNOWN;
+}
+
+AuthAuthorizeHandler *AuthRegistry::get_handler(int peer_type, int method)
+{
+  std::scoped_lock l{lock};
+  ldout(cct,20) << __func__ << " peer_type " << peer_type << " method " << method
+		<< " cluster_methods " << cluster_methods
+		<< " service_methods " << service_methods
+		<< " client_methods " << client_methods
+		<< dendl;
+  if (cct->get_module_type() == CEPH_ENTITY_TYPE_CLIENT) {
+    return nullptr;
+  }
+  switch (peer_type) {
+  case CEPH_ENTITY_TYPE_MON:
+  case CEPH_ENTITY_TYPE_MGR:
+  case CEPH_ENTITY_TYPE_MDS:
+  case CEPH_ENTITY_TYPE_OSD:
+    if (std::find(cluster_methods.begin(), cluster_methods.end(), method) ==
+	cluster_methods.end()) {
+      return nullptr;
+    }
+    break;
+  default:
+    if (std::find(service_methods.begin(), service_methods.end(), method) ==
+	service_methods.end()) {
+      return nullptr;
+    }
+    break;
+  }
+
+  auto iter = authorize_handlers.find(method);
+  if (iter != authorize_handlers.end()) {
+    return iter->second;
+  }
+  AuthAuthorizeHandler *ah = nullptr;
+  switch (method) {
+  case CEPH_AUTH_NONE:
+    ah = new AuthNoneAuthorizeHandler();
+    break;
+  case CEPH_AUTH_CEPHX:
+    ah = new CephxAuthorizeHandler();
+    break;
+#ifdef HAVE_GSSAPI
+  case CEPH_AUTH_GSS:
+    ah = new KrbAuthorizeHandler();
+    break;
+#endif
+  }
+  if (ah) {
+    authorize_handlers[method] = ah;
+  }
+  return ah;
+}
+
diff --git a/src/auth/AuthRegistry.h b/src/auth/AuthRegistry.h
new file mode 100644
index 000000000..fd746c791
--- /dev/null
+++ b/src/auth/AuthRegistry.h
@@ -0,0 +1,81 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#pragma once
+
+#include <map>
+#include <vector>
+
+#include "AuthAuthorizeHandler.h"
+#include "AuthMethodList.h"
+#include "common/ceph_mutex.h"
+#include "common/ceph_context.h"
+#include "common/config_cacher.h"
+
+class AuthRegistry : public md_config_obs_t {
+  CephContext *cct;
+  mutable ceph::mutex lock = ceph::make_mutex("AuthRegistry::lock");
+
+  std::map<int,AuthAuthorizeHandler*> authorize_handlers;
+
+  bool _no_keyring_disabled_cephx = false;
+
+  // CEPH_AUTH_*
+  std::vector<uint32_t> cluster_methods;
+  std::vector<uint32_t> service_methods;
+  std::vector<uint32_t> client_methods;
+
+  // CEPH_CON_MODE_*
+  std::vector<uint32_t> mon_cluster_modes;
+  std::vector<uint32_t> mon_service_modes;
+  std::vector<uint32_t> mon_client_modes;
+  std::vector<uint32_t> cluster_modes;
+  std::vector<uint32_t> service_modes;
+  std::vector<uint32_t> client_modes;
+
+  void _parse_method_list(const std::string& str, std::vector<uint32_t> *v);
+  void _parse_mode_list(const std::string& str, std::vector<uint32_t> *v);
+  void _refresh_config();
+
+public:
+  AuthRegistry(CephContext *cct);
+  ~AuthRegistry();
+
+  void refresh_config() {
+    std::scoped_lock l(lock);
+    _refresh_config();
+  }
+
+  void get_supported_methods(int peer_type,
+			     std::vector<uint32_t> *methods,
+			     std::vector<uint32_t> *modes=nullptr) const;
+  bool is_supported_method(int peer_type, int method) const;
+  bool any_supported_methods(int peer_type) const;
+
+  void get_supported_modes(int peer_type,
+			   uint32_t auth_method,
+			   std::vector<uint32_t> *modes) const;
+
+  uint32_t pick_mode(int peer_type,
+		     uint32_t auth_method,
+		     const std::vector<uint32_t>& preferred_modes);
+
+  static bool is_secure_method(uint32_t method) {
+    return (method == CEPH_AUTH_CEPHX);
+  }
+
+  static bool is_secure_mode(uint32_t mode) {
+    return (mode == CEPH_CON_MODE_SECURE);
+  }
+
+  AuthAuthorizeHandler *get_handler(int peer_type, int method);
+
+  const char** get_tracked_conf_keys() const override;
+  void handle_conf_change(const ConfigProxy& conf,
+                          const std::set<std::string>& changed) override;
+
+  bool no_keyring_disabled_cephx() {
+    std::scoped_lock l(lock);
+    return _no_keyring_disabled_cephx;
+  }
+};
diff --git a/src/auth/AuthServer.h b/src/auth/AuthServer.h
new file mode 100644
index 000000000..a71f6a70a
--- /dev/null
+++ b/src/auth/AuthServer.h
@@ -0,0 +1,51 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#pragma once
+
+#include "AuthRegistry.h"
+#include "include/common_fwd.h"
+
+#include <vector>
+
+class Connection;
+
+class AuthServer {
+public:
+  AuthRegistry auth_registry;
+
+  AuthServer(CephContext *cct) : auth_registry(cct) {}
+  virtual ~AuthServer() {}
+
+  /// Get authentication methods and connection modes for the given peer type
+  virtual void get_supported_auth_methods(
+    int peer_type,
+    std::vector<uint32_t> *methods,
+    std::vector<uint32_t> *modes = nullptr) {
+    auth_registry.get_supported_methods(peer_type, methods, modes);
+  }
+
+  /// Get support connection modes for the given peer type and auth method
+  virtual uint32_t pick_con_mode(
+    int peer_type,
+    uint32_t auth_method,
+    const std::vector<uint32_t>& preferred_modes) {
+    return auth_registry.pick_mode(peer_type, auth_method, preferred_modes);
+  }
+
+  /// return an AuthAuthorizeHandler for the given peer type and auth method
+  AuthAuthorizeHandler *get_auth_authorize_handler(
+    int peer_type,
+    int auth_method) {
+    return auth_registry.get_handler(peer_type, auth_method);
+  }
+
+  /// Handle an authentication request on an incoming connection
+  virtual int handle_auth_request(
+    Connection *con,
+    AuthConnectionMeta *auth_meta,
+    bool more,           ///< true if this is not the first part of the handshake
+    uint32_t auth_method,
+    const ceph::buffer::list& bl,
+    ceph::buffer::list *reply) = 0;
+};
diff --git a/src/auth/AuthServiceHandler.cc b/src/auth/AuthServiceHandler.cc
new file mode 100644
index 000000000..2d1297ee2
--- /dev/null
+++ b/src/auth/AuthServiceHandler.cc
@@ -0,0 +1,81 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#include "AuthServiceHandler.h"
+#include "cephx/CephxServiceHandler.h"
+#ifdef HAVE_GSSAPI
+#include "krb/KrbServiceHandler.hpp"
+#endif
+#include "none/AuthNoneServiceHandler.h"
+#include "common/dout.h"
+
+#define dout_subsys ceph_subsys_auth
+
+
+std::ostream& operator<<(std::ostream& os,
+			 global_id_status_t global_id_status)
+{
+  switch (global_id_status) {
+  case global_id_status_t::NONE:
+    return os << "none";
+  case global_id_status_t::NEW_PENDING:
+    return os << "new_pending";
+  case global_id_status_t::NEW_OK:
+    return os << "new_ok";
+  case global_id_status_t::NEW_NOT_EXPOSED:
+    return os << "new_not_exposed";
+  case global_id_status_t::RECLAIM_PENDING:
+    return os << "reclaim_pending";
+  case global_id_status_t::RECLAIM_OK:
+    return os << "reclaim_ok";
+  case global_id_status_t::RECLAIM_INSECURE:
+    return os << "reclaim_insecure";
+  default:
+    ceph_abort();
+  }
+}
+
+int AuthServiceHandler::start_session(const EntityName& entity_name,
+				      uint64_t global_id,
+				      bool is_new_global_id,
+				      ceph::buffer::list *result,
+				      AuthCapsInfo *caps)
+{
+  ceph_assert(!this->entity_name.get_type() && !this->global_id &&
+	      global_id_status == global_id_status_t::NONE);
+
+  ldout(cct, 10) << __func__ << " entity_name=" << entity_name
+		 << " global_id=" << global_id << " is_new_global_id="
+		 << is_new_global_id << dendl;
+  this->entity_name = entity_name;
+  this->global_id = global_id;
+
+  return do_start_session(is_new_global_id, result, caps);
+}
+
+AuthServiceHandler *get_auth_service_handler(int type, CephContext *cct, KeyServer *ks)
+{
+  switch (type) {
+  case CEPH_AUTH_CEPHX:
+    return new CephxServiceHandler(cct, ks);
+  case CEPH_AUTH_NONE:
+    return new AuthNoneServiceHandler(cct);
+#ifdef HAVE_GSSAPI
+  case CEPH_AUTH_GSS: 
+    return new KrbServiceHandler(cct, ks);
+#endif
+  default:
+    return nullptr;
+  }
+}
diff --git a/src/auth/AuthServiceHandler.h b/src/auth/AuthServiceHandler.h
new file mode 100644
index 000000000..4b3dcccbe
--- /dev/null
+++ b/src/auth/AuthServiceHandler.h
@@ -0,0 +1,83 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_AUTHSERVICEHANDLER_H
+#define CEPH_AUTHSERVICEHANDLER_H
+
+#include <stddef.h>              // for NULL
+#include <stdint.h>              // for uint64_t
+#include "common/entity_name.h"  // for EntityName
+#include "include/common_fwd.h"
+#include "include/buffer_fwd.h"  // for ceph::buffer::list
+
+class KeyServer;
+class CryptoKey;
+struct AuthCapsInfo;
+
+enum class global_id_status_t {
+  NONE,
+  // fresh client (global_id == 0); waiting for CephXAuthenticate
+  NEW_PENDING,
+  // connected client; new enough to correctly reclaim global_id
+  NEW_OK,
+  // connected client; unknown whether it can reclaim global_id correctly
+  NEW_NOT_EXPOSED,
+  // reconnecting client (global_id != 0); waiting for CephXAuthenticate
+  RECLAIM_PENDING,
+  // reconnected client; correctly reclaimed global_id
+  RECLAIM_OK,
+  // reconnected client; did not properly prove prior global_id ownership
+  RECLAIM_INSECURE
+};
+
+std::ostream& operator<<(std::ostream& os,
+			 global_id_status_t global_id_status);
+
+struct AuthServiceHandler {
+protected:
+  CephContext *cct;
+  EntityName entity_name;
+  uint64_t global_id = 0;
+  global_id_status_t global_id_status = global_id_status_t::NONE;
+
+public:
+  explicit AuthServiceHandler(CephContext *cct_) : cct(cct_) {}
+
+  virtual ~AuthServiceHandler() { }
+
+  int start_session(const EntityName& entity_name,
+		    uint64_t global_id,
+		    bool is_new_global_id,
+		    ceph::buffer::list *result,
+		    AuthCapsInfo *caps);
+  virtual int handle_request(ceph::buffer::list::const_iterator& indata,
+			     size_t connection_secret_required_length,
+			     ceph::buffer::list *result,
+			     AuthCapsInfo *caps,
+			     CryptoKey *session_key,
+			     std::string *connection_secret) = 0;
+
+  const EntityName& get_entity_name() { return entity_name; }
+  uint64_t get_global_id() { return global_id; }
+  global_id_status_t get_global_id_status() { return global_id_status; }
+
+private:
+  virtual int do_start_session(bool is_new_global_id,
+			       ceph::buffer::list *result,
+			       AuthCapsInfo *caps) = 0;
+};
+
+extern AuthServiceHandler *get_auth_service_handler(int type, CephContext *cct, KeyServer *ks);
+
+#endif
diff --git a/src/auth/AuthSessionHandler.cc b/src/auth/AuthSessionHandler.cc
new file mode 100644
index 000000000..d7ad21831
--- /dev/null
+++ b/src/auth/AuthSessionHandler.cc
@@ -0,0 +1,54 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#include "common/debug.h"
+#include "AuthSessionHandler.h"
+#include "cephx/CephxSessionHandler.h"
+#ifdef HAVE_GSSAPI
+#include "krb/KrbSessionHandler.hpp"
+#endif
+#include "none/AuthNoneSessionHandler.h"
+
+#include "common/ceph_crypto.h"
+#define dout_subsys ceph_subsys_auth
+
+
+AuthSessionHandler *get_auth_session_handler(
+  CephContext *cct, int protocol,
+  const CryptoKey& key,
+  uint64_t features)
+{
+
+  // Should add code to only print the SHA1 hash of the key, unless in secure debugging mode
+#ifndef WITH_SEASTAR
+  ldout(cct,10) << "In get_auth_session_handler for protocol " << protocol << dendl;
+#endif
+  switch (protocol) {
+  case CEPH_AUTH_CEPHX:
+    // if there is no session key, there is no session handler.
+    if (key.get_type() == CEPH_CRYPTO_NONE) {
+      return nullptr;
+    }
+    return new CephxSessionHandler(cct, key, features);
+  case CEPH_AUTH_NONE:
+    return new AuthNoneSessionHandler();
+#ifdef HAVE_GSSAPI
+  case CEPH_AUTH_GSS: 
+    return new KrbSessionHandler();
+#endif
+  default:
+    return nullptr;
+  }
+}
+
diff --git a/src/auth/AuthSessionHandler.h b/src/auth/AuthSessionHandler.h
new file mode 100644
index 000000000..692ebc288
--- /dev/null
+++ b/src/auth/AuthSessionHandler.h
@@ -0,0 +1,51 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+
+#ifndef CEPH_AUTHSESSIONHANDLER_H
+#define CEPH_AUTHSESSIONHANDLER_H
+
+#include "include/common_fwd.h"
+#include "include/types.h"
+#include "Auth.h"
+
+#define SESSION_SIGNATURE_FAILURE -1
+
+// Defines the security applied to ongoing messages in a session, once the session is established. PLR
+
+class Message;
+
+struct AuthSessionHandler {
+  virtual ~AuthSessionHandler() = default;
+  virtual int sign_message(Message *message) = 0;
+  virtual int check_message_signature(Message *message) = 0;
+};
+
+struct DummyAuthSessionHandler : AuthSessionHandler {
+  int sign_message(Message*) final {
+    return 0;
+  }
+  int check_message_signature(Message*) final {
+    return 0;
+  }
+};
+
+struct DecryptionError : public std::exception {};
+
+extern AuthSessionHandler *get_auth_session_handler(
+  CephContext *cct, int protocol,
+  const CryptoKey& key,
+  uint64_t features);
+
+#endif
diff --git a/src/auth/CMakeLists.txt b/src/auth/CMakeLists.txt
new file mode 100644
index 000000000..1ab294332
--- /dev/null
+++ b/src/auth/CMakeLists.txt
@@ -0,0 +1,25 @@
+set(auth_srcs
+  AuthClientHandler.cc
+  AuthMethodList.cc
+  AuthRegistry.cc
+  AuthSessionHandler.cc
+  Crypto.cc
+  KeyRing.cc
+  RotatingKeyRing.cc
+  cephx/CephxAuthorizeHandler.cc
+  cephx/CephxClientHandler.cc
+  cephx/CephxProtocol.cc
+  cephx/CephxSessionHandler.cc
+  none/AuthNoneAuthorizeHandler.cc)
+
+if(HAVE_GSSAPI)
+  list(APPEND auth_srcs
+    krb/KrbAuthorizeHandler.cpp
+    krb/KrbClientHandler.cpp
+    krb/KrbProtocol.cpp
+    krb/KrbSessionHandler.hpp)
+endif()
+
+add_library(common-auth-objs OBJECT ${auth_srcs})
+target_include_directories(common-auth-objs PRIVATE ${OPENSSL_INCLUDE_DIR})
+add_dependencies(common-auth-objs legacy-option-headers)
diff --git a/src/auth/Crypto.cc b/src/auth/Crypto.cc
new file mode 100644
index 000000000..ce666e8bd
--- /dev/null
+++ b/src/auth/Crypto.cc
@@ -0,0 +1,615 @@
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#include <array>
+#include <sstream>
+#include <limits>
+#include <fcntl.h>
+
+#include <openssl/aes.h>
+
+#include "Crypto.h"
+
+#include "include/ceph_assert.h"
+#include "common/Clock.h"
+#include "common/armor.h"
+#include "common/ceph_context.h"
+#include "common/ceph_crypto.h"
+#include "common/hex.h"
+#include "common/safe_io.h"
+#include "include/ceph_fs.h"
+#include "include/compat.h"
+#include "common/Formatter.h"
+#include "common/debug.h"
+#include <errno.h>
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
+
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wdeprecated-declarations"
+
+using std::ostringstream;
+using std::string;
+
+using ceph::bufferlist;
+using ceph::bufferptr;
+using ceph::Formatter;
+
+
+// use getentropy() if available. it uses the same source of randomness
+// as /dev/urandom without the filesystem overhead
+#ifdef HAVE_GETENTROPY
+
+#include <unistd.h>
+
+static bool getentropy_works()
+{
+  char buf;
+  auto ret = TEMP_FAILURE_RETRY(::getentropy(&buf, sizeof(buf)));
+  if (ret == 0) {
+    return true;
+  } else if (errno == ENOSYS || errno == EPERM) {
+    return false;
+  } else {
+    throw std::system_error(errno, std::system_category());
+  }
+}
+
+CryptoRandom::CryptoRandom() : fd(getentropy_works() ? -1 : open_urandom())
+{}
+
+CryptoRandom::~CryptoRandom()
+{
+  if (fd >= 0) {
+    VOID_TEMP_FAILURE_RETRY(::close(fd));
+  }
+}
+
+void CryptoRandom::get_bytes(char *buf, int len)
+{
+  ssize_t ret = 0;
+  if (unlikely(fd >= 0)) {
+    ret = safe_read_exact(fd, buf, len);
+  } else {
+    // getentropy() reads up to 256 bytes
+    assert(len <= 256);
+    ret = TEMP_FAILURE_RETRY(::getentropy(buf, len));
+  }
+  if (ret < 0) {
+    throw std::system_error(errno, std::system_category());
+  }
+}
+
+#elif defined(_WIN32) // !HAVE_GETENTROPY
+
+#include <bcrypt.h>
+
+CryptoRandom::CryptoRandom() : fd(0) {}
+CryptoRandom::~CryptoRandom() = default;
+
+void CryptoRandom::get_bytes(char *buf, int len)
+{
+  auto ret = BCryptGenRandom (
+    NULL,
+    (unsigned char*)buf,
+    len,
+    BCRYPT_USE_SYSTEM_PREFERRED_RNG);
+  if (ret != 0) {
+    throw std::system_error(ret, std::system_category());
+  }
+}
+
+#else // !HAVE_GETENTROPY && !_WIN32
+// open /dev/urandom once on construction and reuse the fd for all reads
+CryptoRandom::CryptoRandom()
+  : fd{open_urandom()}
+{
+  if (fd < 0) {
+    throw std::system_error(errno, std::system_category());
+  }
+}
+
+CryptoRandom::~CryptoRandom()
+{
+  VOID_TEMP_FAILURE_RETRY(::close(fd));
+}
+
+void CryptoRandom::get_bytes(char *buf, int len)
+{
+  auto ret = safe_read_exact(fd, buf, len);
+  if (ret < 0) {
+    throw std::system_error(-ret, std::system_category());
+  }
+}
+
+#endif
+
+int CryptoRandom::open_urandom()
+{
+  int fd = TEMP_FAILURE_RETRY(::open("/dev/urandom", O_CLOEXEC|O_RDONLY));
+  if (fd < 0) {
+    throw std::system_error(errno, std::system_category());
+  }
+  return fd;
+}
+
+// ---------------------------------------------------
+// fallback implementation of the bufferlist-free
+// interface.
+
+std::size_t CryptoKeyHandler::encrypt(
+  const CryptoKeyHandler::in_slice_t& in,
+  const CryptoKeyHandler::out_slice_t& out) const
+{
+  ceph::bufferptr inptr(reinterpret_cast<const char*>(in.buf), in.length);
+  ceph::bufferlist plaintext;
+  plaintext.append(std::move(inptr));
+
+  ceph::bufferlist ciphertext;
+  std::string error;
+  const int ret = encrypt(plaintext, ciphertext, &error);
+  if (ret != 0 || !error.empty()) {
+    throw std::runtime_error(std::move(error));
+  }
+
+  // we need to specify the template parameter explicitly as ::length()
+  // returns unsigned int, not size_t.
+  const auto todo_len = \
+    std::min<std::size_t>(ciphertext.length(), out.max_length);
+  memcpy(out.buf, ciphertext.c_str(), todo_len);
+
+  return todo_len;
+}
+
+std::size_t CryptoKeyHandler::decrypt(
+  const CryptoKeyHandler::in_slice_t& in,
+  const CryptoKeyHandler::out_slice_t& out) const
+{
+  ceph::bufferptr inptr(reinterpret_cast<const char*>(in.buf), in.length);
+  ceph::bufferlist ciphertext;
+  ciphertext.append(std::move(inptr));
+
+  ceph::bufferlist plaintext;
+  std::string error;
+  const int ret = decrypt(ciphertext, plaintext, &error);
+  if (ret != 0 || !error.empty()) {
+    throw std::runtime_error(std::move(error));
+  }
+
+  // we need to specify the template parameter explicitly as ::length()
+  // returns unsigned int, not size_t.
+  const auto todo_len = \
+    std::min<std::size_t>(plaintext.length(), out.max_length);
+  memcpy(out.buf, plaintext.c_str(), todo_len);
+
+  return todo_len;
+}
+
+sha256_digest_t CryptoKeyHandler::hmac_sha256(
+  const ceph::bufferlist& in) const
+{
+  TOPNSPC::crypto::HMACSHA256 hmac((const unsigned char*)secret.c_str(), secret.length());
+
+  for (const auto& bptr : in.buffers()) {
+    hmac.Update((const unsigned char *)bptr.c_str(), bptr.length());
+  }
+  sha256_digest_t ret;
+  hmac.Final(ret.v);
+
+  return ret;
+}
+
+// ---------------------------------------------------
+
+class CryptoNoneKeyHandler : public CryptoKeyHandler {
+public:
+  CryptoNoneKeyHandler()
+    : CryptoKeyHandler(CryptoKeyHandler::BLOCK_SIZE_0B()) {
+  }
+
+  using CryptoKeyHandler::encrypt;
+  using CryptoKeyHandler::decrypt;
+
+  int encrypt(const bufferlist& in,
+	       bufferlist& out, std::string *error) const override {
+    out = in;
+    return 0;
+  }
+  int decrypt(const bufferlist& in,
+	      bufferlist& out, std::string *error) const override {
+    out = in;
+    return 0;
+  }
+};
+
+class CryptoNone : public CryptoHandler {
+public:
+  CryptoNone() { }
+  ~CryptoNone() override {}
+  int get_type() const override {
+    return CEPH_CRYPTO_NONE;
+  }
+  int create(CryptoRandom *random, bufferptr& secret) override {
+    return 0;
+  }
+  int validate_secret(const bufferptr& secret) override {
+    return 0;
+  }
+  CryptoKeyHandler *get_key_handler(const bufferptr& secret, string& error) override {
+    return new CryptoNoneKeyHandler;
+  }
+};
+
+
+// ---------------------------------------------------
+
+
+class CryptoAES : public CryptoHandler {
+public:
+  CryptoAES() { }
+  ~CryptoAES() override {}
+  int get_type() const override {
+    return CEPH_CRYPTO_AES;
+  }
+  int create(CryptoRandom *random, bufferptr& secret) override;
+  int validate_secret(const bufferptr& secret) override;
+  CryptoKeyHandler *get_key_handler(const bufferptr& secret, string& error) override;
+};
+
+// when we say AES, we mean AES-128
+static constexpr const std::size_t AES_KEY_LEN{16};
+static constexpr const std::size_t AES_BLOCK_LEN{16};
+
+class CryptoAESKeyHandler : public CryptoKeyHandler {
+  AES_KEY enc_key;
+  AES_KEY dec_key;
+
+public:
+  CryptoAESKeyHandler()
+    : CryptoKeyHandler(CryptoKeyHandler::BLOCK_SIZE_16B()) {
+  }
+
+  int init(const bufferptr& s, ostringstream& err) {
+    secret = s;
+
+    const int enc_key_ret = \
+      AES_set_encrypt_key((const unsigned char*)secret.c_str(),
+			  AES_KEY_LEN * CHAR_BIT, &enc_key);
+    if (enc_key_ret != 0) {
+      err << "cannot set OpenSSL encrypt key for AES: " << enc_key_ret;
+      return -1;
+    }
+
+    const int dec_key_ret = \
+      AES_set_decrypt_key((const unsigned char*)secret.c_str(),
+			  AES_KEY_LEN * CHAR_BIT, &dec_key);
+    if (dec_key_ret != 0) {
+      err << "cannot set OpenSSL decrypt key for AES: " << dec_key_ret;
+      return -1;
+    }
+
+    return 0;
+  }
+
+  int encrypt(const ceph::bufferlist& in,
+	      ceph::bufferlist& out,
+              std::string* /* unused */) const override {
+    // we need to take into account the PKCS#7 padding. There *always* will
+    // be at least one byte of padding. This stays even to input aligned to
+    // AES_BLOCK_LEN. Otherwise we would face ambiguities during decryption.
+    // To exemplify:
+    //   16 + p2align(10, 16) -> 16
+    //   16 + p2align(16, 16) -> 32 including 16 bytes for padding.
+    ceph::bufferptr out_tmp{static_cast<unsigned>(
+      AES_BLOCK_LEN + p2align<std::size_t>(in.length(), AES_BLOCK_LEN))};
+
+    // let's pad the data
+    std::uint8_t pad_len = out_tmp.length() - in.length();
+    ceph::bufferptr pad_buf{pad_len};
+    // FIPS zeroization audit 20191115: this memset is not intended to
+    // wipe out a secret after use.
+    memset(pad_buf.c_str(), pad_len, pad_len);
+
+    // form contiguous buffer for block cipher. The ctor copies shallowly.
+    ceph::bufferlist incopy(in);
+    incopy.append(std::move(pad_buf));
+    const auto in_buf = reinterpret_cast<unsigned char*>(incopy.c_str());
+
+    // reinitialize IV each time. It might be unnecessary depending on
+    // actual implementation but at the interface layer we are obliged
+    // to deliver IV as non-const.
+    static_assert(strlen_ct(CEPH_AES_IV) == AES_BLOCK_LEN);
+    unsigned char iv[AES_BLOCK_LEN];
+    memcpy(iv, CEPH_AES_IV, AES_BLOCK_LEN);
+
+    // we aren't using EVP because of performance concerns. Profiling
+    // shows the cost is quite high. Endianness might be an issue.
+    // However, as they would affect Cephx, any fallout should pop up
+    // rather early, hopefully.
+    AES_cbc_encrypt(in_buf, reinterpret_cast<unsigned char*>(out_tmp.c_str()),
+		    out_tmp.length(), &enc_key, iv, AES_ENCRYPT);
+
+    out.append(out_tmp);
+    return 0;
+  }
+
+  int decrypt(const ceph::bufferlist& in,
+	      ceph::bufferlist& out,
+	      std::string* /* unused */) const override {
+    // PKCS#7 padding enlarges even empty plain-text to take 16 bytes.
+    if (in.length() < AES_BLOCK_LEN || in.length() % AES_BLOCK_LEN) {
+      return -1;
+    }
+
+    // needed because of .c_str() on const. It's a shallow copy.
+    ceph::bufferlist incopy(in);
+    const auto in_buf = reinterpret_cast<unsigned char*>(incopy.c_str());
+
+    // make a local, modifiable copy of IV.
+    static_assert(strlen_ct(CEPH_AES_IV) == AES_BLOCK_LEN);
+    unsigned char iv[AES_BLOCK_LEN];
+    memcpy(iv, CEPH_AES_IV, AES_BLOCK_LEN);
+
+    ceph::bufferptr out_tmp{in.length()};
+    AES_cbc_encrypt(in_buf, reinterpret_cast<unsigned char*>(out_tmp.c_str()),
+		    in.length(), &dec_key, iv, AES_DECRYPT);
+
+    // BE CAREFUL: we cannot expose any single bit of information about
+    // the cause of failure. Otherwise we'll face padding oracle attack.
+    // See: https://en.wikipedia.org/wiki/Padding_oracle_attack.
+    const auto pad_len = \
+      std::min<std::uint8_t>(out_tmp[in.length() - 1], AES_BLOCK_LEN);
+    out_tmp.set_length(in.length() - pad_len);
+    out.append(std::move(out_tmp));
+
+    return 0;
+  }
+
+  std::size_t encrypt(const in_slice_t& in,
+		      const out_slice_t& out) const override {
+    if (out.buf == nullptr) {
+      // 16 + p2align(10, 16) -> 16
+      // 16 + p2align(16, 16) -> 32
+      return AES_BLOCK_LEN + p2align<std::size_t>(in.length, AES_BLOCK_LEN);
+    }
+
+    // how many bytes of in.buf hang outside the alignment boundary and how
+    // much padding we need.
+    //   length = 23 -> tail_len = 7, pad_len = 9
+    //   length = 32 -> tail_len = 0, pad_len = 16
+    const std::uint8_t tail_len = in.length % AES_BLOCK_LEN;
+    const std::uint8_t pad_len = AES_BLOCK_LEN - tail_len;
+    static_assert(std::numeric_limits<std::uint8_t>::max() > AES_BLOCK_LEN);
+
+    std::array<unsigned char, AES_BLOCK_LEN> last_block;
+    memcpy(last_block.data(), in.buf + in.length - tail_len, tail_len);
+    // FIPS zeroization audit 20191115: this memset is not intended to
+    // wipe out a secret after use.
+    memset(last_block.data() + tail_len, pad_len, pad_len);
+
+    // need a local copy because AES_cbc_encrypt takes `iv` as non-const.
+    // Useful because it allows us to encrypt in two steps: main + tail.
+    static_assert(strlen_ct(CEPH_AES_IV) == AES_BLOCK_LEN);
+    std::array<unsigned char, AES_BLOCK_LEN> iv;
+    memcpy(iv.data(), CEPH_AES_IV, AES_BLOCK_LEN);
+
+    const std::size_t main_encrypt_size = \
+      std::min(in.length - tail_len, out.max_length);
+    AES_cbc_encrypt(in.buf, out.buf, main_encrypt_size, &enc_key, iv.data(),
+		    AES_ENCRYPT);
+
+    const std::size_t tail_encrypt_size = \
+      std::min(AES_BLOCK_LEN, out.max_length - main_encrypt_size);
+    AES_cbc_encrypt(last_block.data(), out.buf + main_encrypt_size,
+		    tail_encrypt_size, &enc_key, iv.data(), AES_ENCRYPT);
+
+    return main_encrypt_size + tail_encrypt_size;
+  }
+
+  std::size_t decrypt(const in_slice_t& in,
+		      const out_slice_t& out) const override {
+    if (in.length % AES_BLOCK_LEN != 0 || in.length < AES_BLOCK_LEN) {
+      throw std::runtime_error("input not aligned to AES_BLOCK_LEN");
+    } else if (out.buf == nullptr) {
+      // essentially it would be possible to decrypt into a buffer that
+      // doesn't include space for any PKCS#7 padding. We don't do that
+      // for the sake of performance and simplicity.
+      return in.length;
+    } else if (out.max_length < in.length) {
+      throw std::runtime_error("output buffer too small");
+    }
+
+    static_assert(strlen_ct(CEPH_AES_IV) == AES_BLOCK_LEN);
+    std::array<unsigned char, AES_BLOCK_LEN> iv;
+    memcpy(iv.data(), CEPH_AES_IV, AES_BLOCK_LEN);
+
+    AES_cbc_encrypt(in.buf, out.buf, in.length, &dec_key, iv.data(),
+		    AES_DECRYPT);
+
+    // NOTE: we aren't handling partial decrypt. PKCS#7 padding must be
+    // at the end. If it's malformed, don't say a word to avoid risk of
+    // having an oracle. All we need to ensure is valid buffer boundary.
+    const auto pad_len = \
+      std::min<std::uint8_t>(out.buf[in.length - 1], AES_BLOCK_LEN);
+    return in.length - pad_len;
+  }
+};
+
+
+// ------------------------------------------------------------
+
+int CryptoAES::create(CryptoRandom *random, bufferptr& secret)
+{
+  bufferptr buf(AES_KEY_LEN);
+  random->get_bytes(buf.c_str(), buf.length());
+  secret = std::move(buf);
+  return 0;
+}
+
+int CryptoAES::validate_secret(const bufferptr& secret)
+{
+  if (secret.length() < AES_KEY_LEN) {
+    return -EINVAL;
+  }
+
+  return 0;
+}
+
+CryptoKeyHandler *CryptoAES::get_key_handler(const bufferptr& secret,
+					     string& error)
+{
+  CryptoAESKeyHandler *ckh = new CryptoAESKeyHandler;
+  ostringstream oss;
+  if (ckh->init(secret, oss) < 0) {
+    error = oss.str();
+    delete ckh;
+    return NULL;
+  }
+  return ckh;
+}
+
+
+
+
+// --
+
+
+// ---------------------------------------------------
+
+
+void CryptoKey::encode(bufferlist& bl) const
+{
+  using ceph::encode;
+  encode(type, bl);
+  encode(created, bl);
+  __u16 len = secret.length();
+  encode(len, bl);
+  bl.append(secret);
+}
+
+void CryptoKey::decode(bufferlist::const_iterator& bl)
+{
+  using ceph::decode;
+  decode(type, bl);
+  decode(created, bl);
+  __u16 len;
+  decode(len, bl);
+  bufferptr tmp;
+  bl.copy_deep(len, tmp);
+  if (_set_secret(type, tmp) < 0)
+    throw ceph::buffer::malformed_input("malformed secret");
+}
+
+int CryptoKey::set_secret(int type, const bufferptr& s, utime_t c)
+{
+  int r = _set_secret(type, s);
+  if (r < 0)
+    return r;
+  this->created = c;
+  return 0;
+}
+
+int CryptoKey::_set_secret(int t, const bufferptr& s)
+{
+  if (s.length() == 0) {
+    secret = s;
+    ckh.reset();
+    return 0;
+  }
+
+  CryptoHandler *ch = CryptoHandler::create(t);
+  if (ch) {
+    int ret = ch->validate_secret(s);
+    if (ret < 0) {
+      delete ch;
+      return ret;
+    }
+    string error;
+    ckh.reset(ch->get_key_handler(s, error));
+    delete ch;
+    if (error.length()) {
+      return -EIO;
+    }
+  } else {
+      return -EOPNOTSUPP;
+  }
+  type = t;
+  secret = s;
+  return 0;
+}
+
+int CryptoKey::create(CephContext *cct, int t)
+{
+  CryptoHandler *ch = CryptoHandler::create(t);
+  if (!ch) {
+    if (cct)
+      lderr(cct) << "ERROR: cct->get_crypto_handler(type=" << t << ") returned NULL" << dendl;
+    return -EOPNOTSUPP;
+  }
+  bufferptr s;
+  int r = ch->create(cct->random(), s);
+  delete ch;
+  if (r < 0)
+    return r;
+
+  r = _set_secret(t, s);
+  if (r < 0)
+    return r;
+  created = ceph_clock_now();
+  return r;
+}
+
+void CryptoKey::print(std::ostream &out) const
+{
+  out << encode_base64();
+}
+
+void CryptoKey::to_str(std::string& s) const
+{
+  int len = secret.length() * 4;
+  char buf[len];
+  hex2str(secret.c_str(), secret.length(), buf, len);
+  s = buf;
+}
+
+void CryptoKey::encode_formatted(string label, Formatter *f, bufferlist &bl)
+{
+  f->open_object_section(label.c_str());
+  f->dump_string("key", encode_base64());
+  f->close_section();
+  f->flush(bl);
+}
+
+void CryptoKey::encode_plaintext(bufferlist &bl)
+{
+  bl.append(encode_base64());
+}
+
+
+// ------------------
+
+CryptoHandler *CryptoHandler::create(int type)
+{
+  switch (type) {
+  case CEPH_CRYPTO_NONE:
+    return new CryptoNone;
+  case CEPH_CRYPTO_AES:
+    return new CryptoAES;
+  default:
+    return NULL;
+  }
+}
+
+#pragma clang diagnostic pop
+#pragma GCC diagnostic pop
diff --git a/src/auth/Crypto.h b/src/auth/Crypto.h
new file mode 100644
index 000000000..a29ac1abd
--- /dev/null
+++ b/src/auth/Crypto.h
@@ -0,0 +1,223 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_AUTH_CRYPTO_H
+#define CEPH_AUTH_CRYPTO_H
+
+#include "include/common_fwd.h"
+#include "include/types.h"
+#include "include/utime.h"
+#include "include/buffer.h"
+
+#include <string>
+
+class CryptoKeyContext;
+namespace ceph { class Formatter; }
+
+/*
+ * Random byte stream generator suitable for cryptographic use
+ */
+class CryptoRandom {
+public:
+  CryptoRandom(); // throws on failure
+  ~CryptoRandom();
+  /// copy up to 256 random bytes into the given buffer. throws on failure
+  void get_bytes(char *buf, int len);
+private:
+  static int open_urandom();
+  const int fd;
+};
+
+/*
+ * some per-key context that is specific to a particular crypto backend
+ */
+class CryptoKeyHandler {
+public:
+  // The maximum size of a single block for all descendants of the class.
+  static constexpr std::size_t MAX_BLOCK_SIZE {16};
+
+  // A descendant pick-ups one from these and passes it to the ctor template.
+  typedef std::integral_constant<std::size_t,  0> BLOCK_SIZE_0B;
+  typedef std::integral_constant<std::size_t, 16> BLOCK_SIZE_16B;
+
+  struct in_slice_t {
+    const std::size_t length;
+    const unsigned char* const buf;
+  };
+
+  struct out_slice_t {
+    const std::size_t max_length;
+    unsigned char* const buf;
+  };
+
+  ceph::bufferptr secret;
+
+  template <class BlockSizeT>
+  CryptoKeyHandler(BlockSizeT) {
+    static_assert(BlockSizeT::value <= MAX_BLOCK_SIZE);
+  }
+
+  virtual ~CryptoKeyHandler() {}
+
+  virtual int encrypt(const ceph::buffer::list& in,
+		      ceph::buffer::list& out, std::string *error) const = 0;
+  virtual int decrypt(const ceph::buffer::list& in,
+		      ceph::buffer::list& out, std::string *error) const = 0;
+
+  // TODO: provide nullptr in the out::buf to get/estimate size requirements?
+  // Or maybe dedicated methods?
+  virtual std::size_t encrypt(const in_slice_t& in,
+			      const out_slice_t& out) const;
+  virtual std::size_t decrypt(const in_slice_t& in,
+			      const out_slice_t& out) const;
+
+  sha256_digest_t hmac_sha256(const ceph::bufferlist& in) const;
+};
+
+/*
+ * match encoding of struct ceph_secret
+ */
+class CryptoKey {
+protected:
+  __u16 type;
+  utime_t created;
+  ceph::buffer::ptr secret;   // must set this via set_secret()!
+
+  // cache a pointer to the implementation-specific key handler, so we
+  // don't have to create it for every crypto operation.
+  mutable std::shared_ptr<CryptoKeyHandler> ckh;
+
+  int _set_secret(int type, const ceph::buffer::ptr& s);
+
+public:
+  CryptoKey() : type(0) { }
+  CryptoKey(int t, utime_t c, ceph::buffer::ptr& s)
+    : created(c) {
+    _set_secret(t, s);
+  }
+  ~CryptoKey() {
+  }
+
+  void encode(ceph::buffer::list& bl) const;
+  void decode(ceph::buffer::list::const_iterator& bl);
+
+  void clear() {
+    *this = CryptoKey();
+  }
+
+  int get_type() const { return type; }
+  utime_t get_created() const { return created; }
+  void print(std::ostream& out) const;
+
+  int set_secret(int type, const ceph::buffer::ptr& s, utime_t created);
+  const ceph::buffer::ptr& get_secret() { return secret; }
+  const ceph::buffer::ptr& get_secret() const { return secret; }
+
+  bool empty() const { return ckh.get() == nullptr; }
+
+  void encode_base64(std::string& s) const {
+    ceph::buffer::list bl;
+    encode(bl);
+    ceph::bufferlist e;
+    bl.encode_base64(e);
+    e.append('\0');
+    s = e.c_str();
+  }
+  std::string encode_base64() const {
+    std::string s;
+    encode_base64(s);
+    return s;
+  }
+  void decode_base64(const std::string& s) {
+    ceph::buffer::list e;
+    e.append(s);
+    ceph::buffer::list bl;
+    bl.decode_base64(e);
+    auto p = std::cbegin(bl);
+    decode(p);
+  }
+
+  void encode_formatted(std::string label, ceph::Formatter *f,
+			ceph::buffer::list &bl);
+  void encode_plaintext(ceph::buffer::list &bl);
+
+  // --
+  int create(CephContext *cct, int type);
+  int encrypt(CephContext *cct, const ceph::buffer::list& in,
+	      ceph::buffer::list& out,
+	      std::string *error) const {
+    ceph_assert(ckh); // Bad key?
+    return ckh->encrypt(in, out, error);
+  }
+  int decrypt(CephContext *cct, const ceph::buffer::list& in,
+	      ceph::buffer::list& out,
+	      std::string *error) const {
+    ceph_assert(ckh); // Bad key?
+    return ckh->decrypt(in, out, error);
+  }
+
+  using in_slice_t = CryptoKeyHandler::in_slice_t;
+  using out_slice_t = CryptoKeyHandler::out_slice_t;
+
+  std::size_t encrypt(CephContext*, const in_slice_t& in,
+		      const out_slice_t& out) {
+    ceph_assert(ckh);
+    return ckh->encrypt(in, out);
+  }
+  std::size_t decrypt(CephContext*, const in_slice_t& in,
+		      const out_slice_t& out) {
+    ceph_assert(ckh);
+    return ckh->encrypt(in, out);
+  }
+
+  sha256_digest_t hmac_sha256(CephContext*, const ceph::buffer::list& in) {
+    ceph_assert(ckh);
+    return ckh->hmac_sha256(in);
+  }
+
+  static constexpr std::size_t get_max_outbuf_size(std::size_t want_size) {
+    return want_size + CryptoKeyHandler::MAX_BLOCK_SIZE;
+  }
+
+  void to_str(std::string& s) const;
+};
+WRITE_CLASS_ENCODER(CryptoKey)
+
+inline std::ostream& operator<<(std::ostream& out, const CryptoKey& k)
+{
+  k.print(out);
+  return out;
+}
+
+
+/*
+ * Driver for a particular algorithm
+ *
+ * To use these functions, you need to call ceph::crypto::init(), see
+ * common/ceph_crypto.h. common_init_finish does this for you.
+ */
+class CryptoHandler {
+public:
+  virtual ~CryptoHandler() {}
+  virtual int get_type() const = 0;
+  virtual int create(CryptoRandom *random, ceph::buffer::ptr& secret) = 0;
+  virtual int validate_secret(const ceph::buffer::ptr& secret) = 0;
+  virtual CryptoKeyHandler *get_key_handler(const ceph::buffer::ptr& secret,
+					    std::string& error) = 0;
+
+  static CryptoHandler *create(int type);
+};
+
+
+#endif
diff --git a/src/auth/DummyAuth.h b/src/auth/DummyAuth.h
new file mode 100644
index 000000000..237518d41
--- /dev/null
+++ b/src/auth/DummyAuth.h
@@ -0,0 +1,63 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "AuthClient.h"
+#include "AuthServer.h"
+
+class DummyAuthClientServer : public AuthClient,
+			      public AuthServer {
+public:
+  DummyAuthClientServer(CephContext *cct) : AuthServer(cct) {}
+
+  // client
+  int get_auth_request(
+    Connection *con,
+    AuthConnectionMeta *auth_meta,
+    uint32_t *method,
+    std::vector<uint32_t> *preferred_modes,
+    bufferlist *out) override {
+    *method = CEPH_AUTH_NONE;
+    *preferred_modes = { CEPH_CON_MODE_CRC };
+    return 0;
+  }
+
+  int handle_auth_reply_more(
+    Connection *con,
+    AuthConnectionMeta *auth_meta,
+    const bufferlist& bl,
+    bufferlist *reply) override {
+    ceph_abort();
+  }
+
+  int handle_auth_done(
+    Connection *con,
+    AuthConnectionMeta *auth_meta,
+    uint64_t global_id,
+    uint32_t con_mode,
+    const bufferlist& bl,
+    CryptoKey *session_key,
+    std::string *connection_secret) {
+    return 0;
+  }
+
+  int handle_auth_bad_method(
+    Connection *con,
+    AuthConnectionMeta *auth_meta,
+    uint32_t old_auth_method,
+    int result,
+    const std::vector<uint32_t>& allowed_methods,
+    const std::vector<uint32_t>& allowed_modes) override {
+    ceph_abort();
+  }
+
+  // server
+  int handle_auth_request(
+    Connection *con,
+    AuthConnectionMeta *auth_meta,
+    bool more,
+    uint32_t auth_method,
+    const bufferlist& bl,
+    bufferlist *reply) override {
+    return 1;
+  }
+};
diff --git a/src/auth/KeyRing.cc b/src/auth/KeyRing.cc
new file mode 100644
index 000000000..eca429d0b
--- /dev/null
+++ b/src/auth/KeyRing.cc
@@ -0,0 +1,256 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include <errno.h>
+#include <map>
+#include <memory>
+#include <sstream>
+#include <algorithm>
+#include <boost/algorithm/string/replace.hpp>
+#include "auth/KeyRing.h"
+#include "include/stringify.h"
+#include "common/ceph_context.h"
+#include "common/config.h"
+#include "common/debug.h"
+#include "common/errno.h"
+#include "common/Formatter.h"
+
+#define dout_subsys ceph_subsys_auth
+
+#undef dout_prefix
+#define dout_prefix *_dout << "auth: "
+
+using std::map;
+using std::ostream;
+using std::ostringstream;
+using std::string;
+
+using ceph::bufferlist;
+using ceph::Formatter;
+
+int KeyRing::from_ceph_context(CephContext *cct)
+{
+  const auto& conf = cct->_conf;
+  string filename;
+
+  int ret = ceph_resolve_file_search(conf->keyring, filename);
+  if (!ret) {
+    ret = load(cct, filename);
+    if (ret < 0)
+      lderr(cct) << "failed to load " << filename
+		 << ": " << cpp_strerror(ret) << dendl;
+  } else if (conf->key.empty() && conf->keyfile.empty()) {
+    lderr(cct) << "unable to find a keyring on " << conf->keyring
+	       << ": " << cpp_strerror(ret) << dendl;
+  }
+
+  if (!conf->key.empty()) {
+    EntityAuth ea;
+    try {
+      ea.key.decode_base64(conf->key);
+      add(conf->name, ea);
+      return 0;
+    }
+    catch (ceph::buffer::error& e) {
+      lderr(cct) << "failed to decode key '" << conf->key << "'" << dendl;
+      return -EINVAL;
+    }
+  }
+
+  if (!conf->keyfile.empty()) {
+    bufferlist bl;
+    string err;
+    int r = bl.read_file(conf->keyfile.c_str(), &err);
+    if (r < 0) {
+      lderr(cct) << err << dendl;
+      return r;
+    }
+    string k(bl.c_str(), bl.length());
+    EntityAuth ea;
+    try {
+      ea.key.decode_base64(k);
+      add(conf->name, ea);
+    }
+    catch (ceph::buffer::error& e) {
+      lderr(cct) << "failed to decode key '" << k << "'" << dendl;
+      return -EINVAL;
+    }
+    return 0;
+  }
+
+  return ret;
+}
+
+int KeyRing::set_modifier(const char *type,
+			  const char *val,
+			  EntityName& name,
+			  map<string, bufferlist>& caps)
+{
+  if (!val)
+    return -EINVAL;
+
+  if (strcmp(type, "key") == 0) {
+    CryptoKey key;
+    string l(val);
+    try {
+      key.decode_base64(l);
+    } catch (const ceph::buffer::error& err) {
+      return -EINVAL;
+    }
+    set_key(name, key);
+  } else if (strncmp(type, "caps ", 5) == 0) {
+    const char *caps_entity = type + 5;
+    if (!*caps_entity)
+      return -EINVAL;
+    string l(val);
+    bufferlist bl;
+    encode(l, bl);
+    caps[caps_entity] = bl;
+    set_caps(name, caps);
+  } else if (strcmp(type, "auid") == 0) {
+    // just ignore it so we can still decode "old" keyrings that have an auid
+  } else
+    return -EINVAL;
+
+  return 0;
+}
+
+void KeyRing::encode_plaintext(bufferlist& bl)
+{
+  std::ostringstream os;
+  print(os);
+  string str = os.str();
+  bl.append(str);
+}
+
+void KeyRing::encode_formatted(string label, Formatter *f, bufferlist& bl)
+{
+  f->open_array_section(label.c_str());
+  for (const auto &[ename, eauth] : keys) {
+    f->open_object_section("auth_entities");
+    f->dump_string("entity", ename.to_str().c_str());
+    f->dump_string("key", stringify(eauth.key));
+    if (!eauth.pending_key.empty()) {
+      f->dump_string("pending_key", stringify(eauth.pending_key));
+    }
+    f->open_object_section("caps");
+    for (auto& [sys, capsbl] : eauth.caps) {
+      auto dataiter = capsbl.cbegin();
+      string caps;
+      ceph::decode(caps, dataiter);
+      f->dump_string(sys.c_str(), caps);
+    }
+    f->close_section();	/* caps */
+    f->close_section();	/* auth_entities */
+  }
+  f->close_section();	/* auth_dump */
+  f->flush(bl);
+}
+
+void KeyRing::decode(bufferlist::const_iterator& bli)
+{
+  int ret;
+  bufferlist bl;
+  bli.copy_all(bl);
+  ConfFile cf;
+
+  if (cf.parse_bufferlist(&bl, nullptr) != 0) {
+    throw ceph::buffer::malformed_input("cannot parse buffer");
+  }
+
+  for (auto& [name, section] : cf) {
+    if (name == "global")
+      continue;
+
+    EntityName ename;
+    map<string, bufferlist> caps;
+    if (!ename.from_str(name)) {
+      ostringstream oss;
+      oss << "bad entity name in keyring: " << name;
+      throw ceph::buffer::malformed_input(oss.str().c_str());
+    }
+
+    for (auto& [k, val] : section) {
+      if (k.empty())
+        continue;
+      string key;
+      std::replace_copy(k.begin(), k.end(), back_inserter(key), '_', ' ');
+      ret = set_modifier(key.c_str(), val.c_str(), ename, caps);
+      if (ret < 0) {
+	ostringstream oss;
+	oss << "error setting modifier for [" << name << "] type=" << key
+	    << " val=" << val;
+	throw ceph::buffer::malformed_input(oss.str().c_str());
+      }
+    }
+  }
+}
+
+int KeyRing::load(CephContext *cct, const std::string &filename)
+{
+  if (filename.empty())
+    return -EINVAL;
+
+  bufferlist bl;
+  std::string err;
+  int ret = bl.read_file(filename.c_str(), &err);
+  if (ret < 0) {
+    lderr(cct) << "error reading file: " << filename << ": " << err << dendl;
+    return ret;
+  }
+
+  try {
+    auto iter = bl.cbegin();
+    decode(iter);
+  }
+  catch (const ceph::buffer::error& err) {
+    lderr(cct) << "error parsing file " << filename << ": " << err.what() << dendl;
+    return -EIO;
+  }
+
+  ldout(cct, 2) << "KeyRing::load: loaded key file " << filename << dendl;
+  return 0;
+}
+
+void KeyRing::print(ostream& out)
+{
+  for (auto& [ename, eauth] : keys) {
+    out << "[" << ename << "]" << std::endl;
+    out << "\tkey = " << eauth.key << std::endl;
+    if (!eauth.pending_key.empty()) {
+      out << "\tpending key = " << eauth.pending_key << std::endl;
+    }
+
+    for (auto& [sys, capbl] : eauth.caps) {
+      auto dataiter = capbl.cbegin();
+      string caps;
+      ceph::decode(caps, dataiter);
+      boost::replace_all(caps, "\"", "\\\"");
+      out << "\tcaps " << sys << " = \"" << caps << '"' << std::endl;
+    }
+  }
+}
+
+void KeyRing::import(CephContext *cct, KeyRing& other)
+{
+  for (map<EntityName, EntityAuth>::iterator p = other.keys.begin();
+       p != other.keys.end();
+       ++p) {
+    ldout(cct, 10) << " importing " << p->first << dendl;
+    ldout(cct, 30) << "    " << p->second << dendl;
+    keys[p->first] = p->second;
+  }
+}
+
+
diff --git a/src/auth/KeyRing.h b/src/auth/KeyRing.h
new file mode 100644
index 000000000..c0fa28615
--- /dev/null
+++ b/src/auth/KeyRing.h
@@ -0,0 +1,114 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_KEYRING_H
+#define CEPH_KEYRING_H
+
+#include "auth/Auth.h"
+#include "include/common_fwd.h"
+
+class KeyRing : public KeyStore {
+  std::map<EntityName, EntityAuth> keys;
+
+  int set_modifier(const char *type, const char *val, EntityName& name, std::map<std::string, ceph::buffer::list>& caps);
+public:
+  void decode_plaintext(ceph::buffer::list::const_iterator& bl);
+  /* Create a KeyRing from a Ceph context.
+   * We will use the configuration stored inside the context. */
+  int from_ceph_context(CephContext *cct);
+
+  std::map<EntityName, EntityAuth>& get_keys() { return keys; }  // yuck
+
+  int load(CephContext *cct, const std::string &filename);
+  void print(std::ostream& out);
+
+  // accessors
+  bool exists(const EntityName& name) const {
+    auto p = keys.find(name);
+    return p != keys.end();
+  }
+  bool get_auth(const EntityName& name, EntityAuth &a) const {
+    std::map<EntityName, EntityAuth>::const_iterator k = keys.find(name);
+    if (k == keys.end())
+      return false;
+    a = k->second;
+    return true;
+  }
+  bool get_secret(const EntityName& name, CryptoKey& secret) const override {
+    std::map<EntityName, EntityAuth>::const_iterator k = keys.find(name);
+    if (k == keys.end())
+      return false;
+    secret = k->second.key;
+    return true;
+  }
+  bool get_service_secret(uint32_t service_id, uint64_t secret_id,
+			  CryptoKey& secret) const override {
+    return false;
+  }
+  bool get_caps(const EntityName& name,
+		const std::string& type, AuthCapsInfo& caps) const {
+    std::map<EntityName, EntityAuth>::const_iterator k = keys.find(name);
+    if (k == keys.end())
+      return false;
+    std::map<std::string,ceph::buffer::list>::const_iterator i = k->second.caps.find(type);
+    if (i != k->second.caps.end()) {
+      caps.caps = i->second;
+    }
+    return true;
+  }
+  size_t size() const {
+    return keys.size();
+  }
+
+  // modifiers
+  void add(const EntityName& name, const EntityAuth &a) {
+    keys[name] = a;
+  }
+  void add(const EntityName& name, const CryptoKey &k) {
+    EntityAuth a;
+    a.key = k;
+    keys[name] = a;
+  }
+  void add(const EntityName& name, const CryptoKey &k, const CryptoKey &pk) {
+    EntityAuth a;
+    a.key = k;
+    a.pending_key = pk;
+    keys[name] = a;
+  }
+  void remove(const EntityName& name) {
+    keys.erase(name);
+  }
+  void set_caps(EntityName& name, std::map<std::string, ceph::buffer::list>& caps) {
+    keys[name].caps = caps;
+  }
+  void set_key(EntityName& ename, CryptoKey& key) {
+    keys[ename].key = key;
+  }
+  void import(CephContext *cct, KeyRing& other);
+
+  // decode as plaintext
+  void decode(ceph::buffer::list::const_iterator& bl);
+
+  void encode_plaintext(ceph::buffer::list& bl);
+  void encode_formatted(std::string label, ceph::Formatter *f, ceph::buffer::list& bl);
+};
+
+// don't use WRITE_CLASS_ENCODER macro because we don't have an encode
+// macro.  don't juse encode_plaintext in that case because it is not
+// wrappable; it assumes it gets the entire ceph::buffer::list.
+static inline void decode(KeyRing& kr, ceph::buffer::list::const_iterator& p) {
+  kr.decode(p);
+}
+
+#endif
diff --git a/src/auth/RotatingKeyRing.cc b/src/auth/RotatingKeyRing.cc
new file mode 100644
index 000000000..4bc6af6ad
--- /dev/null
+++ b/src/auth/RotatingKeyRing.cc
@@ -0,0 +1,71 @@
+#include <map>
+
+#include "common/debug.h"
+#include "auth/RotatingKeyRing.h"
+#include "auth/KeyRing.h"
+
+#define dout_subsys ceph_subsys_auth
+#undef dout_prefix
+#define dout_prefix *_dout << "auth: "
+
+
+bool RotatingKeyRing::need_new_secrets() const
+{
+  std::lock_guard l{lock};
+  return secrets.need_new_secrets();
+}
+
+bool RotatingKeyRing::need_new_secrets(utime_t now) const
+{
+  std::lock_guard l{lock};
+  return secrets.need_new_secrets(now);
+}
+
+void RotatingKeyRing::set_secrets(RotatingSecrets&& s)
+{
+  std::lock_guard l{lock};
+  secrets = std::move(s);
+  dump_rotating();
+}
+
+void RotatingKeyRing::dump_rotating() const
+{
+  ldout(cct, 10) << "dump_rotating:" << dendl;
+  for (auto iter = secrets.secrets.begin();
+       iter != secrets.secrets.end();
+       ++iter)
+    ldout(cct, 10) << " id " << iter->first << " " << iter->second << dendl;
+}
+
+bool RotatingKeyRing::get_secret(const EntityName& name, CryptoKey& secret) const
+{
+  std::lock_guard l{lock};
+  return keyring->get_secret(name, secret);
+}
+
+bool RotatingKeyRing::get_service_secret(uint32_t service_id_, uint64_t secret_id,
+					 CryptoKey& secret) const
+{
+  std::lock_guard l{lock};
+
+  if (service_id_ != this->service_id) {
+    ldout(cct, 0) << "do not have service " << ceph_entity_type_name(service_id_)
+	    << ", i am " << ceph_entity_type_name(this->service_id) << dendl;
+    return false;
+  }
+
+  auto iter = secrets.secrets.find(secret_id);
+  if (iter == secrets.secrets.end()) {
+    ldout(cct, 0) << "could not find secret_id=" << secret_id << dendl;
+    dump_rotating();
+    return false;
+  }
+
+  secret = iter->second.key;
+  return true;
+}
+
+KeyRing* RotatingKeyRing::get_keyring()
+{
+  return keyring;
+}
diff --git a/src/auth/RotatingKeyRing.h b/src/auth/RotatingKeyRing.h
new file mode 100644
index 000000000..534eb5136
--- /dev/null
+++ b/src/auth/RotatingKeyRing.h
@@ -0,0 +1,53 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_ROTATINGKEYRING_H
+#define CEPH_ROTATINGKEYRING_H
+
+#include "common/ceph_mutex.h"
+#include "auth/Auth.h"
+#include "include/common_fwd.h"
+
+/*
+ * mediate access to a service's keyring and rotating secrets
+ */
+
+class KeyRing;
+
+class RotatingKeyRing : public KeyStore {
+  CephContext *cct;
+  uint32_t service_id;
+  RotatingSecrets secrets;
+  KeyRing *keyring;
+  mutable ceph::mutex lock;
+
+public:
+  RotatingKeyRing(CephContext *cct_, uint32_t s, KeyRing *kr) :
+    cct(cct_),
+    service_id(s),
+    keyring(kr),
+    lock{ceph::make_mutex("RotatingKeyRing::lock")}
+  {}
+
+  bool need_new_secrets() const;
+  bool need_new_secrets(utime_t now) const;
+  void set_secrets(RotatingSecrets&& s);
+  void dump_rotating() const;
+  bool get_secret(const EntityName& name, CryptoKey& secret) const override;
+  bool get_service_secret(uint32_t service_id, uint64_t secret_id,
+			  CryptoKey& secret) const override;
+  KeyRing *get_keyring();
+};
+
+#endif
diff --git a/src/auth/cephx/CephxAuthorizeHandler.cc b/src/auth/cephx/CephxAuthorizeHandler.cc
new file mode 100644
index 000000000..615b87500
--- /dev/null
+++ b/src/auth/cephx/CephxAuthorizeHandler.cc
@@ -0,0 +1,50 @@
+#include "CephxProtocol.h"
+#include "CephxAuthorizeHandler.h"
+#include "common/dout.h"
+
+#define dout_subsys ceph_subsys_auth
+
+bool CephxAuthorizeHandler::verify_authorizer(
+  CephContext *cct,
+  const KeyStore& keys,
+  const ceph::bufferlist& authorizer_data,
+  size_t connection_secret_required_len,
+  ceph::bufferlist *authorizer_reply,
+  EntityName *entity_name,
+  uint64_t *global_id,
+  AuthCapsInfo *caps_info,
+  CryptoKey *session_key,
+  std::string *connection_secret,
+  std::unique_ptr<AuthAuthorizerChallenge> *challenge)
+{
+  auto iter = authorizer_data.cbegin();
+
+  if (!authorizer_data.length()) {
+    ldout(cct, 1) << "verify authorizer, authorizer_data.length()=0" << dendl;
+    return false;
+  }
+
+  CephXServiceTicketInfo auth_ticket_info;
+
+  bool isvalid = cephx_verify_authorizer(cct, keys, iter,
+					 connection_secret_required_len,
+					 auth_ticket_info,
+					 challenge, connection_secret,
+					 authorizer_reply);
+
+  if (isvalid) {
+    *caps_info = auth_ticket_info.ticket.caps;
+    *entity_name = auth_ticket_info.ticket.name;
+    *global_id = auth_ticket_info.ticket.global_id;
+    *session_key = auth_ticket_info.session_key;
+  }
+
+  return isvalid;
+}
+
+// Return type of crypto used for this session's data;  for cephx, symmetric authentication
+
+int CephxAuthorizeHandler::authorizer_session_crypto() 
+{
+  return SESSION_SYMMETRIC_AUTHENTICATE;
+}
diff --git a/src/auth/cephx/CephxAuthorizeHandler.h b/src/auth/cephx/CephxAuthorizeHandler.h
new file mode 100644
index 000000000..626119078
--- /dev/null
+++ b/src/auth/cephx/CephxAuthorizeHandler.h
@@ -0,0 +1,39 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_CEPHXAUTHORIZEHANDLER_H
+#define CEPH_CEPHXAUTHORIZEHANDLER_H
+
+#include "auth/AuthAuthorizeHandler.h"
+#include "include/common_fwd.h"
+
+struct CephxAuthorizeHandler : public AuthAuthorizeHandler {
+  bool verify_authorizer(
+    CephContext *cct,
+    const KeyStore& keys,
+    const ceph::buffer::list& authorizer_data,
+    size_t connection_secret_required_len,
+    ceph::buffer::list *authorizer_reply,
+    EntityName *entity_name,
+    uint64_t *global_id,
+    AuthCapsInfo *caps_info,
+    CryptoKey *session_key,
+    std::string *connection_secret,
+    std::unique_ptr<AuthAuthorizerChallenge> *challenge) override;
+  int authorizer_session_crypto() override;
+};
+
+
+
+#endif
diff --git a/src/auth/cephx/CephxClientHandler.cc b/src/auth/cephx/CephxClientHandler.cc
new file mode 100644
index 000000000..76ccca735
--- /dev/null
+++ b/src/auth/cephx/CephxClientHandler.cc
@@ -0,0 +1,333 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+
+#include <errno.h>
+
+#include "CephxClientHandler.h"
+#include "CephxProtocol.h"
+
+#include "auth/KeyRing.h"
+#include "include/random.h"
+#include "common/ceph_context.h"
+#include "common/config.h"
+#include "common/dout.h"
+
+#define dout_subsys ceph_subsys_auth
+#undef dout_prefix
+#define dout_prefix *_dout << "cephx client: "
+
+using std::string;
+
+using ceph::bufferlist;
+
+void CephxClientHandler::reset()
+{
+  ldout(cct,10) << __func__ << dendl;
+  starting = true;
+  server_challenge = 0;
+}
+
+int CephxClientHandler::build_request(bufferlist& bl) const
+{
+  ldout(cct, 10) << "build_request" << dendl;
+
+  if (need & CEPH_ENTITY_TYPE_AUTH) {
+    /* authenticate */
+    CephXRequestHeader header;
+    header.request_type = CEPHX_GET_AUTH_SESSION_KEY;
+    encode(header, bl);
+
+    CryptoKey secret;
+    const bool got = keyring->get_secret(cct->_conf->name, secret);
+    if (!got) {
+      ldout(cct, 20) << "no secret found for entity: " << cct->_conf->name << dendl;
+      return -ENOENT;
+    }
+
+    // is the key OK?
+    if (!secret.get_secret().length()) {
+      ldout(cct, 20) << "secret for entity " << cct->_conf->name << " is invalid" << dendl;
+      return -EINVAL;
+    }
+
+    CephXAuthenticate req;
+    req.client_challenge = ceph::util::generate_random_number<uint64_t>();
+    std::string error;
+    cephx_calc_client_server_challenge(cct, secret, server_challenge,
+				       req.client_challenge, &req.key, error);
+    if (!error.empty()) {
+      ldout(cct, 20) << "cephx_calc_client_server_challenge error: " << error << dendl;
+      return -EIO;
+    }
+
+    req.old_ticket = ticket_handler->ticket;
+
+    // for nautilus+ servers: request other keys at the same time
+    req.other_keys = need;
+
+    if (req.old_ticket.blob.length()) {
+      ldout(cct, 20) << "old ticket len=" << req.old_ticket.blob.length() << dendl;
+    }
+
+    encode(req, bl);
+
+    ldout(cct, 10) << "get auth session key: client_challenge "
+		   << std::hex << req.client_challenge << std::dec << dendl;
+    return 0;
+  }
+
+  if (_need_tickets()) {
+    /* get service tickets */
+    ldout(cct, 10) << "get service keys: want=" << want << " need=" << need << " have=" << have << dendl;
+
+    CephXRequestHeader header;
+    header.request_type = CEPHX_GET_PRINCIPAL_SESSION_KEY;
+    encode(header, bl);
+
+    CephXAuthorizer *authorizer = ticket_handler->build_authorizer(global_id);
+    if (!authorizer)
+      return -EINVAL;
+    bl.claim_append(authorizer->bl);
+    delete authorizer;
+
+    CephXServiceTicketRequest req;
+    req.keys = need;
+    encode(req, bl);
+  }
+
+  return 0;
+}
+
+bool CephxClientHandler::_need_tickets() const
+{
+  // do not bother (re)requesting tickets if we *only* need the MGR
+  // ticket; that can happen during an upgrade and we want to avoid a
+  // loop.  we'll end up re-requesting it later when the secrets
+  // rotating.
+  return need && need != CEPH_ENTITY_TYPE_MGR;
+}
+
+int CephxClientHandler::handle_response(
+  int ret,
+  bufferlist::const_iterator& indata,
+  CryptoKey *session_key,
+  std::string *connection_secret)
+{
+  ldout(cct, 10) << this << " handle_response ret = " << ret << dendl;
+  
+  if (ret < 0)
+    return ret; // hrm!
+
+  if (starting) {
+    CephXServerChallenge ch;
+    try {
+      decode(ch, indata);
+    } catch (ceph::buffer::error& e) {
+      ldout(cct, 1) << __func__ << " failed to decode CephXServerChallenge: "
+		    << e.what() << dendl;
+      return -EPERM;
+    }
+    server_challenge = ch.server_challenge;
+    ldout(cct, 10) << " got initial server challenge "
+		   << std::hex << server_challenge << std::dec << dendl;
+    starting = false;
+
+    tickets.invalidate_ticket(CEPH_ENTITY_TYPE_AUTH);
+    return -EAGAIN;
+  }
+
+  struct CephXResponseHeader header;
+  try {
+    decode(header, indata);
+  } catch (ceph::buffer::error& e) {
+    ldout(cct, 1) << __func__ << " failed to decode CephXResponseHeader: "
+		  << e.what() << dendl;
+    return -EPERM;
+  }
+
+  switch (header.request_type) {
+  case CEPHX_GET_AUTH_SESSION_KEY:
+    {
+      ldout(cct, 10) << " get_auth_session_key" << dendl;
+      CryptoKey secret;
+      const bool got = keyring->get_secret(cct->_conf->name, secret);
+      if (!got) {
+	ldout(cct, 0) << "key not found for " << cct->_conf->name << dendl;
+	return -ENOENT;
+      }
+	
+      if (!tickets.verify_service_ticket_reply(secret, indata)) {
+	ldout(cct, 0) << "could not verify service_ticket reply" << dendl;
+	return -EACCES;
+      }
+      ldout(cct, 10) << " want=" << want << " need=" << need << " have=" << have << dendl;
+      if (!indata.end()) {
+	bufferlist cbl, extra_tickets;
+	using ceph::decode;
+	try {
+	  decode(cbl, indata);
+	  decode(extra_tickets, indata);
+	} catch (ceph::buffer::error& e) {
+	  ldout(cct, 1) << __func__ << " failed to decode tickets: "
+			<< e.what() << dendl;
+	  return -EPERM;
+	}
+	ldout(cct, 10) << " got connection bl " << cbl.length()
+		       << " and extra tickets " << extra_tickets.length()
+		       << dendl;
+	// for msgr1, both session_key and connection_secret are NULL
+	// so we skip extra_tickets and incur an additional round-trip
+	// to get service tickets via CEPHX_GET_PRINCIPAL_SESSION_KEY
+	// as if talking to a pre-nautilus mon
+	// this wasn't intended but turns out to be needed because in
+	// msgr1 case MonClient doesn't explicitly wait for the monmap
+	// (which is shared together with CEPHX_GET_AUTH_SESSION_KEY
+	// reply)
+	// instead, it waits for CEPHX_GET_PRINCIPAL_SESSION_KEY reply
+	// which comes after the monmap and hence the monmap is always
+	// handled by the time authentication is considered finished
+	// if we start to always process extra_tickets here, MonClient
+	// would have no reason to send CEPHX_GET_PRINCIPAL_SESSION_KEY
+	// and RadosClient::connect() or similar could return with no
+	// actual monmap but just an initial bootstrap stub, leading
+	// to mon commands going out with zero fsid and other issues
+	if (session_key && connection_secret) {
+	  CephXTicketHandler& ticket_handler =
+	    tickets.get_handler(CEPH_ENTITY_TYPE_AUTH);
+	  if (session_key) {
+	    *session_key = ticket_handler.session_key;
+	  }
+	  if (cbl.length() && connection_secret) {
+	    auto p = cbl.cbegin();
+	    string err;
+	    if (decode_decrypt(cct, *connection_secret, *session_key, p,
+			       err)) {
+	      lderr(cct) << __func__ << " failed to decrypt connection_secret"
+			 << dendl;
+	    } else {
+	      ldout(cct, 10) << " got connection_secret "
+			     << connection_secret->size() << " bytes" << dendl;
+	    }
+	  }
+	  if (extra_tickets.length())  {
+	    auto p = extra_tickets.cbegin();
+	    if (!tickets.verify_service_ticket_reply(
+		  *session_key, p)) {
+	      lderr(cct) << "could not verify extra service_tickets" << dendl;
+	    } else {
+	      ldout(cct, 10) << " got extra service_tickets" << dendl;
+	    }
+	  }
+	}
+      }
+      validate_tickets();
+      if (_need_tickets())
+	ret = -EAGAIN;
+      else
+	ret = 0;
+      }
+    break;
+
+  case CEPHX_GET_PRINCIPAL_SESSION_KEY:
+    {
+      CephXTicketHandler& ticket_handler = tickets.get_handler(CEPH_ENTITY_TYPE_AUTH);
+      ldout(cct, 10) << " get_principal_session_key session_key " << ticket_handler.session_key << dendl;
+  
+      if (!tickets.verify_service_ticket_reply(ticket_handler.session_key, indata)) {
+        ldout(cct, 0) << "could not verify service_ticket reply" << dendl;
+        return -EACCES;
+      }
+      validate_tickets();
+      if (!_need_tickets()) {
+	ret = 0;
+      }
+    }
+    break;
+
+  case CEPHX_GET_ROTATING_KEY:
+    {
+      ldout(cct, 10) << " get_rotating_key" << dendl;
+      if (rotating_secrets) {
+	RotatingSecrets secrets;
+	CryptoKey secret_key;
+	const bool got = keyring->get_secret(cct->_conf->name, secret_key);
+        if (!got) {
+          ldout(cct, 0) << "key not found for " << cct->_conf->name << dendl;
+          return -ENOENT;
+        }
+	std::string error;
+	if (decode_decrypt(cct, secrets, secret_key, indata, error)) {
+	  ldout(cct, 0) << "could not set rotating key: decode_decrypt failed. error:"
+	    << error << dendl;
+	  return -EINVAL;
+	} else {
+	  rotating_secrets->set_secrets(std::move(secrets));
+	}
+      }
+    }
+    break;
+
+  default:
+   ldout(cct, 0) << " unknown request_type " << header.request_type << dendl;
+   ceph_abort();
+  }
+  return ret;
+}
+
+
+AuthAuthorizer *CephxClientHandler::build_authorizer(uint32_t service_id) const
+{
+  ldout(cct, 10) << "build_authorizer for service " << ceph_entity_type_name(service_id) << dendl;
+  return tickets.build_authorizer(service_id);
+}
+
+
+bool CephxClientHandler::build_rotating_request(bufferlist& bl) const
+{
+  ldout(cct, 10) << "build_rotating_request" << dendl;
+  CephXRequestHeader header;
+  header.request_type = CEPHX_GET_ROTATING_KEY;
+  encode(header, bl);
+  return true;
+}
+
+void CephxClientHandler::prepare_build_request()
+{
+  ldout(cct, 10) << "validate_tickets: want=" << want << " need=" << need
+		 << " have=" << have << dendl;
+  validate_tickets();
+  ldout(cct, 10) << "want=" << want << " need=" << need << " have=" << have
+		 << dendl;
+
+  ticket_handler = &(tickets.get_handler(CEPH_ENTITY_TYPE_AUTH));
+}
+
+void CephxClientHandler::validate_tickets()
+{
+  // lock should be held for write
+  tickets.validate_tickets(want, have, need);
+}
+
+bool CephxClientHandler::need_tickets()
+{
+  validate_tickets();
+
+  ldout(cct, 20) << "need_tickets: want=" << want
+		 << " have=" << have
+		 << " need=" << need
+		 << dendl;
+
+  return _need_tickets();
+}
diff --git a/src/auth/cephx/CephxClientHandler.h b/src/auth/cephx/CephxClientHandler.h
new file mode 100644
index 000000000..601a5c69f
--- /dev/null
+++ b/src/auth/cephx/CephxClientHandler.h
@@ -0,0 +1,78 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_CEPHXCLIENTHANDLER_H
+#define CEPH_CEPHXCLIENTHANDLER_H
+
+#include "auth/AuthClientHandler.h"
+#include "CephxProtocol.h"
+#include "auth/RotatingKeyRing.h"
+#include "include/common_fwd.h"
+
+class KeyRing;
+
+class CephxClientHandler : public AuthClientHandler {
+  bool starting;
+
+  /* envelope protocol parameters */
+  uint64_t server_challenge;
+
+  CephXTicketManager tickets;
+  CephXTicketHandler* ticket_handler;
+
+  RotatingKeyRing* rotating_secrets;
+  KeyRing *keyring;
+
+public:
+  CephxClientHandler(CephContext *cct_,
+		     RotatingKeyRing *rsecrets)
+    : AuthClientHandler(cct_),
+      starting(false),
+      server_challenge(0),
+      tickets(cct_),
+      ticket_handler(NULL),
+      rotating_secrets(rsecrets),
+      keyring(rsecrets->get_keyring())
+  {
+    reset();
+  }
+
+  CephxClientHandler* clone() const override {
+    return new CephxClientHandler(*this);
+  }
+
+  void reset() override;
+  void prepare_build_request() override;
+  int build_request(ceph::buffer::list& bl) const override;
+  int handle_response(int ret, ceph::buffer::list::const_iterator& iter,
+		      CryptoKey *session_key,
+		      std::string *connection_secret) override;
+  bool build_rotating_request(ceph::buffer::list& bl) const override;
+
+  int get_protocol() const override { return CEPH_AUTH_CEPHX; }
+
+  AuthAuthorizer *build_authorizer(uint32_t service_id) const override;
+
+  bool need_tickets() override;
+
+  void set_global_id(uint64_t id) override {
+    global_id = id;
+    tickets.global_id = id;
+  }
+private:
+  void validate_tickets() override;
+  bool _need_tickets() const;
+};
+
+#endif
diff --git a/src/auth/cephx/CephxKeyServer.cc b/src/auth/cephx/CephxKeyServer.cc
new file mode 100644
index 000000000..236ac451a
--- /dev/null
+++ b/src/auth/cephx/CephxKeyServer.cc
@@ -0,0 +1,479 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#include "common/config.h"
+#include "CephxKeyServer.h"
+#include "common/dout.h"
+#include <sstream>
+
+#define dout_subsys ceph_subsys_auth
+#undef dout_prefix
+#define dout_prefix *_dout << "cephx keyserverdata: "
+
+using std::ostringstream;
+using std::string;
+using std::stringstream;
+
+using ceph::bufferptr;
+using ceph::bufferlist;
+using ceph::Formatter;
+
+bool KeyServerData::get_service_secret(CephContext *cct, uint32_t service_id,
+				       CryptoKey& secret, uint64_t& secret_id,
+				       double& ttl) const
+{
+  auto iter = rotating_secrets.find(service_id);
+  if (iter == rotating_secrets.end()) { 
+    ldout(cct, 10) << "get_service_secret service " << ceph_entity_type_name(service_id) << " not found " << dendl;
+    return false;
+  }
+
+  const RotatingSecrets& secrets = iter->second;
+
+  // second to oldest, unless it's expired
+  auto riter = secrets.secrets.begin();
+  if (secrets.secrets.size() > 1)
+    ++riter;
+
+  utime_t now = ceph_clock_now();
+  if (riter->second.expiration < now)
+    ++riter;   // "current" key has expired, use "next" key instead
+
+  secret_id = riter->first;
+  secret = riter->second.key;
+
+  // ttl may have just been increased by the user
+  // cap it by expiration of "next" key to prevent handing out a ticket
+  // with a bogus, possibly way into the future, validity
+  ttl = service_id == CEPH_ENTITY_TYPE_AUTH ?
+      cct->_conf->auth_mon_ticket_ttl : cct->_conf->auth_service_ticket_ttl;
+  ttl = std::min(ttl, static_cast<double>(
+		     secrets.secrets.rbegin()->second.expiration - now));
+
+  ldout(cct, 30) << __func__ << " service "
+		 << ceph_entity_type_name(service_id) << " secret_id "
+		 << secret_id << " " << riter->second << " ttl " << ttl
+		 << dendl;
+  return true;
+}
+
+bool KeyServerData::get_service_secret(CephContext *cct, uint32_t service_id,
+				uint64_t secret_id, CryptoKey& secret) const
+{
+  auto iter = rotating_secrets.find(service_id);
+  if (iter == rotating_secrets.end()) {
+    ldout(cct, 10) << __func__ << " no rotating_secrets for service " << service_id
+		   << " " << ceph_entity_type_name(service_id) << dendl;
+    return false;
+  }
+
+  const RotatingSecrets& secrets = iter->second;
+  auto riter = secrets.secrets.find(secret_id);
+
+  if (riter == secrets.secrets.end()) {
+    ldout(cct, 10) << "get_service_secret service " << ceph_entity_type_name(service_id)
+	     << " secret " << secret_id << " not found" << dendl;
+    ldout(cct, 30) << " I have:" << dendl;
+    for (auto iter = secrets.secrets.begin();
+	 iter != secrets.secrets.end();
+	 ++iter)
+      ldout(cct, 30) << " id " << iter->first << " " << iter->second << dendl;
+    return false;
+  }
+
+  secret = riter->second.key;
+
+  return true;
+}
+bool KeyServerData::get_auth(const EntityName& name, EntityAuth& auth) const {
+  auto iter = secrets.find(name);
+  if (iter != secrets.end()) {
+    auth = iter->second;
+    return true;
+  }
+  return extra_secrets->get_auth(name, auth);
+}
+
+bool KeyServerData::get_secret(const EntityName& name, CryptoKey& secret) const {
+  auto iter = secrets.find(name);
+  if (iter != secrets.end()) {
+    secret = iter->second.key;
+    return true;
+  }
+  return extra_secrets->get_secret(name, secret);
+}
+
+bool KeyServerData::get_caps(CephContext *cct, const EntityName& name,
+			     const string& type, AuthCapsInfo& caps_info) const
+{
+  caps_info.allow_all = false;
+
+  ldout(cct, 10) << "get_caps: name=" << name.to_str() << dendl;
+  auto iter = secrets.find(name);
+  if (iter != secrets.end()) {
+    ldout(cct, 10) << "get_caps: num of caps=" << iter->second.caps.size() << dendl;
+    auto capsiter = iter->second.caps.find(type);
+    if (capsiter != iter->second.caps.end()) {
+      caps_info.caps = capsiter->second;
+    }
+    return true;
+  }
+
+  return extra_secrets->get_caps(name, type, caps_info);
+}
+
+
+#undef dout_prefix
+#define dout_prefix *_dout << "cephx keyserver: "
+
+
+KeyServer::KeyServer(CephContext *cct_, KeyRing *extra_secrets)
+  : cct(cct_),
+    data(extra_secrets),
+    lock{ceph::make_mutex("KeyServer::lock")}
+{
+}
+
+int KeyServer::start_server()
+{
+  std::scoped_lock l{lock};
+  _dump_rotating_secrets();
+  return 0;
+}
+
+void KeyServer::dump()
+{
+  _dump_rotating_secrets();
+}
+
+void KeyServer::_dump_rotating_secrets()
+{
+  ldout(cct, 30) << "_dump_rotating_secrets" << dendl;
+  for (auto iter = data.rotating_secrets.begin();
+       iter != data.rotating_secrets.end();
+       ++iter) {
+    RotatingSecrets& key = iter->second;
+    for (auto mapiter = key.secrets.begin();
+	 mapiter != key.secrets.end();
+	 ++mapiter)
+      ldout(cct, 30) << "service " << ceph_entity_type_name(iter->first)
+	             << " id " << mapiter->first
+	             << " key " << mapiter->second << dendl;
+  }
+}
+
+int KeyServer::_rotate_secret(uint32_t service_id, KeyServerData &pending_data)
+{
+  RotatingSecrets& r = pending_data.rotating_secrets[service_id];
+  int added = 0;
+  utime_t now = ceph_clock_now();
+  double ttl = service_id == CEPH_ENTITY_TYPE_AUTH ? cct->_conf->auth_mon_ticket_ttl : cct->_conf->auth_service_ticket_ttl;
+
+  while (r.need_new_secrets(now)) {
+    ExpiringCryptoKey ek;
+    generate_secret(ek.key);
+    if (r.empty()) {
+      ek.expiration = now;
+    } else {
+      utime_t next_ttl = now;
+      next_ttl += ttl;
+      ek.expiration = std::max(next_ttl, r.next().expiration);
+    }
+    ek.expiration += ttl;
+    uint64_t secret_id = r.add(ek);
+    ldout(cct, 10) << "_rotate_secret adding " << ceph_entity_type_name(service_id) << dendl;
+    ldout(cct, 30) << "_rotate_secret adding " << ceph_entity_type_name(service_id)
+	           << " id " << secret_id << " " << ek
+	           << dendl;
+    added++;
+  }
+  return added;
+}
+
+bool KeyServer::get_secret(const EntityName& name, CryptoKey& secret) const
+{
+  std::scoped_lock l{lock};
+  return data.get_secret(name, secret);
+}
+
+bool KeyServer::get_auth(const EntityName& name, EntityAuth& auth) const
+{
+  std::scoped_lock l{lock};
+  return data.get_auth(name, auth);
+}
+
+bool KeyServer::get_caps(const EntityName& name, const string& type,
+	      AuthCapsInfo& caps_info) const
+{
+  std::scoped_lock l{lock};
+
+  return data.get_caps(cct, name, type, caps_info);
+}
+
+bool KeyServer::get_service_secret(uint32_t service_id, CryptoKey& secret,
+				   uint64_t& secret_id, double& ttl) const
+{
+  std::scoped_lock l{lock};
+
+  return data.get_service_secret(cct, service_id, secret, secret_id, ttl);
+}
+
+bool KeyServer::get_service_secret(uint32_t service_id,
+		uint64_t secret_id, CryptoKey& secret) const
+{
+  std::scoped_lock l{lock};
+
+  return data.get_service_secret(cct, service_id, secret_id, secret);
+}
+
+void KeyServer::note_used_pending_key(const EntityName& name, const CryptoKey& key)
+{
+  std::scoped_lock l(lock);
+  used_pending_keys[name] = key;
+}
+
+void KeyServer::clear_used_pending_keys()
+{
+  std::scoped_lock l(lock);
+  used_pending_keys.clear();
+}
+
+std::map<EntityName,CryptoKey> KeyServer::get_used_pending_keys()
+{
+  std::map<EntityName,CryptoKey> ret;
+  std::scoped_lock l(lock);
+  ret.swap(used_pending_keys);
+  return ret;
+}
+
+bool KeyServer::generate_secret(CryptoKey& secret)
+{
+  bufferptr bp;
+  CryptoHandler *crypto = cct->get_crypto_handler(CEPH_CRYPTO_AES);
+  if (!crypto)
+    return false;
+
+  if (crypto->create(cct->random(), bp) < 0)
+    return false;
+
+  secret.set_secret(CEPH_CRYPTO_AES, bp, ceph_clock_now());
+
+  return true;
+}
+
+bool KeyServer::generate_secret(EntityName& name, CryptoKey& secret)
+{
+  if (!generate_secret(secret))
+    return false;
+
+  std::scoped_lock l{lock};
+
+  EntityAuth auth;
+  auth.key = secret;
+
+  data.add_auth(name, auth);
+
+  return true;
+}
+
+bool KeyServer::contains(const EntityName& name) const
+{
+  std::scoped_lock l{lock};
+
+  return data.contains(name);
+}
+
+int KeyServer::encode_secrets(Formatter *f, stringstream *ds) const
+{
+  std::scoped_lock l{lock};
+  auto mapiter = data.secrets_begin();
+
+  if (mapiter == data.secrets_end())
+    return -ENOENT;
+
+  if (f)
+    f->open_array_section("auth_dump");
+
+  while (mapiter != data.secrets_end()) {
+    const EntityName& name = mapiter->first;
+    if (ds) {
+      *ds << name.to_str() << std::endl;
+      *ds << "\tkey: " << mapiter->second.key << std::endl;
+    }
+    if (f) {
+      f->open_object_section("auth_entities");
+      f->dump_string("entity", name.to_str());
+      f->dump_stream("key") << mapiter->second.key;
+      f->open_object_section("caps");
+    }
+
+    auto capsiter = mapiter->second.caps.begin();
+    for (; capsiter != mapiter->second.caps.end(); ++capsiter) {
+      // FIXME: need a const_iterator for bufferlist, but it doesn't exist yet.
+      bufferlist *bl = const_cast<bufferlist*>(&capsiter->second);
+      auto dataiter = bl->cbegin();
+      string caps;
+      using ceph::decode;
+      decode(caps, dataiter);
+      if (ds)
+        *ds << "\tcaps: [" << capsiter->first << "] " << caps << std::endl;
+      if (f)
+        f->dump_string(capsiter->first.c_str(), caps);
+    }
+    if (f) {
+      f->close_section(); // caps
+      f->close_section(); // auth_entities
+    }
+
+    ++mapiter;
+  }
+
+  if (f)
+    f->close_section(); // auth_dump
+  return 0;
+}
+
+void KeyServer::encode_formatted(string label, Formatter *f, bufferlist &bl)
+{
+  ceph_assert(f != NULL);
+  f->open_object_section(label.c_str());
+  encode_secrets(f, NULL);
+  f->close_section();
+  f->flush(bl);
+}
+
+void KeyServer::encode_plaintext(bufferlist &bl)
+{
+  stringstream os;
+  encode_secrets(NULL, &os);
+  bl.append(os.str());
+}
+
+bool KeyServer::prepare_rotating_update(bufferlist& rotating_bl)
+{
+  std::scoped_lock l{lock};
+  ldout(cct, 20) << __func__ << " before: data.rotating_ver=" << data.rotating_ver
+		 << dendl;
+
+  KeyServerData pending_data(nullptr);
+  pending_data.rotating_ver = data.rotating_ver + 1;
+  pending_data.rotating_secrets = data.rotating_secrets;
+
+  int added = 0;
+  added += _rotate_secret(CEPH_ENTITY_TYPE_AUTH, pending_data);
+  added += _rotate_secret(CEPH_ENTITY_TYPE_MON, pending_data);
+  added += _rotate_secret(CEPH_ENTITY_TYPE_OSD, pending_data);
+  added += _rotate_secret(CEPH_ENTITY_TYPE_MDS, pending_data);
+  added += _rotate_secret(CEPH_ENTITY_TYPE_MGR, pending_data);
+  if (!added) {
+    return false;
+  }
+
+  ldout(cct, 20) << __func__ << " after: pending_data.rotating_ver="
+		 << pending_data.rotating_ver
+		 << dendl;
+  pending_data.encode_rotating(rotating_bl);
+  return true;
+}
+
+bool KeyServer::get_rotating_encrypted(const EntityName& name,
+	bufferlist& enc_bl) const
+{
+  std::scoped_lock l{lock};
+
+  auto mapiter = data.find_name(name);
+  if (mapiter == data.secrets_end())
+    return false;
+
+  const CryptoKey& specific_key = mapiter->second.key;
+
+  auto rotate_iter = data.rotating_secrets.find(name.get_type());
+  if (rotate_iter == data.rotating_secrets.end())
+    return false;
+
+  RotatingSecrets secrets = rotate_iter->second;
+
+  std::string error;
+  if (encode_encrypt(cct, secrets, specific_key, enc_bl, error))
+    return false;
+
+  return true;
+}
+
+bool KeyServer::_get_service_caps(const EntityName& name, uint32_t service_id,
+				  AuthCapsInfo& caps_info) const
+{
+  string s = ceph_entity_type_name(service_id);
+
+  return data.get_caps(cct, name, s, caps_info);
+}
+
+bool KeyServer::get_service_caps(const EntityName& name, uint32_t service_id,
+				 AuthCapsInfo& caps_info) const
+{
+  std::scoped_lock l{lock};
+  return _get_service_caps(name, service_id, caps_info);
+}
+
+
+int KeyServer::_build_session_auth_info(uint32_t service_id,
+					const AuthTicket& parent_ticket,
+					CephXSessionAuthInfo& info,
+					double ttl)
+{
+  info.service_id = service_id;
+  info.ticket = parent_ticket;
+  info.ticket.init_timestamps(ceph_clock_now(), ttl);
+  info.validity.set_from_double(ttl);
+
+  generate_secret(info.session_key);
+
+  // mon keys are stored externally.  and the caps are blank anyway.
+  if (service_id != CEPH_ENTITY_TYPE_MON) {
+    string s = ceph_entity_type_name(service_id);
+    if (!data.get_caps(cct, info.ticket.name, s, info.ticket.caps)) {
+      return -EINVAL;
+    }
+  }
+  return 0;
+}
+
+int KeyServer::build_session_auth_info(uint32_t service_id,
+				       const AuthTicket& parent_ticket,
+				       CephXSessionAuthInfo& info)
+{
+  double ttl;
+  if (!get_service_secret(service_id, info.service_secret, info.secret_id,
+			  ttl)) {
+    return -EACCES;
+  }
+
+  std::scoped_lock l{lock};
+  return _build_session_auth_info(service_id, parent_ticket, info, ttl);
+}
+
+int KeyServer::build_session_auth_info(uint32_t service_id,
+				       const AuthTicket& parent_ticket,
+				       const CryptoKey& service_secret,
+				       uint64_t secret_id,
+				       CephXSessionAuthInfo& info)
+{
+  info.service_secret = service_secret;
+  info.secret_id = secret_id;
+
+  std::scoped_lock l{lock};
+  return _build_session_auth_info(service_id, parent_ticket, info,
+				  cct->_conf->auth_service_ticket_ttl);
+}
+
diff --git a/src/auth/cephx/CephxKeyServer.h b/src/auth/cephx/CephxKeyServer.h
new file mode 100644
index 000000000..64915c8ce
--- /dev/null
+++ b/src/auth/cephx/CephxKeyServer.h
@@ -0,0 +1,323 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_KEYSSERVER_H
+#define CEPH_KEYSSERVER_H
+
+#include "auth/KeyRing.h"
+#include "CephxProtocol.h"
+#include "common/ceph_mutex.h"
+#include "include/common_fwd.h"
+
+struct KeyServerData {
+  version_t version;
+
+  /* for each entity */
+  std::map<EntityName, EntityAuth> secrets;
+  KeyRing *extra_secrets;
+
+  /* for each service type */
+  version_t rotating_ver;
+  std::map<uint32_t, RotatingSecrets> rotating_secrets;
+
+  explicit KeyServerData(KeyRing *extra)
+    : version(0),
+      extra_secrets(extra),
+      rotating_ver(0) {}
+
+  void encode(ceph::buffer::list& bl) const {
+     __u8 struct_v = 1;
+    using ceph::encode;
+    encode(struct_v, bl);
+    encode(version, bl);
+    encode(rotating_ver, bl);
+    encode(secrets, bl);
+    encode(rotating_secrets, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    __u8 struct_v;
+    decode(struct_v, bl);
+    decode(version, bl);
+    decode(rotating_ver, bl);
+    decode(secrets, bl);
+    decode(rotating_secrets, bl);
+  }
+
+  void encode_rotating(ceph::buffer::list& bl) const {
+    using ceph::encode;
+     __u8 struct_v = 1;
+    encode(struct_v, bl);
+    encode(rotating_ver, bl);
+    encode(rotating_secrets, bl);
+  }
+  void decode_rotating(ceph::buffer::list& rotating_bl) {
+    using ceph::decode;
+    auto iter = rotating_bl.cbegin();
+    __u8 struct_v;
+    decode(struct_v, iter);
+    decode(rotating_ver, iter);
+    decode(rotating_secrets, iter);
+  }
+
+  bool contains(const EntityName& name) const {
+    return (secrets.find(name) != secrets.end());
+  }
+
+  void clear_secrets() {
+    version = 0;
+    secrets.clear();
+    rotating_ver = 0;
+    rotating_secrets.clear();
+  }
+
+  void add_auth(const EntityName& name, EntityAuth& auth) {
+    secrets[name] = auth;
+  }
+
+  void remove_secret(const EntityName& name) {
+    auto iter = secrets.find(name);
+    if (iter == secrets.end())
+      return;
+    secrets.erase(iter);
+  }
+
+  bool get_service_secret(CephContext *cct, uint32_t service_id,
+			  CryptoKey& secret, uint64_t& secret_id,
+			  double& ttl) const;
+  bool get_service_secret(CephContext *cct, uint32_t service_id,
+			  uint64_t secret_id, CryptoKey& secret) const;
+  bool get_auth(const EntityName& name, EntityAuth& auth) const;
+  bool get_secret(const EntityName& name, CryptoKey& secret) const;
+  bool get_caps(CephContext *cct, const EntityName& name,
+		const std::string& type, AuthCapsInfo& caps) const;
+
+  std::map<EntityName, EntityAuth>::iterator secrets_begin()
+  { return secrets.begin(); }
+  std::map<EntityName, EntityAuth>::const_iterator secrets_begin() const 
+  { return secrets.begin(); }
+  std::map<EntityName, EntityAuth>::iterator secrets_end()
+  { return secrets.end(); }
+  std::map<EntityName, EntityAuth>::const_iterator secrets_end() const
+  { return secrets.end(); }
+  std::map<EntityName, EntityAuth>::iterator find_name(const EntityName& name)
+  { return secrets.find(name); }
+  std::map<EntityName, EntityAuth>::const_iterator find_name(const EntityName& name) const
+  { return secrets.find(name); }
+
+
+  // -- incremental updates --
+  typedef enum {
+    AUTH_INC_NOP,
+    AUTH_INC_ADD,
+    AUTH_INC_DEL,
+    AUTH_INC_SET_ROTATING,
+  } IncrementalOp;
+
+  struct Incremental {
+    IncrementalOp op;
+    ceph::buffer::list rotating_bl;  // if SET_ROTATING.  otherwise,
+    EntityName name;
+    EntityAuth auth;
+
+    void encode(ceph::buffer::list& bl) const {
+      using ceph::encode;
+      __u8 struct_v = 1;
+      encode(struct_v, bl);
+     __u32 _op = (__u32)op;
+      encode(_op, bl);
+      if (op == AUTH_INC_SET_ROTATING) {
+	encode(rotating_bl, bl);
+      } else {
+	encode(name, bl);
+	encode(auth, bl);
+      }
+    }
+    void decode(ceph::buffer::list::const_iterator& bl) {
+      using ceph::decode;
+      __u8 struct_v;
+      decode(struct_v, bl);
+      __u32 _op;
+      decode(_op, bl);
+      op = (IncrementalOp)_op;
+      ceph_assert(op >= AUTH_INC_NOP && op <= AUTH_INC_SET_ROTATING);
+      if (op == AUTH_INC_SET_ROTATING) {
+	decode(rotating_bl, bl);
+      } else {
+	decode(name, bl);
+	decode(auth, bl);
+      }
+    }
+  };
+
+  void apply_incremental(Incremental& inc) {
+    switch (inc.op) {
+    case AUTH_INC_ADD:
+      add_auth(inc.name, inc.auth);
+      break;
+      
+    case AUTH_INC_DEL:
+      remove_secret(inc.name);
+      break;
+
+    case AUTH_INC_SET_ROTATING:
+      decode_rotating(inc.rotating_bl);
+      break;
+
+    case AUTH_INC_NOP:
+      break;
+
+    default:
+      ceph_abort();
+    }
+  }
+
+};
+WRITE_CLASS_ENCODER(KeyServerData)
+WRITE_CLASS_ENCODER(KeyServerData::Incremental)
+
+
+
+
+class KeyServer : public KeyStore {
+  CephContext *cct;
+  KeyServerData data;
+  std::map<EntityName, CryptoKey> used_pending_keys;
+  mutable ceph::mutex lock;
+
+  int _rotate_secret(uint32_t service_id, KeyServerData &pending_data);
+  void _dump_rotating_secrets();
+  int _build_session_auth_info(uint32_t service_id, 
+			       const AuthTicket& parent_ticket,
+			       CephXSessionAuthInfo& info,
+			       double ttl);
+  bool _get_service_caps(const EntityName& name, uint32_t service_id,
+	AuthCapsInfo& caps) const;
+public:
+  KeyServer(CephContext *cct_, KeyRing *extra_secrets);
+  bool generate_secret(CryptoKey& secret);
+
+  bool get_secret(const EntityName& name, CryptoKey& secret) const override;
+  bool get_auth(const EntityName& name, EntityAuth& auth) const;
+  bool get_caps(const EntityName& name, const std::string& type, AuthCapsInfo& caps) const;
+  bool get_active_rotating_secret(const EntityName& name, CryptoKey& secret) const;
+
+  void note_used_pending_key(const EntityName& name, const CryptoKey& key);
+  void clear_used_pending_keys();
+  std::map<EntityName,CryptoKey> get_used_pending_keys();
+
+  int start_server();
+  void rotate_timeout(double timeout);
+
+  void dump();
+  
+  int build_session_auth_info(uint32_t service_id,
+			      const AuthTicket& parent_ticket,
+			      CephXSessionAuthInfo& info);
+  int build_session_auth_info(uint32_t service_id,
+			      const AuthTicket& parent_ticket,
+			      const CryptoKey& service_secret,
+			      uint64_t secret_id,
+			      CephXSessionAuthInfo& info);
+
+  /* get current secret for specific service type */
+  bool get_service_secret(uint32_t service_id, CryptoKey& secret,
+			  uint64_t& secret_id, double& ttl) const;
+  bool get_service_secret(uint32_t service_id, uint64_t secret_id,
+			  CryptoKey& secret) const override;
+
+  bool generate_secret(EntityName& name, CryptoKey& secret);
+
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    encode(data, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    std::scoped_lock l{lock};
+    using ceph::decode;
+    decode(data, bl);
+  }
+  bool contains(const EntityName& name) const;
+  int encode_secrets(ceph::Formatter *f, std::stringstream *ds) const;
+  void encode_formatted(std::string label, ceph::Formatter *f, ceph::buffer::list &bl);
+  void encode_plaintext(ceph::buffer::list &bl);
+  int list_secrets(std::stringstream& ds) const {
+    return encode_secrets(NULL, &ds);
+  }
+  version_t get_ver() const {
+    std::scoped_lock l{lock};
+    return data.version;
+  }
+
+  void clear_secrets() {
+    std::scoped_lock l{lock};
+    data.clear_secrets();
+  }
+
+  void apply_data_incremental(KeyServerData::Incremental& inc) {
+    std::scoped_lock l{lock};
+    data.apply_incremental(inc);
+  }
+  void set_ver(version_t ver) {
+    std::scoped_lock l{lock};
+    data.version = ver;
+  }
+
+  void add_auth(const EntityName& name, EntityAuth& auth) {
+    std::scoped_lock l{lock};
+    data.add_auth(name, auth);
+  }
+
+  void remove_secret(const EntityName& name) {
+    std::scoped_lock l{lock};
+    data.remove_secret(name);
+  }
+
+  bool has_secrets() {
+    auto b = data.secrets_begin();
+    return (b != data.secrets_end());
+  }
+  int get_num_secrets() {
+    std::scoped_lock l{lock};
+    return data.secrets.size();
+  }
+
+  void clone_to(KeyServerData& dst) const {
+    std::scoped_lock l{lock};
+    dst = data;
+  }
+  void export_keyring(KeyRing& keyring) {
+    std::scoped_lock l{lock};
+    for (auto p = data.secrets.begin(); p != data.secrets.end(); ++p) {
+      keyring.add(p->first, p->second);
+    }
+  }
+
+  bool prepare_rotating_update(ceph::buffer::list& rotating_bl);
+
+  bool get_rotating_encrypted(const EntityName& name, ceph::buffer::list& enc_bl) const;
+
+  ceph::mutex& get_lock() const { return lock; }
+  bool get_service_caps(const EntityName& name, uint32_t service_id,
+			AuthCapsInfo& caps) const;
+
+  std::map<EntityName, EntityAuth>::iterator secrets_begin()
+  { return data.secrets_begin(); }
+  std::map<EntityName, EntityAuth>::iterator secrets_end()
+  { return data.secrets_end(); }
+};
+WRITE_CLASS_ENCODER(KeyServer)
+
+
+#endif
diff --git a/src/auth/cephx/CephxProtocol.cc b/src/auth/cephx/CephxProtocol.cc
new file mode 100644
index 000000000..87a8b86b9
--- /dev/null
+++ b/src/auth/cephx/CephxProtocol.cc
@@ -0,0 +1,604 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2009-2011 New Dream Network
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "CephxProtocol.h"
+#include "common/Clock.h"
+#include "common/ceph_context.h"
+#include "common/config.h"
+#include "common/debug.h"
+#include "include/buffer.h"
+
+#define dout_subsys ceph_subsys_auth
+#undef dout_prefix
+#define dout_prefix *_dout << "cephx: "
+
+using std::dec;
+using std::hex;
+using std::vector;
+
+using ceph::bufferlist;
+using ceph::decode;
+using ceph::encode;
+
+void cephx_calc_client_server_challenge(CephContext *cct, CryptoKey& secret, uint64_t server_challenge, 
+		  uint64_t client_challenge, uint64_t *key, std::string &error)
+{
+  CephXChallengeBlob b;
+  b.server_challenge = server_challenge;
+  b.client_challenge = client_challenge;
+
+  bufferlist enc;
+  if (encode_encrypt(cct, b, secret, enc, error))
+    return;
+
+  uint64_t k = 0;
+  const ceph_le64 *p = (const ceph_le64 *)enc.c_str();
+  for (int pos = 0; pos + sizeof(k) <= enc.length(); pos+=sizeof(k), p++)
+    k ^= *p;
+  *key = k;
+}
+
+
+/*
+ * Authentication
+ */
+
+bool cephx_build_service_ticket_blob(CephContext *cct, CephXSessionAuthInfo& info,
+				     CephXTicketBlob& blob)
+{
+  CephXServiceTicketInfo ticket_info;
+  ticket_info.session_key = info.session_key;
+  ticket_info.ticket = info.ticket;
+  ticket_info.ticket.caps = info.ticket.caps;
+
+  ldout(cct, 10) << "build_service_ticket service "
+		 << ceph_entity_type_name(info.service_id)
+		 << " secret_id " << info.secret_id
+		 << " ticket_info.ticket.name="
+		 << ticket_info.ticket.name.to_str()
+		 << " ticket.global_id " << info.ticket.global_id << dendl;
+  blob.secret_id = info.secret_id;
+  std::string error;
+  if (!info.service_secret.get_secret().length())
+    error = "invalid key";  // Bad key?
+  else
+    encode_encrypt_enc_bl(cct, ticket_info, info.service_secret, blob.blob, error);
+  if (!error.empty()) {
+    ldout(cct, -1) << "cephx_build_service_ticket_blob failed with error "
+	  << error << dendl;
+    return false;
+  }
+  return true;
+}
+
+/*
+ * AUTH SERVER: authenticate
+ *
+ * Authenticate principal, respond with AuthServiceTicketInfo
+ *
+ * {session key, validity}^principal_secret
+ * {principal_ticket, session key}^service_secret  ... "enc_ticket"
+ */
+bool cephx_build_service_ticket_reply(CephContext *cct,
+                     CryptoKey& principal_secret,
+                     vector<CephXSessionAuthInfo> ticket_info_vec,
+                     bool should_encrypt_ticket,
+                     CryptoKey& ticket_enc_key,
+                     bufferlist& reply)
+{
+  __u8 service_ticket_reply_v = 1;
+  using ceph::encode;
+  encode(service_ticket_reply_v, reply);
+
+  uint32_t num = ticket_info_vec.size();
+  encode(num, reply);
+  ldout(cct, 10) << "build_service_ticket_reply encoding " << num
+	   << " tickets with secret " << principal_secret << dendl;
+
+  for (auto ticket_iter = ticket_info_vec.begin(); 
+       ticket_iter != ticket_info_vec.end();
+       ++ticket_iter) {
+    CephXSessionAuthInfo& info = *ticket_iter;
+    encode(info.service_id, reply);
+
+    __u8 service_ticket_v = 1;
+    encode(service_ticket_v, reply);
+
+    CephXServiceTicket msg_a;
+    msg_a.session_key = info.session_key;
+    msg_a.validity = info.validity;
+    std::string error;
+    if (encode_encrypt(cct, msg_a, principal_secret, reply, error)) {
+      ldout(cct, -1) << "error encoding encrypted: " << error << dendl;
+      return false;
+    }
+
+    bufferlist service_ticket_bl;
+    CephXTicketBlob blob;
+    if (!cephx_build_service_ticket_blob(cct, info, blob)) {
+      return false;
+    }
+    encode(blob, service_ticket_bl);
+
+    ldout(cct, 30) << "service_ticket_blob is ";
+    service_ticket_bl.hexdump(*_dout);
+    *_dout << dendl;
+
+    encode((__u8)should_encrypt_ticket, reply);
+    if (should_encrypt_ticket) {
+      if (encode_encrypt(cct, service_ticket_bl, ticket_enc_key, reply, error)) {
+	ldout(cct, -1) << "error encoding encrypted ticket: " << error << dendl;
+        return false;
+      }
+    } else {
+      encode(service_ticket_bl, reply);
+    }
+  }
+  return true;
+}
+
+/*
+ * PRINCIPAL: verify our attempt to authenticate succeeded.  fill out
+ * this ServiceTicket with the result.
+ */
+bool CephXTicketHandler::verify_service_ticket_reply(
+  CryptoKey& secret,
+  bufferlist::const_iterator& indata)
+{
+  using ceph::decode;
+  try {
+    __u8 service_ticket_v;
+    decode(service_ticket_v, indata);
+
+    CephXServiceTicket msg_a;
+    std::string error;
+    if (decode_decrypt(cct, msg_a, secret, indata, error)) {
+      ldout(cct, 0) << __func__ << " failed decode_decrypt, error is: " << error
+		    << dendl;
+      return false;
+    }
+  
+    __u8 ticket_enc;
+    decode(ticket_enc, indata);
+
+    bufferlist service_ticket_bl;
+    if (ticket_enc) {
+      ldout(cct, 10) << __func__ << " got encrypted ticket" << dendl;
+      std::string error;
+      if (decode_decrypt(cct, service_ticket_bl, session_key, indata, error)) {
+	ldout(cct, 10) << __func__ << " decode_decrypt failed "
+		       << "with " << error << dendl;
+	return false;
+      }
+    } else {
+      decode(service_ticket_bl, indata);
+    }
+    auto iter = service_ticket_bl.cbegin();
+    decode(ticket, iter);
+    ldout(cct, 10) << __func__ << " ticket.secret_id=" <<  ticket.secret_id
+		   << dendl;
+
+    ldout(cct, 10) << __func__ << " service "
+		   << ceph_entity_type_name(service_id)
+		   << " secret_id " << ticket.secret_id
+		   << " session_key " << msg_a.session_key
+		   << " validity=" << msg_a.validity << dendl;
+    session_key = msg_a.session_key;
+    if (!msg_a.validity.is_zero()) {
+      expires = ceph_clock_now();
+      expires += msg_a.validity;
+      renew_after = expires;
+      renew_after -= ((double)msg_a.validity.sec() / 4);
+      ldout(cct, 10) << __func__ << " ticket expires=" << expires
+		     << " renew_after=" << renew_after << dendl;
+    }
+
+    have_key_flag = true;
+    return true;
+  } catch (ceph::buffer::error& e) {
+    ldout(cct, 1) << __func__ << " decode error: " << e.what() << dendl;
+    return false;
+  }
+}
+
+bool CephXTicketHandler::have_key()
+{
+  if (have_key_flag) {
+    have_key_flag = ceph_clock_now() < expires;
+  }
+
+  return have_key_flag;
+}
+
+bool CephXTicketHandler::need_key() const
+{
+  if (have_key_flag) {
+    return (!expires.is_zero()) && (ceph_clock_now() >= renew_after);
+  }
+
+  return true;
+}
+
+bool CephXTicketManager::have_key(uint32_t service_id)
+{
+  auto iter = tickets_map.find(service_id);
+  if (iter == tickets_map.end())
+    return false;
+  return iter->second.have_key();
+}
+
+bool CephXTicketManager::need_key(uint32_t service_id) const
+{
+  auto iter = tickets_map.find(service_id);
+  if (iter == tickets_map.end())
+    return true;
+  return iter->second.need_key();
+}
+
+void CephXTicketManager::set_have_need_key(uint32_t service_id, uint32_t& have, uint32_t& need)
+{
+  auto iter = tickets_map.find(service_id);
+  if (iter == tickets_map.end()) {
+    have &= ~service_id;
+    need |= service_id;
+    ldout(cct, 10) << "set_have_need_key no handler for service "
+		   << ceph_entity_type_name(service_id) << dendl;
+    return;
+  }
+
+  //ldout(cct, 10) << "set_have_need_key service " << ceph_entity_type_name(service_id)
+  //<< " (" << service_id << ")"
+  //<< " need=" << iter->second.need_key() << " have=" << iter->second.have_key() << dendl;
+  if (iter->second.need_key())
+    need |= service_id;
+  else
+    need &= ~service_id;
+
+  if (iter->second.have_key())
+    have |= service_id;
+  else
+    have &= ~service_id;
+}
+
+void CephXTicketManager::invalidate_ticket(uint32_t service_id)
+{
+  auto iter = tickets_map.find(service_id);
+  if (iter != tickets_map.end())
+    iter->second.invalidate_ticket();
+}
+
+/*
+ * PRINCIPAL: verify our attempt to authenticate succeeded.  fill out
+ * this ServiceTicket with the result.
+ */
+bool CephXTicketManager::verify_service_ticket_reply(CryptoKey& secret,
+						     bufferlist::const_iterator& indata)
+{
+  __u8 service_ticket_reply_v;
+  uint32_t num = 0;
+  try {
+    decode(service_ticket_reply_v, indata);
+    decode(num, indata);
+  } catch (ceph::buffer::error& e) {
+    ldout(cct, 10) << __func__ << " failed to decode ticket v or count: "
+		   << e.what() << dendl;
+  }
+  ldout(cct, 10) << "verify_service_ticket_reply got " << num << " keys" << dendl;
+
+  for (int i=0; i<(int)num; i++) {
+    uint32_t type = 0;
+    try {
+      decode(type, indata);
+    } catch (ceph::buffer::error& e) {
+      ldout(cct, 10) << __func__ << " failed to decode ticket type: " << e.what()
+		     << dendl;
+    }
+    ldout(cct, 10) << "got key for service_id " << ceph_entity_type_name(type) << dendl;
+    CephXTicketHandler& handler = get_handler(type);
+    if (!handler.verify_service_ticket_reply(secret, indata)) {
+      return false;
+    }
+    handler.service_id = type;
+  }
+
+  return true;
+}
+
+/*
+ * PRINCIPAL: build authorizer to access the service.
+ *
+ * ticket, {timestamp}^session_key
+ */
+CephXAuthorizer *CephXTicketHandler::build_authorizer(uint64_t global_id) const
+{
+  CephXAuthorizer *a = new CephXAuthorizer(cct);
+  a->session_key = session_key;
+  cct->random()->get_bytes((char*)&a->nonce, sizeof(a->nonce));
+
+  __u8 authorizer_v = 1; // see AUTH_MODE_* in Auth.h
+  encode(authorizer_v, a->bl);
+  encode(global_id, a->bl);
+  encode(service_id, a->bl);
+
+  encode(ticket, a->bl);
+  a->base_bl = a->bl;
+
+  CephXAuthorize msg;
+  msg.nonce = a->nonce;
+
+  std::string error;
+  if (encode_encrypt(cct, msg, session_key, a->bl, error)) {
+    ldout(cct, 0) << "failed to encrypt authorizer: " << error << dendl;
+    delete a;
+    return 0;
+  }
+  return a;
+}
+
+/*
+ * PRINCIPAL: build authorizer to access the service.
+ *
+ * ticket, {timestamp}^session_key
+ */
+CephXAuthorizer *CephXTicketManager::build_authorizer(uint32_t service_id) const
+{
+  auto iter = tickets_map.find(service_id);
+  if (iter == tickets_map.end()) {
+    ldout(cct, 0) << "no TicketHandler for service "
+		  << ceph_entity_type_name(service_id) << dendl;
+    return NULL;
+  }
+
+  const CephXTicketHandler& handler = iter->second;
+  return handler.build_authorizer(global_id);
+}
+
+void CephXTicketManager::validate_tickets(uint32_t mask, uint32_t& have, uint32_t& need)
+{
+  uint32_t i;
+  need = 0;
+  for (i = 1; i<=mask; i<<=1) {
+    if (mask & i) {
+      set_have_need_key(i, have, need);
+    }
+  }
+  ldout(cct, 10) << "validate_tickets want " << mask << " have " << have
+		 << " need " << need << dendl;
+}
+
+bool cephx_decode_ticket(CephContext *cct, KeyStore *keys,
+			 uint32_t service_id,
+			 const CephXTicketBlob& ticket_blob,
+			 CephXServiceTicketInfo& ticket_info)
+{
+  uint64_t secret_id = ticket_blob.secret_id;
+  CryptoKey service_secret;
+
+  if (!ticket_blob.blob.length()) {
+    return false;
+  }
+
+  if (secret_id == (uint64_t)-1) {
+    if (!keys->get_secret(cct->_conf->name, service_secret)) {
+      ldout(cct, 0) << "ceph_decode_ticket could not get general service secret for service_id="
+	      << ceph_entity_type_name(service_id) << " secret_id=" << secret_id << dendl;
+      return false;
+    }
+  } else {
+    if (!keys->get_service_secret(service_id, secret_id, service_secret)) {
+      ldout(cct, 0) << "ceph_decode_ticket could not get service secret for service_id=" 
+	      << ceph_entity_type_name(service_id) << " secret_id=" << secret_id << dendl;
+      return false;
+    }
+  }
+
+  std::string error;
+  decode_decrypt_enc_bl(cct, ticket_info, service_secret, ticket_blob.blob, error);
+  if (!error.empty()) {
+    ldout(cct, 0) << "ceph_decode_ticket could not decrypt ticket info. error:" 
+	<< error << dendl;
+    return false;
+  }
+
+  return true;
+}
+
+/*
+ * SERVICE: verify authorizer and generate reply authorizer
+ *
+ * {timestamp + 1}^session_key
+ */
+bool cephx_verify_authorizer(CephContext *cct, const KeyStore& keys,
+			     bufferlist::const_iterator& indata,
+			     size_t connection_secret_required_len,
+			     CephXServiceTicketInfo& ticket_info,
+			     std::unique_ptr<AuthAuthorizerChallenge> *challenge,
+			     std::string *connection_secret,
+			     bufferlist *reply_bl)
+{
+  __u8 authorizer_v;
+  uint32_t service_id;
+  uint64_t global_id;
+  CryptoKey service_secret;
+  // ticket blob
+  CephXTicketBlob ticket;
+
+  try {
+    decode(authorizer_v, indata);
+    decode(global_id, indata);
+    decode(service_id, indata);
+    decode(ticket, indata);
+  } catch (ceph::buffer::end_of_buffer &e) {
+    // Unable to decode!
+    return false;
+  }
+  ldout(cct, 10) << "verify_authorizer decrypted service "
+	   << ceph_entity_type_name(service_id)
+	   << " secret_id=" << ticket.secret_id << dendl;
+
+  if (ticket.secret_id == (uint64_t)-1) {
+    EntityName name;
+    name.set_type(service_id);
+    if (!keys.get_secret(name, service_secret)) {
+      ldout(cct, 0) << "verify_authorizer could not get general service secret for service "
+	      << ceph_entity_type_name(service_id) << " secret_id=" << ticket.secret_id << dendl;
+      return false;
+    }
+  } else {
+    if (!keys.get_service_secret(service_id, ticket.secret_id, service_secret)) {
+      ldout(cct, 0) << "verify_authorizer could not get service secret for service "
+	      << ceph_entity_type_name(service_id) << " secret_id=" << ticket.secret_id << dendl;
+      if (cct->_conf->auth_debug && ticket.secret_id == 0)
+	ceph_abort_msg("got secret_id=0");
+      return false;
+    }
+  }
+  std::string error;
+  if (!service_secret.get_secret().length())
+    error = "invalid key";  // Bad key?
+  else
+    decode_decrypt_enc_bl(cct, ticket_info, service_secret, ticket.blob, error);
+  if (!error.empty()) {
+    ldout(cct, 0) << "verify_authorizer could not decrypt ticket info: error: "
+      << error << dendl;
+    return false;
+  }
+
+  if (ticket_info.ticket.global_id != global_id) {
+    ldout(cct, 0) << "verify_authorizer global_id mismatch: declared id=" << global_id
+	    << " ticket_id=" << ticket_info.ticket.global_id << dendl;
+    return false;
+  }
+
+  ldout(cct, 10) << "verify_authorizer global_id=" << global_id << dendl;
+
+  // CephXAuthorize
+  CephXAuthorize auth_msg;
+  if (decode_decrypt(cct, auth_msg, ticket_info.session_key, indata, error)) {
+    ldout(cct, 0) << "verify_authorizercould not decrypt authorize request with error: "
+      << error << dendl;
+    return false;
+  }
+
+  if (challenge) {
+    auto *c = static_cast<CephXAuthorizeChallenge*>(challenge->get());
+    if (!auth_msg.have_challenge || !c) {
+      c = new CephXAuthorizeChallenge;
+      challenge->reset(c);
+      cct->random()->get_bytes((char*)&c->server_challenge, sizeof(c->server_challenge));
+      ldout(cct,10) << __func__ << " adding server_challenge " << c->server_challenge
+		    << dendl;
+
+      encode_encrypt_enc_bl(cct, *c, ticket_info.session_key, *reply_bl, error);
+      if (!error.empty()) {
+	ldout(cct, 10) << "verify_authorizer: encode_encrypt error: " << error << dendl;
+	return false;
+      }
+      return false;
+    }
+    ldout(cct, 10) << __func__ << " got server_challenge+1 "
+		   << auth_msg.server_challenge_plus_one
+		   << " expecting " << c->server_challenge + 1 << dendl;
+    if (c->server_challenge + 1 != auth_msg.server_challenge_plus_one) {
+      return false;
+    }
+  }
+
+  /*
+   * Reply authorizer:
+   *  {timestamp + 1}^session_key
+   */
+  CephXAuthorizeReply reply;
+  // reply.trans_id = auth_msg.trans_id;
+  reply.nonce_plus_one = auth_msg.nonce + 1;
+  if (connection_secret) {
+    // generate a connection secret
+    connection_secret->resize(connection_secret_required_len);
+    if (connection_secret_required_len) {
+#ifdef WITH_SEASTAR
+      std::random_device rd;
+      std::generate_n(connection_secret->data(),
+		      connection_secret_required_len,
+		      std::default_random_engine{rd()});
+#else
+      cct->random()->get_bytes(connection_secret->data(),
+			       connection_secret_required_len);
+#endif
+    }
+    reply.connection_secret = *connection_secret;
+  }
+  if (encode_encrypt(cct, reply, ticket_info.session_key, *reply_bl, error)) {
+    ldout(cct, 10) << "verify_authorizer: encode_encrypt error: " << error << dendl;
+    return false;
+  }
+
+  ldout(cct, 10) << "verify_authorizer ok nonce " << hex << auth_msg.nonce << dec
+	   << " reply_bl.length()=" << reply_bl->length() <<  dendl;
+  return true;
+}
+
+bool CephXAuthorizer::verify_reply(bufferlist::const_iterator& indata,
+				   std::string *connection_secret)
+{
+  CephXAuthorizeReply reply;
+
+  std::string error;
+  if (decode_decrypt(cct, reply, session_key, indata, error)) {
+      ldout(cct, 0) << "verify_reply couldn't decrypt with error: " << error << dendl;
+      return false;
+  }
+
+  uint64_t expect = nonce + 1;
+  if (expect != reply.nonce_plus_one) {
+    ldout(cct, 0) << "verify_authorizer_reply bad nonce got " << reply.nonce_plus_one << " expected " << expect
+	    << " sent " << nonce << dendl;
+    return false;
+  }
+
+  if (connection_secret &&
+      reply.connection_secret.size()) {
+    *connection_secret = reply.connection_secret;
+  }
+  return true;
+}
+
+bool CephXAuthorizer::add_challenge(CephContext *cct,
+				    const bufferlist& challenge)
+{
+  bl = base_bl;
+
+  CephXAuthorize msg;
+  msg.nonce = nonce;
+
+  auto p = challenge.begin();
+  if (!p.end()) {
+    std::string error;
+    CephXAuthorizeChallenge ch;
+    decode_decrypt_enc_bl(cct, ch, session_key, challenge, error);
+    if (!error.empty()) {
+      ldout(cct, 0) << "failed to decrypt challenge (" << challenge.length() << " bytes): "
+		    << error << dendl;
+      return false;
+    }
+    msg.have_challenge = true;
+    msg.server_challenge_plus_one = ch.server_challenge + 1;
+  }
+
+  std::string error;
+  if (encode_encrypt(cct, msg, session_key, bl, error)) {
+    ldout(cct, 0) << __func__ << " failed to encrypt authorizer: " << error << dendl;
+    return false;
+  }
+  return true;
+}
diff --git a/src/auth/cephx/CephxProtocol.h b/src/auth/cephx/CephxProtocol.h
new file mode 100644
index 000000000..8a28b7306
--- /dev/null
+++ b/src/auth/cephx/CephxProtocol.h
@@ -0,0 +1,527 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_CEPHXPROTOCOL_H
+#define CEPH_CEPHXPROTOCOL_H
+
+/*
+  Ceph X protocol
+
+  See doc/dev/cephx.rst
+
+*/
+
+/* authenticate requests */
+#define CEPHX_GET_AUTH_SESSION_KEY      0x0100
+#define CEPHX_GET_PRINCIPAL_SESSION_KEY 0x0200
+#define CEPHX_GET_ROTATING_KEY          0x0400
+
+#define CEPHX_REQUEST_TYPE_MASK            0x0F00
+#define CEPHX_CRYPT_ERR			1
+
+#include "auth/Auth.h"
+#include <errno.h>
+#include <sstream>
+
+#include "include/common_fwd.h"
+/*
+ * Authentication
+ */
+
+// initial server -> client challenge
+struct CephXServerChallenge {
+  uint64_t server_challenge;
+
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    __u8 struct_v = 1;
+    encode(struct_v, bl);
+    encode(server_challenge, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    __u8 struct_v;
+    decode(struct_v, bl);
+    decode(server_challenge, bl);
+  }
+};
+WRITE_CLASS_ENCODER(CephXServerChallenge)
+
+
+// request/reply headers, for subsequent exchanges.
+
+struct CephXRequestHeader {
+  __u16 request_type;
+
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    encode(request_type, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    decode(request_type, bl);
+  }
+};
+WRITE_CLASS_ENCODER(CephXRequestHeader)
+
+struct CephXResponseHeader {
+  uint16_t request_type;
+  int32_t status;
+
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    encode(request_type, bl);
+    encode(status, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    decode(request_type, bl);
+    decode(status, bl);
+  }
+};
+WRITE_CLASS_ENCODER(CephXResponseHeader)
+
+struct CephXTicketBlob {
+  uint64_t secret_id;
+  ceph::buffer::list blob;
+
+  CephXTicketBlob() : secret_id(0) {}
+
+  void encode(ceph::buffer::list& bl) const {
+     using ceph::encode;
+     __u8 struct_v = 1;
+     encode(struct_v, bl);
+     encode(secret_id, bl);
+     encode(blob, bl);
+  }
+
+  void decode(ceph::buffer::list::const_iterator& bl) {
+     using ceph::decode;
+     __u8 struct_v;
+     decode(struct_v, bl);
+     decode(secret_id, bl);
+     decode(blob, bl);
+  }
+};
+WRITE_CLASS_ENCODER(CephXTicketBlob)
+
+// client -> server response to challenge
+struct CephXAuthenticate {
+  uint64_t client_challenge;
+  uint64_t key;
+  CephXTicketBlob old_ticket;
+  uint32_t other_keys = 0;  // replaces CephXServiceTicketRequest
+
+  bool old_ticket_may_be_omitted;
+
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    __u8 struct_v = 3;
+    encode(struct_v, bl);
+    encode(client_challenge, bl);
+    encode(key, bl);
+    encode(old_ticket, bl);
+    encode(other_keys, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    __u8 struct_v;
+    decode(struct_v, bl);
+    decode(client_challenge, bl);
+    decode(key, bl);
+    decode(old_ticket, bl);
+    if (struct_v >= 2) {
+      decode(other_keys, bl);
+    }
+
+    // v2 and v3 encodings are the same, but:
+    // - some clients that send v1 or v2 don't populate old_ticket
+    //   on reconnects (but do on renewals)
+    // - any client that sends v3 or later is expected to populate
+    //   old_ticket both on reconnects and renewals
+    old_ticket_may_be_omitted = struct_v < 3;
+  }
+};
+WRITE_CLASS_ENCODER(CephXAuthenticate)
+
+struct CephXChallengeBlob {
+  uint64_t server_challenge, client_challenge;
+  
+  void encode(ceph::buffer::list& bl) const {
+     using ceph::encode;
+    encode(server_challenge, bl);
+    encode(client_challenge, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    decode(server_challenge, bl);
+    decode(client_challenge, bl);
+  }
+};
+WRITE_CLASS_ENCODER(CephXChallengeBlob)
+
+void cephx_calc_client_server_challenge(CephContext *cct, 
+					CryptoKey& secret, uint64_t server_challenge, uint64_t client_challenge,
+					uint64_t *key, std::string &error);
+
+
+/*
+ * getting service tickets
+ */
+struct CephXSessionAuthInfo {
+  uint32_t service_id;
+  uint64_t secret_id;
+  AuthTicket ticket;
+  CryptoKey session_key;
+  CryptoKey service_secret;
+  utime_t validity;
+};
+
+
+extern bool cephx_build_service_ticket_blob(CephContext *cct,
+					    CephXSessionAuthInfo& ticket_info, CephXTicketBlob& blob);
+
+extern void cephx_build_service_ticket_request(CephContext *cct, 
+					       uint32_t keys,
+					       ceph::buffer::list& request);
+
+extern bool cephx_build_service_ticket_reply(CephContext *cct,
+					     CryptoKey& principal_secret,
+					     std::vector<CephXSessionAuthInfo> ticket_info,
+                                             bool should_encrypt_ticket,
+                                             CryptoKey& ticket_enc_key,
+					     ceph::buffer::list& reply);
+
+struct CephXServiceTicketRequest {
+  uint32_t keys;
+
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    __u8 struct_v = 1;
+    encode(struct_v, bl);
+    encode(keys, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    __u8 struct_v;
+    decode(struct_v, bl);
+    decode(keys, bl);
+  }
+};
+WRITE_CLASS_ENCODER(CephXServiceTicketRequest)
+
+
+/*
+ * Authorize
+ */
+
+struct CephXAuthorizeReply {
+  uint64_t nonce_plus_one;
+  std::string connection_secret;
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    __u8 struct_v = 1;
+    if (connection_secret.size()) {
+      struct_v = 2;
+    }
+    encode(struct_v, bl);
+    encode(nonce_plus_one, bl);
+    if (struct_v >= 2) {
+      struct_v = 2;
+      encode(connection_secret, bl);
+    }
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    __u8 struct_v;
+    decode(struct_v, bl);
+    decode(nonce_plus_one, bl);
+    if (struct_v >= 2) {
+      decode(connection_secret, bl);
+    }
+  }
+};
+WRITE_CLASS_ENCODER(CephXAuthorizeReply)
+
+
+struct CephXAuthorizer : public AuthAuthorizer {
+private:
+  CephContext *cct;
+public:
+  uint64_t nonce;
+  ceph::buffer::list base_bl;
+
+  explicit CephXAuthorizer(CephContext *cct_)
+    : AuthAuthorizer(CEPH_AUTH_CEPHX), cct(cct_), nonce(0) {}
+
+  bool build_authorizer();
+  bool verify_reply(ceph::buffer::list::const_iterator& reply,
+		    std::string *connection_secret) override;
+  bool add_challenge(CephContext *cct, const ceph::buffer::list& challenge) override;
+};
+
+
+
+/*
+ * TicketHandler
+ */
+struct CephXTicketHandler {
+  uint32_t service_id;
+  CryptoKey session_key;
+  CephXTicketBlob ticket;        // opaque to us
+  utime_t renew_after, expires;
+  bool have_key_flag;
+
+  CephXTicketHandler(CephContext *cct_, uint32_t service_id_)
+    : service_id(service_id_), have_key_flag(false), cct(cct_) { }
+
+  // to build our ServiceTicket
+  bool verify_service_ticket_reply(CryptoKey& principal_secret,
+				 ceph::buffer::list::const_iterator& indata);
+  // to access the service
+  CephXAuthorizer *build_authorizer(uint64_t global_id) const;
+
+  bool have_key();
+  bool need_key() const;
+
+  void invalidate_ticket() {
+    have_key_flag = false;
+  }
+private:
+  CephContext *cct;
+};
+
+struct CephXTicketManager {
+  typedef std::map<uint32_t, CephXTicketHandler> tickets_map_t;
+  tickets_map_t tickets_map;
+  uint64_t global_id;
+
+  explicit CephXTicketManager(CephContext *cct_) : global_id(0), cct(cct_) {}
+
+  bool verify_service_ticket_reply(CryptoKey& principal_secret,
+				 ceph::buffer::list::const_iterator& indata);
+
+  CephXTicketHandler& get_handler(uint32_t type) {
+    tickets_map_t::iterator i = tickets_map.find(type);
+    if (i != tickets_map.end())
+      return i->second;
+    CephXTicketHandler newTicketHandler(cct, type);
+    std::pair < tickets_map_t::iterator, bool > res =
+	tickets_map.insert(std::make_pair(type, newTicketHandler));
+    ceph_assert(res.second);
+    return res.first->second;
+  }
+  CephXAuthorizer *build_authorizer(uint32_t service_id) const;
+  bool have_key(uint32_t service_id);
+  bool need_key(uint32_t service_id) const;
+  void set_have_need_key(uint32_t service_id, uint32_t& have, uint32_t& need);
+  void validate_tickets(uint32_t mask, uint32_t& have, uint32_t& need);
+  void invalidate_ticket(uint32_t service_id);
+
+private:
+  CephContext *cct;
+};
+
+
+/* A */
+struct CephXServiceTicket {
+  CryptoKey session_key;
+  utime_t validity;
+
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    __u8 struct_v = 1;
+    encode(struct_v, bl);
+    encode(session_key, bl);
+    encode(validity, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    __u8 struct_v;
+    decode(struct_v, bl);
+    decode(session_key, bl);
+    decode(validity, bl);
+  }
+};
+WRITE_CLASS_ENCODER(CephXServiceTicket)
+
+/* B */
+struct CephXServiceTicketInfo {
+  AuthTicket ticket;
+  CryptoKey session_key;
+
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    __u8 struct_v = 1;
+    encode(struct_v, bl);
+    encode(ticket, bl);
+    encode(session_key, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    __u8 struct_v;
+    decode(struct_v, bl);
+    decode(ticket, bl);
+    decode(session_key, bl);
+  }
+};
+WRITE_CLASS_ENCODER(CephXServiceTicketInfo)
+
+struct CephXAuthorizeChallenge : public AuthAuthorizerChallenge {
+  uint64_t server_challenge;
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    __u8 struct_v = 1;
+    encode(struct_v, bl);
+    encode(server_challenge, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    __u8 struct_v;
+    decode(struct_v, bl);
+    decode(server_challenge, bl);
+  }
+};
+WRITE_CLASS_ENCODER(CephXAuthorizeChallenge)
+
+struct CephXAuthorize {
+  uint64_t nonce;
+  bool have_challenge = false;
+  uint64_t server_challenge_plus_one = 0;
+  void encode(ceph::buffer::list& bl) const {
+    using ceph::encode;
+    __u8 struct_v = 2;
+    encode(struct_v, bl);
+    encode(nonce, bl);
+    encode(have_challenge, bl);
+    encode(server_challenge_plus_one, bl);
+  }
+  void decode(ceph::buffer::list::const_iterator& bl) {
+    using ceph::decode;
+    __u8 struct_v;
+    decode(struct_v, bl);
+    decode(nonce, bl);
+    if (struct_v >= 2) {
+      decode(have_challenge, bl);
+      decode(server_challenge_plus_one, bl);
+    }
+  }
+};
+WRITE_CLASS_ENCODER(CephXAuthorize)
+
+/*
+ * Decode an extract ticket
+ */
+bool cephx_decode_ticket(CephContext *cct, KeyStore *keys,
+			 uint32_t service_id,
+			 const CephXTicketBlob& ticket_blob,
+			 CephXServiceTicketInfo& ticket_info);
+
+/*
+ * Verify authorizer and generate reply authorizer
+ */
+extern bool cephx_verify_authorizer(
+  CephContext *cct,
+  const KeyStore& keys,
+  ceph::buffer::list::const_iterator& indata,
+  size_t connection_secret_required_len,
+  CephXServiceTicketInfo& ticket_info,
+  std::unique_ptr<AuthAuthorizerChallenge> *challenge,
+  std::string *connection_secret,
+  ceph::buffer::list *reply_bl);
+
+
+
+
+
+
+/*
+ * encode+encrypt macros
+ */
+static constexpr uint64_t AUTH_ENC_MAGIC = 0xff009cad8826aa55ull;
+
+template <typename T>
+void decode_decrypt_enc_bl(CephContext *cct, T& t, CryptoKey key,
+			   const ceph::buffer::list& bl_enc,
+			   std::string &error)
+{
+  uint64_t magic;
+  ceph::buffer::list bl;
+
+  if (key.decrypt(cct, bl_enc, bl, &error) < 0)
+    return;
+
+  auto iter2 = bl.cbegin();
+  __u8 struct_v;
+  using ceph::decode;
+  decode(struct_v, iter2);
+  decode(magic, iter2);
+  if (magic != AUTH_ENC_MAGIC) {
+    std::ostringstream oss;
+    oss << "bad magic in decode_decrypt, " << magic << " != " << AUTH_ENC_MAGIC;
+    error = oss.str();
+    return;
+  }
+
+  decode(t, iter2);
+}
+
+template <typename T>
+void encode_encrypt_enc_bl(CephContext *cct, const T& t, const CryptoKey& key,
+			   ceph::buffer::list& out, std::string &error)
+{
+  ceph::buffer::list bl;
+  __u8 struct_v = 1;
+  using ceph::encode;
+  encode(struct_v, bl);
+  uint64_t magic = AUTH_ENC_MAGIC;
+  encode(magic, bl);
+  encode(t, bl);
+
+  key.encrypt(cct, bl, out, &error);
+}
+
+template <typename T>
+int decode_decrypt(CephContext *cct, T& t, const CryptoKey& key,
+		    ceph::buffer::list::const_iterator& iter, std::string &error)
+{
+  ceph::buffer::list bl_enc;
+  using ceph::decode;
+  try {
+    decode(bl_enc, iter);
+    decode_decrypt_enc_bl(cct, t, key, bl_enc, error);
+  }
+  catch (ceph::buffer::error &e) {
+    error = "error decoding block for decryption";
+  }
+  if (!error.empty())
+    return CEPHX_CRYPT_ERR;
+  return 0;
+}
+
+template <typename T>
+int encode_encrypt(CephContext *cct, const T& t, const CryptoKey& key,
+		    ceph::buffer::list& out, std::string &error)
+{
+  using ceph::encode;
+  ceph::buffer::list bl_enc;
+  encode_encrypt_enc_bl(cct, t, key, bl_enc, error);
+  if (!error.empty()){
+    return CEPHX_CRYPT_ERR;
+  }
+  encode(bl_enc, out);
+  return 0;
+}
+
+#endif
diff --git a/src/auth/cephx/CephxServiceHandler.cc b/src/auth/cephx/CephxServiceHandler.cc
new file mode 100644
index 000000000..977a43ad6
--- /dev/null
+++ b/src/auth/cephx/CephxServiceHandler.cc
@@ -0,0 +1,421 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+
+#include "CephxServiceHandler.h"
+#include "CephxProtocol.h"
+#include "CephxKeyServer.h"
+#include <errno.h>
+#include <sstream>
+
+#include "include/random.h"
+#include "common/config.h"
+#include "common/debug.h"
+
+#define dout_subsys ceph_subsys_auth
+#undef dout_prefix
+#define dout_prefix *_dout << "cephx server " << entity_name << ": "
+
+using std::dec;
+using std::hex;
+using std::vector;
+
+using ceph::bufferlist;
+using ceph::decode;
+using ceph::encode;
+
+int CephxServiceHandler::do_start_session(
+  bool is_new_global_id,
+  bufferlist *result_bl,
+  AuthCapsInfo *caps)
+{
+  global_id_status = is_new_global_id ? global_id_status_t::NEW_PENDING :
+					global_id_status_t::RECLAIM_PENDING;
+
+  uint64_t min = 1; // always non-zero
+  uint64_t max = std::numeric_limits<uint64_t>::max();
+  server_challenge = ceph::util::generate_random_number<uint64_t>(min, max);
+  ldout(cct, 10) << "start_session server_challenge "
+		 << hex << server_challenge << dec << dendl;
+
+  CephXServerChallenge ch;
+  ch.server_challenge = server_challenge;
+  encode(ch, *result_bl);
+  return 0;
+}
+
+int CephxServiceHandler::verify_old_ticket(
+  const CephXAuthenticate& req,
+  CephXServiceTicketInfo& old_ticket_info,
+  bool& should_enc_ticket)
+{
+  ldout(cct, 20) << " checking old_ticket: secret_id="
+		 << req.old_ticket.secret_id
+		 << " len=" << req.old_ticket.blob.length()
+		 << ", old_ticket_may_be_omitted="
+		 << req.old_ticket_may_be_omitted << dendl;
+  ceph_assert(global_id_status != global_id_status_t::NONE);
+  if (global_id_status == global_id_status_t::NEW_PENDING) {
+    // old ticket is not needed
+    if (req.old_ticket.blob.length()) {
+      ldout(cct, 0) << " superfluous ticket presented" << dendl;
+      return -EINVAL;
+    }
+    if (req.old_ticket_may_be_omitted) {
+      ldout(cct, 10) << " new global_id " << global_id
+		     << " (unexposed legacy client)" << dendl;
+      global_id_status = global_id_status_t::NEW_NOT_EXPOSED;
+    } else {
+      ldout(cct, 10) << " new global_id " << global_id << dendl;
+      global_id_status = global_id_status_t::NEW_OK;
+    }
+    return 0;
+  }
+
+  if (!req.old_ticket.blob.length()) {
+    // old ticket is needed but not presented
+    if (cct->_conf->auth_allow_insecure_global_id_reclaim &&
+	req.old_ticket_may_be_omitted) {
+      ldout(cct, 10) << " allowing reclaim of global_id " << global_id
+		     << " with no ticket presented (legacy client, auth_allow_insecure_global_id_reclaim=true)"
+		     << dendl;
+      global_id_status = global_id_status_t::RECLAIM_INSECURE;
+      return 0;
+    }
+    ldout(cct, 0) << " attempt to reclaim global_id " << global_id
+		  << " without presenting ticket" << dendl;
+    return -EACCES;
+  }
+
+  if (!cephx_decode_ticket(cct, key_server, CEPH_ENTITY_TYPE_AUTH,
+			   req.old_ticket, old_ticket_info)) {
+    if (cct->_conf->auth_allow_insecure_global_id_reclaim &&
+	req.old_ticket_may_be_omitted) {
+      ldout(cct, 10) << " allowing reclaim of global_id " << global_id
+		     << " using bad ticket (legacy client, auth_allow_insecure_global_id_reclaim=true)"
+		     << dendl;
+      global_id_status = global_id_status_t::RECLAIM_INSECURE;
+      return 0;
+    }
+    ldout(cct, 0) << " attempt to reclaim global_id " << global_id
+		  << " using bad ticket" << dendl;
+    return -EACCES;
+  }
+  ldout(cct, 20) << " decoded old_ticket: global_id="
+		 << old_ticket_info.ticket.global_id << dendl;
+  if (global_id != old_ticket_info.ticket.global_id) {
+    if (cct->_conf->auth_allow_insecure_global_id_reclaim &&
+	req.old_ticket_may_be_omitted) {
+      ldout(cct, 10) << " allowing reclaim of global_id " << global_id
+		     << " using mismatching ticket (legacy client, auth_allow_insecure_global_id_reclaim=true)"
+		     << dendl;
+      global_id_status = global_id_status_t::RECLAIM_INSECURE;
+      return 0;
+    }
+    ldout(cct, 0) << " attempt to reclaim global_id " << global_id
+		  << " using mismatching ticket" << dendl;
+    return -EACCES;
+  }
+  ldout(cct, 10) << " allowing reclaim of global_id " << global_id
+		 << " (valid ticket presented, will encrypt new ticket)"
+		 << dendl;
+  global_id_status = global_id_status_t::RECLAIM_OK;
+  should_enc_ticket = true;
+  return 0;
+}
+
+int CephxServiceHandler::handle_request(
+  bufferlist::const_iterator& indata,
+  size_t connection_secret_required_len,
+  bufferlist *result_bl,
+  AuthCapsInfo *caps,
+  CryptoKey *psession_key,
+  std::string *pconnection_secret)
+{
+  int ret = 0;
+
+  struct CephXRequestHeader cephx_header;
+  try {
+    decode(cephx_header, indata);
+  } catch (ceph::buffer::error& e) {
+    ldout(cct, 0) << __func__ << " failed to decode CephXRequestHeader: "
+		  << e.what() << dendl;
+    return -EPERM;
+  }
+
+  switch (cephx_header.request_type) {
+  case CEPHX_GET_AUTH_SESSION_KEY:
+    {
+      ldout(cct, 10) << "handle_request get_auth_session_key for "
+		     << entity_name << dendl;
+
+      CephXAuthenticate req;
+      try {
+	decode(req, indata);
+      } catch (ceph::buffer::error& e) {
+	ldout(cct, 0) << __func__ << " failed to decode CephXAuthenticate: "
+		      << e.what() << dendl;
+	ret = -EPERM;
+	break;
+      }
+
+      EntityAuth eauth;
+      if (!key_server->get_auth(entity_name, eauth)) {
+        ldout(cct, 0) << "couldn't find entity name: " << entity_name << dendl;
+	ret = -EACCES;
+	break;
+      }
+
+      if (!server_challenge) {
+	ret = -EACCES;
+	break;
+      }      
+
+      uint64_t expected_key;
+      CryptoKey *used_key = &eauth.key;
+      std::string error;
+      cephx_calc_client_server_challenge(cct, eauth.key, server_challenge,
+					 req.client_challenge, &expected_key, error);
+      if ((!error.empty() || req.key != expected_key) &&
+	  !eauth.pending_key.empty()) {
+	ldout(cct, 10) << "normal key failed for " << entity_name
+		       << ", trying pending_key" << dendl;
+	// try pending_key instead
+	error.clear();
+	cephx_calc_client_server_challenge(cct, eauth.pending_key,
+					   server_challenge,
+					   req.client_challenge, &expected_key,
+					   error);
+	if (error.empty()) {
+	  used_key = &eauth.pending_key;
+	  key_server->note_used_pending_key(entity_name, eauth.pending_key);
+	}
+      }
+      if (!error.empty()) {
+	ldout(cct, 0) << " cephx_calc_client_server_challenge error: " << error << dendl;
+	ret = -EACCES;
+	break;
+      }
+
+      ldout(cct, 20) << " checking key: req.key=" << hex << req.key
+	       << " expected_key=" << expected_key << dec << dendl;
+      if (req.key != expected_key) {
+        ldout(cct, 0) << " unexpected key: req.key=" << hex << req.key
+		<< " expected_key=" << expected_key << dec << dendl;
+        ret = -EACCES;
+	break;
+      }
+
+      CryptoKey session_key;
+      CephXSessionAuthInfo info;
+      bool should_enc_ticket = false;
+
+      CephXServiceTicketInfo old_ticket_info;
+      ret = verify_old_ticket(req, old_ticket_info, should_enc_ticket);
+      if (ret) {
+	ldout(cct, 0) << " could not verify old ticket" << dendl;
+	break;
+      }
+
+      double ttl;
+      if (!key_server->get_service_secret(CEPH_ENTITY_TYPE_AUTH,
+					  info.service_secret, info.secret_id,
+					  ttl)) {
+        ldout(cct, 0) << " could not get service secret for auth subsystem" << dendl;
+        ret = -EIO;
+        break;
+      }
+
+      info.service_id = CEPH_ENTITY_TYPE_AUTH;
+      info.ticket.name = entity_name;
+      info.ticket.global_id = global_id;
+      info.ticket.init_timestamps(ceph_clock_now(), ttl);
+      info.validity.set_from_double(ttl);
+
+      key_server->generate_secret(session_key);
+
+      info.session_key = session_key;
+      if (psession_key) {
+	*psession_key = session_key;
+      }
+
+      vector<CephXSessionAuthInfo> info_vec;
+      info_vec.push_back(info);
+
+      build_cephx_response_header(cephx_header.request_type, 0, *result_bl);
+      if (!cephx_build_service_ticket_reply(
+	    cct, *used_key, info_vec, should_enc_ticket,
+	    old_ticket_info.session_key, *result_bl)) {
+	ret = -EIO;
+	break;
+      }
+
+      if (!key_server->get_service_caps(entity_name, CEPH_ENTITY_TYPE_MON,
+					*caps)) {
+        ldout(cct, 0) << " could not get mon caps for " << entity_name << dendl;
+        ret = -EACCES;
+	break;
+      } else {
+        char *caps_str = caps->caps.c_str();
+        if (!caps_str || !caps_str[0]) {
+          ldout(cct,0) << "mon caps null for " << entity_name << dendl;
+          ret = -EACCES;
+	  break;
+        }
+
+	if (req.other_keys) {
+	  // nautilus+ client
+	  // generate a connection_secret
+	  bufferlist cbl;
+	  if (pconnection_secret) {
+	    pconnection_secret->resize(connection_secret_required_len);
+	    if (connection_secret_required_len) {
+	      cct->random()->get_bytes(pconnection_secret->data(),
+				       connection_secret_required_len);
+	    }
+	    std::string err;
+	    if (encode_encrypt(cct, *pconnection_secret, session_key, cbl,
+			       err)) {
+	      lderr(cct) << __func__ << " failed to encrypt connection secret, "
+			 << err << dendl;
+	      ret = -EACCES;
+	      break;
+	    }
+	  }
+	  encode(cbl, *result_bl);
+	  // provide requested service tickets at the same time
+	  vector<CephXSessionAuthInfo> info_vec;
+	  for (uint32_t service_id = 1; service_id <= req.other_keys;
+	       service_id <<= 1) {
+	    // skip CEPH_ENTITY_TYPE_AUTH: auth ticket is already encoded
+	    // (possibly encrypted with the old session key)
+	    if ((req.other_keys & service_id) &&
+		service_id != CEPH_ENTITY_TYPE_AUTH) {
+	      ldout(cct, 10) << " adding key for service "
+			     << ceph_entity_type_name(service_id) << dendl;
+	      CephXSessionAuthInfo svc_info;
+	      key_server->build_session_auth_info(
+		service_id,
+		info.ticket,
+		svc_info);
+	      info_vec.push_back(svc_info);
+	    }
+	  }
+	  bufferlist extra;
+	  if (!info_vec.empty()) {
+	    CryptoKey no_key;
+	    cephx_build_service_ticket_reply(
+	      cct, session_key, info_vec, false, no_key, extra);
+	  }
+	  encode(extra, *result_bl);
+	}
+
+	// caller should try to finish authentication
+	ret = 1;
+      }
+    }
+    break;
+
+  case CEPHX_GET_PRINCIPAL_SESSION_KEY:
+    {
+      ldout(cct, 10) << "handle_request get_principal_session_key" << dendl;
+
+      bufferlist tmp_bl;
+      CephXServiceTicketInfo auth_ticket_info;
+      // note: no challenge here.
+      if (!cephx_verify_authorizer(
+	    cct, *key_server, indata, 0, auth_ticket_info, nullptr,
+	    nullptr,
+	    &tmp_bl)) {
+        ret = -EACCES;
+	break;
+      }
+
+      CephXServiceTicketRequest ticket_req;
+      try {
+	decode(ticket_req, indata);
+      } catch (ceph::buffer::error& e) {
+	ldout(cct, 0) << __func__
+		      << " failed to decode CephXServiceTicketRequest: "
+		      << e.what() << dendl;
+	ret = -EPERM;
+	break;
+      }
+      ldout(cct, 10) << " ticket_req.keys = " << ticket_req.keys << dendl;
+
+      ret = 0;
+      vector<CephXSessionAuthInfo> info_vec;
+      int found_services = 0;
+      int service_err = 0;
+      for (uint32_t service_id = 1; service_id <= ticket_req.keys;
+	   service_id <<= 1) {
+        // skip CEPH_ENTITY_TYPE_AUTH: auth ticket must be obtained with
+        // CEPHX_GET_AUTH_SESSION_KEY
+        if ((ticket_req.keys & service_id) &&
+            service_id != CEPH_ENTITY_TYPE_AUTH) {
+	  ldout(cct, 10) << " adding key for service "
+			 << ceph_entity_type_name(service_id) << dendl;
+          CephXSessionAuthInfo info;
+          int r = key_server->build_session_auth_info(
+	    service_id,
+	    auth_ticket_info.ticket,  // parent ticket (client's auth ticket)
+	    info);
+	  // tolerate missing MGR rotating key for the purposes of upgrades.
+          if (r < 0) {
+	    ldout(cct, 10) << "   missing key for service "
+			   << ceph_entity_type_name(service_id) << dendl;
+	    service_err = r;
+	    continue;
+	  }
+          info_vec.push_back(info);
+	  ++found_services;
+        }
+      }
+      if (!found_services && service_err) {
+	ldout(cct, 10) << __func__ << " did not find any service keys" << dendl;
+	ret = service_err;
+      }
+      CryptoKey no_key;
+      build_cephx_response_header(cephx_header.request_type, ret, *result_bl);
+      cephx_build_service_ticket_reply(cct, auth_ticket_info.session_key,
+				       info_vec, false, no_key, *result_bl);
+    }
+    break;
+
+  case CEPHX_GET_ROTATING_KEY:
+    {
+      ldout(cct, 10) << "handle_request getting rotating secret for "
+		     << entity_name << dendl;
+      build_cephx_response_header(cephx_header.request_type, 0, *result_bl);
+      if (!key_server->get_rotating_encrypted(entity_name, *result_bl)) {
+        ret = -EACCES;
+        break;
+      }
+    }
+    break;
+
+  default:
+    ldout(cct, 10) << "handle_request unknown op " << cephx_header.request_type << dendl;
+    return -EINVAL;
+  }
+  return ret;
+}
+
+void CephxServiceHandler::build_cephx_response_header(int request_type, int status, bufferlist& bl)
+{
+  struct CephXResponseHeader header;
+  header.request_type = request_type;
+  header.status = status;
+  encode(header, bl);
+}
diff --git a/src/auth/cephx/CephxServiceHandler.h b/src/auth/cephx/CephxServiceHandler.h
new file mode 100644
index 000000000..e6e093ee4
--- /dev/null
+++ b/src/auth/cephx/CephxServiceHandler.h
@@ -0,0 +1,54 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_CEPHXSERVICEHANDLER_H
+#define CEPH_CEPHXSERVICEHANDLER_H
+
+#include "auth/AuthServiceHandler.h"
+#include "auth/Auth.h"
+
+class KeyServer;
+struct CephXAuthenticate;
+struct CephXServiceTicketInfo;
+
+class CephxServiceHandler  : public AuthServiceHandler {
+  KeyServer *key_server;
+  uint64_t server_challenge;
+
+public:
+  CephxServiceHandler(CephContext *cct_, KeyServer *ks) 
+    : AuthServiceHandler(cct_), key_server(ks), server_challenge(0) {}
+  ~CephxServiceHandler() override {}
+  
+  int handle_request(
+    ceph::buffer::list::const_iterator& indata,
+    size_t connection_secret_required_length,
+    ceph::buffer::list *result_bl,
+    AuthCapsInfo *caps,
+    CryptoKey *session_key,
+    std::string *connection_secret) override;
+
+private:
+  int do_start_session(bool is_new_global_id,
+		       ceph::buffer::list *result_bl,
+		       AuthCapsInfo *caps) override;
+
+  int verify_old_ticket(const CephXAuthenticate& req,
+			CephXServiceTicketInfo& old_ticket_info,
+			bool& should_enc_ticket);
+  void build_cephx_response_header(int request_type, int status,
+				   ceph::buffer::list& bl);
+};
+
+#endif
diff --git a/src/auth/cephx/CephxSessionHandler.cc b/src/auth/cephx/CephxSessionHandler.cc
new file mode 100644
index 000000000..6b2712568
--- /dev/null
+++ b/src/auth/cephx/CephxSessionHandler.cc
@@ -0,0 +1,194 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "CephxSessionHandler.h"
+#include "CephxProtocol.h"
+
+#include <errno.h>
+#include <sstream>
+
+#include "common/config.h"
+#include "include/ceph_features.h"
+#include "msg/Message.h"
+
+#define dout_subsys ceph_subsys_auth
+
+namespace {
+#ifdef WITH_SEASTAR
+  crimson::common::ConfigProxy& conf(CephContext*) {
+    return crimson::common::local_conf();
+  }
+#else
+  ConfigProxy& conf(CephContext* cct) {
+    return cct->_conf;
+  }
+#endif
+}
+
+int CephxSessionHandler::_calc_signature(Message *m, uint64_t *psig)
+{
+  const ceph_msg_header& header = m->get_header();
+  const ceph_msg_footer& footer = m->get_footer();
+
+  if (!HAVE_FEATURE(features, CEPHX_V2)) {
+    // legacy pre-mimic behavior for compatibility
+
+    // optimized signature calculation
+    // - avoid temporary allocated buffers from encode_encrypt[_enc_bl]
+    // - skip the leading 4 byte wrapper from encode_encrypt
+    struct {
+      __u8 v;
+      ceph_le64 magic;
+      ceph_le32 len;
+      ceph_le32 header_crc;
+      ceph_le32 front_crc;
+      ceph_le32 middle_crc;
+      ceph_le32 data_crc;
+    } __attribute__ ((packed)) sigblock = {
+      1, ceph_le64(AUTH_ENC_MAGIC), ceph_le32(4 * 4),
+      ceph_le32(header.crc), ceph_le32(footer.front_crc),
+      ceph_le32(footer.middle_crc), ceph_le32(footer.data_crc)
+    };
+
+    char exp_buf[CryptoKey::get_max_outbuf_size(sizeof(sigblock))];
+
+    try {
+      const CryptoKey::in_slice_t in {
+	sizeof(sigblock),
+	reinterpret_cast<const unsigned char*>(&sigblock)
+      };
+      const CryptoKey::out_slice_t out {
+	sizeof(exp_buf),
+	reinterpret_cast<unsigned char*>(&exp_buf)
+      };
+      key.encrypt(cct, in, out);
+    } catch (std::exception& e) {
+      lderr(cct) << __func__ << " failed to encrypt signature block" << dendl;
+      return -1;
+    }
+
+    *psig = *reinterpret_cast<ceph_le64*>(exp_buf);
+  } else {
+    // newer mimic+ signatures
+    struct {
+      ceph_le32 header_crc;
+      ceph_le32 front_crc;
+      ceph_le32 front_len;
+      ceph_le32 middle_crc;
+      ceph_le32 middle_len;
+      ceph_le32 data_crc;
+      ceph_le32 data_len;
+      ceph_le32 seq_lower_word;
+    } __attribute__ ((packed)) sigblock = {
+      ceph_le32(header.crc),
+      ceph_le32(footer.front_crc),
+      ceph_le32(header.front_len),
+      ceph_le32(footer.middle_crc),
+      ceph_le32(header.middle_len),
+      ceph_le32(footer.data_crc),
+      ceph_le32(header.data_len),
+      ceph_le32(header.seq)
+    };
+
+    char exp_buf[CryptoKey::get_max_outbuf_size(sizeof(sigblock))];
+
+    try {
+      const CryptoKey::in_slice_t in {
+	sizeof(sigblock),
+	reinterpret_cast<const unsigned char*>(&sigblock)
+      };
+      const CryptoKey::out_slice_t out {
+	sizeof(exp_buf),
+	reinterpret_cast<unsigned char*>(&exp_buf)
+      };
+      key.encrypt(cct, in, out);
+    } catch (std::exception& e) {
+      lderr(cct) << __func__ << " failed to encrypt signature block" << dendl;
+      return -1;
+    }
+
+    struct enc {
+      ceph_le64 a, b, c, d;
+    } *penc = reinterpret_cast<enc*>(exp_buf);
+    *psig = penc->a ^ penc->b ^ penc->c ^ penc->d;
+  }
+
+  ldout(cct, 10) << __func__ << " seq " << m->get_seq()
+		 << " front_crc_ = " << footer.front_crc
+		 << " middle_crc = " << footer.middle_crc
+		 << " data_crc = " << footer.data_crc
+		 << " sig = " << *psig
+		 << dendl;
+  return 0;
+}
+
+int CephxSessionHandler::sign_message(Message *m)
+{
+  // If runtime signing option is off, just return success without signing.
+  if (!conf(cct)->cephx_sign_messages) {
+    return 0;
+  }
+
+  uint64_t sig;
+  int r = _calc_signature(m, &sig);
+  if (r < 0)
+    return r;
+
+  ceph_msg_footer& f = m->get_footer();
+  f.sig = sig;
+  f.flags = (unsigned)f.flags | CEPH_MSG_FOOTER_SIGNED;
+  ldout(cct, 20) << "Putting signature in client message(seq # " << m->get_seq()
+		 << "): sig = " << sig << dendl;
+  return 0;
+}
+
+int CephxSessionHandler::check_message_signature(Message *m)
+{
+  // If runtime signing option is off, just return success without checking signature.
+  if (!conf(cct)->cephx_sign_messages) {
+    return 0;
+  }
+  if ((features & CEPH_FEATURE_MSG_AUTH) == 0) {
+    // it's fine, we didn't negotiate this feature.
+    return 0;
+  }
+
+  uint64_t sig;
+  int r = _calc_signature(m, &sig);
+  if (r < 0)
+    return r;
+
+  if (sig != m->get_footer().sig) {
+    // Should have been signed, but signature check failed.  PLR
+    if (!(m->get_footer().flags & CEPH_MSG_FOOTER_SIGNED)) {
+      ldout(cct, 0) << "SIGN: MSG " << m->get_seq() << " Sender did not set CEPH_MSG_FOOTER_SIGNED." << dendl;
+    }
+    ldout(cct, 0) << "SIGN: MSG " << m->get_seq() << " Message signature does not match contents." << dendl;
+    ldout(cct, 0) << "SIGN: MSG " << m->get_seq() << "Signature on message:" << dendl;
+    ldout(cct, 0) << "SIGN: MSG " << m->get_seq() << "    sig: " << m->get_footer().sig << dendl;
+    ldout(cct, 0) << "SIGN: MSG " << m->get_seq() << "Locally calculated signature:" << dendl;
+    ldout(cct, 0) << "SIGN: MSG " << m->get_seq() << "    sig_check:" << sig << dendl;
+
+    // For the moment, printing an error message to the log and
+    // returning failure is sufficient.  In the long term, we should
+    // probably have code parsing the log looking for this kind of
+    // security failure, particularly when there are large numbers of
+    // them, since the latter is a potential sign of an attack.  PLR
+
+    ldout(cct, 0) << "Signature failed." << dendl;
+    return (SESSION_SIGNATURE_FAILURE);
+  }
+
+  return 0;
+}
diff --git a/src/auth/cephx/CephxSessionHandler.h b/src/auth/cephx/CephxSessionHandler.h
new file mode 100644
index 000000000..e09426f1a
--- /dev/null
+++ b/src/auth/cephx/CephxSessionHandler.h
@@ -0,0 +1,44 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+
+#include "auth/AuthSessionHandler.h"
+#include "auth/Auth.h"
+#include "include/common_fwd.h"
+
+class Message;
+
+class CephxSessionHandler  : public AuthSessionHandler {
+  CephContext *cct;
+  int protocol;
+  CryptoKey key;                // per mon authentication
+  uint64_t features;
+
+  int _calc_signature(Message *m, uint64_t *psig);
+
+public:
+  CephxSessionHandler(CephContext *cct,
+		      const CryptoKey& session_key,
+		      const uint64_t features)
+    : cct(cct),
+      protocol(CEPH_AUTH_CEPHX),
+      key(session_key),
+      features(features) {
+  }
+  ~CephxSessionHandler() override = default;
+
+  int sign_message(Message *m) override;
+  int check_message_signature(Message *m) override ;
+};
+
diff --git a/src/auth/krb/KrbAuthorizeHandler.cpp b/src/auth/krb/KrbAuthorizeHandler.cpp
new file mode 100644
index 000000000..8c7523e60
--- /dev/null
+++ b/src/auth/krb/KrbAuthorizeHandler.cpp
@@ -0,0 +1,53 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (c) 2018 SUSE LLC.
+ * Author: Daniel Oliveira <doliveira@suse.com>
+ * 
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "KrbAuthorizeHandler.hpp"
+
+#include "common/debug.h"
+
+#define dout_subsys ceph_subsys_auth 
+
+bool KrbAuthorizeHandler::verify_authorizer(
+  CephContext* ceph_ctx,
+  const KeyStore& keys,
+  const bufferlist& authorizer_data,
+  size_t connection_secret_required_len,
+  bufferlist *authorizer_reply,
+  EntityName *entity_name,
+  uint64_t *global_id,
+  AuthCapsInfo *caps_info,
+  CryptoKey *session_key,
+  std::string *connection_secret,
+  std::unique_ptr<AuthAuthorizerChallenge>* challenge)
+{
+  auto itr(authorizer_data.cbegin());
+
+  try {
+    uint8_t value = (1);
+
+    using ceph::decode;
+    decode(value, itr);
+    decode(*entity_name, itr);
+    decode(*global_id, itr);
+  } catch (const buffer::error& err) {
+    ldout(ceph_ctx, 0) 
+        << "Error: KrbAuthorizeHandler::verify_authorizer() failed!" << dendl;
+    return false;
+  }
+  caps_info->allow_all = true;
+  return true;
+}
+
+
diff --git a/src/auth/krb/KrbAuthorizeHandler.hpp b/src/auth/krb/KrbAuthorizeHandler.hpp
new file mode 100644
index 000000000..448b682e6
--- /dev/null
+++ b/src/auth/krb/KrbAuthorizeHandler.hpp
@@ -0,0 +1,46 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (c) 2018 SUSE LLC.
+ * Author: Daniel Oliveira <doliveira@suse.com>
+ * 
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#ifndef KRB_AUTHORIZE_HANDLER_HPP
+#define KRB_AUTHORIZE_HANDLER_HPP
+
+#include "auth/AuthAuthorizeHandler.h"
+
+class KrbAuthorizeHandler : public AuthAuthorizeHandler {
+  bool verify_authorizer(
+    CephContext*,
+    const KeyStore&,
+    const bufferlist&,
+    size_t,
+    bufferlist *,
+    EntityName *,
+    uint64_t *,
+    AuthCapsInfo *,
+    CryptoKey *,
+    std::string *connection_secret,
+    std::unique_ptr<
+    AuthAuthorizerChallenge>* = nullptr) override;
+
+  int authorizer_session_crypto() override { 
+    return SESSION_SYMMETRIC_AUTHENTICATE; 
+  };
+
+  ~KrbAuthorizeHandler() override = default;
+
+};
+
+
+#endif    //-- KRB_AUTHORIZE_HANDLER_HPP
+
diff --git a/src/auth/krb/KrbClientHandler.cpp b/src/auth/krb/KrbClientHandler.cpp
new file mode 100644
index 000000000..1f728b4dd
--- /dev/null
+++ b/src/auth/krb/KrbClientHandler.cpp
@@ -0,0 +1,253 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (c) 2018 SUSE LLC.
+ * Author: Daniel Oliveira <doliveira@suse.com>
+ * 
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "KrbClientHandler.hpp"
+
+#include <errno.h>
+#include <string>
+#include "KrbProtocol.hpp"
+
+#include "auth/KeyRing.h"
+#include "include/random.h"
+#include "common/ceph_context.h"
+#include "common/config.h"
+#include "common/dout.h"
+
+#define dout_subsys ceph_subsys_auth
+#undef dout_prefix
+#define dout_prefix *_dout << "krb5/gssapi client request: "
+
+struct AuthAuthorizer;
+
+AuthAuthorizer* 
+KrbClientHandler::build_authorizer(uint32_t service_id) const 
+{
+  ldout(cct, 20) 
+      << "KrbClientHandler::build_authorizer(): Service: " 
+      << ceph_entity_type_name(service_id) << dendl; 
+
+  KrbAuthorizer* krb_auth = new KrbAuthorizer();
+  if (krb_auth) {
+    krb_auth->build_authorizer(cct->_conf->name, global_id);
+  }
+  return krb_auth;
+}
+
+
+KrbClientHandler::~KrbClientHandler() 
+{
+  OM_uint32 gss_minor_status(0); 
+
+  gss_release_name(&gss_minor_status, &m_gss_client_name);
+  gss_release_name(&gss_minor_status, &m_gss_service_name);
+  gss_release_cred(&gss_minor_status, &m_gss_credentials); 
+  gss_delete_sec_context(&gss_minor_status, &m_gss_sec_ctx, GSS_C_NO_BUFFER); 
+  gss_release_buffer(&gss_minor_status, 
+                     static_cast<gss_buffer_t>(&m_gss_buffer_out)); 
+}
+
+
+int KrbClientHandler::build_request(bufferlist& buff_list) const
+{
+  ldout(cct, 20) 
+      << "KrbClientHandler::build_request() " << dendl; 
+
+  KrbTokenBlob krb_token; 
+  KrbRequest krb_request; 
+
+  krb_request.m_request_type = 
+      static_cast<int>(GSSAuthenticationRequest::GSS_TOKEN);
+
+  using ceph::encode;
+  encode(krb_request, buff_list);
+
+  if (m_gss_buffer_out.length != 0) {
+    krb_token.m_token_blob.append(buffer::create_static(
+                                    m_gss_buffer_out.length, 
+                                    reinterpret_cast<char*>
+                                      (m_gss_buffer_out.value)));
+
+    encode(krb_token, buff_list);
+    ldout(cct, 20) 
+        << "KrbClientHandler::build_request() : Token Blob: " << "\n"; 
+    krb_token.m_token_blob.hexdump(*_dout);
+    *_dout << dendl;
+  }
+  return 0;
+}
+
+
+int KrbClientHandler::handle_response(
+  int ret,
+  bufferlist::const_iterator& buff_list,
+  CryptoKey *session_key,
+  std::string *connection_secret)
+{
+  auto result(ret);
+  gss_buffer_desc gss_buffer_in = {0, nullptr};
+  gss_OID_set_desc gss_mechs_wanted = {0, nullptr};
+  OM_uint32 gss_major_status(0); 
+  OM_uint32 gss_minor_status(0); 
+  OM_uint32 gss_wanted_flags(GSS_C_MUTUAL_FLAG | 
+                             GSS_C_INTEG_FLAG);
+  OM_uint32 gss_result_flags(0);
+
+  ldout(cct, 20) 
+      << "KrbClientHandler::handle_response() " << dendl; 
+
+  if (result < 0) {
+    return result;
+  }
+
+  gss_mechs_wanted.elements = const_cast<gss_OID>(&GSS_API_SPNEGO_OID_PTR);
+  gss_mechs_wanted.count = 1; 
+
+  KrbResponse krb_response; 
+
+  using ceph::decode;
+  decode(krb_response, buff_list);
+  if (m_gss_credentials == GSS_C_NO_CREDENTIAL) {
+    gss_OID krb_client_type = GSS_C_NT_USER_NAME;
+    std::string krb_client_name(cct->_conf->name.to_str());
+
+    gss_buffer_in.length = krb_client_name.length();
+    gss_buffer_in.value  = (const_cast<char*>(krb_client_name.c_str()));
+
+    if (cct->_conf->name.get_type() == CEPH_ENTITY_TYPE_CLIENT) {
+      gss_major_status = gss_import_name(&gss_minor_status, 
+                                         &gss_buffer_in, 
+                                         krb_client_type, 
+                                         &m_gss_client_name); 
+      if (gss_major_status != GSS_S_COMPLETE) {
+        auto status_str(gss_auth_show_status(gss_major_status, 
+                                             gss_minor_status));
+        ldout(cct, 0) 
+            << "ERROR: KrbClientHandler::handle_response() "
+               "[gss_import_name(gss_client_name)] failed! " 
+            << gss_major_status << " " 
+            << gss_minor_status << " " 
+            << status_str 
+            << dendl;
+      }
+    }
+
+    gss_major_status = gss_acquire_cred(&gss_minor_status, 
+                                        m_gss_client_name, 
+                                        0, 
+                                        &gss_mechs_wanted, 
+                                        GSS_C_INITIATE, 
+                                        &m_gss_credentials, 
+                                        nullptr, 
+                                        nullptr);
+    if (gss_major_status != GSS_S_COMPLETE) {
+      auto status_str(gss_auth_show_status(gss_major_status, 
+                                           gss_minor_status));
+      ldout(cct, 20) 
+          << "ERROR: KrbClientHandler::handle_response() "
+             "[gss_acquire_cred()] failed! " 
+          << gss_major_status << " " 
+          << gss_minor_status << " " 
+          << status_str 
+          << dendl;
+      return (-EACCES);
+    }
+
+    gss_buffer_desc krb_input_name_buff = {0, nullptr};
+    gss_OID krb_input_type = GSS_C_NT_HOSTBASED_SERVICE;
+    std::string gss_target_name(cct->_conf.get_val<std::string>
+                                            ("gss_target_name"));
+    krb_input_name_buff.length = gss_target_name.length(); 
+    krb_input_name_buff.value  = (const_cast<char*>(gss_target_name.c_str()));
+
+    gss_major_status = gss_import_name(&gss_minor_status, 
+                                       &krb_input_name_buff, 
+                                       krb_input_type, 
+                                       &m_gss_service_name); 
+    if (gss_major_status != GSS_S_COMPLETE) {
+      auto status_str(gss_auth_show_status(gss_major_status, 
+                                           gss_minor_status));
+      ldout(cct, 0) 
+          << "ERROR: KrbClientHandler::handle_response() "
+             "[gss_import_name(gss_service_name)] failed! " 
+          << gss_major_status << " " 
+          << gss_minor_status << " " 
+          << status_str 
+          << dendl;
+    }
+  } else {
+    KrbTokenBlob krb_token; 
+
+    using ceph::decode;
+    decode(krb_token, buff_list);
+    ldout(cct, 20) 
+        << "KrbClientHandler::handle_response() : Token Blob: " << "\n"; 
+    krb_token.m_token_blob.hexdump(*_dout);
+    *_dout << dendl;
+
+    gss_buffer_in.length = krb_token.m_token_blob.length();
+    gss_buffer_in.value  = krb_token.m_token_blob.c_str();
+  }
+
+  const gss_OID gss_mech_type = gss_mechs_wanted.elements; 
+  if (m_gss_buffer_out.length != 0) {
+    gss_release_buffer(&gss_minor_status, 
+                       static_cast<gss_buffer_t>(&m_gss_buffer_out)); 
+  }
+
+  gss_major_status = gss_init_sec_context(&gss_minor_status, 
+                                          m_gss_credentials, 
+                                          &m_gss_sec_ctx, 
+                                          m_gss_service_name, 
+                                          gss_mech_type, 
+                                          gss_wanted_flags, 
+                                          0, 
+                                          nullptr, 
+                                          &gss_buffer_in, 
+                                          nullptr, 
+                                          &m_gss_buffer_out, 
+                                          &gss_result_flags, 
+                                          nullptr);
+  switch (gss_major_status) {
+    case GSS_S_CONTINUE_NEEDED: 
+      ldout(cct, 20) 
+          << "KrbClientHandler::handle_response() : "
+             "[gss_init_sec_context(GSS_S_CONTINUE_NEEDED)] " << dendl; 
+      result = (-EAGAIN);
+      break;
+
+    case GSS_S_COMPLETE: 
+      ldout(cct, 20) 
+          << "KrbClientHandler::handle_response() : "
+             "[gss_init_sec_context(GSS_S_COMPLETE)] " << dendl; 
+      result = 0;
+      break;
+
+    default: 
+      auto status_str(gss_auth_show_status(gss_major_status, 
+                                           gss_minor_status));
+      ldout(cct, 0) 
+          << "ERROR: KrbClientHandler::handle_response() "
+             "[gss_init_sec_context()] failed! " 
+          << gss_major_status << " " 
+          << gss_minor_status << " " 
+          << status_str 
+          << dendl;
+      result = (-EACCES);
+      break;
+  }
+
+  return result;
+}
+
diff --git a/src/auth/krb/KrbClientHandler.hpp b/src/auth/krb/KrbClientHandler.hpp
new file mode 100644
index 000000000..58e531116
--- /dev/null
+++ b/src/auth/krb/KrbClientHandler.hpp
@@ -0,0 +1,84 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (c) 2018 SUSE LLC.
+ * Author: Daniel Oliveira <doliveira@suse.com>
+ * 
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#ifndef KRB_CLIENT_HANDLER_HPP
+#define KRB_CLIENT_HANDLER_HPP
+
+#include "auth/AuthClientHandler.h"
+#include "auth/RotatingKeyRing.h"
+#include "include/common_fwd.h"
+
+#include "KrbProtocol.hpp"
+
+#include <gssapi.h>
+#include <gssapi/gssapi_generic.h>
+#include <gssapi/gssapi_krb5.h>
+#include <gssapi/gssapi_ext.h>
+
+
+class Keyring;
+
+
+class KrbClientHandler : public AuthClientHandler {
+
+  public:
+    KrbClientHandler(CephContext* ceph_ctx = nullptr) 
+      : AuthClientHandler(ceph_ctx) {
+      reset();
+    }
+    ~KrbClientHandler() override;
+
+    KrbClientHandler* clone() const override {
+      return new KrbClientHandler(*this);
+    }
+
+    int get_protocol() const override { return CEPH_AUTH_GSS; }
+    void reset() override {
+      m_gss_client_name = GSS_C_NO_NAME; 
+      m_gss_service_name = GSS_C_NO_NAME; 
+      m_gss_credentials = GSS_C_NO_CREDENTIAL;
+      m_gss_sec_ctx = GSS_C_NO_CONTEXT;
+      m_gss_buffer_out = {0, 0};
+    }
+
+    void prepare_build_request() override { };
+    int build_request(bufferlist& buff_list) const override;
+    int handle_response(int ret, 
+                        bufferlist::const_iterator& buff_list,
+			CryptoKey *session_key,
+			std::string *connection_secret) override;
+
+    bool build_rotating_request(bufferlist& buff_list) const override { 
+      return false; 
+    }
+
+    AuthAuthorizer* build_authorizer(uint32_t service_id) const override;
+    bool need_tickets() override { return false; }
+    void set_global_id(uint64_t guid) override { global_id = guid; }
+
+
+  private:
+    gss_name_t m_gss_client_name; 
+    gss_name_t m_gss_service_name; 
+    gss_cred_id_t m_gss_credentials; 
+    gss_ctx_id_t m_gss_sec_ctx; 
+    gss_buffer_desc m_gss_buffer_out; 
+
+  protected:
+    void validate_tickets() override { } 
+};
+
+#endif    //-- KRB_CLIENT_HANDLER_HPP
+
diff --git a/src/auth/krb/KrbProtocol.cpp b/src/auth/krb/KrbProtocol.cpp
new file mode 100644
index 000000000..6988d3556
--- /dev/null
+++ b/src/auth/krb/KrbProtocol.cpp
@@ -0,0 +1,86 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (c) 2018 SUSE LLC.
+ * Author: Daniel Oliveira <doliveira@suse.com>
+ * 
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "KrbProtocol.hpp"
+
+#include "common/Clock.h"
+#include "common/config.h"
+#include "common/debug.h"
+#include "include/buffer.h"
+
+#define dout_subsys ceph_subsys_auth
+#undef dout_prefix
+#define dout_prefix *_dout << "krb5/gssapi protocol: "
+
+
+std::string gss_auth_show_status(const OM_uint32 gss_major_status, 
+                                 const OM_uint32 gss_minor_status)
+{
+  const std::string STR_DOT(".");
+  const std::string STR_BLANK(" ");
+
+  gss_buffer_desc gss_str_status = {0, nullptr};
+  OM_uint32 gss_maj_status(0); 
+  OM_uint32 gss_min_status(0);
+  OM_uint32 gss_ctx_message(-1);
+
+  std::string str_status("");
+
+  const auto gss_complete_status_str_format = [&](const uint32_t gss_status) {
+    if (gss_status == GSS_S_COMPLETE) {
+      std::string str_tmp("");
+      str_tmp.append(reinterpret_cast<char*>(gss_str_status.value), 
+                     gss_str_status.length);
+      str_tmp += STR_DOT;
+      if (gss_ctx_message != 0) {
+        str_tmp += STR_BLANK;
+      }
+      return str_tmp;
+    }
+    return STR_BLANK;
+  };
+
+  while (gss_ctx_message != 0) {
+    gss_maj_status = gss_display_status(&gss_min_status, 
+                                        gss_major_status, 
+                                        GSS_C_GSS_CODE, 
+                                        GSS_C_NO_OID, 
+                                        &gss_ctx_message, 
+                                        &gss_str_status); 
+    
+    if (gss_maj_status == GSS_S_COMPLETE) {
+      str_status += gss_complete_status_str_format(gss_maj_status);
+      gss_release_buffer(&gss_min_status, &gss_str_status);
+    }
+  }
+
+  if (gss_major_status == GSS_S_FAILURE) {
+    gss_ctx_message = -1;
+    while (gss_ctx_message != 0) {
+      gss_maj_status = gss_display_status(&gss_min_status, 
+                                          gss_minor_status, 
+                                          GSS_C_MECH_CODE,
+                                          const_cast<gss_OID>(&GSS_API_KRB5_OID_PTR),
+                                          &gss_ctx_message, 
+                                          &gss_str_status); 
+      if (gss_maj_status == GSS_S_COMPLETE) {
+        str_status += gss_complete_status_str_format(gss_maj_status);
+        gss_release_buffer(&gss_min_status, &gss_str_status);
+      }
+    }
+  }
+  return str_status;
+}
+
diff --git a/src/auth/krb/KrbProtocol.hpp b/src/auth/krb/KrbProtocol.hpp
new file mode 100644
index 000000000..abddb5848
--- /dev/null
+++ b/src/auth/krb/KrbProtocol.hpp
@@ -0,0 +1,160 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (c) 2018 SUSE LLC.
+ * Author: Daniel Oliveira <doliveira@suse.com>
+ * 
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#ifndef KRB_PROTOCOL_HPP
+#define KRB_PROTOCOL_HPP
+
+#include "auth/Auth.h"
+
+#include <errno.h>
+#include <gssapi.h>
+#include <gssapi/gssapi_generic.h>
+#include <gssapi/gssapi_krb5.h>
+#include <gssapi/gssapi_ext.h>
+
+#include <map>
+#include <sstream> 
+#include <string>
+
+/* 
+  Kerberos Version 5 GSS-API Mechanism
+  OID {1.2.840.113554.1.2.2}
+  RFC https://tools.ietf.org/html/rfc1964
+*/
+static const gss_OID_desc GSS_API_KRB5_OID_PTR = 
+    { 9, (void *)"\052\206\110\206\367\022\001\002\002" };
+
+/* 
+  Kerberos Version 5 GSS-API Mechanism
+  Simple and Protected GSS-API Negotiation Mechanism 
+  OID {1.3.6.1.5.5.2}
+  RFC https://tools.ietf.org/html/rfc4178
+*/
+static const gss_OID_desc GSS_API_SPNEGO_OID_PTR =
+    {6, (void *)"\x2b\x06\x01\x05\x05\x02"};
+
+static const std::string KRB_SERVICE_NAME("kerberos/gssapi");
+static const std::string GSS_API_SPNEGO_OID("{1.3.6.1.5.5.2}");
+static const std::string GSS_API_KRB5_OID("{1.2.840.113554.1.2.2}");
+
+enum class GSSAuthenticationRequest {
+  GSS_CRYPTO_ERR    = 1,
+  GSS_MUTUAL        = 0x100,
+  GSS_TOKEN         = 0x200,
+  GSS_REQUEST_MASK  = 0x0F00
+};
+
+enum class GSSKeyExchange {
+  USERAUTH_GSSAPI_RESPONSE = 70,
+  USERAUTH_GSSAPI_TOKEN,
+  USERAUTH_GSSAPI_EXCHANGE_COMPLETE,
+  USERAUTH_GSSAPI_ERROR, 
+  USERAUTH_GSSAPI_ERRTOK, 
+  USERAUTH_GSSAPI_MIC, 
+};
+static constexpr auto CEPH_GSS_OIDTYPE(0x07);
+
+struct AuthAuthorizer;
+
+
+class KrbAuthorizer : public AuthAuthorizer {
+
+  public:
+    KrbAuthorizer() : AuthAuthorizer(CEPH_AUTH_GSS) { }
+    ~KrbAuthorizer() = default; 
+    bool build_authorizer(const EntityName& entity_name, 
+                          const uint64_t guid) {
+      uint8_t value = (1);
+
+      using ceph::encode;
+      encode(value, bl, 0);
+      encode(entity_name, bl, 0); 
+      encode(guid, bl, 0);
+      return false;
+    }
+
+    bool verify_reply(bufferlist::const_iterator& buff_list,
+		      std::string *connection_secret) override {
+      return true; 
+    }
+    bool add_challenge(CephContext* ceph_ctx, 
+                       const bufferlist& buff_list) override {
+      return true; 
+    }
+};
+
+class KrbRequest {
+
+  public:
+    void decode(bufferlist::const_iterator& buff_list) {
+      using ceph::decode;
+      decode(m_request_type, buff_list);
+    }
+
+    void encode(bufferlist& buff_list) const {
+      using ceph::encode;
+      encode(m_request_type, buff_list);
+    }
+
+    uint16_t m_request_type; 
+};
+WRITE_CLASS_ENCODER(KrbRequest);
+
+class KrbResponse {
+
+  public: 
+    void decode(bufferlist::const_iterator& buff_list) {
+      using ceph::decode;
+      decode(m_response_type, buff_list); 
+    }    
+
+    void encode(bufferlist& buff_list) const {
+      using ceph::encode;
+      encode(m_response_type, buff_list); 
+    }
+
+    uint16_t m_response_type;
+};
+WRITE_CLASS_ENCODER(KrbResponse);
+
+class KrbTokenBlob {
+
+  public:
+    void decode(bufferlist::const_iterator& buff_list) {
+      uint8_t value = (0); 
+     
+      using ceph::decode; 
+      decode(value, buff_list);
+      decode(m_token_blob, buff_list);
+    }
+        
+    void encode(bufferlist& buff_list) const {
+      uint8_t value = (1); 
+      
+      using ceph::encode;
+      encode(value, buff_list, 0);
+      encode(m_token_blob, buff_list, 0);
+    }
+
+    bufferlist m_token_blob;
+};
+WRITE_CLASS_ENCODER(KrbTokenBlob);
+
+
+std::string gss_auth_show_status(const OM_uint32 gss_major_status, 
+                                 const OM_uint32 gss_minor_status); 
+
+#endif    //-- KRB_PROTOCOL_HPP
+
diff --git a/src/auth/krb/KrbServiceHandler.cpp b/src/auth/krb/KrbServiceHandler.cpp
new file mode 100644
index 000000000..c2ca3bbf2
--- /dev/null
+++ b/src/auth/krb/KrbServiceHandler.cpp
@@ -0,0 +1,225 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (c) 2018 SUSE LLC.
+ * Author: Daniel Oliveira <doliveira@suse.com>
+ * 
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "KrbServiceHandler.hpp"
+#include "KrbProtocol.hpp"
+#include <errno.h>
+#include <sstream>
+
+#include "common/config.h"
+#include "common/debug.h"
+
+#define dout_subsys ceph_subsys_auth
+#undef dout_prefix
+#define dout_prefix *_dout << "krb5/gssapi service: " << entity_name <<  " : "
+
+
+int KrbServiceHandler::handle_request(
+  bufferlist::const_iterator& indata,
+  size_t connection_secret_required_length,
+  bufferlist *buff_list,
+  AuthCapsInfo *caps,
+  CryptoKey *session_key,
+  std::string *connection_secret)
+{
+  auto result(0);
+  gss_buffer_desc gss_buffer_in = {0, nullptr};
+  gss_name_t gss_client_name = GSS_C_NO_NAME;
+  gss_OID gss_object_id = {0};
+  OM_uint32 gss_major_status(0); 
+  OM_uint32 gss_minor_status(0);
+  OM_uint32 gss_result_flags(0); 
+  std::string status_str(" ");
+
+  ldout(cct, 20) 
+      << "KrbServiceHandler::handle_request() " << dendl; 
+
+  KrbRequest krb_request; 
+  KrbTokenBlob krb_token; 
+
+  using ceph::decode;
+  decode(krb_request, indata); 
+  decode(krb_token, indata);
+
+  gss_buffer_in.length = krb_token.m_token_blob.length();
+  gss_buffer_in.value  = krb_token.m_token_blob.c_str();
+
+  ldout(cct, 20) 
+      << "KrbClientHandler::handle_request() : Token Blob: " 
+      << "\n"; 
+  krb_token.m_token_blob.hexdump(*_dout); 
+  *_dout << dendl;
+
+  if (m_gss_buffer_out.length != 0) {
+    gss_release_buffer(&gss_minor_status, 
+                       static_cast<gss_buffer_t>(&m_gss_buffer_out)); 
+  }
+
+  gss_major_status = gss_accept_sec_context(&gss_minor_status, 
+                                            &m_gss_sec_ctx, 
+                                            m_gss_credentials, 
+                                            &gss_buffer_in, 
+                                            GSS_C_NO_CHANNEL_BINDINGS, 
+                                            &gss_client_name, 
+                                            &gss_object_id, 
+                                            &m_gss_buffer_out, 
+                                            &gss_result_flags, 
+                                            nullptr, 
+                                            nullptr);
+  switch (gss_major_status) {
+    case GSS_S_CONTINUE_NEEDED: 
+      {
+        ldout(cct, 20) 
+            << "KrbServiceHandler::handle_response() : "
+               "[KrbServiceHandler(GSS_S_CONTINUE_NEEDED)] " << dendl;
+        result = 0;
+        break;
+      }
+
+    case GSS_S_COMPLETE: 
+      {
+        result = 0;
+        ldout(cct, 20) 
+            << "KrbServiceHandler::handle_response() : "
+               "[KrbServiceHandler(GSS_S_COMPLETE)] " << dendl; 
+        if (!m_key_server->get_service_caps(entity_name, 
+                                            CEPH_ENTITY_TYPE_MON, 
+                                            *caps)) {
+          result = (-EACCES);
+          ldout(cct, 0)
+              << "KrbServiceHandler::handle_response() : "
+                 "ERROR: Could not get MONITOR CAPS : " << entity_name << dendl;
+        } else {
+          if (!caps->caps.c_str()) {
+            result = (-EACCES);
+            ldout(cct, 0)
+                << "KrbServiceHandler::handle_response() : "
+                   "ERROR: MONITOR CAPS invalid : " << entity_name << dendl;
+          }
+        }
+        break;
+      }
+            
+    default: 
+      {
+        status_str = gss_auth_show_status(gss_major_status, 
+                                          gss_minor_status);
+        ldout(cct, 0) 
+            << "ERROR: KrbServiceHandler::handle_response() "
+               "[gss_accept_sec_context()] failed! " 
+            << gss_major_status << " " 
+            << gss_minor_status << " " 
+            << status_str 
+            << dendl;
+        result = (-EACCES);
+        break;
+      }
+  }
+
+  if (m_gss_buffer_out.length != 0) {
+    KrbResponse krb_response;
+    KrbTokenBlob krb_token;
+    krb_response.m_response_type = 
+        static_cast<int>(GSSAuthenticationRequest::GSS_TOKEN);
+
+    using ceph::encode;
+    encode(krb_response, *buff_list);
+
+    krb_token.m_token_blob.append(buffer::create_static(
+                                    m_gss_buffer_out.length, 
+                                    reinterpret_cast<char*>
+                                      (m_gss_buffer_out.value)));
+    encode(krb_token, *buff_list);
+    ldout(cct, 20) 
+        << "KrbServiceHandler::handle_request() : Token Blob: " << "\n"; 
+    krb_token.m_token_blob.hexdump(*_dout);
+    *_dout << dendl;
+  }
+  gss_release_name(&gss_minor_status, &gss_client_name);
+  return result;
+}
+
+int KrbServiceHandler::do_start_session(
+  bool is_new_global_id,
+  bufferlist *buff_list,
+  AuthCapsInfo *caps)
+{
+  gss_buffer_desc gss_buffer_in = {0, nullptr};
+  gss_OID gss_object_id = GSS_C_NT_HOSTBASED_SERVICE;
+  gss_OID_set gss_mechs_wanted = GSS_C_NO_OID_SET;
+  OM_uint32 gss_major_status(0); 
+  OM_uint32 gss_minor_status(0);
+  std::string gss_service_name(cct->_conf.get_val<std::string>
+                                            ("gss_target_name"));
+
+  gss_buffer_in.length = gss_service_name.length();
+  gss_buffer_in.value  = (const_cast<char*>(gss_service_name.c_str()));
+
+  gss_major_status = gss_import_name(&gss_minor_status, 
+                                     &gss_buffer_in, 
+                                     gss_object_id, 
+                                     &m_gss_service_name);
+  if (gss_major_status != GSS_S_COMPLETE) {
+    auto status_str(gss_auth_show_status(gss_major_status, 
+                                         gss_minor_status));
+    ldout(cct, 0) 
+        << "ERROR: KrbServiceHandler::start_session() "
+           "[gss_import_name(gss_client_name)] failed! " 
+        << gss_major_status << " " 
+        << gss_minor_status << " " 
+        << status_str 
+        << dendl;
+  }
+
+  gss_major_status = gss_acquire_cred(&gss_minor_status, 
+                                      m_gss_service_name, 
+                                      0, 
+                                      gss_mechs_wanted, 
+                                      GSS_C_ACCEPT, 
+                                      &m_gss_credentials, 
+                                      nullptr, 
+                                      nullptr);
+  if (gss_major_status != GSS_S_COMPLETE) {
+    auto status_str(gss_auth_show_status(gss_major_status, 
+                                         gss_minor_status));
+    ldout(cct, 0) 
+        << "ERROR: KrbServiceHandler::start_session() "
+           "[gss_acquire_cred()] failed! " 
+        << gss_major_status << " " 
+        << gss_minor_status << " " 
+        << status_str 
+        << dendl;
+    return (-EACCES);
+  } else {
+    KrbResponse krb_response;
+    krb_response.m_response_type = 
+        static_cast<int>(GSSAuthenticationRequest::GSS_MUTUAL);
+
+    using ceph::encode;
+    encode(krb_response, *buff_list);
+    return (CEPH_AUTH_GSS);
+  }
+}
+
+KrbServiceHandler::~KrbServiceHandler()
+{
+  OM_uint32 gss_minor_status(0); 
+
+  gss_release_name(&gss_minor_status, &m_gss_service_name);
+  gss_release_cred(&gss_minor_status, &m_gss_credentials); 
+  gss_delete_sec_context(&gss_minor_status, &m_gss_sec_ctx, GSS_C_NO_BUFFER); 
+  gss_release_buffer(&gss_minor_status, static_cast<gss_buffer_t>(&m_gss_buffer_out)); 
+}
+
diff --git a/src/auth/krb/KrbServiceHandler.hpp b/src/auth/krb/KrbServiceHandler.hpp
new file mode 100644
index 000000000..85cf5a1a6
--- /dev/null
+++ b/src/auth/krb/KrbServiceHandler.hpp
@@ -0,0 +1,61 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (c) 2018 SUSE LLC.
+ * Author: Daniel Oliveira <doliveira@suse.com>
+ * 
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#ifndef KRB_SERVICE_HANDLER_HPP
+#define KRB_SERVICE_HANDLER_HPP
+
+#include "auth/AuthServiceHandler.h"
+#include "auth/Auth.h"
+#include "auth/cephx/CephxKeyServer.h"
+
+#include <gssapi.h>
+#include <gssapi/gssapi_generic.h>
+#include <gssapi/gssapi_krb5.h>
+#include <gssapi/gssapi_ext.h>
+
+
+class KrbServiceHandler : public AuthServiceHandler {
+
+  public:
+    explicit KrbServiceHandler(CephContext* ceph_ctx, KeyServer* kserver) : 
+      AuthServiceHandler(ceph_ctx), 
+      m_gss_buffer_out({0, nullptr}), 
+      m_gss_credentials(GSS_C_NO_CREDENTIAL), 
+      m_gss_sec_ctx(GSS_C_NO_CONTEXT), 
+      m_gss_service_name(GSS_C_NO_NAME), 
+      m_key_server(kserver) { }
+    ~KrbServiceHandler();
+    int handle_request(bufferlist::const_iterator& indata,
+		       size_t connection_secret_required_length,
+		       bufferlist *buff_list,
+                       AuthCapsInfo *caps,
+		       CryptoKey *session_key,
+		       std::string *connection_secret) override;
+
+  private:
+    int do_start_session(bool is_new_global_id,
+			 ceph::buffer::list *buff_list,
+			 AuthCapsInfo *caps) override;
+
+    gss_buffer_desc m_gss_buffer_out;
+    gss_cred_id_t m_gss_credentials; 
+    gss_ctx_id_t m_gss_sec_ctx; 
+    gss_name_t m_gss_service_name; 
+    KeyServer* m_key_server;
+
+};
+
+#endif    //-- KRB_SERVICE_HANDLER_HPP
+
diff --git a/src/auth/krb/KrbSessionHandler.hpp b/src/auth/krb/KrbSessionHandler.hpp
new file mode 100644
index 000000000..ee80d7909
--- /dev/null
+++ b/src/auth/krb/KrbSessionHandler.hpp
@@ -0,0 +1,37 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (c) 2018 SUSE LLC.
+ * Author: Daniel Oliveira <doliveira@suse.com>
+ * 
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#ifndef KRB_SESSION_HANDLER_HPP
+#define KRB_SESSION_HANDLER_HPP
+
+#include "auth/AuthSessionHandler.h"
+#include "auth/Auth.h"
+
+#include "KrbProtocol.hpp"
+#include <errno.h>
+#include <sstream>
+
+#include "common/config.h"
+#include "include/ceph_features.h"
+#include "msg/Message.h"
+ 
+#define dout_subsys ceph_subsys_auth
+
+struct KrbSessionHandler : DummyAuthSessionHandler {
+};
+
+#endif    //-- KRB_SESSION_HANDLER_HPP
+
+
diff --git a/src/auth/none/AuthNoneAuthorizeHandler.cc b/src/auth/none/AuthNoneAuthorizeHandler.cc
new file mode 100644
index 000000000..bc553fed6
--- /dev/null
+++ b/src/auth/none/AuthNoneAuthorizeHandler.cc
@@ -0,0 +1,56 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2009-2011 New Dream Network
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "AuthNoneAuthorizeHandler.h"
+#include "common/debug.h"
+
+#define dout_subsys ceph_subsys_auth
+
+bool AuthNoneAuthorizeHandler::verify_authorizer(
+  CephContext *cct,
+  const KeyStore& keys,
+  const ceph::buffer::list& authorizer_data,
+  size_t connection_secret_required_len,
+  ceph::buffer::list *authorizer_reply,
+  EntityName *entity_name,
+  uint64_t *global_id,
+  AuthCapsInfo *caps_info,
+  CryptoKey *session_key,
+  std::string *connection_secret,
+  std::unique_ptr<AuthAuthorizerChallenge> *challenge)
+{
+  using ceph::decode;
+  auto iter = authorizer_data.cbegin();
+
+  try {
+    __u8 struct_v = 1;
+    decode(struct_v, iter);
+    decode(*entity_name, iter);
+    decode(*global_id, iter);
+  } catch (const ceph::buffer::error &err) {
+    ldout(cct, 0) << "AuthNoneAuthorizeHandle::verify_authorizer() failed to decode" << dendl;
+    return false;
+  }
+
+  caps_info->allow_all = true;
+
+  return true;
+}
+
+// Return type of crypto used for this session's data;  for none, no crypt used
+
+int AuthNoneAuthorizeHandler::authorizer_session_crypto() 
+{
+  return SESSION_CRYPTO_NONE;
+}
diff --git a/src/auth/none/AuthNoneAuthorizeHandler.h b/src/auth/none/AuthNoneAuthorizeHandler.h
new file mode 100644
index 000000000..3d5ca58d5
--- /dev/null
+++ b/src/auth/none/AuthNoneAuthorizeHandler.h
@@ -0,0 +1,39 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_AUTHNONEAUTHORIZEHANDLER_H
+#define CEPH_AUTHNONEAUTHORIZEHANDLER_H
+
+#include "auth/AuthAuthorizeHandler.h"
+#include "include/common_fwd.h"
+
+struct AuthNoneAuthorizeHandler : public AuthAuthorizeHandler {
+  bool verify_authorizer(
+    CephContext *cct,
+    const KeyStore& keys,
+    const ceph::buffer::list& authorizer_data,
+    size_t connection_secret_required_len,
+    ceph::buffer::list *authorizer_reply,
+    EntityName *entity_name,
+    uint64_t *global_id,
+    AuthCapsInfo *caps_info,
+    CryptoKey *session_key,
+    std::string *connection_secret,
+    std::unique_ptr<AuthAuthorizerChallenge> *challenge) override;
+  int authorizer_session_crypto() override;
+};
+
+
+
+#endif
diff --git a/src/auth/none/AuthNoneClientHandler.h b/src/auth/none/AuthNoneClientHandler.h
new file mode 100644
index 000000000..66b4d59fc
--- /dev/null
+++ b/src/auth/none/AuthNoneClientHandler.h
@@ -0,0 +1,61 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_AUTHNONECLIENTHANDLER_H
+#define CEPH_AUTHNONECLIENTHANDLER_H
+
+#include "auth/AuthClientHandler.h"
+#include "AuthNoneProtocol.h"
+#include "common/ceph_context.h"
+#include "common/config.h"
+
+class AuthNoneClientHandler : public AuthClientHandler {
+
+public:
+  AuthNoneClientHandler(CephContext *cct_)
+    : AuthClientHandler(cct_) {}
+
+  AuthNoneClientHandler* clone() const override {
+    return new AuthNoneClientHandler(*this);
+  }
+
+  void reset() override { }
+
+  void prepare_build_request() override {}
+  int build_request(ceph::buffer::list& bl) const override { return 0; }
+  int handle_response(int ret, ceph::buffer::list::const_iterator& iter,
+		      CryptoKey *session_key,
+		      std::string *connection_secret) override { return 0; }
+  bool build_rotating_request(ceph::buffer::list& bl) const override { return false; }
+
+  int get_protocol() const override { return CEPH_AUTH_NONE; }
+  
+  AuthAuthorizer *build_authorizer(uint32_t service_id) const override {
+    AuthNoneAuthorizer *auth = new AuthNoneAuthorizer();
+    if (auth) {
+      auth->build_authorizer(cct->_conf->name, global_id);
+    }
+    return auth;
+  }
+
+  bool need_tickets() override { return false; }
+
+  void set_global_id(uint64_t id) override {
+    global_id = id;
+  }
+private:
+  void validate_tickets() override {}
+};
+
+#endif
diff --git a/src/auth/none/AuthNoneProtocol.h b/src/auth/none/AuthNoneProtocol.h
new file mode 100644
index 000000000..d23fdcc67
--- /dev/null
+++ b/src/auth/none/AuthNoneProtocol.h
@@ -0,0 +1,38 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_AUTHNONEPROTOCOL_H
+#define CEPH_AUTHNONEPROTOCOL_H
+
+#include "auth/Auth.h"
+#include "include/common_fwd.h"
+
+struct AuthNoneAuthorizer : public AuthAuthorizer {
+  AuthNoneAuthorizer() : AuthAuthorizer(CEPH_AUTH_NONE) { }
+  bool build_authorizer(const EntityName &ename, uint64_t global_id) {
+    __u8 struct_v = 1; // see AUTH_MODE_* in Auth.h
+    using ceph::encode;
+    encode(struct_v, bl);
+    encode(ename, bl);
+    encode(global_id, bl);
+    return 0;
+  }
+  bool verify_reply(ceph::buffer::list::const_iterator& reply,
+		    std::string *connection_secret) override { return true; }
+  bool add_challenge(CephContext *cct, const ceph::buffer::list& ch) override {
+    return true;
+  }
+};
+
+#endif
diff --git a/src/auth/none/AuthNoneServiceHandler.h b/src/auth/none/AuthNoneServiceHandler.h
new file mode 100644
index 000000000..e6fcee8fc
--- /dev/null
+++ b/src/auth/none/AuthNoneServiceHandler.h
@@ -0,0 +1,46 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#ifndef CEPH_AUTHNONESERVICEHANDLER_H
+#define CEPH_AUTHNONESERVICEHANDLER_H
+
+#include "auth/AuthServiceHandler.h"
+#include "auth/Auth.h"
+#include "include/common_fwd.h"
+
+class AuthNoneServiceHandler  : public AuthServiceHandler {
+public:
+  explicit AuthNoneServiceHandler(CephContext *cct_)
+    : AuthServiceHandler(cct_) {}
+  ~AuthNoneServiceHandler() override {}
+  
+  int handle_request(ceph::buffer::list::const_iterator& indata,
+		     size_t connection_secret_required_length,
+		     ceph::buffer::list *result_bl,
+		     AuthCapsInfo *caps,
+		     CryptoKey *session_key,
+		     std::string *connection_secret) override {
+    return 0;
+  }
+
+private:
+  int do_start_session(bool is_new_global_id,
+		       ceph::buffer::list *result_bl,
+		       AuthCapsInfo *caps) override {
+    caps->allow_all = true;
+    return 1;
+  }
+};
+
+#endif
diff --git a/src/auth/none/AuthNoneSessionHandler.h b/src/auth/none/AuthNoneSessionHandler.h
new file mode 100644
index 000000000..ca1451f79
--- /dev/null
+++ b/src/auth/none/AuthNoneSessionHandler.h
@@ -0,0 +1,19 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2009 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#include "auth/AuthSessionHandler.h"
+
+struct AuthNoneSessionHandler : DummyAuthSessionHandler {
+};
+
diff --git a/src/auth/scheme.txt b/src/auth/scheme.txt
new file mode 100644
index 000000000..0df00addc
--- /dev/null
+++ b/src/auth/scheme.txt
@@ -0,0 +1,87 @@
+
+client_name = foo   (mon has some corresponding shared secret)
+client_addr = ip address, port, pid
+
+
+monitor has:
+
+client_auth {
+       client_name;
+       client capabilities;
+       client secret;
+};
+map<client_name, client_auth> users;
+
+struct secret {
+        bufferlist secret;
+        utime_t created;
+};
+map<entity_name, secret> entity_secrets;
+
+struct service_secret_set {
+       secret[3];
+};
+map<string, service_secret_set> svc_secrets;
+
+/*
+svcsecret will be a rotating key.  we regenerate every time T, and keep
+keys for 3*T.  client always get the second-newest key.  all 3 are
+considered valid.  clients and services renew/reverify key at least one
+every time T.
+*/
+
+
+client_ticket {
+       client_addr;
+       map<svc name or type, blob> client_capabilities;
+};
+
+
+
+authenticate principle:
+
+C->M : client_name, client_addr.  authenticate me.
+ ...monitor does lookup in database...
+M->C : A= {client/mon session key, validity}^clientsecret
+       B= {client ticket, validity, client/mon session key}^monsecret
+
+
+authorize principle to do something on monitor:
+
+C->M : B, {client_addr, timestamp}^client/mon session key.  do foo (assign id)
+M->C : result.  and {timestamp+1}^client/mon session key
+
+
+authorize for service:
+
+C->M : B, {client_addr, timestamp}^client/mon session key.  authorize me!
+M->C : E= {svc ticket}^svcsecret
+       F= {svc session key, validity}^client/mon session key
+
+svc ticket = (client addr, validity, svc session key)
+
+
+on opening session to service:
+
+C->O : E + {client_addr, timestamp}^svc session key
+O->C : {timestamp+1}^svc session key
+
+
+
+
+
+To authenticate:
+
+ client -> auth:
+   {client_name, client_addr}^client_secret
+ auth -> client:
+   {session key, validity, nonce}^client_secret
+   {client_ticket, session key}^service_secret  ... "enc_ticket"
+
+where client_ticket is { client_addr, created, expires, none, capabilities }.
+
+To gain access using our ticket:
+
+
+   
+