package org.apache.kafka.common.security.substitutions;
 
/**
 * The map-like interface that the input to {@link SubstitutableValues} must
 * implement.
 */
 
public interface UnderlyingValues {
    /**
     * Return the value associated with the given key, if any, otherwise null
     * 
     * @param key
     *            the mandatory key
     * @return the value associated with the given key, if any, otherwise null
     */
    Object get(String key);
}

package org.apache.kafka.common.security.substitutions;

/**
 * Adds support for substitution within the values of an instance of
 * {@link UnderlyingValues}. Substitution is accomplished via delimited text
 * within a value of the following form:
 *
 * <pre>
 * &lt;OPENING_DELIMITER&gt;&lt;TYPE&gt;&lt;OPTIONAL_MODIFIERS&gt;=&lt;OPTIONAL_IDENTIFIER&gt;&lt;CLOSING_DELIMITER&gt;
 * </pre>
 *
 * Where the above elements are defined as follows:
 *
 * <pre>
 * OPENING_DELIMITER: $[, $[[, $[[[, $[[[[, or $[[[[[
 * CLOSING_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
 *                     punctuation character except for the equal sign (=), and ending
 *                     with the same punctuation character followed immediately by an
 *                     equal sign. The same punctuation character delimits individual
 *                     modifiers, which come in two flavors: flags, which do not contain
 *                     an equal sign, and name=value arguments, which do.
 *
 * OPTIONAL_IDENTIFIER: the optional section immediately after any modifiers and the equal
 *                      sign (=).
 * </pre>
 *
 * For example:
 *
 * <pre>
 * $[envVar=THE_ENV_VAR]
 * $[envVar/notBlank/redact/=THE_ENV_VAR]
 * $[envVar/defaultValue = theDefaultValue/=THE_ENV_VAR]
 * $[envVar/defaultKey = theDefaultKey/=THE_ENV_VAR]
 * $[file|redact|notBlank|=/the/path/to/the/file]
 * </pre>
 *
 * Once an underlying value is retrieved and its calculated value -- whether
 * different from the raw underlying value due to a substitution or not -- is
 * determined, this instance will not retrieve the underlying value again; the
 * calculated value will be used if it is referred to. This means if the
 * underlying values are expected to change then to see those changes a new
 * instance of this class must be allocated.
 * <p>
 * Working left to right, once the delimiters are defined for a value (for
 * example, {@code $[} and {@code ]}), only those delimiters are recognized for
 * the rest of that value (and they are always recognized as meaning
 * substitution for the rest of that value). 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 value to the
 * one indicated. For example, to force the delimiter to {@code $[[} and
 * {@code ]]} (and prevent {@code $[} and {@code ]} from causing substitution)
 * for a value:
 *
 * <pre>
 * someKey = "$[[]]These $[ and ] delimiters do not cause substitution"
 * </pre>
 *
 * The following built-in substitution types are supported, though it is
 * straightforward to add others (see below):
 *
 * <ul>
 * <li>{@code envVar}: substitute an environment variable, typically indicated
 * by the identifier</li>
 * <li>{@code sysProp}: substitute a system property, typically indicated by the
 * identifier</li>
 * <li>{@code file}: substitute the contents of a file, typically indicated by
 * the identifier</li>
 * <li>{@code keyValue}: substitute the contents of another key's value,
 * typically indicated by the identifier (and that key's value has substitution
 * performed on it if necessary)</li>
 * </ul>
 *
 * The built-in substitution types support the following flags, which are
 * trimmed and may be redundantly specified:
 * <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 fromValueOfKey}: provides a level of indirection so that the
 * identifier, instead of being used directly (i.e. read the indicated file, or
 * the indicated environment variable), identifies another key whose value is
 * used as the identifier instead. This allows, for example, the filename,
 * system property name, etc. to potentially be generated from multiple
 * substitutions concatenated together.</li>
 * </ul>
 * The built-in substitution types support the following arguments, whose names
 * are trimmed but whose values are not; it is an error to specify the same
 * named argument multiple times (even if the values are identical):
 * <ul>
 * <li>{@code defaultValue=<value>}: substitute the given literal value if the
 * substitution cannot otherwise be made (either because the identifier
 * indicates something that does not exist or the determined value was
 * disallowed because it was empty or blank). The substituted default value must
 * satisfy any {@code notBlank} or {@code notEmpty} modifiers that act as
 * constraints, otherwise it is an error.</li>
 * <li>{@code defaultKey=<key>}: substitute the value associated with the
 * indicated key if the substitution cannot otherwise be made (either because
 * the identifier indicates something that does not exist or the determined
 * value was disallowed because it was empty or blank). The value that is
 * ultimately substituted must satisfy any {@code notBlank} or {@code notEmpty}
 * modifiers that act as constraints, otherwise it is an error.</li>
 * </ul>
 *
 * To add new substitutions beyond the built-in ones mentioned above simply
 * define a key/value pair of the following form:
 *
 * <pre>
 * [optionalTypeDefinitionKeyPrefix]&lt;type&gt;SubstituterType = "fully.qualified.class.name"
 * </pre>
 *
 * For example:
 *
 * <pre>
 * fooSubstituterType = "org.example.FooSubstituterType"
 * </pre>
 *
 * The indicated class must implement {@link SubstituterType}, and you can
 * invoke the substitution in a value like this:
 *
 * <pre>
 * $[foo/optional/modifiers/=optionalValue]
 * </pre>
 *
 * The type definition prefix is defined at construction time and may be empty.
 *
 * A parsing error within recognized delimiters results in the delimiters and
 * the intervening text that could not be parsed being left alone. For example,
 * the following text would be passed through unchanged because the delimited
 * text cannot be parsed as a valid substitution request:
 * {@code qw$[asd_4Q!]uH6}.
 *
 * @see SubstituterType
 * @see SubstituterTypeHelper
 */
