auth: add functions_resource to resources

This commit adds "functions" resource to our authorization
resources. The implementation strives to be compatible
with Cassandra both from CQL level and serialization,
i.e. so that entries in system_auth.role_permissions table
will be identical if CassandraAuthorizer is used.

This commit adds a way of representing these resources
in-memory, but they are not enforced as permissions yet.

The following permissions are supported:
```
CREATE ALL FUNCTIONS
CREATE ALL FUNCTIONS IN KEYSPACE <ks>

ALTER ALL FUNCTIONS
ALTER ALL FUNCTIONS IN KEYSPACE <ks>
ALTER FUNCTION <f>

DROP ALL FUNCTIONS
DROP ALL FUNCTIONS IN KEYSPACE <ks>
DROP FUNCTION <f>

AUTHORIZE ALL FUNCTIONS
AUTHORIZE ALL FUNCTIONS IN KEYSPACE <ks>
AUTHORIZE FUNCTION <f>

EXECUTE ALL FUNCTIONS
EXECUTE ALL FUNCTIONS IN KEYSPACE <ks>
EXECUTE FUNCTION <f>
```
as per
https://cassandra.apache.org/doc/latest/cassandra/cql/security.html#cql-permissions

auth/permission.cc

@@ -21,7 +21,8 @@ const auth::permission_set auth::permissions::ALL = auth::permission_set::of<
         auth::permission::SELECT,
         auth::permission::MODIFY,
         auth::permission::AUTHORIZE,
-        auth::permission::DESCRIBE>();
+        auth::permission::DESCRIBE,
+        auth::permission::EXECUTE>();
 
 const auth::permission_set auth::permissions::NONE;
 
