auth: Check for unsupported authentication options

While it's undefined behavior to pass an unsupported option to a
specific authenticator directly, the `auth::service` layer will check
options and throw this exception. It is turned into a
`invalid_request_exception` by the CQL layer.

auth/authentication_options.hh

@@ -23,9 +23,11 @@
 
 #include <iosfwd>
 #include <optional>
+#include <stdexcept>
 #include <unordered_map>
 #include <unordered_set>
 
+#include <seastar/core/print.hh>
 #include <seastar/core/sstring.hh>
 
 #include "seastarx.hh"
@@ -50,4 +52,11 @@ inline bool any_authentication_options(const authentication_options& aos) noexce
     return aos.password || aos.options;
 }
 
+class unsupported_authentication_option : public std::invalid_argument {
+public:
+    explicit unsupported_authentication_option(authentication_option k)
+            : std::invalid_argument(sprint("The %s option is not supported.", k)) {
+    }
+};
+
 }

auth/service.cc

@@ -291,6 +291,25 @@ bool is_enforcing(const service& ser)  {
     return enforcing_authorizer || enforcing_authenticator;
 }
 
+static void validate_authentication_options_are_supported(
+        const authentication_options& options,
+        const authentication_option_set& supported) {
+    const auto check = [&supported](authentication_option k) {
+        if (supported.count(k) == 0) {
+            throw unsupported_authentication_option(k);
+        }
+    };
+
+    if (options.password) {
+        check(authentication_option::password);
+    }
+
+    if (options.options) {
+        check(authentication_option::options);
+    }
+}
+
+
 future<> create_role(
         service& ser,
         const authenticated_user& performer,
@@ -305,11 +324,14 @@ future<> create_role(
             return make_ready_future<>();
         }
 
-        return ser.underlying_authenticator().create(
-                sstring(name),
-                options).handle_exception([&ser, &performer, &name](std::exception_ptr ep) {
-             // Roll-back.
-             return ser.underlying_role_manager().drop(performer, name).then([ep = std::move(ep)] {
+        return futurize_apply(
+                &validate_authentication_options_are_supported,
+                options,
+                ser.underlying_authenticator().supported_options()).then([&ser, name, &options] {
+            return ser.underlying_authenticator().create(sstring(name), options);
+        }).handle_exception([&ser, &performer, &name](std::exception_ptr ep) {
+            // Roll-back.
+            return ser.underlying_role_manager().drop(performer, name).then([ep = std::move(ep)] {
                 std::rethrow_exception(ep);
             });
         });
@@ -327,7 +349,12 @@ future<> alter_role(
             return make_ready_future<>();
         }
 
-        return ser.underlying_authenticator().alter(sstring(name), options);
+        return futurize_apply(
+                &validate_authentication_options_are_supported,
+                options,
+                ser.underlying_authenticator().supported_options()).then([&ser, name, &options] {
+            return ser.underlying_authenticator().alter(sstring(name), options);
+        });
     });
 }
 

auth/service.hh

@@ -183,6 +183,8 @@ bool is_enforcing(const service&);
 ///
 /// \returns an exceptional future with \ref role_already_exists if the user or role exists.
 ///
+/// \returns an exceptional future with \ref unsupported_authentication_option if an unsupported option is included.
+///
 future<> create_role(
         service&,
         const authenticated_user& performer,
@@ -195,6 +197,8 @@ future<> create_role(
 ///
 /// \returns an exceptional future with \ref nonexistant_role if the named role does not exist.
 ///
+/// \returns an exceptional future with \ref unsupported_authentication_option if an unsupported option is included.
+///
 future<> alter_role(
         service&,
         const authenticated_user& performer,

cql3/statements/role-management-statements.cc

@@ -122,6 +122,8 @@ create_role_statement::execute(distributed<service::storage_proxy>&,
             }
 
             return void_result_message();
+        }).handle_exception_type([](const auth::unsupported_authentication_option& e) {
+            return make_exception_future<result_message_ptr>(exceptions::invalid_request_exception(e.what()));
         });
     });
 }
@@ -192,6 +194,8 @@ alter_role_statement::execute(distributed<service::storage_proxy>&, service::que
             return void_result_message();
         }).handle_exception_type([](const auth::roles_argument_exception& e) {
             return make_exception_future<result_message_ptr>(exceptions::invalid_request_exception(e.what()));
+        }).handle_exception_type([](const auth::unsupported_authentication_option& e) {
+            return make_exception_future<result_message_ptr>(exceptions::invalid_request_exception(e.what()));
         });
     });
 }