Page History

package org.apache.kafka.common.security.oauthbearer.smo;

 
    /**
 * A {@code Callback} related *to Identifyintrospection thatrequests theagainst optiona identifiedJAAS
 by* {@link #optionName()}, on the instance
     * returned by {@link #substitutableModuleOptions()}, has had substitution
     * performed for it resulting in the given password-related value. This method
     * can be successfully invoked (and is idempotent) only when the option existsconfiguration
 *
 * @see SubstitutableModuleOptionsCallbackHandler
 * @see SubstitutableModuleOptions
 */
public class SubstitutableModuleOptionsCallback implements Callback {
    private final SubstitutableModuleOptions substitutableModuleOptions;
    private final String optionName;
    private final boolean optionRequiredToExist;

    /**
     * Constructor
     *
     * @param substitutableModuleOptions
     * and its value isn't being changed.            the mandatory substitutable module options
     * @param optionName
     *            the requested option
     * @param substitutionValueoptionRequiredToExist
     *            if true then the mandatoryrequested password-relatedoption substitutionis valuerequired to setexist
     */
    public void setSubstitutionValue(Password substitutionValue SubstitutableModuleOptionsCallback(SubstitutableModuleOptions substitutableModuleOptions, String optionName,
            boolean optionRequiredToExist) {
        this.substitutableModuleOptions = Objects.setSubstitutionValue(optionName,requireNonNull(substitutableModuleOptions);
        this.optionName = Objects.requireNonNull(substitutionValue))optionName);
        this.optionRequiredToExist = optionRequiredToExist;
    }
 
    /**
     * Return the substitutable substitution value, if any has been set, otherwise null. Anymodule options provided at construction time
     *
     * non-null@return valuethe willsubstitutable bemodule eitheroptions aprovided {@codeat String} or a {@code Password}. Noteconstruction time
     * that the value may not have been set via a call to/
    public * {@link #setSubstitutionValue(Password)} orSubstitutableModuleOptions substitutableModuleOptions() {
     * {@link #setSubstitutionValue(String)}; it is possible that the value was return substitutableModuleOptions;
    }

     /**
 already set prior to construction* ofReturn thisthe instance,option inname whichprovided caseat that value
     * will be returned here.construction time
     *
     * @return the substitutionoption value,name ifprovided anyat has been set, otherwise nullconstruction time
     */
    public ObjectString substitutionValueoptionName() {
        return substitutableModuleOptions.moduleOptionSubstitutionState().get(optionName);
    }
 
