AuthMonitor: allow "fs authorize" to update caps

When "fs authorize" subcommand is executed for the second time with
different caps, the subcommand exits with error. Modify the behaviour so
that the caps passed every subsequent time is incorporated in to the caps
that are already present in the entity's keyring.

Behaviour before this commit -

$ ./bin/ceph fs authorize a client.x1 / rw
[client.x1]
	key = AQBirqxg5KHeFxAAgOm6lHMYych6OTI+y1HJKw==

$ ./bin/ceph fs authorize b client.x1 / rw
Error EINVAL: client.x1 already has fs capabilities that differ from those supplied. To generate a new auth key for client.x1, first remove client.x1 from configuration files, execute 'ceph auth rm client.x1', then execute this command again.

$ ./bin/ceph auth get client.x1
[client.x1]
	key = AQBirqxg5KHeFxAAgOm6lHMYych6OTI+y1HJKw==
	caps mds = "allow rw fsname=a"
	caps mon = "allow r fsname=a"
	caps osd = "allow rw tag cephfs data=a"
exported keyring for client.x1

After this commit -

$ ./bin/ceph fs authorize a client.x1 / rw
[client.x1]
	key = AQDvrqxgU3I3FBAAJWwF6ZtcOVeHH8TA8CwWmQ==

$ ./bin/ceph fs authorize b client.x1 / rw
updated caps for client.x1

$ ./bin/ceph auth get client.x1
[client.x1]
	key = AQDvrqxgU3I3FBAAJWwF6ZtcOVeHH8TA8CwWmQ==
	caps mds = "allow rw fsname=a, allow rw fsname=b"
	caps mon = "allow r fsname=a, allow r fsname=b"
	caps osd = "allow rw tag cephfs data=a, allow rw tag cephfs data=b"
exported keyring for client.x1

Fixes: https://tracker.ceph.com/issues/47264
Signed-off-by: Rishabh Dave <ridave@redhat.com>

MDSAuthCaps: bug fixes

Signed-off-by: Rishabh Dave <ridave@redhat.com>

9 files changed, 343 insertions, 6 deletions

diff --git a/PendingReleaseNotes b/PendingReleaseNotes
index 2ea831bc401..947f412d7eb 100644
--- a/PendingReleaseNotes
+++ b/PendingReleaseNotes
@@ -173,6 +173,11 @@
   than the number mentioned against the config tunable `mds_max_snaps_per_dir`
   so that a new snapshot can be created and retained during the next schedule
   run.
+* cephfs: Running the command "ceph fs authorize" for an existing entity now
+  upgrades the entity's capabilities instead of printing an error. It can now
+  also change read/write permissions in a capability that the entity already
+  holds. If the capability passed by user is same as one of the capabilities
+  that the entity already holds, idempotency is maintained.
 
 >=17.2.1
 
diff --git a/src/mds/MDSAuthCaps.cc b/src/mds/MDSAuthCaps.cc
index e4287072475..6d2ea07a852 100644
--- a/src/mds/MDSAuthCaps.cc
+++ b/src/mds/MDSAuthCaps.cc
@@ -362,6 +362,88 @@ bool MDSAuthCaps::parse(string_view str, ostream *err)
   }
 }
 
