Copilot commented on code in PR #1137: URL: https://github.com/apache/ranger/pull/1137#discussion_r3786999654
########## agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanValidator.java: ########## @@ -0,0 +1,144 @@ +/* + * 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.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.ranger.audit.partition.exception.PartitionPlanException; +import org.apache.ranger.audit.partition.model.PartitionPlan; +import org.apache.ranger.audit.partition.model.PluginEntry; + +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Checks partition plan shape and append-only updates. */ +public final class PartitionPlanValidator { + private PartitionPlanValidator() { + } + + public static void validate(PartitionPlan plan) { + validate(plan, null); + } + + /** + * When kafkaPartitionCount is set, it must be at least plan.topicPartitionCount. + * Extra live Kafka partitions (e.g. after static-mode migration) are allowed. + */ + public static void validate(PartitionPlan plan, Integer kafkaPartitionCount) { + if (plan == null || StringUtils.isBlank(plan.getTopic()) || plan.getVersion() < AuditPartitionPlanConstants.INITIAL_PLAN_VERSION || plan.getTopicPartitionCount() < 1) { + throw new PartitionPlanException("Invalid partition plan"); + } + if (kafkaPartitionCount != null && kafkaPartitionCount < plan.getTopicPartitionCount()) { + throw new PartitionPlanException("Kafka topic has fewer partitions than plan requires"); + } + + Set<Integer> assigned = new HashSet<>(); + registerPartitions(plan.getBuffer().getPartitions(), assigned, true); + for (Map.Entry<String, PluginEntry> entry : plan.getPlugins().entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + throw new PartitionPlanException("Plugin id is required"); + } + registerPartitions(entry.getValue().getPartitions(), assigned, false); + } Review Comment: `validate()` can throw a NullPointerException if the `plugins` map contains a non-blank pluginId mapped to a null `PluginEntry` (e.g., malformed/hostile JSON). This should fail validation with a clear `PartitionPlanException` instead of NPE. ########## agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanValidator.java: ########## @@ -0,0 +1,144 @@ +/* + * 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.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.ranger.audit.partition.exception.PartitionPlanException; +import org.apache.ranger.audit.partition.model.PartitionPlan; +import org.apache.ranger.audit.partition.model.PluginEntry; + +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Checks partition plan shape and append-only updates. */ +public final class PartitionPlanValidator { + private PartitionPlanValidator() { + } + + public static void validate(PartitionPlan plan) { + validate(plan, null); + } + + /** + * When kafkaPartitionCount is set, it must be at least plan.topicPartitionCount. + * Extra live Kafka partitions (e.g. after static-mode migration) are allowed. + */ + public static void validate(PartitionPlan plan, Integer kafkaPartitionCount) { + if (plan == null || StringUtils.isBlank(plan.getTopic()) || plan.getVersion() < AuditPartitionPlanConstants.INITIAL_PLAN_VERSION || plan.getTopicPartitionCount() < 1) { + throw new PartitionPlanException("Invalid partition plan"); + } + if (kafkaPartitionCount != null && kafkaPartitionCount < plan.getTopicPartitionCount()) { + throw new PartitionPlanException("Kafka topic has fewer partitions than plan requires"); + } + + Set<Integer> assigned = new HashSet<>(); + registerPartitions(plan.getBuffer().getPartitions(), assigned, true); + for (Map.Entry<String, PluginEntry> entry : plan.getPlugins().entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + throw new PartitionPlanException("Plugin id is required"); + } + registerPartitions(entry.getValue().getPartitions(), assigned, false); + } + if (assigned.size() != plan.getTopicPartitionCount()) { + throw new PartitionPlanException("topicPartitionCount must equal the union of all assigned partition ids"); + } + validateServiceUniqueness(plan.getPlugins()); + validateServiceAllowedUsers(plan.getServiceAllowedUsers()); + } + + /** + * When a service repo is listed in {@code serviceAllowedUsers}, it must have at least one + * allowed short username (from Admin {@code policy.download.auth.users}). + */ + public static void validateServiceAllowedUsers(Map<String, List<String>> serviceAllowedUsers) { + if (serviceAllowedUsers == null || serviceAllowedUsers.isEmpty()) { + return; + } + for (Map.Entry<String, List<String>> entry : serviceAllowedUsers.entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + throw new PartitionPlanException("Service repo name is required"); + } + List<String> users = entry.getValue(); + if (users == null || users.isEmpty()) { + throw new PartitionPlanException( + "allowedUsers must not be empty for service '" + entry.getKey().trim() + "'"); + } Review Comment: `validateServiceAllowedUsers()` only checks that the list is non-empty, but does not enforce that it contains at least one *effective* user after applying the same parsing rules used elsewhere (trim, drop blanks, ignore `*`). This allows invalid allow-lists like `["*"]` or `[" "]` to pass validation. ########## agents-common/src/main/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtil.java: ########## @@ -0,0 +1,74 @@ +/* + * 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.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.ranger.plugin.model.RangerService; + +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +/** Parses {@code policy.download.auth.users} service config for audit ingestor authorization. */ +public final class PolicyDownloadAuthUsersUtil { + public static final String CONFIG_NAME = "policy.download.auth.users"; + + private PolicyDownloadAuthUsersUtil() { + } + + public static List<String> parseUsers(RangerService service) { + if (service == null || service.getConfigs() == null) { + return Collections.emptyList(); + } + return parseUsers(service.getConfigs().get(CONFIG_NAME)); + } + + public static List<String> parseUsers(String configValue) { + if (StringUtils.isBlank(configValue)) { + return Collections.emptyList(); + } + return Arrays.stream(configValue.split(",")) + .map(String::trim) + .filter(StringUtils::isNotBlank) + .filter(user -> !"*".equals(user)) + .collect(Collectors.toList()); + } + + public static Map<String, List<String>> normalizeServiceAllowedUsers(Map<String, List<String>> serviceAllowedUsers) { + if (serviceAllowedUsers == null || serviceAllowedUsers.isEmpty()) { + return Collections.emptyMap(); + } + Map<String, List<String>> normalized = new LinkedHashMap<>(); + for (Map.Entry<String, List<String>> entry : serviceAllowedUsers.entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + continue; + } + List<String> users = entry.getValue() == null ? Collections.emptyList() : parseUsers(String.join(",", entry.getValue())); Review Comment: `normalizeServiceAllowedUsers()` builds a comma-separated string via `String.join(",", entry.getValue())`. `String.join` throws if the list contains any null element, so a malformed list like `["hive", null]` will cause an unexpected NullPointerException instead of being normalized/validated cleanly. ########## agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanValidator.java: ########## @@ -0,0 +1,144 @@ +/* + * 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.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.ranger.audit.partition.exception.PartitionPlanException; +import org.apache.ranger.audit.partition.model.PartitionPlan; +import org.apache.ranger.audit.partition.model.PluginEntry; + +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Checks partition plan shape and append-only updates. */ +public final class PartitionPlanValidator { + private PartitionPlanValidator() { + } + + public static void validate(PartitionPlan plan) { + validate(plan, null); + } + + /** + * When kafkaPartitionCount is set, it must be at least plan.topicPartitionCount. + * Extra live Kafka partitions (e.g. after static-mode migration) are allowed. + */ + public static void validate(PartitionPlan plan, Integer kafkaPartitionCount) { + if (plan == null || StringUtils.isBlank(plan.getTopic()) || plan.getVersion() < AuditPartitionPlanConstants.INITIAL_PLAN_VERSION || plan.getTopicPartitionCount() < 1) { + throw new PartitionPlanException("Invalid partition plan"); + } + if (kafkaPartitionCount != null && kafkaPartitionCount < plan.getTopicPartitionCount()) { + throw new PartitionPlanException("Kafka topic has fewer partitions than plan requires"); + } + + Set<Integer> assigned = new HashSet<>(); + registerPartitions(plan.getBuffer().getPartitions(), assigned, true); + for (Map.Entry<String, PluginEntry> entry : plan.getPlugins().entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + throw new PartitionPlanException("Plugin id is required"); + } + registerPartitions(entry.getValue().getPartitions(), assigned, false); + } + if (assigned.size() != plan.getTopicPartitionCount()) { + throw new PartitionPlanException("topicPartitionCount must equal the union of all assigned partition ids"); + } + validateServiceUniqueness(plan.getPlugins()); + validateServiceAllowedUsers(plan.getServiceAllowedUsers()); Review Comment: Validation currently only checks that the number of distinct partition IDs equals `topicPartitionCount`, but it does not enforce that IDs are exactly the contiguous 1-based range `1..topicPartitionCount`. A plan like `{topicPartitionCount: 3, buffer: [2,3,4]}` would pass size checks but break routing assumptions (logical IDs are documented as 1..N). ########## agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanValidator.java: ########## @@ -0,0 +1,144 @@ +/* + * 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.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.ranger.audit.partition.exception.PartitionPlanException; +import org.apache.ranger.audit.partition.model.PartitionPlan; +import org.apache.ranger.audit.partition.model.PluginEntry; + +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Checks partition plan shape and append-only updates. */ +public final class PartitionPlanValidator { + private PartitionPlanValidator() { + } + + public static void validate(PartitionPlan plan) { + validate(plan, null); + } + + /** + * When kafkaPartitionCount is set, it must be at least plan.topicPartitionCount. + * Extra live Kafka partitions (e.g. after static-mode migration) are allowed. + */ + public static void validate(PartitionPlan plan, Integer kafkaPartitionCount) { + if (plan == null || StringUtils.isBlank(plan.getTopic()) || plan.getVersion() < AuditPartitionPlanConstants.INITIAL_PLAN_VERSION || plan.getTopicPartitionCount() < 1) { + throw new PartitionPlanException("Invalid partition plan"); + } + if (kafkaPartitionCount != null && kafkaPartitionCount < plan.getTopicPartitionCount()) { + throw new PartitionPlanException("Kafka topic has fewer partitions than plan requires"); + } + + Set<Integer> assigned = new HashSet<>(); + registerPartitions(plan.getBuffer().getPartitions(), assigned, true); + for (Map.Entry<String, PluginEntry> entry : plan.getPlugins().entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + throw new PartitionPlanException("Plugin id is required"); + } + registerPartitions(entry.getValue().getPartitions(), assigned, false); + } + if (assigned.size() != plan.getTopicPartitionCount()) { + throw new PartitionPlanException("topicPartitionCount must equal the union of all assigned partition ids"); + } + validateServiceUniqueness(plan.getPlugins()); + validateServiceAllowedUsers(plan.getServiceAllowedUsers()); + } + + /** + * When a service repo is listed in {@code serviceAllowedUsers}, it must have at least one + * allowed short username (from Admin {@code policy.download.auth.users}). + */ + public static void validateServiceAllowedUsers(Map<String, List<String>> serviceAllowedUsers) { + if (serviceAllowedUsers == null || serviceAllowedUsers.isEmpty()) { + return; + } + for (Map.Entry<String, List<String>> entry : serviceAllowedUsers.entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + throw new PartitionPlanException("Service repo name is required"); + } + List<String> users = entry.getValue(); + if (users == null || users.isEmpty()) { + throw new PartitionPlanException( + "allowedUsers must not be empty for service '" + entry.getKey().trim() + "'"); + } + } + } + + /** Each Ranger service repo name may appear in at most one plugin entry. */ + public static void validateServiceUniqueness(Map<String, PluginEntry> plugins) { + if (plugins == null || plugins.isEmpty()) { + return; + } + Set<String> seenServices = new HashSet<>(); + for (Map.Entry<String, PluginEntry> entry : plugins.entrySet()) { + for (String serviceName : entry.getValue().getServices()) { + if (!seenServices.add(serviceName)) { + throw new PartitionPlanException("Service '" + serviceName + "' is assigned to more than one plugin"); + } + } + } + } + + /** New plan must only add tail partitions; existing plugin lists stay unchanged in order. */ + public static void validateAppendOnly(PartitionPlan current, PartitionPlan proposed) { + if (current == null || proposed == null) { + throw new PartitionPlanException("Current and proposed plans are required"); + } + if (proposed.getTopicPartitionCount() < current.getTopicPartitionCount() || proposed.getVersion() != current.getVersion() + 1) { + throw new PartitionPlanException("Plan must grow partition count and increment version by one"); + } Review Comment: The `validateAppendOnly()` exception message says the plan "must grow partition count", but the guard condition only rejects *shrinks* (`proposed.topicPartitionCount < current.topicPartitionCount`). This is confusing for callers (e.g., metadata-only updates like `serviceAllowedUsers` keep the same partition count). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
