Add secured namespaces feature

michelin · Dec 17, 2024 · e1f7bb8 · e1f7bb8
1 parent 0d0c2b9
commit e1f7bb8
Show file tree

Hide file tree

Showing 6 changed files with 277 additions and 3 deletions.
diff --git a/src/main/java/com/michelin/ns4kafka/controller/acl/AclController.java b/src/main/java/com/michelin/ns4kafka/controller/acl/AclController.java
@@ -1,8 +1,10 @@
 package com.michelin.ns4kafka.controller.acl;
 
+import static com.michelin.ns4kafka.service.AclService.PUBLIC_GRANTED_TO;
 import static com.michelin.ns4kafka.util.FormatErrorUtils.invalidAclDeleteOnlyAdmin;
 import static com.michelin.ns4kafka.util.FormatErrorUtils.invalidImmutableField;
 import static com.michelin.ns4kafka.util.FormatErrorUtils.invalidNotFound;
+import static com.michelin.ns4kafka.util.FormatErrorUtils.invalidSecuredNamespaceGrantAcl;
 import static com.michelin.ns4kafka.util.FormatErrorUtils.invalidSelfAssignedAclDelete;
 import static com.michelin.ns4kafka.util.enumation.Kind.ACCESS_CONTROL_ENTRY;
 import static io.micronaut.core.util.StringUtils.EMPTY_STRING;
@@ -105,10 +107,19 @@ public HttpResponse<AccessControlEntry> apply(Authentication authentication,
                                                   @QueryValue(defaultValue = "false") boolean dryrun) {
         Namespace ns = getNamespace(namespace);
 
+        boolean grantorIsSecured = getNamespace(accessControlEntry.getMetadata().getNamespace()).getSpec().isSecured();
+
+        boolean granteeIsSecured = !PUBLIC_GRANTED_TO.equals(accessControlEntry.getSpec().getGrantedTo())
+            && getNamespace(accessControlEntry.getSpec().getGrantedTo()).getSpec().isSecured();
         boolean isAdmin = authentication.getRoles().contains(ResourceBasedSecurityRule.IS_ADMIN);
         boolean isSelfAssigned = namespace.equals(accessControlEntry.getSpec().getGrantedTo());
 
         List<String> validationErrors;
+
+        if (grantorIsSecured && !granteeIsSecured) {
+            throw new ResourceValidationException(accessControlEntry, invalidSecuredNamespaceGrantAcl());
+        }
+
         if (isAdmin && isSelfAssigned) {
             // Validate overlapping OWNER
             validationErrors = aclService.validateAsAdmin(accessControlEntry, ns);

diff --git a/src/main/java/com/michelin/ns4kafka/model/Namespace.java b/src/main/java/com/michelin/ns4kafka/model/Namespace.java
@@ -50,6 +50,8 @@ public static class NamespaceSpec {
         @NotBlank
         private String kafkaUser;
         @Builder.Default
+        private boolean isSecured = Boolean.FALSE;
+        @Builder.Default
         private List<String> connectClusters = List.of();
         private TopicValidator topicValidator;
         private ConnectValidator connectValidator;

diff --git a/src/main/java/com/michelin/ns4kafka/util/FormatErrorUtils.java b/src/main/java/com/michelin/ns4kafka/util/FormatErrorUtils.java
@@ -638,6 +638,16 @@ public static String invalidQuotaOperationCannotAdd(ResourceQuota.ResourceQuotaS
                 toAdd));
     }
 
+    /**
+     * Invalid secured namespace grant ACL to basic namespaces.
+     *
+     * @return the error message
+     */
+    public static String invalidSecuredNamespaceGrantAcl() {
+        return String.format(INVALID_OPERATION, OPERATION_APPLY,
+            "secured namespace cannot grant ACL to basic namespaces");
+    }
+
     /**
      * Invalid schema suffix.
      *