package org.apache.kafka.common.security.oauthbearer;
 
/**
 * Exception thrown when there is a problem with the configuration (an invalid
 * option in a JAAS config, for example).
 */
public class OAuthBearerConfigException extends OAuthBearerException {
    // etc...
}

package org.apache.kafka.common.security.oauthbearer;
 
/**
 * Exception thrown when token validation fails due to a problem with the token
 * itself (as opposed to a missing remote resource or a configuration problem)
 */
public class OAuthBearerIllegalTokenException extends OAuthBearerException {
    /**
     * Constructor
     *
     * @param reason
     *            the mandatory reason for the validation failure; it must indicate
     *            failure
     */
    public OAuthBearerIllegalTokenException(OAuthBearerValidationResult reason) {
        super(Objects.requireNonNull(reason).failureDescription());
        if (reason.success())
            throw new IllegalArgumentException("The reason indicates success; it must instead indicate failure");
        this.reason = reason;
    }

    /**
     * Return the (always non-null) reason for the validation failure
     *
     * @return the reason for the validation failure
     */
    public OAuthBearerValidationResult reason() {
        return reason;
    }
}

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

/**
 * The result of some kind of token validation
 */
public class OAuthBearerValidationResult implements Serializable {
 
    // etc...

    /**
     * Return an instance indicating success
     *
     * @return an instance indicating success
     */
    public static OAuthBearerValidationResult newSuccess() {
        return new OAuthBearerValidationResult(true, null, null, null);
    }

    /**
     * Return a new validation failure instance
     *
     * @param failureDescription
     *            optional description of the failure
     * @return a new validation failure instance
     */
    public static OAuthBearerValidationResult newFailure(String failureDescription) {
        return newFailure(failureDescription, null, null);
    }

    /**
     * Return a new validation failure instance
     *
     * @param failureDescription
     *            optional description of the failure
     * @param failureScope
     *            optional scope to be reported with the failure
     * @param failureOpenIdConfig
     *         Section 2.1</a>
  optional OpenID Connect configuration*/
 to be reported with theString value();

     /**
     * The token's scope of access, as failureper
     * @return a new validation failure instance<a href="https://tools.ietf.org/html/rfc6749#section-1.4">RFC 6749 Section
     * 1.4</a>
    public *
 static OAuthBearerValidationResult newFailure(String failureDescription, String* failureScope,
@return the token's (always non-null but potentially empty) scope of access,
  String failureOpenIdConfig) {
 *       return new OAuthBearerValidationResult(false, failureDescription, failureScope, failureOpenIdConfig);as per <a href="https://tools.ietf.org/html/rfc6749#section-1.4">RFC
    }
 
 *    private OAuthBearerValidationResult(boolean success, String failureDescription, String6749 failureScope,
            String failureOpenIdConfig) {
 Section 1.4</a>. Note that all values in the returned set will
     *       // etc...
    }
    /**  be trimmed of preceding and trailing whitespace, and the result will
     * Return true if this instance indicates success, otherwise false
never contain the empty  *string.
     */
 @return true if this instance indicates success, otherwise falseSet<String> scope();

    /**
     */
 The token's lifetime, publicexpressed booleanas success() {
    the number of milliseconds since the
    return success;
* epoch, as per }

    /**<a href="https://tools.ietf.org/html/rfc6749#section-1.4">RFC
     * Return6749 the (potentially null) descriptive message for the failureSection 1.4</a>
     *
     *
 @return the token'slifetime, expressed *as @return the (potentiallynumber null)of descriptivemilliseconds messagesince
 for the failure
  *   */
    public String failureDescription() {
  the epoch, as per
     * return failureDescription;
    }

   <a /**href="https://tools.ietf.org/html/rfc6749#section-1.4">RFC 6749
     * Return the (potentially null) scope to be reported with the failureSection 1.4</a>.
     */
     * @return the (potentially null) scope to be reported with the failure
     */
    public String failureScope() {long lifetimeMs();

    /**
     * The name of the principal to which this credential applies
     *
   return failureScope;
 * @return the }

    /**always non-null/non-empty principal name
     */
 Return   theString principalName(potentially null);

 OpenID Connect configuration to be reported /**
     * withWhen the failure
     *credential became valid, in terms of the number of milliseconds
     * @returnsince the (potentiallyepoch, if known, otherwise null). OpenIDAn Connectexpiring configurationcredential tomay be reportednot
     * necessarily indicate when it was created -- just withwhen theit failure
expires -- so   */we
    public String* failureOpenIdConfig() {
        return failureOpenIdConfig;
    }
 
need to support a null return value here.
     /**
     * Raise@return anthe exceptiontime ifwhen thisthe instancecredential indicatesbecame failurevalid, otherwisein doterms nothing
of the number   *of
     * @throws OAuthBearerIllegalTokenException
     *  milliseconds since          if this instance indicates failurethe epoch, if known, otherwise null
     */
    publicLong void throwExceptionIfFailed() throws OAuthBearerIllegalTokenException {
        if (!success())
            throw new OAuthBearerIllegalTokenException(this);
    }
}

package org.apache.kafka.common.security.oauthbearer.refresh;
 
/**
 * A credential that expires and that can potentially be refreshed
 *
 * @see ExpiringCredentialRefreshingLogin
 */
public interface ExpiringCredential {
    /**
     * The name of the principal to which this credential applies (used only for
     * logging)
     *
     * @return the always non-null/non-empty principal name
     */
    String principalName();
    /**
     * When the credential became valid, in terms of the number of milliseconds
     * since the epoch, if known otherwise null
     *
     * @return the time when the credential became valid, in terms of the number of
     *         milliseconds since the epoch, if known otherwise null
     */
    Long startTimeMillis();
 
    /**
     * When the credential expires, in terms of the number of milliseconds since the
     * epoch
     *
     * @return the time when the credential expires, in terms of the number of
     *         milliseconds since the epoch
     */
    long expireTimeMillis();
 
    /**
     * The point after which the credential can no longer be refreshed, in terms of
     * the number of milliseconds since the epoch, if any, otherwise null.
     *
     * @return the point after which the credential can no longer be refreshed, in
     *         terms of the number of milliseconds since the epoch, if any,
     *         otherwise null
     */
    Long absoluteLastRefreshMillis();
}

;

