content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.kafka.controller;
  
  import org.apache.kafka.clients.admin.ScramMechanism;
  import org.apache.kafka.common.message.AlterUserScramCredentialsRequestData;
  import org.apache.kafka.common.message.AlterUserScramCredentialsRequestData.ScramCredentialDeletion;
  import org.apache.kafka.common.message.AlterUserScramCredentialsRequestData.ScramCredentialUpsertion;
  import org.apache.kafka.common.message.AlterUserScramCredentialsResponseData;
  import org.apache.kafka.common.message.AlterUserScramCredentialsResponseData.AlterUserScramCredentialsResult;
  import org.apache.kafka.common.metadata.RemoveUserScramCredentialRecord;
  import org.apache.kafka.common.metadata.UserScramCredentialRecord;
  import org.apache.kafka.common.requests.ApiError;
  import org.apache.kafka.common.security.scram.internals.ScramFormatter;
  import org.apache.kafka.common.utils.LogContext;
  import org.apache.kafka.server.common.ApiMessageAndVersion;
  import org.apache.kafka.server.common.MetadataVersion;
  import org.apache.kafka.timeline.SnapshotRegistry;
  import org.apache.kafka.timeline.TimelineHashMap;
  
  import org.slf4j.Logger;
  
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Map.Entry;
  import java.util.Objects;
  
  import static org.apache.kafka.common.protocol.Errors.DUPLICATE_RESOURCE;
  import static org.apache.kafka.common.protocol.Errors.NONE;
  import static org.apache.kafka.common.protocol.Errors.RESOURCE_NOT_FOUND;
  import static org.apache.kafka.common.protocol.Errors.UNACCEPTABLE_CREDENTIAL;
  import static org.apache.kafka.common.protocol.Errors.UNSUPPORTED_SASL_MECHANISM;
  import static org.apache.kafka.common.protocol.Errors.UNSUPPORTED_VERSION;
  
  
  /**
   * Manages SCRAM credentials.
   */
  public class ScramControlManager {
      static final int MAX_ITERATIONS = 16384;
  
      static class Builder {
          private LogContext logContext = null;
          private SnapshotRegistry snapshotRegistry = null;
  
          Builder setLogContext(LogContext logContext) {
              this.logContext = logContext;
              return this;
          }
  
          Builder setSnapshotRegistry(SnapshotRegistry snapshotRegistry) {
              this.snapshotRegistry = snapshotRegistry;
              return this;
          }
  
          ScramControlManager build() {
              if (logContext == null) logContext = new LogContext();
              if (snapshotRegistry == null) snapshotRegistry = new SnapshotRegistry(logContext);
              return new ScramControlManager(logContext,
                  snapshotRegistry);
          }
      }
  
      static class ScramCredentialKey {
          private final String username;
          private final ScramMechanism mechanism;
  
          ScramCredentialKey(String username, ScramMechanism mechanism) {
              this.username = username;
              this.mechanism = mechanism;
          }
  
          @Override
          public int hashCode() {
              return Objects.hash(username, mechanism);
          }
  
          @Override
          public boolean equals(Object o) {
              if (o == null) return false;
              if (!(o.getClass() == this.getClass())) return false;
              ScramCredentialKey other = (ScramCredentialKey) o;
              return username.equals(other.username) &&
                  mechanism.equals(other.mechanism);
          }
  
          @Override
          public String toString() {
              return "ScramCredentialKey" +
                  "(username=" + username +
                  ", mechanism=" + mechanism +
                  ")";
          }
      }
  
      static class ScramCredentialValue {
          private final byte[] salt;
          private final byte[] storedKey;
          private final byte[] serverKey;
          private final int iterations;
  
          ScramCredentialValue(
              byte[] salt,
              byte[] storedKey,
              byte[] serverKey,
              int iterations
          ) {
              this.salt = salt;
              this.storedKey = storedKey;
              this.serverKey = serverKey;
              this.iterations = iterations;
          }
  
          @Override
          public int hashCode() {
              return Objects.hash(
                  Arrays.hashCode(salt),
                  Arrays.hashCode(storedKey),
                  Arrays.hashCode(serverKey),
                  iterations
              );
          }
  
          @Override
          public boolean equals(Object o) {
              if (o == null) return false;
              if (!(o.getClass() == this.getClass())) return false;
              ScramCredentialValue other = (ScramCredentialValue) o;
              return Arrays.equals(salt, other.salt) &&
                  Arrays.equals(storedKey, other.storedKey) &&
                  Arrays.equals(serverKey, other.serverKey) &&
                  iterations == other.iterations;
          }
  
          @Override
          public String toString() {
              return "ScramCredentialValue" +
                  "(salt=" + "[hidden]" +
                  ", storedKey=" + "[hidden]" +
                  ", serverKey=" + "[hidden]" +
                  ", iterations=" + "[hidden]" +
                  ")";
          }
      }
  
      private final Logger log;
      private final TimelineHashMap<ScramCredentialKey, ScramCredentialValue> credentials;
  
      private ScramControlManager(
          LogContext logContext,
          SnapshotRegistry snapshotRegistry
      ) {
          this.log = logContext.logger(ScramControlManager.class);
          this.credentials = new TimelineHashMap<>(snapshotRegistry, 0);
      }
  
      /*
       * Pass in the MetadataVersion so that we can return a response to the caller 
       * if the current metadataVersion is too low.
       */
      public ControllerResult<AlterUserScramCredentialsResponseData> alterCredentials(
          AlterUserScramCredentialsRequestData request,
          MetadataVersion metadataVersion
      ) {
          boolean scramIsSupported = metadataVersion.isScramSupported();
          Map<String, ScramCredentialDeletion> userToDeletion = new HashMap<>();
          Map<String, ScramCredentialUpsertion> userToUpsert = new HashMap<>();
          Map<String, ApiError> userToError = new HashMap<>();
  
          for (ScramCredentialDeletion deletion : request.deletions()) {
              if (!userToError.containsKey(deletion.name())) {
                  if (userToDeletion.remove(deletion.name()) != null) {
                      userToError.put(deletion.name(), new ApiError(DUPLICATE_RESOURCE,
                          "A user credential cannot be altered twice in the same request"));
                  } else {
                      if (!scramIsSupported) {
                          userToError.put(deletion.name(), new ApiError(UNSUPPORTED_VERSION,
                              "The current metadata.version does not support SCRAM"));
                      } else {
                          ApiError error = validateDeletion(deletion);
                          if (error.isFailure()) {
                              userToError.put(deletion.name(), error);
                          } else {
                              userToDeletion.put(deletion.name(), deletion);
                          }
                      }
                  }
              }
          }
          for (ScramCredentialUpsertion upsertion : request.upsertions()) {
              if (!userToError.containsKey(upsertion.name())) {
                  if (userToDeletion.remove(upsertion.name()) != null ||
                          userToUpsert.remove(upsertion.name()) != null) {
                      userToError.put(upsertion.name(), new ApiError(DUPLICATE_RESOURCE,
                          "A user credential cannot be altered twice in the same request"));
                  } else {
                      if (!scramIsSupported) {
                          userToError.put(upsertion.name(), new ApiError(UNSUPPORTED_VERSION,
                              "The current metadata.version does not support SCRAM"));
                      } else {
                          ApiError error = validateUpsertion(upsertion);
                          if (error.isFailure()) {
                              userToError.put(upsertion.name(), error);
                          } else {
                              userToUpsert.put(upsertion.name(), upsertion);
                          }
                      }
                  }
              }
          }
          AlterUserScramCredentialsResponseData response = new AlterUserScramCredentialsResponseData();
          List<ApiMessageAndVersion> records = new ArrayList<>();
          for (ScramCredentialDeletion deletion : userToDeletion.values()) {
              response.results().add(new AlterUserScramCredentialsResult().
                  setUser(deletion.name()).
                  setErrorCode(NONE.code()).
                  setErrorMessage(null));
              records.add(new ApiMessageAndVersion(new RemoveUserScramCredentialRecord().
                  setName(deletion.name()).
                  setMechanism(deletion.mechanism()), (short) 0));
          }
          for (ScramCredentialUpsertion upsertion : userToUpsert.values()) {
              ApiError error = finishUpsertion(records, upsertion);
              if (!error.isFailure()) {
                  response.results().add(new AlterUserScramCredentialsResult().
                      setUser(upsertion.name()).
                      setErrorCode(NONE.code()).
                      setErrorMessage(null));
              } else {
                  userToError.put(upsertion.name(), error);
              }
          }
          for (Entry<String, ApiError> entry : userToError.entrySet()) {
              response.results().add(new AlterUserScramCredentialsResult().
                  setUser(entry.getKey()).
                  setErrorCode(entry.getValue().error().code()).
                  setErrorMessage(entry.getValue().message()));
          }
          return ControllerResult.atomicOf(records, response);
      }
  
      static ApiError finishUpsertion(List<ApiMessageAndVersion> records, ScramCredentialUpsertion upsertion) {
          org.apache.kafka.common.security.scram.internals.ScramMechanism internalMechanism = 
                  org.apache.kafka.common.security.scram.internals.ScramMechanism.forMechanismName(
                  ScramMechanism.fromType(upsertion.mechanism()).mechanismName());
  
          try { // Convert from saltedPassword to storedKey and serverKey
              ScramFormatter formatter = new ScramFormatter(internalMechanism);
  
              records.add(new ApiMessageAndVersion(new UserScramCredentialRecord().
                  setName(upsertion.name()).
                  setMechanism(upsertion.mechanism()).
                  setSalt(upsertion.salt()).
  
                  // Convert from saltedPassword to storedKey and serverKey
                  setStoredKey(formatter.storedKey(formatter.clientKey(upsertion.saltedPassword()))).
                  setServerKey(formatter.serverKey(upsertion.saltedPassword())).
                  setIterations(upsertion.iterations()), (short) 0));
  
          } catch (Throwable e) {
              return ApiError.fromThrowable(e);
          }
          return ApiError.NONE;
      }
  
      static ApiError validateUpsertion(ScramCredentialUpsertion upsertion) {
          ScramMechanism mechanism = ScramMechanism.fromType(upsertion.mechanism());
          ApiError error = validateScramUsernameAndMechanism(upsertion.name(), mechanism);
          if (error.isFailure()) return error;
          org.apache.kafka.common.security.scram.internals.ScramMechanism internalMechanism =
              org.apache.kafka.common.security.scram.internals.ScramMechanism.forMechanismName(mechanism.mechanismName());
          if (upsertion.iterations() < internalMechanism.minIterations()) {
              return new ApiError(UNACCEPTABLE_CREDENTIAL, "Too few iterations");
          } else if (upsertion.iterations() > MAX_ITERATIONS) {
              return new ApiError(UNACCEPTABLE_CREDENTIAL, "Too many iterations");
          }
          return ApiError.NONE;
      }
  
      ApiError validateDeletion(ScramCredentialDeletion deletion) {
          ApiError error = validateScramUsernameAndMechanism(deletion.name(),
              ScramMechanism.fromType(deletion.mechanism()));
          if (error.isFailure()) return error;
          ScramCredentialKey key = new ScramCredentialKey(deletion.name(),
              ScramMechanism.fromType(deletion.mechanism()));
          if (!credentials.containsKey(key)) {
              return new ApiError(RESOURCE_NOT_FOUND,
                  "Attempt to delete a user credential that does not exist");
          }
          return ApiError.NONE;
      }
  
      static ApiError validateScramUsernameAndMechanism(
          String username,
          ScramMechanism mechanism
      ) {
          if (username.isEmpty()) {
              return new ApiError(UNACCEPTABLE_CREDENTIAL, "Username must not be empty");
          }
          if (mechanism == ScramMechanism.UNKNOWN) {
              return new ApiError(UNSUPPORTED_SASL_MECHANISM, "Unknown SCRAM mechanism");
          }
          return ApiError.NONE;
      }
  
      public void replay(RemoveUserScramCredentialRecord record) {
          ScramCredentialKey key = new ScramCredentialKey(record.name(),
              ScramMechanism.fromType(record.mechanism()));
          if (credentials.remove(key) == null) {
              throw new RuntimeException("Unable to find credential to delete: " + key);
          }
          log.info("Replayed RemoveUserScramCredentialRecord for {} with mechanism {}.",
              key.username, key.mechanism);
      }
  
      public void replay(UserScramCredentialRecord record) {
          ScramCredentialKey key = new ScramCredentialKey(record.name(),
              ScramMechanism.fromType(record.mechanism()));
          ScramCredentialValue value = new ScramCredentialValue(record.salt(),
              record.storedKey(),
              record.serverKey(),
              record.iterations());
          if (credentials.put(key, value) == null) {
              log.info("Replayed UserScramCredentialRecord creating new entry for {} with " +
                  "mechanism {}.", key.username, key.mechanism);
          } else {
              log.info("Replayed UserScramCredentialRecord modifying existing entry for {} " +
                  "with mechanism {}.", key.username, key.mechanism);
          }
      }
  
  }
  