Cleanup code in KerberosInit, HDFSCommon and StorageKafka; update English and Russian documentation.

This commit is contained in:
Roman Vasin 2022-06-08 17:57:45 +03:00
parent 2b76d0c6a9
commit 3cfea6e76f
10 changed files with 104 additions and 134 deletions

View File

@ -186,7 +186,6 @@ Similar to GraphiteMergeTree, the HDFS engine supports extended configuration us
| - | - |
|hadoop\_kerberos\_keytab | "" |
|hadoop\_kerberos\_principal | "" |
|hadoop\_kerberos\_kinit\_command | kinit |
|libhdfs3\_conf | "" |
### Limitations {#limitations}
@ -200,8 +199,7 @@ Note that due to libhdfs3 limitations only old-fashioned approach is supported,
datanode communications are not secured by SASL (`HADOOP_SECURE_DN_USER` is a reliable indicator of such
security approach). Use `tests/integration/test_storage_kerberized_hdfs/hdfs_configs/bootstrap.sh` for reference.
If `hadoop_kerberos_keytab`, `hadoop_kerberos_principal` or `hadoop_kerberos_kinit_command` is specified, `kinit` will be invoked. `hadoop_kerberos_keytab` and `hadoop_kerberos_principal` are mandatory in this case. `kinit` tool and krb5 configuration files are required.
If `hadoop_kerberos_keytab`, `hadoop_kerberos_principal` or `hadoop_security_kerberos_ticket_cache_path` are specified, Kerberos authentication will be used. `hadoop_kerberos_keytab` and `hadoop_kerberos_principal` are mandatory in this case.
## HDFS Namenode HA support {#namenode-ha}
libhdfs3 support HDFS namenode HA.

View File

