Principal = User:ANONYMOUS is Denied operation

[MariaIRubio]

Hi! I am trying to integrate keycloak and strimzi in my application, but I always get this errror in the Cluster

2025-01-31 17:27:18,277 INFO Principal = User:ANONYMOUS is Denied operation = DESCRIBE from host = 10.233.96.37 on resource = Cluster:LITERAL:kafka-cluster for request = DescribeLogDirs with resourceRefCount = 1 based on rule DefaultDeny (kafka.authorizer.logger) [data-plane-kafka-request-handler-1]

and in consecuence this error in my application

2025-01-31T18:20:08.701+01:00 ERROR 433593 --- [integrationKafkaKeycloak] [ad | producer-1] org.apache.kafka.clients.Metadata : [Producer clientId=producer-1] Topic authorization failed for topics [x_topic]

This is the configuration of my cluster that concerns authentication and authorization:

authentication:
          type: oauth
          clientId: kafka
          clientSecret:
            key: secret
            secretName: broker-oauth-secret
          validIssuerUri: https://dev.ingo.nentec.de/auth/realms/InGOad
          jwksEndpointUri: https://dev.ingo.nentec.de/auth/realms/InGOad/protocol/openid-connect/certs
          userNameClaim: preferred_username
          tlsTrustedCertificates:
          - secretName: ca-truststore
            certificate: ca.crt
        configuration:
          bootstrap:
            nodePort: 32303
          brokers:
            - broker: 0
              advertisedHost: 172.20.82.131
            - broker: 1
              advertisedHost: 172.20.82.132
            - broker: 2
              advertisedHost: 172.20.82.133
    authorization:
      type: keycloak
      clientId: kafka
      tokenEndpointUri: https://dev.ingo.nentec.de/auth/realms/InGOad/protocol/openid-connect/token
      delegateToKafkaAcls: true
      superUsers:
      - service-account-kafka

In keycloak I have three clients:
- kafka: that has the authorization activated with a client_credentials flow and manages the roles and policies
- team-a-client: that simulates a producer with role Dev Team A - User:service-account-team-a-client
- team-a-client: that simulates a consumer with role Dev Team B - User:service-account-team-b-client

In my application I get the token throught a request

@Configuration
public class TokenRequest {

    @Value("${spring.security.oauth2.client.provider.keycloak.token-uri}")
    private String tokenUri;
    @Value("${spring.security.oauth2.client.registration.keycloak.client-id}")
    private String clientId;
    @Value("${spring.security.oauth2.client.registration.keycloak.client-secret}")
    private String clientSecret;

    private static final Logger logger = LoggerFactory.getLogger(TokenRequest.class);

    public void executeTokenRequest() {
        String apiUrl = null;
        try {
            // 1. Request to get the access token
            final HttpHeaders authHeaders = new HttpHeaders();
            authHeaders.setContentType(MediaType.APPLICATION_FORM_URLENCODED);

            final Map<String, String> authBody = new HashMap<>();
            authBody.put("grant_type", "client_credentials");
            authBody.put("client_id", clientId);
            authBody.put("client_secret", clientSecret);

            final StringBuilder authRequestBody = new StringBuilder();
            authBody.forEach((k, v) -> authRequestBody.append(k).append("=").append(v).append("&"));
            final String authRequestBodyString = authRequestBody.toString().replaceAll("&$", "");

            final HttpEntity<String> authRequestEntity =
                    new HttpEntity<>(authRequestBodyString, authHeaders);

            final RestTemplate restTemplate = new RestTemplate();

            final ResponseEntity<Map> authResponse =
                    restTemplate.exchange(tokenUri, HttpMethod.POST, authRequestEntity, Map.class);
            final String accessToken = authResponse.getBody().get("access_token").toString();
            if (authResponse.getStatusCode() == HttpStatus.OK) {
                logger.info("Successfully authenticated token");
                logger.info("Access token: {}", accessToken);
            } else {
                logger.info("Failed to authenticated token");
            }

            // 2. Retrieve UserInfo
            Map<String, Object> userInfo = getUserInfo(accessToken);
            if (userInfo != null) {
                logger.info("User Info: {}", userInfo);
            }

            // 3. Use the token to access the endpoint
            apiUrl = "http://localhost:8081/teamA/produce";
            final HttpHeaders apiHeaders = new HttpHeaders();
            apiHeaders.add("Authorization", "Bearer " + accessToken);

            final HttpEntity<String> apiRequestEntity = new HttpEntity<>(apiHeaders);

            final ResponseEntity<String> apiResponse =
                    restTemplate.exchange(apiUrl, HttpMethod.GET, apiRequestEntity, String.class);

            logger.info("ACCESS AUTHORIZED - Status code: {}", apiResponse.getStatusCode());
            logger.info("Response body: {}", apiResponse.getBody());
        } catch (HttpClientErrorException e) {
            switch (e.getStatusCode()) {
                case UNAUTHORIZED:
                    logger.error("Error 401: Not authorized. Verify Credentials.");
                    break;
                case FORBIDDEN:
                    logger.error("Error 403: Access Forbidden. Verify Credentials.");
                    break;
                default:
                    logger.error("Error HTTP: {}", e.getStatusCode());
            }
        }
    }

