This is an automated email from the ASF dual-hosted git repository.
pradeepagrawal8184 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ranger.git
The following commit(s) were added to refs/heads/master by this push:
new 3bc00377d RANGER-5862 - Updated the tests and did functional test
refactoring (#1342)
3bc00377d is described below
commit 3bc00377db985c696548dd55af5a70dfd5445ac3
Author: harshal1172 <[email protected]>
AuthorDate: Fri Oct 30 16:27:22 2026 +0530
RANGER-5862 - Updated the tests and did functional test refactoring (#1342)
---
functional-tests/conftest.py | 15 +
functional-tests/hbase/test_hbase_01.py | 97 ++---
functional-tests/hive/test_hive_01.py | 160 +++-----
.../ranger_test_utils/access_audit_utils.py | 161 ++++++++-
functional-tests/ranger_test_utils/hbase_utils.py | 248 ++++++++++---
functional-tests/ranger_test_utils/hive_utils.py | 246 ++++++++++---
.../ranger_test_utils/plugin_conftest.py | 146 ++++++++
.../ranger_test_utils/plugin_enforcement_utils.py | 61 ++++
.../ranger_test_utils/plugin_test_users.py | 401 +++++++++++++++++++++
functional-tests/ranger_test_utils/utils.py | 43 ++-
functional-tests/readme.md | 1 +
11 files changed, 1313 insertions(+), 266 deletions(-)
diff --git a/functional-tests/conftest.py b/functional-tests/conftest.py
index 1d7c35856..79e415b48 100644
--- a/functional-tests/conftest.py
+++ b/functional-tests/conftest.py
@@ -20,6 +20,7 @@
import pytest
+from ranger_test_utils.plugin_conftest import
ensure_plugin_test_users_for_session
from ranger_test_utils.utils import LOGS_DIR, ensure_logs_directory
LOG_FORMAT = "%(asctime)s %(levelname)s [%(name)s] %(message)s"
@@ -38,6 +39,20 @@ def pytest_configure(config):
config.option.log_file = None
[email protected](scope="session", autouse=True)
+def _ensure_plugin_test_users_for_enforcement_suites(request):
+ """
+ Before plugin enforcement tests (markers: hive, hbase, kafka, hdfs_policy),
+ create shared Ranger xusers and Kerberos keytabs via plugin_conftest.
+ REST-only suites (api_test, kms, …) are unchanged.
+ """
+ if not request.session.items:
+ yield
+ return
+ ensure_plugin_test_users_for_session(request.session)
+ yield
+
+
@pytest.fixture(scope="module", autouse=True)
def _log_to_module_file(request):
"""Write logger.info (and above) to logs/<test_file_name>.log for each
test module."""
diff --git a/functional-tests/hbase/test_hbase_01.py
b/functional-tests/hbase/test_hbase_01.py
index a6739f1db..735cb642c 100644
--- a/functional-tests/hbase/test_hbase_01.py
+++ b/functional-tests/hbase/test_hbase_01.py
@@ -14,13 +14,10 @@
# limitations under the License.
"""
-Authorization is asserted via HBase shell output only. Access-audit asserts
are deferred until
-HBase plugin audits are reliably indexed in Docker — use
ranger_test_utils.access_audit_utils
-when ready.
+HBase plugin enforcement via HBase shell and optional access-audit validation.
-AUDIT_FILTERS (future):
- test_01_01: accessType=createTable, accessResult=0, aclEnforcer=ranger-acl,
- resourceName=<table>, repoName=dev_hbase, requestUser=hrt_21
+Set ``AUDIT_INDEX_STORE`` to ``opensearch`` or ``solr`` when starting Docker
+(see functional-tests readme) so access audits are indexed and asserted.
"""
import pytest
@@ -28,35 +25,30 @@
from ranger_test_utils.hbase_utils import (
HBASE_AUTHZ_DENIED_SUBSTR,
HBASE_SERVICE_NAME,
- HBASE_TEST_USER,
+ HBasePluginTest,
+ HBasePluginTestContext,
cleanup_hbase_test_table,
create_audit_only_policy,
- ensure_kerberos_user,
preflight_hbase_stack,
prepare_hbase_test_table,
- run_hbase_shell,
wait_for_hbase_policy_propagation,
)
from ranger_test_utils.hive_utils import delete_policies_by_id,
set_all_policies_enabled
-from ranger_test_utils.utils import (
- DOCKER_PLUGIN_TEST_USER_PASSWORD,
+from ranger_test_utils.plugin_enforcement_utils import (
configure_test_logging,
create_ranger_admin_session,
- ensure_user_exists,
- get_test_logger,
- log_testcase_begin_for_pytest,
- unique_suffix,
+ logger,
)
pytestmark = pytest.mark.hbase
-logger = get_test_logger(__name__)
+TEST_CASE = None
+
+HBASE_TEST_TABLE = "iemployee"
_admin_session = None
_audit_policy_id = None
_audit_suffix = None
-_run_suffix = None
-_test_table = None
def setup_module():
@@ -64,24 +56,18 @@ def setup_module():
logger.info("=== setup_module: start ===")
preflight_hbase_stack()
- global _admin_session, _audit_policy_id, _audit_suffix
- global _run_suffix, _test_table
+ global TEST_CASE, _admin_session, _audit_policy_id, _audit_suffix
+ TEST_CASE = HBasePluginTestContext()
_admin_session = create_ranger_admin_session()
- _run_suffix = unique_suffix(4)
- _test_table = "iemployee_" + _run_suffix
- logger.info("Test table: %s", _test_table)
-
- logger.info("Ensuring Ranger user and Kerberos principal for %s",
HBASE_TEST_USER)
- ensure_user_exists(
- _admin_session,
- HBASE_TEST_USER,
- password=DOCKER_PLUGIN_TEST_USER_PASSWORD,
- last_name="hive_test",
+ TEST_CASE.admin_session = _admin_session
+
+ logger.info(
+ "Plugin user %s (Ranger + Kerberos) ensured by
functional-tests/conftest.py",
+ HBasePluginTestContext.USER_USER1,
)
- ensure_kerberos_user(HBASE_TEST_USER)
- logger.info("HBase setup: drop table %s if present", _test_table)
- prepare_hbase_test_table(_test_table)
+ logger.info("HBase setup: drop table %s if present", HBASE_TEST_TABLE)
+ prepare_hbase_test_table(HBASE_TEST_TABLE)
logger.info(
"Disabling all policies on %s and creating audit-only policy",
@@ -99,9 +85,8 @@ def teardown_module():
return
logger.info("=== teardown_module: start ===")
- if _test_table is not None:
- logger.info("HBase cleanup: drop table %s", _test_table)
- cleanup_hbase_test_table(_test_table)
+ logger.info("HBase cleanup: drop table %s", HBASE_TEST_TABLE)
+ cleanup_hbase_test_table(HBASE_TEST_TABLE)
if _audit_policy_id is not None:
logger.info("Deleting audit policy id=%s", _audit_policy_id)
@@ -113,26 +98,20 @@ def teardown_module():
logger.info("=== teardown_module: done ===")
-def test_01_01_create_hbase_table_case1a_deny(request):
- log_testcase_begin_for_pytest(logger, request.node.name, suite="HBASE")
- create_cmd = (
- "create '"
- + _test_table
- + "','personal','payroll','medical'"
- )
- logger.info(
- "Running HBase CREATE TABLE as %s (expect deny): %s",
- HBASE_TEST_USER,
- create_cmd,
- )
- exit_code, output = run_hbase_shell(
- HBASE_TEST_USER,
- [create_cmd],
- expect_substring=HBASE_AUTHZ_DENIED_SUBSTR,
- )
- assert HBASE_AUTHZ_DENIED_SUBSTR in output, (
- "Expected authorization denial in HBase shell output; exit_code="
- + str(exit_code)
- + " output="
- + output
- )
+def test_01_01_CreateHbaseTableCase1A():
+ HBasePluginTest(
+ "HBASE.test_01_01_CreateHbaseTableCase1A",
+ TEST_CASE,
+ hBaseCmds=["create 'iemployee','personal','payroll','medical'"],
+ reqUser=HBasePluginTestContext.USER_USER1,
+ verifyCriteria=HBasePluginTest.STDOUT_CONTAINS,
+ verifyStr=HBASE_AUTHZ_DENIED_SUBSTR,
+ checkAuditLog=False,
+ newAuditLogMustExist=True,
+ auditLogFilter={
+ "accessType": TEST_CASE.access_type["CREATE_TABLE"],
+ "accessResult": TEST_CASE.result_denied,
+ "accessEnforcer": "ranger-acl",
+ "resourceName": "iemployee",
+ },
+ ).execute()
diff --git a/functional-tests/hive/test_hive_01.py
b/functional-tests/hive/test_hive_01.py
index 525133963..73b5debcb 100644
--- a/functional-tests/hive/test_hive_01.py
+++ b/functional-tests/hive/test_hive_01.py
@@ -14,15 +14,10 @@
# limitations under the License.
"""
-Authorization is asserted via beeline output only. Access-audit asserts are
deferred until
-Hive plugin audits are reliably indexed in Docker — use
ranger_test_utils.access_audit_utils
-when ready.
-
-AUDIT_FILTERS (future):
- test_01_01: accessType=CREATE, accessResult=0, aclEnforcer=ranger-acl,
- resourcePath=<db>/<table>, repoName=dev_hive, requestUser=hrt_21
- test_01_02: accessType=USE, accessResult=0, aclEnforcer=ranger-acl,
- resourcePath=<db>, repoName=dev_hive, requestUser=hrt_21
+Hive plugin enforcement via beeline and optional access-audit validation.
+
+Set ``AUDIT_INDEX_STORE`` to ``opensearch`` or ``solr`` when starting Docker
+(see functional-tests readme) so access audits are indexed and asserted.
"""
import pytest
@@ -30,37 +25,28 @@
from ranger_test_utils.hive_utils import (
HIVE_AUTHZ_DENIED_SUBSTR,
HIVE_SERVICE_NAME,
- HIVE_TEST_USER,
- cleanup_hive_test_namespace,
+ HivePluginTest,
+ HivePluginTestContext,
create_audit_only_policy,
delete_policies_by_id,
- ensure_kerberos_user,
preflight_hive_stack,
- prepare_hive_test_namespace,
run_beeline,
set_all_policies_enabled,
wait_for_policy_propagation,
)
-from ranger_test_utils.utils import (
- DOCKER_PLUGIN_TEST_USER_PASSWORD,
+from ranger_test_utils.plugin_enforcement_utils import (
configure_test_logging,
create_ranger_admin_session,
- ensure_user_exists,
- get_test_logger,
- log_testcase_begin_for_pytest,
- unique_suffix,
+ logger,
)
pytestmark = pytest.mark.hive
-logger = get_test_logger(__name__)
+TEST_CASE = None
_admin_session = None
_audit_policy_id = None
_audit_suffix = None
-_run_suffix = None
-_test_database = None
-_test_table = None
def setup_module():
@@ -68,34 +54,18 @@ def setup_module():
logger.info("=== setup_module: start ===")
preflight_hive_stack()
- global _admin_session, _audit_policy_id, _audit_suffix
- global _run_suffix, _test_database, _test_table
+ global TEST_CASE, _admin_session, _audit_policy_id, _audit_suffix
+ TEST_CASE = HivePluginTestContext()
_admin_session = create_ranger_admin_session()
- _run_suffix = unique_suffix(4)
- _test_database = "db_" + _run_suffix
- _test_table = "employee_" + _run_suffix
- logger.info(
- "Test resources: database=%s table=%s",
- _test_database,
- _test_table,
- )
-
- logger.info("Ensuring Ranger user and Kerberos principal for %s",
HIVE_TEST_USER)
- ensure_user_exists(
- _admin_session,
- HIVE_TEST_USER,
- password=DOCKER_PLUGIN_TEST_USER_PASSWORD,
- last_name="hive_test",
- )
- ensure_kerberos_user(HIVE_TEST_USER)
+ TEST_CASE.admin_session = _admin_session
logger.info(
- "Hive service setup: create database %s and drop table %s.%s if
present",
- _test_database,
- _test_database,
- _test_table,
+ "Plugin user %s (Ranger + Kerberos) ensured by
functional-tests/conftest.py",
+ HivePluginTestContext.USER_USER1,
)
- prepare_hive_test_namespace(_test_database, _test_table)
+
+ logger.info("Hive service setup: drop default.employee if present")
+ run_beeline(None, "drop table if exists default.employee;",
service_user=True)
logger.info("Disabling all policies on %s and creating audit-only policy",
HIVE_SERVICE_NAME)
set_all_policies_enabled(_admin_session, False, HIVE_SERVICE_NAME)
@@ -110,15 +80,6 @@ def teardown_module():
return
logger.info("=== teardown_module: start ===")
- if _test_database is not None and _test_table is not None:
- logger.info(
- "Hive service cleanup: drop %s.%s and database %s",
- _test_database,
- _test_table,
- _test_database,
- )
- cleanup_hive_test_namespace(_test_database, _test_table)
-
if _audit_policy_id is not None:
logger.info("Deleting audit policy id=%s", _audit_policy_id)
delete_policies_by_id(_admin_session, [_audit_policy_id])
@@ -129,49 +90,44 @@ def teardown_module():
logger.info("=== teardown_module: done ===")
-def test_01_01_table_create_employee_no_policy_as_user1_deny(request):
- log_testcase_begin_for_pytest(logger, request.node.name)
- sql = (
- "create table "
- + _test_database
- + "."
- + _test_table
- + "(name String);"
- )
- logger.info(
- "Running beeline CREATE TABLE as %s (expect deny): %s",
- HIVE_TEST_USER,
- sql,
- )
- exit_code, output = run_beeline(
- HIVE_TEST_USER,
- sql,
- expect_substring=HIVE_AUTHZ_DENIED_SUBSTR,
- )
- assert HIVE_AUTHZ_DENIED_SUBSTR in output, (
- "Expected authorization denial in beeline output; exit_code="
- + str(exit_code)
- + " output="
- + output
- )
-
-
-def test_01_02_database_use_default_no_policy_as_user1_deny(request):
- log_testcase_begin_for_pytest(logger, request.node.name)
- sql = "use " + _test_database + ";"
- logger.info(
- "Running beeline USE database as %s (expect deny): %s",
- HIVE_TEST_USER,
- sql,
- )
- exit_code, output = run_beeline(
- HIVE_TEST_USER,
- sql,
- expect_substring=HIVE_AUTHZ_DENIED_SUBSTR,
- )
- assert HIVE_AUTHZ_DENIED_SUBSTR in output, (
- "Expected authorization denial in beeline output; exit_code="
- + str(exit_code)
- + " output="
- + output
- )
[email protected](
+ ("validateImpala", "verifyStr"),
+ [(False, HIVE_AUTHZ_DENIED_SUBSTR)],
+)
+def test_01_01_TableCreateEmployeeNoPolicyAsUser1Deny(validateImpala,
verifyStr):
+ HivePluginTest(
+ "HIVE.test_01_01_TableCreateEmployeeNoPolicyAsUser1Deny",
+ TEST_CASE,
+ "create table default.employee(name String);",
+ HivePluginTestContext.USER_USER1,
+ HivePluginTestContext.PASS_USER1,
+ verifyCriteria=HivePluginTest.STDERR_CONTAINS,
+ verifyStr=verifyStr,
+ newAuditLogMustExist=True,
+ validateImpala=validateImpala,
+ auditLogFilter={
+ "accessType": TEST_CASE.access_type["CREATE"],
+ "accessResult": TEST_CASE.result_denied,
+ "accessEnforcer": "ranger-acl",
+ "resourceName": "default/employee",
+ },
+ ).execute()
+
+
+def test_01_02_DatabaseUseDefaultNoPolicyAsUser1Deny():
+ HivePluginTest(
+ "HIVE.test_01_02_DatabaseUseDefaultNoPolicyAsUser1Deny",
+ TEST_CASE,
+ "use default;",
+ HivePluginTestContext.USER_USER1,
+ HivePluginTestContext.PASS_USER1,
+ verifyCriteria=HivePluginTest.STDERR_CONTAINS,
+ verifyStr=HIVE_AUTHZ_DENIED_SUBSTR,
+ newAuditLogMustExist=True,
+ auditLogFilter={
+ "accessType": TEST_CASE.access_type["USE"],
+ "accessResult": TEST_CASE.result_denied,
+ "accessEnforcer": "ranger-acl",
+ "resourceName": "default",
+ },
+ ).execute()
diff --git a/functional-tests/ranger_test_utils/access_audit_utils.py
b/functional-tests/ranger_test_utils/access_audit_utils.py
index c37d3f765..7e249eca4 100644
--- a/functional-tests/ranger_test_utils/access_audit_utils.py
+++ b/functional-tests/ranger_test_utils/access_audit_utils.py
@@ -16,16 +16,16 @@
"""
Ranger Admin access-audit query helpers (Docker functional-tests).
-Hive plugin tests do not call these yet — Hive access audits are not reliably
-indexed in the default Docker stack. Wire into hive tests once audit pipeline
works.
+Audits are indexed in Solr or OpenSearch when ``AUDIT_INDEX_STORE`` is not
+``none``; tests poll the Admin ``/assets/accessAudit`` REST API.
"""
-import logging
+import os
import time
-from ranger_test_utils.utils import RANGER_ADMIN_BASE_URL
+from ranger_test_utils.utils import RANGER_ADMIN_BASE_URL, get_test_logger
-logger = logging.getLogger(__name__)
+logger = get_test_logger(__name__)
ACCESS_AUDIT_PATH = RANGER_ADMIN_BASE_URL + "/assets/accessAudit"
@@ -121,6 +121,155 @@ def wait_for_access_audits(
def assert_access_audits_found(session, **kwargs):
- """Convenience assert for future hive audit tests."""
+ """Assert at least one access audit row matches the given filters."""
audits = wait_for_access_audits(session, **kwargs)
assert audits, "Expected access audit events matching filters: " +
str(kwargs)
+
+
+def access_audit_pipeline_enabled():
+ """True when Docker was started with Solr/OpenSearch audit indexing."""
+ return os.environ.get("AUDIT_INDEX_STORE", "opensearch").lower() != "none"
+
+
+def audit_filters_to_query_kwargs(audit_filter, username, repo_name):
+ """Map plugin test ``audit_filter`` dict keys to
``wait_for_access_audits`` kwargs."""
+ if not audit_filter:
+ return {}
+ return {
+ "request_user": audit_filter.get("request_user", username),
+ "repo_name": audit_filter.get("repoName",
audit_filter.get("repo_name", repo_name)),
+ "access_type": audit_filter.get("accessType",
audit_filter.get("access_type")),
+ "resource_path": audit_filter.get(
+ "resourceName", audit_filter.get("resource_path")
+ ),
+ "access_result": audit_filter.get(
+ "accessResult", audit_filter.get("access_result")
+ ),
+ "acl_enforcer": audit_filter.get(
+ "aclEnforcer", audit_filter.get("acl_enforcer")
+ ),
+ "resource_type": audit_filter.get(
+ "resourceType", audit_filter.get("resource_type")
+ ),
+ "max_iterations": audit_filter.get("max_iterations", 12),
+ "sleep_time": audit_filter.get("sleep_time", 5),
+ }
+
+
+def verify_access_audits_after_operation(
+ session,
+ audit_filter,
+ username,
+ repo_name,
+ *,
+ after_event_time,
+ expect_audit_event=True,
+):
+ """
+ Poll access audits newer than ``after_event_time``; assert
presence/absence.
+ """
+ query_kwargs = audit_filters_to_query_kwargs(audit_filter, username,
repo_name)
+ query_kwargs["after_event_time"] = after_event_time
+ audits = wait_for_access_audits(session, **query_kwargs)
+ if expect_audit_event:
+ assert audits, (
+ "Expected access audit events matching filters: " +
str(query_kwargs)
+ )
+ else:
+ assert not audits, (
+ "Unexpected access audit events matching filters: " +
str(query_kwargs)
+ )
+ return audits
+
+
+class AccessAuditCapture:
+ """
+ Baseline Solr/OpenSearch-backed access audits before a plugin client
action.
+
+ Use from any enforcement suite (Hive, HBase, Kafka, HDFS):
+
+ capture = begin_access_audit_capture(admin_session, repo_name,
request_user)
+ ... run beeline / hbase shell / kafka producer / hdfs dfs ...
+ capture.verify(audit_filter)
+ """
+
+ def __init__(self, session, repo_name, request_user):
+ self.session = session
+ self.repo_name = repo_name
+ self.request_user = request_user
+ self.enabled = (
+ session is not None
+ and access_audit_pipeline_enabled()
+ )
+ self.after_event_time = None
+ if self.enabled:
+ self.after_event_time = fetch_last_access_audit_event_time(session)
+ logger.info(
+ "Access audit baseline eventTime=%s repo=%s requestUser=%s",
+ self.after_event_time,
+ repo_name,
+ request_user,
+ )
+ elif session is not None and not access_audit_pipeline_enabled():
+ logger.info(
+ "Skipping access audit checks (AUDIT_INDEX_STORE=none); "
+ "use opensearch or solr when starting Docker to enable"
+ )
+
+ def verify(
+ self,
+ audit_filter,
+ *,
+ expect_audit_event=True,
+ request_user=None,
+ repo_name=None,
+ ):
+ """Assert access audit rows in Solr/OpenSearch (via Admin REST) after
the action."""
+ if not audit_filter:
+ return None
+ if not self.enabled:
+ return None
+ user = request_user if request_user is not None else self.request_user
+ repo = repo_name if repo_name is not None else self.repo_name
+ logger.info("Verifying access audits: %s", audit_filter)
+ return verify_access_audits_after_operation(
+ self.session,
+ audit_filter,
+ user,
+ repo,
+ after_event_time=self.after_event_time,
+ expect_audit_event=expect_audit_event,
+ )
+
+
+def begin_access_audit_capture(session, repo_name, request_user):
+ """
+ Record newest access audit timestamp before a plugin enforcement action.
+
+ Returns an :class:`AccessAuditCapture` for ``verify()`` after the client
runs.
+ """
+ return AccessAuditCapture(session, repo_name, request_user)
+
+
+def verify_plugin_access_audit(
+ session,
+ audit_filter,
+ *,
+ request_user,
+ repo_name,
+ after_event_time,
+ expect_audit_event=True,
+):
+ """
+ One-shot access audit verify when the baseline timestamp is already known.
+
+ Prefer :func:`begin_access_audit_capture` + ``capture.verify()`` for new
tests.
+ """
+ return verify_access_audits_after_operation(
+ session,
+ audit_filter,
+ request_user,
+ repo_name,
+ after_event_time=after_event_time,
+ expect_audit_event=expect_audit_event,
+ )
diff --git a/functional-tests/ranger_test_utils/hbase_utils.py
b/functional-tests/ranger_test_utils/hbase_utils.py
index 511f188fe..ba3fbf91c 100644
--- a/functional-tests/ranger_test_utils/hbase_utils.py
+++ b/functional-tests/ranger_test_utils/hbase_utils.py
@@ -13,7 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-"""HBase Docker helpers for Ranger functional-tests (HBase shell + plugin
enforcement)."""
+"""HBase Docker helpers and declarative plugin scenarios (shell, policies,
``HBasePluginTest``)."""
import base64
import socket
@@ -21,18 +21,23 @@
import pytest
+from ranger_test_utils.plugin_enforcement_utils import (
+ logger as plugin_enforcement_logger,
+ normalize_audit_log_filter,
+)
from ranger_test_utils.utils import (
POLICY_URL,
assert_http_ok,
get_test_logger,
+ log_testcase_begin,
unique_suffix,
)
from ranger_test_utils.hive_utils import (
- HIVE_TEST_USER,
KDC_CONTAINER,
POLICY_PROPAGATION_SLEEP_SEC,
get_docker_client,
)
+from ranger_test_utils.plugin_test_users import PLUGIN_TEST_USER
logger = get_test_logger(__name__)
@@ -47,7 +52,7 @@
HBASE_SERVICE_PRINCIPAL = "hbase/ranger-hbase.rangernw@" + KERBEROS_REALM
HBASE_SERVICE_KEYTAB = "/etc/keytabs/hbase.keytab"
-HBASE_TEST_USER = HIVE_TEST_USER
+HBASE_TEST_USER = PLUGIN_TEST_USER
HBASE_AUTHZ_DENIED_SUBSTR = "AccessDeniedException"
@@ -120,53 +125,12 @@ def _log_multiline_block(header, text,
max_chars=HBASE_SHELL_LOG_OUTPUT_MAX_CHAR
def ensure_kerberos_user(username):
"""Create a user Kerberos principal and keytab on KDC, visible in
ranger-hbase."""
- kdc = get_docker_client().containers.get(KDC_CONTAINER)
- principal = _user_principal(username)
- keytab_in_kdc = "/etc/keytabs/ranger-hbase/" + username + ".keytab"
- keytab_in_hbase = _user_keytab_in_hbase_container(username)
-
- logger.info(
- "Kerberos setup on %s: addprinc/ktadd for %s -> %s (visible in hbase
as %s)",
- KDC_CONTAINER,
- principal,
- keytab_in_kdc,
- keytab_in_hbase,
+ from ranger_test_utils.plugin_test_users import (
+ PLUGIN_KERBEROS_STACK_HBASE,
+ ensure_kerberos_keytab_for_stack,
)
- add_cmd = 'kadmin.local -q "addprinc -randkey ' + principal + '"'
- exit_add, out_add = kdc.exec_run(add_cmd, user="root", demux=True)
- _log_multiline_block(
- "kadmin addprinc exit_code=%s" % exit_add,
- ((out_add[0] or b"") + (out_add[1] or b"")).decode(),
- max_chars=2000,
- )
-
- kdc.exec_run("rm -f " + keytab_in_kdc, user="root")
- ktadd_cmd = (
- 'kadmin.local -q "ktadd -k '
- + keytab_in_kdc
- + " "
- + principal
- + '"'
- )
- exit_kt, out_kt = kdc.exec_run(ktadd_cmd, user="root", demux=True)
- kt_out = ((out_kt[0] or b"") + (out_kt[1] or b"")).decode()
- _log_multiline_block(
- "kadmin ktadd exit_code=%s" % exit_kt,
- kt_out,
- max_chars=2000,
- )
- if exit_kt != 0:
- raise RuntimeError(
- "ktadd failed for "
- + principal
- + " (keytab "
- + keytab_in_hbase
- + "): "
- + kt_out
- )
- kdc.exec_run("chmod 444 " + keytab_in_kdc, user="root")
- verify_kerberos_keytab_in_hbase(username)
+ ensure_kerberos_keytab_for_stack(username, PLUGIN_KERBEROS_STACK_HBASE)
def verify_kerberos_keytab_in_hbase(username):
@@ -355,3 +319,191 @@ def create_audit_only_policy(session, suffix=None,
service_name=HBASE_SERVICE_NA
def wait_for_hbase_policy_propagation(seconds=POLICY_PROPAGATION_SLEEP_SEC):
logger.info("Waiting %ss for HBase plugin policy propagation", seconds)
time.sleep(seconds)
+
+
+def execute_hbase_test(
+ shell_lines,
+ username=HBASE_TEST_USER,
+ *,
+ admin_session=None,
+ verify_output_contains=None,
+ verify_output_not_contains=None,
+ audit_filter=None,
+ expect_audit_event=True,
+ repo_name=HBASE_SERVICE_NAME,
+ service_user=False,
+):
+ """
+ Run HBase shell as ``username``, assert on output, optionally validate
access audits.
+
+ Uses
:func:`ranger_test_utils.access_audit_utils.begin_access_audit_capture`
+ (Solr/OpenSearch via Admin REST when the audit pipeline is enabled).
+ """
+ from ranger_test_utils.access_audit_utils import begin_access_audit_capture
+
+ if isinstance(shell_lines, str):
+ shell_lines = [shell_lines]
+
+ audit_capture = None
+ if (
+ audit_filter is not None
+ and admin_session is not None
+ and not service_user
+ ):
+ audit_capture = begin_access_audit_capture(
+ admin_session,
+ repo_name,
+ username,
+ )
+
+ exit_code, output = run_hbase_shell(
+ username,
+ shell_lines,
+ service_user=service_user,
+ expect_substring=verify_output_contains,
+ )
+
+ if verify_output_contains is not None:
+ assert verify_output_contains in output, (
+ "Expected text not found in HBase shell output; exit_code="
+ + str(exit_code)
+ + " output="
+ + output
+ )
+ if verify_output_not_contains is not None:
+ assert verify_output_not_contains not in output, (
+ "Unexpected text in HBase shell output; exit_code="
+ + str(exit_code)
+ + " output="
+ + output
+ )
+
+ if audit_capture is not None:
+ audit_capture.verify(
+ audit_filter,
+ expect_audit_event=expect_audit_event,
+ )
+
+ return exit_code, output
+
+
+class HBasePluginTestContext:
+ """
+ Per-module context (assign ``admin_session`` in ``setup_module``).
+
+ Class attributes mirror common ranger-qe ``XAHBaseTestCase`` fields used
in ports.
+ """
+
+ USER_USER1 = HBASE_TEST_USER
+ USER_USER2 = "hrt_22"
+
+ result_denied = "0"
+ result_allowed = "1"
+
+ access_type = {
+ "CREATE_TABLE": "createTable",
+ "PUT": "put",
+ "SCAN": "scannerOpen",
+ "GET": "get",
+ "DISABLE_TABLE": "disableTable",
+ "DELETE_TABLE": "deleteTable",
+ "GRANT": "grant",
+ "REVOKE": "revoke",
+ }
+
+ def __init__(self):
+ self.admin_session = None
+ self.repo_name = HBASE_SERVICE_NAME
+
+
+class HBasePluginTest:
+ """One HBase enforcement scenario: log id, shell cmds, verify output,
optional audit."""
+
+ STDERR_CONTAINS = "stderr_contains"
+ STDOUT_CONTAINS = "stdout_contains"
+ STDERR_NOT_CONTAINS = "stderr_not_contains"
+ STDOUT_NOT_CONTAINS = "stdout_not_contains"
+
+ def __init__(
+ self,
+ test_id,
+ test_case,
+ shell_cmds=None,
+ req_user=None,
+ *,
+ verify_criteria=STDOUT_CONTAINS,
+ verify_str=None,
+ new_audit_log_must_exist=True,
+ audit_log_filter=None,
+ logoutput=True,
+ verifyCriteria=None,
+ verifyStr=None,
+ newAuditLogMustExist=None,
+ auditLogFilter=None,
+ hBaseCmds=None,
+ reqUser=None,
+ checkAuditLog=None,
+ ):
+ self.test_id = test_id
+ self.test_case = test_case
+ if hBaseCmds is not None:
+ self.shell_cmds = hBaseCmds
+ elif shell_cmds is not None:
+ self.shell_cmds = shell_cmds
+ else:
+ raise ValueError("Provide shell_cmds or hBaseCmds")
+ self.req_user = reqUser if reqUser is not None else req_user
+ if self.req_user is None:
+ raise ValueError("Provide req_user or reqUser")
+ self.logoutput = logoutput
+
+ self.verify_criteria = (
+ verifyCriteria if verifyCriteria is not None else verify_criteria
+ )
+ self.verify_str = (
+ verifyStr
+ if verifyStr is not None
+ else (verify_str if verify_str is not None else
HBASE_AUTHZ_DENIED_SUBSTR)
+ )
+ self.new_audit_log_must_exist = (
+ newAuditLogMustExist
+ if newAuditLogMustExist is not None
+ else new_audit_log_must_exist
+ )
+ self.audit_log_filter = (
+ auditLogFilter if auditLogFilter is not None else audit_log_filter
+ )
+ _ = checkAuditLog
+
+ def execute(self):
+ log_testcase_begin(plugin_enforcement_logger, self.test_id)
+
+ verify_contains = None
+ verify_not_contains = None
+ if self.verify_criteria in (
+ self.STDERR_CONTAINS,
+ self.STDOUT_CONTAINS,
+ ):
+ verify_contains = self.verify_str
+ elif self.verify_criteria in (
+ self.STDERR_NOT_CONTAINS,
+ self.STDOUT_NOT_CONTAINS,
+ ):
+ verify_not_contains = self.verify_str
+ else:
+ raise ValueError("Unsupported verify_criteria: " +
str(self.verify_criteria))
+
+ admin_session = getattr(self.test_case, "admin_session", None)
+ repo_name = getattr(self.test_case, "repo_name", HBASE_SERVICE_NAME)
+ audit_filter = normalize_audit_log_filter(self.audit_log_filter)
+
+ execute_hbase_test(
+ self.shell_cmds,
+ username=self.req_user,
+ admin_session=admin_session,
+ verify_output_contains=verify_contains,
+ verify_output_not_contains=verify_not_contains,
+ audit_filter=audit_filter,
+ expect_audit_event=self.new_audit_log_must_exist,
+ repo_name=repo_name,
+ )
diff --git a/functional-tests/ranger_test_utils/hive_utils.py
b/functional-tests/ranger_test_utils/hive_utils.py
index a6e408625..b939d9b43 100644
--- a/functional-tests/ranger_test_utils/hive_utils.py
+++ b/functional-tests/ranger_test_utils/hive_utils.py
@@ -13,19 +13,25 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-"""Hive Docker helpers for Ranger functional-tests (HS2 + plugin
enforcement)."""
+"""Hive Docker helpers and declarative plugin scenarios (beeline, policies,
``HivePluginTest``)."""
import socket
import time
import pytest
+from ranger_test_utils.plugin_test_users import PLUGIN_TEST_USER
+from ranger_test_utils.plugin_enforcement_utils import (
+ logger as plugin_enforcement_logger,
+ normalize_audit_log_filter,
+)
from ranger_test_utils.utils import (
DOCKER_PLUGIN_TEST_USER_PASSWORD,
POLICY_URL,
RANGER_ADMIN_BASE_URL,
assert_http_ok,
get_test_logger,
+ log_testcase_begin,
unique_suffix,
)
@@ -47,7 +53,7 @@
"jdbc:hive2://localhost:10000/default;principal=" + HIVE_HS2_PRINCIPAL
)
-HIVE_TEST_USER = "hrt_21"
+HIVE_TEST_USER = PLUGIN_TEST_USER
HIVE_TEST_USER_PASSWORD = DOCKER_PLUGIN_TEST_USER_PASSWORD
HIVE_AUTHZ_DENIED_SUBSTR = "Permission denied"
@@ -152,8 +158,8 @@ def _log_text_block(title, text,
max_chars=BEELINE_COMPLETE_LOG_MAX_CHARS):
logger.info("%s", line)
-def _log_beeline_qe_style(exit_code, stdout, stderr, combined, sql,
beeline_cmd, expect_substring=None):
- """beeline logging: query, command, exit code, stdout, stderr."""
+def _log_beeline_run_detail(exit_code, stdout, stderr, combined, sql,
beeline_cmd, expect_substring=None):
+ """Log beeline query, command, exit code, stdout, and stderr."""
logger.info("Exit Code: %s", exit_code)
_log_text_block("Output of beeline command is:", stdout)
_log_text_block("Error of beeline command is:", stderr)
@@ -190,53 +196,12 @@ def _log_beeline_qe_style(exit_code, stdout, stderr,
combined, sql, beeline_cmd,
def ensure_kerberos_user(username):
"""Create a user Kerberos principal and keytab on KDC, visible in
ranger-hive."""
- kdc = get_docker_client().containers.get(KDC_CONTAINER)
- principal = _user_principal(username)
- keytab_in_kdc = "/etc/keytabs/ranger-hive/" + username + ".keytab"
- keytab_in_hive = _user_keytab_in_hive_container(username)
-
- logger.info(
- "Kerberos setup on %s: addprinc/ktadd for %s -> %s (visible in hive as
%s)",
- KDC_CONTAINER,
- principal,
- keytab_in_kdc,
- keytab_in_hive,
- )
-
- add_cmd = 'kadmin.local -q "addprinc -randkey ' + principal + '"'
- exit_add, out_add = kdc.exec_run(add_cmd, user="root", demux=True)
- _log_multiline_block(
- "kadmin addprinc exit_code=%s" % exit_add,
- ((out_add[0] or b"") + (out_add[1] or b"")).decode(),
- max_chars=2000,
+ from ranger_test_utils.plugin_test_users import (
+ PLUGIN_KERBEROS_STACK_HIVE,
+ ensure_kerberos_keytab_for_stack,
)
- kdc.exec_run("rm -f " + keytab_in_kdc, user="root")
- ktadd_cmd = (
- 'kadmin.local -q "ktadd -k '
- + keytab_in_kdc
- + " "
- + principal
- + '"'
- )
- exit_kt, out_kt = kdc.exec_run(ktadd_cmd, user="root", demux=True)
- kt_out = ((out_kt[0] or b"") + (out_kt[1] or b"")).decode()
- _log_multiline_block(
- "kadmin ktadd exit_code=%s" % exit_kt,
- kt_out,
- max_chars=2000,
- )
- if exit_kt != 0:
- raise RuntimeError(
- "ktadd failed for "
- + principal
- + " (keytab "
- + keytab_in_hive
- + "): "
- + kt_out
- )
- kdc.exec_run("chmod 444 " + keytab_in_kdc, user="root")
- verify_kerberos_keytab_in_hive(username)
+ ensure_kerberos_keytab_for_stack(username, PLUGIN_KERBEROS_STACK_HIVE)
def verify_kerberos_keytab_in_hive(username):
@@ -356,7 +321,7 @@ def run_beeline(username, sql, service_user=False,
expect_substring=None):
stderr = (output[1] or b"").decode()
combined = stdout + stderr
- _log_beeline_qe_style(
+ _log_beeline_run_detail(
exit_code,
stdout,
stderr,
@@ -467,3 +432,184 @@ def delete_policies_by_id(session, policy_ids):
def wait_for_policy_propagation(seconds=POLICY_PROPAGATION_SLEEP_SEC):
logger.info("Waiting %ss for Hive plugin policy propagation", seconds)
time.sleep(seconds)
+
+
+def execute_hive_test(
+ sql,
+ username=HIVE_TEST_USER,
+ *,
+ admin_session=None,
+ verify_output_contains=None,
+ verify_output_not_contains=None,
+ audit_filter=None,
+ expect_audit_event=True,
+ repo_name=HIVE_SERVICE_NAME,
+):
+ """
+ Run beeline as ``username``, assert on CLI output, optionally validate
access audits.
+
+ Uses
:func:`ranger_test_utils.access_audit_utils.begin_access_audit_capture`
+ for Solr/OpenSearch-backed audit checks (shared with other plugin suites).
+ """
+ from ranger_test_utils.access_audit_utils import begin_access_audit_capture
+
+ audit_capture = None
+ if audit_filter is not None and admin_session is not None:
+ audit_capture = begin_access_audit_capture(
+ admin_session,
+ repo_name,
+ username,
+ )
+
+ exit_code, output = run_beeline(
+ username,
+ sql,
+ expect_substring=verify_output_contains,
+ )
+
+ if verify_output_contains is not None:
+ assert verify_output_contains in output, (
+ "Expected text not found in beeline output; exit_code="
+ + str(exit_code)
+ + " output="
+ + output
+ )
+ if verify_output_not_contains is not None:
+ assert verify_output_not_contains not in output, (
+ "Unexpected text in beeline output; exit_code="
+ + str(exit_code)
+ + " output="
+ + output
+ )
+
+ if audit_capture is not None:
+ audit_capture.verify(
+ audit_filter,
+ expect_audit_event=expect_audit_event,
+ )
+
+ return exit_code, output
+
+
+class HivePluginTestContext:
+ """
+ Per-module context (assign ``admin_session`` in ``setup_module``).
+
+ Class attributes mirror common ranger-qe ``XAHiveTestCase`` fields used in
ports.
+ """
+
+ USER_USER1 = HIVE_TEST_USER
+ USER_USER2 = "hrt_22"
+ PASS_USER1 = None
+ PASS_USER2 = None
+
+ result_denied = "0"
+ result_allowed = "1"
+
+ access_type = {
+ "CREATE": "CREATE",
+ "USE": "USE",
+ "DROP": "DROP",
+ "SELECT": "SELECT",
+ "UPDATE": "UPDATE",
+ "ALTER": "ALTER",
+ "INSERT": "INSERT",
+ "DELETE": "DELETE",
+ "GRANT": "GRANT",
+ "REVOKE": "REVOKE",
+ }
+
+ def __init__(self):
+ self.admin_session = None
+ self.repo_name = HIVE_SERVICE_NAME
+
+
+class HivePluginTest:
+ """One Hive enforcement scenario: log id, beeline SQL, verify output,
optional audit."""
+
+ STDERR_CONTAINS = "stderr_contains"
+ STDOUT_CONTAINS = "stdout_contains"
+ STDERR_NOT_CONTAINS = "stderr_not_contains"
+ STDOUT_NOT_CONTAINS = "stdout_not_contains"
+
+ def __init__(
+ self,
+ test_id,
+ test_case,
+ sql,
+ req_user,
+ req_user_pass=None,
+ *,
+ verify_criteria=STDERR_CONTAINS,
+ verify_str=None,
+ new_audit_log_must_exist=True,
+ audit_log_filter=None,
+ validate_impala=False,
+ verifyCriteria=None,
+ verifyStr=None,
+ newAuditLogMustExist=None,
+ auditLogFilter=None,
+ validateImpala=None,
+ hSql=None,
+ ):
+ self.test_id = test_id
+ self.test_case = test_case
+ self.sql = hSql if hSql is not None else sql
+ self.req_user = req_user
+ self.req_user_pass = req_user_pass
+
+ self.verify_criteria = (
+ verifyCriteria if verifyCriteria is not None else verify_criteria
+ )
+ self.verify_str = (
+ verifyStr
+ if verifyStr is not None
+ else (verify_str if verify_str is not None else
HIVE_AUTHZ_DENIED_SUBSTR)
+ )
+ self.new_audit_log_must_exist = (
+ newAuditLogMustExist
+ if newAuditLogMustExist is not None
+ else new_audit_log_must_exist
+ )
+ self.audit_log_filter = (
+ auditLogFilter if auditLogFilter is not None else audit_log_filter
+ )
+ self.validate_impala = (
+ validateImpala if validateImpala is not None else validate_impala
+ )
+
+ def execute(self):
+ if self.validate_impala:
+ pytest.skip("Impala validation is not enabled in Docker
functional-tests")
+
+ log_testcase_begin(plugin_enforcement_logger, self.test_id)
+
+ verify_contains = None
+ verify_not_contains = None
+ if self.verify_criteria in (
+ self.STDERR_CONTAINS,
+ self.STDOUT_CONTAINS,
+ ):
+ verify_contains = self.verify_str
+ elif self.verify_criteria in (
+ self.STDERR_NOT_CONTAINS,
+ self.STDOUT_NOT_CONTAINS,
+ ):
+ verify_not_contains = self.verify_str
+ else:
+ raise ValueError("Unsupported verify_criteria: " +
str(self.verify_criteria))
+
+ admin_session = getattr(self.test_case, "admin_session", None)
+ repo_name = getattr(self.test_case, "repo_name", HIVE_SERVICE_NAME)
+ audit_filter = normalize_audit_log_filter(self.audit_log_filter)
+
+ execute_hive_test(
+ self.sql,
+ username=self.req_user,
+ admin_session=admin_session,
+ verify_output_contains=verify_contains,
+ verify_output_not_contains=verify_not_contains,
+ audit_filter=audit_filter,
+ expect_audit_event=self.new_audit_log_must_exist,
+ repo_name=repo_name,
+ )
diff --git a/functional-tests/ranger_test_utils/plugin_conftest.py
b/functional-tests/ranger_test_utils/plugin_conftest.py
new file mode 100644
index 000000000..7cbcd7910
--- /dev/null
+++ b/functional-tests/ranger_test_utils/plugin_conftest.py
@@ -0,0 +1,146 @@
+# 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.
+
+"""
+Session setup for Docker plugin enforcement suites (Hive, HDFS, HBase, Kafka).
+
+Used by ``functional-tests/conftest.py`` when collected tests carry a
registered
+plugin marker. Suites can also call ``ensure_plugin_test_users_for_markers``
from
+``setup_module`` if they need explicit control.
+"""
+
+import sys
+
+import pytest
+import requests
+
+from ranger_test_utils.plugin_test_users import (
+ PLUGIN_KERBEROS_STACK_HADOOP,
+ PLUGIN_KERBEROS_STACK_HBASE,
+ PLUGIN_KERBEROS_STACK_HIVE,
+ PLUGIN_KERBEROS_STACK_KAFKA,
+ ensure_plugin_test_users,
+)
+from ranger_test_utils.utils import (
+ create_ranger_admin_session,
+ get_test_logger,
+ preflight_ranger_admin_http,
+)
+
+logger = get_test_logger(__name__)
+
+
+def _plugin_setup_progress(message):
+ """Visible on console during session setup (log_cli is off until module
tests run)."""
+ print("[plugin-setup] " + message, file=sys.stderr, flush=True)
+
+# pytest marker name -> Kerberos stacks to provision for that suite
+PLUGIN_MARKER_KERBEROS_STACKS = {
+ "hive": (PLUGIN_KERBEROS_STACK_HIVE,),
+ "hbase": (PLUGIN_KERBEROS_STACK_HBASE,),
+ "kafka": (PLUGIN_KERBEROS_STACK_KAFKA,),
+ "hdfs_policy": (PLUGIN_KERBEROS_STACK_HADOOP,),
+}
+
+PLUGIN_ENFORCEMENT_MARKERS = frozenset(PLUGIN_MARKER_KERBEROS_STACKS.keys())
+
+
+def _preflight_for_marker(marker_name):
+ if marker_name == "hive":
+ from ranger_test_utils.hive_utils import preflight_hive_stack
+
+ preflight_hive_stack()
+ elif marker_name == "hbase":
+ from ranger_test_utils.hbase_utils import preflight_hbase_stack
+
+ preflight_hbase_stack()
+ elif marker_name == "kafka":
+ from ranger_test_utils.kafka_utils import preflight_kafka_stack
+
+ preflight_kafka_stack()
+ elif marker_name == "hdfs_policy":
+ from ranger_test_utils.hdfs_utils import preflight_hdfs_plugin_stack
+
+ preflight_hdfs_plugin_stack()
+
+
+def markers_present_in_session(session):
+ """Return plugin enforcement marker names present on collected tests."""
+ found = set()
+ for item in session.items:
+ for marker_name in PLUGIN_ENFORCEMENT_MARKERS:
+ if item.get_closest_marker(marker_name) is not None:
+ found.add(marker_name)
+ return found
+
+
+def kerberos_stacks_for_markers(marker_names):
+ """Merge Kerberos stack tuples for the given marker names (stable
order)."""
+ stacks = []
+ seen = set()
+ for marker_name in sorted(marker_names):
+ for stack in PLUGIN_MARKER_KERBEROS_STACKS.get(marker_name, ()):
+ if stack not in seen:
+ seen.add(stack)
+ stacks.append(stack)
+ return tuple(stacks)
+
+
+def ensure_plugin_test_users_for_markers(marker_names):
+ """
+ Run stack preflights, then Ranger xusers + Kerberos for merged stacks.
+
+ Safe to call from pytest session setup or from a suite ``setup_module``.
+ """
+ marker_names = set(marker_names)
+ if not marker_names:
+ return
+
+ _plugin_setup_progress(
+ "starting for markers: " + ", ".join(sorted(marker_names))
+ )
+
+ for marker_name in sorted(marker_names):
+ logger.info("Plugin preflight for marker: %s", marker_name)
+ _plugin_setup_progress("preflight: " + marker_name)
+ _preflight_for_marker(marker_name)
+
+ _plugin_setup_progress("checking Ranger Admin HTTP")
+ preflight_ranger_admin_http()
+
+ stacks = kerberos_stacks_for_markers(marker_names)
+ _plugin_setup_progress(
+ "Ranger xusers + Kerberos for stacks: " + ", ".join(stacks)
+ )
+ logger.info(
+ "Ensuring plugin test users (Ranger + Kerberos) for stacks: %s",
+ stacks,
+ )
+ try:
+ session = create_ranger_admin_session()
+ ensure_plugin_test_users(session, kerberos_stacks=stacks)
+ except (requests.RequestException, RuntimeError) as exc:
+ pytest.skip("Plugin user setup failed: " + str(exc))
+
+ _plugin_setup_progress("done")
+ logger.info("Plugin test users ready for markers: %s",
sorted(marker_names))
+
+
+def ensure_plugin_test_users_for_session(pytest_session):
+ """If any plugin enforcement tests were collected, provision shared
users."""
+ marker_names = markers_present_in_session(pytest_session)
+ if not marker_names:
+ return
+ ensure_plugin_test_users_for_markers(marker_names)
diff --git a/functional-tests/ranger_test_utils/plugin_enforcement_utils.py
b/functional-tests/ranger_test_utils/plugin_enforcement_utils.py
new file mode 100644
index 000000000..5dc09d575
--- /dev/null
+++ b/functional-tests/ranger_test_utils/plugin_enforcement_utils.py
@@ -0,0 +1,61 @@
+# 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.
+
+"""
+Shared logging, access-audit checks, and pytest helpers for plugin enforcement
suites.
+
+Test modules should import ``logger`` from here (same pattern as
``api_test_utils``
+for ``api_tests``).
+
+Access audits (Solr/OpenSearch via Admin REST) for any service::
+
+ from ranger_test_utils.plugin_enforcement_utils import (
+ begin_access_audit_capture,
+ logger,
+ )
+
+ capture = begin_access_audit_capture(admin_session, "dev_hbase", "hrt_21")
+ ... hbase shell / kafka producer / hdfs dfs ...
+ capture.verify({"accessType": "...", "accessResult": "0", ...})
+"""
+
+from ranger_test_utils.access_audit_utils import (
+ AccessAuditCapture,
+ access_audit_pipeline_enabled,
+ assert_access_audits_found,
+ begin_access_audit_capture,
+ fetch_last_access_audit_event_time,
+ verify_plugin_access_audit,
+ wait_for_access_audits,
+)
+from ranger_test_utils.utils import (
+ configure_test_logging,
+ create_ranger_admin_session,
+ get_test_logger,
+ log_testcase_begin_for_pytest,
+ unique_suffix,
+)
+
+logger = get_test_logger(__name__)
+
+
+def normalize_audit_log_filter(audit_log_filter):
+ """Map ranger-qe ``accessEnforcer`` to functional-tests ``aclEnforcer``."""
+ if not audit_log_filter:
+ return None
+ normalized = dict(audit_log_filter)
+ if "accessEnforcer" in normalized and "aclEnforcer" not in normalized:
+ normalized["aclEnforcer"] = normalized.pop("accessEnforcer")
+ return normalized
diff --git a/functional-tests/ranger_test_utils/plugin_test_users.py
b/functional-tests/ranger_test_utils/plugin_test_users.py
new file mode 100644
index 000000000..7c40d5e77
--- /dev/null
+++ b/functional-tests/ranger_test_utils/plugin_test_users.py
@@ -0,0 +1,401 @@
+# 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.
+
+"""
+Shared Ranger xusers and Kerberos principals for Docker plugin enforcement
suites.
+
+Call ``ensure_plugin_test_users`` from suite ``conftest.py`` (session scope) or
+``setup_module`` before service-specific policy setup. Kerberos uses ``ktadd
+-norandkey`` on KDC volume paths so the same principal can receive keytabs on
+multiple stacks without invalidating existing keytabs.
+"""
+
+import subprocess
+import sys
+import tempfile
+
+from ranger_test_utils.utils import (
+ ARGUS_POLICY_USERS,
+ DOCKER_PLUGIN_TEST_USER_PASSWORD,
+ ensure_group_exists,
+ ensure_user_assigned_to_groups,
+ ensure_user_exists,
+ get_test_logger,
+)
+
+logger = get_test_logger(__name__)
+
+KDC_CONTAINER = "ranger-kdc"
+HADOOP_CONTAINER = "ranger-hadoop"
+KERBEROS_REALM = "EXAMPLE.COM"
+
+# Primary enforcement user (Hive/HBase tests); second user used by HDFS/Kafka
deny cases.
+PLUGIN_TEST_USER = ARGUS_POLICY_USERS[0]
+PLUGIN_TEST_USER_ALT = ARGUS_POLICY_USERS[1]
+
+# Policy / multi-user Hive-HDFS scenarios in ranger-qe (hrt_1 .. hrt_10).
+PLUGIN_HRT_USERS_1_TO_10 = ["hrt_" + str(i) for i in range(1, 11)]
+
+# Ranger xusers provisioned on QE clusters (ansible create-test-users role
test_users).
+# Source: ranger-qe/.qaas/ansible/roles/create-test-users/vars/main.yml
+PLUGIN_RANGER_CLUSTER_TEST_USERS = [
+ "hiveuser",
+ "hrt_21",
+ "hrt_22",
+ "hivetest",
+ "impala",
+ "hbaseuser",
+ "hrt_qa",
+]
+
+
+def _unique_usernames_in_order(names):
+ seen = set()
+ result = []
+ for name in names:
+ if name not in seen:
+ seen.add(name)
+ result.append(name)
+ return result
+
+
+# Default Ranger xusers for plugin session setup: hrt_1..10 plus QE cluster
test_users.
+PLUGIN_RANGER_DEFAULT_TEST_USERS = _unique_usernames_in_order(
+ PLUGIN_HRT_USERS_1_TO_10 + PLUGIN_RANGER_CLUSTER_TEST_USERS
+)
+
+# Subset that need Kerberos keytabs for current Docker plugin enforcement
suites.
+PLUGIN_KERBEROS_TEST_USERS = list(ARGUS_POLICY_USERS)
+
+# Ranger group membership aligned with QE ranger_test_users_groups (hrt_21 /
hrt_22).
+PLUGIN_RANGER_USER_GROUP_MAP = {
+ "hrt_21": ["finance"],
+ "hrt_22": ["audit"],
+}
+
+# Kerberos stacks plugin suites can request (see ``KDC_VOLUME_KEYTAB_DIRS`` /
hadoop).
+PLUGIN_KERBEROS_STACK_HIVE = "hive"
+PLUGIN_KERBEROS_STACK_HBASE = "hbase"
+PLUGIN_KERBEROS_STACK_KAFKA = "kafka"
+PLUGIN_KERBEROS_STACK_HADOOP = "hadoop"
+
+PLUGIN_KERBEROS_STACKS_ALL = (
+ PLUGIN_KERBEROS_STACK_HIVE,
+ PLUGIN_KERBEROS_STACK_HBASE,
+ PLUGIN_KERBEROS_STACK_KAFKA,
+ PLUGIN_KERBEROS_STACK_HADOOP,
+)
+
+# KDC paths are volume-mounted into service containers as
/etc/keytabs/<user>.keytab
+KDC_VOLUME_KEYTAB_DIRS = {
+ PLUGIN_KERBEROS_STACK_HIVE: "/etc/keytabs/ranger-hive",
+ PLUGIN_KERBEROS_STACK_HBASE: "/etc/keytabs/ranger-hbase",
+ PLUGIN_KERBEROS_STACK_KAFKA: "/etc/keytabs/ranger-kafka",
+}
+
+_docker_client = None
+
+
+def get_docker_client():
+ global _docker_client
+ if _docker_client is None:
+ import docker
+
+ _docker_client = docker.from_env(timeout=10)
+ try:
+ _docker_client.ping()
+ except Exception as exc:
+ raise RuntimeError(
+ "Docker is not available (is the daemon running?): " + str(exc)
+ ) from exc
+ return _docker_client
+
+
+def user_kerberos_principal(username):
+ return username + "@" + KERBEROS_REALM
+
+
+def user_keytab_in_service_container(username):
+ return "/etc/keytabs/" + username + ".keytab"
+
+
+def ensure_plugin_ranger_groups(session):
+ """Create finance/audit groups and assign hrt_21 / hrt_22 like QE cluster
seed."""
+ for group_name in ("finance", "audit"):
+ ensure_group_exists(session, group_name)
+ for username, group_names in PLUGIN_RANGER_USER_GROUP_MAP.items():
+ ensure_user_assigned_to_groups(session, username, group_names)
+
+
+def ensure_plugin_test_users(
+ session,
+ usernames=None,
+ kerberos_usernames=None,
+ kerberos_stacks=None,
+ ranger_password=DOCKER_PLUGIN_TEST_USER_PASSWORD,
+ verify_keytabs=True,
+ ensure_groups=True,
+):
+ """
+ Create Ranger xusers and Kerberos keytabs for plugin enforcement tests.
+
+ Parameters
+ ----------
+ session:
+ Ranger Admin ``requests.Session`` from ``create_ranger_admin_session``.
+ usernames:
+ Ranger user names to ensure (default:
``PLUGIN_RANGER_DEFAULT_TEST_USERS``:
+ ``hrt_1``..``hrt_10`` plus ranger-qe ansible ``test_users``).
+ kerberos_usernames:
+ Users to receive keytabs (default: ``PLUGIN_KERBEROS_TEST_USERS`` /
hrt_21, hrt_22).
+ kerberos_stacks:
+ Subset of ``PLUGIN_KERBEROS_STACKS_ALL``; default is Hive only.
+ verify_keytabs:
+ When True, kinit in the target container after each stack keytab is
installed.
+ ensure_groups:
+ When True, ensure finance/audit groups and hrt_21/hrt_22 membership.
+ """
+ if usernames is None:
+ usernames = list(PLUGIN_RANGER_DEFAULT_TEST_USERS)
+ if kerberos_usernames is None:
+ kerberos_usernames = list(PLUGIN_KERBEROS_TEST_USERS)
+ if kerberos_stacks is None:
+ kerberos_stacks = (PLUGIN_KERBEROS_STACK_HIVE,)
+
+ logger.info(
+ "Ensuring plugin test users in Ranger: %s; Kerberos users: %s; stacks:
%s",
+ usernames,
+ kerberos_usernames,
+ kerberos_stacks,
+ )
+ for username in usernames:
+ print(
+ "[plugin-setup] Ranger user: " + username,
+ file=sys.stderr,
+ flush=True,
+ )
+ ensure_user_exists(
+ session,
+ username,
+ password=ranger_password,
+ last_name="plugin_test",
+ )
+
+ if ensure_groups:
+ print("[plugin-setup] Ranger groups: finance, audit", file=sys.stderr,
flush=True)
+ ensure_plugin_ranger_groups(session)
+
+ for username in kerberos_usernames:
+ for stack in kerberos_stacks:
+ print(
+ "[plugin-setup] Kerberos "
+ + username
+ + " @ stack "
+ + stack,
+ file=sys.stderr,
+ flush=True,
+ )
+ ensure_kerberos_keytab_for_stack(
+ username,
+ stack,
+ verify=verify_keytabs,
+ )
+
+
+def ensure_kerberos_keytab_for_stack(username, stack, verify=True):
+ """Idempotent principal + keytab for one plugin stack."""
+ ensure_kerberos_principal(username)
+ if stack in KDC_VOLUME_KEYTAB_DIRS:
+ _install_keytab_on_kdc_volume(username, KDC_VOLUME_KEYTAB_DIRS[stack])
+ elif stack == PLUGIN_KERBEROS_STACK_HADOOP:
+ _install_keytab_on_hadoop_container(username)
+ else:
+ raise ValueError("Unknown Kerberos stack: " + str(stack))
+
+ if not verify:
+ return
+
+ if stack == PLUGIN_KERBEROS_STACK_HIVE:
+ from ranger_test_utils.hive_utils import verify_kerberos_keytab_in_hive
+
+ verify_kerberos_keytab_in_hive(username)
+ elif stack == PLUGIN_KERBEROS_STACK_HBASE:
+ from ranger_test_utils.hbase_utils import
verify_kerberos_keytab_in_hbase
+
+ verify_kerberos_keytab_in_hbase(username)
+ elif stack == PLUGIN_KERBEROS_STACK_KAFKA:
+ _verify_kerberos_keytab_in_kafka(username)
+ elif stack == PLUGIN_KERBEROS_STACK_HADOOP:
+ _verify_kerberos_keytab_in_hadoop(username)
+
+
+def ensure_kerberos_principal(username):
+ """Create user@REALM in KDC if it does not exist."""
+ kdc = get_docker_client().containers.get(KDC_CONTAINER)
+ principal = user_kerberos_principal(username)
+ exit_code, output = kdc.exec_run(
+ 'kadmin.local -q "getprinc ' + principal + '"',
+ user="root",
+ )
+ combined = (output or b"").decode()
+ if exit_code == 0 and "does not exist" not in combined:
+ logger.info("Kerberos principal already exists: %s", principal)
+ return
+
+ exit_add, out_add = kdc.exec_run(
+ 'kadmin.local -q "addprinc -randkey ' + principal + '"',
+ user="root",
+ demux=True,
+ )
+ add_out = ((out_add[0] or b"") + (out_add[1] or b"")).decode()
+ if exit_add != 0:
+ raise RuntimeError(
+ "addprinc failed for " + principal + ": " + add_out
+ )
+ logger.info("Created Kerberos principal: %s", principal)
+
+
+def _install_keytab_on_kdc_volume(username, kdc_keytab_dir):
+ kdc = get_docker_client().containers.get(KDC_CONTAINER)
+ principal = user_kerberos_principal(username)
+ keytab_path = kdc_keytab_dir + "/" + username + ".keytab"
+
+ logger.info(
+ "Kerberos keytab on KDC volume: principal=%s path=%s",
+ principal,
+ keytab_path,
+ )
+ kdc.exec_run("rm -f " + keytab_path, user="root")
+ exit_code, output = kdc.exec_run(
+ 'kadmin.local -q "ktadd -norandkey -k '
+ + keytab_path
+ + " "
+ + principal
+ + '"',
+ user="root",
+ demux=True,
+ )
+ kt_out = ((output[0] or b"") + (output[1] or b"")).decode()
+ if exit_code != 0:
+ raise RuntimeError(
+ "ktadd failed for " + principal + " at " + keytab_path + ": " +
kt_out
+ )
+ kdc.exec_run("chmod 444 " + keytab_path, user="root")
+
+
+def _install_keytab_on_hadoop_container(username):
+ kdc = get_docker_client().containers.get(KDC_CONTAINER)
+ hadoop = get_docker_client().containers.get(HADOOP_CONTAINER)
+ principal = user_kerberos_principal(username)
+ keytab_name = username + ".keytab"
+ kdc_tmp = "/tmp/" + keytab_name
+ dest_path = user_keytab_in_service_container(username)
+
+ ensure_kerberos_principal(username)
+
+ kdc.exec_run("rm -f " + kdc_tmp, user="root")
+ exit_code, output = kdc.exec_run(
+ 'kadmin.local -q "xst -k ' + kdc_tmp + " " + principal + '"',
+ user="root",
+ )
+ if exit_code != 0:
+ raise RuntimeError(
+ "keytab export failed for "
+ + principal
+ + ": "
+ + (output or b"").decode()
+ )
+
+ local = tempfile.NamedTemporaryFile(delete=False, suffix=".keytab")
+ local.close()
+ try:
+ subprocess.check_call(
+ ["docker", "cp", KDC_CONTAINER + ":" + kdc_tmp, local.name]
+ )
+ subprocess.check_call(
+ ["docker", "cp", local.name, HADOOP_CONTAINER + ":" + dest_path]
+ )
+ finally:
+ import os
+
+ os.remove(local.name)
+
+ hadoop.exec_run("chmod 444 " + dest_path, user="root")
+ hadoop.exec_run(
+ "chown " + username + ":" + username + " " + dest_path,
+ user="root",
+ )
+ logger.info(
+ "Kerberos keytab on %s: principal=%s path=%s",
+ HADOOP_CONTAINER,
+ principal,
+ dest_path,
+ )
+
+
+def _verify_kerberos_keytab_in_kafka(username):
+ from ranger_test_utils.kafka_utils import KAFKA_CONTAINER
+
+ container = get_docker_client().containers.get(KAFKA_CONTAINER)
+ principal = user_kerberos_principal(username)
+ keytab = user_keytab_in_service_container(username)
+ shell = (
+ "kdestroy -A 2>/dev/null || true; "
+ "kinit -kt "
+ + keytab
+ + " "
+ + principal
+ + "; klist"
+ )
+ exit_code, output = container.exec_run(
+ ["bash", "-c", shell], user="root", demux=True
+ )
+ combined = ((output[0] or b"") + (output[1] or b"")).decode()
+ if exit_code != 0:
+ raise RuntimeError(
+ "Kerberos kinit failed in "
+ + KAFKA_CONTAINER
+ + " for "
+ + principal
+ + ": "
+ + combined
+ )
+
+
+def _verify_kerberos_keytab_in_hadoop(username):
+ hadoop = get_docker_client().containers.get(HADOOP_CONTAINER)
+ principal = user_kerberos_principal(username)
+ keytab = user_keytab_in_service_container(username)
+ shell = (
+ "kdestroy -A 2>/dev/null || true; "
+ "kinit -kt "
+ + keytab
+ + " "
+ + principal
+ + "; klist"
+ )
+ exit_code, output = hadoop.exec_run(
+ ["bash", "-c", shell], user=username, demux=True
+ )
+ combined = ((output[0] or b"") + (output[1] or b"")).decode()
+ if exit_code != 0:
+ raise RuntimeError(
+ "Kerberos kinit failed in "
+ + HADOOP_CONTAINER
+ + " for "
+ + principal
+ + ": "
+ + combined
+ )
diff --git a/functional-tests/ranger_test_utils/utils.py
b/functional-tests/ranger_test_utils/utils.py
index d751f5a62..a791aca8b 100644
--- a/functional-tests/ranger_test_utils/utils.py
+++ b/functional-tests/ranger_test_utils/utils.py
@@ -38,6 +38,9 @@
DEFAULT_ADMIN_AUTH = ("admin", "rangerR0cks!")
+# Avoid indefinite hangs when Ranger Admin or Docker is not up (plugin session
setup).
+RANGER_ADMIN_REQUEST_TIMEOUT_SEC = 30
+
DEFAULT_JSON_HEADERS = {
"Accept": "application/json",
"Content-Type": "application/json",
@@ -121,6 +124,14 @@ def assert_http_ok(response, ok_codes, step, url=""):
raise AssertionError(message)
+class _RangerAdminSession(requests.Session):
+ """requests.Session with a default timeout on every call."""
+
+ def request(self, method, url, **kwargs):
+ kwargs.setdefault("timeout", RANGER_ADMIN_REQUEST_TIMEOUT_SEC)
+ return super().request(method, url, **kwargs)
+
+
def create_ranger_admin_session(auth=None, headers=None):
"""
Build a requests.Session for Ranger Admin REST calls (Docker defaults).
@@ -131,12 +142,42 @@ def create_ranger_admin_session(auth=None, headers=None):
auth = DEFAULT_ADMIN_AUTH
if headers is None:
headers = DEFAULT_JSON_HEADERS
- session = requests.Session()
+ session = _RangerAdminSession()
session.auth = auth
session.headers.update(headers)
return session
+def preflight_ranger_admin_http():
+ """Skip plugin tests when Ranger Admin is not reachable on Docker
defaults."""
+ import pytest
+
+ url = RANGER_ADMIN_BASE_URL + "/plugins/definitions?page=0&pageSize=1"
+ try:
+ response = requests.get(
+ url,
+ auth=DEFAULT_ADMIN_AUTH,
+ headers=DEFAULT_JSON_HEADERS,
+ timeout=RANGER_ADMIN_REQUEST_TIMEOUT_SEC,
+ )
+ except requests.RequestException as exc:
+ pytest.skip(
+ "Ranger Admin is not reachable at "
+ + RANGER_ADMIN_BASE_URL
+ + " ("
+ + str(exc)
+ + "). Start the stack: cd functional-tests && ./run-tests.sh
postgres hive"
+ )
+ if response.status_code not in (200, 401, 403):
+ pytest.skip(
+ "Ranger Admin returned HTTP "
+ + str(response.status_code)
+ + " at "
+ + url
+ + ". Wait for containers to finish starting, then retry."
+ )
+
+
# --- Test run IDs ---
diff --git a/functional-tests/readme.md b/functional-tests/readme.md
index c9f9e25af..a4f296f4a 100644
--- a/functional-tests/readme.md
+++ b/functional-tests/readme.md
@@ -218,6 +218,7 @@ Brought up based on what you enter at the containers prompt
(or `SELECTED_CONTAI
- Tests assert on client output only (beeline, HBase shell, Kafka console
producer). Access-audit checks are deferred; helpers are in
`ranger_test_utils/access_audit_utils.py`.
- `hive` and `hbase` start `ranger-hadoop` automatically. `kafka` starts
`ranger-kafka`.
- Example: `SELECTED_CONTAINERS="hive hbase kafka" ./run-tests.sh postgres
hive hbase kafka`
+- **Shared plugin user/Kerberos setup** (session hook, markers, flow
diagrams): [docs/plugin-enforcement-setup.md](docs/plugin-enforcement-setup.md)
### HDFS suite notes