+bool MDSAuthCaps::merge(MDSAuthCaps newcap)
+{
+  ceph_assert(newcap.grants.size() == 1);
+  auto ng = newcap.grants[0];
+
+  for (auto& g : grants) {
+    if (g.match.fs_name == ng.match.fs_name && g.match.path == ng.match.path) {
+      if (g.spec.get_caps() == ng.spec.get_caps()) {
+	// no update required. maintaining idempotency.
+	return false;
+       } else {
+	// cap for given fs name is present, let's update it.
+	g.spec.set_caps(ng.spec.get_caps());
+	return true;
+      }
+    }
+  }
+
+  // cap for given fs name and/or path is absent, let's add a new cap for it.
+  grants.push_back(MDSCapGrant(
+    MDSCapSpec(ng.spec.get_caps()),
+    MDSCapMatch(ng.match.fs_name, ng.match.path, ng.match.root_squash),
+    {}));
+
+  return true;
+}
+
+string MDSCapMatch::to_string()
+{
+  string str = "";
+
+  if (!fs_name.empty())   { str += " fsname=" + fs_name; }
+  if (!path.empty())   { str += " path=" + path; }
+  if (root_squash)   { str += " root_squash"; }
+  if (uid != MDS_AUTH_UID_ANY) { str += " uid=" + std::to_string(uid); }
+  if (!gids.empty()) {
+    str += " gids=";
+    for (size_t i = 0; i < gids.size(); ++i) {
+      str += std::to_string(gids[i]);
+      if (i < gids.size() - 1) {
+	str += ",";
+      }
+    }
+  }
+
+  return str;
+}
+
+string MDSCapSpec::to_string()
+{
+  string str = "";
+
+  if (allow_all()) {
+    str += "*";
+  } else {
+    if (allow_read()) { str +="r"; }
+    if (allow_write()) { str +="w"; }
+    if (allow_full()) { str +="f"; }
+    if (allow_set_vxattr()) { str +="p"; }
+    if (allow_snapshot()) { str +="s"; }
+  }
+
+  return str;
+}
+
+string MDSCapGrant::to_string()
+{
+  return "allow " + spec.to_string() + match.to_string();
+}
+
+string MDSAuthCaps::to_string()
+{
+  string str = "";
+
+  for (size_t i = 0; i < grants.size(); ++i) {
+    str += grants[i].to_string();
+    if (i < grants.size() - 1)
+      str += ", ";
+  }
+
+  return str;
+}
 
 bool MDSAuthCaps::allow_all() const
 {
diff --git a/src/mds/MDSAuthCaps.h b/src/mds/MDSAuthCaps.h
index 6cfdf489f9a..1571ab56105 100644
--- a/src/mds/MDSAuthCaps.h
+++ b/src/mds/MDSAuthCaps.h
@@ -93,6 +93,17 @@ struct MDSCapSpec {
   bool allow_full() const {
     return (caps & FULL);
   }
+
+  unsigned get_caps() {
+    return caps;
+  }
+
+  void set_caps(unsigned int _caps) {
+    caps = _caps;
+  }
+
+  std::string to_string();
+
 private:
   unsigned caps = 0;
 };
@@ -135,6 +146,7 @@ struct MDSCapMatch {
    * @param target_path filesystem path without leading '/'
    */
   bool match_path(std::string_view target_path) const;
+  std::string to_string();
 
   // Require UID to be equal to this, if !=MDS_AUTH_UID_ANY
   int64_t uid = MDS_AUTH_UID_ANY;
@@ -156,6 +168,7 @@ struct MDSCapGrant {
   MDSCapGrant() {}
 
   void parse_network();
+  std::string to_string();
 
   MDSCapSpec spec;
   MDSCapMatch match;
@@ -181,6 +194,7 @@ public:
 
   void set_allow_all();
   bool parse(std::string_view str, std::ostream *err);
+  bool merge(MDSAuthCaps newcap);
 
   bool allow_all() const;
   bool is_capable(std::string_view inode_path,
@@ -211,7 +225,9 @@ public:
     return false;
   }
 
+
   friend std::ostream &operator<<(std::ostream &out, const MDSAuthCaps &cap);
+  std::string to_string();
 private:
   std::vector<MDSCapGrant> grants;
 };
diff --git a/src/mon/AuthMonitor.cc b/src/mon/AuthMonitor.cc
index 2fa2d78d613..6bc88619792 100644
--- a/src/mon/AuthMonitor.cc
+++ b/src/mon/AuthMonitor.cc
@@ -1789,9 +1789,9 @@ bool AuthMonitor::prepare_command(MonOpRequestRef op)
 
     EntityAuth entity_auth;
     if (mon.key_server.get_auth(entity, entity_auth)) {
-      for (const auto &sys_cap : encoded_caps) {
-	if (entity_auth.caps.count(sys_cap.first) == 0 ||
-	    !entity_auth.caps[sys_cap.first].contents_equal(sys_cap.second)) {
+      for (const auto& [cap_entity, cap] : encoded_caps) {
+	if (entity_auth.caps.count(cap_entity) == 0 ||
+	    !entity_auth.caps[cap_entity].contents_equal(cap)) {
 	  ss << entity << " already has fs capabilities that differ from "
 	     << "those supplied. To generate a new auth key for " << entity
 	     << ", first remove " << entity << " from configuration files, "
@@ -1802,9 +1802,28 @@ bool AuthMonitor::prepare_command(MonOpRequestRef op)
 	}
       }
 
-      _encode_key(entity, entity_auth, rdata, f.get(), false, &encoded_caps);
-      err = 0;
-      goto done;
+      int rv = _gen_wanted_caps(entity_auth, newcaps, ss);
+      ceph_assert(rv == CAPS_UPDATE_REQD or rv == CAPS_UPDATE_NOT_REQD or
+	          rv == CAPS_PARSING_ERR);
+      if (rv == CAPS_PARSING_ERR) {
+	goto done;
+      } else if (rv == CAPS_UPDATE_NOT_REQD) {
+	ss << "no update for caps of " << entity;
+	err = 0;
+	goto done;
+      }
+
+      dout(20) << "caps that will be enforced -" << dendl;
+      for (const auto& it : newcaps) {
+	dout(20) << it.first << " cap = \"" << it.second << "\"" << dendl;
+      }
+
+      err = _update_caps(entity, newcaps, op, ds, &rdata, f.get());
+      if (err == 0) {
+	return true;
+      } else {
+	goto done;
+      }
     }
 
     err = _create_entity(entity, newcaps, op, ds, &rdata, f.get());
@@ -1842,6 +1861,75 @@ done:
   return false;
 }
 
+template<typename CAP_ENTITY_CLASS>
+AuthMonitor::caps_update AuthMonitor::_merge_caps(const string& cap_entity,
+  const string& new_cap_str, const string& cur_cap_str,
+  map<string, string>& newcaps, ostream& out)
+{
+  CAP_ENTITY_CLASS cur_cap, new_cap;
+
+  if (not cur_cap.parse(cur_cap_str, &out)) {
+    out << "error parsing " << cap_entity << "caps client already holds";
+    return CAPS_PARSING_ERR;
+  }
+  if (not new_cap.parse(new_cap_str, &out)) {
+    out << "error parsing new " << cap_entity << "caps";
+    return CAPS_PARSING_ERR;
+  }
+
+  if (cur_cap.merge(new_cap)) {
+    newcaps[cap_entity] = cur_cap.to_string();
+    return CAPS_UPDATE_REQD;
+  } else {
+    newcaps[cap_entity] = cur_cap_str;
+    return CAPS_UPDATE_NOT_REQD;
+  }
+}
+
+/* Generate the caps that should be present in the entity's auth keyring
+ * after running the "fs authorize" command. This is done by merging the
+ * caps already present in the client's auth keyring with the new caps
+ * provided by the user at "fs authorize" command.
+ */
+AuthMonitor::caps_update AuthMonitor::_gen_wanted_caps(EntityAuth& e_auth,
+  map<string, string>& newcaps, ostream& out)
+{
+  caps_update is_caps_update_reqd = CAPS_UPDATE_NOT_REQD;
+
+  if (e_auth.caps.empty()) {
+    return CAPS_UPDATE_REQD;
+  }
+
+  // new_cap_str is the new cap to be added to the current cap
+  for (const auto& [cap_entity, new_cap_str] : newcaps) {
+    string cur_cap_str; // current cap held by entity's auth keyring
+
+    if (e_auth.caps.count(cap_entity) == 0) {
+      is_caps_update_reqd = CAPS_UPDATE_REQD;
+      continue;
+    }
+
+    auto iter = e_auth.caps[cap_entity].cbegin();
+    decode(cur_cap_str, iter);
+    if (cur_cap_str == new_cap_str) {
+      continue;
+    }
+
+    if (cap_entity == "mon") {
+      is_caps_update_reqd = _merge_caps<MonCap>(cap_entity, new_cap_str,
+	cur_cap_str, newcaps, out);
+    } else if (cap_entity == "osd") {
+      is_caps_update_reqd = _merge_caps<OSDCap>(cap_entity, new_cap_str,
+	cur_cap_str, newcaps, out);
+    } else if (cap_entity == "mds") {
+      is_caps_update_reqd = _merge_caps<MDSAuthCaps>(cap_entity, new_cap_str,
+	cur_cap_str, newcaps, out);
+    }
+  }
+
+  return is_caps_update_reqd;
+}
+
 void AuthMonitor::_encode_keyring(KeyRing& kr, const EntityName& entity,
   bufferlist& rdata, Formatter* fmtr, map<string, bufferlist>* caps)
 {
diff --git a/src/mon/AuthMonitor.h b/src/mon/AuthMonitor.h
index 51fba994977..b48ab914e9b 100644
--- a/src/mon/AuthMonitor.h
+++ b/src/mon/AuthMonitor.h
@@ -31,7 +31,12 @@ class Monitor;
 #define MIN_GLOBAL_ID 0x1000
 
 class AuthMonitor : public PaxosService {
+
 public:
+  typedef enum {
+    CAPS_UPDATE_NOT_REQD, CAPS_UPDATE_REQD, CAPS_PARSING_ERR
+  } caps_update;
+
   enum IncType {
     GLOBAL_ID,
     AUTH_DATA,
@@ -195,6 +200,13 @@ private:
     const std::map<std::string, std::string>& caps, MonOpRequestRef op,
     std::stringstream& ds, bufferlist* rdata, Formatter* fmtr);
 
+  caps_update _gen_wanted_caps(EntityAuth& e_auth,
+    std::map<std::string, std::string>& newcaps, std::ostream& out);
+  template<typename CAP_ENTITY_CLASS>
+  caps_update _merge_caps(const std::string& cap_entity,
+    const std::string& new_cap_str, const std::string& cur_cap_str,
+    std::map<std::string, std::string>& newcaps, std::ostream& out);
+
   bool check_rotate();
   void process_used_pending_keys(const std::map<EntityName,CryptoKey>& keys);
 
diff --git a/src/mon/MonCap.cc b/src/mon/MonCap.cc
index a232db9efc9..0d941f2c282 100644
--- a/src/mon/MonCap.cc
+++ b/src/mon/MonCap.cc
@@ -690,3 +690,62 @@ bool MonCap::parse(const string& str, ostream *err)
   return false; 
 }
 
+bool MonCap::merge(MonCap newcap)
+{
+  ceph_assert(newcap.grants.size() == 1);
+  auto& ng = newcap.grants[0];
+
+  for (auto& g : grants) {
+    /* TODO: check case where cap is "allow rw *". */
+
+    if (g.fs_name == ng.fs_name) {
+      if (g.allow == ng.allow) {
+	// no update required; maintain idempotency.
+	return false;
+      } else {
+	// cap for given fs name is present, let's update it.
+	g.allow = ng.allow;
+	return true;
+      }
+    }
+  }
+
+  // cap for given fs name is absent, let's add a new cap for it.
+  grants.push_back(MonCapGrant(ng.allow, ng.fs_name));
+  return true;
+}
+
+string MonCapGrant::to_string()
+{
+  string str = "allow ";
+
+  if (allow & MON_CAP_R) {
+      str+= "r";
+  } else if (allow & MON_CAP_W) {
+      str+= "w";
+  } else if (allow & MON_CAP_X) {
+      str+= "x";
+  } else if (allow == MON_CAP_ANY) {
+      str+= "*";
+  }
+
+  if (not fs_name.empty()) {
+    str += " fsname=" + fs_name;
+  }
+
+  return str;
+}
+
+string MonCap::to_string()
+{
+  string str;
+
+  for (size_t i = 0; i < grants.size(); ++i) {
+    str += grants[i].to_string();
+    if (i < grants.size () - 1) {
+      str += ", ";
+    }
+  }
+
+  return str;
+}
diff --git a/src/mon/MonCap.h b/src/mon/MonCap.h
index 570f788ad59..45063dbed9e 100644
--- a/src/mon/MonCap.h
+++ b/src/mon/MonCap.h
@@ -137,6 +137,8 @@ struct MonCapGrant {
       command.length() == 0 &&
       fs_name.empty();
   }
+
+  std::string to_string();
 };
 
 std::ostream& operator<<(std::ostream& out, const MonCapGrant& g);
@@ -151,10 +153,12 @@ struct MonCap {
   std::string get_str() const {
     return text;
   }
+  std::string to_string();
 
   bool is_allow_all() const;
   void set_allow_all();
   bool parse(const std::string& str, std::ostream *err=NULL);
+  bool merge(MonCap newcap);
 
   /**
    * check if we are capable of something
diff --git a/src/osd/OSDCap.cc b/src/osd/OSDCap.cc
index 677981f8015..6c30cd84938 100644
--- a/src/osd/OSDCap.cc
+++ b/src/osd/OSDCap.cc
@@ -539,3 +539,71 @@ bool OSDCap::parse(const string& str, ostream *err)
 
   return false; 
 }
+
+bool OSDCap::merge(OSDCap newcap)
+{
+  ceph_assert(newcap.grants.size() == 1);
+  auto ng = newcap.grants[0];
+
+  for (auto& g : grants) {
+    /* TODO: check case where cap is "allow rw tag cephfs *". */
+
+    if (g.match.pool_tag.application == ng.match.pool_tag.application and
+	g.match.pool_tag.key == ng.match.pool_tag.key and
+	g.match.pool_tag.value == ng.match.pool_tag.value) {
+      if (g.spec.allow == ng.spec.allow) {
+	// no update required, maintaining idempotency.
+	return false;
+      } else if (g.spec.allow != ng.spec.allow) {
+	// cap for given application is present, let's update it.
+	g.spec.allow = ng.spec.allow;
+	return true;
+      }
+    }
+  }
+
+  // cap for given application is absent, let's add a new cap for it.
+  grants.push_back(OSDCapGrant(
+    OSDCapMatch(OSDCapPoolTag(ng.match.pool_tag.application,
+      ng.match.pool_tag.key, ng.match.pool_tag.value)),
+    OSDCapSpec(ng.spec.allow)));
+  return true;
+}
+
+string OSDCapGrant::to_string() {
+  string str = "allow ";
+
+  if (spec.allow & OSD_CAP_R)
+    str += "r";
+  if (spec.allow & OSD_CAP_W)
+    str += "w";
+
+  if ((spec.allow & OSD_CAP_X) == OSD_CAP_X)
+    str += "x";
+  else {
+    if (spec.allow & OSD_CAP_CLS_R)
+      str += " class-read";
+    else if (spec.allow & OSD_CAP_CLS_W)
+      str += " class-write";
+  }
+
+  if (not (match.pool_tag.application.empty() and match.pool_tag.key.empty()
+	   and match.pool_tag.value.empty()))
+    str += " tag " + match.pool_tag.application + " " + \
+	   match.pool_tag.key + "=" + match.pool_tag.value;
+
+  return str;
+}
+
+string OSDCap::to_string()
+{
+  string str;
+
+  for (size_t i = 0; i < grants.size(); ++i) {
+    str += grants[i].to_string();
+    if (i < grants.size() - 1)
+      str += ", ";
+  }
+
+  return str;
+}
diff --git a/src/osd/OSDCap.h b/src/osd/OSDCap.h
index 394b1a72635..caf6cd788d7 100644
--- a/src/osd/OSDCap.h
+++ b/src/osd/OSDCap.h
@@ -216,6 +216,7 @@ struct OSDCapGrant {
                   std::vector<bool>* class_allowed) const;
 
   void expand_profile();
+  std::string to_string();
 };
 
 ostream& operator<<(ostream& out, const OSDCapGrant& g);
@@ -230,6 +231,8 @@ struct OSDCap {
   bool allow_all() const;
   void set_allow_all();
   bool parse(const std::string& str, ostream *err=NULL);
+  bool merge(OSDCap newcap);
+  std::string to_string();
 
   /**
    * check if we are capable of something