    /**
     * Return true if the substitutionthe text,requested ifoption anyis hasrequired beento setexist, otherwise null.false
     * 
     * @return thetrue substitutionif value,the if any has been setthe requested option is required to exist, otherwise null.
     */
      public String substitutionText() {false
     */
   Object substitutionValuepublic =boolean substitutionValueoptionRequiredToExist(); {
        return substitutionValueoptionRequiredToExist;
 instanceof Password ? ((Password) substitutionValue).value() }
 
    /**
     * Identify that the option identified by :{@link #optionName(String) substitutionValue;}, on the instance
    }
}

package org.apache.kafka.common.security.oauthbearer.smo;

/**
 *returned Aby {@code@link CallbackHandler#substitutableModuleOptions()}, thathas handleshad introspectionsubstitution
 requests against a JAAS
 * configuration
performed *
for *it @seeyielding SubstitutableModuleOptionsCallback
the *given @seeresult. SubstitutableModuleOptions
 */
public class SubstitutableModuleOptionsCallbackHandler implements AuthenticateCallbackHandler {
    @Override
    public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
        for (Callback callback : callbacks) {
            if (callback instanceof SubstitutableModuleOptionsCallback) {This method is idempotent;
     * invoking it with a substitution result equal to the option's current
     * substitution result (as defined by {@code Object.equals()}) has no effect.
     * The substitution result for an option cannot be changed (again, as defined by
     * {@code Object.equals()}) once it has been set; an attempt to SubstitutableModuleOptionsCallbackdo substitutableModuleOptionsCallbackso = (SubstitutableModuleOptionsCallback) callback;will raise
     * an exception.
     *
    Object substitutionValue* =@param getSubstitutionValue(substitutionResult
     *            the mandatory substitution result    substitutableModuleOptionsCallback.substitutableModuleOptions(),to set
     */
    public void setSubstitutionResult(RedactableObject substitutionResult) {
           substitutableModuleOptionsCallback.optionName(),substitutableModuleOptions.setSubstitutionResult(optionName, Objects.requireNonNull(substitutionResult));
    }

    /**
     * Return the substitution result, if any has been set, otherwise substitutableModuleOptionsCallback.optionRequiredToExist());
 null. Note
               // depend on idempotence here because* that the substitutionresult valuemay hasnot alreadyhave been
 set via a call to
     * {@link #setSubstitutionResult(RedactableObject)}; it is possible // set on the underlying SubstitutableModuleOptions instancethat the
     * result was already set prior to construction of this instance, ifin (substitutionValue instanceof String)which case
     * that result will be returned here.
      *
   substitutableModuleOptionsCallback.setSubstitutionValue((String) substitutionValue);
 * @return the substitution result, if any has been set, otherwise null
    else if (substitutionValue instanceof Password) */
    public RedactableObject substitutionResult() {
        return     substitutableModuleOptionsCallback.setSubstitutionValue((Password) substitutionValuesubstitutableModuleOptions.substitutionResults().get(optionName);
    } 
}

package org.apache.kafka.common.security.oauthbearer.smo;

/**
 * A {@code CallbackHandler} that handles introspection requests against a JAAS
 * configuration and supports substitution into option values via delimited text
 * of the following form:
 *
 * <pre>
 * &lt;OPENING_DELIMITER&gt;&lt;TYPE&gt;&lt;OPTIONAL_MODIFIERS&gt;=&lt;OPTIONAL_VALUE&gt;&lt;CLOSING_DELIMITER&gt;
 * </pre>
 *
 * Where the above elements are defined as follows:
 *
 * <pre>
 * OPENING_DELIMITER: $[, $[[, $[[[, $[[[[, or $[[[[[
 * CLLOSING_DELIMITER: ],  ]],  ]]],  ]]]], or  ]]]]] (number of brackets must match)
 *
 * TYPE: everything up to (but not including) the first punctuation character
 *
 * OPTIONAL_MODIFIERS: the optional section immediately after the TYPE, starting with any
        } else
                throw new UnsupportedCallbackException(callback,
                        String.format("Unrecognized Callback type: %s", callback.getClass().getName()));
        }
    }
 
    /**
     * Return the server configuration provided during
     * {@link #configure(Map, String, List)}, if any, otherwise null
     *
     * @return the server configuration provided during
     *         {@link #configure(Map, String, List)}, if any, otherwise null
     */
punctuation character except for publicthe Map<String,equal ?>sign serverConfig(=), {
and ending
 *         return serverConfig;
    }

    /**
   with the *same Returnpunctuation thecharacter SASLfollowed mechanismimmediately providedby duringan
 *     * {@link #configure(Map, String, List)}, if any, otherwise null
     *
     * @return the SASL mechanism provided during
             equal sign. The same punctuation character delimits individual
 *           *         {@link #configure(Mapmodifiers, String, List)}, if any, otherwise null
     */
which come in two flavors: flags, which do not contain
 *     public String mechanism() {
        return mechanism;
    }

an equal sign,  /**
     * Return the JAAS login module configurations provided during
 and name=value arguments, which do.
 *
 * OPTIONAL_VALUE: the optional section immediately after the modifier section-delimiting
 *    * {@link #configure(Map, String, List)}, if any, otherwise null
     *
punctuation character (if any) and *the @returnequal the JAAS login module configurations provided during
     *         {@link #configure(Map, String, List)}, if any, otherwise null
     */
    public List<AppConfigurationEntry> jaasConfigEntries() {
        return jaasConfigEntries;
    }
 
    /**
     * Convenience method to perform substitution without using callbacks and a
     * callback handler.
     *
     * @param substitutableModuleOptions
     *            the mandatory substitutable module options to query
     * @param optionName
     *            the mandatory requested option
     * @param optionRequiredToExist
     *            if true then the requested option is required to exist
     * @return the given option value, after any required substitution is applied,
     *         or null if the option does not exist and it was not required to exist
     * @throws IOException
     *             if a required substitution cannot be performed, including if the
     *             given -- or another -- required option does not exist
     */
    public static Object getSubstitutionValue(SubstitutableModuleOptions substitutableModuleOptions, String optionName,
            boolean optionRequiredToExist) throws IOException {
        if (!substitutableModuleOptions.optionExists(optionName)) {
            if (!optionRequiredToExist)
                return null;
            throw new IOException(String.format("Mandatory option does not exist: %s", optionName));
        }
        return getSubstitutionValue(substitutableModuleOptions, optionName, new HashSet<String>());
    }
 
    /**
     * Convenience method to perform substitution without using callbacks and a
     * callback handler.
     *
     * @param substitutableModuleOptions
     *            the mandatory substitutable module options to query
     * @param optionName
     *            the mandatory requested option
     * @param optionRequiredToExist
     *            if true then the requested option is required to exist
     * @return the given option text, after any required substitution is applied, or
     *sign (=). It is not trimmed.
 * </pre>
 *
 * For example:
 *
 * <pre>
 * $[envvar=THE_ENV_VAR]
 * $[envvar/notBlank/redact/=THE_ENV_VAR]
 * $[envvar/defaultValue = theDefaultValue/=THE_ENV_VAR]
 * $[envvar/defaultOption = theOptionName/=THE_ENV_VAR]
 * $[file|redact|notBlank|=/the/path/to/the/file]
 * </pre>
 *
 * Working left to right, once the delimiter is defined for a module option (for
 * example, {@code $[} and {@code ]}), only that delimiter is recognized for the
 * rest of that option (and it is always recognized as meaning substitution for
 * the rest of that module option). A special "empty" substitution does nothing
 * except, when it appears to the left of every other occurrence of matching
 * delimiters, it serves to force the delimiter for that module option to the
 * one indicated . For example, to force the delimiter to {@code $[[} and
 * {@code ]]} (and prevent {@code $[} and {@code ]} from causing substitution)
 * for a module option:
 *
 * <pre>
 * optionName = "$[[]]These $[ and ] delimiters do not cause substitution"
 * </pre>
 *
 * The following substitutions are supported, though it is straightforward to
 * add others (see below):
 *
 * <ul>
 * <li>{@code envvar}: substitute the value of an environment variable</li>
 * <li>{@code sysprop}: substitute the value of a system property</li>
 * <li>{@code file}: substitute the contents of a file</li>
 * <li>{@code option}: substitute the contents of another option</li>
 * </ul>
 *
 * The above substitutions support the following modifiers:
 *
 * <ul>
 * <li>{@code redact}: prevent values from being logged</li>
 * <li>{@code notEmpty}: the value must not be empty</li>
 * <li>{@code notBlank}: the value must not be blank (i.e. consisting only of
 * whitespace); implies {@code notEmpty}.</li>
 * <li>{@code defaultValue=<value>}: substitute the given literal value if the
 * substitution cannot otherwise be made (either because the value does not
 * exist or the determined value was disallowed because it was empty or blank).
 * The substituted value must satisfy any modifiers that act as
 * constraints.</li>
 * <li>{@code defaultOption=<optionName>}: substitute the value of the indicated
 * option if the substitution cannot otherwise be made (either because the value
 * does not exist or the determined value was disallowed because it was empty or
 * blank). The substituted value must satisfy any modifiers that act as
 * constraints.</li>
 * <li>{@code fromOption}: provides a level of indirection so that the value,
 * instead of always being literally specified (i.e. read this file, or this
 * environment variable), can be determined via some other option. This allows,
 * for example, the filename, system property name, etc. to be generated
 * potentially from multiple substitutions concatenated together.</li>
 * </ul>
 *
 * To add new substitutions simply define a module option of the following form:
 *
 * <pre>
 * typeModuleOptionSubstituter = "fully.qualified.class.name"
 * </pre>
 *
 * For example:
 *
 * <pre>
 * fooModuleOptionSubstituter = "org.example.FooSubstituter"
 * </pre>
 *
 * The indicated class must implement {@link ModuleOptionSubstituter}, and you
 * can invoke the substitution like this:
 *
 * <pre>
 * $[foo/optional/modifiers/=optionalValue]
 * </pre>
 *
 * @see SubstitutableModuleOptionsCallback
 * @see SubstitutableModuleOptions
 * @see ModuleOptionSubstituter
 */
public class SubstitutableModuleOptionsCallbackHandler implements AuthenticateCallbackHandler {
    // etc...
 
    @Override
    public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
        for (Callback callback : callbacks) {
           null if the(callback optioninstanceof doesSubstitutableModuleOptionsCallback) not{
 exist and it was not required to exist
     * @throws IOException
 SubstitutableModuleOptionsCallback substitutableModuleOptionsCallback =  *(SubstitutableModuleOptionsCallback) callback;
             if a required substitutionRedactableObject cannotsubstitutionResult be performed, including if the= getSubstitutionResult(
     *             given -- or another -- required option does not exist
 substitutableModuleOptionsCallback.substitutableModuleOptions(),
              */
    public static String getSubstitutionText(SubstitutableModuleOptions substitutableModuleOptions, String substitutableModuleOptionsCallback.optionName(),
            boolean optionRequiredToExist) throws IOException {
        Object substitutionValue = getSubstitutionValue(substitutableModuleOptions, optionName, optionRequiredToExistsubstitutableModuleOptionsCallback.optionRequiredToExist(), forceDebugLogForTesting);
        if (substitutionValue instanceof String)
     if (substitutionObject != null)
    return (String) substitutionValue;
        if (substitutionValue instanceof Password)
   // depend on idempotence here because the substitution object return ((Password) substitutionValue).value();has already been
        return null;
    }
 
    /*
   // set *on Handlethe substitution,underlying dealingSubstitutableModuleOptions withinstance
 circular references, constraints, etc. It
     * is an error if the indicated option does not existsubstitutableModuleOptionsCallback.setSubstitutionResult(substitutionObject);
     */
    private static Object getSubstitutionValue(SubstitutableModuleOptions substitutableModuleOptions,} else
            String optionNameRequiredToExist, HashSet<String> optionsSeen) throwsthrow IOException {
new UnsupportedCallbackException(callback,
         // etc...
      }
 
    // etc...
}

package org.apache.kafka.common.security.oauthbearer.smo;

/**
 * An object whose text value can be redacted
 */
public class RedactableObject {
    static final String REDACTED = "[redacted]";
    private final Object object;
    private final boolean redacted;

    /**
     * Constructor
     *
     * @param object
     *            the mandatory object
     * @param redact
     *            when true the object's value will be redacted in
     *            {@link #redactedText()}
     */
    public RedactableObject(Object object, boolean redact) {
        this.object = Objects.requireNonNull(object);
        this.redacted = redact;
    }


    /**
     * Return the (always non-null) underlying object provided during instance
     * construction
     *
     * @return the (always non-null) underlying object provided during instance
     *         construction
     */
    public Object object() {
        return object;
    }
 
    /**
     * Return true if this instance contains information that will be redacted when
     * {@link #redactedText()} is invoked, otherwise false
     *
     * @return true if this instance contains information that will be redacted when
     *         {@link #redactedText()} is invoked, otherwise false
     */
    public boolean isRedacted() {
        return redacted;
    }

    /**
     * Return the redacted text for this instance, if redaction is required,
     * otherwise return the {@link #value()}
     *
     * @return the redacted text for this instance, if redaction is required,
     *         otherwise return the {@link #value()}
     */
    public String redactedText() {
        return redacted ? REDACTED : value();
    }

    /**
     * Return the {@code String} value of this instance, including information that
     * would otherwise be redacted
     *
     * @return the {@code String} value of this instance, including information that
     *         would otherwise be redacted
     */
    public String value() {
        if (object instanceof String)
            return (String) object;
        throw new IllegalStateException(
                String.format("Unknown substitution result object type: %s", object.getClass().getName()));
    }
 
    /**
     * Return true if this result is considered to be empty, otherwise false
     *
     * @return true if this result is considered to be empty, otherwise false
     */
    public boolean isEmpty() {
        return value().isEmpty();
    }

    /**
     * Return true if this result is considered to be blank (containing at most just
     * whitespace), otherwise false
     *
     * @return true if this result is considered to be blank (containing at most
     *         just whitespace), otherwise false
     */
    public boolean isBlank() {
        return value().trim().isEmpty();
    }

    /**
     * Return this instance if it is redacted according to {@link #isRedacted()},
     * otherwise return a new, redacted instance with the same underlying object
     *
     * @return this instance if it is redacted according to {@link #isRedacted()},
     *         otherwise return a new, redacted instance with the same underlying
     *         object
     */
    public RedactableObject redactedVersion() {
        return redacted ? this : new RedactableObject(object, true);
    }
 
    @Override
    public String toString() {
        // be sure to redact information as required
        return redactedText();
    }


    @Override
    public boolean equals(Object obj) {
        if (!(obj instanceof RedactableObject))
            return false;
        RedactableObject other = (RedactableObject) obj;
        /*
         * Note that differences in redaction cause inequality
         */
        return redacted == other.redacted && object.equals(other.object);
    }


    @Override
    public int hashCode() {
        return object.hashCode();
    }
} 

package org.apache.kafka.common.security.oauthbearer.smo;

/**
 * The interface that pluggable module option substituter classes must
 * implement.
 *
 * @see SubstitutableModuleOptionsCallbackHandler
 */
public interface ModuleOptionSubstituter {
    /**
     * Perform the substitution of the given type on the given options using the
     * given modifiers and value
     *
     * @param type
     *            the (always non-null) type of substitution to perform
     * @param modifiers
     *            the (always non-null but potentially empty) modifiers to apply, if
     *            any. They are presented exactly as they appear in the
     *            configuration, with no whitespace trimming applied.
     * @param value
     *            the always non-null (but potentially empty) value
     * @param substitutableModuleOptions
     *            the module options and their current substitution state
     * @return the (always non-null) result of performing the substitution
     * @throws IOException
     *             if the substitution cannot be performed
     */
    RedactableObject doSubstitution(String type, List<String> modifiers, String value,
            SubstitutableModuleOptions substitutableModuleOptions) throws IOException;
}

package org.apache.kafka.common.security.oauthbearer.smo;

/**
 * A template {@code ModuleOptionSubstituter} that handles the following
 * modifiers:
 * <ul>
 * <li>{@code redact} -- when enabled, values are stored such that they are
 * prevented from being logged</li>
 * <li>{@code notBlank} -- when enabled, blank (only whitespace) or non-existent
 * values are replaced by default values. Implies {@code notEmpty}.</li>
 * <li>{@code notEmpty} -- when enabled, either explicitly or via
 * {@code notBlank}, empty {@code ""} or non-existent values are replaced by
 * default values.</li>
 * <li>{@code fromOption} -- when enabled, provides a level of indirection such
 * that the option's value is taken as an option name from which the actual
 * value will be retrieved</li>
 * <li>{@code defaultValue=<value>} -- when enabled, the provided value is used
 * as a default value in case the requested value either does not exist or is
 * disallowed via {@code notBlank} or {@code notEmpty}</li>
 * <li>{@code defaultOption=<value>} -- when enabled, the indicated option is
 * evaluated as a default value in case the requested value either does not
 * exist or is disallowed via {@code notBlank} or {@code notEmpty}</li>
 * </ul>
 *
 * Flags (modifiers without an equal sign) are trimmed, so "{@code redact}" and
 * "{@code  redact }" are recognized as being the same. Arguments (modifiers
 * with an equal sign) have their name trimmed but not their value, so
 * "{@code name=value}" and "{@code  name = value }" are both recognized as
 * setting the {@code name} argument (though their values do not match due to
 * whitespace differences).
 * <p>
 * It is an error to set the same named argument multiple times (even if the
 * values are the same). Redundantly specifying the same flag is acceptable.
 * <p>
 * Flags and arguments are presented to the substitution's implementation via
 * the
 * {@link #retrieveResult(String, String, boolean, SubstitutableModuleOptions, Set, Map)}
 * method.
 */
public abstract class SubstituterHelper implements ModuleOptionSubstituter {
    /**
     * Retrieve the substitution result associated with the given key, if any,
     * otherwise null
     *
     * @param type
     *            the (always non-null) type of substitution to perform
     * @param key
     *            the required key
     * @param redact
     *            if the result must be redacted regardless of any information to
     *            the contrary
     * @param substitutableModuleOptions
     *            the module options and their current substitution state
     * @param flags
     *            the flags specified, if any, beyond the standard {@code redact},
     *            {@code notBlank}, {@code notEmpty}, and {@code fromOption} flags
     * @param args
     *            the arguments specified, if any, beyond the standard
     *            {@code defaultValue} and {@code defaultOption} arguments
     * @return the substitution result associated with the given key, if any,
     *         otherwise null
     * @throws IOException
     *             if the request cannot be performed such that the use of a default
     *             value would be inappropriate
     */
    public abstract RedactableObject retrieveResult(String type, String key, boolean redact,
            SubstitutableModuleOptions substitutableModuleOptions, Set<String> flags, Map<String, String> args)
            throws IOException;


    @Override
    public RedactableObject doSubstitution(String type, List<String> modifiers, String value,
            SubstitutableModuleOptions substitutableModuleOptions) throws IOException {
        // etc...
    }
 
    // etc...
}

package org.apache.kafka.common.security.oauthbearer.smo;

/**
 * A {@code ModuleOptionSubstituter} that handles module option substitution.
 * <p>
 * Note that the {@code fromOption} modifier, if present, still adds an
 * additional level of indirection: the value indicates the option whose value,
 * in turn, indicates the option text to retrieve.
 */
public class OptionSubstituter implements ModuleOptionSubstituter {
    private final SubstituterHelper substituterHelper;
    public OptionSubstituter() {
        substituterHelper = new SubstituterHelper() {
            @Override
            public RedactableObject retrieveResult(String type, String key, boolean redact,
                    SubstitutableModuleOptions substitutableModuleOptions, Set<String> flags, Map<String, String> args)
                    throws IOException {
                if (!flags.isEmpty() || !args.isEmpty())
                    throw new IOException(String.format("Unknown extra flags/args: %s; %s", flags, args));
                RedactableObject substitutionResult = SubstitutableModuleOptionsCallbackHandler
                        .getSubstitutionResult(substitutableModuleOptions, key, false);
                return redact ? substitutionResult.redactedVersion() : substitutionResult;
            }
        };
    }
    @Override
    public RedactableObject doSubstitution(String type, List<String> modifiers, String value,
            SubstitutableModuleOptions substitutableModuleOptions) throws IOException {
        return substituterHelper.doSubstitution(type, modifiers, value, substitutableModuleOptions);
    }
}

package org.apache.kafka.common.security.oauthbearer.smo;
 
/**
 * A {@code ModuleOptionSubstituter} that handles file content substitution.
 */
public class FileContentSubstituter implements ModuleOptionSubstituter {
    private final SubstituterHelper substituterHelper;
 
    public FileContentSubstituter() {
        substituterHelper = new SubstituterHelper() {
            @Override
            public RedactableObject retrieveResult(String type, String key, boolean redact,
                    SubstitutableModuleOptions substitutableModuleOptions, Set<String> flags, Map<String, String> args)
                    throws IOException {
                if (!flags.isEmpty() || !args.isEmpty())
                    throw new IOException(String.format("Unknown extra flags/args: %s; %s", flags, args));
                Path path = Paths.get(key);
                if (!Files.exists(path))
                    return null;
                BasicFileAttributes fileAttributes = Files.getFileAttributeView(path, BasicFileAttributeView.class)
                        .readAttributes();
                if (!fileAttributes.isRegularFile())
                    return null;
                long fileSize = fileAttributes.size();
                int maxSize = 1024 * 1024;
                if (fileSize > maxSize) {
                    throw new IOException(String.format("Type=%s: key=%s: file size exceeds max of %d: %d", type, key,
                            maxSize, fileSize));
                }
                String retval = new String(Files.readAllBytes(path), StandardCharsets.UTF_8);
                return new RedactableObject(retval, redact);
            }
        };
    }

    @Override
    public RedactableObject doSubstitution(String type, List<String> modifiers, String value,
            SubstitutableModuleOptions substitutableModuleOptions) throws IOException {
        return substituterHelper.doSubstitution(type, modifiers, value, substitutableModuleOptions);
    }
}

package org.apache.kafka.common.security.oauthbearer.smo;
 
/**
 * A {@code ModuleOptionSubstituter} that handles system property substitution.
 * The JAAS config file already supports system property substitution via the
 * '{@code ${}' and '}' delimiters, but we support it here for syntactic
 * consistency as well as to add the functionality associated with modifiers.
 */
public class SystemPropertySubstituter implements ModuleOptionSubstituter {
    private final SubstituterHelper substituterHelper;

    public SystemPropertySubstituter() {
        substituterHelper = new SubstituterHelper() {
            @Override
            public RedactableObject retrieveResult(String type, String key, boolean redact,
                    SubstitutableModuleOptions substitutableModuleOptions, Set<String> flags, Map<String, String> args)
                    throws IOException {
                if (!flags.isEmpty() || !args.isEmpty())
                    throw new IOException(String.format("Unknown extra flags/args: %s; %s", flags, args));
                String retval = System.getProperty(key);
                return retval != null ? new RedactableObject(retval, redact) : null;
            }
        };
    }

    @Override
    public RedactableObject doSubstitution(String type, List<String> modifiers, String value,
            SubstitutableModuleOptions substitutableModuleOptions) throws IOException {
        return substituterHelper.doSubstitution(type, modifiers, value, substitutableModuleOptions);
    }
}

package org.apache.kafka.common.security.oauthbearer.smo;
 
/**
 * A {@code ModuleOptionSubstituter} that handles environment variable
 * substitution
 */
public class EnvironmentVariableSubstituter implements ModuleOptionSubstituter {
    private final SubstituterHelper substituterHelper;


    public EnvironmentVariableSubstituter() {
        substituterHelper = new SubstituterHelper() {
            @Override
            public RedactableObject retrieveResult(String type, String key, boolean redact,
                    SubstitutableModuleOptions substitutableModuleOptions, Set<String> flags, Map<String, String> args)
                    throws IOException {
                if (!flags.isEmpty() || !args.isEmpty())
                    throw new IOException(String.format("Unknown extra flags/args: %s; %s", flags, args));
                String retval = System.getenv(key);
                return retval != null ? new RedactableObject(retval, redact) : null;
            }
        };
    }

    @Override
    public RedactableObject doSubstitution(String type, List<String> modifiers, String value,
            SubstitutableModuleOptions substitutableModuleOptions) throws IOException {
        return substituterHelper.doSubstitution(type, modifiers, value, substitutableModuleOptions);
    }
}

SubstitutableModuleOptions options = new SubstitutableModuleOptions(moduleOptionsMap);
boolean requiredToExist = true;
SubstitutableModuleOptionsCallback callback = new SubstitutableModuleOptionsCallback(options, "thePassword", requiredToExist);
CallbackHandler callbackHandler = new SubstitutableModuleOptionsCallbackHandler()
callbackHandler.handle(new Callback[] {callback});
String thePassword = callback.substitutionResult().value();

SubstitutableModuleOptions options = new SubstitutableModuleOptions(moduleOptionsMap);
boolean requiredToExist = true;
String thePassword = SubstitutableModuleOptionsCallbackHandler.getSubstitutionResult(substitutableModuleOptions,
        "thePassword", requiredToExist).value();