Apache Ranger + Trino: Centralised Data Security for a Banking Governance Pipeline

docker compose --project-directory C:\reg_repo\platform-solution\corep-governance-pipeline ps | grep ranger
# Expected: ranger   running   0.0.0.0:6080->6080/tcp

# Default credentials — CHANGE IN PRODUCTION
# URL: http://localhost:6080
# User: admin   Password: rangerR0cks!

# docker/trino/ranger-trino-plugin/ranger-trino-security.xml
# This file tells the Trino plugin where to find the Ranger Admin server

<?xml version="1.0" encoding="UTF-8"?>
<configuration>

  <property>
    <name>ranger.plugin.trino.service.name</name>
    <value>corep-trino</value>
    <!-- Must match the service name created in Ranger Admin UI -->
  </property>

  <property>
    <name>ranger.plugin.trino.policy.source.impl</name>
    <value>org.apache.ranger.admin.client.RangerAdminRESTClient</value>
  </property>

  <property>
    <name>ranger.plugin.trino.policy.rest.url</name>
    <value>http://ranger:6080</value>
    <!-- ranger is the Docker service name -->
  </property>

  <property>
    <name>ranger.plugin.trino.policy.pollIntervalMs</name>
    <value>30000</value>
    <!-- Trino pulls policy updates every 30 seconds -->
  </property>

  <property>
    <name>ranger.plugin.trino.access.cluster.name</name>
    <value>corep-governance-pipeline</value>
  </property>

</configuration>

# docker/trino/ranger-trino-plugin/ranger-trino-audit.xml
# Configures audit log destination — write to local file + MinIO S3

<?xml version="1.0" encoding="UTF-8"?>
<configuration>

  <property>
    <name>xasecure.audit.is.enabled</name>
    <value>true</value>
  </property>

  <property>
    <name>xasecure.audit.solr.is.enabled</name>
    <value>false</value>
    <!-- Not using Solr in this stack -->
  </property>

  <property>
    <name>xasecure.audit.hdfs.is.enabled</name>
    <value>false</value>
  </property>

  <property>
    <name>xasecure.audit.log4j.is.enabled</name>
    <value>true</value>
    <!-- Write to container log — Docker log driver captures it -->
  </property>

  <property>
    <name>xasecure.audit.destination.db.is.enabled</name>
    <value>true</value>
  </property>

  <property>
    <name>xasecure.audit.destination.db.jdbc.url</name>
    <value>jdbc:postgresql://postgres:5432/corep</value>
  </property>

  <property>
    <name>xasecure.audit.destination.db.jdbc.driver</name>
    <value>org.postgresql.Driver</value>
  </property>

  <property>
    <name>xasecure.audit.destination.db.user</name>
    <value>corep_admin</value>
  </property>

  <property>
    <name>xasecure.audit.destination.db.password</name>
    <value>${POSTGRES_PASSWORD}</value>
    <!-- Injected from Docker env at container start -->
  </property>

</configuration>

# catalog/scripts/create_ranger_service.py
# Run once to register the Trino service in Ranger Admin

import requests, os, json

RANGER_URL = os.environ.get("RANGER_URL", "http://localhost:6080")
RANGER_AUTH = (
    os.environ.get("RANGER_ADMIN_USER", "admin"),
    os.environ.get("RANGER_ADMIN_PASSWORD", "rangerR0cks!"),
)
HEADERS = {"Content-Type": "application/json", "Accept": "application/json"}

service_payload = {
    "name":        "corep-trino",
    "displayName": "COREP Governance Pipeline — Trino",
    "type":        "trino",
    "description": "Trino query engine for COREP regulatory pipeline. All analyst queries go through this service.",
    "isEnabled":   True,
    "configs": {
        "username":    "trino_ranger_user",  # service account Ranger uses to validate resources
        "password":    os.environ.get("TRINO_RANGER_PASSWORD", ""),
        "jdbc.driverClassName": "io.trino.jdbc.TrinoDriver",
        "jdbc.url":    "jdbc:trino://trino:8080",
    },
}

r = requests.post(
    f"{RANGER_URL}/service/plugins/services",
    json=service_payload,
    headers=HEADERS,
    auth=RANGER_AUTH,
)
r.raise_for_status()
service_id = r.json()["id"]
print(f"Ranger service created: corep-trino (id={service_id})")

# catalog/scripts/create_ranger_roles.py