@ -168,7 +168,7 @@ For a list of possible configuration options, see the [librdkafka configuration
### Kerberos support {#kafka-kerberos-support}
To deal with Kerberos-aware Kafka, add `security_protocol` child element with `sasl_plaintext` value. It is enough if Kerberos ticket-granting ticket is obtained and cached by OS facilities.
ClickHouse is able to maintain Kerberos credentials using a keytab file. Consider `sasl_kerberos_service_name`, `sasl_kerberos_keytab`, `sasl_kerberos_principal` and `sasl.kerberos.kinit.cmd` child elements.
ClickHouse is able to maintain Kerberos credentials using a keytab file. Consider `sasl_kerberos_service_name`, `sasl_kerberos_keytab` and `sasl_kerberos_principal` child elements.
Example:

View File

@ -183,7 +183,6 @@ CREATE TABLE big_table (name String, value UInt32) ENGINE = HDFS('hdfs://hdfs1:9
| - | - |
|hadoop\_kerberos\_keytab | "" |
|hadoop\_kerberos\_principal | "" |
|hadoop\_kerberos\_kinit\_command | kinit |
### Ограничения {#limitations}
* `hadoop_security_kerberos_ticket_cache_path` и `libhdfs3_conf` могут быть определены только на глобальном, а не на пользовательском уровне
@ -196,7 +195,7 @@ CREATE TABLE big_table (name String, value UInt32) ENGINE = HDFS('hdfs://hdfs1:9
коммуникация с узлами данных не защищена SASL (`HADOOP_SECURE_DN_USER` надежный показатель такого
подхода к безопасности). Используйте `tests/integration/test_storage_kerberized_hdfs/hdfs_configs/bootstrap.sh` для примера настроек.
Если `hadoop_kerberos_keytab`, `hadoop_kerberos_principal` или `hadoop_kerberos_kinit_command` указаны в настройках, `kinit` будет вызван. `hadoop_kerberos_keytab` и `hadoop_kerberos_principal` обязательны в этом случае. Необходимо также будет установить `kinit` и файлы конфигурации krb5.
Если `hadoop_kerberos_keytab`, `hadoop_kerberos_principal` или `hadoop_security_kerberos_ticket_cache_path` указаны в настройках, будет использоваться аутентификация с помощью Kerberos. `hadoop_kerberos_keytab` и `hadoop_kerberos_principal` обязательны в этом случае.
## Виртуальные столбцы {#virtual-columns}

View File

@ -167,7 +167,7 @@ Kafka(kafka_broker_list, kafka_topic_list, kafka_group_name, kafka_format
### Поддержка Kerberos {#kafka-kerberos-support}
Чтобы начать работу с Kafka с поддержкой Kerberos, добавьте дочерний элемент `security_protocol` со значением `sasl_plaintext`. Этого будет достаточно, если получен тикет на получение тикета (ticket-granting ticket) Kerberos и он кэшируется средствами ОС.
ClickHouse может поддерживать учетные данные Kerberos с помощью файла keytab. Рассмотрим дочерние элементы `sasl_kerberos_service_name`, `sasl_kerberos_keytab`, `sasl_kerberos_principal` и `sasl.kerberos.kinit.cmd`.
ClickHouse может поддерживать учетные данные Kerberos с помощью файла keytab. Рассмотрим дочерние элементы `sasl_kerberos_service_name`, `sasl_kerberos_keytab` и `sasl_kerberos_principal`.
Пример:

View File

@ -5,11 +5,17 @@
#include <Loggers/Loggers.h>
#include <filesystem>
using namespace DB;
std::mutex KerberosInit::kinit_mtx;
int KerberosInit::init(const String & keytab_file, const String & principal, const String & cache_name)
{
auto adqm_log = &Poco::Logger::get("ADQM");
LOG_DEBUG(adqm_log,"KerberosInit: begin");
//LOG_DEBUG(adqm_log,"KerberosInit: do nothing"); return 0;
// Using mutex to prevent cache file corruptions
std::unique_lock<std::mutex> lck(kinit_mtx);
auto log = &Poco::Logger::get("ADQM");
LOG_DEBUG(log,"Trying to authenticate to Kerberos v5");
krb5_error_code ret;
@ -17,182 +23,152 @@ int KerberosInit::init(const String & keytab_file, const String & principal, con
int flags = 0;
if (!std::filesystem::exists(keytab_file))
throw DB::Exception(0, "Error keytab file does not exist");
throw Exception("Error keytab file does not exist", ErrorCodes::KERBEROS_ERROR);
memset(&k5d, 0, sizeof(k5d));
k5 = &k5d;
// begin
ret = krb5_init_context(&k5->ctx);
memset(&k5, 0, sizeof(k5));
ret = krb5_init_context(&k5.ctx);
if (ret)
throw DB::Exception(0, "Error while initializing Kerberos 5 library");
throw Exception("Error while initializing Kerberos 5 library", ErrorCodes::KERBEROS_ERROR);
if (!cache_name.empty())
{
ret = krb5_cc_resolve(k5->ctx, cache_name.c_str(), &k5->out_cc);
// todo: analyze return code
LOG_DEBUG(adqm_log,"Resolved cache");
ret = krb5_cc_resolve(k5.ctx, cache_name.c_str(), &k5.out_cc);
if (ret)
throw Exception("Error in resolving cache", ErrorCodes::KERBEROS_ERROR);
LOG_DEBUG(log,"Resolved cache");
}
else
{
// Resolve the default ccache and get its type and default principal (if it is initialized).
ret = krb5_cc_default(k5->ctx, &defcache);
ret = krb5_cc_default(k5.ctx, &defcache);
if (ret)
throw DB::Exception(0, "Error while getting default ccache");
LOG_DEBUG(adqm_log,"Resolved default cache");
deftype = krb5_cc_get_type(k5->ctx, defcache);
if (krb5_cc_get_principal(k5->ctx, defcache, &defcache_princ) != 0)
throw Exception("Error while getting default ccache", ErrorCodes::KERBEROS_ERROR);
LOG_DEBUG(log,"Resolved default cache");
deftype = krb5_cc_get_type(k5.ctx, defcache);
if (krb5_cc_get_principal(k5.ctx, defcache, &defcache_princ) != 0)
defcache_princ = nullptr;
}
// Use the specified principal name.
ret = krb5_parse_name_flags(k5->ctx, principal.c_str(), flags, &k5->me);
ret = krb5_parse_name_flags(k5.ctx, principal.c_str(), flags, &k5.me);
if (ret)
throw DB::Exception(0, "Error when parsing principal name " + principal);
throw Exception("Error when parsing principal name " + principal, ErrorCodes::KERBEROS_ERROR);
// Cache related commands
if (k5->out_cc == nullptr && krb5_cc_support_switch(k5->ctx, deftype))
if (k5.out_cc == nullptr && krb5_cc_support_switch(k5.ctx, deftype))
{
// Use an existing cache for the client principal if we can.
ret = krb5_cc_cache_match(k5->ctx, k5->me, &k5->out_cc);
ret = krb5_cc_cache_match(k5.ctx, k5.me, &k5.out_cc);
if (ret && ret != KRB5_CC_NOTFOUND)
throw DB::Exception(0, "Error while searching for ccache for " + principal);
throw Exception("Error while searching for cache for " + principal, ErrorCodes::KERBEROS_ERROR);
if (!ret)
{
LOG_DEBUG(adqm_log,"Using default cache: {}", krb5_cc_get_name(k5->ctx, k5->out_cc));
k5->switch_to_cache = 1;
LOG_DEBUG(log,"Using default cache: {}", krb5_cc_get_name(k5.ctx, k5.out_cc));
k5.switch_to_cache = 1;
}
else if (defcache_princ != nullptr)
{
// Create a new cache to avoid overwriting the initialized default cache.
ret = krb5_cc_new_unique(k5->ctx, deftype, nullptr, &k5->out_cc);
ret = krb5_cc_new_unique(k5.ctx, deftype, nullptr, &k5.out_cc);
if (ret)
throw DB::Exception(0, "Error while generating new ccache");
LOG_DEBUG(adqm_log,"Using default cache: {}", krb5_cc_get_name(k5->ctx, k5->out_cc));
k5->switch_to_cache = 1;
throw Exception("Error while generating new cache", ErrorCodes::KERBEROS_ERROR);
LOG_DEBUG(log,"Using default cache: {}", krb5_cc_get_name(k5.ctx, k5.out_cc));
k5.switch_to_cache = 1;
}
}
// Use the default cache if we haven't picked one yet.
if (k5->out_cc == nullptr)
if (k5.out_cc == nullptr)
{
k5->out_cc = defcache;
k5.out_cc = defcache;
defcache = nullptr;
LOG_DEBUG(adqm_log,"Using default cache: {}", krb5_cc_get_name(k5->ctx, k5->out_cc));
LOG_DEBUG(log,"Using default cache: {}", krb5_cc_get_name(k5.ctx, k5.out_cc));
}
ret = krb5_unparse_name(k5->ctx, k5->me, &k5->name);
ret = krb5_unparse_name(k5.ctx, k5.me, &k5.name);
if (ret)
throw DB::Exception(0, "Error when unparsing name");
throw Exception("Error when unparsing name", ErrorCodes::KERBEROS_ERROR);
LOG_DEBUG(log,"Using principal: {}", k5.name);
LOG_DEBUG(adqm_log,"KerberosInit: Using principal: {}", k5->name);
// init:
memset(&my_creds, 0, sizeof(my_creds));
ret = krb5_get_init_creds_opt_alloc(k5->ctx, &options);
ret = krb5_get_init_creds_opt_alloc(k5.ctx, &options);
if (ret)
throw DB::Exception(0, "Error in options allocation");
throw Exception("Error in options allocation", ErrorCodes::KERBEROS_ERROR);
// todo
/*
#ifndef _WIN32
if (strncmp(opts->keytab_name, "KDB:", 4) == 0) {
ret = kinit_kdb_init(&k5->ctx, k5->me->realm.data);
errctx = k5->ctx;
if (ret) {
com_err(progname, ret,
_("while setting up KDB keytab for realm %s"),
k5->me->realm.data);
goto cleanup;
}
}
#endif
*/
// Resolve keytab
ret = krb5_kt_resolve(k5->ctx, keytab_file.c_str(), &keytab);
ret = krb5_kt_resolve(k5.ctx, keytab_file.c_str(), &keytab);
if (ret)
throw DB::Exception(0, "Error resolving keytab "+keytab_file);
throw Exception("Error in resolving keytab "+keytab_file, ErrorCodes::KERBEROS_ERROR);
LOG_DEBUG(log,"Using keytab: {}", keytab_file);
LOG_DEBUG(adqm_log,"KerberosInit: Using keytab: {}", keytab_file);
if (k5->in_cc)
if (k5.in_cc)
{
ret = krb5_get_init_creds_opt_set_in_ccache(k5->ctx, options, k5->in_cc);
ret = krb5_get_init_creds_opt_set_in_ccache(k5.ctx, options, k5.in_cc);
if (ret)
throw DB::Exception(0, "Error in setting input credential cache");
throw Exception("Error in setting input credential cache", ErrorCodes::KERBEROS_ERROR);
}
ret = krb5_get_init_creds_opt_set_out_ccache(k5->ctx, options, k5->out_cc);
ret = krb5_get_init_creds_opt_set_out_ccache(k5.ctx, options, k5.out_cc);
if (ret)
throw DB::Exception(0, "Error in setting output credential cache");
throw Exception("Error in setting output credential cache", ErrorCodes::KERBEROS_ERROR);
// action: init or renew
LOG_DEBUG(adqm_log,"Trying to renew credentials");
ret = krb5_get_renewed_creds(k5->ctx, &my_creds, k5->me, k5->out_cc, nullptr);
// Action: init or renew
LOG_DEBUG(log,"Trying to renew credentials");
ret = krb5_get_renewed_creds(k5.ctx, &my_creds, k5.me, k5.out_cc, nullptr);
if (ret)
{
LOG_DEBUG(adqm_log,"Renew failed, making init credentials");
ret = krb5_get_init_creds_keytab(k5->ctx, &my_creds, k5->me, keytab, 0, nullptr, options);
LOG_DEBUG(log,"Renew failed, trying to get initial credentials");
ret = krb5_get_init_creds_keytab(k5.ctx, &my_creds, k5.me, keytab, 0, nullptr, options);
if (ret)
throw DB::Exception(0, "Error in init");
throw Exception("Error in getting initial credentials", ErrorCodes::KERBEROS_ERROR);
else
LOG_DEBUG(adqm_log,"Getting initial credentials");
LOG_DEBUG(log,"Got initial credentials");
}
else
{
LOG_DEBUG(adqm_log,"Successfull reviewal");
ret = krb5_cc_initialize(k5->ctx, k5->out_cc, k5->me);
LOG_DEBUG(log,"Successfull renewal");
ret = krb5_cc_initialize(k5.ctx, k5.out_cc, k5.me);
if (ret)
throw DB::Exception(0, "Error when initializing cache");
LOG_DEBUG(adqm_log,"Initialized cache");
ret = krb5_cc_store_cred(k5->ctx, k5->out_cc, &my_creds);
throw Exception("Error when initializing cache", ErrorCodes::KERBEROS_ERROR);
LOG_DEBUG(log,"Initialized cache");
ret = krb5_cc_store_cred(k5.ctx, k5.out_cc, &my_creds);
if (ret)
LOG_DEBUG(adqm_log,"Error while storing credentials");
LOG_DEBUG(adqm_log,"Stored credentials");
LOG_DEBUG(log,"Error while storing credentials");
LOG_DEBUG(log,"Stored credentials");
}
if (k5->switch_to_cache) {
ret = krb5_cc_switch(k5->ctx, k5->out_cc);
if (k5.switch_to_cache) {
ret = krb5_cc_switch(k5.ctx, k5.out_cc);
if (ret)
throw DB::Exception(0, "Error while switching to new ccache");
throw Exception("Error while switching to new ccache", ErrorCodes::KERBEROS_ERROR);
}
LOG_DEBUG(adqm_log,"Authenticated to Kerberos v5");
LOG_DEBUG(adqm_log,"KerberosInit: end");
LOG_DEBUG(log,"Authenticated to Kerberos v5");
return 0;
}
KerberosInit::~KerberosInit()
{
if (k5->ctx)
std::unique_lock<std::mutex> lck(kinit_mtx);
if (k5.ctx)
{
//begin. cleanup:
if (defcache)
krb5_cc_close(k5->ctx, defcache);
krb5_free_principal(k5->ctx, defcache_princ);
krb5_cc_close(k5.ctx, defcache);
krb5_free_principal(k5.ctx, defcache_princ);
// init. cleanup:
//todo:
/*
#ifndef _WIN32
kinit_kdb_fini();
#endif
*/
if (options)
krb5_get_init_creds_opt_free(k5->ctx, options);
if (my_creds.client == k5->me)
krb5_get_init_creds_opt_free(k5.ctx, options);
if (my_creds.client == k5.me)
my_creds.client = nullptr;
krb5_free_cred_contents(k5->ctx, &my_creds);
krb5_free_cred_contents(k5.ctx, &my_creds);
if (keytab)
krb5_kt_close(k5->ctx, keytab);
krb5_kt_close(k5.ctx, keytab);
// end. cleanup:
krb5_free_unparsed_name(k5->ctx, k5->name);
krb5_free_principal(k5->ctx, k5->me);
if (k5->in_cc != nullptr)
krb5_cc_close(k5->ctx, k5->in_cc);
if (k5->out_cc != nullptr)
krb5_cc_close(k5->ctx, k5->out_cc);
krb5_free_context(k5->ctx);
krb5_free_unparsed_name(k5.ctx, k5.name);
krb5_free_principal(k5.ctx, k5.me);
if (k5.in_cc != nullptr)
krb5_cc_close(k5.ctx, k5.in_cc);
if (k5.out_cc != nullptr)
krb5_cc_close(k5.ctx, k5.out_cc);
krb5_free_context(k5.ctx);
}
memset(k5, 0, sizeof(*k5));
}

View File

@ -4,10 +4,16 @@
#include <base/types.h>
//#include <k5-int.h>
//#include "k5-platform.h"
#include <krb5.h>
//#include <extern.h>
#include <mutex>
namespace DB
{
namespace ErrorCodes
{
extern const int KERBEROS_ERROR;
}
}
struct k5_data
{
@ -24,13 +30,14 @@ public:
int init(const String & keytab_file, const String & principal, const String & cache_name = "");
~KerberosInit();
private:
struct k5_data * k5 = nullptr;
//struct k5_data k5;
struct k5_data k5d;
//struct k5_data * k5 = nullptr;
struct k5_data k5;
//struct k5_data k5d;
krb5_ccache defcache = nullptr;
krb5_get_init_creds_opt *options = nullptr;
krb5_creds my_creds;
krb5_keytab keytab = nullptr;
krb5_principal defcache_princ = nullptr;
static std::mutex kinit_mtx;
};

View File

@ -13,6 +13,7 @@ int main(int argc, char ** argv)
if (argc < 3)
{
std::cout << "kerberos_init obtains and caches an initial ticket-granting ticket for principal." << "\n\n";
std::cout << "Usage:" << "\n" << " kerberos_init keytab principal [cache]" << "\n";
return 0;
}
@ -31,6 +32,7 @@ int main(int argc, char ** argv)
k_init.init(argv[1], argv[2], cache_name);
} catch (const Exception & e) {
std::cout << "KerberosInit failure: " << getExceptionMessage(e, false) << "\n";
return -1;
}
std::cout << "Done" << "\n";
return 0;

View File

@ -52,12 +52,6 @@ void HDFSBuilderWrapper::loadFromConfig(const Poco::Util::AbstractConfiguration
hdfsBuilderSetPrincipal(hdfs_builder, hadoop_kerberos_principal.c_str());
continue;
}
else if (key == "hadoop_kerberos_kinit_command")
{
need_kinit = true;
hadoop_kerberos_kinit_command = config.getString(key_path);
continue;
}
else if (key == "hadoop_security_kerberos_ticket_cache_path")
{
if (isUser)
@ -80,13 +74,12 @@ void HDFSBuilderWrapper::loadFromConfig(const Poco::Util::AbstractConfiguration
void HDFSBuilderWrapper::runKinit()
{
LOG_DEBUG(&Poco::Logger::get("HDFSClient"), "ADQM: running KerberosInit");
std::unique_lock<std::mutex> lck(kinit_mtx);
KerberosInit k_init;
try
{
k_init.init(hadoop_kerberos_keytab,hadoop_kerberos_principal,hadoop_security_kerberos_ticket_cache_path);
} catch (const DB::Exception & e) {
throw Exception("KerberosInit failure: "+ DB::getExceptionMessage(e, false), ErrorCodes::BAD_ARGUMENTS);
throw Exception("KerberosInit failure: "+ getExceptionMessage(e, false), ErrorCodes::KERBEROS_ERROR);
}
LOG_DEBUG(&Poco::Logger::get("HDFSClient"), "ADQM: finished KerberosInit");
}
@ -168,8 +161,6 @@ HDFSBuilderWrapper createHDFSBuilder(const String & uri_str, const Poco::Util::A
return builder;
}
std::mutex HDFSBuilderWrapper::kinit_mtx;
HDFSFSPtr createHDFSFS(hdfsBuilder * builder)
{
HDFSFSPtr fs(hdfsBuilderConnect(builder));

View File

@ -9,7 +9,6 @@
#include <hdfs/hdfs.h>
#include <base/types.h>
#include <mutex>
#include <Interpreters/Context.h>
#include <Poco/Util/AbstractConfiguration.h>
@ -80,10 +79,8 @@ private:
hdfsBuilder * hdfs_builder;
String hadoop_kerberos_keytab;
String hadoop_kerberos_principal;
String hadoop_kerberos_kinit_command = "kinit";
String hadoop_security_kerberos_ticket_cache_path;
static std::mutex kinit_mtx;
std::vector<std::pair<String, String>> config_stor;
bool need_kinit{false};
};

View File

@ -528,8 +528,8 @@ void StorageKafka::updateConfiguration(cppkafka::Configuration & conf)
try
{
k_init.init(keytab,principal);
} catch (const DB::Exception & e) {
LOG_ERROR(log, "ADQM: KerberosInit failure: {}", DB::getExceptionMessage(e, false));
} catch (const Exception & e) {
LOG_ERROR(log, "ADQM: KerberosInit failure: {}", getExceptionMessage(e, false));
}
LOG_DEBUG(log, "ADQM: finished KerberosInit");
conf.set("sasl.kerberos.kinit.cmd","");