/**
 * The {@code LoginModule} for the SASL/OAUTHBEARER mechanism. When a client
 * (whether a non-broker client or a broker when SASL/OAUTHBEARER is the
 * inter-broker protocol) connects to Kafka the {@code OAuthBearerLoginModule}
 * instance asks its configured {@link AuthenticateCallbackHandler}
 * implementation to handle an instance of {@link OAuthBearerTokenCallback} and
 * return an instance of {@link OAuthBearerToken}. A default, builtin
 * {@link AuthenticateCallbackHandler} implementation creates an unsecured token
 * as defined by these JAAS module options:
 * <table>
 * <tr>
 * <th>JAAS Module Option for Unsecured Token Retrieval</th>
 * <th>Documentation</th>
 * </tr>
 * <tr>
 * <td>{@code unsecuredLoginStringClaim_<claimname>="value"}</td>
 * <td>Creates a {@code String} claim with the given name and value. Any valid
 * claim name can be specified except '{@code iat}' and '{@code exp}' (these are
 * automatically generated).</td>
 * </tr>
 * <tr>
 * <td>{@code unsecuredLoginNumberClaim_<claimname>="value"}</td>
 * <td>Creates a {@code Number} claim with the given name and value. Any valid
 * claim name can be specified except '{@code iat}' and '{@code exp}' (these are
 * automatically generated).</td>
 * </tr>
 * <tr>
 * <td>{@code unsecuredLoginListClaim_<claimname>="value"}</td>
 * <td>Creates a {@code String List} claim with the given name and values parsed
 * from the given value where the first character is taken as the delimiter. For
 * example: {@code unsecuredLoginListClaim_fubar="|value1|value2"}. Any valid
 * claim name can be specified except '{@code iat}' and '{@code exp}' (these are
 * automatically generated).</td>
 * </tr>
 * <tr>
 * <td>{@code unsecuredLoginPrincipalClaimName}</td>
 * <td>Set to a custom claim name if you wish the name of the String claim
 * holding the principal name to be something other than '{@code sub}'.</td>
 * </tr>
 * <tr>
 * <td>{@code unsecuredLoginLifetimeSeconds}</td>
 * <td>Set to an integer value if the token expiration is to be set to something
 * other than the default value of 3600 seconds (which is 1 hour). The
 * '{@code exp}' claim will be set to reflect the expiration time.</td>
 * </tr>
 * <tr>
 * <td>{@code unsecuredLoginScopeClaimName}</td>
 * <td>Set to a custom claim name if you wish the name of the String or String
 * List claim holding any token scope to be something other than
 * '{@code scope}'.</td>
 * </tr>
 * </table>
 * Production use cases will require writing an implementation of
 * {@link AuthenticateCallbackHandler} that can handle an instance of
 * {@link OAuthBearerTokenCallback} and declaring it via either the
 * {@code sasl.login.callback.handler.class} configuration option for a
 * non-broker client or via the
 * {@code listener.name.sasl_ssl.oauthbearer.sasl.login.callback.handler.class}
 * configuration option for brokers (when SASL/OAUTHBEARER is the inter-broker
 * protocol).
 * <p>
 * Here is a typical, basic JAAS configuration for a client leveraging unsecured
 * SASL/OAUTHBEARER authentication:
 *
 * <pre>
 * KafkaClient {
 *      org.apache.kafka.common.security.oauthbearer.

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

/**
 * Individual refresh-related configuration properties defining how
 * {@link ExpiringCredentialRefreshingLogin} refreshes instances ofOAuthBearerLoginModule Required
 *      unsecuredLoginStringClaim_sub="thePrincipalName";
 * };
 * </pre>
 *
 * An implementation of the {@link Login} interface specific to the
 * {@link@code ExpiringCredentialOAUTHBEARER}. Eachmechanism valueis hasautomatically aapplied; type, optional min/max/defaultit periodically
 * values,refreshes anany orderedtoken listbefore ofit stringexpires keysso (containingthat the enumclient can namecontinue andto themake
 * camelCaseconnections versionto ofbrokers. theThe name,parameters in that order)impact sohow itthe can be picked out of arefresh algorithm
 * mapoperates are keyedspecified withas Strings,part andof the producer/consumer/broker configuration
 * are as follows. See the documentation for these properties elsewhere for
 * details.
 * <table>
 * <tr>
 * <th>Producer/Consumer/Broker Configuration Property</th>
 * </tr>
 * <tr>
 * <td>{@code sasl.login.refresh.window.factor}</td>
 * </tr>
 * <tr>
 * <td>{@code sasl.login.refresh.window.jitter}</td>
 * </tr>
 * <tr>
 * <td>{@code sasl.login.refresh.min.period.seconds}</td>
 * </tr>
 * <tr>
 * <td>{@code sasl.login.refresh.min.buffer.seconds}</td>
 * </tr>
 * </table>
 * When a broker accepts a SASL/OAUTHBEARER connection the instance of the
 * builtin {@code SaslServer} implementation asks its configured
 * {@link AuthenticateCallbackHandler} implementation to handle an instance of
 * {@link OAuthBearerValidatorCallback} constructed with the OAuth 2 Bearer
 * Token's compact serialization and return an instance of
 * {@link OAuthBearerToken} if the value validates. A default, builtin
 * {@link AuthenticateCallbackHandler} implementation validates an unsecured
 * token as defined by these JAAS module options:
 * <table>
 * <tr>
 * <th>JAAS Module Option for Unsecured Token Validation</th>
 * <th>Documentation</th>
 * </tr>
 * <tr>
 * <td>{@code unsecuredValidatorPrincipalClaimName="value"}</td>
 * <td>Set to a non-empty value if you wish a particular String claim holding a
 * principal name to be checked for existence; the default is to check for the
 * existence of the '{@code sub}' claim.</td>
 * </tr>
 * <tr>
 * <td>{@code unsecuredValidatorScopeClaimName="value"}</td>
 * <td>Set to a custom claim name if you wish the name of the String or String
 * List claim holding any token scope to be something other than
 * '{@code scope}'.</td>
 * </tr>
 * <tr>
 * <td>{@code unsecuredValidatorRequiredScope="value"}</td>
 * <td>Set to a space-delimited list of scope values if you wish the
 * String/String List claim holding the token scope to be checked to make sure
 * it contains certain values.</td>
 * </tr>
 * <tr>
 * <td>{@code unsecuredValidatorAllowableClockSkewMs="value"}</td>
 * <td>Set to a positive integer value if you wish to allow up to some number of
 * positive milliseconds of clock skew (the default is 0).</td>
 * </tr>
 * </table>
 * Here is a typical, basic JAAS configuration for a broker leveraging unsecured
 * SASL/OAUTHBEARER validation:
 *
 * <pre>
 * KafkaServer {
 *      org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule Required
 *      unsecuredLoginStringClaim_sub="thePrincipalName";
 * };
 * </pre>
 *
 * Production use cases will require writing an implementation of
 * {@link AuthenticateCallbackHandler} that can handle an instance of
 * {@link OAuthBearerValidatorCallback} and declaring it via the
 * {@code listener.name.sasl_ssl.oauthbearer.sasl.server.callback.handler.class}
 * configuration option.
 *
 * @see SaslConfigs#SASL_LOGIN_REFRESH_WINDOW_FACTOR_DOC
 * @see SaslConfigs#SASL_LOGIN_REFRESH_WINDOW_JITTER_DOC
 * @see SaslConfigs#SASL_LOGIN_REFRESH_MIN_PERIOD_SECONDS_DOC
 * @see SaslConfigs#SASL_LOGIN_REFRESH_MIN_BUFFER_SECONDS_DOC
 */
