OSD/auth caps: Add OSD auth caps based on pool tag

Extend the OSD auth caps syntax to include RADOS pool tags. New syntax:

allow rw tag <application> <key>=<value>

Access is granted if the pool contains the <key>:<value> in its
application metadata.

Feature: http://tracker.ceph.com/issues/21084
Signed-off-by: Douglas Fuller <dfuller@redhat.com>

doc/man/8/ceph-authtool.rst

@@ -139,7 +139,8 @@ In general, an osd capability follows the grammar::
 
         osdcap  := grant[,grant...]
         grant   := allow (match capspec | capspec match)
-        match   := [pool[=]<poolname> | object_prefix <prefix>]
+        match   := [pool[=]<poolname> | object_prefix <prefix>
+                    | tag <application-name> <key>=<value> ]
         capspec := * | [r][w][x] [class-read] [class-write]
 
 The capspec determines what kind of operations the entity can perform::

doc/rados/operations/user-management.rst

@@ -92,9 +92,10 @@ Authorization (Capabilities)
 
 Ceph uses the term "capabilities" (caps) to describe authorizing an
 authenticated user to exercise the functionality of the monitors, OSDs and
-metadata servers. Capabilities can also restrict access to data within a pool or
-a namespace within a pool. A Ceph administrative user sets a user's
-capabilities when creating or updating a user.
+metadata servers. Capabilities can also restrict access to data within a pool,
+a namespace within a pool, or a set of pools based on their application tags.
+A Ceph administrative user sets a user's capabilities when creating or updating
+a user.
 
 Capability syntax follows the form::
 
@@ -110,7 +111,7 @@ Capability syntax follows the form::
   ``class-write`` access settings or ``profile {name}``. Additionally, OSD
   capabilities also allow for pool and namespace settings. ::
 
-	osd 'allow {access} [pool={pool-name} [namespace={namespace-name}]]'
+	osd 'allow {access} [pool={pool-name} [namespace={namespace-name}]] [tag {application} {key}={value}]'
 	osd 'profile {name} [pool={pool-name} [namespace={namespace-name}]]'
 
 - **Metadata Server Caps:** For administrators, use ``allow *``.  For all
@@ -216,6 +217,12 @@ OpenStack, a typical deployment would have pools for volumes, images, backups
 and virtual machines, and users such as ``client.glance``, ``client.cinder``,
 etc.
 
+Application Tags
+----------------
+
+Access may be restricted to specific pools as defined by their application
+metadata. The ``*`` wildcard may be used for the ``key`` argument, the
+``value`` argument, or both.
 
 Namespace
 ---------

src/osd/OSDCap.cc

@@ -70,12 +70,21 @@ ostream& operator<<(ostream& out, const OSDCapPoolNamespace& pns)
   return out;
 }
 
