Merge pull request #133 from mstruk/groups

Add groups extraction and expose them via OAuthKafkaPrincipal

README.md

@@ -454,6 +454,40 @@ For example:
 
 See [JsonPathFilterQuery JavaDoc](oauth-common/src/main/java/io/strimzi/kafka/oauth/jsonpath/JsonPathFilterQuery.java) for more information about the syntax.
 
+###### Group extraction
+
+When using custom authorization (by installing a custom authorizer) you may want to take user's group membership into account when making the authorization decisions.
+One way is to obtain and inspect a parsed JWT token from `io.strimzi.kafka.oauth.server.OAuthKafkaPrincipal` object available through `AuthorizableRequestContext` passed to your `authorize()` method. 
+This gives you full access to token claims.
+
+Another way is to configure group extraction at authentication time, and get groups as a set of group names from `OAuthKafkaPrincipal` object. If your custom authorizer only needs group information from the token, and that information is present in the token in the form where you can use a JSONPath query to extract it, then you may prefer to avoid extracting this information from the token yourself as part of each call to `authorize()` method, and rather use the one already extracted during authentication.
+
+There are two configuration parameters for configuring group extraction:
+
+- `oauth.groups.claim` (e.g.: `$.roles.client-roles.kafka`)
+- `oauth.groups.claim.delimiter` (a delimiter to parse the value of the groups claim when it's a single delimited string. E.g.: `,` - that's the default value)
+
+Use `oauth.groups.claim` to specify a JSONPath query pointing to the claim containing an array of strings, or a delimited single string.
+Use `oauth.groups.claim.delimiter` to specify a delimiter to use for parsing groups when they are specified as a delimited string.
+
+By default, no group extraction is performed. When you configure `oauth.groups.claim` the group extraction is enabled and occurs during authentication.
+The extracted groups are stored into `OAuthKafkaPrincipal` object. Here is an example how you can extract them in your custom authorizer:
+```
+    public List<AuthorizationResult> authorize(AuthorizableRequestContext requestContext, List<Action> actions) {
+    
+        KafkaPrincipal principal = requestContext.principal();
+        if (principal instanceof OAuthKafkaPrincipal) {
+            OAuthKafkaPrincipal p = (OAuthKafkaPrincipal) principal;
+            
+            for (String group: p.getGroups()) {
+                System.out.println("Group: " + group);
+            }
+        }
+    }
+```
+
+See [JsonPathQuery JavaDoc](oauth-common/src/main/java/io/strimzi/kafka/oauth/jsonpath/JsonPathQuery.java) for more information about the syntax.
+
 ###### Configuring the `OAuth over PLAIN`
 
 When configuring the listener for `SASL/PLAIN` using `org.apache.kafka.common.security.plain.PlainLoginModule` in its `jaas.sasl.config` (as [explained previously](#configuring-the-listeners)), the `oauth.*` options are the same as when configuring the listener for SASL/OAUTHBEARER.

oauth-common/src/main/java/io/strimzi/kafka/oauth/common/BearerTokenWithPayload.java

@@ -5,6 +5,8 @@
 package io.strimzi.kafka.oauth.common;
 
 import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.util.Set;
 
 /**
  * This extension of OAuthBearerToken provides a way to associate any additional information with the token
@@ -21,10 +23,35 @@
  */
 public interface BearerTokenWithPayload extends OAuthBearerToken {
 
+    /**
+     * Get the usage dependent object previously associated with this instance by calling {@link BearerTokenWithPayload#setPayload(Object)}
+     *
+     * @return The associated object
+     */
     Object getPayload();
 
+    /**
+     * Associate a usage dependent object with this instance
+     *
+     * @param payload The object to associate with this instance
+     */
     void setPayload(Object payload);
 
+    /**
+     * Get groups associated with this token (principal).
+     *
+     * @return The groups for the user
+     */
+    Set<String> getGroups();
+
+    /**
+     * The token claims as a JSON object. For JWT tokens it contains the content of the JWT Payload part of the token.
+     * If introspection is used, it contains the introspection endpoint response.
+     *
+     * @return Token content / details as a JSON object
+     */
+    ObjectNode getJSON();
+
     /**
      * This method returns an id of the current instance of this object.
      * It is used for debugging purposes - e.g. logging that allows tracking of an individual instance

oauth-common/src/main/java/io/strimzi/kafka/oauth/common/JSONUtil.java

@@ -16,6 +16,8 @@
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 public class JSONUtil {
 
@@ -87,18 +89,46 @@ public static String getClaimFromJWT(JsonNode node, String... path) {
         return node.asText();
     }
 
-    public static List<String> asListOfString(JsonNode arrayNode) {
+    /**
+     * This method takes a JsonNode representing an array, or a string, and converts it into a List of String items.
+     *
+     * If the passed node is a TextNode, the text is parsed into a list of items by using ' ' (space) as a delimiter.
+     * The resulting list can contain empty strings if two delimiters are present next to one another.
+     *
+     * If the JsonNode is neither an ArrayNode, nor a TextNode an IllegalArgumentException is thrown.
+     *
+     * @param arrayOrString A JsonNode to convert into a list of String
+     * @return A list of String
+     */
+    public static List<String> asListOfString(JsonNode arrayOrString) {
+        return asListOfString(arrayOrString, " ");
+    }
+
+    /**
+     * This method takes a JsonNode representing an array, or a string, and converts it into a List of String items.
+     *
+     * The <tt>delimiter</tt> parameter is only used if the passed node is a TextNode. It is used to parse the node content
+     * as a list of strings. The resulting list can contain empty strings if two delimiters are present next to one another.
+     *
+     * If the JsonNode is neither an ArrayNode, nor a TextNode an IllegalArgumentException is thrown.
+     *
+     * @param arrayOrString A JsonNode to convert into a list of String
+     * @param delimiter A delimiter to use for parsing the TextNode
+     * @return A list of String
+     */
+    public static List<String> asListOfString(JsonNode arrayOrString, String delimiter) {
 
         ArrayList<String> result = new ArrayList<>();
 
-        if (arrayNode.isTextual()) {
-            result.addAll(Arrays.asList(arrayNode.asText().split(" ")));
+        if (arrayOrString.isTextual()) {
+            result.addAll(Arrays.asList(arrayOrString.asText().split(Pattern.quote(delimiter)))
+                    .stream().map(String::trim).collect(Collectors.toList()));
         } else {
-            if (!arrayNode.isArray()) {
-                throw new IllegalArgumentException("JsonNode not a text node, nor an array node: " + arrayNode);
+            if (!arrayOrString.isArray()) {
+                throw new IllegalArgumentException("JsonNode not a text node, nor an array node: " + arrayOrString);
             }
 
-            Iterator<JsonNode> it = arrayNode.iterator();
+            Iterator<JsonNode> it = arrayOrString.iterator();
             while (it.hasNext()) {
                 JsonNode n = it.next();
                 if (n.isTextual()) {

oauth-common/src/main/java/io/strimzi/kafka/oauth/common/OAuthAuthenticator.java

@@ -38,7 +38,7 @@ public static TokenInfo loginWithAccessToken(String token, boolean isJwt, Princi
             }
         }
 
-        return new TokenInfo(token, "undefined", "undefined", System.currentTimeMillis(), System.currentTimeMillis() + 365 * 24 * 3600000);
+        return new TokenInfo(token, "undefined", "undefined", null, System.currentTimeMillis(), System.currentTimeMillis() + 365 * 24 * 3600000);
     }
 
     public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFactory socketFactory,
@@ -163,7 +163,7 @@ private static TokenInfo post(URI tokenEndpointUri, SSLSocketFactory socketFacto
             }
         }
 
-        return new TokenInfo(token.asText(), scope != null ? scope.asText() : null, "undefined", now, now + expiresIn.asLong() * 1000L);
+        return new TokenInfo(token.asText(), scope != null ? scope.asText() : null, "undefined", null, now, now + expiresIn.asLong() * 1000L);
     }
 
     public static String base64encode(String value) {

oauth-common/src/main/java/io/strimzi/kafka/oauth/common/TokenInfo.java

@@ -5,6 +5,7 @@
 package io.strimzi.kafka.oauth.common;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 
 import java.util.Collections;
 import java.util.HashSet;
@@ -24,21 +25,32 @@ public class TokenInfo {
     private Set<String> scopes = new HashSet<>();
     private long expiresAt;
     private String principal;
+    private Set<String> groups;
     private long issuedAt;
-    private JsonNode payload;
+    private ObjectNode payload;
 
     public TokenInfo(JsonNode payload, String token, String principal) {
+        this(payload, token, principal, null);
+    }
+
+    public TokenInfo(JsonNode payload, String token, String principal, Set<String> groups) {
         this(token,
                 payload.has(SCOPE) ? payload.get(SCOPE).asText() : null,
                 principal,
+                groups,
                 payload.has(IAT) ? payload.get(IAT).asInt(0) * 1000L : 0L,
                 payload.get(EXP).asInt(0) * 1000L);
-        this.payload = payload;
+
+        if (!(payload instanceof ObjectNode)) {
+            throw new IllegalArgumentException("Unexpected JSON Node type (not ObjectNode): " + payload.getClass());
+        }
+        this.payload = (ObjectNode) payload;
     }
 
-    public TokenInfo(String token, String scope, String principal, long issuedAtMs, long expiresAtMs) {
+    public TokenInfo(String token, String scope, String principal, Set<String> groups, long issuedAtMs, long expiresAtMs) {
         this.token = token;
         this.principal = principal;
+        this.groups = groups != null ? Collections.unmodifiableSet(groups) : null;
         this.issuedAt = issuedAtMs;
         this.expiresAt = expiresAtMs;
 
@@ -65,11 +77,15 @@ public String principal() {
         return principal;
     }
 
+    public Set<String> groups() {
+        return groups;
+    }
+
     public long issuedAtMs() {
         return issuedAt;
     }
 
-    public JsonNode payload() {
+    public ObjectNode payload() {
         return payload;
     }
 }

oauth-common/src/main/java/io/strimzi/kafka/oauth/jsonpath/JsonPathFilterQuery.java

@@ -86,10 +86,13 @@
  * </pre>
  *
  * Query is parsed in the first line and any error during parsing results
- * in {@link JsonPathFilterQueryException}.
+ * in {@link JsonPathQueryException}.
  *
  * Matching is thread safe. The normal usage pattern is to initialise the JsonPathFilterQuery object once,
  * and query it many times concurrently against json objects.
+ *
+ * In addition to filtering this helper can also be used to apply JsonPath query to extract a result containing the matching keys.
+ *
  */
 public class JsonPathFilterQuery {
 
@@ -106,7 +109,7 @@ private JsonPathFilterQuery(String query) {
         try {
             this.matcher = new Matcher(ctx, query);
         } catch (JsonPathException e) {
-            throw new JsonPathFilterQueryException("Failed to parse filter query: \"" + query + "\"", e);
+            throw new JsonPathQueryException("Failed to parse filter query: \"" + query + "\"", e);
         }
     }
 

oauth-common/src/main/java/io/strimzi/kafka/oauth/jsonpath/JsonPathQuery.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2017-2021, Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.kafka.oauth.jsonpath;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.jayway.jsonpath.Configuration;
+import com.jayway.jsonpath.JsonPathException;
+import com.jayway.jsonpath.Option;
+import com.jayway.jsonpath.ParseContext;
+import com.jayway.jsonpath.spi.json.JacksonJsonNodeJsonProvider;
+import com.jayway.jsonpath.spi.mapper.JacksonMappingProvider;
+
+import static com.jayway.jsonpath.JsonPath.using;
+
+/**
+ * This class implements the support for JSONPath querying as implemented by:
+ *
+ *    https://github.com/json-path/JsonPath
+ *
+ * Given the following content of the JWT token:
+ * <pre>
+ *   {
+ *     "aud": ["uma_authorization", "kafka"],
+ *     "iss": "https://auth-server/sso",
+ *     "iat": 0,
+ *     "exp": 600,
+ *     "sub": "username",
+ *     "custom": "custom-value",
+ *     "roles": {
+ *       "client-roles": {
+ *         "kafka": ["kafka-user"]
+ *       }
+ *     },
+ *     "custom-level": 9
+ *   }
+ * </pre>
+ *
+ * Some examples of valid queries are:
+ *
+ * <pre>
+ *   {@literal $}.custom
+ *   {@literal $}['aud']
+ *   {@literal $}.roles.client-roles.kafka
+ *   {@literal $}['roles']['client-roles']['kafka']
+ * </pre>
+ *
+ * See <a href="https://github.com/json-path/JsonPath">Jayway JsonPath project</a> for full syntax.
+ * <p>
+ * This class takes a JSONPath expression and applies it as-is.
+ * <p>
+ * The JWT token is used in its original payload form, for example:
+ * <pre>
+ *   {
+ *      "sub": "username",
+ *      "iss": "https://auth-server/sso",
+ *      "custom": "custom value",
+ *      ...
+ *   }
+ * </pre>
+ *
+ * Usage:
+ * <pre>
+ *   JsonPathQuery query = new JsonPathQuery("$.roles");
+ *   JsonNode result = query.apply(jsonObject);
+ * </pre>
+ *
+ * Query is parsed in the first line and any error during parsing results
+ * in {@link JsonPathQueryException}.
+ *
+ * Matching is thread safe. The normal usage pattern is to initialise the JsonPathFilterQuery object once,
+ * and query it many times concurrently against json objects.
+ *
+ * In addition to filtering this helper can also be used to apply JsonPath query to extract a result containing the matching keys.
+ *
+ */
+public class JsonPathQuery {
+
+    private final Matcher matcher;
+
+    private JsonPathQuery(String query) {
+        Configuration conf = Configuration.builder()
+                .jsonProvider(new JacksonJsonNodeJsonProvider())
+                .mappingProvider(new JacksonMappingProvider())
+                .options(Option.SUPPRESS_EXCEPTIONS)
+                .build();
+
+        ParseContext ctx = using(conf);
+        try {
+            this.matcher = new Matcher(ctx, query, false);
+        } catch (JsonPathException e) {
+            throw new JsonPathQueryException("Failed to parse filter query: \"" + query + "\"", e);
+        }
+    }
+
+    /**
+     * Construct a new JsonPathQuery
+     *
+     * @param query The query using the JSONPath syntax
+     * @return New JsonPathQuery instance
+     */
+    public static JsonPathQuery parse(String query) {
+        return new JsonPathQuery(query);
+    }
+
+    /**
+     * Apply the JsonPath query to passed object
+     *
+     * @param jsonObject Jackson DataBind object
+     * @return The Jackson JsonNode object with the result
+     */
+    public JsonNode apply(JsonNode jsonObject) {
+        return matcher.apply(jsonObject);
+    }
+
+    @Override
+    public String toString() {
+        return matcher.toString();
+    }
+}

oauth-common/src/main/java/io/strimzi/kafka/oauth/jsonpath/JsonPathFilterQueryException.java → oauth-common/src/main/java/io/strimzi/kafka/oauth/jsonpath/JsonPathQueryException.java

@@ -5,11 +5,11 @@
 package io.strimzi.kafka.oauth.jsonpath;
 
 /**
- * The exception signalling a syntactic or semantic error during JsonPathFilterQuery parsing
+ * The exception signalling a syntactic or semantic error during JsonPathQuery or JsonPathFilterQuery parsing
  */
-public class JsonPathFilterQueryException extends RuntimeException {
+public class JsonPathQueryException extends RuntimeException {
 
-    public JsonPathFilterQueryException(String message, Throwable cause) {
+    public JsonPathQueryException(String message, Throwable cause) {
         super(message, cause);
     }
 }