ROLES = [
    {"name": "corep_reporting", "description": "Regulatory reporting team — COREP submission"},
    {"name": "risk_analyst",     "description": "Risk management — capital and liquidity analysis"},
    {"name": "data_engineer",   "description": "Pipeline engineering team — full access with audit"},
    {"name": "auditor",         "description": "Internal/external audit and ECB SREP inspectors"},
    {"name": "pipeline_service","description": "Airflow DAG service account"},
]

for role in ROLES:
    r = requests.post(
        f"{RANGER_URL}/service/roles/roles",
        json={
            "name":        role["name"],
            "description": role["description"],
            "users":       [],  # assign users via LDAP/SCIM in production
            "groups":      [],
            "roles":       [],
        },
        headers=HEADERS,
        auth=RANGER_AUTH,
    )
    r.raise_for_status()
    print(f"  Role created: {role['name']}")

# catalog/scripts/create_ranger_policies.py — column masking policies

def create_column_mask_policy(
    name: str,
    schema: str,
    table: str,
    column: str,
    roles: list,
    mask_type: str,       # "MASK" (hash), "MASK_SHOW_FIRST_4", "MASK_NULL", "CUSTOM"
    mask_expr: str = "",  # for CUSTOM mask type only
) -> dict:
    return {
        "service":     "corep-trino",
        "name":        name,
        "policyType":  1,  # 0=Access, 1=Masking, 2=RowFilter
        "description": f"Column masking: {schema}.{table}.{column}",
        "isEnabled":   True,
        "resources": {
            "catalog":    {"values": ["postgresql"], "isExcludes": False, "isRecursive": False},
            "schema":     {"values": [schema],       "isExcludes": False, "isRecursive": False},
            "table":      {"values": [table],        "isExcludes": False, "isRecursive": False},
            "column":     {"values": [column],       "isExcludes": False, "isRecursive": False},
        },
        "dataMaskPolicyItems": [
            {
                "dataMaskInfo": {
                    "dataMaskType": mask_type,
                    "valueExpr":   mask_expr,
                },
                "roles":      roles,
                "groups":     [],
                "users":      [],
                "accesses":   [{"type": "select", "isAllowed": True}],
                "conditions": [],
                "delegateAdmin": False,
            }
        ],
    }


MASK_POLICIES = [
    # LEI → SHA-256 hash for analysts (pseudonymised, not anonymised)
    create_column_mask_policy(
        name="mask-lei-analyst",
        schema="raw", table="counterparties", column="lei",
        roles=["risk_analyst", "auditor"],
        mask_type="MASK",   # Ranger built-in SHA-256 hash
    ),
    # Counterparty name → first 3 chars + *** for analysts
    create_column_mask_policy(
        name="mask-name-analyst",
        schema="raw", table="counterparties", column="name",
        roles=["risk_analyst", "auditor"],
        mask_type="MASK_SHOW_FIRST_4",
    ),
    # Borrower name in loans → NULL for all non-engineer roles
    create_column_mask_policy(
        name="mask-borrower-name",
        schema="raw", table="loans", column="borrower_name",
        roles=["risk_analyst", "corep_reporting", "auditor"],
        mask_type="MASK_NULL",
    ),
    # Borrower ID → hash (still usable for COUNT DISTINCT, GROUP BY)
    create_column_mask_policy(
        name="mask-borrower-id",
        schema="raw", table="loans", column="borrower_id",
        roles=["risk_analyst", "corep_reporting", "auditor"],
        mask_type="MASK",
    ),
]

for policy in MASK_POLICIES:
    r = requests.post(
        f"{RANGER_URL}/service/plugins/policies",
        json=policy, headers=HEADERS, auth=RANGER_AUTH,
    )
    r.raise_for_status()
    print(f"  Masking policy created: {policy['name']}")

# catalog/scripts/create_ranger_policies.py — row filter policies

def create_row_filter_policy(
    name: str,
    schema: str,
    table: str,
    roles: list,
    filter_expr: str,
) -> dict:
    return {
        "service":     "corep-trino",
        "name":        name,
        "policyType":  2,   # 2 = RowFilter
        "description": f"Row filter: {schema}.{table} for {', '.join(roles)}",
        "isEnabled":   True,
        "resources": {
            "catalog": {"values": ["postgresql"], "isExcludes": False, "isRecursive": False},
            "schema":  {"values": [schema],       "isExcludes": False, "isRecursive": False},
            "table":   {"values": [table],        "isExcludes": False, "isRecursive": False},
        },
        "rowFilterPolicyItems": [
            {
                "rowFilterInfo": {"filterExpr": filter_expr},
                "roles":       roles,
                "groups":      [],
                "users":       [],
                "accesses":    [{"type": "select", "isAllowed": True}],
                "conditions":  [],
                "delegateAdmin": False,
            }
        ],
    }


