auth: Add TLS certificate authenticator

Fixes #10099

Adds the com.scylladb.auth.CertificateAuthenticator type. If set as authenticator,
will extract roles from TLS authentication certificate (not wire cert - those are
server side) subject, based on configurable regex.

Example:

scylla.yaml:

authenticator: com.scylladb.auth.CertificateAuthenticator
auth_superuser_name: <name>
auth_certificate_role_queries:
	- source: SUBJECT
	  query: CN=([^,\s]+)

client_encryption_options:
  enabled: True
  certificate: <server cert>
  keyfile: <server key>
  truststore: <shared trust>
  require_client_auth: True

In a client, then use a certificate signed with the <shared trust>
store as auth cert, with the common name <name>. I.e. for cqlsh
set "usercert" and "userkey" to these certificate files.

No user/password needs to be sent, but role will be picked up
from auth certificate. If none is present, the transport will
reject the connection. If the certificate subject does not
contain a recongnized role name (from config or set in tables)
the authenticator mechanism will reject it.

Otherwise, connection becomes the role described.

auth/certificate_authenticator.cc

@@ -0,0 +1,181 @@
+/*
+ * Copyright (C) 2022-present ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+#include "auth/certificate_authenticator.hh"
+
+#include <regex>
+
+#include "utils/class_registrator.hh"
+#include "data_dictionary/data_dictionary.hh"
+#include "cql3/query_processor.hh"
+#include "db/config.hh"
+
+static const auto CERT_AUTH_NAME = "com.scylladb.auth.CertificateAuthenticator";
+const std::string_view auth::certificate_authenticator_name(CERT_AUTH_NAME);
+
+static logging::logger clogger("certificate_authenticator");
+
+static const std::string cfg_source_attr = "source";
+static const std::string cfg_query_attr = "query";
+
+static const std::string cfg_source_subject = "SUBJECT";
+static const std::string cfg_source_altname = "ALTNAME";
+
+static const class_registrator<auth::authenticator
+    , auth::certificate_authenticator
+    , cql3::query_processor&
+    , ::service::migration_manager&> cert_auth_reg(CERT_AUTH_NAME);
+
+enum class auth::certificate_authenticator::query_source {
+    subject, altname
+};
+
+auth::certificate_authenticator::certificate_authenticator(cql3::query_processor& qp, ::service::migration_manager&)
+    : _queries([&] {
+        auto& conf = qp.db().get_config();
+        auto queries = conf.auth_certificate_role_queries();
+
+        if (queries.empty()) {
+            throw std::invalid_argument("No role extraction queries specified.");
+        }
+
+        std::vector<std::pair<query_source, boost::regex>> res;
+
+        for (auto& map : queries) {
+            // first, check for any invalid config keys
+            if (map.size() == 2) {
+                try {
+                    auto& source = map.at(cfg_source_attr);
+                    std::string query = map.at(cfg_query_attr);
+
+                    std::transform(source.begin(), source.end(), source.begin(), ::toupper);
+
+                    boost::regex ex(query);
+                    if (ex.mark_count() != 1) {
+                        throw std::invalid_argument("Role query must have exactly one mark expression");
+                    }
+
+                    clogger.debug("Append role query: {} : {}", source, query);
+
+                    if (source == cfg_source_subject) {
+                        res.emplace_back(query_source::subject, std::move(ex));
+                    } else if (source == cfg_source_altname) {
+                        res.emplace_back(query_source::altname, std::move(ex));
+                    } else {
+                        throw std::invalid_argument(fmt::format("Invalid source: {}", map.at(cfg_source_attr)));
+                    }
+                    continue;
+                } catch (std::out_of_range&) {
+                    // just fallthrough
+                } catch (std::regex_error&) {
+                    std::throw_with_nested(std::invalid_argument(fmt::format("Invalid query expression: {}", map.at(cfg_query_attr))));
+                }
+            }
+            throw std::invalid_argument(fmt::format("Invalid query: {}", map));
+        }
+        return res;
+    }())
+{}
+
+auth::certificate_authenticator::~certificate_authenticator() = default;
+
+future<> auth::certificate_authenticator::start() {
+    co_return;
+}
+
+future<> auth::certificate_authenticator::stop() {
+    co_return;
+}
+
+std::string_view auth::certificate_authenticator::qualified_java_name() const {
+    return certificate_authenticator_name;
+}
+
+bool auth::certificate_authenticator::require_authentication() const {
+    return true;
+}
+
+auth::authentication_option_set auth::certificate_authenticator::supported_options() const {
+    return {};
+}
+
+auth::authentication_option_set auth::certificate_authenticator::alterable_options() const {
+    return {};
+}
+
+future<std::optional<auth::authenticated_user>> auth::certificate_authenticator::authenticate(session_dn_func f) const {
+    if (!f) {
+        co_return std::nullopt;
+    }
+    auto dninfo = co_await f();
+    if (!dninfo) {
+        throw exceptions::authentication_exception("No valid certificate found");
+    }
+
+    auto& subject = dninfo->subject;
+    std::optional<std::string> altname ;
+
+    const std::string* source_str = nullptr;
+
+    for (auto& [source, expr] : _queries) {
+        switch (source) {
+            default:
+            case query_source::subject:
+                source_str = &subject;
+                break;
+            case query_source::altname:
+                if (!altname) {
+                    altname = dninfo->get_alt_names ? co_await dninfo->get_alt_names() : std::string{};
+                }
+                source_str = &*altname;
+                break;
+        }
+
+        clogger.debug("Checking {}: {}", int(source), *source_str);
+
+        boost::smatch m;
+        if (boost::regex_search(*source_str, m, expr)) {
+            auto username = m[1].str();
+            clogger.debug("Return username: {}", username);
+            co_return username;
+        }
+    }
+    throw exceptions::authentication_exception(format("Subject '{}'/'{}' does not match any query expression", subject, altname));
+}
+
+
+future<auth::authenticated_user> auth::certificate_authenticator::authenticate(const credentials_map&) const {
+    throw exceptions::authentication_exception("Cannot authenticate using attribute map");
+}
+
+future<> auth::certificate_authenticator::create(std::string_view role_name, const authentication_options& options) const {
+    // TODO: should we keep track of roles/enforce existence? Role manager should deal with this...
+    co_return;
+}
+
+future<> auth::certificate_authenticator::alter(std::string_view role_name, const authentication_options& options) const {
+    co_return;
+}
+
+future<> auth::certificate_authenticator::drop(std::string_view role_name) const {
+    co_return;
+}
+
+future<auth::custom_options> auth::certificate_authenticator::query_custom_options(std::string_view) const {
+    co_return auth::custom_options{};
+}
+
+const auth::resource_set& auth::certificate_authenticator::protected_resources() const {
+    static const resource_set resources;
+    return resources;
+}
+
+::shared_ptr<auth::sasl_challenge> auth::certificate_authenticator::new_sasl_challenge() const {
+    throw exceptions::authentication_exception("Login authentication not supported");
+}

auth/certificate_authenticator.hh

@@ -0,0 +1,62 @@
+/*
+ * Copyright (C) 2022-present ScyllaDB
+ *
+ */
+
+/*
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+#pragma once
+
+#include <boost/regex.hpp>
+#include "auth/authenticator.hh"
+
+namespace cql3 {
+
+class query_processor;
+
+} // namespace cql3
+
+namespace service {
+class migration_manager;
+}
+
+namespace auth {
+
+extern const std::string_view certificate_authenticator_name;
+
+class certificate_authenticator : public authenticator {
+    enum class query_source;
+    std::vector<std::pair<query_source, boost::regex>> _queries;
+public:
+    certificate_authenticator(cql3::query_processor&, ::service::migration_manager&);
+    ~certificate_authenticator();
+
+    future<> start() override;
+    future<> stop() override;
+
+    std::string_view qualified_java_name() const override;
+
+    bool require_authentication() const override;
+
+    authentication_option_set supported_options() const override;
+    authentication_option_set alterable_options() const override;
+
+    future<authenticated_user> authenticate(const credentials_map& credentials) const override;
+    future<std::optional<authenticated_user>> authenticate(session_dn_func) const override;
+
+    future<> create(std::string_view role_name, const authentication_options& options) const override;
+    future<> alter(std::string_view role_name, const authentication_options& options) const override;
+    future<> drop(std::string_view role_name) const override;
+
+    future<custom_options> query_custom_options(std::string_view role_name) const override;
+
+    const resource_set& protected_resources() const override;
+
+    ::shared_ptr<sasl_challenge> new_sasl_challenge() const override;
+private:
+};
+
+}
+

configure.py

@@ -1016,6 +1016,7 @@ scylla_core = (['message/messaging_service.cc',
                 'auth/transitional.cc',
                 'auth/role_or_anonymous.cc',
                 'auth/sasl_challenge.cc',
+                'auth/certificate_authenticator.cc',
                 'tracing/tracing.cc',
                 'tracing/trace_keyspace_helper.cc',
                 'tracing/trace_state.cc',

db/config.cc

@@ -126,6 +126,9 @@ const config_type config_type_for<std::vector<sstring>> = config_type("string li
 template <>
 const config_type config_type_for<std::unordered_map<sstring, sstring>> = config_type("string map", value_to_json<std::unordered_map<sstring, sstring>>);
 
+template <>
+const config_type config_type_for<std::vector<std::unordered_map<sstring, sstring>>> = config_type("string map list", value_to_json<std::vector<std::unordered_map<sstring, sstring>>>);
+
 template <>
 const config_type config_type_for<std::unordered_map<sstring, log_level>> = config_type("string map", log_level_map_to_json);
 
@@ -756,9 +759,10 @@ db::config::config(std::shared_ptr<db::extensions> exts)
         "\n"
         "\torg.apache.cassandra.auth.AllowAllAuthenticator : Disables authentication; no checks are performed.\n"
         "\torg.apache.cassandra.auth.PasswordAuthenticator : Authenticates users with user names and hashed passwords stored in the system_auth.credentials table. If you use the default, 1, and the node with the lone replica goes down, you will not be able to log into the cluster because the system_auth keyspace was not replicated.\n"
+        "\tcom.scylladb.auth.CertificateAuthenticator : Authenticates users based on TLS certificate authentication subject. Roles and permissions still need to be defined as normal. Super user can be set using the 'auth_superuser_name' configuration value. Query to extract role name from subject string is set using 'auth_certificate_role_queries'.\n"
         "\tcom.scylladb.auth.TransitionalAuthenticator : Wraps around the PasswordAuthenticator, logging them in if username/password pair provided is correct and treating them as anonymous users otherwise.\n"
         "Related information: Internal authentication"
-        , {"AllowAllAuthenticator", "PasswordAuthenticator", "org.apache.cassandra.auth.PasswordAuthenticator", "org.apache.cassandra.auth.AllowAllAuthenticator", "com.scylladb.auth.TransitionalAuthenticator"})
+        , {"AllowAllAuthenticator", "PasswordAuthenticator", "CertificateAuthenticator", "org.apache.cassandra.auth.PasswordAuthenticator", "org.apache.cassandra.auth.AllowAllAuthenticator", "com.scylladb.auth.TransitionalAuthenticator", "com.scylladb.auth.CertificateAuthenticator"})
     , internode_authenticator(this, "internode_authenticator", value_status::Unused, "enabled",
         "Internode authentication backend. It implements org.apache.cassandra.auth.AllowAllInternodeAuthenticator to allows or disallow connections from peer nodes.")
     , authorizer(this, "authorizer", value_status::Used, "org.apache.cassandra.auth.AllowAllAuthorizer",
@@ -998,6 +1002,8 @@ db::config::config(std::shared_ptr<db::extensions> exts)
     , auth_superuser_salted_password(this, "auth_superuser_salted_password", value_status::Used, "", 
         "Initial authentication super user salted password. Create using mkpassword or similar. The hashing algorithm used must be available on the node host. "
         "Ignored if authentication tables already contain a super user password.")
+    , auth_certificate_role_queries(this, "auth_certificate_role_queries", value_status::Used, { { { "source", "SUBJECT" }, {"query", "CN=([^,]+)" } } },
+        "Regular expression used by CertificateAuthenticator to extract role name from an accepted transport authentication certificate subject info.")
     , error_injections_at_startup(this, "error_injections_at_startup", error_injection_value_status, {}, "List of error injections that should be enabled on startup.")
     , default_log_level(this, "default_log_level", value_status::Used)
     , logger_log_level(this, "logger_log_level", value_status::Used)

db/config.hh

@@ -445,6 +445,8 @@ public:
     named_value<std::string> auth_superuser_name;
     named_value<std::string> auth_superuser_salted_password;
 
+    named_value<std::vector<std::unordered_map<sstring, sstring>>> auth_certificate_role_queries;
+
     seastar::logging_settings logging_settings(const log_cli::options&) const;
 
     const db::extensions& extensions() const;