public class OAuthBearerLoginModule implements LoginModule {
    static {
        OAuthBearerSaslClientProvider.initialize(); // not part of public API
        OAuthBearerSaslServerProvider.initialize(); // not part of public API
    }

    // etc...
}

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

/**
 * A JSON Web Token, which may either be a JWS or a JWE. Neither signature
 * validation nor decryption are performed here.
 *
 * @see <a href="https://tools.ietf.org/html/rfc7519">RFC 7519</a>
 */
public abstract class OAuthBearerJwt implements OAuthBearerToken {
    /**
     * Constructor with the principal taken from the '{@code sub}' claim (if any)
     * and no scope claim name
     *
     * @param compactSerialization
     *            the compact serialization to parse as a JWT
     * @throws OAuthBearerIllegalTokenException
     *             if compactSerialization cannot be either a JWS or JWE (due to
     *             there not being either 3 or 5 dot-separated values; or the JWT
     *             header either not being a valid Base64 URL-encoded value or not
     *             valid JSON after decoding)
     */
    public OAuthBearerJwt(String compactSerialization) throws OAuthBearerIllegalTokenException {
        this(compactSerialization, null);
    }

    /**
     * Constructor with the given principal and scope claim names
     *
     * @param compactSerialization
     *            the compact serialization to parse as a JWT
     * @param principalClaimName
     *            the optional principal claim name
     * @param scopeClaimName
     *            the optional scope claim name
     * @throws OAuthBearerIllegalTokenException
     *             if compactSerialization cannot be either a JWS or JWE (due to
     *             there not being either 3 or 5 dot-separated values; or the JWT
     *             header either not being a valid Base64 URL-encoded value or not
     *             valid JSON after decoding)
     */
    public OAuthBearerJwt(String compactSerialization, String principalClaimName, String scopeClaimName)
            throws OAuthBearerIllegalTokenException {
        // etc...
    }
 
    /**
     * Return the 3 or 5 dot-separated sections of the JWT compact serialization
     * 
     * @return the 3 or 5 dot-separated sections of the JWT compact serialization
     */
    public List<String> splits() {
        return splits;
    }

    /**
     * Return the JOSE Header as a {@code Map}
     *
     * @return the JOSE header
     */
    public Map<String, Object> header() {
        return header;
    }

    /**
     * Return the JWT Claim Set as a {@code Map}
     *
     * @return the (always non-null but possibly empty) claims
     */
    public abstract Map<String, Object> claims();

    /**
     * Return the scope claim name, if any, otherwise null
     *
     * @return the scope claim name, if any, otherwise null
     */
    public String scopeClaimName() {
        return scopeClaimName;
    }

    /**
     * Indicate if the claim exists and is the given type
     *
     * @param claimName
     *            the mandatory JWT claim name
     * @param type
     *            the mandatory type, which should either be String.class,
     *            Number.class, or List.class
     * @return true if the claim exists and is the given type, otherwise false
     */
    public boolean isClaimType(String claimName, Class<?> type) {
        // etc...
    }

    /**
     * Extract a claim of the given type
     *
     * @param claimName
     *            the mandatory JWT claim name
     * @param type
     *            the mandatory type, which must either be String.class,
     *            Number.class, or List.class
     * @return the claim if it exists, otherwise null
     * @throws OAuthBearerIllegalTokenException
     *             if the claim exists but is not the given type
     */
    public <T> T getClaim(String claimName, Class<T> type) throws OAuthBearerIllegalTokenException {
        // etc...
    }

    /**
     * Extract a claim that could be either a String or a String List as a String
     * List (if it was a String it will be returned as a list of size 1).
     *
     * @param claimName
     *            the mandatory JWT claim name
     * @return the claim in the form of a String List, if it exists, otherwise null
     * @throws OAuthBearerIllegalTokenException
     *             if the claim exists but is neither a String nor a List
     */
    public List<String> getListClaimFromStringOrList(String claimName) throws OAuthBearerIllegalTokenException {
        // etc...
    }
 
    /**
     * Extract a claim in its raw form
     *
     * @param claimName
     *            the mandatory JWT claim name
     * @return the raw claim value, if it exists, otherwise null
     */
    public Object getRawClaim(String claimName) {
        return claims().get(Objects.requireNonNull(claimName));
    }

    /**
     * Return the
     * <a href="https://tools.ietf.org/html/rfc7519#section-4.1.3">Audience</a>
     * claim as a String List
     *
     * @return the <a href=
     *         "https://tools.ietf.org/html/rfc7519#section-4.1.3">Audience</a>
     *         claim as a String List if available, otherwise null. An Audience
     *         claim that is a String will be returned as a List of size 1.
     * @throws OAuthBearerIllegalTokenException
     *             if the claim value is the incorrect type
     */
    public List<String> audience() throws OAuthBearerIllegalTokenException {
        return getListClaimFromStringOrList(JwtClaim.AUDIENCE.claimName());
    }

    /**
     * Return the
     * <a href="https://tools.ietf.org/html/rfc7519#section-4.1.4">Expiration
     * Time</a> claim
     *
     * @return the <a href=
     *         "https://tools.ietf.org/html/rfc7519#section-4.1.4">Expiration
     *         Time</a> claim if available, otherwise null
     * @throws OAuthBearerIllegalTokenException
     *             if the claim value is the incorrect type
     */
    public Number expirationTime() throws OAuthBearerIllegalTokenException {
        return getClaim(JwtClaim.EXPIRATION_TIME.claimName(), Number.class);
    }

    /**
     * Return the <a href="https://tools.ietf.org/html/rfc7519#section-4.1.6">Issued
     * At</a> claim
     *
     * @return the
     *         <a href= "https://tools.ietf.org/html/rfc7519#section-4.1.6">Issued
     *         At</a> claim if available, otherwise null
     * @throws OAuthBearerIllegalTokenException
     *             if the claim value is the incorrect type
     */
    public Number issuedAt() throws OAuthBearerIllegalTokenException {
        return getClaim(JwtClaim.ISSUED_AT.claimName(), Number.class);
    }
 
    /**
     * Return the
     * <a href="https://tools.ietf.org/html/rfc7519#section-4.1.1">Issuer</a> claim
     *
     * @return the <a href=
     *         "https://tools.ietf.org/html/rfc7519#section-4.1.1">Issuer</a> claim
     *         if available, otherwise null
     * @throws OAuthBearerIllegalTokenException
     *             if the claim value is the incorrect type
     */
    public String issuer() throws OAuthBearerIllegalTokenException {
        return getClaim(JwtClaim.ISSUER.claimName(), String.class);
    }