ROW_FILTER_POLICIES = [
    create_row_filter_policy(
        name="filter-rwa-credit-only",
        schema="raw", table="rwa_exposures",
        roles=["risk_analyst"],
        filter_expr="exposure_class IN ('central_governments','institutions','corporates','retail','real_estate')",
    ),
    create_row_filter_policy(
        name="filter-capital-valid-tiers",
        schema="raw", table="capital_instruments",
        roles=["risk_analyst", "corep_reporting"],
        filter_expr="tier IN ('CET1', 'AT1', 'T2')",
    ),
    create_row_filter_policy(
        name="filter-liquidity-hqla-only",
        schema="raw", table="liquidity_assets",
        roles=["risk_analyst"],
        filter_expr="hqla_level IN ('1', '2A', '2B')",
    ),
]

for policy in ROW_FILTER_POLICIES:
    r = requests.post(
        f"{RANGER_URL}/service/plugins/policies",
        json=policy, headers=HEADERS, auth=RANGER_AUTH,
    )
    r.raise_for_status()
    print(f"  Row filter policy created: {policy['name']}")

# Schema-level access policy — corep_reporting can see mart.* and intermediate.*
# but NOT raw.* (only the data_engineer and pipeline_service can)

ACCESS_POLICIES = [
    {
        "service":    "corep-trino",
        "name":       "allow-mart-to-reporting",
        "policyType": 0,
        "isEnabled":  True,
        "resources": {
            "catalog": {"values": ["postgresql"]},
            "schema":  {"values": ["mart", "intermediate"]},
            "table":   {"values": ["*"]},
            "column":  {"values": ["*"]},
        },
        "policyItems": [{
            "roles":   ["corep_reporting", "risk_analyst", "auditor"],
            "accesses": [{"type": "select", "isAllowed": True}],
            "conditions": [], "delegateAdmin": False,
        }],
    },
    {
        "service":    "corep-trino",
        "name":       "allow-raw-to-engineers-only",
        "policyType": 0,
        "isEnabled":  True,
        "resources": {
            "catalog": {"values": ["postgresql"]},
            "schema":  {"values": ["raw"]},
            "table":   {"values": ["*"]},
            "column":  {"values": ["*"]},
        },
        "policyItems": [{
            "roles":   ["data_engineer", "pipeline_service"],
            "accesses": [
                {"type": "select",    "isAllowed": True},
                {"type": "insert",    "isAllowed": True},
                {"type": "update",    "isAllowed": True},
                {"type": "delete",    "isAllowed": True},
                {"type": "create",    "isAllowed": True},
                {"type": "drop",      "isAllowed": True},
            ],
            "conditions": [], "delegateAdmin": False,
        }],
        # Explicit deny for everyone else — deny overrides allow in Ranger
        "denyPolicyItems": [{
            "roles":   ["risk_analyst", "corep_reporting", "auditor"],
            "accesses": [{"type": "select", "isAllowed": True}],
            "conditions": [], "delegateAdmin": False,
        }],
    },
    {
        "service":    "corep-trino",
        "name":       "allow-audit-schema-to-auditors",
        "policyType": 0,
        "isEnabled":  True,
        "resources": {
            "catalog": {"values": ["postgresql"]},
            "schema":  {"values": ["audit"]},
            "table":   {"values": ["*"]},
            "column":  {"values": ["*"]},
        },
        "policyItems": [{
            "roles":   ["auditor", "data_engineer"],
            "accesses": [{"type": "select", "isAllowed": True}],
            "conditions": [], "delegateAdmin": False,
        }],
    },
]

for policy in ACCESS_POLICIES:
    r = requests.post(
        f"{RANGER_URL}/service/plugins/policies",
        json=policy, headers=HEADERS, auth=RANGER_AUTH,
    )
    r.raise_for_status()
    print(f"  Access policy created: {policy['name']}")

"""
modules/security.py — Apply and verify Ranger security policies as a pipeline step.

This module:
  - Creates the Ranger service if it does not exist
  - Pushes all column masking, row filter, and access allow policies
  - Verifies policies are active by polling the Ranger API
  - Exports current policy state to MinIO for audit evidence
"""

