From a1764e0b56363911e21dcbee5f9f7cb3dc761f88 Mon Sep 17 00:00:00 2001
From: Alexis SOUQUIERE <alexis.souquiere@gmail.com>
Date: Tue, 3 Oct 2023 09:42:30 +0200
Subject: [PATCH] Add key/value suffix for schema filtering on LITERAL ACL on
 v3 claim (#328)

* Add key/value suffix for schema filtering on LITERAL ACL on v3 claim

* Updating literal/prefixed tests
---
 .../AkhqClaimProviderController.java          | 26 ++++++++++++++-----
 .../AkhqClaimProviderControllerV3Test.java    |  4 +--
 2 files changed, 21 insertions(+), 9 deletions(-)

diff --git a/src/main/java/com/michelin/ns4kafka/controllers/AkhqClaimProviderController.java b/src/main/java/com/michelin/ns4kafka/controllers/AkhqClaimProviderController.java
index 0371a864..c738ea02 100644
--- a/src/main/java/com/michelin/ns4kafka/controllers/AkhqClaimProviderController.java
+++ b/src/main/java/com/michelin/ns4kafka/controllers/AkhqClaimProviderController.java
@@ -189,13 +189,25 @@ public AkhqClaimResponseV3 generateClaimV3(@Valid @Body AkhqClaimRequest request
 
         // Add the same pattern and cluster filtering for SCHEMA as the TOPIC ones
         result.addAll(result.stream()
-            .filter(g -> g.role.equals(config.getRoles().get(AccessControlEntry.ResourceType.TOPIC)))
-            .map(g -> AkhqClaimResponseV3.Group.builder()
-                .role(config.getRoles().get(AccessControlEntry.ResourceType.SCHEMA))
-                .patterns(g.getPatterns())
-                .clusters(g.getClusters())
-                .build()
-            ).toList());
+                .filter(g -> g.role.equals(config.getRoles().get(AccessControlEntry.ResourceType.TOPIC)))
+                .map(g -> {
+                            // Takes all the PREFIXED patterns as-is
+                            List<String> patterns = new ArrayList<>(
+                                    g.getPatterns().stream().filter(p -> p.endsWith("\\E.*$")).toList());
+
+                            // Add -key or -value prefix to the schema pattern for LITERAL patterns
+                            patterns.addAll(g.getPatterns().stream()
+                                    .filter(p -> p.endsWith("\\E$"))
+                                    .map(p -> p.replace("\\E$", "-\\E(key|value)$"))
+                                    .toList());
+
+                            return AkhqClaimResponseV3.Group.builder()
+                                    .role(config.getRoles().get(AccessControlEntry.ResourceType.SCHEMA))
+                                    .patterns(patterns)
+                                    .clusters(g.getClusters())
+                                    .build();
+                        }
+                ).toList());
 
         return AkhqClaimResponseV3.builder()
             .groups(result.isEmpty() ? null : Map.of("group", result))
diff --git a/src/test/java/com/michelin/ns4kafka/controllers/AkhqClaimProviderControllerV3Test.java b/src/test/java/com/michelin/ns4kafka/controllers/AkhqClaimProviderControllerV3Test.java
index e25a516f..43db538a 100644
--- a/src/test/java/com/michelin/ns4kafka/controllers/AkhqClaimProviderControllerV3Test.java
+++ b/src/test/java/com/michelin/ns4kafka/controllers/AkhqClaimProviderControllerV3Test.java
@@ -493,8 +493,8 @@ void generateClaimAndOptimizePatterns() {
         );
         Assertions.assertEquals("registry-read", groups.get(2).getRole());
         Assertions.assertEquals(
-            List.of("^\\Qproject1.\\E.*$", "^\\Qproject2.topic2\\E$", "^\\Qproject2.topic2a\\E$",
-                "^\\Qproject2.topic3\\E$", "^\\Qproject3.\\E.*$"),
+            List.of("^\\Qproject1.\\E.*$", "^\\Qproject3.\\E.*$", "^\\Qproject2.topic2-\\E(key|value)$",
+                    "^\\Qproject2.topic2a-\\E(key|value)$", "^\\Qproject2.topic3-\\E(key|value)$"),
             groups.get(2).getPatterns()
         );
     }