    /**
     * Return the <a href="https://tools.ietf.org/html/rfc7519#section-4.1.7">JWT
     * ID</a> claim
     *
     * @return the <a href= "https://tools.ietf.org/html/rfc7519#section-4.1.7">JWT
     *         ID</a> claim if available, otherwise null
     * @throws OAuthBearerIllegalTokenException
     *             if the claim value is the incorrect type
     */
    public String jwtId() throws OAuthBearerIllegalTokenException {
        return getClaim(JwtClaim.JWT_ID.claimName(), String.class);
    }

    /**
     * Return the <a href="https://tools.ietf.org/html/rfc7519#section-4.1.5">Not
     * Before</a> claim
     *
     * @return the <a href= "https://tools.ietf.org/html/rfc7519#section-4.1.5">Not
     *         Before</a> claim if available, otherwise null
     * @throws OAuthBearerIllegalTokenException
     *             if the claim value is the incorrect type
     */
    public Number notBefore() throws OAuthBearerIllegalTokenException {
        return getClaim(JwtClaim.NOT_BEFORE.claimName(), Number.class);
    }

    /**
     * Return the
     * <a href="https://tools.ietf.org/html/rfc7519#section-4.1.2">Subject</a> claim
     *
     * @return the <a href=
     *         "https://tools.ietf.org/html/rfc7519#section-4.1.2">Subject</a> claim
     *         if available, otherwise null
     * @throws OAuthBearerIllegalTokenException
     *             if the claim value is the incorrect type
     */
    public String subject() throws OAuthBearerIllegalTokenException {
        return getClaim(JwtClaim.SUBJECT.claimName(), String.class);
    }
 
    /**
     * Decode the given Base64URL-encoded value, parse the resulting JSON as a JSON
     * object, and return the map of member names to their values (each value being
     * represented as either a String, a Number, or a List of Strings).
     *
     * @param split
     *            the value to decode and parse
     * @return the map of JSON member names to their String, Number, or String List
     *         value
     * @throws OAuthBearerIllegalTokenException
     *             if the given Base64URL-encoded value cannot be decoded or parsed
     */
    public static Map<String, Object> toMap(String split) throws OAuthBearerIllegalTokenException {
        // etc...
    }
}

package org.apache.kafka.common.security.oauthbearer;
 
/**
 * @see <a href="https://tools.ietf.org/html/rfc7515">RFC 7515</a>. Note that
 *      digital signature validation is not performed.
 */
public class OAuthBearerJws extends OAuthBearerJwt {
    /**
     * Constructor with no scope claim name
     *
     * @param compactSerialization
     *            the compact serialization to parse as a JWT
     * @throws OAuthBearerIllegalTokenException
     *             if the compact serialization is not a valid JWS (meaning it did
     *             not have 3 dot-separated Base64URL sections; or the header or
     *             claims either are not valid Base 64 URL encoded values or are not
     *             JSON after decoding; or the existence or absence of a digital
     *             signature is inconsistent with the mandatory 'alg' header value)
     */
    public OAuthBearerJws(String compactSerialization) throws OAuthBearerIllegalTokenException {
        this(compactSerialization, JwtClaim.SUBJECT.claimName(), null);
    }

    /**
     * Constructor with the given principal and scope claim names
     *
     * @param compactSerialization
     *            the compact serialization to parse as a JWS
     * @param principalClaimName
     *            the optional principal claim name
     * @param scopeClaimName
     *            the optional scope claim name
     * @throws OAuthBearerIllegalTokenException
     *             if compactSerialization cannot be either a JWS or JWE (due to
     *             there not being either 3 or 5 dot-separated values; or the JWT
     *             header either not being a valid Base64 URL-encoded value or not
     *             valid JSON after decoding)
     */
    public OAuthBearerJws(String compactSerialization, String principalClaimName, String scopeClaimName)
            throws OAuthBearerIllegalTokenException {
        // etc...
    }
}

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

/**
 * JSON Web Signature and Encryption Header Parameters
 *
 * @see <a href="https://www.iana.org/assignments/jose/jose.xhtml">JSON Web
 *      Signature and Encryption Header Parameters</a>
 */
public enum JwtHeaderParameter {
    ALGORITHM("alg"),
    JWK_SET_URL("jku"),
    JSON_WEB_KEY("jwk"),
    KEY_ID("kid"),
    X_509_URL("x5u"),
    X_509_CERTIFICATE_CHAIN("x5c"),
    X_509_CERTIFICATE_SHA1_THUMBPRINT("x5t"),
    X_509_CERTIFICATE_SHA256_THUMBPRINT("x5t#S256"),
    TYPE("typ"),
    CONTENT_TYPE("cty"),
    CRITICAL("crit"),
    ENCRYPTION_ALGORITHM("enc"),
    COMPRESSION_ALGORITHM("zip"),
    ISSUER("iss"),
    SUBJECT("sub"),
    AUDIENCE("aud");

    private String headerParameterName;

    private JwtHeaderParameter(String headerParameterName) {
        this.headerParameterName = headerParameterName;
    }

    String headerParameterName() {
        return headerParameterName;
    }
}

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

/**
 * JSON Web Token Claims
 *
 * @see <a href="https://www.iana.org/assignments/jwt/jwt.xhtml">JSON Web Token
 *      Claims</a>
 */
public enum JwtClaim {
    ISSUER("iss"),
    SUBJECT("sub"),
    AUDIENCE("aud"),
    EXPIRATION_TIME("exp"),
    NOT_BEFORE("nbf"),
    ISSUED_AT("iat"),
    JWT_ID("jti");

    private String claimName;

    private JwtClaim(String claimName) {
        this.claimName = claimName;
    }

    public String claimName() {
        return claimName;
    }
}

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

/**
 * Holds state regarding which {@code LoginModule} <a href=
 * "https://docs.oracle.com/javase/9/docs/api/javax/security/auth/login/Configuration.html">module
 * options</a> have been substituted. Instances of this class are thread-safe.
 *
 * @see SubstitutableModuleOptionsCallbackHandler
 * @see SubstitutableModuleOptionsCallback
 */
public class SubstitutableModuleOptions {
    /**
     * Constructor
     *
     * @param moduleOptionsMap
     *            the mandatory map representation of the <a href=
     *            "https://docs.oracle.com/javase/9/docs/api/javax/security/auth/login/Configuration.html">module
     *            options</a>
     */
    public SubstitutableModuleOptions(Map<String, String> moduleOptionsMap) {
        this.moduleOptionsMap = Collections.unmodifiableMap(Objects.requireNonNull(moduleOptionsMap));
    }

    /**
     * Return (an unmodifiable copy of) the original module options map provided
     * during construction
     *
     * @return (an unmodifiable copy of) the original module options map provided
     *         during construction
     */
    public Map<String, String> moduleOptionsMap() {
        return moduleOptionsMap;
    }
 