@@ -34,7 +35,8 @@ static const std::unordered_map<sstring, auth::permission> permission_names({
         {"SELECT", auth::permission::SELECT},
         {"MODIFY", auth::permission::MODIFY},
         {"AUTHORIZE", auth::permission::AUTHORIZE},
-        {"DESCRIBE", auth::permission::DESCRIBE}});
+        {"DESCRIBE", auth::permission::DESCRIBE},
+        {"EXECUTE", auth::permission::EXECUTE}});
 
 const sstring& auth::permissions::to_string(permission p) {
     for (auto& v : permission_names) {

auth/permission.hh

@@ -38,6 +38,8 @@ enum class permission {
     AUTHORIZE, // required for GRANT and REVOKE.
     DESCRIBE, // required on the root-level role resource to list all roles.
 
+    // function/aggregate/procedure calls
+    EXECUTE,
 };
 
 typedef enum_set<
@@ -51,7 +53,8 @@ typedef enum_set<
                 permission::SELECT,
                 permission::MODIFY,
                 permission::AUTHORIZE,
-                permission::DESCRIBE>> permission_set;
+                permission::DESCRIBE,
+                permission::EXECUTE>> permission_set;
 
 bool operator<(const permission_set&, const permission_set&);
 

auth/resource.cc

@@ -16,8 +16,11 @@
 
 #include <boost/algorithm/string/join.hpp>
 #include <boost/algorithm/string/split.hpp>
+#include <boost/algorithm/string/classification.hpp>
 
 #include "service/storage_proxy.hh"
+#include "data_dictionary/user_types_metadata.hh"
+#include "cql3/util.hh"
 
 namespace auth {
 
@@ -26,6 +29,7 @@ std::ostream& operator<<(std::ostream& os, resource_kind kind) {
         case resource_kind::data: os << "data"; break;
         case resource_kind::role: os << "role"; break;
         case resource_kind::service_level: os << "service_level"; break;
+        case resource_kind::functions: os << "functions"; break;
     }
 
     return os;
@@ -34,12 +38,14 @@ std::ostream& operator<<(std::ostream& os, resource_kind kind) {
 static const std::unordered_map<resource_kind, std::string_view> roots{
         {resource_kind::data, "data"},
         {resource_kind::role, "roles"},
-        {resource_kind::service_level, "service_levels"}};
+        {resource_kind::service_level, "service_levels"},
+        {resource_kind::functions, "functions"}};
 
 static const std::unordered_map<resource_kind, std::size_t> max_parts{
         {resource_kind::data, 2},
         {resource_kind::role, 1},
-        {resource_kind::service_level, 0}};
+        {resource_kind::service_level, 0},
+        {resource_kind::functions, 2}};
 
 static permission_set applicable_permissions(const data_resource_view& dv) {
     if (dv.table()) {
@@ -82,6 +88,15 @@ static permission_set applicable_permissions(const service_level_resource_view &
             permission::AUTHORIZE>();
 }
 
+static permission_set applicable_permissions(const functions_resource_view& fv) {
+    return permission_set::of<
+            permission::CREATE,
+            permission::ALTER,
+            permission::DROP,
+            permission::AUTHORIZE,
+            permission::EXECUTE>();
+}
+
 resource::resource(resource_kind kind) : _kind(kind) {
     _parts.emplace_back(roots.at(kind));
 }
@@ -106,6 +121,39 @@ resource::resource(role_resource_t, std::string_view role) : resource(resource_k
 resource::resource(service_level_resource_t): resource(resource_kind::service_level) {
 }
 
+resource::resource(functions_resource_t) : resource(resource_kind::functions) {
+}
+
+resource::resource(functions_resource_t, std::string_view keyspace) : resource(resource_kind::functions) {
+    _parts.emplace_back(keyspace);
+}
+
+resource::resource(functions_resource_t, std::string_view keyspace, std::string_view function_name) : resource(resource_kind::functions) {
+    _parts.emplace_back(keyspace);
+    _parts.emplace_back(function_name);
+}
+
+resource::resource(functions_resource_t, std::string_view keyspace, std::string_view function_name, std::vector<sstring> function_signature) : resource(resource_kind::functions) {
+    _parts.emplace_back(keyspace);
+    sstring encoded_signature = format("{}[{}]",
+            function_name,
+            ::join("^", function_signature));
+    _parts.emplace_back(encoded_signature);
+}
+
+resource make_functions_resource(std::string_view keyspace, std::string_view function_name, std::vector<::shared_ptr<cql3::cql3_type::raw>> function_signature) {
+    if (keyspace.empty()) {
+        throw exceptions::invalid_request_exception("In this context function name must be explictly qualified by a keyspace");
+    }
+    std::vector<sstring> args_types;
+    for (auto& raw_type : function_signature) {
+        // FIXME(sarna): provide information on user-defined types - this is tricky, because this information
+        // is kept in a database instance, and is thus dynamic
+        args_types.emplace_back(raw_type->prepare_internal(sstring(keyspace), data_dictionary::user_types_metadata{}).get_type()->name());
+    }
+    return resource(functions_resource_t{}, keyspace, function_name, std::move(args_types));
+}
+
 sstring resource::name() const {
     return boost::algorithm::join(_parts, "/");
 }
@@ -127,6 +175,7 @@ permission_set resource::applicable_permissions() const {
         case resource_kind::data: ps = ::auth::applicable_permissions(data_resource_view(*this)); break;
         case resource_kind::role: ps = ::auth::applicable_permissions(role_resource_view(*this)); break;
         case resource_kind::service_level: ps = ::auth::applicable_permissions(service_level_resource_view(*this)); break;
+        case resource_kind::functions: ps = ::auth::applicable_permissions(functions_resource_view(*this)); break;
     }
 
     return ps;
@@ -149,6 +198,7 @@ std::ostream& operator<<(std::ostream& os, const resource& r) {
         case resource_kind::data: return os << data_resource_view(r);
         case resource_kind::role: return os << role_resource_view(r);
         case resource_kind::service_level: return os << service_level_resource_view(r);
+        case resource_kind::functions: return os << functions_resource_view(r);
     }
 
     return os;
@@ -165,6 +215,75 @@ std::ostream &operator<<(std::ostream &os, const service_level_resource_view &v)
     return os;
 }
 
+sstring encode_signature(std::string_view name, std::vector<data_type> args) {
+    return format("{}[{}]", name,
+            ::join("^", args | boost::adaptors::transformed([] (const data_type t) {
+                return t->name();
+            })));
+}
+
+std::pair<sstring, std::vector<data_type>> decode_signature(std::string_view encoded_signature) {
+    auto name_delim = encoded_signature.find_last_of('[');
+    std::string_view function_name = encoded_signature.substr(0, name_delim);
+    encoded_signature.remove_prefix(name_delim + 1);
+    encoded_signature.remove_suffix(1);
+    std::vector<std::string_view> raw_types;
+    boost::split(raw_types, encoded_signature, boost::is_any_of("^"));
+    std::vector<data_type> decoded_types = boost::copy_range<std::vector<data_type>>(
+        raw_types | boost::adaptors::transformed([] (std::string_view raw_type) {
+            return abstract_type::parse_type(sstring(raw_type));
+        })
+    );
+    return {sstring(function_name), decoded_types};
+}
+
+// Purely for Cassandra compatibility, types in the function signature are
+// decoded from their verbose form (org.apache.cassandra.db.marshal.Int32Type)
+// to the short form (int)
+static sstring decoded_signature_string(std::string_view encoded_signature) {
+    auto [function_name, arg_types] = decode_signature(encoded_signature);
+    return format("{}({})", cql3::util::maybe_quote(sstring(function_name)),
+            boost::algorithm::join(arg_types | boost::adaptors::transformed([] (data_type t) {
+                return t->cql3_type_name();
+            }), ", "));
+}
+
+std::ostream &operator<<(std::ostream &os, const functions_resource_view &v) {
+    const auto keyspace = v.keyspace();
+    const auto function_signature = v.function_signature();
+
+    if (!keyspace) {
+        os << "<all functions>";
+    } else if (!function_signature) {
+        os << "<all functions in " << *keyspace << '>';
+    } else {
+        os << "<function " << *keyspace << '.' << decoded_signature_string(*function_signature) << '>';
+    }
+    return os;
+}
+
+functions_resource_view::functions_resource_view(const resource& r) : _resource(r) {
+    if (r._kind != resource_kind::functions) {
+        throw resource_kind_mismatch(resource_kind::functions, r._kind);
+    }
+}
+
+std::optional<std::string_view> functions_resource_view::keyspace() const {
+    if (_resource._parts.size() == 1) {
+        return {};
+    }
+
+    return _resource._parts[1];
+}
+
+std::optional<std::string_view> functions_resource_view::function_signature() const {
+    if (_resource._parts.size() <= 2) {
+        return {};
+    }
+
+    return _resource._parts[2];
+}
+
 data_resource_view::data_resource_view(const resource& r) : _resource(r) {
     if (r._kind != resource_kind::data) {
         throw resource_kind_mismatch(resource_kind::data, r._kind);

auth/resource.hh

@@ -25,6 +25,7 @@
 #include "seastarx.hh"
 #include "utils/hash.hh"
 #include "utils/small_vector.hh"
+#include "cql3/cql3_type.hh"
 
 namespace auth {
 
@@ -36,7 +37,7 @@ public:
 };
 
 enum class resource_kind {
-    data, role, service_level
+    data, role, service_level, functions
 };
 
 std::ostream& operator<<(std::ostream&, resource_kind);
@@ -56,10 +57,15 @@ struct role_resource_t final {};
 ///
 struct service_level_resource_t final {};
 
+///
+/// Type tag for constructing function resources.
+///
+struct functions_resource_t final {};
+
 ///
 /// Resources are entities that users can be granted permissions on.
 ///
-/// There are data (keyspaces and tables) and role resources. There may be other kinds of resources in the future.
+/// There are data (keyspaces and tables), role and function resources. There may be other kinds of resources in the future.
 ///
 /// When they are stored as system metadata, resources have the form `root/part_0/part_1/.../part_n`. Each kind of
 /// resource has a specific root prefix, followed by a maximum of `n` parts (where `n` is distinct for each kind of
@@ -83,6 +89,11 @@ public:
     resource(data_resource_t, std::string_view keyspace, std::string_view table);
     resource(role_resource_t, std::string_view role);
     resource(service_level_resource_t);
+    explicit resource(functions_resource_t);
+    resource(functions_resource_t, std::string_view keyspace);
+    resource(functions_resource_t, std::string_view keyspace, std::string_view function_name);
+    resource(functions_resource_t, std::string_view keyspace, std::string_view function_name,
+            std::vector<sstring> function_signature);
 
     resource_kind kind() const noexcept {
         return _kind;
@@ -104,6 +115,7 @@ private:
     friend class data_resource_view;
     friend class role_resource_view;
     friend class service_level_resource_view;
+    friend class functions_resource_view;
 
     friend bool operator<(const resource&, const resource&);
     friend bool operator==(const resource&, const resource&);
@@ -182,6 +194,23 @@ public:
 
 std::ostream& operator<<(std::ostream&, const service_level_resource_view&);
 
+///
+/// A "function" view of \ref resource.
+///
+class functions_resource_view final {
+    const resource& _resource;
+public:
+    ///
+    /// \throws \ref resource_kind_mismatch if the argument is not a "function" resource.
+    ///
+    explicit functions_resource_view(const resource&);
+
+    std::optional<std::string_view> keyspace() const;
+    std::optional<std::string_view> function_signature() const;
+};
+
+std::ostream& operator<<(std::ostream&, const functions_resource_view&);
+
 ///
 /// Parse a resource from its name.
 ///
@@ -210,6 +239,26 @@ inline resource make_service_level_resource() {
     return resource(service_level_resource_t{});
 }
 
+const resource& root_function_resource();
+
+inline resource make_functions_resource() {
+    return resource(functions_resource_t{});
+}
+
+inline resource make_functions_resource(std::string_view keyspace) {
+    return resource(functions_resource_t{}, keyspace);
+}
+
+inline resource make_functions_resource(std::string_view keyspace, std::string_view function_name) {
+    return resource(functions_resource_t{}, keyspace, function_name);
+}
+
+resource make_functions_resource(std::string_view keyspace, std::string_view function_name, std::vector<::shared_ptr<cql3::cql3_type::raw>> function_signature);
+
+sstring encode_signature(std::string_view name, std::vector<data_type> args);
+
+std::pair<sstring, std::vector<data_type>> decode_signature(std::string_view encoded_signature);
+
 }
 
 namespace std {
@@ -228,13 +277,18 @@ struct hash<auth::resource> {
             return utils::tuple_hash()(std::make_tuple(auth::resource_kind::service_level));
     }
 
+    static size_t hash_function(const auth::functions_resource_view& fv) {
+        return utils::tuple_hash()(std::make_tuple(auth::resource_kind::functions, fv.keyspace(), fv.function_signature()));
+    }
+
     size_t operator()(const auth::resource& r) const {
         std::size_t value;
 
         switch (r._kind) {
         case auth::resource_kind::data: value = hash_data(auth::data_resource_view(r)); break;
         case auth::resource_kind::role: value = hash_role(auth::role_resource_view(r)); break;
         case auth::resource_kind::service_level: value = hash_service_level(auth::service_level_resource_view(r)); break;
+        case auth::resource_kind::functions: value = hash_function(auth::functions_resource_view(r)); break;
         }
 
         return value;

auth/service.cc

@@ -20,10 +20,12 @@
 #include "auth/allow_all_authorizer.hh"
 #include "auth/common.hh"
 #include "auth/role_or_anonymous.hh"
+#include "cql3/functions/functions.hh"
 #include "cql3/query_processor.hh"
 #include "cql3/untyped_result_set.hh"
 #include "db/config.hh"
 #include "db/consistency_level_type.hh"
+#include "db/functions/function_name.hh"
 #include "exceptions/exceptions.hh"
 #include "log.hh"
 #include "service/migration_manager.hh"
@@ -346,6 +348,22 @@ future<bool> service::exists(const resource& r) const {
         }
         case resource_kind::service_level:
             return make_ready_future<bool>(true);
+
+        case resource_kind::functions: {
+            const auto& db = _qp.db();
+
+            functions_resource_view v(r);
+            const auto keyspace = v.keyspace();
+            if (!keyspace) {
+                return make_ready_future<bool>(true);
+            }
+            const auto function_signature = v.function_signature();
+            if (!function_signature) {
+                return make_ready_future<bool>(db.has_keyspace(sstring(*keyspace)));
+            }
+            auto [name, function_args] = auth::decode_signature(*function_signature);
+            return make_ready_future<bool>(cql3::functions::functions::find(db::functions::function_name{sstring(*keyspace), name}, function_args));
+        }
     }
 
     return make_ready_future<bool>(false);