import json, logging, os, time
from pathlib import Path

import requests

from modules.base import BaseModule

log = logging.getLogger(__name__)

RANGER_URL   = os.environ.get("RANGER_URL", "http://localhost:6080")
RANGER_AUTH  = (
    os.environ.get("RANGER_ADMIN_USER",     "admin"),
    os.environ.get("RANGER_ADMIN_PASSWORD",  "rangerR0cks!"),
)
HEADERS      = {"Content-Type": "application/json", "Accept": "application/json"}
POLICY_DIR   = Path(os.environ.get("RANGER_POLICY_DIR", "catalog/ranger_policies"))


class SecurityModule(BaseModule):
    MODULE_NAME = "security"

    # Policy JSON files — loaded from POLICY_DIR
    # Each file is a Ranger policy payload (policyType 0/1/2)
    _POLICY_FILES = [
        "access_allow_policies.json",
        "column_mask_policies.json",
        "row_filter_policies.json",
    ]

    def input_check(self) -> None:
        """Verify Ranger Admin is reachable and policy files exist."""
        try:
            r = requests.get(
                f"{RANGER_URL}/service/plugins/services",
                auth=RANGER_AUTH, timeout=10
            )
            r.raise_for_status()
        except Exception as exc:
            raise RuntimeError(
                f"[security] Ranger Admin not reachable at {RANGER_URL}: {exc}"
            )

        missing = [
            str(POLICY_DIR / f)
            for f in self._POLICY_FILES
            if not (POLICY_DIR / f).exists()
        ]
        if missing:
            raise RuntimeError(
                "[security] Missing policy JSON files: " + ", ".join(missing)
                + ". Run: python catalog/scripts/export_ranger_policies.py"
            )
        log.info("[security] Ranger reachable. %d policy files present.", len(self._POLICY_FILES))

    def _execute(self) -> None:
        """Push all policies to Ranger. Existing policies are updated (PUT), new ones created (POST)."""
        for policy_file in self._POLICY_FILES:
            policies = json.loads((POLICY_DIR / policy_file).read_text())
            log.info("[security] Applying %d policies from %s", len(policies), policy_file)

            for policy in policies:
                existing = self._get_existing_policy(policy["name"])
                if existing:
                    policy["id"] = existing["id"]
                    r = requests.put(
                        f"{RANGER_URL}/service/plugins/policies/{existing['id']}",
                        json=policy, headers=HEADERS, auth=RANGER_AUTH,
                    )
                else:
                    r = requests.post(
                        f"{RANGER_URL}/service/plugins/policies",
                        json=policy, headers=HEADERS, auth=RANGER_AUTH,
                    )
                r.raise_for_status()
                log.info(
                    "[security] Policy %s: %s",
                    policy["name"],
                    "updated" if existing else "created",
                )

        time.sleep(5)   # Trino polls Ranger every 30s; allow propagation before output_check
        self._export_policies_to_minio()

    def _get_existing_policy(self, name: str) -> dict | None:
        r = requests.get(
            f"{RANGER_URL}/service/plugins/policies",
            params={"serviceName": "corep-trino", "policyName": name},
            headers=HEADERS, auth=RANGER_AUTH,
        )
        r.raise_for_status()
        policies = r.json().get("policies", [])
        return policies[0] if policies else None

    def _export_policies_to_minio(self) -> None:
        """Export current Ranger policy state to MinIO for versioned audit evidence."""
        try:
            from minio import Minio
            import io
            from datetime import datetime, timezone

            r = requests.get(
                f"{RANGER_URL}/service/plugins/policies",
                params={"serviceName": "corep-trino"},
                headers=HEADERS, auth=RANGER_AUTH,
            )
            r.raise_for_status()
            policy_json = json.dumps(r.json(), indent=2).encode()

            client = Minio(
                os.environ.get("MINIO_ENDPOINT", "minio:9000"),
                access_key=os.environ.get("MINIO_ROOT_USER", "minioadmin"),
                secret_key=os.environ.get("MINIO_ROOT_PASSWORD", "minioadmin"),
                secure=False,
            )
            bucket = "corep-ranger-audit"
            if not client.bucket_exists(bucket):
                client.make_bucket(bucket)

            ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
            object_name = f"policies/ranger_policies_{ts}.json"
            client.put_object(
                bucket, object_name,
                io.BytesIO(policy_json), length=len(policy_json),
                content_type="application/json",
            )
            log.info("[security] Ranger policies exported → minio://%s/%s", bucket, object_name)
        except Exception as exc:
            log.warning("[security] MinIO export failed (non-fatal): %s", exc)

    def emit_lineage(self) -> None:
        log.info("[security] No lineage event — policy enforcement is orthogonal to data lineage.")

    def output_check(self) -> None:
        """Verify all expected policy names are active in Ranger."""
        expected = {"mask-lei-analyst", "mask-name-analyst", "mask-borrower-name",
                    "filter-rwa-credit-only", "allow-mart-to-reporting",
                    "allow-raw-to-engineers-only", "allow-audit-schema-to-auditors"}

        r = requests.get(
            f"{RANGER_URL}/service/plugins/policies",
            params={"serviceName": "corep-trino"},
            headers=HEADERS, auth=RANGER_AUTH,
        )
        r.raise_for_status()
        active = {p["name"] for p in r.json().get("policies", [])}

        missing = expected - active
        if missing:
            raise RuntimeError(
                f"[security] Policies not active in Ranger: {', '.join(missing)}"
            )
        log.info("[security] All %d policies verified active in Ranger.", len(expected))