    public Map<String, Object> getUserInfo(String accessToken){
        String userInfoEndpoint = "https://dev.ingo.nentec.de/auth/realms/InGOad/protocol/openid-connect/userinfo";

        try {
            HttpHeaders headers = new HttpHeaders();
            headers.add("Authorization", "Bearer " + accessToken);

            HttpEntity<String> requestEntity = new HttpEntity<>(headers);
            RestTemplate restTemplate = new RestTemplate();

            ResponseEntity<Map> response = restTemplate.exchange(userInfoEndpoint, HttpMethod.GET, requestEntity, Map.class);

            if(response.getStatusCode() == HttpStatus.OK){
                logger.info("UserInfo retrieved successfully");
                return response.getBody();
            }else {
                logger.error("Failed to retrieve UserInfo. Status: {}: {}", response.getStatusCode());
            }
        }catch (HttpClientErrorException e){
            switch (e.getStatusCode()){
                case UNAUTHORIZED:
                    logger.error("Error 401: Not authorized. Verify Credentials.");
                    break;
                case FORBIDDEN:
                    logger.error("Error 403: Access Forbidden. Verify Credentials.");
                    break;
                default:
                     logger.error("Error HTTP: {}", e.getStatusCode());

            }
        }
        return null;
    }
}

And this is the Security Configuration that I have to integrate keycloak with my app, decode the token and grant the authentication/authorization:

@Configuration
public class SecurityConfig {

    private final Logger logger = LoggerFactory.getLogger(SecurityConfig.class);
    @Value("${spring.security.oauth2.client.registration.keycloak.client-id}")
    private String clientId;
    @Value("${spring.security.oauth2.client.registration.keycloak.client-secret}")
    private String clientSecret;
    @Value("${spring.security.oauth2.client.provider.keycloak.issuer-uri}")
    private String issuerUri;
    @Value("${spring.security.oauth2.client.provider.keycloak.authorization-uri}")
    private String authorizationUri;
    @Value("${spring.security.oauth2.client.provider.keycloak.jwk-set-uri}")
    private String jwkSetUri;
    @Value("${spring.security.oauth2.client.provider.keycloak.token-uri}")
    private String tokenUri;
    @Value("${spring.security.oauth2.client.provider.keycloak.user-info-uri}")
    private String userInfoUri;
    @Value("${keycloak.registrationid}")
    private String registrationId;

    @Bean
    public SecurityFilterChain securityFilterChain(HttpSecurity http) throws Exception {

        return http.cors(cors -> cors.disable()).csrf(csrf -> csrf.disable()).authorizeHttpRequests(
                        httpRequests -> httpRequests.requestMatchers("/teamA/**").hasRole("Dev Team A")
                                .requestMatchers("/teamB/**").hasRole("Dev Team B")
                                .requestMatchers("/admin/**").hasRole("AdminRole").anyRequest()
                                .authenticated()).sessionManagement(
                        session -> session.sessionCreationPolicy(SessionCreationPolicy.STATELESS))
                .oauth2ResourceServer(oauth2ResourceServer -> oauth2ResourceServer.jwt(
                        jwt -> jwt.jwtAuthenticationConverter(jwtAuthenticationConverter())))
                .build();
    }

