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.cassandra.hints;
  
  import java.io.DataInput;
  import java.io.IOException;
  import java.nio.file.Files;
  import java.nio.file.Path;
  import java.util.HashMap;
  import java.util.Map;
  import java.util.Optional;
  import java.util.UUID;
  import java.util.regex.Pattern;
  import java.util.zip.CRC32;
  import javax.crypto.Cipher;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.base.MoreObjects;
  import com.google.common.base.Objects;
  import com.google.common.collect.ImmutableMap;
  
  import org.apache.cassandra.io.util.File;
  import org.apache.cassandra.io.util.FileInputStreamPlus;
  import com.google.common.io.ByteStreams;
  import com.google.common.io.CountingOutputStream;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.config.ParameterizedClass;
  import org.apache.cassandra.db.TypeSizes;
  import org.apache.cassandra.io.FSReadError;
  import org.apache.cassandra.io.compress.ICompressor;
  import org.apache.cassandra.io.util.DataOutputPlus;
  import org.apache.cassandra.net.MessagingService;
  import org.apache.cassandra.schema.CompressionParams;
  import org.apache.cassandra.security.EncryptionContext;
  import org.apache.cassandra.serializers.MarshalException;
  import org.apache.cassandra.utils.Hex;
  import org.apache.cassandra.utils.JsonUtils;
  
  import static org.apache.cassandra.utils.FBUtilities.updateChecksumInt;
  
  /**
   * Describes the host id, the version, the timestamp of creation, and an arbitrary map of JSON-encoded parameters of a
   * hints file.
   *
   * Written in the beginning of each hints file.
   */
  final class HintsDescriptor
  {
      private static final Logger logger = LoggerFactory.getLogger(HintsDescriptor.class);
  
      static final int VERSION_30 = 1;
      static final int VERSION_40 = 2;
      static final int VERSION_50 = 3;
      static final int VERSION_51 = 4;
      static final int CURRENT_VERSION = DatabaseDescriptor.getStorageCompatibilityMode().isBefore(5) ? VERSION_40 : VERSION_51;
  
      static final String COMPRESSION = "compression";
      static final String ENCRYPTION = "encryption";
  
      static final Pattern pattern =
          Pattern.compile("^[a-fA-F0-9]{8}\\-[a-fA-F0-9]{4}\\-[a-fA-F0-9]{4}\\-[a-fA-F0-9]{4}\\-[a-fA-F0-9]{12}\\-(\\d+)\\-(\\d+)\\.hints$");
  
      final UUID hostId;
      final int version;
      final long timestamp;
      final String hintsFileName;
      final String crc32FileName;
  
      final ImmutableMap<String, Object> parameters;
      final ParameterizedClass compressionConfig;
  
      private final Cipher cipher;
      private final ICompressor compressor;
  
      HintsDescriptor(UUID hostId, int version, long timestamp, ImmutableMap<String, Object> parameters)
      {
          this.hostId = hostId;
          this.version = version;
          this.timestamp = timestamp;
          hintsFileName = hostId + "-" + timestamp + '-' + version + ".hints";
          crc32FileName = hostId + "-" + timestamp + '-' + version + ".crc32";
          compressionConfig = createCompressionConfig(parameters);
  
          EncryptionData encryption = createEncryption(parameters);
          if (encryption == null)
          {
              cipher = null;
              compressor = null;
          }
          else
          {
              if (compressionConfig != null)
                  throw new IllegalStateException("a hints file cannot be configured for both compression and encryption");
              cipher = encryption.cipher;
              compressor = encryption.compressor;
              parameters = encryption.params;
          }
  
          this.parameters = parameters;
      }
  
      HintsDescriptor(UUID hostId, long timestamp, ImmutableMap<String, Object> parameters)
      {
          this(hostId, CURRENT_VERSION, timestamp, parameters);
      }
  
      HintsDescriptor(UUID hostId, long timestamp)
      {
          this(hostId, CURRENT_VERSION, timestamp, ImmutableMap.<String, Object>of());
      }
  
      @SuppressWarnings("unchecked")
      static ParameterizedClass createCompressionConfig(Map<String, Object> params)
      {
          if (params.containsKey(COMPRESSION))
          {
              Map<String, Object> compressorConfig = (Map<String, Object>) params.get(COMPRESSION);
              return new ParameterizedClass((String) compressorConfig.get(ParameterizedClass.CLASS_NAME),
                                            (Map<String, String>) compressorConfig.get(ParameterizedClass.PARAMETERS));
          }
          else
          {
              return null;
          }
      }
  
      /**
       * Create, if necessary, the required encryption components (for either decrpyt or encrypt operations).
       * Note that in the case of encyption (this is, when writing out a new hints file), we need to write
       * the cipher's IV out to the header so it can be used when decrypting. Thus, we need to add an additional
       * entry to the {@code params} map.
       *
       * @param params the base parameters into the descriptor.
       * @return null if not using encryption; else, the initialized {@link Cipher} and a possibly updated version
       * of the {@code params} map.
       */
      @SuppressWarnings("unchecked")
      static EncryptionData createEncryption(ImmutableMap<String, Object> params)
      {
          if (params.containsKey(ENCRYPTION))
          {
              Map<?, ?> encryptionConfig = (Map<?, ?>) params.get(ENCRYPTION);
              EncryptionContext encryptionContext = EncryptionContext.createFromMap(encryptionConfig, DatabaseDescriptor.getEncryptionContext());
  
              try
              {
                  Cipher cipher;
                  if (encryptionConfig.containsKey(EncryptionContext.ENCRYPTION_IV))
                  {
                      cipher = encryptionContext.getDecryptor();
                  }
                  else
                  {
                      cipher = encryptionContext.getEncryptor();
                      ImmutableMap<String, Object> encParams = ImmutableMap.<String, Object>builder()
                                                                   .putAll(encryptionContext.toHeaderParameters())
                                                                   .put(EncryptionContext.ENCRYPTION_IV, Hex.bytesToHex(cipher.getIV()))
                                                                   .build();
  
                      Map<String, Object> map = new HashMap<>(params);
                      map.put(ENCRYPTION, encParams);
                      params = ImmutableMap.<String, Object>builder().putAll(map).build();
                  }
                  return new EncryptionData(cipher, encryptionContext.getCompressor(), params);
              }
              catch (IOException ioe)
              {
                  logger.warn("failed to create encyption context for hints file. ignoring encryption for hints.", ioe);
                  return null;
              }
          }
          else
          {
              return null;
          }
      }
  
      private static final class EncryptionData
      {
          final Cipher cipher;
          final ICompressor compressor;
          final ImmutableMap<String, Object> params;
  
          private EncryptionData(Cipher cipher, ICompressor compressor, ImmutableMap<String, Object> params)
          {
              this.cipher = cipher;
              this.compressor = compressor;
              this.params = params;
          }
      }
  
      String fileName()
      {
          return hintsFileName;
      }
  
      String checksumFileName()
      {
          return crc32FileName;
      }
  
      File file(File hintsDirectory)
      {
          return new File(hintsDirectory, fileName());
      }
  
      File checksumFile(File hintsDirectory)
      {
          return new File(hintsDirectory, checksumFileName());
      }
  
      /** cached size of the represented hints file */
      private transient volatile long hintsFileSize = -1L;
  
      long hintsFileSize(File hintsDirectory)
      {
          long size = hintsFileSize;
          if (size == -1L) // we may race and duplicate lookup the first time the size is being queried, but that is fine
              hintsFileSize = size = file(hintsDirectory).length();
          return size;
      }
  
      void hintsFileSize(long value)
      {
          hintsFileSize = value;
      }
  
      int messagingVersion()
      {
          return messagingVersion(version);
      }
  
      static int messagingVersion(int hintsVersion)
      {
          switch (hintsVersion)
          {
              case VERSION_30:
                  return MessagingService.VERSION_30;
              case VERSION_40:
                  return MessagingService.VERSION_40;
              case VERSION_50:
                  return MessagingService.VERSION_50;
              case VERSION_51:
                  return MessagingService.VERSION_51;
              default:
                  throw new AssertionError();
          }
      }
  
      static boolean isHintFileName(Path path)
      {
          return pattern.matcher(path.getFileName().toString()).matches();
      }
  
      static Optional<HintsDescriptor> readFromFileQuietly(Path path)
      {
          try (FileInputStreamPlus raf = new FileInputStreamPlus(path))
          {
              return Optional.of(deserialize(raf));
          }
          catch (ChecksumMismatchException e)
          {
              throw new FSReadError(e, path);
          }
          catch (IOException e)
          {
              handleDescriptorIOE(e, path);
              return Optional.empty();
          }
      }
  
      @VisibleForTesting
      static void handleDescriptorIOE(IOException e, Path path)
      {
          try
          {
              if (Files.size(path) > 0)
              {
                  String newFileName = path.getFileName().toString().replace(".hints", ".corrupt.hints");
                  Path target = path.getParent().resolve(newFileName);
                  logger.error("Failed to deserialize hints descriptor {} - saving file as {}", path.toString(), target, e);
                  Files.move(path, target);
              }
              else
              {
                  logger.warn("Found empty hints file {} on startup, removing", path.toString());
                  Files.delete(path);
              }
          }
          catch (IOException ex)
          {
              logger.error("Error handling corrupt hints file {}", path.toString(), ex);
          }
      }
  
      static HintsDescriptor readFromFile(File path)
      {
          try (FileInputStreamPlus raf = new FileInputStreamPlus(path))
          {
              return deserialize(raf);
          }
          catch (IOException e)
          {
              throw new FSReadError(e, path);
          }
      }
  
      public boolean isCompressed()
      {
          return compressionConfig != null;
      }
  
      public boolean isEncrypted()
      {
          return cipher != null;
      }
  
      public ICompressor createCompressor()
      {
          if (isCompressed())
              return CompressionParams.createCompressor(compressionConfig);
          if (isEncrypted())
              return compressor;
          return null;
      }
  
      public Cipher getCipher()
      {
          return isEncrypted() ? cipher : null;
      }
  
      @Override
      public String toString()
      {
          return MoreObjects.toStringHelper(this)
                            .add("hostId", hostId)
                            .add("version", version)
                            .add("timestamp", timestamp)
                            .add("parameters", parameters)
                            .toString();
      }
  
      @Override
      public boolean equals(Object o)
      {
          if (this == o)
              return true;
  
          if (!(o instanceof HintsDescriptor))
              return false;
  
          HintsDescriptor hd = (HintsDescriptor) o;
  
          return Objects.equal(hostId, hd.hostId)
              && Objects.equal(version, hd.version)
              && Objects.equal(timestamp, hd.timestamp)
              && Objects.equal(parameters, hd.parameters);
      }
  
      @Override
      public int hashCode()
      {
          return Objects.hashCode(hostId, version, timestamp, parameters);
      }
  
      void serialize(DataOutputPlus out) throws IOException
      {
          CRC32 crc = new CRC32();
  
          out.writeInt(version);
          updateChecksumInt(crc, version);
  
          out.writeLong(timestamp);
          updateChecksumLong(crc, timestamp);
  
          out.writeLong(hostId.getMostSignificantBits());
          updateChecksumLong(crc, hostId.getMostSignificantBits());
          out.writeLong(hostId.getLeastSignificantBits());
          updateChecksumLong(crc, hostId.getLeastSignificantBits());
  
          byte[] paramsBytes = JsonUtils.writeAsJsonBytes(parameters);
          out.writeInt(paramsBytes.length);
          updateChecksumInt(crc, paramsBytes.length);
          out.writeInt((int) crc.getValue());
  
          out.write(paramsBytes);
          crc.update(paramsBytes, 0, paramsBytes.length);
  
          out.writeInt((int) crc.getValue());
      }
  
      int serializedSize()
      {
          int size = TypeSizes.sizeof(version);
          size += TypeSizes.sizeof(timestamp);
  
          size += TypeSizes.sizeof(hostId.getMostSignificantBits());
          size += TypeSizes.sizeof(hostId.getLeastSignificantBits());
  
          // Let's avoid allocation of serialized output, use counting output stream
          int serializedParamsLength;
          try (CountingOutputStream out = new CountingOutputStream(ByteStreams.nullOutputStream()))
          {
              JsonUtils.JSON_OBJECT_MAPPER.writeValue(out, parameters);
              serializedParamsLength = (int) out.getCount();
          }
          catch (IOException e)
          {
              throw new RuntimeException(e); // should never happen
          }
          size += TypeSizes.sizeof(serializedParamsLength);
          size += 4; // size checksum
          size += serializedParamsLength;
          size += 4; // total checksum
  
          return size;
      }
  
      static HintsDescriptor deserialize(DataInput in) throws IOException
      {
          CRC32 crc = new CRC32();
  
          int version = in.readInt();
          updateChecksumInt(crc, version);
  
          long timestamp = in.readLong();
          updateChecksumLong(crc, timestamp);
  
          long msb = in.readLong();
          updateChecksumLong(crc, msb);
          long lsb = in.readLong();
          updateChecksumLong(crc, lsb);
  
          UUID hostId = new UUID(msb, lsb);
  
          int paramsLength = in.readInt();
          updateChecksumInt(crc, paramsLength);
          validateCRC(in.readInt(), (int) crc.getValue());
  
          byte[] paramsBytes = new byte[paramsLength];
          in.readFully(paramsBytes, 0, paramsLength);
          crc.update(paramsBytes, 0, paramsLength);
          validateCRC(in.readInt(), (int) crc.getValue());
  
          return new HintsDescriptor(hostId, version, timestamp, decodeJSONBytes(paramsBytes));
      }
  
      @SuppressWarnings("unchecked")
      private static ImmutableMap<String, Object> decodeJSONBytes(byte[] bytes)
      {
          // note: There is a Jackson module (datatype-guava) for directly reading into ImmutableMap,
          // but would require adding dependency to that
          try
          {
              return ImmutableMap.copyOf(JsonUtils.fromJsonMap(bytes));
          }
          catch (MarshalException e)
          {
              // Couple of options here: up to 4.0 simply returned null and caller failed with NPE.
              // Seems cleaner to throw an exception
              throw new MarshalException("Corrupt HintsDescriptor serialization, problem: " + e.getMessage(), e);
          }
      }
  
      private static void updateChecksumLong(CRC32 crc, long value)
      {
          updateChecksumInt(crc, (int) (value & 0xFFFFFFFFL));
          updateChecksumInt(crc, (int) (value >>> 32));
      }
  
      private static void validateCRC(int expected, int actual) throws IOException
      {
          if (expected != actual)
              throw new ChecksumMismatchException("Hints Descriptor CRC Mismatch");
      }
  }
  