-- Connect as risk_analyst role
-- Trino CLI: trino --user risk_analyst_user --catalog postgresql --schema raw

-- Test 1: Column masking on lei
SELECT instrument_id, lei, name
FROM raw.counterparties
LIMIT 3;

-- Expected output for risk_analyst_user:
-- instrument_id | lei                                                              | name
-- CP001         | a665a45920422f9d417e4867efdc4fb8a04a1f3fff1fa07e998e86f7f7a27ae | Deu***
-- CP002         | 2c624232cdd221771294dfbb310acbc8d2e4b5f0d1af56af893d69ccb20b43 | Bar***
-- CP003         | 19581e27de7ced00ff1ce50b2047e7a567c76b1cbaebabe5ef03f7c3017bb5b | Soc***
-- Note: lei is SHA-256 hash. name is truncated. GDPR enforced.

-- Test 2: Row filter on rwa_exposures
SELECT DISTINCT exposure_class FROM raw.rwa_exposures;
-- Expected: only credit classes — equity and other are filtered out

-- Test 3: Schema-level deny for raw schema (analyst tries to access raw directly)
SELECT * FROM raw.capital_instruments LIMIT 1;
-- Expected: ERROR: Access Denied: Cannot select from table: postgresql.raw.capital_instruments
-- (raw schema is blocked — analyst must go through mart)

-- Test 4: Mart access works for risk_analyst (unmasked, aggregated — no PII)
SELECT * FROM mart.corep_c0300;
-- Expected: full result — cet1_ratio, tier1_ratio, etc. No PII in mart.

-- Connect as corep_reporting role

-- Test 5: Reporting role sees full unmasked LEI
SELECT lei FROM raw.counterparties LIMIT 3;
-- Expected: 529900T8BM49AURSDO55  (real LEI — unmasked)
-- corep_reporting has the ALLOW policy for this column

-- Test 6: Reporting role cannot see raw.loans borrower data (masked to NULL)
SELECT borrower_name, borrower_id FROM raw.loans LIMIT 3;
-- Expected: NULL | a665a45920... (name NULL, id hashed)

-- Query Ranger audit log from Trino (audit schema in PostgreSQL)
-- Auditors can run this themselves — no Ranger admin access needed

SELECT
    event_time,
    request_user,
    access_type,
    resource_path,
    access_result,
    policy_id,
    tags
FROM audit.ranger_access_audit
WHERE
    event_time >= DATE_TRUNC('month', CURRENT_DATE)   -- current reporting month
    AND resource_path LIKE '%lei%'                     -- all LEI column accesses
ORDER BY event_time DESC
LIMIT 100;

-- Cross-check: any ALLOW on lei where user is not in corep_reporting or data_engineer?
SELECT DISTINCT request_user, access_result, policy_id
FROM audit.ranger_access_audit
WHERE resource_path LIKE '%lei%'
  AND access_result = 'ALLOWED'
  AND request_user NOT IN (
      SELECT username FROM audit.role_assignments
      WHERE role_name IN ('corep_reporting', 'data_engineer', 'pipeline_service')
  );
-- This query should always return 0 rows. If it doesn't, you have a policy gap.