public class SubstitutableValues {
    /**
     * Constructor where the type definition key prefix is empty
     *
     * @param underlyingMap
     *            the mandatory underlying map. It is not copied, so it should be
     *            immutable. Results are unspecified if it is mutated in any manner.
     */
    public SubstitutableValues(UnderlyingValues underlyingValues) {
        this("", underlyingValues);
    }

    /**
     * Constructor with the given type definition key prefix
     *
     * @param typeDefinitionKeyPrefix
     *            the mandatory (but possibly empty) type definition key prefix
     * @param underlyingMap
     *            the mandatory underlying map. It is not copied, so it should be
     *            immutable. Results are unspecified if it is mutated in any manner.
     */
    public SubstitutableValues(String typeDefinitionKeyPrefix, UnderlyingValues underlyingValues) {
        // etc...
    }
    /**
     * Return the always non-null (but possibly empty) type definition key prefix
     *
     * @return the always non-null (but possibly empty) type definition key prefix
     */
    public String typeDefinitionKeyPrefix() {
        return typeDefinitionKeyPrefix;
    }


    /**
     * Return the underlying values provided during construction
     *
     * @return the underlying values provided during construction
     */
    public UnderlyingValues underlyingValues() {
        return underlyingValues;
    }
    /**
     * Return an unmodifiable map identifying which keys have been processed for
     * substitution and the corresponding result (if any). A key is guaranteed to
     * have been processed for substitution and its name will appear as a key in the
     * returned map only after {@link #getSubstitutionResult(String)} has been
     * invoked for that key either directly or indirectly because some other key's
     * substitution result depends on the substitution result of the key.
     *
     * @return an unmodifiable map identifying which keys have been processed for
     *         substitution and the corresponding result (if any)
     */
    public Map<String, RedactableObject> substitutionResults() {
        // etc...
    }

    /**
     * Perform substitution if necessary and return the resulting value for the
     * given key
     *
     * @param key
     *            the mandatory requested key
     * @param requiredToExist
     *            if true then the requested key is required to exist
     * @return the given key's substitution result, after any required substitution
     *         is applied, or null if the key does not exist and it was not required
     *         to exist
     * @throws IOException
     *             if a required substitution cannot be performed, including if the
     *             given (or any other) required key does not exist
     */
    public RedactableObject getSubstitutionResult(String key, boolean requiredToExist) throws IOException {
        // etc...
    }