    // Validates the JWT tokens
    @Bean
    public JwtDecoder jwtDecoder() {
        return NimbusJwtDecoder.withJwkSetUri(jwkSetUri).build();
    }

    // Authentication based on JWT
    private JwtAuthenticationConverter jwtAuthenticationConverter() {
        final JwtGrantedAuthoritiesConverter authoritiesConverter =
                new JwtGrantedAuthoritiesConverter();
        authoritiesConverter.setAuthorityPrefix("ROLE_");
        authoritiesConverter.setAuthoritiesClaimName("realm_access.roles");

        final JwtAuthenticationConverter jwtConverter = new JwtAuthenticationConverter();
        jwtConverter.setJwtGrantedAuthoritiesConverter(jwt -> {
            logger.info("JWT claims: {}", jwt.getClaims());
            final Collection<GrantedAuthority> authorities = new ArrayList<>();

            final String preferredUsername =
                    jwt.getClaimAsString(StandardClaimNames.PREFERRED_USERNAME);
            authorities.add(new SimpleGrantedAuthority("User:" + preferredUsername));
            logger.info("preferredUsername: {}", preferredUsername);

            final Map<String, Object> realmAccess = jwt.getClaim("realm_access");
            extractRoles(authorities, realmAccess);

            final Map<String, Object> resourceAccess = jwt.getClaim("resource_access");
            if (resourceAccess != null) {
                final Map<String, Object> kafkaAccess =
                        (Map<String, Object>) resourceAccess.get("kafka");
                extractRoles(authorities, kafkaAccess);
            }
            logger.info("Granted authorities for {}: {}", preferredUsername, authorities);
            final Authentication authentication =
                    new JwtAuthenticationToken(jwt, authorities, preferredUsername);
            logger.info("Authentication: {}", authentication.getName());
            return authorities;
        });
        jwtConverter.setPrincipalClaimName(StandardClaimNames.PREFERRED_USERNAME);
        return jwtConverter;
    }



    private void extractRoles(Collection<GrantedAuthority> authorities,
            Map<String, Object> rolesMap) {
        if (rolesMap != null) {
            final List<String> roles = (List<String>) rolesMap.get("roles");
            if (roles != null) {
                for (final String role : roles) {
                    authorities.add(new SimpleGrantedAuthority("ROLE_" + role));
                }
            }
        }
    }


    // Manage OAuth2 client configuration for keycloak authentication
    @Bean
    public ClientRegistrationRepository clientRegistrationRepository() {
        return new InMemoryClientRegistrationRepository(this.keycloakRegistration());
    }

    private ClientRegistration keycloakRegistration() {
        return ClientRegistration.withRegistrationId("keycloak").clientId(clientId)
                .clientSecret(clientSecret)
                .authorizationGrantType(AuthorizationGrantType.CLIENT_CREDENTIALS)
                .scope("openid", "profile", "email", "offline_access").issuerUri(issuerUri)
                .authorizationUri(authorizationUri).jwkSetUri(jwkSetUri).tokenUri(tokenUri)
                .userInfoUri(userInfoUri)
                .userNameAttributeName(StandardClaimNames.PREFERRED_USERNAME).build();
    }
}

And my Producer, where I add the jaasConfig to connect it with strimzi:

@Component
public class Producer {

    @Value("${spring.security.oauth2.client.registration.keycloak.client-id}")
    private String clientId;
    @Value("${spring.security.oauth2.client.registration.keycloak.client-secret}")
    private String clientSecret;
    @Value("${spring.security.oauth2.client.provider.keycloak.token-uri}")
    private String tokenUri;

    private static final Logger logger = LoggerFactory.getLogger(Producer.class);

    @Autowired
    private JaasConfig jaasConfig;