+ostream& operator<<(ostream &out, const OSDCapPoolTag &pt)
+{
+  out << "app " << pt.application << " key " << pt.key << " val " << pt.value
+      << " ";
+  return out;
+}
+
 ostream& operator<<(ostream& out, const OSDCapMatch& m)
 {
   if (m.auid != -1LL) {
     out << "auid " << m.auid << " ";
-  } else {
+  } else if (!m.pool_namespace.pool_name.empty() || m.pool_namespace.nspace) {
     out << m.pool_namespace;
+  } else if (!m.pool_tag.application.empty()) {
+    out << m.pool_tag;
   }
 
   if (m.object_prefix.length()) {
@@ -116,14 +125,52 @@ bool OSDCapPoolNamespace::is_match_all() const
   return true;
 }
 
+bool OSDCapPoolTag::is_match(const app_map_t& app_map) const
+{
+  if (application.empty()) {
+    return true;
+  }
+  auto kv_map = app_map.find(application);
+  if (kv_map == app_map.end()) {
+    return false;
+  }
+  if (!key.compare("*") && !value.compare("*")) {
+    return true;
+  }
+  if (!key.compare("*")) {
+    for (auto it : kv_map->second) {
+      if (it.second == value) {
+	return true;
+      }
+    }
+    return false;
+  }
+  auto kv_val = kv_map->second.find(key);
+  if (kv_val == kv_map->second.end()) {
+    return false;
+  }
+  if (!value.compare("*")) {
+    return true;
+  }
+  return kv_val->second == value;
+}
+
+bool OSDCapPoolTag::is_match_all() const {
+  return application.empty();
+}
+
 bool OSDCapMatch::is_match(const string& pn, const string& ns,
-                           int64_t pool_auid, const string& object) const
+                           int64_t pool_auid,
+			   const OSDCapPoolTag::app_map_t& app_map,
+			   const string& object) const
 {
   if (auid >= 0) {
     if (auid != pool_auid)
       return false;
   } else if (!pool_namespace.is_match(pn, ns)) {
     return false;
+  } else if (!pool_tag.is_match(app_map)) {
+    return false;
   }
 
   if (object_prefix.length()) {
@@ -139,6 +186,8 @@ bool OSDCapMatch::is_match_all() const
     return false;
   } else if (!pool_namespace.is_match_all()) {
     return false;
+  } else if (!pool_tag.is_match_all()) {
+    return false;
   }
 
   if (object_prefix.length()) {
@@ -180,7 +229,9 @@ bool OSDCapGrant::allow_all() const
 }
 
 bool OSDCapGrant::is_capable(const string& pool_name, const string& ns,
-                             int64_t pool_auid, const string& object,
+                             int64_t pool_auid,
+			     const OSDCapPoolTag::app_map_t& application_metadata,
+			     const string& object,
                              bool op_may_read, bool op_may_write,
                              const std::vector<OpRequest::ClassInfo>& classes,
                              std::vector<bool>* class_allowed) const
@@ -189,11 +240,14 @@ bool OSDCapGrant::is_capable(const string& pool_name, const string& ns,
   if (profile.is_valid()) {
     return std::any_of(profile_grants.cbegin(), profile_grants.cend(),
                        [&](const OSDCapGrant& grant) {
-        return grant.is_capable(pool_name, ns, pool_auid, object, op_may_read,
-                                op_may_write, classes, class_allowed);
-      });
+			   return grant.is_capable(pool_name, ns, pool_auid,
+						   application_metadata,
+						   object, op_may_read,
+						   op_may_write, classes,
+						   class_allowed);
+		       });
   } else {
-    if (match.is_match(pool_name, ns, pool_auid, object)) {
+    if (match.is_match(pool_name, ns, pool_auid, application_metadata, object)) {
       allow = allow | spec.allow;
       if ((op_may_read && !(allow & OSD_CAP_R)) ||
           (op_may_write && !(allow & OSD_CAP_W))) {
@@ -283,13 +337,15 @@ void OSDCap::set_allow_all()
 }
 
 bool OSDCap::is_capable(const string& pool_name, const string& ns,
-                        int64_t pool_auid, const string& object,
+                        int64_t pool_auid,
+			const OSDCapPoolTag::app_map_t& application_metadata,
+			const string& object,
                         bool op_may_read, bool op_may_write,
 			const std::vector<OpRequest::ClassInfo>& classes) const
 {
   std::vector<bool> class_allowed(classes.size(), false);
   for (auto &grant : grants) {
-    if (grant.is_capable(pool_name, ns, pool_auid, object, op_may_read,
+    if (grant.is_capable(pool_name, ns, pool_auid, application_metadata, object, op_may_read,
                          op_may_write, classes, &class_allowed)) {
       return true;
     }
@@ -337,8 +393,13 @@ struct OSDCapParser : qi::grammar<Iterator, OSDCap()>
     // match := [pool[=]<poolname> [namespace[=]<namespace>] | auid <123>] [object_prefix <prefix>]
     auid %= (spaces >> lit("auid") >> spaces >> int_);
     object_prefix %= -(spaces >> lit("object_prefix") >> spaces >> str);
+    pooltag %= (spaces >> lit("tag")
+		>> spaces >> str // application
+		>> spaces >> (str | char_('*')) // key
+		>> -spaces >> lit('=') >> -spaces >> (str | char_('*'))); // value
 
     match = (
+      pooltag                                 [_val = phoenix::construct<OSDCapMatch>(_1)] |
       (auid >> object_prefix)                 [_val = phoenix::construct<OSDCapMatch>(_1, _2)] |
       (pool_name >> nspace >> object_prefix)  [_val = phoenix::construct<OSDCapMatch>(_1, _2, _3)] |
       (pool_name >> object_prefix)            [_val = phoenix::construct<OSDCapMatch>(_1, _2)]);
@@ -390,6 +451,7 @@ struct OSDCapParser : qi::grammar<Iterator, OSDCap()>
   qi::rule<Iterator, string()> pool_name;
   qi::rule<Iterator, string()> nspace;
   qi::rule<Iterator, string()> object_prefix;
+  qi::rule<Iterator, OSDCapPoolTag()> pooltag;
   qi::rule<Iterator, OSDCapMatch()> match;
   qi::rule<Iterator, string()> profile_name;
   qi::rule<Iterator, OSDCapProfile()> profile;

src/osd/OSDCap.h

@@ -36,6 +36,7 @@ using std::ostream;
 #include <list>
 #include <vector>
 #include <boost/optional.hpp>
+#include <boost/fusion/include/adapt_struct.hpp>
 
 static const __u8 OSD_CAP_R     = (1 << 1);      // read
 static const __u8 OSD_CAP_W     = (1 << 2);      // write
@@ -95,14 +96,38 @@ struct OSDCapPoolNamespace {
 
 ostream& operator<<(ostream& out, const OSDCapPoolNamespace& pns);
 
+struct OSDCapPoolTag {
+  typedef std::map<std::string, std::map<std::string, std::string> > app_map_t;
+  std::string application;
+  std::string key;
+  std::string value;
+
+  OSDCapPoolTag () {}
+  OSDCapPoolTag(const std::string& application, const std::string& key,
+		const std::string& value) :
+    application(application), key(key), value(value) {}
+
+  bool is_match(const app_map_t& app_map) const;
+  bool is_match_all() const;
+};
+// adapt for parsing with boost::spirit::qi in OSDCapParser
+BOOST_FUSION_ADAPT_STRUCT(OSDCapPoolTag,
+			  (std::string, application)
+			  (std::string, key)
+			  (std::string, value))
+
+ostream& operator<<(ostream& out, const OSDCapPoolTag& pt);
 
 struct OSDCapMatch {
+  typedef std::map<std::string, std::map<std::string, std::string> > app_map_t;
   // auid and pool_name/nspace are mutually exclusive
   int64_t auid = CEPH_AUTH_UID_DEFAULT;
   OSDCapPoolNamespace pool_namespace;
+  OSDCapPoolTag pool_tag;
   std::string object_prefix;
 
   OSDCapMatch() {}
+  OSDCapMatch(const OSDCapPoolTag& pt) : pool_tag(pt) {}
   OSDCapMatch(const OSDCapPoolNamespace& pns) : pool_namespace(pns) {}
   OSDCapMatch(const std::string& pl, const std::string& pre)
     : pool_namespace(pl), object_prefix(pre) {}
@@ -111,6 +136,9 @@ struct OSDCapMatch {
     : pool_namespace(pl, ns), object_prefix(pre) {}
   OSDCapMatch(uint64_t auid, const std::string& pre)
     : auid(auid), object_prefix(pre) {}
+  OSDCapMatch(const std::string& dummy, const std::string& app,
+	      const std::string& key, const std::string& val)
+    : pool_tag(app, key, val) {}
 
   /**
    * check if given request parameters match our constraints
@@ -122,7 +150,8 @@ struct OSDCapMatch {
    * @return true if we match, false otherwise
    */
   bool is_match(const std::string& pool_name, const std::string& nspace_name,
-                int64_t pool_auid, const std::string& object) const;
+                int64_t pool_auid, const app_map_t& app_map,
+		const std::string& object) const;
   bool is_match_all() const;
 };
 
@@ -165,6 +194,7 @@ struct OSDCapGrant {
 
   bool allow_all() const;
   bool is_capable(const string& pool_name, const string& ns, int64_t pool_auid,
+		  const OSDCapPoolTag::app_map_t& application_metadata,
                   const string& object, bool op_may_read, bool op_may_write,
                   const std::vector<OpRequest::ClassInfo>& classes,
                   std::vector<bool>* class_allowed) const;
@@ -202,6 +232,7 @@ struct OSDCap {
    * @return true if the operation is allowed, false otherwise
    */
   bool is_capable(const string& pool_name, const string& ns, int64_t pool_auid,
+		  const OSDCapPoolTag::app_map_t& application_metadata,
 		  const string& object, bool op_may_read, bool op_may_write,
 		  const std::vector<OpRequest::ClassInfo>& classes) const;
 };

src/osd/PG.cc

@@ -1881,7 +1881,9 @@ bool PG::op_has_sufficient_caps(OpRequestRef& op)
     req->get_hobj().get_key();
 
   bool cap = caps.is_capable(pool.name, req->get_hobj().nspace,
-                             pool.auid, key,
+                             pool.auid,
+			     pool.info.application_metadata,
+			     key,
 			     op->need_read_cap(),
 			     op->need_write_cap(),
 			     op->classes());
@@ -1891,6 +1893,7 @@ bool PG::op_has_sufficient_caps(OpRequestRef& op)
            << " pool=" << pool.id << " (" << pool.name
            << " " << req->get_hobj().nspace
 	   << ") owner=" << pool.auid
+	   << " pool_app_metadata=" << pool.info.application_metadata
 	   << " need_read_cap=" << op->need_read_cap()
 	   << " need_write_cap=" << op->need_write_cap()
 	   << " classes=" << op->classes()

src/vstart.sh

@@ -584,6 +584,12 @@ start_mon() {
 			--cap mgr 'allow *' \
 			"$keyring_fn"
 
+		prun $SUDO "$CEPH_BIN/ceph-authtool" --gen-key --name=client.fs --set-uid=0 \
+		   --cap mon 'allow r' \
+			--cap osd 'allow rw tag cephfs data=*' \
+			--cap mds 'allow rwp' \
+			"$keyring_fn"
+
 		prun $SUDO "$CEPH_BIN/ceph-authtool" --gen-key --name=client.rgw \
 		    --cap mon 'allow rw' \
 		    --cap osd 'allow rwx' \
@@ -731,7 +737,7 @@ EOF
 EOF
 	        fi
 	        prun $SUDO "$CEPH_BIN/ceph-authtool" --create-keyring --gen-key --name="mds.$name" "$key_fn"
-	        ceph_adm -i "$key_fn" auth add "mds.$name" mon 'allow profile mds' osd 'allow *' mds 'allow' mgr 'allow profile mds'
+	        ceph_adm -i "$key_fn" auth add "mds.$name" mon 'allow profile mds' osd 'allow rw tag cephfs *=*' mds 'allow' mgr 'allow profile mds'
 	        if [ "$standby" -eq 1 ]; then
 			    prun $SUDO "$CEPH_BIN/ceph-authtool" --create-keyring --gen-key --name="mds.${name}s" \
 				     "$CEPH_DEV_DIR/mds.${name}s/keyring"