    /**
     * Return an unmodifiable map identifying which module options have been
     * processed for substitution and the resulting value after substitution (if
     * any) was applied. A module option is guaranteed to have been processed for
     * substitution and its name will appear as a key in the returned map only after
     * {@link #setSubstitutionValue(String, String)} or
     * {@link #setSubstitutionValue(String, Password)} is called.
     *
     * @return an unmodifiable map identifying which module options have been
     *         processed for substitution and the resulting value after substitution
     *         (if any) was applied
     */
    public Map<String, Object> moduleOptionSubstitutionState() {
        return Collections.unmodifiableMap(moduleOptionSubstitutionState);
    }
 
    /**
     * Identify that the option with the given name has had substitution performed
     * for it resulting in the given non-password value. This method can be
     * successfully invoked (and is idempotent) only if invoked before
     * {@link #setSubstitutionValue(String, Password)}; the option value cannot be
     * changed.
     *
     * @param optionName
     *            the mandatory option name, which must exist in the map returned by
     *            {@link #moduleOptionsMap()}
     * @param substitutionValue
     *            the mandatory substitution value to set
     */
    public void setSubstitutionValue(String optionName, String substitutionValue) {
        setSubstitutionValueInternal(Objects.requireNonNull(optionName), Objects.requireNonNull(substitutionValue));
    }
 
    /**
     * Identify that the option with the given name has had substitution performed
     * for it resulting in the given password-related value. This method can be
     * successfully invoked (and is idempotent) only if invoked before
     * {@link #setSubstitutionValue(String, String)}; the option value cannot be
     * changed.
     *
     * @param optionName
     *            the mandatory option name, which must exist in the map returned by
     *            {@link #moduleOptionsMap()}
     * @param substitutionValue
     *            the mandatory substitution value to set
     */
    public void setSubstitutionValue(String optionName, Password substitutionValue) {
        setSubstitutionValueInternal(Objects.requireNonNull(optionName), Objects.requireNonNull(substitutionValue));
    }
 
    // etc..
}

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

/**
 * A {@code Callback} related to introspection requests against a JAAS
 * configuration
 *
 * @see SubstitutableModuleOptionsCallbackHandler
 * @see SubstitutableModuleOptions
 */
public class SubstitutableModuleOptionsCallback implements Callback {
    /**
     * Constructor
     * 
     * @param substitutableModuleOptions
     *            the mandatory substitutable module options
     * @param optionName
     *            the requested option
     * @param optionRequiredToExist
     *            if true then the requested option is required to exist
     */
    public SubstitutableModuleOptionsCallback(SubstitutableModuleOptions substitutableModuleOptions, String optionName,
            boolean optionRequiredToExist) {
        // etc...
    }
 
    /**
     * Return the substitutable module options provided at construction time
     *
     * @return the substitutable module options provided at construction time
     */
    public SubstitutableModuleOptions substitutableModuleOptions() {
        return substitutableModuleOptions;
    }

    /**
     * Return the option name provided at construction time
     *
     * @return the option name provided at construction time
     */
    public String optionName() {
        return optionName;
    }
 
    /**

     * Return true if the the requested option is required to exist, otherwise false
     * 
     * @return true if the the requested option is required to exist, otherwise
     *         false
     */
    public boolean optionRequiredToExist() {
        return optionRequiredToExist;
    }
 
    /**
     * Identify that the option identified by {@link #optionName()}, on the instance
     * returned by {@link #substitutableModuleOptions()}, has had substitution
     * performed for it resulting in the given non-password value. This method can
     * be successfully invoked (and is idempotent) only when the option exists and
     * its value isn't being changed.
     *
     * @param substitutionValue
     *            the mandatory non-password substitution value to set
     */
    public void setSubstitutionValue(String substitutionValue) {
        substitutableModuleOptions.setSubstitutionValue(optionName, Objects.requireNonNull(substitutionValue));
    }
 
    /**
     * Identify that the option identified 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 exists
     * and its value isn't being changed.
     *
     * @param substitutionValue
     *            the mandatory password-related substitution value to set
     */
    public void setSubstitutionValue(Password substitutionValue) {
        substitutableModuleOptions.setSubstitutionValue(optionName, Objects.requireNonNull(substitutionValue));
    }
 
    /**
     * Return the substitution value, if any has been set, otherwise null. Any
     * non-null value will be either a {@code String} or a {@code Password}. Note
     * that the value may not have been set via a call to
     * {@link #setSubstitutionValue(Password)} or
     * {@link #setSubstitutionValue(String)}; it is possible that the value was
     * already set prior to construction of this instance, in which case that value
     * will be returned here.
     *
     * @return the substitution value, if any has been set, otherwise null
     */
    public Object substitutionValue() {
        return substitutableModuleOptions.moduleOptionSubstitutionState().get(optionName);
    }

    /**
     * Return the substitution text, if any has been set, otherwise null.
     * 
     * @return the substitution value, if any has been set, otherwise null.
     */
    public String substitutionText() {
        Object substitutionValue = substitutionValue();
        return substitutionValue instanceof Password ? ((Password) substitutionValue).value()
                : (String) substitutionValue;
    }
}

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

/**
 * A {@code CallbackHandler} that handles introspection requests against a JAAS
 * configuration
 *
 * @see SubstitutableModuleOptionsCallback
 * @see SubstitutableModuleOptions
 */