    public void produce(){

        String topic = "x_topic";

        Properties properties = PgcKafkaUtils.getProducerProperties();
        //properties.putAll(jaasConfig.getJaasProperties(clientId));
        properties.put("sasl.oauthbearer.sub.claim.name", "preferred_username");
        //properties.put("sasl.oauthbearer.expected.audience", clientId);
        properties.put("sasl.oauthbearer.jwks.endpoint.url", "https://dev.ingo.nentec.de/auth/realms/InGOad/protocol/openid-connect/certs");
        properties.put("sasl.oauthbearer.token.endpoint.url", "https://dev.ingo.nentec.de/auth/realms/InGOad/protocol/openid-connect/token");
        properties.put(Config.OAUTH_CLIENT_SECRET, clientSecret);
        properties.put(Config.OAUTH_CLIENT_ID, clientId);
        properties.put(Config.OAUTH_USERNAME_CLAIM, "preferred_username");
        properties.put(Config.OAUTH_USERNAME_PREFIX, "User:");
        properties.put(ClientConfig.OAUTH_TOKEN_ENDPOINT_URI, tokenUri);
        properties.put("security.protocol", "SASL_SSL");
        properties.put("sasl.mechanism", "OAUTHBEARER");
        properties.put("sasl.jaas.config",
                "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required "
                        + "oauth.client.id=\"team-a-client\" "
                        + "oauth.client.secret=\"team-a-client-secret\" "
                        + "oauth.username.prefix=\"User:\" "
                        + "oauth.username.claim=\"preferred_username\""
                        + "oauth.token.endpoint.uri=\"https://dev.ingo.nentec.de/auth/realms/InGOad/protocol/openid-connect/token\";");

        Config externals = new Config(properties);
        ConfigProperties.resolve(properties);
        ConfigProperties.resolveAndExportToSystemProperties(properties);

        String externalHost = externals.getValue("keycloak.host", "NO VALUE");
        String externalRealm = externals.getValue("realm", "NO VALUE");
        String externalToken = externals.getValue(ClientConfig.OAUTH_ACCESS_TOKEN, "NO VALUE");

        logger.info("Externals: {}", externals);
        logger.info("ExternalHost: {}", externalHost);
        logger.info("ExternalRealm: {}", externalRealm);
        logger.info("ExternalToken: {}", externalToken);

        PgcProducer<String> producer = new PgcProducer<>(topic, properties);
        try {
            producer.sendMessage("key3", "message3");
        }catch (AuthenticationException | AuthorizationException e){
            producer.close();
            producer = new PgcProducer<>(topic, properties);
        }


    }

}

I tried to follow the examples of https://github.com/strimzi/strimzi-kafka-oauth/blob/main/examples/producer/src/main/java/io/strimzi/examples/producer/ExampleProducer.java

but I get the externals from Config (import io.strimzi.kafka.oauth.client.ClientConfig;
import io.strimzi.kafka.oauth.common.Config;
import io.strimzi.kafka.oauth.common.ConfigProperties;) as null.

If someone have an idea it is more than welcome!!! Thank you in advance

[scholzj]

User:ANONYMOUS suggests your user is not authenticated. So you should look at the Kafka CR if it is correctly configured, if you connect to the right port etc.

[scholzj]

What I pass as a token in the Producer is the JaasConfig with the tokenUri. Isn`t that right? So first I have a request where I get the token and use it to authenticate the client, and then I pass it in the jaasConfig with the credentials and the tokenUri.

What do I have too complicated, the Producer? I also don´t understand very well the Config externals = new Config(); from the example. In my case the externals are null.

All our OAuth examples are just few lines of code. You shared 3 different classes and I do not understad how they really fit together. They do not seem to be needed. You should not need any interaction with the OAuth server. The OAuth client should do that for you.

But at the end, whatever you do produces a bad username that seems to be Dev Team A based on the exception. So either your token is wrong or your settings for extracting the username from it are wrong.

I'm afraid I cannot help you more with that. @mstruk might have some ideas when he gets to this. But I think the easiest way would be to:

• Eliminate the other listeners and unrelated authorization errors as I suggested
• Start from our examples to try to figure out what might be wrong.

[MariaIRubio]

So the only thing needed, up to the example is the JaasConfig in the producer and nothing else?

Thank you very very much for your time and your patience! I do really value and appreciate it.

[scholzj]

No, there is not just the JAAS config. There is also the callback for example.

[MariaIRubio]

And where is the Config externals = new Config() coming from? From Strimzi I guess because of the import. But should it give me the back the configuration from keycloak? Host, etc? In my case it gives null.

[scholzj]

I think that just loads values from env vars and system properties.