Added configs

dlg99 · tiagomlalves · commit a71136009936 · 2025-07-10T11:00:36.000+01:00
diff --git a/common/src/main/java/com/datastax/oss/common/sink/config/AuthenticatorConfig.java b/common/src/main/java/com/datastax/oss/common/sink/config/AuthenticatorConfig.java
@@ -41,14 +41,69 @@ public class AuthenticatorConfig extends AbstractConfig {
   public static final String PRINCIPAL_OPT = "auth.gssapi.principal";
   public static final String SERVICE_OPT = "auth.gssapi.service";
 
+  /*
+    # oidc_options:
+  #     # OIDC Authentication server url (required when "oidc" authentication scheme is used).
+  #     issuer: https://auth.example.com/realms/my-realm
+  #
+  #     # Array of accepted OIDC clients (by default all clients are accepted).
+  #     accepted_audience: []
+  #
+  #     # Access token claim used to identify the user in the database (by default the "preferred_username"
+  #     # claim is used). When defined, the claim must exist at the top-level of the access token's JWT.
+  #     user_name_claim:
+  #
+  #     # Array of claims used to identify the user roles in the database. For example, setting this
+  #     # configuration to "[ realm_roles.roles ]" will assume all OIDC realm roles. If the claim doesn't
+  #     # exist or does not identify an array of strings in the JWT, the setting will be ignored. Roles
+  #     # defined in these claims must be previously defined in the database such that any privilege grant
+  #     # is recognized and validated.
+  #     user_roles_claims: []
+  #
+  #     # Set to true to initiate a TLS encrypted connections. Defaults to false. If case of
+  #     # truststore_path being set, it will default to true.
+  #     use_tls: false
+  #
+  #     # Sets the keystore path with valid certificates to establish encrypted connections with the OIDC Server.
+  #     # When TLS is enabled, failure to set this option or an invalid path will result in an error.
+  #     truststore_path:
+  #     # Keystore password
+  #     truststore_password:
+  #     # Keystore type (JKS or PKCS12). By default, the trustore type is inferred from the keystore
+  #     # extension. For `.p12`, `.pkcs12`, or `.pfx` files, PKCS12 is used. For `.keystore` or
+  #     # `.jks` files, JKS is used. In case the keystore type cannot be inferred from the extension,
+  #     # this option must be set.
+  #     truststore_type: jks
+  #
+  #     # Compatibility option, allowing older drivers to use OIDC authentication via password
+  #     # authentication, by using a pre-configurable username and passing an OIDC JWT access token as
+  #     # password. Defaults to false.
+  #     # allow_token_via_password_authentication: false
+  #     # When the above option is true, it allows configuring the username which will be used
+  #     # to detect that an OIDC JWT access token is being sent as password.
+  #     # token_username: oauth-bearer
+     */
+
+  public static final String OIDC_ISSUER = "auth.oidc.issuer";
+  public static final String OIDC_ACCEPTED_AUDIENCE = "auth.oidc.accepted_audience";
+  public static final String OIDC_USER_NAME_CLAIM = "auth.oidc.user_name_claim";
+  public static final String OIDC_USER_ROLES_CLAIMS = "auth.oidc.user_roles_claims";
+  public static final String OIDC_USE_TLS = "auth.oidc.use_tls";
+  public static final String OIDC_TRUSTSTORE_PATH = "auth.oidc.truststore_path";
+  public static final String OIDC_TRUSTSTORE_PASSWORD = "auth.oidc.truststore_password";
+  public static final String OIDC_TRUSTSTORE_TYPE = "auth.oidc.truststore_type";
+  public static final String OIDC_ALLOW_TOKEN_VIA_PASSWORD_AUTHENTICATION =
+      "auth.oidc.allow_token_via_password_authentication";
+  public static final String OIDC_TOKEN_USERNAME = "auth.oidc.token_username";
+
   private static final Logger log = LoggerFactory.getLogger(AuthenticatorConfig.class);
   public static final ConfigDef CONFIG_DEF =
       new ConfigDef()
           .define(
               PROVIDER_OPT,
               ConfigDef.Type.STRING,
               "None",
-              ConfigDef.ValidString.in("None", "PLAIN", "GSSAPI"),
+              ConfigDef.ValidString.in("None", "PLAIN", "GSSAPI", "OIDC"),
               ConfigDef.Importance.HIGH,
               "Authentication provider")
           .define(
@@ -80,7 +135,74 @@ public class AuthenticatorConfig extends AbstractConfig {
               ConfigDef.Type.STRING,
               "dse",
               ConfigDef.Importance.HIGH,
-              "SASL service name to use for GSSAPI provider authentication");
+              "SASL service name to use for GSSAPI provider authentication")
+          .define(
+              OIDC_ISSUER,
+              ConfigDef.Type.STRING,
+              "",
+              ConfigDef.Importance.HIGH,
+              "OIDC Authentication server url")
+          .define(
+              OIDC_ACCEPTED_AUDIENCE,
+              ConfigDef.Type.LIST,
+              "",
+              ConfigDef.Importance.HIGH,
+              "Array of accepted OIDC clients")
+          .define(
+              OIDC_USER_NAME_CLAIM,
+              ConfigDef.Type.STRING,
+              "preferred_username",
+              ConfigDef.Importance.HIGH,
+              "Access token claim used to identify the user in the database")
+          .define(
+              OIDC_USER_ROLES_CLAIMS,
+              ConfigDef.Type.LIST,
+              "",
+              ConfigDef.Importance.HIGH,
+              "Array of claims used to identify the user roles in the database")
+          .define(
+              OIDC_USE_TLS,
+              ConfigDef.Type.BOOLEAN,
+              false,
+              ConfigDef.Importance.HIGH,
+              "Set to true to initiate a TLS encrypted connections. Defaults to false. If case of "
+                  + "truststore_path being set, it will default to true.")
+          .define(
+              OIDC_TRUSTSTORE_PATH,
+              ConfigDef.Type.STRING,
+              "",
+              ConfigDef.Importance.HIGH,
+              "Sets the keystore path with valid certificates to establish encrypted connections with the OIDC Server.")
+          .define(
+              OIDC_TRUSTSTORE_PASSWORD,
+              ConfigDef.Type.PASSWORD,
+              "",
+              ConfigDef.Importance.HIGH,
+              "Keystore password")
+          .define(
+              OIDC_TRUSTSTORE_TYPE,
+              ConfigDef.Type.STRING,
+              "jks",
+              ConfigDef.Importance.HIGH,
+              "Keystore type (JKS or PKCS12). By default, the trustore type is inferred from the keystore "
+                  + "extension. For `.p12`, `.pkcs12`, or `.pfx` files, PKCS12 is used. For `.keystore` or "
+                  + "`.jks` files, JKS is used. In case the keystore type cannot be inferred from the extension, "
+                  + "this option must be set.")
+          .define(
+              OIDC_ALLOW_TOKEN_VIA_PASSWORD_AUTHENTICATION,
+              ConfigDef.Type.BOOLEAN,
+              false,
+              ConfigDef.Importance.HIGH,
+              "Compatibility option, allowing older drivers to use OIDC authentication via password "
+                  + "authentication, by using a pre-configurable username and passing an OIDC JWT access token as "
+                  + "password.")
+          .define(
+              OIDC_TOKEN_USERNAME,
+              ConfigDef.Type.STRING,
+              "oauth-bearer",
+              ConfigDef.Importance.HIGH,
+              "When the above option is true, it allows configuring the username which will be used "
+                  + "to detect that an OIDC JWT access token is being sent as password.");
 
   @Nullable private final Path keyTabPath;
 
@@ -101,9 +223,14 @@ public AuthenticatorConfig(Map<String, String> authSettings) {
       if (getService().isEmpty()) {
         throw new ConfigException(SERVICE_OPT, "<empty>", "is required");
       }
-
       assertAccessibleFile(keyTabPath, KEYTAB_OPT);
     }
+
+    if (provider == Provider.OIDC) {
+      if (getString(OIDC_ISSUER).isEmpty()) {
+        throw new ConfigException(OIDC_ISSUER, "<empty>", "is required");
+      }
+    }
   }
 
   /**
@@ -139,6 +266,19 @@ private static Map<String, String> sanitizeAuthSettings(Map<String, String> auth
         mutated.put(PRINCIPAL_OPT, getPrincipalFromKeyTab(keyTabPath.toString()));
       }
     }
+
+    if ("OIDC".equals(provider)) {
+      // OIDC provider requires issuer to be set.
+      if (!mutated.containsKey(OIDC_ISSUER) || mutated.get(OIDC_ISSUER).isEmpty()) {
+        throw new ConfigException(OIDC_ISSUER, "<empty>", "is required for OIDC authentication");
+      }
+      // If truststore path is set, use TLS.
+      if (mutated.containsKey(OIDC_TRUSTSTORE_PATH)
+          && !mutated.get(OIDC_TRUSTSTORE_PATH).isEmpty()) {
+        mutated.put(OIDC_USE_TLS, "true");
+      }
+    }
+
     return ImmutableMap.<String, String>builder().putAll(mutated).build();
   }
 
@@ -191,7 +331,7 @@ public Provider getProvider() {
       return Provider.valueOf(providerString);
     } catch (IllegalArgumentException e) {
       throw new ConfigException(
-          PROVIDER_OPT, providerString, "valid values are None, PLAIN, GSSAPI");
+          PROVIDER_OPT, providerString, "valid values are None, PLAIN, GSSAPI, OIDC");
     }
   }
 
@@ -226,12 +366,23 @@ public String toString() {
         PASSWORD_OPT,
         KEYTAB_OPT,
         PRINCIPAL_OPT,
-        SERVICE_OPT);
+        SERVICE_OPT,
+        // OIDC opts
+        OIDC_ISSUER,
+        OIDC_ACCEPTED_AUDIENCE,
+        OIDC_USER_NAME_CLAIM,
+        OIDC_USE_TLS,
+        OIDC_TRUSTSTORE_PATH,
+        OIDC_TRUSTSTORE_PASSWORD,
+        OIDC_TRUSTSTORE_TYPE,
+        OIDC_ALLOW_TOKEN_VIA_PASSWORD_AUTHENTICATION,
+        OIDC_TOKEN_USERNAME);
   }
 
   public enum Provider {
     None,
     PLAIN,
-    GSSAPI
+    GSSAPI,
+    OIDC
   }
 }
diff --git a/common/src/main/java/com/datastax/oss/common/sink/state/LifeCycleManager.java b/common/src/main/java/com/datastax/oss/common/sink/state/LifeCycleManager.java
@@ -620,11 +620,20 @@ private static void processAuthenticatorConfig(
           .withStringMap(
               AUTH_PROVIDER_SASL_PROPERTIES, ImmutableMap.of("javax.security.sasl.qop", "auth"))
           .withStringMap(DseDriverOption.AUTH_PROVIDER_LOGIN_CONFIGURATION, loginConfig);
+    } else if (authConfig.getProvider() == AuthenticatorConfig.Provider.OIDC) {
+      /*
+      TODO: OidcApiAuthProvider.
+        configLoaderBuilder
+          .withClass(AUTH_PROVIDER_CLASS, OidcApiAuthProvider.class)
+          .withString(AUTH_PROVIDER_SERVICE, authConfig.getService())
+          ....
+       */
+      throw new RuntimeException("TODO: OIDC authentication is not supported yet.");
     }
   }
 
   /**
-   * Prepare insert or update (depending on whether or not the table is a COUNTER table), and delete
+   * Prepare insert or update (depending on whether the table is a COUNTER table), and delete
    * statements asynchronously.
    *
    * @param session the session
