mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-11-14 03:25:15 +00:00
92 lines
2.2 KiB
Python
92 lines
2.2 KiB
Python
import pytest
|
|
from helpers.cluster import ClickHouseCluster, is_arm
|
|
|
|
if is_arm():
|
|
pytestmark = pytest.mark.skip
|
|
|
|
|
|
cluster = ClickHouseCluster(__file__)
|
|
instance1 = cluster.add_instance(
|
|
"instance1",
|
|
main_configs=["configs/kerberos_with_keytab.xml"],
|
|
user_configs=["configs/users.xml"],
|
|
with_kerberos_kdc=True,
|
|
)
|
|
instance2 = cluster.add_instance(
|
|
"instance2",
|
|
main_configs=["configs/kerberos_without_keytab.xml"],
|
|
user_configs=["configs/users.xml"],
|
|
with_kerberos_kdc=True,
|
|
)
|
|
instance3 = cluster.add_instance(
|
|
"instance3",
|
|
main_configs=["configs/kerberos_bad_path_to_keytab.xml"],
|
|
user_configs=["configs/users.xml"],
|
|
with_kerberos_kdc=True,
|
|
)
|
|
client = cluster.add_instance(
|
|
"client",
|
|
main_configs=["configs/kerberos_without_keytab.xml"],
|
|
user_configs=["configs/users.xml"],
|
|
with_kerberos_kdc=True,
|
|
)
|
|
|
|
|
|
# Fixtures
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def kerberos_cluster():
|
|
try:
|
|
cluster.start()
|
|
yield cluster
|
|
finally:
|
|
cluster.shutdown()
|
|
|
|
|
|
# Tests
|
|
|
|
|
|
def make_auth(instance):
|
|
instance_ip = cluster.get_instance_ip(instance.name)
|
|
|
|
client.exec_in_container(
|
|
["bash", "-c", f"echo '{instance_ip} {instance.hostname}' >> /etc/hosts"]
|
|
)
|
|
|
|
client.exec_in_container(
|
|
["bash", "-c", "kinit -k -t /tmp/keytab/kuser.keytab kuser"]
|
|
)
|
|
return client.exec_in_container(
|
|
[
|
|
"bash",
|
|
"-c",
|
|
f"echo 'select currentUser()' | curl --negotiate -u : http://{instance.hostname}:8123/ --data-binary @-",
|
|
]
|
|
)
|
|
|
|
|
|
def test_kerberos_auth_with_keytab(kerberos_cluster):
|
|
assert make_auth(instance1) == "kuser\n"
|
|
|
|
|
|
def test_kerberos_auth_without_keytab(kerberos_cluster):
|
|
assert (
|
|
"DB::Exception: : Authentication failed: password is incorrect, or there is no user with such name."
|
|
in make_auth(instance2)
|
|
)
|
|
|
|
|
|
def test_bad_path_to_keytab(kerberos_cluster):
|
|
assert (
|
|
"DB::Exception: : Authentication failed: password is incorrect, or there is no user with such name."
|
|
in make_auth(instance3)
|
|
)
|
|
assert instance3.contains_in_log("Keytab file not found")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cluster.start()
|
|
input("Cluster created, press any key to destroy...")
|
|
cluster.shutdown()
|