    // etc...
}

package org.apache.kafka.common.security.substitutions;

/**
 * The single-method interface that pluggable substituter types must implement.
 */
public interface SubstituterType {
    /**
     * Perform the substitution of the given type using the given modifiers and
     * value on the given options
     *
     * @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 identifier
     *            the always non-null (but potentially empty) identifier, which is
     *            interpreted in the context of the substitution of the indicated
     *            type. For example, it may indicate an environment variable name, a
     *            filename, etc.
     * @param substitutableValues
     *            the values 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 identifier,
            SubstitutableValues substitutableValues) throws IOException;
}

package org.apache.kafka.common.security.substitutions;

/**
 * A template {@code SubstituterType} that handles the following modifiers:
 * <ul>
 * <li>{@code redact} -- when enabled, results are stored such that they are
 * prevented from being logged</li>
 * <li>{@code notBlank} -- when enabled, blank (only whitespace) or non-existent
 * results 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 results are replaced by
 * default values.</li>
 * <li>{@code fromValueOfKey} -- provides a level of indirection so that the
 * identifier, instead of always being literally specified (i.e. read this
 * particular file, or this particular environment variable), can be determined
 * via some other key's value. This allows, for example, the filename, system
 * property name, etc. to potentially be generated from multiple substitutions
 * concatenated together.</li>
 * <li>{@code defaultValue=<value>} -- when enabled, the provided literal value
 * is used as a default value in case the result either does not exist or is
 * disallowed via {@code notBlank} or {@code notEmpty}</li>
 * <li>{@code defaultKey=<key>} -- when enabled, the indicated key is evaluated
 * as a default value in case the result 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 type's implementation
 * via the
 * {@link #retrieveResult(String, String, boolean, Set, Map, SubstitutableValues)}
 * method.
 * <p>
 * Implementations of the {@link SubstituterType} interface that wish to
 * leverage the help provided by this class can either extend this class
 * directly or delegate to an anonymous class that extends this one.
 */
public abstract class SubstituterTypeHelper implements SubstituterType {
    /**
     * Retrieve the substitution result associated with the given identifier, if
     * any, otherwise null
     *
     * @param type
     *            the (always non-null/non-blank) type of substitution being
     *            performed
     * @param identifier
     *            the required (though potentially empty) identifier as interpreted
     *            by the substitution implementation for the given type. For
     *            example, it may be a filename, system property name, environment
     *            variable name, etc.
     * @param redact
     *            if the result must be redacted regardless of any information to
     *            the contrary
     * @param additionalFlags
     *            the flags specified, if any, beyond the standard {@code redact},
     *            {@code notBlank}, {@code notEmpty}, and {@code fromValueOfKey}
     *            flags
     * @param additionalArgs
     *            the arguments specified, if any, beyond the standard
     *            {@code defaultValue} and {@code defaultKey} arguments
     * @param substitutableValues
     *            the key/value mappings and their current substitution state
     * @return the substitution result associated with the given identifier, 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 identifier, boolean redact,
            Set<String> additionalFlags, Map<String, String> additionalArgs, SubstitutableValues substitutableValues)
            throws IOException;

    @Override
    public RedactableObject doSubstitution(String type, List<String> modifiers, String identifier,
            SubstitutableValues substitutableValues) throws IOException {
        // etc...
    }

    // etc...
}

package org.apache.kafka.common.security.substitutions;

/**
 * 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 for an instance that will be redacted only if the given object is
     * of type {@link Password}.
     *
     * @param object
     *            the mandatory object
     */
    public RedactableObject(Object object) {
        this(Objects.requireNonNull(object), object instanceof Password);
    }
 
    /**
     * 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;
        if (object instanceof Password)
            return ((Password) object).value();
        return object.toString();
    }

    /**
     * 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();
    }

    // etc...
}