public class SubstitutableModuleOptionsCallbackHandler implements AuthenticateCallbackHandler {
    @Override
    public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
        for (Callback callback : callbacks) {
            if (callback instanceof SubstitutableModuleOptionsCallback) {
                SubstitutableModuleOptionsCallback substitutableModuleOptionsCallback = (SubstitutableModuleOptionsCallback) callback;
                SubstitutableModuleOptions substitutableModuleOptions = substitutableModuleOptionsCallback
                        .substitutableModuleOptions();
                String optionName = substitutableModuleOptionsCallback.optionName();
                boolean optionRequiredToExist = substitutableModuleOptionsCallback.optionRequiredToExist();
                boolean optionExists = substitutableModuleOptions.moduleOptionsMap().containsKey(optionName);
                if (!optionExists) {
                    if (!optionRequiredToExist)
                        continue;
                    else
                        throw new IOException(String
                                .format("The requested option was required to exist but does not: %s", optionName));
                }
                Object substitutionValue = getSubstitutionValue(substitutableModuleOptions, optionName,
                        new HashSet<String>());
                if (substitutionValue instanceof String)
                    substitutableModuleOptionsCallback.setSubstitutionValue((String) substitutionValue);
                else
                    substitutableModuleOptionsCallback.setSubstitutionValue((Password) substitutionValue);
            } 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
     */
    public Map<String, ?> serverConfig() {
        return serverConfig;
    }

    /**
     * Return the SASL mechanism provided during
     * {@link #configure(Map, String, List)}, if any, otherwise null
     *
     * @return the SASL mechanism provided during
     *         {@link #configure(Map, String, List)}, if any, otherwise null
     */
    public String mechanism() {
        return mechanism;
    }

    /**
     * Return the JAAS login module configurations provided during
     * {@link #configure(Map, String, List)}, if any, otherwise null
     *
     * @return 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 requested option
     * @param optionRequiredToExist
     *            if true, and the requested option does not exist, then an
     *            exception is raised; if false, and the requested option does not
     *            exist, then the empty string is returned.
     * @return the given option value, after any required substitution is applied,
     *         or the empty string if the option does not exist and it was not
     *         required to exist
     * @throws IOException
     *             if a required substitution cannot be performed (including if a
     *             required option does not exist)
     */
    public static Object getSubstitutionValue(SubstitutableModuleOptions substitutableModuleOptions, String optionName,
            boolean optionRequiredToExist) throws IOException {
        boolean optionExists = substitutableModuleOptions.moduleOptionsMap().containsKey(optionName);
        if (!optionExists) {
            if (!optionRequiredToExist)
                return "";
            else
                throw new IOException(
                        String.format("The requested option was required to exist but does not: %s", optionName));
        }
        return getSubstitutionValue(substitutableModuleOptions, optionName, new HashSet<String>());
    }
 
    /*
     * Handle substitution, dealing with circular references, constraints, etc.
     */
    private static Object getSubstitutionValue(SubstitutableModuleOptions substitutableModuleOptions, String optionName,
            HashSet<String> optionsSeen) throws IOException {
        // etc...
    }
 
    // etc...
}

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

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

import org.apache.kafka.common.security.oauthbearer.internal.OAuthBearerSaslClientProvider;
import org.apache.kafka.common.security.oauthbearer.internal.OAuthBearerSaslServerProvider;
/**
 * A {@code LoginModule} for the SASL/OAUTHBEARER mechanism.
 * <p>
 * Example use on a non-broker client:
 *
 * <pre>
 * KafkaClient {
 *      org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule Required
 *      reloginAllowedBeforeLogout="true"
 *      tokenRetriever="org.apache.kafka.common.security.oauthbearer.OAuthBearerUnsecuredJwtRetriever"
 *          stringClaim_sub="theClientPrincipalName"
 *          listClaim_scope="|scopeValue1|scopeValue2";
 * };
 * </pre>
 *
 * Example use on a broker:
 *
 * <pre>
 * KafkaServer {
 *      org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule Required
 *      reloginAllowedBeforeLogout="true"
 *      tokenRetriever="org.apache.kafka.common.security.oauthbearer.OAuthBearerUnsecuredJwtRetriever"
 *          stringClaim_sub="theBrokerPrincipalName"
 *          listClaim_scope="|scopeValue1|scopeValue2"
 *      tokenValidator="org.apache.kafka.common.security.oauthbearer.OAuthBearerUnsecuredJwtValidator"
 *          principalClaimName="sub";
 * };
 * </pre>
 *
 * @see RefreshConfigProp#RELOGIN_ALLOWED_BEFORE_LOGOUT
 * @see OAuthBearerTokenRetriever
 * @see OAuthBearerTokenValidator
 */
public class OAuthBearerLoginModule implements ExpiringCredentialLoginModule {
    public static final String TOKEN_RETRIEVER_CLASS_NAME_OPTION = "tokenRetriever";
    public static final String TOKEN_VALIDATOR_CLASS_NAME_OPTION = "tokenValidator";
    private static final Set<String> MECHANISMS = Collections
            .unmodifiableSet(new HashSet<>(Arrays.asList("OAUTHBEARER")));
 
    static {
        OAuthBearerSaslClientProvider.initialize(); // not part of public API
        OAuthBearerSaslServerProvider.initialize(); // not part of public API
    }
 
    @Override
    public Set<String> mechanisms() {
        return MECHANISMS;
    }
    @Override
    public CallbackHandler newCallbackHandler() {
        return new SubstitutableModuleOptionsCallbackHandler();
    }
 
    // etc...

}

package org.apache.kafka.common.security.oauthbearer;
/**
 * An implementation of this interface must be configured for broker and
 * non-broker clients via JAAS when using the {@link OAuthBearerLoginModule}.
 * The configuration is done via the
 * {@value OAuthBearerLoginModule#TOKEN_RETRIEVER_CLASS_NAME_OPTION} option.
 */
public interface OAuthBearerTokenRetriever {
    /**
     * Retrieve a token using the given callback handler and JAAS login module
     * options.
     *
     * @param callbackHandler
     *            the mandatory callback handler. It will typically be capable of
     *            handling instances of {@link SubstitutableModuleOptionsCallback},
     *            though different implementations of this interface are free to
     *            differ in their requirements.
     * @param moduleOptionsMap
     *            the mandatory map representation of the <a href=
     *            "https://docs.oracle.com/javase/9/docs/api/javax/security/auth/login/Configuration.html">module
     *            options</a>
     * @return the result of the attempt. It will be an instance of
     *         {@link OAuthBearerToken} if the token is successfully retrieved;
     *         otherwise it will be a String describing why the token could not be
     *         retrieved (which will be a different reason than an unavailable
     *         remote resource or a configuration problem -- these issues result in
     *         exceptions as described below).
     * @throws IOException
     *             if one or more networked resources required to perform the
     *             retrieval (e.g. a web service) is unavailable.
     * @throws UnsupportedCallbackException
     *             if the provided {@code CallbackHandler} cannot handle
     *             {@link SubstitutableModuleOptionsCallback}.
     * @throws OAuthBearerConfigException
     *             if there is a configuration problem that prevents this instance
     *             from functioning (a missing mandatory parameter, for example)
     * @throws LoginException
     *             if the retrieval fails for any other reason (if the token
     *             endpoint rejects the provided credentials, for example)
     */
    OAuthBearerToken retrieve(CallbackHandler callbackHandler, Map<String, String> moduleOptionsMap)
            throws IOException, UnsupportedCallbackException, OAuthBearerConfigException, LoginException;
}

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

/**
 * An implementation of {@link OAuthBearerTokenRetriever} that generates an
 * unsecured JWT. Claims and their values can be specified using
 * stringClaim_<claimname>, numberClaim_<claimname>, and
 * listClaim_<claimname> options. The first character of the value is
 * taken as the delimiter for list claims. You may define any claim name and
 * value except '{@code iat}' and '{@code exp}', both of which are calculated
 * automatically.
 * <p>
 * This implementation also accepts the following options:
 * <ul>
 * <li><code>{@value #PRINCIPAL_CLAIM_NAME_OPTION}</code> set to a custom claim
 * name if you wish the name of the String claim holding the principal name to
 * be something other than <code>"sub"</code>.</li>
 * <li><code>{@value #LIFETIME_SECONDS_OPTION}</code> set to an integer value if
 * the token expiration is to be set to something other than the default value
 * of 3600 seconds (which is 1 hour). The 'exp' claim reflects the expiration
 * time.</li>
 * <li><code>{@value #SCOPE_CLAIM_NAME_OPTION}</code> set to a custom claim name
 * if you wish the name of the String or String List claim holding any token
 * scope to be something other than "<code>scope</code>"</li>
 * </ul>
 * For example:
 *
 * <pre>
 * KafkaClient {
 *      org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule Required
 *      tokenRetriever="org.apache.kafka.common.security.oauthbearer.OAuthBearerUnsecuredJwtRetriever"
 *      reloginAllowedBeforeLogout="true"
 *         stringClaim_sub="thePrincipalName"
 *         listClaim_scope="|scopeValue1|scopeValue2"
 *         lifetimeSeconds="60";
 * };
 *
 * </pre>
 */
public class OAuthBearerUnsecuredJwtRetriever implements OAuthBearerTokenRetriever {
    public static final String PRINCIPAL_CLAIM_NAME_OPTION = "principalClaimName";
    public static final String LIFETIME_SECONDS_OPTION = "lifetimeSeconds";
    public static final String SCOPE_CLAIM_NAME_OPTION = "scopeClaimName";
 
