mirror of
https://github.com/apache/impala.git
synced 2026-02-01 12:00:22 -05:00
05dfb208ff
If Sentry authorization is enabled, users with multi group-policies will take time to get the result of 'show tables/databases'. It seems that ResourceAuthorizationProvider.hasAccess performs bad for users with complex group-policies, IMPALA-9242 will target to address this problem. This patch provides a config option 'num_check_authorization_threads' to accelerate 'show tables/databases' by using multithreading. This configuration is applicable only when authorization is enabled. A value of 1 disables multi-threaded execution for checking access. However, a small value of larger than 1 may limit the parallism of FE requests when checking authorization with a high concurrency. The value must be in the range of 1 to 128. The default value of 'num_check_access_threads' is 1. Change-Id: I860e0d18afa0421665f8b3b1c5561d6bdacc5e96 Reviewed-on: http://gerrit.cloudera.org:8080/14846 Reviewed-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com> Tested-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>
678 lines
33 KiB
Python
678 lines
33 KiB
Python
# Licensed to the Apache Software Foundation (ASF) under one
|
|
# or more contributor license agreements. See the NOTICE file
|
|
# distributed with this work for additional information
|
|
# regarding copyright ownership. The ASF licenses this file
|
|
# to you under the Apache License, Version 2.0 (the
|
|
# "License"); you may not use this file except in compliance
|
|
# with the License. You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing,
|
|
# software distributed under the License is distributed on an
|
|
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
|
# KIND, either express or implied. See the License for the
|
|
# specific language governing permissions and limitations
|
|
# under the License.
|
|
#
|
|
# Client tests for SQL statement authorization
|
|
|
|
import os
|
|
import pytest
|
|
import tempfile
|
|
import grp
|
|
import re
|
|
import random
|
|
import sys
|
|
import subprocess
|
|
import urllib
|
|
|
|
from getpass import getuser
|
|
from ImpalaService import ImpalaHiveServer2Service
|
|
from TCLIService import TCLIService
|
|
from thrift.transport.TSocket import TSocket
|
|
from thrift.transport.TTransport import TBufferedTransport
|
|
from thrift.protocol import TBinaryProtocol
|
|
from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
|
|
from tests.common.file_utils import assert_file_in_dir_contains,\
|
|
assert_no_files_in_dir_contain
|
|
|
|
SENTRY_CONFIG_DIR = os.getenv('IMPALA_HOME') + '/fe/src/test/resources/'
|
|
SENTRY_BASE_LOG_DIR = os.getenv('IMPALA_CLUSTER_LOGS_DIR') + "/sentry"
|
|
SENTRY_CONFIG_FILE = SENTRY_CONFIG_DIR + 'sentry-site.xml'
|
|
SENTRY_CONFIG_FILE_OO = SENTRY_CONFIG_DIR + 'sentry-site_oo.xml'
|
|
PRIVILEGES = ['all', 'alter', 'drop', 'insert', 'refresh', 'select']
|
|
ADMIN = "admin"
|
|
|
|
class TestAuthorization(CustomClusterTestSuite):
|
|
def setup(self):
|
|
host, port = (self.cluster.impalads[0].service.hostname,
|
|
self.cluster.impalads[0].service.hs2_port)
|
|
self.socket = TSocket(host, port)
|
|
self.transport = TBufferedTransport(self.socket)
|
|
self.transport.open()
|
|
self.protocol = TBinaryProtocol.TBinaryProtocol(self.transport)
|
|
self.hs2_client = ImpalaHiveServer2Service.Client(self.protocol)
|
|
|
|
def teardown(self):
|
|
if self.socket:
|
|
self.socket.close()
|
|
|
|
def __execute_hs2_stmt(self, statement, verify=True):
|
|
"""
|
|
Executes an hs2 statement
|
|
|
|
:param statement: the statement to execute
|
|
:param verify: If set to true, will thrown an exception on a failed hs2 execution
|
|
:return: the result of execution
|
|
"""
|
|
from tests.hs2.test_hs2 import TestHS2
|
|
execute_statement_req = TCLIService.TExecuteStatementReq()
|
|
execute_statement_req.sessionHandle = self.session_handle
|
|
execute_statement_req.statement = statement
|
|
result = self.hs2_client.ExecuteStatement(execute_statement_req)
|
|
if verify:
|
|
TestHS2.check_response(result)
|
|
return result
|
|
|
|
def __open_hs2(self, user, configuration, verify=True):
|
|
"""
|
|
Open a session with hs2
|
|
|
|
:param user: the user to open the session
|
|
:param configuration: the configuration for the session
|
|
:param verify: If set to true, will thrown an exception on failed session open
|
|
:return: the result of opening the session
|
|
"""
|
|
from tests.hs2.test_hs2 import TestHS2
|
|
open_session_req = TCLIService.TOpenSessionReq()
|
|
open_session_req.username = user
|
|
open_session_req.configuration = configuration
|
|
resp = self.hs2_client.OpenSession(open_session_req)
|
|
if verify:
|
|
TestHS2.check_response(resp)
|
|
return resp
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server_name=server1 "
|
|
"--sentry_config={0} "
|
|
"--authorization_policy_provider_class="
|
|
"org.apache.impala.testutil.TestSentryResourceAuthorizationProvider"
|
|
.format(SENTRY_CONFIG_FILE),
|
|
catalogd_args="--sentry_config={0} "
|
|
"--authorization_policy_provider_class="
|
|
"org.apache.impala.testutil.TestSentryResourceAuthorizationProvider"
|
|
.format(SENTRY_CONFIG_FILE),
|
|
sentry_config=SENTRY_CONFIG_FILE)
|
|
def test_custom_authorization_provider(self, unique_role):
|
|
try:
|
|
self.session_handle = self.__open_hs2(getuser(), dict()).sessionHandle
|
|
self.__execute_hs2_stmt("create role {0}".format(unique_role))
|
|
self.__execute_hs2_stmt("grant role {0} to group {1}"
|
|
.format(unique_role, grp.getgrnam(getuser()).gr_name))
|
|
self.__execute_hs2_stmt("grant select on table tpch.lineitem to role {0}"
|
|
.format(unique_role))
|
|
|
|
bad_resp = self.__execute_hs2_stmt("describe tpch_seq.lineitem", False)
|
|
assert 'User \'%s\' does not have privileges to access' % getuser() in \
|
|
str(bad_resp)
|
|
self.__execute_hs2_stmt("describe tpch.lineitem")
|
|
finally:
|
|
self.__execute_hs2_stmt("drop role {0}".format(unique_role))
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server_name=server1 "
|
|
"--authorized_proxy_user_config=hue={0}".format(getuser()),
|
|
catalogd_args="--sentry_config=" + SENTRY_CONFIG_FILE,
|
|
sentry_config=SENTRY_CONFIG_FILE)
|
|
def test_access_runtime_profile(self, unique_role, unique_name):
|
|
unique_db = unique_name + "_db"
|
|
|
|
try:
|
|
self.session_handle = self.__open_hs2(getuser(), dict()).sessionHandle
|
|
self.__execute_hs2_stmt("create role {0}".format(unique_role))
|
|
self.__execute_hs2_stmt("grant create on server to role {0}".format(unique_role))
|
|
self.__execute_hs2_stmt("grant all on database tpch to role {0}"
|
|
.format(unique_role))
|
|
self.__execute_hs2_stmt("grant select on table functional.complex_view to role {0}"
|
|
.format(unique_role))
|
|
self.__execute_hs2_stmt("grant role {0} to group {1}"
|
|
.format(unique_role, grp.getgrnam(getuser()).gr_name))
|
|
# Create db with permissions
|
|
self.__execute_hs2_stmt("create database {0}".format(unique_db))
|
|
self.__execute_hs2_stmt("grant all on database {0} to role {1}"
|
|
.format(unique_db, unique_role))
|
|
|
|
# Current user can't access view's underlying tables
|
|
bad_resp = self.__execute_hs2_stmt("explain select * from functional.complex_view",
|
|
False)
|
|
assert 'User \'%s\' does not have privileges to EXPLAIN' % getuser() in \
|
|
str(bad_resp)
|
|
# User should not have access to the runtime profile
|
|
self.__run_stmt_and_verify_profile_access("select * from functional.complex_view",
|
|
False, False)
|
|
self.__run_stmt_and_verify_profile_access("select * from functional.complex_view",
|
|
False, True)
|
|
|
|
# Repeat as a delegated user
|
|
self.session_handle = \
|
|
self.__open_hs2('hue', {'impala.doas.user': getuser()}).sessionHandle
|
|
# User should not have access to the runtime profile
|
|
self.__run_stmt_and_verify_profile_access("select * from functional.complex_view",
|
|
False, False)
|
|
self.__run_stmt_and_verify_profile_access("select * from functional.complex_view",
|
|
False, True)
|
|
|
|
# Create a view for which the user has access to the underlying tables.
|
|
self.session_handle = self.__open_hs2(getuser(), dict()).sessionHandle
|
|
self.__execute_hs2_stmt(
|
|
"create view if not exists {0}.customer_view as select * from tpch.customer "
|
|
"limit 1".format(unique_db))
|
|
|
|
# User should be able to run EXPLAIN
|
|
self.__execute_hs2_stmt("explain select * from {0}.customer_view"
|
|
.format(unique_db))
|
|
|
|
# User should have access to the runtime profile and exec summary
|
|
self.__run_stmt_and_verify_profile_access("select * from {0}.customer_view"
|
|
.format(unique_db), True, False)
|
|
self.__run_stmt_and_verify_profile_access("select * from {0}.customer_view"
|
|
.format(unique_db), True, True)
|
|
|
|
# Repeat as a delegated user
|
|
self.session_handle = \
|
|
self.__open_hs2('hue', {'impala.doas.user': getuser()}).sessionHandle
|
|
# Delegated user is the current user
|
|
self.__run_stmt_and_verify_profile_access("select * from {0}.customer_view"
|
|
.format(unique_db), True, False)
|
|
self.__run_stmt_and_verify_profile_access("select * from {0}.customer_view"
|
|
.format(unique_db), True, True)
|
|
finally:
|
|
self.__execute_hs2_stmt("grant all on server to role {0}".format(unique_role))
|
|
self.__execute_hs2_stmt("drop view if exists {0}.customer_view".format(unique_db))
|
|
self.__execute_hs2_stmt("drop table if exists {0}.customer".format(unique_db))
|
|
self.__execute_hs2_stmt("drop database if exists {0}".format(unique_db))
|
|
self.__execute_hs2_stmt("drop role {0}".format(unique_role))
|
|
|
|
def __run_stmt_and_verify_profile_access(self, stmt, has_access, close_operation):
|
|
"""Runs 'stmt' and retrieves the runtime profile and exec summary. If
|
|
'has_access' is true, it verifies that no runtime profile or exec summary are
|
|
returned. If 'close_operation' is true, make sure the operation is closed before
|
|
retrieving the profile and exec summary."""
|
|
from tests.hs2.test_hs2 import TestHS2
|
|
execute_statement_resp = self.__execute_hs2_stmt(stmt, False)
|
|
|
|
if close_operation:
|
|
close_operation_req = TCLIService.TCloseOperationReq()
|
|
close_operation_req.operationHandle = execute_statement_resp.operationHandle
|
|
TestHS2.check_response(self.hs2_client.CloseOperation(close_operation_req))
|
|
|
|
get_profile_req = ImpalaHiveServer2Service.TGetRuntimeProfileReq()
|
|
get_profile_req.operationHandle = execute_statement_resp.operationHandle
|
|
get_profile_req.sessionHandle = self.session_handle
|
|
get_profile_resp = self.hs2_client.GetRuntimeProfile(get_profile_req)
|
|
|
|
if has_access:
|
|
TestHS2.check_response(get_profile_resp)
|
|
assert "Plan: " in get_profile_resp.profile
|
|
else:
|
|
assert "User %s is not authorized to access the runtime profile or "\
|
|
"execution summary." % (getuser()) in str(get_profile_resp)
|
|
|
|
exec_summary_req = ImpalaHiveServer2Service.TGetExecSummaryReq()
|
|
exec_summary_req.operationHandle = execute_statement_resp.operationHandle
|
|
exec_summary_req.sessionHandle = self.session_handle
|
|
exec_summary_resp = self.hs2_client.GetExecSummary(exec_summary_req)
|
|
|
|
if has_access:
|
|
TestHS2.check_response(exec_summary_resp)
|
|
else:
|
|
assert "User %s is not authorized to access the runtime profile or "\
|
|
"execution summary." % (getuser()) in str(exec_summary_resp)
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server_name=server1 --sentry_config=" + SENTRY_CONFIG_FILE,
|
|
catalogd_args="--sentry_config=" + SENTRY_CONFIG_FILE,
|
|
impala_log_dir=tempfile.mkdtemp(prefix="test_deprecated_none_",
|
|
dir=os.getenv("LOG_DIR")))
|
|
def test_deprecated_flag_doesnt_show(self):
|
|
assert_no_files_in_dir_contain(self.impala_log_dir, "Ignoring removed flag "
|
|
"authorization_policy_file")
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args("--server_name=server1\
|
|
--authorization_policy_file=ignored_file",
|
|
impala_log_dir=tempfile.mkdtemp(prefix="test_deprecated_",
|
|
dir=os.getenv("LOG_DIR")))
|
|
def test_deprecated_flags(self):
|
|
assert_file_in_dir_contains(self.impala_log_dir, "Ignoring removed flag "
|
|
"authorization_policy_file")
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server_name=server1 --sentry_config=%s" % SENTRY_CONFIG_FILE,
|
|
catalogd_args="--sentry_config=%s" % SENTRY_CONFIG_FILE,
|
|
impala_log_dir=tempfile.mkdtemp(prefix="test_catalog_restart_",
|
|
dir=os.getenv("LOG_DIR")))
|
|
def test_catalog_restart(self, unique_role):
|
|
"""IMPALA-7713: Tests that a catalogd restart when authorization is enabled should
|
|
reset the previous privileges stored in impalad's catalog to avoid stale privilege
|
|
data in the impalad's catalog."""
|
|
def assert_privileges():
|
|
result = self.client.execute("show grant role %s_foo" % unique_role)
|
|
TestAuthorization._check_privileges(result, [["database", "functional",
|
|
"", "", "", "all", "false"]])
|
|
|
|
result = self.client.execute("show grant role %s_bar" % unique_role)
|
|
TestAuthorization._check_privileges(result, [["database", "functional_kudu",
|
|
"", "", "", "all", "false"]])
|
|
|
|
result = self.client.execute("show grant role %s_baz" % unique_role)
|
|
TestAuthorization._check_privileges(result, [["database", "functional_avro",
|
|
"", "", "", "all", "false"]])
|
|
|
|
self.role_cleanup(unique_role)
|
|
try:
|
|
self.client.execute("create role %s_foo" % unique_role)
|
|
self.client.execute("create role %s_bar" % unique_role)
|
|
self.client.execute("create role %s_baz" % unique_role)
|
|
self.client.execute("grant all on database functional to role %s_foo" %
|
|
unique_role)
|
|
self.client.execute("grant all on database functional_kudu to role %s_bar" %
|
|
unique_role)
|
|
self.client.execute("grant all on database functional_avro to role %s_baz" %
|
|
unique_role)
|
|
|
|
assert_privileges()
|
|
self._start_impala_cluster(["--catalogd_args=--sentry_config=%s" %
|
|
SENTRY_CONFIG_FILE, "--restart_catalogd_only"])
|
|
assert_privileges()
|
|
finally:
|
|
self.role_cleanup(unique_role)
|
|
|
|
def role_cleanup(self, role_name_match):
|
|
"""Cleans up any roles that match the given role name."""
|
|
for role_name in self.client.execute("show roles").data:
|
|
if role_name_match in role_name:
|
|
self.client.execute("drop role %s" % role_name)
|
|
|
|
@staticmethod
|
|
def _check_privileges(result, expected):
|
|
def columns(row):
|
|
cols = row.split("\t")
|
|
return cols[0:len(cols) - 1]
|
|
assert map(columns, result.data) == expected
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server_name=server1 --sentry_config=%s" % SENTRY_CONFIG_FILE,
|
|
catalogd_args="--sentry_config=%s" % SENTRY_CONFIG_FILE,
|
|
impala_log_dir=tempfile.mkdtemp(prefix="test_catalog_restart_",
|
|
dir=os.getenv("LOG_DIR")))
|
|
def test_catalog_object(self, unique_role):
|
|
"""IMPALA-7721: Tests /catalog_object web API for principal and privilege"""
|
|
self.role_cleanup(unique_role)
|
|
try:
|
|
self.client.execute("create role %s" % unique_role)
|
|
self.client.execute("grant select on database functional to role %s" % unique_role)
|
|
for service in [self.cluster.catalogd.service,
|
|
self.cluster.get_first_impalad().service]:
|
|
obj_dump = service.get_catalog_object_dump("PRINCIPAL", "%s.ROLE" % unique_role)
|
|
assert "catalog_version" in obj_dump
|
|
|
|
# Get the privilege associated with that principal ID.
|
|
principal_id = re.search(r"principal_id \(i32\) = (\d+)", obj_dump)
|
|
assert principal_id is not None
|
|
obj_dump = service.get_catalog_object_dump("PRIVILEGE", urllib.quote(
|
|
"server=server1->db=functional->action=select->grantoption=false.%s.ROLE" %
|
|
principal_id.group(1)))
|
|
assert "catalog_version" in obj_dump
|
|
|
|
# Get the principal that does not exist.
|
|
obj_dump = service.get_catalog_object_dump("PRINCIPAL", "doesnotexist.ROLE")
|
|
assert "CatalogException" in obj_dump
|
|
|
|
# Get the privilege that does not exist.
|
|
obj_dump = service.get_catalog_object_dump("PRIVILEGE", urllib.quote(
|
|
"server=server1->db=doesntexist->action=select->grantoption=false.%s.ROLE" %
|
|
principal_id.group(1)))
|
|
assert "CatalogException" in obj_dump
|
|
finally:
|
|
self.role_cleanup(unique_role)
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server_name=server1 --sentry_config=%s" % SENTRY_CONFIG_FILE,
|
|
catalogd_args="--sentry_config=%s --sentry_catalog_polling_frequency_s=3600" %
|
|
SENTRY_CONFIG_FILE,
|
|
impala_log_dir=tempfile.mkdtemp(prefix="test_invalidate_metadata_sentry_unavailable_",
|
|
dir=os.getenv("LOG_DIR")))
|
|
def test_invalidate_metadata_sentry_unavailable(self, unique_role):
|
|
"""IMPALA-7824: Tests that running INVALIDATE METADATA when Sentry is unavailable
|
|
should not cause Impala to hang."""
|
|
self.role_cleanup(unique_role)
|
|
try:
|
|
group_name = grp.getgrnam(getuser()).gr_name
|
|
self.client.execute("create role %s" % unique_role)
|
|
self.client.execute("grant all on server to role %s" % unique_role)
|
|
self.client.execute("grant role %s to group `%s`" % (unique_role, group_name))
|
|
|
|
self._stop_sentry_service()
|
|
# Calling INVALIDATE METADATA when Sentry is unavailable should return an error.
|
|
result = self.execute_query_expect_failure(self.client, "invalidate metadata")
|
|
result_str = str(result)
|
|
assert "MESSAGE: CatalogException: Error refreshing authorization policy:" \
|
|
in result_str
|
|
assert "CAUSED BY: ImpalaRuntimeException: Error refreshing authorization policy." \
|
|
" Sentry is unavailable. Ensure Sentry is up:" in result_str
|
|
|
|
self._start_sentry_service(SENTRY_CONFIG_FILE)
|
|
# Calling INVALIDATE METADATA after Sentry is up should not return an error.
|
|
self.execute_query_expect_success(self.client, "invalidate metadata")
|
|
finally:
|
|
self.role_cleanup(unique_role)
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server_name=server1 --sentry_config=%s" % SENTRY_CONFIG_FILE,
|
|
catalogd_args="--sentry_config=%s --sentry_catalog_polling_frequency_s=3600 " %
|
|
SENTRY_CONFIG_FILE,
|
|
impala_log_dir=tempfile.mkdtemp(prefix="test_refresh_authorization_",
|
|
dir=os.getenv("LOG_DIR")))
|
|
def test_refresh_authorization(self, unique_role):
|
|
"""Tests refresh authorization statement by adding and removing roles and privileges
|
|
externally. The long Sentry polling is used so that any authorization metadata
|
|
updated externally does not get polled by Impala in order to test an an explicit
|
|
call to refresh authorization statement."""
|
|
group_name = grp.getgrnam(getuser()).gr_name
|
|
self.role_cleanup(unique_role)
|
|
for sync_ddl in [1, 0]:
|
|
query_options = {'sync_ddl': sync_ddl}
|
|
clients = []
|
|
if sync_ddl:
|
|
# When sync_ddl is True, we want to ensure the changes are propagated to all
|
|
# coordinators.
|
|
for impalad in self.cluster.impalads:
|
|
clients.append(impalad.service.create_beeswax_client())
|
|
else:
|
|
clients.append(self.client)
|
|
try:
|
|
self.client.execute("create role %s" % unique_role)
|
|
self.client.execute("grant role %s to group `%s`" % (unique_role, group_name))
|
|
self.client.execute("grant refresh on server to %s" % unique_role)
|
|
|
|
self.validate_refresh_authorization_roles(unique_role, query_options, clients)
|
|
self.validate_refresh_authorization_privileges(unique_role, query_options,
|
|
clients)
|
|
finally:
|
|
self.role_cleanup(unique_role)
|
|
|
|
def validate_refresh_authorization_roles(self, unique_role, query_options, clients):
|
|
"""This method tests refresh authorization statement by adding and removing
|
|
roles externally."""
|
|
try:
|
|
# Create two roles inside Impala.
|
|
self.client.execute("create role %s_internal1" % unique_role)
|
|
self.client.execute("create role %s_internal2" % unique_role)
|
|
# Drop an existing role (_internal1) outside Impala.
|
|
role = "%s_internal1" % unique_role
|
|
subprocess.check_call(
|
|
["/bin/bash", "-c",
|
|
"%s/bin/sentryShell --conf %s/sentry-site.xml -dr -r %s" %
|
|
(os.getenv("SENTRY_HOME"), os.getenv("SENTRY_CONF_DIR"), role)],
|
|
stdout=sys.stdout, stderr=sys.stderr)
|
|
|
|
result = self.execute_query_expect_success(self.client, "show roles")
|
|
assert any(role in x for x in result.data)
|
|
self.execute_query_expect_success(self.client, "refresh authorization",
|
|
query_options=query_options)
|
|
for client in clients:
|
|
result = self.execute_query_expect_success(client, "show roles")
|
|
assert not any(role in x for x in result.data)
|
|
|
|
# Add a new role outside Impala.
|
|
role = "%s_external" % unique_role
|
|
subprocess.check_call(
|
|
["/bin/bash", "-c",
|
|
"%s/bin/sentryShell --conf %s/sentry-site.xml -cr -r %s" %
|
|
(os.getenv("SENTRY_HOME"), os.getenv("SENTRY_CONF_DIR"), role)],
|
|
stdout=sys.stdout, stderr=sys.stderr)
|
|
|
|
result = self.execute_query_expect_success(self.client, "show roles")
|
|
assert not any(role in x for x in result.data)
|
|
self.execute_query_expect_success(self.client, "refresh authorization",
|
|
query_options=query_options)
|
|
for client in clients:
|
|
result = self.execute_query_expect_success(client, "show roles")
|
|
assert any(role in x for x in result.data)
|
|
finally:
|
|
for suffix in ["internal1", "internal2", "external"]:
|
|
self.role_cleanup("%s_%s" % (unique_role, suffix))
|
|
|
|
def validate_refresh_authorization_privileges(self, unique_role, query_options,
|
|
clients):
|
|
"""This method tests refresh authorization statement by adding and removing
|
|
privileges externally."""
|
|
# Grant select privilege outside Impala.
|
|
subprocess.check_call(
|
|
["/bin/bash", "-c",
|
|
"%s/bin/sentryShell --conf %s/sentry-site.xml -gpr -p "
|
|
"'server=server1->db=functional->table=alltypes->action=select' -r %s" %
|
|
(os.getenv("SENTRY_HOME"), os.getenv("SENTRY_CONF_DIR"), unique_role)],
|
|
stdout=sys.stdout, stderr=sys.stderr)
|
|
|
|
# Before refresh authorization, there should only be one refresh privilege.
|
|
result = self.execute_query_expect_success(self.client, "show grant role %s" %
|
|
unique_role)
|
|
assert len(result.data) == 1
|
|
assert any("refresh" in x for x in result.data)
|
|
|
|
for client in clients:
|
|
self.execute_query_expect_failure(client,
|
|
"select * from functional.alltypes limit 1")
|
|
|
|
self.execute_query_expect_success(self.client, "refresh authorization",
|
|
query_options=query_options)
|
|
|
|
for client in clients:
|
|
# Ensure select privilege was granted after refresh authorization.
|
|
result = self.execute_query_expect_success(client, "show grant role %s" %
|
|
unique_role)
|
|
assert len(result.data) == 2
|
|
assert any("select" in x for x in result.data)
|
|
assert any("refresh" in x for x in result.data)
|
|
self.execute_query_expect_success(client,
|
|
"select * from functional.alltypes limit 1")
|
|
|
|
@staticmethod
|
|
def _verify_show_dbs(result, unique_name, visibility_privileges=PRIVILEGES):
|
|
""" Helper function for verifying the results of SHOW DATABASES below.
|
|
Only show databases with privileges implying any of the visibility_privileges.
|
|
"""
|
|
for priv in PRIVILEGES:
|
|
# Result lines are in the format of "db_name\tdb_comment"
|
|
db_name = 'db_%s_%s\t' % (unique_name, priv)
|
|
if priv != 'all' and priv not in visibility_privileges:
|
|
assert db_name not in result.data
|
|
else:
|
|
assert db_name in result.data
|
|
|
|
def _test_sentry_show_stmts_helper(self, unique_role, unique_name,
|
|
visibility_privileges):
|
|
unique_db = unique_name + "_db"
|
|
# TODO: can we create and use a temp username instead of using root?
|
|
another_user = 'root'
|
|
another_user_grp = 'root'
|
|
self.role_cleanup(unique_role)
|
|
try:
|
|
self.client.execute("create role %s" % unique_role)
|
|
self.client.execute("grant create on server to role %s" % unique_role)
|
|
self.client.execute("grant drop on server to role %s" % unique_role)
|
|
self.client.execute("grant role %s to group %s" %
|
|
(unique_role, grp.getgrnam(getuser()).gr_name))
|
|
|
|
self.client.execute("drop database if exists %s cascade" % unique_db)
|
|
self.client.execute("create database %s" % unique_db)
|
|
for priv in PRIVILEGES:
|
|
self.client.execute("create database db_%s_%s" % (unique_name, priv))
|
|
self.client.execute("grant {0} on database db_{1}_{2} to role {3}"
|
|
.format(priv, unique_name, priv, unique_role))
|
|
self.client.execute("create table %s.tbl_%s (i int)" % (unique_db, priv))
|
|
self.client.execute("grant {0} on table {1}.tbl_{2} to role {3}"
|
|
.format(priv, unique_db, priv, unique_role))
|
|
self.client.execute("grant role %s to group %s" %
|
|
(unique_role, another_user_grp))
|
|
|
|
# Owner (current user) can still see all the owned databases and tables
|
|
result = self.client.execute("show databases")
|
|
TestAuthorization._verify_show_dbs(result, unique_name)
|
|
result = self.client.execute("show tables in %s" % unique_db)
|
|
assert result.data == ["tbl_%s" % p for p in PRIVILEGES]
|
|
|
|
# Check SHOW DATABASES and SHOW TABLES using another username
|
|
# Create another client so we can user another username
|
|
root_impalad_client = self.create_impala_client()
|
|
result = self.execute_query_expect_success(
|
|
root_impalad_client, "show databases", user=another_user)
|
|
TestAuthorization._verify_show_dbs(result, unique_name, visibility_privileges)
|
|
result = self.execute_query_expect_success(
|
|
root_impalad_client, "show tables in %s" % unique_db, user=another_user)
|
|
# Only show tables with privileges implying any of the visibility privileges
|
|
assert 'tbl_all' in result.data # ALL can imply to any privilege
|
|
for p in visibility_privileges:
|
|
assert 'tbl_%s' % p in result.data
|
|
finally:
|
|
self.client.execute("drop database if exists %s cascade" % unique_db)
|
|
for priv in PRIVILEGES:
|
|
self.client.execute(
|
|
"drop database if exists db_%s_%s cascade" % (unique_name, priv))
|
|
self.role_cleanup(unique_role)
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server_name=server1 --sentry_config=%s "
|
|
"--authorized_proxy_user_config=%s=* "
|
|
"--min_privilege_set_for_show_stmts=select" %
|
|
(SENTRY_CONFIG_FILE, getuser()),
|
|
catalogd_args="--sentry_config={0}".format(SENTRY_CONFIG_FILE),
|
|
sentry_config=SENTRY_CONFIG_FILE_OO, # Enable Sentry Object Ownership
|
|
sentry_log_dir="{0}/test_sentry_show_stmts_with_select".format(SENTRY_BASE_LOG_DIR))
|
|
def test_sentry_show_stmts_with_select(self, unique_role, unique_name):
|
|
self._test_sentry_show_stmts_helper(unique_role, unique_name, ['select'])
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server_name=server1 --sentry_config=%s "
|
|
"--authorized_proxy_user_config=%s=* "
|
|
"--min_privilege_set_for_show_stmts=select,insert" %
|
|
(SENTRY_CONFIG_FILE, getuser()),
|
|
catalogd_args="--sentry_config={0}".format(SENTRY_CONFIG_FILE),
|
|
sentry_config=SENTRY_CONFIG_FILE_OO, # Enable Sentry Object Ownership
|
|
sentry_log_dir="{0}/test_sentry_show_stmts_with_select_insert"
|
|
.format(SENTRY_BASE_LOG_DIR))
|
|
def test_sentry_show_stmts_with_select_insert(self, unique_role, unique_name):
|
|
self._test_sentry_show_stmts_helper(unique_role, unique_name,
|
|
['select', 'insert'])
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server_name=server1 --sentry_config=%s "
|
|
"--authorized_proxy_user_config=%s=* "
|
|
"--min_privilege_set_for_show_stmts=any" %
|
|
(SENTRY_CONFIG_FILE, getuser()),
|
|
catalogd_args="--sentry_config={0}".format(SENTRY_CONFIG_FILE),
|
|
sentry_config=SENTRY_CONFIG_FILE_OO, # Enable Sentry Object Ownership
|
|
sentry_log_dir="{0}/test_sentry_show_stmts_with_any".format(SENTRY_BASE_LOG_DIR))
|
|
def test_sentry_show_stmts_with_any(self, unique_role, unique_name):
|
|
self._test_sentry_show_stmts_helper(unique_role, unique_name, PRIVILEGES)
|
|
|
|
def _test_ranger_show_stmts_helper(self, unique_name, visibility_privileges):
|
|
unique_db = unique_name + "_db"
|
|
admin_client = self.create_impala_client()
|
|
try:
|
|
admin_client.execute("drop database if exists %s cascade" % unique_db, user=ADMIN)
|
|
admin_client.execute("create database %s" % unique_db, user=ADMIN)
|
|
for priv in PRIVILEGES:
|
|
admin_client.execute("create database db_%s_%s" % (unique_name, priv))
|
|
admin_client.execute("grant {0} on database db_{1}_{2} to user {3}"
|
|
.format(priv, unique_name, priv, getuser()))
|
|
admin_client.execute("create table %s.tbl_%s (i int)" % (unique_db, priv))
|
|
admin_client.execute("grant {0} on table {1}.tbl_{2} to user {3}"
|
|
.format(priv, unique_db, priv, getuser()))
|
|
|
|
# Admin can still see all the databases and tables
|
|
result = admin_client.execute("show databases")
|
|
TestAuthorization._verify_show_dbs(result, unique_name)
|
|
result = admin_client.execute("show tables in %s" % unique_db)
|
|
assert result.data == ["tbl_%s" % p for p in PRIVILEGES]
|
|
|
|
# Check SHOW DATABASES and SHOW TABLES using another username
|
|
result = self.client.execute("show databases")
|
|
TestAuthorization._verify_show_dbs(result, unique_name, visibility_privileges)
|
|
result = self.client.execute("show tables in %s" % unique_db)
|
|
# Only show tables with privileges implying any of the visibility privileges
|
|
assert 'tbl_all' in result.data # ALL can imply to any privilege
|
|
for p in visibility_privileges:
|
|
assert 'tbl_%s' % p in result.data
|
|
finally:
|
|
admin_client.execute("drop database if exists %s cascade" % unique_db)
|
|
for priv in PRIVILEGES:
|
|
admin_client.execute(
|
|
"drop database if exists db_%s_%s cascade" % (unique_name, priv))
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server-name=server1 --ranger_service_type=hive "
|
|
"--ranger_app_id=impala --authorization_provider=ranger "
|
|
"--min_privilege_set_for_show_stmts=select",
|
|
catalogd_args="--server-name=server1 --ranger_service_type=hive "
|
|
"--ranger_app_id=impala --authorization_provider=ranger")
|
|
def test_ranger_show_stmts_with_select(self, unique_name):
|
|
self._test_ranger_show_stmts_helper(unique_name, ['select'])
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server-name=server1 --ranger_service_type=hive "
|
|
"--ranger_app_id=impala --authorization_provider=ranger "
|
|
"--min_privilege_set_for_show_stmts=select,insert",
|
|
catalogd_args="--server-name=server1 --ranger_service_type=hive "
|
|
"--ranger_app_id=impala --authorization_provider=ranger")
|
|
def test_ranger_show_stmts_with_select_insert(self, unique_name):
|
|
self._test_ranger_show_stmts_helper(unique_name, ['select', 'insert'])
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server-name=server1 --ranger_service_type=hive "
|
|
"--ranger_app_id=impala --authorization_provider=ranger "
|
|
"--min_privilege_set_for_show_stmts=any",
|
|
catalogd_args="--server-name=server1 --ranger_service_type=hive "
|
|
"--ranger_app_id=impala --authorization_provider=ranger")
|
|
def test_ranger_show_stmts_with_any(self, unique_name):
|
|
self._test_ranger_show_stmts_helper(unique_name, PRIVILEGES)
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server-name=server1 --ranger_service_type=hive "
|
|
"--ranger_app_id=impala --authorization_provider=ranger "
|
|
"--num_check_authorization_threads=%d" % (random.randint(2, 128)),
|
|
catalogd_args="--server-name=server1 --ranger_service_type=hive "
|
|
"--ranger_app_id=impala --authorization_provider=ranger")
|
|
def test_num_check_authorization_threads_with_ranger(self, unique_name):
|
|
self._test_ranger_show_stmts_helper(unique_name, PRIVILEGES)
|
|
|
|
@pytest.mark.execute_serially
|
|
@CustomClusterTestSuite.with_args(
|
|
impalad_args="--server_name=server1 --sentry_config=%s "
|
|
"--authorized_proxy_user_config=%s=* "
|
|
"--num_check_authorization_threads=%d" %
|
|
(SENTRY_CONFIG_FILE, getuser(), random.randint(2, 128)),
|
|
catalogd_args="--sentry_config={0}".format(SENTRY_CONFIG_FILE),
|
|
sentry_config=SENTRY_CONFIG_FILE_OO, # Enable Sentry Object Ownership
|
|
sentry_log_dir="{0}/test_num_check_authorization_threads_with_sentry"
|
|
.format(SENTRY_BASE_LOG_DIR))
|
|
def test_num_check_authorization_threads_with_sentry(self, unique_role, unique_name):
|
|
self._test_sentry_show_stmts_helper(unique_role, unique_name, PRIVILEGES)
|