This is an automated email from the ASF dual-hosted git repository. ramackri pushed a commit to branch ranger-kafka-dynamic-partition-plan in repository https://gitbox.apache.org/repos/asf/ranger.git
commit 6b4d09155eca6d1181cc40aeeeed50095d5a0f94 Author: ramk <[email protected]> AuthorDate: Tue Jun 16 22:01:36 2026 +0530 RANGER: Unify service allowlist into dynamic partition plan registry. Store plugins, buffer, and services in ranger_audit_partition_plan with one version and /api/audit/partition-plan REST (including onboard-repo); merge operator guides and extend partition-plan E2E to assert services on GET. Co-authored-by: Cursor <[email protected]> --- .../README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md | 135 ++--- .../README-DYNAMIC-SERVICE-ALLOWLIST-GUIDE.md | 650 +-------------------- .../README-KAFKA-DYNAMIC-PARTITION-PLAN.md | 2 + .../README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md | 268 ++++++--- .../README-KAFKA-PARTITION-PLAN-IMPLEMENTATION.md | 6 +- .../ranger/audit/server/AuditServerConstants.java | 1 + .../kafka/partition/PartitionPlanAllocator.java | 38 +- .../kafka/partition/PartitionPlanBootstrap.java | 1 + .../kafka/partition/PartitionPlanHolder.java | 26 + .../kafka/partition/PartitionPlanKafkaConfig.java | 18 + .../kafka/partition/PartitionPlanService.java | 24 +- .../kafka/partition/PartitionPlanValidator.java | 18 + .../kafka/partition/PartitionPlanWatcher.java | 2 + .../kafka/partition/ServiceAllowlistBootstrap.java | 97 +++ ...ePluginRequest.java => OnboardRepoRequest.java} | 21 +- .../kafka/partition/model/PartitionPlan.java | 35 +- .../model/PartitionPlanReplaceRequest.java | 34 +- .../partition/model/PromotePluginRequest.java | 21 +- .../partition/model/ServiceAllowlistEntry.java | 96 +++ .../org/apache/ranger/audit/rest/AuditREST.java | 159 +++-- .../resources/conf/ranger-audit-ingestor-site.xml | 9 + .../PartitionPlanServiceMutationTest.java | 39 ++ .../partition/ServiceAllowlistBootstrapTest.java | 94 +++ .../partition/model/PartitionPlanJsonTest.java | 16 + .../scripts/audit/verify-partition-plan-e2e.sh | 8 + 25 files changed, 964 insertions(+), 854 deletions(-) diff --git a/audit-server/README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md b/audit-server/README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md index dcebc000e..770fb3772 100644 --- a/audit-server/README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md +++ b/audit-server/README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md @@ -23,13 +23,13 @@ under the License. **Audience:** Architects, operators, and reviewers planning runtime plugin onboarding without ingestor restart. -**Operator guide (everyone):** [README-DYNAMIC-SERVICE-ALLOWLIST-GUIDE.md](README-DYNAMIC-SERVICE-ALLOWLIST-GUIDE.md) — plain-language companion to this design doc (mirrors [partition plan guide](README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md)). +**Operator guide (everyone):** [README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md](README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md) — unified ingestor registry guide (partition routing + service allowlist; [Confluence](https://cloudera.atlassian.net/wiki/spaces/ENG/pages/12043681813)). **Related docs:** | Doc | Relationship | |-----|----------------| -| [README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md](README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md) | Dynamic **Kafka routing** (partition plan) — orthogonal; companion operator guide | +| [README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md](README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md) | Unified operator guide (routing + allowlist in one registry) | | [DESIGN-KAFKA-DYNAMIC-PARTITIONING.md](DESIGN-KAFKA-DYNAMIC-PARTITIONING.md) | Partition plan architecture | | [README-KAFKA-PARTITION-PLAN-REGISTRY-REST.md](README-KAFKA-PARTITION-PLAN-REGISTRY-REST.md) | Partition-plan Kafka topic + REST | | [README-KAFKA-PARTITION-PLAN-IMPLEMENTATION.md](README-KAFKA-PARTITION-PLAN-IMPLEMENTATION.md) | Phase 6: **partition-plan admin** allowlist (separate surface) | @@ -214,32 +214,36 @@ Kerberos (or JWT/basic) success does **not** imply authorization. Surface 1 (`al --- -## 5. Proposed architecture +## 5. Proposed architecture (unified registry) -### 5.1 High-level (mirror partition plan) +### 5.1 Single durable document -| Component | Partition plan (exists / planned) | Service allowlist (proposed) | -|-----------|-----------------------------------|------------------------------| -| **Durable store** | Kafka compacted `ranger_audit_partition_plan` | Kafka compacted `ranger_audit_service_allowlist` (name TBD) | -| **In-memory cache** | `PartitionPlanService` + watcher | `ServiceAllowlistService` + watcher | -| **Bootstrap** | Seed v1 from site XML if topic empty | Seed from site XML `service.*.allowed.users` if topic empty | -| **Admin REST** | `GET/PUT /api/audit/partition-plan` | `GET/PUT /api/audit/service-allowlist` | -| **Request path** | `AuditPartitioner` reads cache | `AuditREST.isAllowedServiceUser()` reads cache | -| **AuthZ on REST** | `partition.plan.allowed.users` (Phase 6) | `service.allowlist.admin.users` (proposed) | +Service allowlist and partition routing share **one** Kafka compacted topic and **one** in-memory cache: + +| Component | Unified registry | +|-----------|------------------| +| **Durable store** | Kafka compacted `ranger_audit_partition_plan` (`plugins`, `buffer`, `services`) | +| **In-memory cache** | `PartitionPlanHolder` + `PartitionPlanWatcher` | +| **Bootstrap** | Seed v1 from site XML (`configured.plugins` + `service.*.allowed.users`) when topic empty | +| **Admin REST** | `GET/PUT/POST /api/audit/partition-plan` (+ `onboard-repo`) | +| **Plugin authz path** | `AuditREST.isAllowedServiceUser()` reads `plan.services` when dynamic enabled | +| **Routing path** | `AuditPartitioner` reads `plan.plugins` / `buffer` | +| **AuthZ on REST** | `kafka.partition.plan.allowed.users` (when configured) | + +No separate `ranger_audit_service_allowlist` topic or `/service-allowlist` REST surface. ### 5.2 Control plane vs data plane ```mermaid flowchart LR subgraph control [Control plane — rare updates] - AdminREST[Admin REST] - AllowTopic[(ranger_audit_service_allowlist compacted)] - PlanTopic[(ranger_audit_partition_plan compacted)] - Watcher[ServiceAllowlistWatcher] - Mem[(In-memory allowlist map)] - - AdminREST --> AllowTopic - Watcher --> AllowTopic + AdminREST["/api/audit/partition-plan"] + PlanTopic[(ranger_audit_partition_plan)] + Watcher[PartitionPlanWatcher] + Mem[(PartitionPlanHolder)] + + AdminREST --> PlanTopic + Watcher --> PlanTopic Watcher --> Mem end @@ -249,46 +253,42 @@ flowchart LR Kafka[(ranger_audits)] Plugin --> REST - REST -->|isAllowedServiceUser| Mem + REST -->|isAllowedServiceUser from services| Mem REST -->|on success| Kafka end ``` -**Takeaway:** Allowlist topic = small, rare config (like partition plan). Audit events still flow through `ranger_audits`; dispatchers unchanged. +**Takeaway:** One compacted config topic; audit events still flow through `ranger_audits`; dispatchers unchanged. -### 5.3 Registry document shape (proposed JSON) +### 5.3 Registry document shape (unified JSON) -Key: audit topic name or fixed key `default` (single global allowlist document). +Kafka key: audit topic name (e.g. `ranger_audits`). ```json { - "version": 3, + "topic": "ranger_audits", + "version": 8, + "topicPartitionCount": 48, "updatedAt": "2026-06-15T12:00:00Z", "updatedBy": "admin", + "plugins": { + "hive": { "partitions": [6, 7, 8] } + }, + "buffer": { "partitions": [40, 41, 42] }, "services": { - "dev_hive": { - "allowedUsers": ["hive"], - "source": "policy-manager", - "notes": "synced from dev_hive policy.download.auth.users" - }, - "dev_trino": { - "allowedUsers": ["trino"], - "source": "rest", - "notes": "onboarded via PUT /service-allowlist" - }, - "dev_ozone": { - "allowedUsers": ["om", "ozone"], - "source": "rest" - } + "dev_hive": { "allowedUsers": ["hive"] }, + "dev_trino": { "allowedUsers": ["trino"], "source": "rest" }, + "dev_ozone": { "allowedUsers": ["om", "ozone"] } } } ``` **Rules:** -- `version` monotonic; `PUT` uses optimistic concurrency (`expectedVersion`) like partition plan. -- `allowedUsers` = short names **after** `auth_to_local` (same semantics as today’s XML values). -- Empty or missing repo entry → **403** for that `serviceName` (fail closed). +- Single `version` for routing **and** allowlist mutations (`expectedVersion` on all writes). +- `allowedUsers` = short names **after** `auth_to_local`. +- Missing repo in `services` when registry is active → **403** for that `serviceName` (fail closed). +- Brownfield plans without `services`: lazy in-memory merge from XML until first admin PUT. --- @@ -317,62 +317,41 @@ AuditREST.isAllowedServiceUser(serviceName, authenticatedUser) - Remove `static final allowedServiceUsers`; inject `ServiceAllowlistService` (same pattern as `PartitionPlanService`). - Watcher polls or consumes compacted topic; on change, swap in-memory map atomically. - **Read path:** O(1) map lookup; no Kafka read per audit POST. -- **Fallback:** If dynamic mode disabled (`service.allowlist.dynamic.enabled=false`), keep current static XML behavior. +- **Fallback:** If dynamic mode disabled (`kafka.partition.plan.dynamic.enabled=false`), keep current static XML behavior. --- -## 7. Admin REST API (proposed) +## 7. Admin REST API (unified `/partition-plan`) ### Endpoints | Method | Path | Auth | Description | |--------|------|------|-------------| -| `GET` | `/api/audit/service-allowlist` | Admin allowlist | Return current document + version | -| `PUT` | `/api/audit/service-allowlist` | Admin allowlist | Replace or merge; write to Kafka topic | -| `PUT` | `/api/audit/service-allowlist/services/{repo}` | Admin allowlist | Upsert one repo (convenience) | -| `DELETE` | `/api/audit/service-allowlist/services/{repo}` | Admin allowlist | Remove repo (optional; fail-closed for plugins) | - -**PUT body (full replace example):** - -```json -{ - "expectedVersion": 2, - "services": { - "dev_trino": { - "allowedUsers": ["trino"] - } - } -} -``` - -**Responses:** Same patterns as partition plan — `200` OK, `409` version conflict, `401` unauthenticated, `403` not admin. - -### Unified **onboard plugin** API (optional Phase 2) +| `GET` | `/api/audit/partition-plan` | `partition.plan.allowed.users` | Return full plan (`plugins`, `buffer`, `services`, `version`) | +| `PUT` | `/api/audit/partition-plan` | Admin | Replace plan; optional `services` map in body | +| `POST` | `/api/audit/partition-plan/promote` | Admin | Promote plugin; optional `repo` + `allowedUsers` | +| `POST` | `/api/audit/partition-plan/onboard-repo` | Admin | Upsert `services[repo]` + promote plugin (one version) | +| `POST` | `/api/audit/partition-plan/scale` | Admin | Append tail partitions (routing only) | -Single operation for ops when adding `dev_trino`: +**Onboard repo example:** ```http -POST /api/audit/onboard-plugin +POST /api/audit/partition-plan/onboard-repo ``` ```json { "repo": "dev_trino", - "appIds": ["trino"], + "pluginId": "trino", "allowedUsers": ["trino"], "partitionCount": 3, - "expectedPlanVersion": 5, - "expectedAllowlistVersion": 2 + "expectedVersion": 5 } ``` -**Atomic intent (best-effort across two topics):** +**Atomic intent:** Single Kafka write bumps one `version` covering both `services` and `plugins` — no cross-topic ordering problem. -1. Upsert `dev_trino` in service allowlist registry. -2. Append/allocate partition range for `trino` in partition plan. -3. Optionally grow `ranger_audits` partition count. - -If step 2 succeeds and step 1 fails, return `207 Multi-Status` with rollback guidance (ops retry allowlist). True distributed transaction is **not** required for v1; document ordering: **allowlist first, then plan** (fail closed: no audits accepted until allowlist exists). +**Responses:** Same patterns as partition plan — `200` OK, `409` version conflict, `401` unauthenticated, `403` not admin. --- @@ -386,7 +365,7 @@ On service create/update REST handler in `security-admin`: 1. Read `policy.download.auth.users` for the service. 2. Map Kerberos principals → short names (same rules as ingestor `auth_to_local`, or store short names only in service def). -3. Call ingestor `PUT /api/audit/service-allowlist/services/{repo}` (internal HTTP, service account). +3. Call ingestor `POST /api/audit/partition-plan/onboard-repo` (internal HTTP, service account). **Pros:** Single source of truth in Policy Manager. **Cons:** Requires Admin → ingestor coupling and credentials. @@ -536,7 +515,7 @@ Set `dynamic.enabled=false` and restart; static XML values apply again. Kafka to ## 15. Open questions for review -1. **Topic naming:** `ranger_audit_service_allowlist` vs single document key in an existing config topic? +1. **Topic naming:** **Resolved** — embed `services` in existing `ranger_audit_partition_plan` document (no second topic). 2. **Strict vs permissive Admin sync:** Reject allowlist users not in `policy.download.auth.users`? 3. **Delete repo:** Should DELETE remove partition plan entry too (onboard-plugin rollback)? 4. **JWT/basic auth:** Allowlist still keyed by short username; document that JWT subjects must match allowlist entries. diff --git a/audit-server/README-DYNAMIC-SERVICE-ALLOWLIST-GUIDE.md b/audit-server/README-DYNAMIC-SERVICE-ALLOWLIST-GUIDE.md index d05befd8e..24bd279ae 100644 --- a/audit-server/README-DYNAMIC-SERVICE-ALLOWLIST-GUIDE.md +++ b/audit-server/README-DYNAMIC-SERVICE-ALLOWLIST-GUIDE.md @@ -17,651 +17,15 @@ specific language governing permissions and limitations under the License. --> -# Dynamic service allowlist for Ranger audit-ingestor — guide for everyone +# Dynamic service allowlist — merged into unified ingestor registry guide -This guide explains **why** and **how** Ranger can move from a **static** per-repo allowlist (loaded once at ingestor startup) to a **dynamic** allowlist (change at runtime without restarting the audit ingestor). +**This guide has been merged** into the unified operator guide: -It is written for operators, architects, and reviewers who need a shared mental model — **without reading the codebase**. +- **Repo:** [README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md](README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md) — *Dynamic ingestor registry for Ranger audit-ingestor* +- **Confluence:** [Dynamic Ingestor Registry Guide](https://cloudera.atlassian.net/wiki/spaces/ENG/pages/12043681813) (page `12043681813`) -**Confluence:** [Dynamic Service Allowlist Guide (Ranger Audit Ingestor)](https://cloudera.atlassian.net/wiki/spaces/ENG/pages/12055576591/Dynamic+Service+Allowlist+Guide+Ranger+Audit+Ingestor) (child of [Ranger Engineering](https://cloudera.atlassian.net/wiki/spaces/ENG/pages/759726545/Ranger+Engineering)) +Service allowlist (`services` map) and Kafka partition routing (`plugins` / `buffer`) now share one document on **`ranger_audit_partition_plan`** and one REST surface: **`/api/audit/partition-plan`** (including `POST .../onboard-repo`). -**Status:** Proposal (not fully implemented). Complements [Dynamic Kafka Partitioning Guide](README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md) ([Confluence](https://cloudera.atlassian.net/wiki/spaces/ENG/pages/12043681813/Dynamic+Kafka+Partitioning+Guide+Ranger+Audit+Plugins)); does **not** replace authorization on `POST /api/audit/access`. +**Design detail (allowlist authorization layers, security, API):** [README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md](README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md) -**Related design doc:** [README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md](README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md) - ---- - -## 1. What problem are we solving? - -Ranger plugins (HDFS, Hive, Ozone OM, Trino, etc.) POST audit events to **audit-ingestor**: - -```http -POST /api/audit/access?serviceName=<repo>&appId=<agent> -``` - -After Kerberos/JWT/basic **authentication** (401 on failure), ingestor performs **authorization**: the authenticated short username must appear in the allowlist for that repo. - -**Today (static mode):** - -- Who may POST audits for repo `dev_hive` is defined in **XML** at startup: - - `ranger.audit.ingestor.service.dev_hive.allowed.users` = `hive` -- Adding a new Ranger service repo (e.g. `dev_trino`) means: edit `ranger-audit-ingestor-site.xml` → **restart ingestor** on every pod. -- [PR #1017](https://github.com/apache/ranger/pull/1017) (RANGER-5645) added missing static entries for Docker E2E (e.g. `dev_ozone` needed `om`). - -**Goal (dynamic mode):** - -- Change **who may claim audits for a repo** while ingestor is running. -- Onboard a new repo without rolling restart. -- Keep all ingestor replicas consistent via a shared durable registry (same pattern as the partition plan). - -**What this does *not* solve:** Kafka partition routing. That is the [partition plan guide](README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md) ([Confluence](https://cloudera.atlassian.net/wiki/spaces/ENG/pages/12043681813/Dynamic+Kafka+Partitioning+Guide+Ranger+Audit+Plugins)). Dynamic partition mapping does **not** remove the allowlist check — and that is **intentional**. - ---- - -## 2. Core ideas (plain language) - -### Service repo name (`serviceName`) - -The `serviceName` query parameter on `/access` is the **Ranger Policy Manager service name** — e.g. `dev_hive`, `dev_ozone`, `dev_trino`. It must match the repo name in Ranger Admin, not the service *type* (`hive`, `ozone`). - -### Short username (after `auth_to_local`) - -Ingestor maps the Kerberos principal to a **short name** before checking the allowlist — e.g. `hive/[email protected]` → `hive`. - -Allowlist values are **short names**, not full principals. - -### Service allowlist: `ranger.audit.ingestor.service.<repo>.allowed.users` - -This is the authorization gate on every plugin audit POST. - -| Item | Detail | -|------|--------| -| **Property** | `ranger.audit.ingestor.service.<repo>.allowed.users` | -| **Example** | `ranger.audit.ingestor.service.dev_hive.allowed.users` = `hive` | -| **Caller** | Service daemons (`hive`, `hdfs`, `om`, `trino`, …) | -| **When checked** | On **every** audit POST, **before** Kafka produce | -| **Failure** | **403** (authentication already succeeded) | - -**What it prevents:** any principal with a valid Kerberos ticket from forging audits for a repo it does not own. Without this check, a compromised `kafka` daemon could POST events as `serviceName=dev_hive`. - -### Partition plan (orthogonal — different guide) - -After the service allowlist check passes, ingestor publishes to Kafka and the **partition plan** decides *which partition*. See [README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md](README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md). - -A plugin can have a valid partition range and still get **403** on `/access` if its short name is not in the service allowlist. - -### Service allowlist registry (proposed source of truth) - -The live allowlist lives in **durable shared storage** that all ingestor replicas read — a Kafka **compacted** topic (`ranger_audit_service_allowlist`). - -- No new database dependency. -- Survives pod restarts. -- All replicas see the same latest allowlist document. - -### Consistency rule (ops discipline — Phase 1) - -```text -allowedUsers for repo R ⊆ { short names from policy.download.auth.users for R in Ranger Admin } -``` - -Ranger Admin does **not** sync this automatically in Phase 1 (Option C). Operators copy the right short names when onboarding. Ingestor may **reject** REST writes that violate the subset rule (strict mode). - -### Three layers — do not merge - -| Layer | Who | Purpose | -|---------|-----|---------| -| **Service allowlist** (plugin POST) | Daemons (`hive`, `om`, …) | May this principal POST audits for repo `R`? | -| **Partition plan** (Kafka routing) | Ingestor internal | Which Kafka partition after accept? | -| **Admin REST APIs** | Ops (`admin`, `ops`, …) | Who may change allowlist or partition plan via REST? | - -Plugins must **not** call allowlist-admin or partition-plan REST. One combined list would let a daemon rewire routing or impersonate other repos. - ---- - -## 3. Today vs proposed (at a glance) - -| | Static (today) | Dynamic (proposed) | -|---|----------------|---------------------| -| **Where allowlist lives** | XML on each pod at startup | Shared document in Kafka compacted topic | -| **Change allowlist** | Edit XML + restart | REST API; no restart | -| **Add new repo `dev_trino`** | New XML property + restart all pods | `PUT /service-allowlist` or seed topic | -| **Multi-replica ingestor** | Same XML if synced via ConfigMap | All pods watch same Kafka topic | -| **Relation to partition plan** | Separate manual step | Still separate; optional `onboard-plugin` bundles both later | -| **Authorization on `/access`** | Required | **Still required** | -| **Feature flag** | Default behavior | `ranger.audit.ingestor.service.allowlist.dynamic.enabled=true` | - ---- - -## 4. Request path — why partition plan does not replace allowlist - -```text -Plugin POST /api/audit/access?serviceName=dev_hive&appId=hiveServer2 - │ - ├─ 401 Authentication failed (Kerberos / JWT / basic) - │ - ├─ 403 Service allowlist: auth_to_local(principal) → "hive" - │ "hive" ∉ allowed set for dev_hive → STOP - │ (partition plan is never consulted) - │ - └─ 200/202 Allowlist passed → Kafka producer - │ - └─ Partition plan: router picks partition -``` - -| If you see… | Layer | Fix | -|-------------|-------|-----| -| **401** on `/access` | Authentication | Plugin / ingestor Kerberos, keytabs, SPNEGO | -| **403** on `/access` | **Service allowlist** | Add short name to `service.<repo>.allowed.users` | -| Audits accepted but wrong Kafka partition | **Partition plan** | Update partition plan ([partition guide](README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md)) | -| New repo in Admin, audits still **403** | Service allowlist not onboarded | Partition plan alone is **not** enough | -| Allowlist OK, audits in buffer partition | Partition plan not onboarded | Allowlist alone is **not** enough | - -**Takeaway:** Onboard **two things** per repo — **service allowlist** and **partition range**. - ---- - -## 5. Static → dynamic cutover — direct answers - -**Goal:** Turn on dynamic mode so every plugin keeps the **same** authorization behavior it has today — no surprise 403s on cutover day. - -### When dynamic mode is off - -| Question | Answer | -|----------|--------| -| Is `ranger_audit_service_allowlist` created? | **No** — the allowlist topic is not created or used. | -| How is authz decided? | From XML at startup (`initializeAllowedUsers()`), same as today. | -| Is there a background allowlist sync? | **No**. | - -### When dynamic mode is on - -| Question | Answer | -|----------|--------| -| Is the allowlist topic created? | **Yes** — on first startup that needs the registry. | -| Where does every ingestor get the allowlist? | From `ranger_audit_service_allowlist` (compacted topic), kept in memory on each pod. | -| Can XML edits change live authz? | **No** (once a document exists in Kafka). Runtime changes go through REST. | - -### How “same allowed users per repo” is achieved on cutover - -Preservation works because the **first published document** is built from the **same XML properties** used today: - -| Piece | Static today | Dynamic (first document from XML) | -|-------|--------------|-------------------------------------| -| Repo key | `service.<repo>.allowed.users` property name | Same repo name in JSON `services` map | -| User values | Comma-separated short names in XML | Same names in `allowedUsers` array | -| Unknown repo | No XML entry → **403** | No registry entry → **403** (fail closed) | -| Per-request check | `isAllowedServiceUser()` | Same method; reads in-memory cache | - -**Brownfield:** Export current `service.*.allowed.users` from `ranger-audit-ingestor-site.xml`, seed JSON v1 into the allowlist topic **before** or during cutover, then enable dynamic mode and verify plugin POSTs still return **200**. - -### Allowlist already in Kafka vs empty registry - -```mermaid -sequenceDiagram - participant Pod as Ingestor pod - participant Allow as ranger_audit_service_allowlist - participant XML as site.xml - - Pod->>Allow: create topic if missing (Race A) - Pod->>Allow: read allowlist document - alt document exists - Allow-->>Pod: use stored allowlist — skip XML bootstrap - else registry empty - Pod->>XML: build first document from service.*.allowed.users - Pod->>Allow: re-read (Race B — peer may have published) - Pod->>Allow: publish first document if still empty - Pod->>Allow: mandatory read-back - end - Pod->>Pod: install map in memory -``` - -| Situation | What each pod does | -|-----------|-------------------| -| Document **already** in `ranger_audit_service_allowlist` | Read and use it; **do not** publish a new document from XML | -| Topic exists but **no message** yet | One pod publishes the first document; others re-read and adopt (**Race B**) | -| Several pods create the topic at once | Idempotent create — **Race A** | - -After the first document is stored in Kafka, **Kafka is the source of truth** for the allowlist. - ---- - -## 6. How dynamic mode works (end-to-end) - -| Plane | Kafka topic | Traffic | Who reads/writes | -|-------|-------------|---------|------------------| -| **Data** | `ranger_audits` | High — every audit event | Plugins → ingestor → dispatchers | -| **Control (allowlist)** | `ranger_audit_service_allowlist` (compacted) | Low — rare allowlist changes | Admin REST + background sync on ingestor pods | -| **Control (partition plan)** | `ranger_audit_partition_plan` (compacted) | Low — rare routing changes | Separate admin REST ([partition guide](README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md)) | - -The service allowlist is **configuration**, not audit data. Ingestor keeps the current map **in memory**; only the background sync thread and REST handlers touch the allowlist topic. - -### Architecture (control plane vs data plane) - -```mermaid -flowchart TB - subgraph plugins [Ranger plugins] - HS2[HiveServer2] - OM[Ozone OM] - end - - subgraph ops [Ops or automation] - Admin[Admin REST client] - end - - subgraph ingestor [Each audit-ingestor pod] - Access["POST /api/audit/access"] - AllowREST["GET/PUT /api/audit/service-allowlist"] - Svc[Allowlist update service] - Watcher[Background allowlist sync] - Mem[(In-memory allowlist map)] - Part[Partition router] - KafkaProd[Kafka producer] - - AllowREST --> Svc - Svc --> AllowTopic - Watcher -->|poll / consume| AllowTopic - Watcher -->|atomic swap| Mem - HS2 -->|serviceName=dev_hive| Access - OM -->|serviceName=dev_ozone| Access - Access -->|isAllowedServiceUser| Mem - Access -->|on 200/202| KafkaProd - KafkaProd --> Part - Part --> AuditTopic - end - - subgraph kafka [Kafka] - AllowTopic[(ranger_audit_service_allowlist<br/>compacted, low volume)] - AuditTopic[(ranger_audits<br/>high volume)] - end - - Admin -->|GET / PUT via load balancer| AllowREST -``` - -### First startup — seeding the allowlist (once per cluster) - -When dynamic mode starts and the allowlist registry is **empty**: - -1. Ingestor enables dynamic allowlist mode. -2. Background sync finds no document in `ranger_audit_service_allowlist`. -3. Ingestor scans XML for `ranger.audit.ingestor.service.*.allowed.users` properties. -4. Ingestor builds and publishes the **first document** to the compacted topic. -5. Ingestor loads that map into memory and begins enforcing authz on `/access`. - -**After that:** additional pods and restarts **read Kafka only** — they do not re-build from XML. - -### Every audit — the hot path - -The allowlist topic is **not** read on this path. - -1. Plugin POSTs audit to ingestor (Kerberos/JWT/basic auth). -2. Ingestor maps principal → short name (`auth_to_local`). -3. `isAllowedServiceUser(serviceName, shortName)` reads the **in-memory** map. -4. If allowed → accept → Kafka produce → partition router. -5. If denied → **403**. - -### Changing the allowlist — admin or automation - -**On the pod that receives the REST call:** - -1. Read current document from Kafka. -2. Validate change (subset rule if strict mode enabled). -3. Write new document **version** to compacted topic. -4. Return **200 OK** or **409 Conflict** (stale `expectedVersion`). - -**On every ingestor pod (~30s sync interval):** - -1. Background sync picks up new document version. -2. Swaps map in memory — **no restart**. - -```mermaid -sequenceDiagram - participant Ops as Ops / automation - participant REST as Ingestor REST (one pod) - participant Allow as ranger_audit_service_allowlist - participant W as Background sync (each pod) - participant Mem as In-memory allowlist - participant Plugin as Plugin POST /access - - Ops->>REST: PUT service-allowlist (add dev_trino) - REST->>Allow: read current version - REST->>Allow: write new version - REST-->>Ops: 200 OK - - loop Every ingestor pod - W->>Allow: poll latest document - W->>Mem: atomic swap - Plugin->>Mem: isAllowedServiceUser? - end -``` - -### Rules to remember - -- **Allowlist topic = who may POST**; audit topic = audit data. -- **Memory on the hot path** — no per-audit read of the allowlist topic. -- **Kafka is the source of truth** after the first document is published. -- **Fail closed** — unknown repo or empty allowlist → **403**. -- **All pods must agree** — every ingestor syncs from the same compacted topic. -- **Partition plan is separate** — changing routing does not change who may POST. - ---- - -## 7. Admin REST API (control plane) - -When dynamic mode is on, operators change the allowlist through the **ingestor admin API** on **any** pod (usually via load balancer). Mutations are written to `ranger_audit_service_allowlist`; every pod picks up changes through background sync (~30s). - -**Auth:** Kerberos or JWT. Caller must be in **`service.allowlist.admin.users`** — **not** the same list as plugin users. Dynamic mode off → allowlist REST returns **503**. - -### Endpoints (proposed) - -| Method | Path | Use when | -|--------|------|----------| -| `GET` | `/api/audit/service-allowlist` | Read current document + version | -| `PUT` | `/api/audit/service-allowlist` | Replace or merge full document | -| `PUT` | `/api/audit/service-allowlist/services/{repo}` | Upsert one repo (convenience) | -| `DELETE` | `/api/audit/service-allowlist/services/{repo}` | Remove repo (plugins get **403** for that repo) | - -Base URL example: `https://<ingestor-host>:7081/api/audit/service-allowlist` - -### Registry document shape - -```json -{ - "version": 3, - "updatedAt": "2026-06-15T12:00:00Z", - "updatedBy": "admin", - "services": { - "dev_hive": { - "allowedUsers": ["hive"] - }, - "dev_ozone": { - "allowedUsers": ["om", "ozone"] - }, - "dev_trino": { - "allowedUsers": ["trino"] - } - } -} -``` - -### `expectedVersion` (all writes) - -Every `PUT` must include the **version** you read from the last `GET`. If another admin changed the allowlist first, the server returns **409 Conflict** with the current document — refresh and retry. - -### Common operations - -**Read allowlist** - -```http -GET /api/audit/service-allowlist -→ 200 + JSON document (note the "version" field) -``` - -**Onboard repo** — e.g. allow `trino` for `dev_trino`: - -```json -PUT /api/audit/service-allowlist -{ - "expectedVersion": 2, - "services": { - "dev_trino": { - "allowedUsers": ["trino"] - } - } -} -``` - -Or upsert one repo: - -```http -PUT /api/audit/service-allowlist/services/dev_trino -{ - "expectedVersion": 2, - "allowedUsers": ["trino"] -} -``` - -Success → **200** + updated document (version incremented). -Stale version → **409** + current document in body. -Strict subset violation → **400**. - -### What happens inside one REST call - -| Step | What the ingestor does | -|------|------------------------| -| 1 | Authenticate caller; check `service.allowlist.admin.users` | -| 2 | Read current document from `ranger_audit_service_allowlist` | -| 3 | Reject if `expectedVersion` does not match | -| 4 | Validate `allowedUsers` ⊆ `policy.download.auth.users` (strict mode, optional) | -| 5 | Publish new document version to Kafka | -| 6 | Return updated JSON | - -**GET** is cheap (memory). **PUT** always goes through Kafka so all pods converge on the same allowlist. - ---- - -## 8. Operator workflow: onboarding a new repo - -Use this **in addition to** the [partition plan onboarding workflow](README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md#7-operator-workflow-onboarding-a-plugin). - -### Stage 0 — Create service in Ranger Admin - -1. Create service `dev_trino` in Policy Manager. -2. Set `policy.download.auth.users` = `trino` (comma-separated short names). -3. Configure plugin audit destination → ingestor URL (`:7081`). - -Ranger Admin does **not** push to ingestor in Phase 1. You must onboard the allowlist separately. - -### Stage 1 — Add allowlist entry - -**Static mode (today):** - -```xml -<property> - <name>ranger.audit.ingestor.service.dev_trino.allowed.users</name> - <value>trino</value> -</property> -``` - -Restart all ingestor pods. - -**Dynamic mode (proposed):** - -```http -PUT /api/audit/service-allowlist/services/dev_trino -{ "expectedVersion": N, "allowedUsers": ["trino"] } -``` - -All ingestors apply within ~30s. **No restart** required. - -### Stage 2 — Verify plugin POST - -```bash -# From plugin host (after kinit as trino) -curl --negotiate -u : -X POST \ - "http://<ingestor>:7081/api/audit/access?serviceName=dev_trino&appId=trino" \ - -H "Content-Type: application/json" \ - -d '[...]' -``` - -Expect **200/202**, not **403**. - -### Stage 3 — Partition plan - -Promote or scale in the partition plan if dedicated partitions are needed. See [partition plan onboarding workflow](README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md#7-operator-workflow-onboarding-a-plugin) ([Confluence](https://cloudera.atlassian.net/wiki/spaces/ENG/pages/12043681813/Dynamic+Kafka+Partitioning+Guide+Ranger+Audit+Plugins)). - -### Onboard checklist - -| Check | Example | -|-------|---------| -| Admin `policy.download.auth.users` | `trino` | -| Ingestor allowlist ⊆ Admin download users | `dev_trino` → `{trino}` | -| `auth_to_local` maps plugin principal | `trino/host@REALM` → `trino` | -| Plugin `serviceName` query param | `dev_trino` (Admin repo name) | -| Partition plan entry (if needed) | `trino` promoted from buffer | - -**Do not** edit `ranger-audit-ingestor-site.xml` on one pod for runtime changes when dynamic mode is on. XML is only for **initial bootstrap** when the allowlist registry is empty. - ---- - -## 9. Configuration (dynamic mode) - -| Property | Purpose | Example | -|----------|---------|---------| -| `ranger.audit.ingestor.service.allowlist.dynamic.enabled` | Turn dynamic allowlist on/off | `false` (default) = static XML | -| `ranger.audit.ingestor.service.allowlist.topic` | Compacted allowlist topic name | `ranger_audit_service_allowlist` | -| `ranger.audit.ingestor.service.allowlist.refresh.interval.ms` | How often pods reload allowlist | `30000` | -| `ranger.audit.ingestor.service.allowlist.admin.users` | Who may call allowlist REST | `admin,ops` | -| `ranger.audit.ingestor.service.<repo>.allowed.users` | Static bootstrap per repo | `hive`, `om,ozone`, … | -| `ranger.audit.ingestor.auth.to.local` | Principal → short name rules | Same as Hadoop `hadoop.security.auth_to_local` | - -When dynamic is **off**, authz is fixed from XML at startup and the allowlist topic is not used. - -When dynamic is **on** and the registry is **empty**, the first ingestor pod seeds the document from XML. Later pods and restarts read **Kafka only**. - ---- - -## 10. FAQ - -### Basics - -**Why do we need an allowlist if Kerberos already authenticates the plugin?** -Authentication proves *who* connected. Authorization proves they may **claim audits for this repo**. Kerberos success alone would let any daemon POST as any `serviceName`. - -**What is the difference between static and dynamic allowlist?** -Static: map loaded once from XML at startup; changes need restart. Dynamic: map lives in Kafka; ops change it via REST while ingestor keeps running. - -**Does dynamic partition plan remove the allowlist check?** -**No.** Partition routing runs **after** `/access` accepts the batch. **403** on `/access` is always a **service allowlist** failure. - -**How does this relate to PR #1017 (RANGER-5645)?** -#1017 added missing **static** XML entries for Docker (service allowlist). Partition plan work is orthogonal. - -**Why not store the allowlist in Ranger Admin or Postgres?** -Phase 1 (Option C): no Admin → ingestor coupling. Kafka compacted topic matches the partition-plan pattern ingestor already uses. - -**Why not edit XML on a running pod?** -Each pod has its own copy; edits are not shared, not durable, and are lost on restart. Runtime changes belong in the allowlist topic via REST. - -### Two topics and sync - -**What are the Kafka topics involved?** -`ranger_audits` = audit data (high volume). `ranger_audit_service_allowlist` = allowlist config (low volume, compacted). Partition plan uses a third topic — see partition guide. - -**Does every audit POST read the allowlist topic?** -No. Each audit uses the map already in **memory** on that pod. Only background sync and REST mutations touch the allowlist topic. - -**How do all ingestor pods stay in sync?** -Every pod watches the same compacted allowlist topic (default every 30s) and swaps the new map into memory. - -**What happens when a pod restarts?** -It reads the latest document from Kafka (if dynamic is on). The allowlist survives in Kafka across crashes. - -### Allowlist content - -**What goes in `allowedUsers`?** -Short names after `auth_to_local` — same values as `policy.download.auth.users` on the Ranger service (or a subset). - -**What if a repo is missing from the allowlist?** -**403** for that `serviceName`. Fail closed. - -**Can I use `*` as a wildcard?** -Not recommended. Use explicit short names per repo. - -**Should tag services get an allowlist entry?** -No for repos that never POST resource audits (e.g. tag services per RANGER-2481). - -### REST and concurrency - -**Do I need to restart ingestor after PUT allowlist?** -No. Background sync applies the new map within about one refresh interval (~30s). - -**What is `expectedVersion`?** -The document `version` you believe is current when you write. Stale version → **409**. - -**What should I do on HTTP 409?** -Another writer published a newer document. Use the body from the 409 response (or `GET` again), note the new `version`, and retry. - -**Who can call service-allowlist REST?** -Ops principals in `service.allowlist.admin.users` (Kerberos/JWT). Plugin users (`hive`, `om`) send audits via `/access` — they must **not** mutate the allowlist. - -**Who can call `POST /access`?** -Only principals in `service.<repo>.allowed.users` for the given `serviceName`. - -### Cutover and bootstrap - -**When is `ranger_audit_service_allowlist` created?** -Only when dynamic allowlist mode is enabled. - -**Can I publish the allowlist to Kafka before enabling dynamic mode?** -Yes — recommended for brownfield. Ingestor will read your pre-loaded document and will not replace it with a fresh XML bootstrap. - -**How do I verify cutover succeeded?** -`GET /api/audit/service-allowlist` on every pod (same `version`); plugin POSTs still **200**; remove a user via REST → plugin gets **403** → restore → **200** without restart. - -**How do I roll back to static mode?** -Export current document via `GET`, align XML to that layout, set `dynamic.enabled=false`, rolling restart. Allowlist topic remains but is ignored. - -### Troubleshooting - -**Plugin gets 401 on `/access`** -Authentication problem — keytabs, SPNEGO, ingestor Kerberos config. Not an allowlist issue. - -**Plugin gets 403 on `/access` but Kerberos works** -Service allowlist: short name not in `service.<repo>.allowed.users`, or wrong `serviceName`, or `auth_to_local` mismatch. - -**New repo in Admin UI, still 403** -Service allowlist not onboarded. Creating the service in Admin does not auto-update ingestor (Phase 1). - -**Audits accepted but not in expected Solr partition** -Partition plan issue, not allowlist. See [partition guide](README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md). - -**Pods show different allowlist versions** -Wait one refresh interval. If still mismatched, check allowlist topic readability and watcher logs on the lagging pod. - -**Does Ranger Admin need the audit-ingestor URL?** -No. Admin reads audits from Solr (or DB/ES). Plugins point at ingestor; allowlist is ingestor-local config. - -### Relation to partition plan - -**Do I need both allowlist and partition plan for a new plugin?** -Yes for full production onboarding. Allowlist = may POST; partition plan = where in Kafka. - -**Is there a bundled API?** -Proposed Phase 4: `POST /api/audit/onboard-plugin` updates allowlist **first**, then partition plan (fail closed if allowlist missing). - -**What order on partial failure?** -Allowlist first, then plan. No audits accepted until allowlist exists for the repo. - ---- - -## 11. Phased rollout (summary) - -| Phase | Deliverable | -|-------|-------------| -| **0** | Static XML ([#1017](https://github.com/apache/ranger/pull/1017) Docker entries) — **today** | -| **1** | Kafka topic + watcher; `AuditREST` reads cache; bootstrap from XML | -| **2** | `GET/PUT /api/audit/service-allowlist` + admin authZ | -| **3** | Ops runbook + brownfield migration | -| **4** | Optional `POST /api/audit/onboard-plugin` (allowlist + partition plan) | -| **5** | Optional Ranger Admin push sync — **out of scope for Phase 1–3** | - -**Explicit non-goals for Phase 1–3:** - -- Removing authorization on `/access` -- Merging plugin allowlist with partition-plan admin allowlist -- Ranger Admin automatic sync (Option C only) - ---- - -## Docs - -| Doc | Purpose | -|-----|---------| -| [README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md](README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md) | Full design proposal (authorization layers, security, API detail) | -| [README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md](README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md) | Partition plan operator guide | -| [DESIGN-KAFKA-DYNAMIC-PARTITIONING.md](DESIGN-KAFKA-DYNAMIC-PARTITIONING.md) | Partition plan architecture | -| [DESIGN-KAFKA-AUDIT-SERVER.md](DESIGN-KAFKA-AUDIT-SERVER.md) | End-to-end audit pipeline | -| [PR #1017](https://github.com/apache/ranger/pull/1017) (RANGER-5645) | Static Docker allowlist fix | - ---- - -*Operator guide for the dynamic service allowlist proposal. Implementation tracked separately from partition plan phases.* +**Former Confluence page** (redirect stub): [Dynamic Service Allowlist Guide](https://cloudera.atlassian.net/wiki/spaces/ENG/pages/12055576591) → merged into page `12043681813`. diff --git a/audit-server/README-KAFKA-DYNAMIC-PARTITION-PLAN.md b/audit-server/README-KAFKA-DYNAMIC-PARTITION-PLAN.md index d6e2dd5f2..424838edc 100644 --- a/audit-server/README-KAFKA-DYNAMIC-PARTITION-PLAN.md +++ b/audit-server/README-KAFKA-DYNAMIC-PARTITION-PLAN.md @@ -112,6 +112,7 @@ On first startup with an empty registry, the **first ingestor pod** publishes an } ``` +- **`services`** — per-repo allowlist for `POST /api/audit/access` (`allowedUsers` short names). - **`version`** — increments on every successful REST mutation; clients send `expectedVersion` for optimistic locking. - **`topicPartitionCount`** — must match Kafka partition count for `ranger_audits`. - **`buffer`** — sticky hash target for plugins not yet promoted. @@ -127,6 +128,7 @@ Base: `https://<ingestor-host>:<port>/api/audit` | `GET` | `/partition-plan` | Read in-memory plan on this pod | | `PUT` | `/partition-plan` | Replace full plan (advanced) | | `POST` | `/partition-plan/promote` | Promote plugin from buffer to dedicated partitions | +| `POST` | `/partition-plan/onboard-repo` | Upsert `services[repo]` + promote plugin (single version) | | `POST` | `/partition-plan/scale` | Add tail partitions to an existing plugin | Example promote: diff --git a/audit-server/README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md b/audit-server/README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md index a0d26aa7f..197f7b336 100644 --- a/audit-server/README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md +++ b/audit-server/README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md @@ -17,30 +17,44 @@ specific language governing permissions and limitations under the License. --> -# Dynamic Kafka partitioning for Ranger audit plugins — guide for everyone +# Dynamic ingestor registry for Ranger audit-ingestor — guide for everyone -This guide explains **why** and **how** Ranger can move from **static** plugin-to-partition mapping (set once at startup) to **dynamic** mapping (change at runtime without restarting the audit ingestor). +This guide explains **why** and **how** Ranger moves from **static** ingestor configuration (XML at startup) to a **dynamic unified registry** at runtime — **without restarting** audit-ingestor. + +It covers both: + +1. **Kafka partition routing** — which plugin uses which `ranger_audits` partitions +2. **Service allowlist** — which principals may `POST /api/audit/access` for each Ranger repo + +Both live in one Kafka document on **`ranger_audit_partition_plan`** (`plugins`, `buffer`, and `services`). One REST API: **`/api/audit/partition-plan`**. It is written for operators, architects, and reviewers who need a shared mental model — **without reading the codebase**. ---- +**Confluence:** [Dynamic Ingestor Registry Guide (Ranger Audit Ingestor)](https://cloudera.atlassian.net/wiki/spaces/ENG/pages/12043681813) (child of [Ranger Engineering](https://cloudera.atlassian.net/wiki/spaces/ENG/pages/759726545/Ranger+Engineering)) + +**Related design docs:** [README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md](README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md) · [DESIGN-KAFKA-DYNAMIC-PARTITIONING.md](DESIGN-KAFKA-DYNAMIC-PARTITIONING.md) ## 1. What problem are we solving? -Ranger plugins (HDFS, Hive, Trino, etc.) send audit events to **audit-ingestor**, which writes them to a Kafka topic (default: `ranger_audits`). +Ranger plugins (HDFS, Hive, Trino, Ozone OM, etc.) send audit events to **audit-ingestor**, which writes them to Kafka (`ranger_audits`). + +Ingestor performs **two** configuration jobs: -To keep high-volume plugins from starving others, ingestor can assign **dedicated Kafka partitions** per plugin. Unlisted plugins share a **buffer** pool of partitions. +| Job | Question | Static today | Dynamic goal | +|-----|----------|--------------|--------------| +| **Service allowlist** | May this Kerberos principal POST audits claiming repo `R`? | XML `service.<repo>.allowed.users` at startup | `services` map in unified registry | +| **Partition routing** | After accept, which Kafka partition? | XML `configured.plugins` at startup | `plugins` / `buffer` in unified registry | **Today (static mode):** -- Which plugin uses which partitions is defined in **XML config** at startup. -- Adding a plugin or giving a hot plugin more partitions usually means: edit config → increase topic partitions → **restart ingestor**. +- Allowlist and partition layout are defined in **XML** at startup. +- Adding a repo or plugin usually means: edit XML → **restart ingestor** on every pod. **Goal (dynamic mode):** -- Change partition assignments **while ingestor is running**. -- Onboard a new plugin without reshuffling partitions already used by other plugins. -- Grow capacity by adding **new partitions at the end** of the topic (append-only), not by recomputing everyone’s ranges. +- Change allowlist **and** partition assignments **while ingestor is running**. +- Onboard a new plugin/repo without reshuffling existing partition assignments (append-only). +- Keep all ingestor replicas consistent via one shared Kafka compacted topic. --- @@ -59,12 +73,12 @@ Kafka allows **increasing** partition count; it does **not** support shrinking. The plugin id (also called agent id / app id) identifies the source of the audit — e.g. `hdfs`, `hiveServer2`, `trino`. -### Partition plan +### Partition plan (unified registry document) -A **partition plan** is a versioned JSON document that answers: +A **partition plan** is a versioned JSON document stored in `ranger_audit_partition_plan`. It answers: -- For each known plugin: **which partition numbers** may receive its audits? -- Which partition numbers are reserved for **unknown / new** plugins (the buffer)? +- For each known plugin: **which partition numbers** may receive its audits? (`plugins` + `buffer`) +- For each Ranger service repo: **which short usernames** may `POST /access`? (`services`) - What is the current **`topicPartitionCount`** (must match Kafka)? Example (simplified): @@ -78,15 +92,24 @@ Example (simplified): "hdfs": { "partitions": [0, 1, 2, 3, 4, 5] }, "hiveServer2": { "partitions": [6, 7, 8, 9, 10, 11] } }, - "buffer": { "partitions": [12, 13, 14, "... through 47 ..."] } + "buffer": { "partitions": [12, 13, 14, "... through 47 ..."] }, + "services": { + "dev_hive": { "allowedUsers": ["hive"] }, + "dev_ozone": { "allowedUsers": ["om", "ozone"] }, + "dev_trino": { "allowedUsers": ["trino"] } + } } ``` -The `version` field increments on every successful admin change (optimistic locking). +The `version` field increments on every successful admin change (optimistic locking) — **one version** for routing and allowlist mutations. + +Every ingestor pod uses the **same** document so routing and authorization stay consistent. -Every ingestor pod uses the **same** plan so routing stays consistent. +### Service allowlist (`POST /api/audit/access`) -### Partition registry (source of truth) +Plugins authenticate (Kerberos/JWT), then ingestor checks `services[serviceName].allowedUsers` (short names after `auth_to_local`). Failure → **403** before any Kafka produce. This check is **orthogonal** to partition routing but stored in the **same** registry document when dynamic mode is on. + +### Unified registry (source of truth) The live plan lives in **durable shared storage** that all ingestor replicas read — a Kafka **compacted** topic (`ranger_audit_partition_plan`). @@ -110,13 +133,13 @@ Plugins not yet in the plan (or newly appearing in the fleet) go to **buffer** p ## 3. Today vs proposed (at a glance) -| | Static (today) | Dynamic (proposed) | -|---|----------------|---------------------| -| **Where mapping lives** | XML on each pod at startup | Shared plan in Kafka compacted topic | -| **Change mapping** | Edit XML + restart | REST API; no restart | -| **Add new plugin** | Edit `configured.plugins`, often reshuffles ranges | Promote from buffer; allocate from tail only | -| **Scale hot plugin** | Edit overrides + restart | Add tail partitions + update plan | -| **Multi-replica ingestor** | Same XML if synced via ConfigMap | All pods watch same Kafka plan | +| | Static (today) | Dynamic (unified registry) | +|---|----------------|---------------------------| +| **Where config lives** | XML on each pod at startup | `ranger_audit_partition_plan` (`plugins` + `services`) | +| **Change allowlist or routing** | Edit XML + restart | `/api/audit/partition-plan` REST; no restart | +| **Add new plugin/repo** | XML + restart | `POST .../onboard-repo` or promote + `services` update | +| **Scale hot plugin** | Edit overrides + restart | `POST .../scale` (append-only tail partitions) | +| **Multi-replica ingestor** | Same XML if synced via ConfigMap | All pods watch same Kafka document | | **Feature flag** | Default behavior | `ranger.audit.ingestor.kafka.partition.plan.dynamic.enabled=true` | --- @@ -179,7 +202,7 @@ sequenceDiagram alt plan message exists Plan-->>Pod: use stored plan — skip XML bootstrap else registry empty - Pod->>XML: build first plan from XML layout + Pod->>XML: build first plan from XML (plugins + services) Pod->>Plan: re-read (Race B — peer may have published) Pod->>Plan: publish first plan if still empty Pod->>Plan: mandatory read-back @@ -214,16 +237,17 @@ After the first plan is stored in Kafka, **Kafka is the source of truth** for ro | Plane | Kafka topic | Traffic | Who reads/writes | |-------|-------------|---------|------------------| | **Data** | `ranger_audits` | High — every audit event | Plugins → ingestor → dispatchers | -| **Control** | `ranger_audit_partition_plan` (compacted) | Low — rare plan changes | Admin REST + background sync on ingestor pods | +| **Control (unified registry)** | `ranger_audit_partition_plan` (compacted) | Low — rare routing + allowlist changes | `/api/audit/partition-plan` REST + `PartitionPlanWatcher` | -The partition plan is **configuration**, not audit data. Ingestor keeps the current plan **in memory**; only the background sync thread and REST handlers touch the plan topic. +The registry is **configuration**, not audit data. Ingestor keeps the current document **in memory**; only the watcher and REST handlers touch the plan topic. ### Architecture (control plane vs data plane) ```mermaid flowchart TB subgraph plugins [Ranger plugins] - PluginIn[HDFS / Hive / new plugins] + HS2[HiveServer2] + OM[Ozone OM] end subgraph ops [Ops or automation] @@ -231,10 +255,11 @@ flowchart TB end subgraph ingestor [Each audit-ingestor pod] - REST[Partition-plan REST API] - Svc[Plan update service] - Watcher[Background plan sync] - Mem[(In-memory plan)] + Access["POST /api/audit/access"] + REST["GET/PUT/POST /partition-plan"] + Svc[PartitionPlanService] + Watcher[PartitionPlanWatcher] + Mem[(In-memory PartitionPlan<br/>plugins + services)] Part[Partition router] Queue[Audit queue] @@ -242,15 +267,18 @@ flowchart TB Svc --> PlanTopic Watcher -->|poll / consume| PlanTopic Watcher -->|atomic swap| Mem + HS2 -->|serviceName=dev_hive| Access + OM -->|serviceName=dev_ozone| Access + Access -->|isAllowedServiceUser from services| Mem + Access -->|on 200/202| Queue Mem --> Part - PluginIn -->|POST /access| Queue Queue --> Part - Part -->|Kafka produce key=plugin id| AuditTopic + Part -->|key=plugin id| AuditTopic end subgraph kafka [Kafka] - PlanTopic[(ranger_audit_partition_plan<br/>compacted, low volume)] - AuditTopic[(ranger_audits<br/>high volume)] + PlanTopic[(ranger_audit_partition_plan<br/>plugins + services)] + AuditTopic[(ranger_audits)] end subgraph consumers [Downstream — unchanged] @@ -258,7 +286,7 @@ flowchart TB HDFSdisp[HDFS dispatcher] end - Admin -->|GET / PUT via load balancer| REST + Admin -->|via load balancer| REST Svc -->|createPartitions if needed| AuditTopic AuditTopic --> Solr AuditTopic --> HDFSdisp @@ -269,10 +297,10 @@ flowchart TB When dynamic mode starts and the plan registry is **empty**: 1. Ingestor enables dynamic mode. -2. Background sync finds no plan in `ranger_audit_partition_plan`. -3. Ingestor reads XML (`configured.plugins`, overrides, buffer). -4. Ingestor builds and publishes the **first plan** to the compacted topic. -5. Ingestor loads that plan into memory and begins routing audits. +2. Watcher finds no document in `ranger_audit_partition_plan`. +3. Ingestor reads XML (`configured.plugins`, overrides, buffer, and `service.*.allowed.users`). +4. Ingestor builds and publishes the **first document** (`plugins` + `services`) to the compacted topic. +5. Ingestor loads that document into memory and begins enforcing allowlist + routing. **After that:** additional pods and restarts **read Kafka only** — they do not re-build from XML. @@ -280,10 +308,18 @@ When dynamic mode starts and the plan registry is **empty**: The plan topic is **not** read on this path. -1. Plugin POSTs audit to ingestor. -2. Partition router reads the **in-memory** plan. -3. Known plugin → round-robin within its partition list. -4. Unknown plugin → buffer partition. +```text +Plugin POST /api/audit/access?serviceName=dev_hive&appId=hiveServer2 + │ + ├─ 401 Authentication failed + ├─ 403 services[dev_hive] does not include short username → STOP + └─ 200/202 Allowlist passed → partition router → ranger_audits +``` + +1. Plugin POSTs audit to ingestor (authenticate). +2. `isAllowedServiceUser` reads **in-memory** `services` map (or static XML when dynamic off). +3. Partition router reads **in-memory** `plugins` / `buffer`. +4. Known plugin → round-robin within its partition list; unknown → buffer. 5. Record written to `ranger_audits`. ### Changing the plan — admin or automation @@ -312,7 +348,7 @@ sequenceDiagram participant Mem as In-memory plan participant Part as Partition router - Admin->>REST: promote / scale / PUT plan + Admin->>REST: promote / scale / onboard-repo / PUT plan REST->>Plan: read current plan version REST->>Audit: createPartitions (if needed) REST->>Plan: write new plan version @@ -328,28 +364,30 @@ sequenceDiagram ### Rules to remember -- **Two topics, two jobs** — plan topic = config; audit topic = data. +- **Two topics, two jobs** — plan topic = unified config (`plugins` + `services`); audit topic = data. - **Memory on the hot path** — no per-audit read of the plan topic. -- **Kafka is the source of truth** after the first plan is published. +- **Kafka is the source of truth** after the first document is published. - **Append-only growth** — new partitions only at the tail of `ranger_audits`. - **All pods must agree** — every ingestor syncs from the same compacted topic. +- **Allowlist and routing are separate checks** — 403 on `/access` is allowlist; wrong partition is routing. --- ## 6. Admin REST API (control plane) -When dynamic mode is on, operators change routing through the **ingestor admin API** on **any** pod (usually via load balancer). Mutations are written to `ranger_audit_partition_plan`; every pod picks up changes through background sync (~30s). +When dynamic mode is on, operators change routing **and** allowlists through **`/api/audit/partition-plan`** on **any** pod (usually via load balancer). Mutations are written to `ranger_audit_partition_plan`; every pod picks up changes through `PartitionPlanWatcher` (~30s). -**Auth:** Kerberos or JWT (same ingestor admin pattern as other audit APIs). Dynamic mode off → all partition-plan calls return **503**. +**Auth:** Kerberos or JWT. When `kafka.partition.plan.allowed.users` is set, only those short names may call partition-plan REST (plugin users must not). Dynamic mode off → all partition-plan calls return **503**. ### Endpoints | Method | Path | Use when | |--------|------|----------| -| `GET` | `/api/audit/partition-plan` | Read current plan (from in-memory copy on that pod) | -| `PUT` | `/api/audit/partition-plan` | Replace the full plan (advanced / corrections) | -| `POST` | `/api/audit/partition-plan/promote` | Give a **new** plugin dedicated partitions (from buffer) | -| `POST` | `/api/audit/partition-plan/scale` | Add more partitions to a plugin **already** in the plan | +| `GET` | `/api/audit/partition-plan` | Read current document (`plugins`, `buffer`, `services`, `version`) | +| `PUT` | `/api/audit/partition-plan` | Replace full document (advanced); optional `services` in body | +| `POST` | `/api/audit/partition-plan/promote` | Promote plugin; optional `repo` + `allowedUsers` | +| `POST` | `/api/audit/partition-plan/onboard-repo` | Upsert `services[repo]` + promote plugin (one version) | +| `POST` | `/api/audit/partition-plan/scale` | Add tail partitions to existing plugin | Base URL example: `https://<ingestor-host>:7081/api/audit/partition-plan` @@ -366,7 +404,20 @@ GET /api/audit/partition-plan → 200 + JSON plan (note the "version" field) ``` -**Promote** — e.g. onboard `trino` from buffer with 3 dedicated partitions: +**Onboard repo** — allowlist + promote in one call: + +```json +POST /api/audit/partition-plan/onboard-repo +{ + "repo": "dev_trino", + "pluginId": "trino", + "allowedUsers": ["trino"], + "partitionCount": 3, + "expectedVersion": 4 +} +``` + +**Promote** — e.g. onboard `trino` from buffer with 3 dedicated partitions (optional allowlist): ```json POST /api/audit/partition-plan/promote @@ -411,7 +462,7 @@ flowchart LR | Step | What the ingestor does | |------|------------------------| -| 1 | Authenticate the caller | +| 1 | Authenticate caller; check `kafka.partition.plan.allowed.users` when configured | | 2 | Read current plan from `ranger_audit_partition_plan` | | 3 | Reject if `expectedVersion` does not match | | 4 | Compute new plugin lists (append-only — no reshuffling existing slots) | @@ -423,7 +474,38 @@ flowchart LR --- -## 7. Operator workflow: onboarding a plugin +## 7. Operator workflow: onboarding a plugin or repo + +**Recommended:** one `POST /partition-plan/onboard-repo` when you need **both** allowlist and dedicated partitions. + +### Stage 0 — Create service in Ranger Admin + +1. Create service `dev_trino` in Policy Manager; set `policy.download.auth.users`. +2. Configure plugin audit destination → ingestor URL (`:7081`). + +### Stage 1 — Onboard via unified registry + +```http +POST /api/audit/partition-plan/onboard-repo +``` + +All ingestors apply within ~30s. **No restart** required. + +### Stage 2 — Verify plugin POST + +Expect **200/202** on `/access`, not **403**. + +### Stage 3 — Scale (optional) + +Call `POST /api/audit/partition-plan/scale` for hot plugins. + +**Do not** edit `ranger-audit-ingestor-site.xml` on one pod for runtime changes. XML is only for **initial bootstrap** when the registry is empty. + +--- + +## 8. Operator workflow: partition-only changes + +Use when allowlist already exists and you only need routing changes. ### Stage 0 — Plugin appears (unknown) @@ -444,21 +526,24 @@ flowchart LR --- -## 8. Configuration (dynamic mode) +## 9. Configuration (dynamic mode) | Property | Purpose | Example | |----------|---------|---------| -| `ranger.audit.ingestor.kafka.partition.plan.dynamic.enabled` | Turn dynamic mode on/off | `false` (default) = static XML | -| `ranger.audit.ingestor.kafka.partition.plan.topic` | Compacted plan topic name | `ranger_audit_partition_plan` | +| `ranger.audit.ingestor.kafka.partition.plan.dynamic.enabled` | Turn dynamic unified registry on/off | `false` (default) = static XML | +| `ranger.audit.ingestor.kafka.partition.plan.topic` | Compacted registry topic name | `ranger_audit_partition_plan` | | `ranger.audit.ingestor.kafka.partition.plan.refresh.interval.ms` | How often pods reload plan | `30000` | +| `ranger.audit.ingestor.kafka.partition.plan.allowed.users` | Who may call partition-plan REST | `admin,ops` | +| `ranger.audit.ingestor.service.<repo>.allowed.users` | Static bootstrap per-repo allowlist | `hive`, `om,ozone`, … | +| `ranger.audit.ingestor.auth.to.local` | Principal → short name rules | Same as Hadoop `hadoop.security.auth_to_local` | -When dynamic is **off**, routing is fixed from XML at startup and the plan topic is not used. +When dynamic is **off**, routing and allowlist are fixed from XML at startup; the plan topic is not used. -When dynamic is **on** and the registry is **empty**, the first ingestor pod seeds the plan from XML. Later pods and restarts read **Kafka only**. +When dynamic is **on** and the registry is **empty**, the first ingestor pod seeds the plan (`plugins` + `services`) from XML. Later pods and restarts read **Kafka only**. --- -## 9. FAQ +## 10. FAQ ### Basics @@ -578,17 +663,64 @@ No. Per-pod spool and retry when Kafka is briefly unavailable works as today. No. Batch size, linger, and compression are separate settings. **Who can call partition-plan REST?** -Authenticated admin callers on ingestor (Kerberos/JWT). Plugin audit POST users are a different path — they send audits, not plan changes. +Ops principals in `kafka.partition.plan.allowed.users` when configured (Kerberos/JWT). Plugin users (`hive`, `om`) send audits via `/access` — they must **not** mutate the registry. + +### Service allowlist (`POST /api/audit/access`) + +**Why do we need an allowlist if Kerberos already authenticates the plugin?** +Authentication proves *who* connected. Authorization proves they may **claim audits for this repo**. Kerberos success alone would let any daemon POST as any `serviceName`. + +**Does dynamic partition plan remove the allowlist check?** +**No.** Partition routing runs **after** `/access` accepts the batch. **403** on `/access` is always a **service allowlist** failure (not routing). + +**What is `serviceName`?** +The Ranger Policy Manager **repo name** (e.g. `dev_hive`), not the service type (`hive`). + +**What goes in `allowedUsers`?** +Short names after `auth_to_local` — same values as `policy.download.auth.users` on the Ranger service (or a subset). + +**Consistency rule (ops discipline):** +`allowedUsers for repo R ⊆ { short names from policy.download.auth.users for R in Ranger Admin }`. Ranger Admin does not auto-sync in Phase 1; ingestor may reject REST writes that violate subset rules (strict mode). + +**Three layers — do not merge:** + +| Layer | Who | Purpose | +|-------|-----|---------| +| **Service allowlist** (plugin POST) | Daemons (`hive`, `om`, …) | May this principal POST audits for repo `R`? | +| **Partition plan** (Kafka routing) | Ingestor internal | Which Kafka partition after accept? | +| **Admin REST APIs** | Ops (`admin`, `ops`, …) | Who may change the unified registry via `/api/audit/partition-plan`? | + +**If you see…** + +| Symptom | Layer | Fix | +|---------|-------|-----| +| **401** on `/access` | Authentication | Plugin / ingestor Kerberos, keytabs, SPNEGO | +| **403** on `/access` | **Service allowlist** | Add short name via `onboard-repo` or `PUT` `services` | +| Audits accepted but wrong Kafka partition | **Partition plan** | `promote` / `scale` | +| New repo in Admin, audits still **403** | Allowlist not onboarded | `POST .../onboard-repo` (partition plan alone is not enough) | +| Allowlist OK, audits in buffer partition | Partition plan not onboarded | `promote` plugin (allowlist alone is not enough) | + +**Plugin gets 403 but Kerberos works — what now?** +Short name not in `services[repo].allowedUsers`, wrong `serviceName`, or `auth_to_local` mismatch. + +**New repo in Admin UI, still 403?** +Creating the service in Admin does not auto-update ingestor (Phase 1). Call `POST /api/audit/partition-plan/onboard-repo`. + +**Do I need both allowlist and partition plan for a new plugin?** +Yes for full production onboarding. Use `onboard-repo` to update both in **one** plan version. + +**Is there a bundled API?** +`POST /api/audit/partition-plan/onboard-repo` upserts `services[repo]` and promotes plugin partitions atomically. --- -## Docs +## 11. Docs | Doc | Purpose | |-----|---------| +| [README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md](README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md) | Service allowlist design (unified `services` map) | | [DESIGN-KAFKA-DYNAMIC-PARTITIONING.md](DESIGN-KAFKA-DYNAMIC-PARTITIONING.md) | Partition plan architecture | | [README-KAFKA-PARTITION-PLAN-REGISTRY-REST.md](README-KAFKA-PARTITION-PLAN-REGISTRY-REST.md) | Partition-plan Kafka topic + REST | | [README-KAFKA-PARTITION-PLAN-IMPLEMENTATION.md](README-KAFKA-PARTITION-PLAN-IMPLEMENTATION.md) | Partition-plan implementation phases | -| [README-DYNAMIC-SERVICE-ALLOWLIST-GUIDE.md](README-DYNAMIC-SERVICE-ALLOWLIST-GUIDE.md) | **Service allowlist** operator guide (Surface 1 — orthogonal to partition routing) | -| [README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md](README-DYNAMIC-SERVICE-ALLOWLIST-DESIGN.md) | Service allowlist design proposal | | [DESIGN-KAFKA-AUDIT-SERVER.md](DESIGN-KAFKA-AUDIT-SERVER.md) | End-to-end audit pipeline | +| [PR #1017](https://github.com/apache/ranger/pull/1017) (RANGER-5645) | Static Docker allowlist fix | diff --git a/audit-server/README-KAFKA-PARTITION-PLAN-IMPLEMENTATION.md b/audit-server/README-KAFKA-PARTITION-PLAN-IMPLEMENTATION.md index 117e4ac66..8d102a912 100644 --- a/audit-server/README-KAFKA-PARTITION-PLAN-IMPLEMENTATION.md +++ b/audit-server/README-KAFKA-PARTITION-PLAN-IMPLEMENTATION.md @@ -301,11 +301,7 @@ When `dynamic.enabled=false`: return **404** or **503 Feature disabled**; do not ## Phase 6 — AuthZ + `/status` -**Status:** **Deferred** — implement in a follow-up PR. Phases 1–5, 7, and 8 are sufficient for initial dynamic-mode rollout. - -**Today:** `security-applicationContext.xml` requires authentication for `/api/audit/partition-plan/**`; `AuditDelegationTokenFilter` skips auth only for `/health` and `/status`. Any authenticated principal can call partition-plan (no separate admin allow-list yet). Use `GET /api/audit/partition-plan` for plan health until `/status` is extended. - -**Goal:** Restrict partition-plan API to admin users; expose plan health on status endpoint. +**Phase 6 — AuthZ:** **Done** — `kafka.partition.plan.allowed.users` enforced on partition-plan REST when configured. ### AuthZ (proposed property) diff --git a/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java b/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java index 5cdaf8960..e3dc9a2ef 100644 --- a/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java +++ b/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java @@ -67,6 +67,7 @@ private AuditServerConstants() {} public static final String PROP_PARTITION_PLAN_REFRESH_INTERVAL_MS = "kafka.partition.plan.refresh.interval.ms"; public static final String PROP_PARTITION_PLAN_CONSUMER_POLL_TIMEOUT_MS = "kafka.partition.plan.consumer.poll.timeout.ms"; public static final String PROP_PARTITION_PLAN_DYNAMIC_ENABLED = "kafka.partition.plan.dynamic.enabled"; + public static final String PROP_PARTITION_PLAN_ALLOWED_USERS = "kafka.partition.plan.allowed.users"; public static final String DEFAULT_PARTITION_PLAN_TOPIC = "ranger_audit_partition_plan"; public static final int DEFAULT_PARTITION_PLAN_REFRESH_INTERVAL_MS = 30000; public static final int DEFAULT_PARTITION_PLAN_CONSUMER_POLL_TIMEOUT_MS = 500; diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanAllocator.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanAllocator.java index 37e870fde..0d55e45b1 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanAllocator.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanAllocator.java @@ -23,6 +23,7 @@ import org.apache.ranger.audit.producer.kafka.partition.exception.PartitionPlanException; import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; import org.apache.ranger.audit.producer.kafka.partition.model.PluginPartitionAssignment; +import org.apache.ranger.audit.producer.kafka.partition.model.ServiceAllowlistEntry; import java.time.Instant; import java.util.ArrayList; @@ -35,10 +36,15 @@ public final class PartitionPlanAllocator { private PartitionPlanAllocator() { } + public static PartitionPlan promotePlugin(PartitionPlan current, String pluginId, int partitionCount, String updatedBy) { + return promotePlugin(current, pluginId, partitionCount, updatedBy, null, null); + } + /** * Give a plugin its own partitions. Uses buffer IDs first; adds new tail IDs when buffer is too small. + * Optionally upserts {@code services[repo]} in the same plan version when {@code repo} and {@code allowedUsers} are set. */ - public static PartitionPlan promotePlugin(PartitionPlan current, String pluginId, int partitionCount, String updatedBy) { + public static PartitionPlan promotePlugin(PartitionPlan current, String pluginId, int partitionCount, String updatedBy, String repo, List<String> allowedUsers) { requireMutationInputs(current, pluginId, partitionCount, updatedBy); if (current.getPlugins().containsKey(pluginId)) { throw new PartitionPlanException("Plugin '" + pluginId + "' already has dedicated partitions"); @@ -48,7 +54,22 @@ public static PartitionPlan promotePlugin(PartitionPlan current, String pluginId List<Integer> newPluginIds = takeFromBuffer(remainingBuffer, partitionCount); int topicPartitionCount = appendTailPartitions(newPluginIds, current.getTopicPartitionCount(), partitionCount - newPluginIds.size()); - return commitPlanUpdate(current, updatedBy, topicPartitionCount, addPluginAssignment(current, pluginId, newPluginIds), remainingBuffer); + Map<String, PluginPartitionAssignment> plugins = addPluginAssignment(current, pluginId, newPluginIds); + Map<String, ServiceAllowlistEntry> services = mergeServiceAllowlist(current.getServices(), repo, allowedUsers); + return commitPlanUpdate(current, updatedBy, topicPartitionCount, plugins, remainingBuffer, services); + } + + /** + * Onboard a Ranger service repo: promote plugin partitions and upsert service allowlist atomically. + */ + public static PartitionPlan onboardRepo(PartitionPlan current, String repo, String pluginId, int partitionCount, List<String> allowedUsers, String updatedBy) { + if (StringUtils.isBlank(repo)) { + throw new PartitionPlanException("repo is required"); + } + if (allowedUsers == null || allowedUsers.isEmpty()) { + throw new PartitionPlanException("allowedUsers are required"); + } + return promotePlugin(current, pluginId, partitionCount, updatedBy, repo, allowedUsers); } /** @@ -63,7 +84,7 @@ public static PartitionPlan scalePlugin(PartitionPlan current, String pluginId, List<Integer> pluginIds = new ArrayList<>(current.getPlugins().get(pluginId).getPartitions()); int topicPartitionCount = appendTailPartitions(pluginIds, current.getTopicPartitionCount(), additionalPartitions); - return commitPlanUpdate(current, updatedBy, topicPartitionCount, addPluginAssignment(current, pluginId, pluginIds), current.getBuffer().getPartitions()); + return commitPlanUpdate(current, updatedBy, topicPartitionCount, addPluginAssignment(current, pluginId, pluginIds), current.getBuffer().getPartitions(), current.getServices()); } /** Full plan replace (REST PUT) with append-only checks against the current plan. */ @@ -103,12 +124,13 @@ private static Map<String, PluginPartitionAssignment> addPluginAssignment(Partit return plugins; } - private static PartitionPlan commitPlanUpdate(PartitionPlan current, String updatedBy, int topicPartitionCount, Map<String, PluginPartitionAssignment> plugins, List<Integer> bufferIds) { + private static PartitionPlan commitPlanUpdate(PartitionPlan current, String updatedBy, int topicPartitionCount, Map<String, PluginPartitionAssignment> plugins, List<Integer> bufferIds, Map<String, ServiceAllowlistEntry> services) { PartitionPlan next = current.toBuilder() .version(current.getVersion() + 1) .topicPartitionCount(topicPartitionCount) .plugins(plugins) .buffer(new PluginPartitionAssignment(bufferIds)) + .services(services != null ? services : current.getServices()) .updatedAt(Instant.now().toString()) .updatedBy(updatedBy) .build(); @@ -117,6 +139,14 @@ private static PartitionPlan commitPlanUpdate(PartitionPlan current, String upda return next; } + private static Map<String, ServiceAllowlistEntry> mergeServiceAllowlist(Map<String, ServiceAllowlistEntry> currentServices, String repo, List<String> allowedUsers) { + Map<String, ServiceAllowlistEntry> services = new LinkedHashMap<>(currentServices); + if (StringUtils.isNotBlank(repo) && allowedUsers != null && !allowedUsers.isEmpty()) { + services.put(repo.trim(), ServiceAllowlistEntry.ofUsers(allowedUsers)); + } + return services; + } + private static void requireMutationInputs(PartitionPlan current, String pluginId, int partitionCount, String updatedBy) { if (current == null) { throw new PartitionPlanException("Current plan is required"); diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrap.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrap.java index 3850f6808..b3aee9777 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrap.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrap.java @@ -64,6 +64,7 @@ public static PartitionPlan createInitialPlan(PartitionPlanBootstrapConfig confi .updatedBy(PartitionPlanConstants.BOOTSTRAP_UPDATED_BY) .plugins(plugins) .buffer(PluginPartitionAssignment.ofRange(nextPartition, topicPartitionCount - 1)) + .services(ServiceAllowlistBootstrap.loadFromAuditServerConfig()) .build(); PartitionPlanValidator.validate(plan); diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanHolder.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanHolder.java index 8ce99689c..9af9b1d4e 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanHolder.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanHolder.java @@ -20,7 +20,11 @@ package org.apache.ranger.audit.producer.kafka.partition; import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; +import org.apache.ranger.audit.producer.kafka.partition.model.ServiceAllowlistEntry; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.Set; import java.util.concurrent.atomic.AtomicReference; /** Hot-path in-memory plan for {@code AuditPartitioner} and the background watcher. */ @@ -52,6 +56,28 @@ public void install(PartitionPlan plan, Integer kafkaPartitionCount) { lastInstalledVersion = plan.getVersion(); } + /** + * Returns allowed short usernames for a service repo from the in-memory registry document. + * {@code null} when the plan has no {@code services} block (caller should fall back to static XML). + */ + public Set<String> getAllowedUsersForService(String serviceName) { + PartitionPlan plan = planRef.get(); + if (plan == null || plan.getServices() == null || plan.getServices().isEmpty()) { + return null; + } + ServiceAllowlistEntry entry = plan.getServices().get(serviceName); + if (entry == null || entry.getAllowedUsers().isEmpty()) { + return Collections.emptySet(); + } + return Collections.unmodifiableSet(new LinkedHashSet<>(entry.getAllowedUsers())); + } + + /** True when the loaded plan carries a non-empty service allowlist map. */ + public boolean hasServiceAllowlist() { + PartitionPlan plan = planRef.get(); + return plan != null && plan.getServices() != null && !plan.getServices().isEmpty(); + } + /** Clears holder state between unit tests. */ public void resetForTests() { planRef.set(null); diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanKafkaConfig.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanKafkaConfig.java index fd1e6573c..9fb496302 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanKafkaConfig.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanKafkaConfig.java @@ -26,8 +26,11 @@ import org.apache.ranger.audit.server.AuditServerConstants; import org.apache.ranger.audit.utils.AuditMessageQueueUtils; +import java.util.Collections; +import java.util.LinkedHashSet; import java.util.Map; import java.util.Properties; +import java.util.Set; /** Kafka client settings for the partition-plan registry topic. */ public final class PartitionPlanKafkaConfig { @@ -44,6 +47,21 @@ public static boolean isDynamicPartitionPlanEnabled(Properties props, String pro return MiscUtil.getBooleanProperty(props, propPrefix + "." + AuditServerConstants.PROP_PARTITION_PLAN_DYNAMIC_ENABLED, false); } + /** Resolves short usernames allowed to call partition-plan admin REST (empty = any authenticated principal). */ + public static Set<String> resolvePartitionPlanAdminUsers(Properties props, String propPrefix) { + String configured = MiscUtil.getStringProperty(props, propPrefix + "." + AuditServerConstants.PROP_PARTITION_PLAN_ALLOWED_USERS, ""); + if (configured == null || configured.isBlank()) { + return Collections.emptySet(); + } + Set<String> users = new LinkedHashSet<>(); + for (String user : configured.split(",")) { + if (user != null && !user.isBlank()) { + users.add(user.trim()); + } + } + return users; + } + /** Returns whether the Kafka producer partitioner should use the in-memory dynamic plan. */ public static boolean isDynamicPartitionPlanEnabled(Map<String, ?> configs, String ingestorPropPrefix) { String key = ingestorPropPrefix + "." + AuditServerConstants.PROP_PARTITION_PLAN_DYNAMIC_ENABLED; diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanService.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanService.java index 855e75121..74760ee67 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanService.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanService.java @@ -21,6 +21,7 @@ import org.apache.ranger.audit.producer.kafka.partition.exception.PartitionPlanConflictException; import org.apache.ranger.audit.producer.kafka.partition.exception.PartitionPlanException; +import org.apache.ranger.audit.producer.kafka.partition.model.OnboardRepoRequest; import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlanReplaceRequest; import org.apache.ranger.audit.producer.kafka.partition.model.PromotePluginRequest; @@ -31,6 +32,7 @@ import org.springframework.stereotype.Component; import java.util.Properties; +import java.util.Set; /** REST mutations and reads for the dynamic Kafka partition plan. */ @Component @@ -88,7 +90,7 @@ public PartitionPlan promotePlugin(PromotePluginRequest request, String updatedB String auditTopic = resolveAuditTopicName(); try (PartitionPlanRegistry registry = registryFactory.open(configProps, INGESTOR_PROP_PREFIX)) { PartitionPlan current = requirePlan(registry, auditTopic); - PartitionPlan next = PartitionPlanAllocator.promotePlugin(current, request.getPluginId(), request.getPartitionCount(), updatedBy); + PartitionPlan next = PartitionPlanAllocator.promotePlugin(current, request.getPluginId(), request.getPartitionCount(), updatedBy, request.getRepo(), request.getAllowedUsers()); return publishMutation(registry, auditTopic, request.getExpectedVersion(), current, next); } catch (PartitionPlanException e) { throw e; @@ -97,6 +99,26 @@ public PartitionPlan promotePlugin(PromotePluginRequest request, String updatedB } } + /** Onboards a service repo: upsert allowlist and promote plugin partitions in one plan version. */ + public PartitionPlan onboardRepo(OnboardRepoRequest request, String updatedBy) { + requireDynamicEnabled(); + String auditTopic = resolveAuditTopicName(); + try (PartitionPlanRegistry registry = registryFactory.open(configProps, INGESTOR_PROP_PREFIX)) { + PartitionPlan current = requirePlan(registry, auditTopic); + PartitionPlan next = PartitionPlanAllocator.onboardRepo(current, request.getRepo(), request.getPluginId(), request.getPartitionCount(), request.getAllowedUsers(), updatedBy); + return publishMutation(registry, auditTopic, request.getExpectedVersion(), current, next); + } catch (PartitionPlanException e) { + throw e; + } catch (Exception e) { + throw new PartitionPlanException("Failed to onboard repo in partition plan for audit topic '" + auditTopic + "'", e); + } + } + + /** Returns configured admin short usernames for partition-plan REST (empty = not restricted beyond authentication). */ + public Set<String> getPartitionPlanAdminUsers() { + return PartitionPlanKafkaConfig.resolvePartitionPlanAdminUsers(configProps, INGESTOR_PROP_PREFIX); + } + /** Appends tail partitions to a plugin already present in the plan. */ public PartitionPlan scalePlugin(ScalePluginRequest request, String updatedBy) { requireDynamicEnabled(); diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanValidator.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanValidator.java index 2b79d100c..78fffd9ad 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanValidator.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanValidator.java @@ -24,6 +24,7 @@ import org.apache.ranger.audit.producer.kafka.partition.exception.PartitionPlanException; import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; import org.apache.ranger.audit.producer.kafka.partition.model.PluginPartitionAssignment; +import org.apache.ranger.audit.producer.kafka.partition.model.ServiceAllowlistEntry; import java.util.HashSet; import java.util.List; @@ -59,6 +60,23 @@ public static void validate(PartitionPlan plan, Integer kafkaPartitionCount) { if (assigned.size() != plan.getTopicPartitionCount()) { throw new PartitionPlanException("Partition plan must assign every topic partition exactly once"); } + validateServices(plan.getServices()); + } + + /** When present, each service entry must declare at least one allowed short username. */ + public static void validateServices(Map<String, ServiceAllowlistEntry> services) { + if (services == null || services.isEmpty()) { + return; + } + for (Map.Entry<String, ServiceAllowlistEntry> entry : services.entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + throw new PartitionPlanException("Service repo name is required"); + } + ServiceAllowlistEntry allowlistEntry = entry.getValue(); + if (allowlistEntry == null || allowlistEntry.getAllowedUsers().isEmpty()) { + throw new PartitionPlanException("allowedUsers must not be empty for service '" + entry.getKey() + "'"); + } + } } /** New plan must only add tail partitions; existing plugin lists stay unchanged in order. */ diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanWatcher.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanWatcher.java index d09101df6..667c5f6a0 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanWatcher.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanWatcher.java @@ -71,6 +71,7 @@ public void startBlocking() throws Exception { registry = new KafkaPartitionPlanRegistry(props, propPrefix); Map<String, Object> producerConfig = buildProducerConfigMap(); PartitionPlan plan = PartitionPlanBootstrap.bootstrapIfEmpty(registry, auditTopicKey, producerConfig); + plan = ServiceAllowlistBootstrap.enrichServicesFromXmlIfMissing(plan, props); int kafkaPartitionCount = resolveAuditTopicPartitionCount(); partitionPlanHolder.install(plan, kafkaPartitionCount); openConsumerAtLogEnd(); @@ -137,6 +138,7 @@ private void applyRecordIfNewer(ConsumerRecord<String, String> record) { } try { PartitionPlan plan = PartitionPlan.fromJson(record.value()); + plan = ServiceAllowlistBootstrap.enrichServicesFromXmlIfMissing(plan, props); if (plan.getVersion() <= partitionPlanHolder.getLastInstalledVersion()) { return; } diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrap.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrap.java new file mode 100644 index 000000000..b47dafedf --- /dev/null +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrap.java @@ -0,0 +1,97 @@ +/* + * 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. + */ + +package org.apache.ranger.audit.producer.kafka.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; +import org.apache.ranger.audit.producer.kafka.partition.model.ServiceAllowlistEntry; +import org.apache.ranger.audit.server.AuditServerConfig; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import static org.apache.ranger.audit.server.AuditServerConstants.PROP_PREFIX_AUDIT_SERVER_SERVICE; +import static org.apache.ranger.audit.server.AuditServerConstants.PROP_SUFFIX_ALLOWED_USERS; + +/** Loads service allowlist entries from ingestor site XML for registry bootstrap and brownfield merge. */ +public final class ServiceAllowlistBootstrap { + private static final String BOOTSTRAP_SOURCE = "xml-bootstrap"; + + private ServiceAllowlistBootstrap() { + } + + /** Scans {@code ranger.audit.ingestor.service.<repo>.allowed.users} properties. */ + public static Map<String, ServiceAllowlistEntry> loadFromProperties(Properties props) { + Map<String, ServiceAllowlistEntry> services = new LinkedHashMap<>(); + if (props == null) { + return services; + } + for (String key : props.stringPropertyNames()) { + if (!key.startsWith(PROP_PREFIX_AUDIT_SERVER_SERVICE) || !key.endsWith(PROP_SUFFIX_ALLOWED_USERS)) { + continue; + } + String repo = key.substring(PROP_PREFIX_AUDIT_SERVER_SERVICE.length(), key.length() - PROP_SUFFIX_ALLOWED_USERS.length()); + if (StringUtils.isBlank(repo)) { + continue; + } + String value = props.getProperty(key); + if (StringUtils.isBlank(value)) { + continue; + } + List<String> users = new ArrayList<>(); + for (String part : value.split(",")) { + if (part != null) { + String trimmed = part.trim(); + if (StringUtils.isNotBlank(trimmed)) { + users.add(trimmed); + } + } + } + if (users.isEmpty()) { + continue; + } + services.put(repo.trim(), new ServiceAllowlistEntry(users, BOOTSTRAP_SOURCE, null)); + } + return services; + } + + /** Loads allowlist entries from the running ingestor configuration singleton. */ + public static Map<String, ServiceAllowlistEntry> loadFromAuditServerConfig() { + return loadFromProperties(AuditServerConfig.getInstance().getProperties()); + } + + /** + * Brownfield helper: when the Kafka plan has no {@code services} block, merge XML entries in memory only + * (does not bump version or write to Kafka). + */ + public static PartitionPlan enrichServicesFromXmlIfMissing(PartitionPlan plan, Properties props) { + if (plan == null || (plan.getServices() != null && !plan.getServices().isEmpty())) { + return plan; + } + Map<String, ServiceAllowlistEntry> fromXml = loadFromProperties(props); + if (fromXml.isEmpty()) { + return plan; + } + return plan.toBuilder().services(fromXml).build(); + } +} diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PromotePluginRequest.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/OnboardRepoRequest.java similarity index 65% copy from audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PromotePluginRequest.java copy to audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/OnboardRepoRequest.java index bc706c2f8..7a63c2a7b 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PromotePluginRequest.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/OnboardRepoRequest.java @@ -22,19 +22,30 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; -/** POST /api/audit/partition-plan/promote body. */ -public final class PromotePluginRequest { +import java.util.Collections; +import java.util.List; + +/** POST /api/audit/partition-plan/onboard-repo body. */ +public final class OnboardRepoRequest { + private final String repo; private final String pluginId; private final int partitionCount; + private final List<String> allowedUsers; private final int expectedVersion; @JsonCreator - public PromotePluginRequest(@JsonProperty("pluginId") String pluginId, @JsonProperty("partitionCount") int partitionCount, @JsonProperty("expectedVersion") int expectedVersion) { + public OnboardRepoRequest(@JsonProperty("repo") String repo, @JsonProperty("pluginId") String pluginId, @JsonProperty("partitionCount") int partitionCount, @JsonProperty("allowedUsers") List<String> allowedUsers, @JsonProperty("expectedVersion") int expectedVersion) { + this.repo = repo; this.pluginId = pluginId; this.partitionCount = partitionCount; + this.allowedUsers = allowedUsers == null ? Collections.emptyList() : List.copyOf(allowedUsers); this.expectedVersion = expectedVersion; } + public String getRepo() { + return repo; + } + public String getPluginId() { return pluginId; } @@ -43,6 +54,10 @@ public int getPartitionCount() { return partitionCount; } + public List<String> getAllowedUsers() { + return allowedUsers; + } + public int getExpectedVersion() { return expectedVersion; } diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlan.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlan.java index d3dd3a37f..2aef30632 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlan.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlan.java @@ -41,9 +41,10 @@ public final class PartitionPlan { private final String updatedBy; private final Map<String, PluginPartitionAssignment> plugins; private final PluginPartitionAssignment buffer; + private final Map<String, ServiceAllowlistEntry> services; @JsonCreator - public PartitionPlan(@JsonProperty("topic") String topic, @JsonProperty("version") int version, @JsonProperty("topicPartitionCount") int topicPartitionCount, @JsonProperty("updatedAt") String updatedAt, @JsonProperty("updatedBy") String updatedBy, @JsonProperty("plugins") Map<String, PluginPartitionAssignment> plugins, @JsonProperty("buffer") PluginPartitionAssignment buffer) { + public PartitionPlan(@JsonProperty("topic") String topic, @JsonProperty("version") int version, @JsonProperty("topicPartitionCount") int topicPartitionCount, @JsonProperty("updatedAt") String updatedAt, @JsonProperty("updatedBy") String updatedBy, @JsonProperty("plugins") Map<String, PluginPartitionAssignment> plugins, @JsonProperty("buffer") PluginPartitionAssignment buffer, @JsonProperty("services") Map<String, ServiceAllowlistEntry> services) { this.topic = topic; this.version = version; this.topicPartitionCount = topicPartitionCount; @@ -51,6 +52,7 @@ public PartitionPlan(@JsonProperty("topic") String topic, @JsonProperty("version this.updatedBy = updatedBy; this.plugins = copyPlugins(plugins); this.buffer = buffer != null ? buffer : PluginPartitionAssignment.empty(); + this.services = copyServices(services); } private static Map<String, PluginPartitionAssignment> copyPlugins(Map<String, PluginPartitionAssignment> plugins) { @@ -60,6 +62,13 @@ private static Map<String, PluginPartitionAssignment> copyPlugins(Map<String, Pl return Collections.unmodifiableMap(new LinkedHashMap<>(plugins)); } + private static Map<String, ServiceAllowlistEntry> copyServices(Map<String, ServiceAllowlistEntry> services) { + if (services == null || services.isEmpty()) { + return Collections.emptyMap(); + } + return Collections.unmodifiableMap(new LinkedHashMap<>(services)); + } + public String getTopic() { return topic; } @@ -88,6 +97,10 @@ public PluginPartitionAssignment getBuffer() { return buffer; } + public Map<String, ServiceAllowlistEntry> getServices() { + return services; + } + public Builder toBuilder() { return new Builder(this); } @@ -127,17 +140,17 @@ public boolean equals(Object o) { return false; } PartitionPlan that = (PartitionPlan) o; - return version == that.version && topicPartitionCount == that.topicPartitionCount && Objects.equals(topic, that.topic) && Objects.equals(updatedAt, that.updatedAt) && Objects.equals(updatedBy, that.updatedBy) && Objects.equals(plugins, that.plugins) && Objects.equals(buffer, that.buffer); + return version == that.version && topicPartitionCount == that.topicPartitionCount && Objects.equals(topic, that.topic) && Objects.equals(updatedAt, that.updatedAt) && Objects.equals(updatedBy, that.updatedBy) && Objects.equals(plugins, that.plugins) && Objects.equals(buffer, that.buffer) && Objects.equals(services, that.services); } @Override public int hashCode() { - return Objects.hash(topic, version, topicPartitionCount, updatedAt, updatedBy, plugins, buffer); + return Objects.hash(topic, version, topicPartitionCount, updatedAt, updatedBy, plugins, buffer, services); } @Override public String toString() { - return "PartitionPlan{topic='" + topic + "', version=" + version + ", topicPartitionCount=" + topicPartitionCount + ", plugins=" + plugins.keySet() + ", bufferSize=" + buffer.size() + '}'; + return "PartitionPlan{topic='" + topic + "', version=" + version + ", topicPartitionCount=" + topicPartitionCount + ", plugins=" + plugins.keySet() + ", bufferSize=" + buffer.size() + ", services=" + services.keySet() + '}'; } public static final class Builder { @@ -148,6 +161,7 @@ public static final class Builder { private String updatedBy; private Map<String, PluginPartitionAssignment> plugins = new LinkedHashMap<>(); private PluginPartitionAssignment buffer = PluginPartitionAssignment.empty(); + private Map<String, ServiceAllowlistEntry> services = new LinkedHashMap<>(); private Builder() { } @@ -160,6 +174,7 @@ private Builder(PartitionPlan plan) { this.updatedBy = plan.updatedBy; this.plugins = new LinkedHashMap<>(plan.plugins); this.buffer = plan.buffer; + this.services = new LinkedHashMap<>(plan.services); } public Builder topic(String topic) { @@ -202,8 +217,18 @@ public Builder buffer(PluginPartitionAssignment buffer) { return this; } + public Builder services(Map<String, ServiceAllowlistEntry> services) { + this.services = services == null ? new LinkedHashMap<>() : new LinkedHashMap<>(services); + return this; + } + + public Builder putService(String repo, ServiceAllowlistEntry entry) { + this.services.put(repo, entry); + return this; + } + public PartitionPlan build() { - return new PartitionPlan(topic, version, topicPartitionCount, updatedAt, updatedBy, plugins, buffer); + return new PartitionPlan(topic, version, topicPartitionCount, updatedAt, updatedBy, plugins, buffer, services); } } } diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlanReplaceRequest.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlanReplaceRequest.java index 0ab8833dc..f3f76e28e 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlanReplaceRequest.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlanReplaceRequest.java @@ -33,13 +33,19 @@ public final class PartitionPlanReplaceRequest { private final int topicPartitionCount; private final Map<String, PluginPartitionAssignment> plugins; private final PluginPartitionAssignment buffer; + private final Map<String, ServiceAllowlistEntry> services; @JsonCreator - public PartitionPlanReplaceRequest(@JsonProperty("expectedVersion") int expectedVersion, @JsonProperty("topicPartitionCount") int topicPartitionCount, @JsonProperty("plugins") Map<String, PluginPartitionAssignment> plugins, @JsonProperty("buffer") PluginPartitionAssignment buffer) { + public PartitionPlanReplaceRequest(@JsonProperty("expectedVersion") int expectedVersion, @JsonProperty("topicPartitionCount") int topicPartitionCount, @JsonProperty("plugins") Map<String, PluginPartitionAssignment> plugins, @JsonProperty("buffer") PluginPartitionAssignment buffer, @JsonProperty("services") Map<String, ServiceAllowlistEntry> services) { this.expectedVersion = expectedVersion; this.topicPartitionCount = topicPartitionCount; this.plugins = copyPlugins(plugins); this.buffer = buffer != null ? buffer : PluginPartitionAssignment.empty(); + this.services = copyServices(services); + } + + public PartitionPlanReplaceRequest(int expectedVersion, int topicPartitionCount, Map<String, PluginPartitionAssignment> plugins, PluginPartitionAssignment buffer) { + this(expectedVersion, topicPartitionCount, plugins, buffer, null); } private static Map<String, PluginPartitionAssignment> copyPlugins(Map<String, PluginPartitionAssignment> plugins) { @@ -49,6 +55,16 @@ private static Map<String, PluginPartitionAssignment> copyPlugins(Map<String, Pl return Collections.unmodifiableMap(new LinkedHashMap<>(plugins)); } + private static Map<String, ServiceAllowlistEntry> copyServices(Map<String, ServiceAllowlistEntry> services) { + if (services == null) { + return null; + } + if (services.isEmpty()) { + return Collections.emptyMap(); + } + return Collections.unmodifiableMap(new LinkedHashMap<>(services)); + } + public int getExpectedVersion() { return expectedVersion; } @@ -65,15 +81,25 @@ public PluginPartitionAssignment getBuffer() { return buffer; } + /** When null, the current plan's services map is preserved. */ + public Map<String, ServiceAllowlistEntry> getServices() { + return services; + } + /** Converts the REST PUT body into a proposed plan for append-only validation. */ public PartitionPlan toProposedPlan(PartitionPlan current, String updatedBy) { - return PartitionPlan.builder() + PartitionPlan.Builder builder = PartitionPlan.builder() .topic(current.getTopic()) .topicPartitionCount(topicPartitionCount) .plugins(plugins) .buffer(buffer) .updatedAt(Instant.now().toString()) - .updatedBy(updatedBy) - .build(); + .updatedBy(updatedBy); + if (services != null) { + builder.services(services); + } else { + builder.services(current.getServices()); + } + return builder.build(); } } diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PromotePluginRequest.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PromotePluginRequest.java index bc706c2f8..14051b6d5 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PromotePluginRequest.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PromotePluginRequest.java @@ -22,17 +22,28 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; +import java.util.Collections; +import java.util.List; + /** POST /api/audit/partition-plan/promote body. */ public final class PromotePluginRequest { private final String pluginId; private final int partitionCount; private final int expectedVersion; + private final String repo; + private final List<String> allowedUsers; @JsonCreator - public PromotePluginRequest(@JsonProperty("pluginId") String pluginId, @JsonProperty("partitionCount") int partitionCount, @JsonProperty("expectedVersion") int expectedVersion) { + public PromotePluginRequest(@JsonProperty("pluginId") String pluginId, @JsonProperty("partitionCount") int partitionCount, @JsonProperty("expectedVersion") int expectedVersion, @JsonProperty("repo") String repo, @JsonProperty("allowedUsers") List<String> allowedUsers) { this.pluginId = pluginId; this.partitionCount = partitionCount; this.expectedVersion = expectedVersion; + this.repo = repo; + this.allowedUsers = allowedUsers == null ? Collections.emptyList() : List.copyOf(allowedUsers); + } + + public PromotePluginRequest(String pluginId, int partitionCount, int expectedVersion) { + this(pluginId, partitionCount, expectedVersion, null, null); } public String getPluginId() { @@ -46,4 +57,12 @@ public int getPartitionCount() { public int getExpectedVersion() { return expectedVersion; } + + public String getRepo() { + return repo; + } + + public List<String> getAllowedUsers() { + return allowedUsers; + } } diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/ServiceAllowlistEntry.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/ServiceAllowlistEntry.java new file mode 100644 index 000000000..c83ab56a5 --- /dev/null +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/ServiceAllowlistEntry.java @@ -0,0 +1,96 @@ +/* + * 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. + */ + +package org.apache.ranger.audit.producer.kafka.partition.model; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Objects; +import java.util.Set; + +/** Per-repo service allowlist entry stored in the unified partition-plan registry document. */ +@JsonInclude(JsonInclude.Include.NON_NULL) +public final class ServiceAllowlistEntry { + private final List<String> allowedUsers; + private final String source; + private final String notes; + + @JsonCreator + public ServiceAllowlistEntry(@JsonProperty("allowedUsers") List<String> allowedUsers, @JsonProperty("source") String source, @JsonProperty("notes") String notes) { + this.allowedUsers = copyAllowedUsers(allowedUsers); + this.source = source; + this.notes = notes; + } + + public static ServiceAllowlistEntry ofUsers(String... users) { + return new ServiceAllowlistEntry(List.of(users), null, null); + } + + public static ServiceAllowlistEntry ofUsers(List<String> users) { + return new ServiceAllowlistEntry(users, null, null); + } + + private static List<String> copyAllowedUsers(List<String> allowedUsers) { + if (allowedUsers == null || allowedUsers.isEmpty()) { + return Collections.emptyList(); + } + Set<String> unique = new LinkedHashSet<>(); + for (String user : allowedUsers) { + if (user != null && !user.isBlank()) { + unique.add(user.trim()); + } + } + return Collections.unmodifiableList(new ArrayList<>(unique)); + } + + public List<String> getAllowedUsers() { + return allowedUsers; + } + + public String getSource() { + return source; + } + + public String getNotes() { + return notes; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + ServiceAllowlistEntry that = (ServiceAllowlistEntry) o; + return Objects.equals(allowedUsers, that.allowedUsers) && Objects.equals(source, that.source) && Objects.equals(notes, that.notes); + } + + @Override + public int hashCode() { + return Objects.hash(allowedUsers, source, notes); + } +} diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/rest/AuditREST.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/rest/AuditREST.java index 1410730d5..5c06dd6b7 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/rest/AuditREST.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/rest/AuditREST.java @@ -23,9 +23,11 @@ import org.apache.hadoop.security.authentication.util.KerberosName; import org.apache.ranger.audit.model.AuthzAuditEvent; import org.apache.ranger.audit.producer.AuditDestinationMgr; +import org.apache.ranger.audit.producer.kafka.partition.PartitionPlanHolder; import org.apache.ranger.audit.producer.kafka.partition.PartitionPlanService; import org.apache.ranger.audit.producer.kafka.partition.exception.PartitionPlanConflictException; import org.apache.ranger.audit.producer.kafka.partition.exception.PartitionPlanException; +import org.apache.ranger.audit.producer.kafka.partition.model.OnboardRepoRequest; import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlanReplaceRequest; import org.apache.ranger.audit.producer.kafka.partition.model.PromotePluginRequest; @@ -255,17 +257,22 @@ public Response logAccessAudit(@QueryParam("serviceName") String serviceName, @Q @GET @Path("/partition-plan") @Produces("application/json") - public Response getPartitionPlan() { + public Response getPartitionPlan(@Context HttpServletRequest httpRequest) { LOG.debug("==> AuditREST.getPartitionPlan()"); Response ret; if (!partitionPlanService.isDynamicPartitionPlanEnabled()) { ret = partitionPlanDisabled("GET /partition-plan"); } else { - try { - ret = Response.ok(partitionPlanService.getPartitionPlan().toJson()).build(); - } catch (PartitionPlanException e) { - LOG.error("Partition plan GET failed", e); - ret = Response.status(Response.Status.SERVICE_UNAVAILABLE).entity(buildErrorResponse(e.getMessage())).build(); + Response authFailure = authorizePartitionPlanAdmin(httpRequest, "GET /partition-plan"); + if (authFailure != null) { + ret = authFailure; + } else { + try { + ret = Response.ok(partitionPlanService.getPartitionPlan().toJson()).build(); + } catch (PartitionPlanException e) { + LOG.error("Partition plan GET failed", e); + ret = Response.status(Response.Status.SERVICE_UNAVAILABLE).entity(buildErrorResponse(e.getMessage())).build(); + } } } LOG.debug("<== AuditREST.getPartitionPlan(): status={}", ret.getStatus()); @@ -283,15 +290,20 @@ public Response putPartitionPlan(PartitionPlanReplaceRequest request, @Context H if (!partitionPlanService.isDynamicPartitionPlanEnabled()) { ret = partitionPlanDisabled("PUT /partition-plan"); } else { - try { - ret = toSuccessfulPartitionPlanResponse(partitionPlanService.replacePartitionPlan(request, resolveUpdatedBy(httpRequest))); - } catch (PartitionPlanConflictException e) { - ret = toPartitionPlanConflictResponse("PUT /partition-plan", e); - } catch (PartitionPlanException e) { - ret = toPartitionPlanErrorResponse("PUT /partition-plan", e); - } catch (Exception e) { - LOG.error("Unexpected error replacing partition plan", e); - ret = Response.status(Response.Status.INTERNAL_SERVER_ERROR).entity(buildErrorResponse("Failed to replace partition plan")).build(); + Response authFailure = authorizePartitionPlanAdmin(httpRequest, "PUT /partition-plan"); + if (authFailure != null) { + ret = authFailure; + } else { + try { + ret = toSuccessfulPartitionPlanResponse(partitionPlanService.replacePartitionPlan(request, resolveUpdatedBy(httpRequest))); + } catch (PartitionPlanConflictException e) { + ret = toPartitionPlanConflictResponse("PUT /partition-plan", e); + } catch (PartitionPlanException e) { + ret = toPartitionPlanErrorResponse("PUT /partition-plan", e); + } catch (Exception e) { + LOG.error("Unexpected error replacing partition plan", e); + ret = Response.status(Response.Status.INTERNAL_SERVER_ERROR).entity(buildErrorResponse("Failed to replace partition plan")).build(); + } } } LOG.debug("<== AuditREST.putPartitionPlan(): status={}", ret.getStatus()); @@ -309,15 +321,20 @@ public Response promotePlugin(PromotePluginRequest request, @Context HttpServlet if (!partitionPlanService.isDynamicPartitionPlanEnabled()) { ret = partitionPlanDisabled("POST /partition-plan/promote"); } else { - try { - ret = toSuccessfulPartitionPlanResponse(partitionPlanService.promotePlugin(request, resolveUpdatedBy(httpRequest))); - } catch (PartitionPlanConflictException e) { - ret = toPartitionPlanConflictResponse("POST /partition-plan/promote", e); - } catch (PartitionPlanException e) { - ret = toPartitionPlanErrorResponse("POST /partition-plan/promote", e); - } catch (Exception e) { - LOG.error("Unexpected error promoting plugin in partition plan", e); - ret = Response.status(Response.Status.INTERNAL_SERVER_ERROR).entity(buildErrorResponse("Failed to promote plugin in partition plan")).build(); + Response authFailure = authorizePartitionPlanAdmin(httpRequest, "POST /partition-plan/promote"); + if (authFailure != null) { + ret = authFailure; + } else { + try { + ret = toSuccessfulPartitionPlanResponse(partitionPlanService.promotePlugin(request, resolveUpdatedBy(httpRequest))); + } catch (PartitionPlanConflictException e) { + ret = toPartitionPlanConflictResponse("POST /partition-plan/promote", e); + } catch (PartitionPlanException e) { + ret = toPartitionPlanErrorResponse("POST /partition-plan/promote", e); + } catch (Exception e) { + LOG.error("Unexpected error promoting plugin in partition plan", e); + ret = Response.status(Response.Status.INTERNAL_SERVER_ERROR).entity(buildErrorResponse("Failed to promote plugin in partition plan")).build(); + } } } LOG.debug("<== AuditREST.promotePlugin(): status={}", ret.getStatus()); @@ -335,21 +352,57 @@ public Response scalePlugin(ScalePluginRequest request, @Context HttpServletRequ if (!partitionPlanService.isDynamicPartitionPlanEnabled()) { ret = partitionPlanDisabled("POST /partition-plan/scale"); } else { - try { - ret = toSuccessfulPartitionPlanResponse(partitionPlanService.scalePlugin(request, resolveUpdatedBy(httpRequest))); - } catch (PartitionPlanConflictException e) { - ret = toPartitionPlanConflictResponse("POST /partition-plan/scale", e); - } catch (PartitionPlanException e) { - ret = toPartitionPlanErrorResponse("POST /partition-plan/scale", e); - } catch (Exception e) { - LOG.error("Unexpected error scaling plugin in partition plan", e); - ret = Response.status(Response.Status.INTERNAL_SERVER_ERROR).entity(buildErrorResponse("Failed to scale plugin in partition plan")).build(); + Response authFailure = authorizePartitionPlanAdmin(httpRequest, "POST /partition-plan/scale"); + if (authFailure != null) { + ret = authFailure; + } else { + try { + ret = toSuccessfulPartitionPlanResponse(partitionPlanService.scalePlugin(request, resolveUpdatedBy(httpRequest))); + } catch (PartitionPlanConflictException e) { + ret = toPartitionPlanConflictResponse("POST /partition-plan/scale", e); + } catch (PartitionPlanException e) { + ret = toPartitionPlanErrorResponse("POST /partition-plan/scale", e); + } catch (Exception e) { + LOG.error("Unexpected error scaling plugin in partition plan", e); + ret = Response.status(Response.Status.INTERNAL_SERVER_ERROR).entity(buildErrorResponse("Failed to scale plugin in partition plan")).build(); + } } } LOG.debug("<== AuditREST.scalePlugin(): status={}", ret.getStatus()); return ret; } + /** Onboards a service repo: upsert allowlist and promote plugin partitions in one plan version. */ + @POST + @Path("/partition-plan/onboard-repo") + @Consumes("application/json") + @Produces("application/json") + public Response onboardRepo(OnboardRepoRequest request, @Context HttpServletRequest httpRequest) { + LOG.debug("==> AuditREST.onboardRepo(repo={}, pluginId={})", request != null ? request.getRepo() : null, request != null ? request.getPluginId() : null); + Response ret; + if (!partitionPlanService.isDynamicPartitionPlanEnabled()) { + ret = partitionPlanDisabled("POST /partition-plan/onboard-repo"); + } else { + Response authFailure = authorizePartitionPlanAdmin(httpRequest, "POST /partition-plan/onboard-repo"); + if (authFailure != null) { + ret = authFailure; + } else { + try { + ret = toSuccessfulPartitionPlanResponse(partitionPlanService.onboardRepo(request, resolveUpdatedBy(httpRequest))); + } catch (PartitionPlanConflictException e) { + ret = toPartitionPlanConflictResponse("POST /partition-plan/onboard-repo", e); + } catch (PartitionPlanException e) { + ret = toPartitionPlanErrorResponse("POST /partition-plan/onboard-repo", e); + } catch (Exception e) { + LOG.error("Unexpected error onboarding repo in partition plan", e); + ret = Response.status(Response.Status.INTERNAL_SERVER_ERROR).entity(buildErrorResponse("Failed to onboard repo in partition plan")).build(); + } + } + } + LOG.debug("<== AuditREST.onboardRepo(): status={}", ret.getStatus()); + return ret; + } + /** Returns HTTP 200 with the updated plan JSON body. */ private Response toSuccessfulPartitionPlanResponse(PartitionPlan updatedPlan) { return Response.ok(updatedPlan.toJson()).build(); @@ -379,6 +432,24 @@ private Response partitionPlanDisabled(String operation) { return Response.status(Response.Status.SERVICE_UNAVAILABLE).entity(buildErrorResponse("Dynamic partition plan is not enabled")).build(); } + /** + * When {@code kafka.partition.plan.allowed.users} is configured, restrict partition-plan REST to those short names. + * When unset, any authenticated principal may call partition-plan (backward compatible). + */ + private Response authorizePartitionPlanAdmin(HttpServletRequest request, String operation) { + String user = getAuthenticatedUser(request); + if (StringUtils.isBlank(user)) { + LOG.error("{} rejected: authentication required", operation); + return Response.status(Response.Status.UNAUTHORIZED).entity(buildErrorResponse("Authentication required")).build(); + } + Set<String> adminUsers = partitionPlanService.getPartitionPlanAdminUsers(); + if (!adminUsers.isEmpty() && !adminUsers.contains(user)) { + LOG.error("{} rejected: user '{}' is not in partition plan admin allowlist", operation, user); + return Response.status(Response.Status.FORBIDDEN).entity(buildErrorResponse("User is not authorized to manage partition plan")).build(); + } + return null; + } + /** Maps service/infrastructure failures to 503; client validation mistakes to 400. */ private static Response.Status resolvePartitionPlanErrorStatus(PartitionPlanException error) { if (error.getCause() != null) { @@ -487,18 +558,22 @@ private static void initializeAuthToLocal() { * @return true if user is allowed, false otherwise */ private boolean isAllowedServiceUser(String serviceName, String userName) { - boolean ret; - - if (StringUtils.isNotBlank(serviceName) && StringUtils.isNotBlank(userName)) { - Set<String> allowedUsers = allowedServiceUsers.get(serviceName); - - ret = allowedUsers != null && allowedUsers.contains(userName); - } else { - ret = false; + if (StringUtils.isBlank(serviceName) || StringUtils.isBlank(userName)) { + return false; } - LOG.debug("isAllowedServiceUser(serviceName={}, userName={}): ret={}", serviceName, userName, ret); + if (partitionPlanService != null && partitionPlanService.isDynamicPartitionPlanEnabled()) { + Set<String> registryUsers = PartitionPlanHolder.getInstance().getAllowedUsersForService(serviceName); + if (registryUsers != null) { + boolean ret = registryUsers.contains(userName); + LOG.debug("isAllowedServiceUser(serviceName={}, userName={}) from registry: ret={}", serviceName, userName, ret); + return ret; + } + } + Set<String> allowedUsers = allowedServiceUsers.get(serviceName); + boolean ret = allowedUsers != null && allowedUsers.contains(userName); + LOG.debug("isAllowedServiceUser(serviceName={}, userName={}) from static XML: ret={}", serviceName, userName, ret); return ret; } diff --git a/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml b/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml index 943f57147..fbb58022d 100644 --- a/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml +++ b/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml @@ -460,6 +460,15 @@ <value>500</value> <description>Kafka consumer poll timeout when draining the compacted plan topic (milliseconds). Default 500.</description> </property> + <property> + <name>ranger.audit.ingestor.kafka.partition.plan.allowed.users</name> + <value>admin,ops</value> + <description> + Comma-separated short usernames allowed to call GET/PUT/POST /api/audit/partition-plan when dynamic.enabled=true. + The unified plan document includes plugin partition routing and service allowlist (services map). + Plugin daemon users must not appear here. + </description> + </property> --> <property> diff --git a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanServiceMutationTest.java b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanServiceMutationTest.java index 3873ee283..c2bf96d0a 100644 --- a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanServiceMutationTest.java +++ b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanServiceMutationTest.java @@ -19,11 +19,13 @@ import org.apache.ranger.audit.producer.kafka.partition.exception.PartitionPlanConflictException; import org.apache.ranger.audit.producer.kafka.partition.exception.PartitionPlanException; +import org.apache.ranger.audit.producer.kafka.partition.model.OnboardRepoRequest; import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlanReplaceRequest; import org.apache.ranger.audit.producer.kafka.partition.model.PluginPartitionAssignment; import org.apache.ranger.audit.producer.kafka.partition.model.PromotePluginRequest; import org.apache.ranger.audit.producer.kafka.partition.model.ScalePluginRequest; +import org.apache.ranger.audit.producer.kafka.partition.model.ServiceAllowlistEntry; import org.apache.ranger.audit.server.AuditServerConstants; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -96,6 +98,43 @@ public void testReplacePlanHonorsAppendOnly() throws Exception { assertIterableEquals(List.of(3, 4, 5, 15, 16, 17), result.getPlugins().get("hiveServer2").getPartitions()); } + @Test + public void testOnboardRepoPromotesPluginAndUpsertsAllowlist() throws Exception { + MutableRegistry registry = new MutableRegistry(initialPlan); + PartitionPlanService service = service(registry, new NoOpAuditTopicPartitionGrower()); + + PartitionPlan result = service.onboardRepo(new OnboardRepoRequest("dev_trino", "trino", 3, List.of("trino"), 1), "ops"); + + assertEquals(2, result.getVersion()); + assertIterableEquals(List.of(6, 7, 8), result.getPlugins().get("trino").getPartitions()); + assertIterableEquals(List.of("trino"), result.getServices().get("dev_trino").getAllowedUsers()); + } + + @Test + public void testPromoteWithAllowlistUpsertsServices() throws Exception { + MutableRegistry registry = new MutableRegistry(initialPlan); + PartitionPlanService service = service(registry, new NoOpAuditTopicPartitionGrower()); + + PartitionPlan result = service.promotePlugin(new PromotePluginRequest("trino", 3, 1, "dev_trino", List.of("trino")), "ops"); + + assertEquals(2, result.getVersion()); + assertIterableEquals(List.of("trino"), result.getServices().get("dev_trino").getAllowedUsers()); + } + + @Test + public void testReplacePlanCanUpdateServices() throws Exception { + MutableRegistry registry = new MutableRegistry(initialPlan); + PartitionPlanService service = service(registry, new NoOpAuditTopicPartitionGrower()); + Map<String, ServiceAllowlistEntry> services = new LinkedHashMap<>(); + services.put("dev_hive", ServiceAllowlistEntry.ofUsers("hive")); + PartitionPlanReplaceRequest request = new PartitionPlanReplaceRequest(1, initialPlan.getTopicPartitionCount(), initialPlan.getPlugins(), initialPlan.getBuffer(), services); + + PartitionPlan result = service.replacePartitionPlan(request, "ops"); + + assertEquals(2, result.getVersion()); + assertIterableEquals(List.of("hive"), result.getServices().get("dev_hive").getAllowedUsers()); + } + @Test public void testStaleExpectedVersionReturnsConflict() { MutableRegistry registry = new MutableRegistry(initialPlan); diff --git a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrapTest.java b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrapTest.java new file mode 100644 index 000000000..b05ab8ad3 --- /dev/null +++ b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrapTest.java @@ -0,0 +1,94 @@ +/* + * 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. + */ + +package org.apache.ranger.audit.producer.kafka.partition; + +import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; +import org.apache.ranger.audit.producer.kafka.partition.model.PluginPartitionAssignment; +import org.apache.ranger.audit.producer.kafka.partition.model.ServiceAllowlistEntry; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import static org.apache.ranger.audit.server.AuditServerConstants.PROP_PREFIX_AUDIT_SERVER_SERVICE; +import static org.apache.ranger.audit.server.AuditServerConstants.PROP_SUFFIX_ALLOWED_USERS; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertIterableEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class ServiceAllowlistBootstrapTest { + @AfterEach + public void tearDown() { + PartitionPlanHolder.getInstance().resetForTests(); + } + + @Test + public void testLoadFromPropertiesParsesServiceRepos() { + Properties props = new Properties(); + props.setProperty(PROP_PREFIX_AUDIT_SERVER_SERVICE + "dev_hive" + PROP_SUFFIX_ALLOWED_USERS, "hive"); + props.setProperty(PROP_PREFIX_AUDIT_SERVER_SERVICE + "dev_ozone" + PROP_SUFFIX_ALLOWED_USERS, "om, ozone"); + + Map<String, ServiceAllowlistEntry> services = ServiceAllowlistBootstrap.loadFromProperties(props); + + assertEquals(2, services.size()); + assertIterableEquals(List.of("hive"), services.get("dev_hive").getAllowedUsers()); + assertIterableEquals(List.of("om", "ozone"), services.get("dev_ozone").getAllowedUsers()); + } + + @Test + public void testEnrichServicesFromXmlIfMissingMergesInMemoryOnly() { + PartitionPlan plan = PartitionPlan.builder() + .topic("ranger_audits") + .version(3) + .topicPartitionCount(6) + .plugins(Map.of("hdfs", PluginPartitionAssignment.of(0, 1, 2))) + .buffer(PluginPartitionAssignment.of(3, 4, 5)) + .build(); + + Properties props = new Properties(); + props.setProperty(PROP_PREFIX_AUDIT_SERVER_SERVICE + "dev_hive" + PROP_SUFFIX_ALLOWED_USERS, "hive"); + + PartitionPlan enriched = ServiceAllowlistBootstrap.enrichServicesFromXmlIfMissing(plan, props); + + assertEquals(3, enriched.getVersion()); + assertNotNull(enriched.getServices().get("dev_hive")); + assertTrue(PartitionPlanHolder.getInstance().getAllowedUsersForService("dev_hive") == null); + } + + @Test + public void testHolderReturnsRegistryUsersWhenServicesPresent() { + Map<String, ServiceAllowlistEntry> services = new LinkedHashMap<>(); + services.put("dev_hive", ServiceAllowlistEntry.ofUsers("hive")); + PartitionPlan plan = PartitionPlan.builder() + .topic("ranger_audits") + .version(1) + .topicPartitionCount(6) + .plugins(Map.of("hdfs", PluginPartitionAssignment.of(0, 1, 2))) + .buffer(PluginPartitionAssignment.of(3, 4, 5)) + .services(services) + .build(); + PartitionPlanHolder.getInstance().install(plan, 6); + + assertIterableEquals(List.of("hive"), PartitionPlanHolder.getInstance().getAllowedUsersForService("dev_hive")); + assertTrue(PartitionPlanHolder.getInstance().getAllowedUsersForService("dev_trino").isEmpty()); + } +} diff --git a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlanJsonTest.java b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlanJsonTest.java index b8e0bb1ee..2778bd833 100644 --- a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlanJsonTest.java +++ b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlanJsonTest.java @@ -23,6 +23,7 @@ import org.junit.jupiter.api.Test; import java.util.List; +import java.util.Map; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertIterableEquals; @@ -41,6 +42,21 @@ public void testRoundTripPreservesPlan() { assertIterableEquals(List.of(6, 7, 8, 9, 10, 11, 12, 13, 14), parsed.getBuffer().getPartitions()); } + @Test + public void testRoundTripPreservesServices() { + PartitionPlan plan = PartitionPlan.builder() + .topic("ranger_audits") + .version(2) + .topicPartitionCount(6) + .plugins(Map.of("hdfs", PluginPartitionAssignment.of(0, 1, 2))) + .buffer(PluginPartitionAssignment.of(3, 4, 5)) + .services(Map.of("dev_hive", ServiceAllowlistEntry.ofUsers("hive"))) + .build(); + PartitionPlan parsed = PartitionPlan.fromJson(plan.toJson()); + + assertIterableEquals(List.of("hive"), parsed.getServices().get("dev_hive").getAllowedUsers()); + } + @Test public void testFromJsonRejectsInvalidPlan() { assertThrows(PartitionPlanException.class, () -> PartitionPlan.fromJson("{\"topic\":\"ranger_audits\",\"version\":1}")); diff --git a/dev-support/ranger-docker/scripts/audit/verify-partition-plan-e2e.sh b/dev-support/ranger-docker/scripts/audit/verify-partition-plan-e2e.sh index 7d40bb224..fc7c3668c 100755 --- a/dev-support/ranger-docker/scripts/audit/verify-partition-plan-e2e.sh +++ b/dev-support/ranger-docker/scripts/audit/verify-partition-plan-e2e.sh @@ -136,6 +136,14 @@ run_dynamic_tests() { pp_record_fail "topicPartitionCount=${topic_count} kafka=${kafka_parts}" fi + local services_count + services_count="$(printf '%s' "${HTTP_BODY}" | python3 -c "import sys,json; print(len(json.load(sys.stdin).get('services') or {}))" 2>/dev/null || echo "0")" + if [[ "${services_count}" -ge 1 ]]; then + pp_record_pass "plan includes services allowlist (${services_count} repos)" + else + pp_record_fail "plan missing services allowlist map" + fi + promote_plugin="${PROMOTE_PLUGIN}" if [[ "${PROMOTE_PLUGIN}" == "storm" ]]; then promote_plugin="$(pp_pick_buffer_promote_plugin "${CONTAINER}" "${PLAN_URL}")"