    // etc...
 
}

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

import javax.security.sasl.SaslServer;

/**
 * The public interface of SaslServer implementations of SASL/OAUTHBEARER for
 * Kafka. The validated access token will be available via the {@link #token()}
 * method. The existence of this interface keeps the actual implementation out
 * of the public API.
 */
public interface OAuthBearerSaslServer extends SaslServer {
    /**
     * Return the validated access token. Exposing the token here is necessary if we
     * wish to make it available for authorization decisions.
     *
     * @return the validated access token
     * @see <a href=
     *      "https://cwiki.apache.org/confluence/display/KAFKA/KIP-189%3A+Improve+principal+builder+interface+and+add+support+for+SASL">KIP-189:
     *      Improve principal builder interface and add support for SASL</a>
     */
    public OAuthBearerToken token();
}

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

/**
 * An implementation of this interface must be configured for brokers via JAAS
 * when using the {@link OAuthBearerLoginModule}. The configuration is done via
 * the {@value OAuthBearerLoginModule#TOKEN_VALIDATOR_CLASS_NAME_OPTION} option.
 */
public interface OAuthBearerTokenValidator {
    /**
     * Validate a token using the given callback handler and JAAS login module
     * options.
     *
     * @param tokenValue
     *            the <code>b64token</code> value as defined in
     *            <a href="https://tools.ietf.org/html/rfc6750#section-2.1">RFC 6750
     *            Section 2.1</a> to validate
     * @param callbackHandler
     *            the mandatory callback handler. It will typically be capable of
     *            handling instances of {@link SubstitutableModuleOptionsCallback},
     *            though different implementations of this interface are free to
     *            differ in their requirements. Set the callback handler via the
     *            {@code sasl.server.callback.handler.class.map} option in the
     *            broker properties file.
     * @param moduleOptionsMap
     *            the mandatory map representation of the <a href=
     *            "https://docs.oracle.com/javase/9/docs/api/javax/security/auth/login/Configuration.html">module
     *            options</a>
     * @return the result of the attempt. It will be an instance of
     *         {@link OAuthBearerToken} if the token is successfully validated;
     *         otherwise it will be an instance of
     *         {@link OAuthBearerValidationResult} describing why the token could
     *         not be validated (which will be a different reason than an
     *         unavailable remote resource or a configuration problem -- these
     *         issues result in exceptions as described below). Additional keys and
     *         values are ignored by {@link OAuthBearerSaslServerImpl}.
     * @throws IOException
     *             if one or more networked resources required to perform the
     *             validation (e.g. a web service) is unavailable.
     * @throws UnsupportedCallbackException
     *             if the provided {@code CallbackHandler} cannot handle
     *             {@link SubstitutableModuleOptionsCallback}.
     * @throws OAuthBearerConfigException
     *             if there is a configuration problem that prevents this instance
     *             from functioning (a missing mandatory parameter, for example)
     * @throws OAuthBearerIllegalTokenException
     *             if there is a problem with the token itself (it cannot be parsed
     *             or it otherwise fails validation)
     * @see BrokerSecurityConfigs#SASL_SERVER_CALLBACK_HANDLER_CLASS_MAP_DOC
     */
    OAuthBearerToken validate(String tokenValue, CallbackHandler callbackHandler, Map<String, String> moduleOptionsMap)
            throws IOException, UnsupportedCallbackException, OAuthBearerConfigException,
            OAuthBearerIllegalTokenException;
}

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

/**
 * An implementation of {@link OAuthBearerTokenValidator} that validates a JWS
 * without regard to its algorithm and signature. It requires there to be an
 * <code>"exp" (Expiration Time)</code> claim of type Number. If
 * <code>"iat" (Issued At)</code> or <code>"nbf" (Not Before)</code> claims are
 * present each must be a number that precedes the Expiration Time claim, and if
 * both are present the Not Before claim must not precede the Issued At claim.
 * It also accepts the following options, none of which are required:
 * <ul>
 * <li><code>{@value #PRINCIPAL_CLAIM_NAME_OPTION}</code> set to a non-empty
 * value if you wish a String claim holding a principal name to be checked for
 * existence; the default is to perform no such check</li>
 * <li><code>{@value #SCOPE_CLAIM_NAME_OPTION}</code> set to a custom claim name
 * if you wish the name of the String/String List claim holding any token scope
 * to be something other than <code>"scope"</code></li>
 * <li><code>{@value #REQUIRED_SCOPE_OPTION}</code> set to a space-delimited
 * list of scope values if you wish the String/String List claim holding the
 * token scope to be checked to make sure it contains certain values</li>
 * <li><code>{@value #ALLOWABLE_CLOCK_SKEW_MILLIS_OPTION}</code> set to a
 * positive integer value if you wish to allow up to some number of positive
 * milliseconds of clock skew (the default is 0)</li>
 * </ul>
 * For example:
 *
 * <pre>
 * KafkaServer {
 *      org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule Required
 *      reloginAllowedBeforeLogout="true"
 *      tokenRetriever="org.apache.kafka.common.security.oauthbearer.OAuthBearerUnsecuredJwtRetriever"
 *         stringClaim_sub="thePrincipalName"
 *         listClaim_scope=",KAFKA_BROKER,LOGIN_TO_KAFKA"
 *      tokenValidator="org.apache.kafka.common.security.oauthbearer.OAuthBearerUnsecuredJwtValidator"
 *          principalClaimName="sub"
 *          requiredScope="LOGIN_TO_KAFKA"
 *          allowableClockSkewMillis="3000";
 * };
 * </pre>
 */
public class OAuthBearerUnsecuredJwtValidator implements OAuthBearerTokenValidator {
    public static final String PRINCIPAL_CLAIM_NAME_OPTION = "principalClaimName";
    public static final String SCOPE_CLAIM_NAME_OPTION = "scopeClaimName";
    public static final String REQUIRED_SCOPE_OPTION = "requiredScope";
    public static final String ALLOWABLE_CLOCK_SKEW_MILLIS_OPTION = "allowableClockSkewMillis";
 
