ListenersValidator { @SuppressWarnings({"checkstyle:BooleanExpressionComplexity", "checkstyle:NPathComplexity", "checkstyle:CyclomaticComplexity"}) private static void validateOauth(Set<String> errors, GenericKafkaListener listener) { if (listener.getAuth() != null && KafkaListenerAuthenticationOAuth.TYPE_OAUTH.equals(listener.getAuth().getType())) { KafkaListenerAuthenticationOAuth oAuth = (KafkaListenerAuthenticationOAuth) listener.getAuth(); String listenerName = listener.getName(); boolean hasJwksRefreshSecondsValidInput = oAuth.getJwksRefreshSeconds() != null && oAuth.getJwksRefreshSeconds() > 0; boolean hasJwksExpirySecondsValidInput = oAuth.getJwksExpirySeconds() != null && oAuth.getJwksExpirySeconds() > 0; boolean hasJwksMinRefreshPauseSecondsValidInput = oAuth.getJwksMinRefreshPauseSeconds() != null && oAuth.getJwksMinRefreshPauseSeconds() >= 0; if (oAuth.getIntrospectionEndpointUri() == null && oAuth.getJwksEndpointUri() == null) { errors.add("listener " + listenerName + ": Introspection endpoint URI or JWKS endpoint URI has to be specified"); } if (oAuth.getValidIssuerUri() == null && oAuth.isCheckIssuer()) { errors.add("listener " + listenerName + ": Valid Issuer URI has to be specified or 'checkIssuer' set to false"); } if (oAuth.getIntrospectionEndpointUri() != null && (oAuth.getClientId() == null || oAuth.getClientSecret() == null)) { errors.add("listener " + listenerName + ": Introspection Endpoint URI needs to be configured together with clientId and clientSecret"); } if (oAuth.getUserInfoEndpointUri() != null && oAuth.getIntrospectionEndpointUri() == null) { errors.add("listener " + listenerName + ": User Info Endpoint URI can only be used if Introspection Endpoint URI is also configured"); } if (oAuth.getJwksEndpointUri() == null && (hasJwksRefreshSecondsValidInput || hasJwksExpirySecondsValidInput || hasJwksMinRefreshPauseSecondsValidInput)) { errors.add("listener " + listenerName + ": jwksRefreshSeconds, jwksExpirySeconds and jwksMinRefreshPauseSeconds can only be used together with jwksEndpointUri"); } if (oAuth.getJwksRefreshSeconds() != null && !hasJwksRefreshSecondsValidInput) { errors.add("listener " + listenerName + ": jwksRefreshSeconds needs to be a positive integer (set to: " + oAuth.getJwksRefreshSeconds() + ")"); } if (oAuth.getJwksExpirySeconds() != null && !hasJwksExpirySecondsValidInput) { errors.add("listener " + listenerName + ": jwksExpirySeconds needs to be a positive integer (set to: " + oAuth.getJwksExpirySeconds() + ")"); } if (oAuth.getJwksMinRefreshPauseSeconds() != null && !hasJwksMinRefreshPauseSecondsValidInput) { errors.add("listener " + listenerName + ": jwksMinRefreshPauseSeconds needs to be a positive integer or zero (set to: " + oAuth.getJwksMinRefreshPauseSeconds() + ")"); } if ((hasJwksExpirySecondsValidInput && hasJwksRefreshSecondsValidInput && oAuth.getJwksExpirySeconds() < oAuth.getJwksRefreshSeconds() + 60) || (!hasJwksExpirySecondsValidInput && hasJwksRefreshSecondsValidInput && KafkaListenerAuthenticationOAuth.DEFAULT_JWKS_EXPIRY_SECONDS < oAuth.getJwksRefreshSeconds() + 60) || (hasJwksExpirySecondsValidInput && !hasJwksRefreshSecondsValidInput && oAuth.getJwksExpirySeconds() < KafkaListenerAuthenticationOAuth.DEFAULT_JWKS_REFRESH_SECONDS + 60)) { errors.add("listener " + listenerName + ": The refresh interval has to be at least 60 seconds shorter then the expiry interval specified in `jwksExpirySeconds`"); } if (!oAuth.isAccessTokenIsJwt()) { if (oAuth.getJwksEndpointUri() != null) { errors.add("listener " + listenerName + ": accessTokenIsJwt=false can not be used together with jwksEndpointUri"); } if (!oAuth.isCheckAccessTokenType()) { errors.add("listener " + listenerName + ": checkAccessTokenType can not be set to false when accessTokenIsJwt is false"); } } if (!oAuth.isCheckAccessTokenType() && oAuth.getIntrospectionEndpointUri() != null) { errors.add("listener " + listenerName + ": checkAccessTokenType=false can not be used together with introspectionEndpointUri"); } if (oAuth.getValidTokenType() != null && oAuth.getIntrospectionEndpointUri() == null) { errors.add("listener " + listenerName + ": validTokenType can only be used with introspectionEndpointUri"); } } }  static void validate(int replicas, List<GenericKafkaListener> listeners); final static List<Integer> FORBIDDEN_PORTS; final static int LOWEST_ALLOWED_PORT_NUMBER; }
@Test public void testValidateOauth() { GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName("listener1") .withPort(9900) .withType(KafkaListenerType.INTERNAL) .withAuth(new KafkaListenerAuthenticationOAuthBuilder().build()) .build(); List<GenericKafkaListener> listeners = asList(listener1); Exception exception = assertThrows(InvalidResourceException.class, () -> ListenersValidator.validate(3, listeners)); assertThat(exception.getMessage(), allOf( containsString("listener listener1: Introspection endpoint URI or JWKS endpoint URI has to be specified"), containsString("listener listener1: Valid Issuer URI has to be specified or 'checkIssuer' set to false"))); }