update README and example and make 'env:' dereference safer

Signed-off-by: Robbert van Waveren <robbert.van.waveren@alliander.com>

README.md

@@ -149,6 +149,8 @@ The authorization configuration for KeycloakRBACAuthorizer is specified as `serv
 Both authentication and authorization configuration specific to Strimzi Kafka OAuth can also be set as ENV vars, or as Java system properties.
 The limitation here is that authentication configuration specified in this manner can not be listener-scoped. 
 
+Note that property-values starting with `env:` are interpreted as references to existing ENV vars.
+
 ### Configuring the Kafka Broker authentication
 
 Note: Strimzi Kafka OAuth can not be used for Kafka Broker to Zookeeper authentication. It only supports Kafka Client to Kafka Broker authentication (including inter-broker communication).
@@ -870,20 +872,47 @@ Strimzi Kafka OAuth supports four ways to configure authentication on the client
 
 #### Client Credentials
 
-The first is to specify the client ID and secret configured on the authorization server specifically for the individual client deployment. This is also called `client credentials grant`.
+The first is to specify the client ID configured on the authorization server specifically for the individual client deployment. This is also called `client credentials grant`.
 
 This is achieved by specifying the following:
 - `oauth.client.id` (e.g.: "my-client")
+
+together with one of authentication options below 
+
+When client starts to establish the connection with the Kafka Broker it will first obtain an access token from the configured Token Endpoint, authenticating with the configured client ID and configured authentication option using client_credentials grant type.
+
+##### Option 1: Using a Client Secret 
+
+Specify the client secret.
+
 - `oauth.client.secret` (e.g.: "my-client-secret")
 
-When client starts to establish the connection with the Kafka Broker it will first obtain an access token from the configured Token Endpoint, authenticating with the configured client ID and secret using client_credentials grant type.
+##### Option 2: Using a Client Assertion (a.k.a. private_key_jwt)
+
+Specify the client assertion (JWT token) either directly through
+
+- `oauth.client.assertion`
+
+or pointing to a file on the filesystem
+
+- `oauth.client.assertion.location`
+
+the exact type of the token must also be communicated to the token endpoint and defaults to `urn:ietf:params:oauth:client-assertion-type:jwt-bearer`. 
+
+This can be overridden using property
+
+-  `oauth.client.assertion.type` (i.e. use `urn:ietf:params:oauth:client-assertion-type:saml2-bearer` for SAML 2 tokens)
 
 #### Refresh Token
 
-The second way is to manually obtain and set a refresh token:
+The second way is to manually obtain and set a refresh token either directly through
 
 - `oauth.refresh.token`
 
+or pointing to a file on the filesystem
+
+- `oauth.refresh.token.location`
+
 When using this approach you are not limited to OAuth2 client_credentials grant type for obtaining a token.
 You can use a password grant type and authenticate as an individual user, rather than a client application.
 There is a [simple CLI tool](examples/docker/kafka-oauth-strimzi/kafka/oauth.sh) you can use to obtain the refresh token or an access token. 
@@ -892,10 +921,14 @@ When client starts to establish the connection with the Kafka Broker it will fir
 
 #### Access Token
 
-The third way is to manually obtain and set an access token:
+The third way is to manually obtain and set an access token either directly through:
 
 - `oauth.access.token`
 
+or pointing to a file on the filesystem
+
+- `oauth.access.token.location`
+
 Access tokens are supposed to be short-lived in order to prevent unauthorized access if the token leaks.
 It is up to you, your environment, and how you plan to run your Kafka client application to consider if using long-lived access tokens is appropriate.
 

examples/consumer/src/main/java/io/strimzi/examples/consumer/ExampleConsumer.java

@@ -44,8 +44,9 @@ public static void main(String[] args) {
 
         defaults.setProperty(ClientConfig.OAUTH_TOKEN_ENDPOINT_URI, tokenEndpointUri);
 
-        //  By defaut this client uses preconfigured clientId and secret to authenticate.
-        //  You can set OAUTH_ACCESS_TOKEN or OAUTH_REFRESH_TOKEN to override default authentication.
+        //  By default, this client uses preconfigured clientId and secret to authenticate.
+        //  You can set OAUTH_ACCESS_TOKEN(_LOCATION) or OAUTH_REFRESH_TOKEN(_LOCATION)
+        //  or OAUTH_CLIENT_ASSERTION(_LOCATION) to override default authentication behavior.
         //
         //  If access token is configured, it is passed directly to Kafka broker
         //  If refresh token is configured, it is used in conjunction with clientId and secret
@@ -56,7 +57,12 @@ public static void main(String[] args) {
 
         if (accessToken == null) {
             defaults.setProperty(Config.OAUTH_CLIENT_ID, "kafka-consumer-client");
+
+            // use a secret for client_credentials authentication
             defaults.setProperty(Config.OAUTH_CLIENT_SECRET, "kafka-consumer-client-secret");
+
+            // use private_key_jwt for client_credentials authentication
+            //defaults.setProperty(ClientConfig.OAUTH_CLIENT_ASSERTION, "jwt-signed-by-trusted-key");
         }
 
         // Use 'preferred_username' rather than 'sub' for principal name

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

@@ -120,7 +120,10 @@ public String getValue(String key, String fallback) {
 
         if (result != null && result.startsWith("env:")) {
             // try reference to environment variable
-            result = System.getenv(result.substring(4));
+            final String envResult = System.getenv(result.substring(4));
+            if (envResult != null) {
+                result = envResult;
+            }
         }
 
         return result != null ? result : fallback;