    // etc...
}

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

public class OAuthBearerValidationUtils {
    /**
     * Validate the given claim for existence and type. It can be required to exist
     * in the given claims, and if it exists it must be one of the types indicated
     *
     * @param jwt
     *            the mandatory JWT to which the validation will be applied
     * @param required
     *            true if the claim is required to exist
     * @param claimName
     *            the required claim name identifying the clam to be checked
     * @param allowedTypes
     *            one or more of {@code String.class}, {@code Number.class}, and
     *            {@code List.class} identifying the type(s) that the claim value is
     *            allowed to be if it exists
     * @return the result of the validation
     */
    public static OAuthBearerValidationResult validateClaimForExistenceAndType(OAuthBearerJwt jwt, boolean required,
            String claimName, Class<?>... allowedTypes) {
        // etc...
    }
 
    /**
     * Validate the 'iat' (Issued At) claim. It can be required to exist in the
     * given claims, and if it exists it must be a (potentially fractional) number
     * of seconds since the epoch defining when the JWT was issued; it is a
     * validation error if the Issued At time is after the time at which the check
     * is being done (plus any allowable clock skew).
     *
     * @param jwt
     *            the mandatory JWT to which the validation will be applied
     * @param required
     *            true if the claim is required to exist
     * @param whenCheckTimeMillis
     *            the time relative to which the validation is to occur
     * @param allowableClockSkewMillis
     *            non-negative number to take into account some potential clock skew
     * @return the result of the validation
     * @throws OAuthBearerConfigException
     *             if the given allowable clock skew is negative
     */
    public static OAuthBearerValidationResult validateIssuedAt(OAuthBearerJwt jwt, boolean required,
            long whenCheckTimeMillis, int allowableClockSkewMillis) throws OAuthBearerConfigException {
        // etc...
    }
 
    /**
     * Validate the 'nbf' (Not Before) claim. It can be required to exist in the
     * given claims, and if it exists it must be a (potentially fractional) number
     * of seconds since the epoch defining the point at which the JWT becomes valid.
     * It is a validation error if the time at which the check is being done (plus
     * any allowable clock skew) is before the Not Before time.
     *
     * @param jwt
     *            the mandatory JWT to which the validation will be applied
     * @param required
     *            true if the claim is required to exist
     * @param whenCheckTimeMillis
     *            the time relative to which the validation is to occur
     * @param allowableClockSkewMillis
     *            non-negative number to take into account some potential clock skew
     * @return the result of the validation
     * @throws OAuthBearerConfigException
     *             if the given allowable clock skew is negative
     */
    public static OAuthBearerValidationResult validateNotBefore(OAuthBearerJwt jwt, boolean required,
            long whenCheckTimeMillis, int allowableClockSkewMillis) throws OAuthBearerConfigException {
        // etc...
    }
 
    /**
     * Validate the 'exp' (Expiration Time) claim. It can be required to exist in
     * the given claims, and if it exists it must be a (potentially fractional)
     * number of seconds defining the point at which the JWT expires. It is a
     * validation error if the time at which the check is being done (minus any
     * allowable clock skew) is on or after the Expiration Time time.
     *
     * @param jwt
     *            the mandatory JWT to which the validation will be applied
     * @param required
     *            true if the claim is required to exist
     * @param whenCheckTimeMillis
     *            the time relative to which the validation is to occur
     * @param allowableClockSkewMillis
     *            non-negative number to take into account some potential clock skew
     * @return the result of the validation
     * @throws OAuthBearerConfigException
     *             if the given allowable clock skew is negative
     */
    public static OAuthBearerValidationResult validateExpirationTime(OAuthBearerJwt jwt, boolean required,
            long whenCheckTimeMillis, int allowableClockSkewMillis) throws OAuthBearerConfigException {
        // etc...
    }
 
    /**
     * Validate the 'iat' (Issued At), 'nbf' (Not Before), and 'exp' (Expiration
     * Time) claims for internal consistency. For claim pairs that exist, the
     * following must be true:
     * <ul>
     * <li>nbf &gt;= iat</li>
     * <li>exp &gt; iat</li>
     * <li>exp &gt; nbf</li>
     * </ul>
     *
     * @param jwt
     *            the mandatory JWT to which the validation will be applied
     * @return the result of the validation
     */
    public static OAuthBearerValidationResult validateTimeConsistency(OAuthBearerJwt jwt) {
        // etc...
    }
 
    /**
     * Validate the given scope against the required scope. Every required scope
     * element (if any) must exist in the provided scope for the validation to
     * succeed.
     *
     * @param scope
     *            the optional scope to validate
     * @param requiredScope
     *            the optional required scope against which the given scope will be
     *            validated
     * @return the result of the validation
     * @throws IOException
     *             if one or more required networked resources (e.g. to re-hydrate
     *             an opaque token) is unavailable.
     * @throws OAuthBearerIllegalTokenException
     *             if there is something fundamentally wrong with the token (if it
     *             is malformed, for example)
     */
    public static OAuthBearerValidationResult validateScope(OAuthBearerToken token, List<String> requiredScope)
            throws IOException, OAuthBearerIllegalTokenException {
        // etc...
    }
 
    // etc...
 
    private OAuthBearerValidationUtils() {
        // empty
    }
}

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

/**
 * Utility class for help dealing with
 * <a href="https://tools.ietf.org/html/rfc6749#section-3.3">Access Token
 * Scopes</a>
 */
public class OAuthBearerScopeUtils {
    private static final Pattern INDIVIDUAL_SCOPE_ITEM_PATTERN = Pattern.compile("[\\x23-\\x5B\\x5D-\\x7E\\x21]+");

    /**
     * Return true if the given value meets the definition of a valid scope item as
     * per <a href="https://tools.ietf.org/html/rfc6749#section-3.3">RFC 6749
     * Section 3.3</a>, otherwise false
     *
     * @param scopeItem
     *            the mandatory scope item to check for validity
     * @return true if the given value meets the definition of a valid scope item,
     *         otherwise false
     */
    public static boolean isValidScopeItem(String scopeItem) {
        return INDIVIDUAL_SCOPE_ITEM_PATTERN.matcher(Objects.requireNonNull(scopeItem)).matches();
    }

    /**
     * Convert a space-delimited list of scope values (for example,
     * <code>"scope1 scope2"</code>) to a List containing the individual elements
     * (<code>"scope1"</code> and <code>"scope2"</code>)
     *
     * @param spaceDelimitedScope
     *            the mandatory (but possibly empty) space-delimited scope values,
     *            each of which must be valid according to
     *            {@link #isValidScopeItem(String)}
     * @return the list of the given (possibly empty) space-delimited values
     * @throws OAuthBearerConfigException
     *             if any of the individual scope values are malformed/illegal
     */
    public static List<String> parseScope(String spaceDelimitedScope) throws OAuthBearerConfigException {
        // etc...
    }


    private OAuthBearerScopeUtils() {
        // empty
    }
}