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 com.cloud.agent.transport;
  
  import java.io.ByteArrayInputStream;
  import java.io.ByteArrayOutputStream;
  import java.io.IOException;
  import java.io.StringReader;
  import java.lang.reflect.Type;
  import java.nio.ByteBuffer;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.Iterator;
  import java.util.List;
  import java.util.zip.GZIPInputStream;
  import java.util.zip.GZIPOutputStream;
  
  import org.apache.commons.lang3.StringUtils;
  import org.apache.logging.log4j.Level;
  import org.apache.logging.log4j.Logger;
  import org.apache.logging.log4j.LogManager;
  
  import com.cloud.agent.api.Answer;
  import com.cloud.agent.api.BadCommand;
  import com.cloud.agent.api.Command;
  import com.cloud.agent.api.SecStorageFirewallCfgCommand.PortConfig;
  import com.cloud.exception.UnsupportedVersionException;
  import com.cloud.serializer.GsonHelper;
  import com.cloud.utils.HumanReadableJson;
  import com.cloud.utils.NumbersUtil;
  import com.cloud.utils.Pair;
  import com.cloud.utils.exception.CloudRuntimeException;
  import com.google.gson.Gson;
  import com.google.gson.JsonArray;
  import com.google.gson.JsonDeserializationContext;
  import com.google.gson.JsonDeserializer;
  import com.google.gson.JsonElement;
  import com.google.gson.JsonNull;
  import com.google.gson.JsonParseException;
  import com.google.gson.JsonSerializationContext;
  import com.google.gson.JsonSerializer;
  import com.google.gson.stream.JsonReader;
  
  /**
   * Request is a simple wrapper around command and answer to add sequencing,
   * versioning, and flags. Note that the version here represents the changes
   * in the over the wire protocol. For example, if we decide to not use Gson.
   * It does not version the changes in the actual commands. That's expected
   * to be done by adding new classes to the command and answer list.
   *
   * A request looks as follows:
   * 1. Version - 1 byte;
   * 2. Flags - 3 bytes;
   * 3. Sequence - 8 bytes;
   * 4. Length - 4 bytes;
   * 5. ManagementServerId - 8 bytes;
   * 6. AgentId - 8 bytes;
   * 7. Data Package.
   *
   */
  public class Request {
      protected static Logger LOGGER = LogManager.getLogger(Request.class);
  
      protected static final Gson s_gson = GsonHelper.getGson();
      protected static final Gson s_gogger = GsonHelper.getGsonLogger();
      protected static final Logger s_gsonLogger = GsonHelper.getLogger();
  
      public enum Version {
          v1, // using gson to marshall
          v2, // now using gson as marshalled.
          v3; // Adding routing information into the Request data structure.
  
          public static Version get(final byte ver) throws UnsupportedVersionException {
              for (final Version version : Version.values()) {
                  if (ver == version.ordinal()) {
                      return version;
                  }
              }
              throw new UnsupportedVersionException("Can't lookup version: " + ver, UnsupportedVersionException.UnknownVersion);
          }
      };
  
      protected static final short FLAG_RESPONSE = 0x0;
      protected static final short FLAG_REQUEST = 0x1;
      protected static final short FLAG_STOP_ON_ERROR = 0x2;
      protected static final short FLAG_IN_SEQUENCE = 0x4;
      protected static final short FLAG_FROM_SERVER = 0x20;
      protected static final short FLAG_CONTROL = 0x40;
      protected static final short FLAG_COMPRESSED = 0x80;
  
      protected Version _ver;
      protected long _session;
      protected long _seq;
      protected short _flags;
      protected long _mgmtId;
      protected long _via;
      protected long _agentId;
      protected Command[] _cmds;
      protected String _content;
      protected String _agentName;
  
      protected Request() {
      }
  
      protected Request(Version ver, long seq, long agentId, long mgmtId, long via, short flags, final Command[] cmds) {
          _ver = ver;
          _cmds = cmds;
          _flags = flags;
          _seq = seq;
          _via = via;
          _agentId = agentId;
          _mgmtId = mgmtId;
          setInSequence(cmds);
      }
  
      protected Request(Version ver, long seq, long agentId, long mgmtId, short flags, final Command[] cmds) {
          this(ver, seq, agentId, mgmtId, agentId, flags, cmds);
      }
  
      protected Request(Version ver, long seq, long agentId, long mgmtId, long via, short flags, final String content) {
          this(ver, seq, agentId, mgmtId, via, flags, (Command[])null);
          _content = content;
      }
  
      public Request(long agentId, long mgmtId, Command command, boolean fromServer) {
          this(agentId, mgmtId, new Command[] {command}, true, fromServer);
      }
  
      public Request(long agentId, long mgmtId, Command[] cmds, boolean stopOnError, boolean fromServer) {
          this(Version.v1, -1l, agentId, mgmtId, (short)0, cmds);
          setStopOnError(stopOnError);
          setFromServer(fromServer);
      }
  
      public Request(long agentId, String agentName, long mgmtId, Command[] cmds, boolean stopOnError, boolean fromServer) {
          this(agentId, mgmtId, cmds, stopOnError, fromServer);
          setAgentName(agentName);
      }
  
      public void setSequence(long seq) {
          _seq = seq;
      }
  
      protected void setInSequence(Command[] cmds) {
          if (cmds == null) {
              return;
          }
          for (Command cmd : cmds) {
              if (cmd.executeInSequence()) {
                  setInSequence(true);
                  break;
              }
          }
      }
  
      protected Request(final Request that, final Command[] cmds) {
          _ver = that._ver;
          _seq = that._seq;
          setInSequence(that.executeInSequence());
          setStopOnError(that.stopOnError());
          _cmds = cmds;
          _mgmtId = that._mgmtId;
          _via = that._via;
          _agentId = that._agentId;
          _agentName = that._agentName;
          setFromServer(!that.isFromServer());
      }
  
      private final void setStopOnError(boolean stopOnError) {
          _flags |= (stopOnError ? FLAG_STOP_ON_ERROR : 0);
      }
  
      private final void setAgentName(String agentName) {
          _agentName = agentName;
      }
  
      private final void setInSequence(boolean inSequence) {
          _flags |= (inSequence ? FLAG_IN_SEQUENCE : 0);
      }
  
      public boolean isControl() {
          return (_flags & FLAG_CONTROL) > 0;
      }
  
      public void setControl(boolean control) {
          _flags |= (control ? FLAG_CONTROL : 0);
      }
  
      private final void setFromServer(boolean fromServer) {
          _flags |= (fromServer ? FLAG_FROM_SERVER : 0);
      }
  
      public long getManagementServerId() {
          return _mgmtId;
      }
  
      public boolean isFromServer() {
          return (_flags & FLAG_FROM_SERVER) > 0;
      }
  
      public Version getVersion() {
          return _ver;
      }
  
      public void setAgentId(long agentId) {
          _agentId = agentId;
      }
  
      public void setVia(long viaId) {
          _via = viaId;
      }
  
      public boolean executeInSequence() {
          return (_flags & FLAG_IN_SEQUENCE) > 0;
      }
  
      public long getSequence() {
          return _seq;
      }
  
      public boolean stopOnError() {
          return (_flags & FLAG_STOP_ON_ERROR) > 0;
      }
  
      public Command getCommand() {
          getCommands();
          return _cmds[0];
      }
  
      public Command[] getCommands() {
          if (_cmds == null) {
              try {
                  StringReader reader = new StringReader(_content);
                  JsonReader jsonReader = new JsonReader(reader);
                  jsonReader.setLenient(true);
                  _cmds = s_gson.fromJson(jsonReader, (Type)Command[].class);
              } catch (JsonParseException e) {
                  LOGGER.error("Caught problem while parsing JSON command " + _content, e);
                  _cmds = new Command[] { new BadCommand() };
              } catch (RuntimeException e) {
                  LOGGER.error("Caught problem with " + _content, e);
                  throw e;
              }
          }
          return _cmds;
      }
  
      protected String getType() {
          return "Cmd ";
      }
  
      protected ByteBuffer serializeHeader(final int contentSize) {
          final ByteBuffer buffer = ByteBuffer.allocate(40);
          buffer.put(getVersionInByte());
          buffer.put((byte)0);
          buffer.putShort(getFlags());
          buffer.putLong(_seq);
          // The size here is uncompressed size, if the data is compressed.
          buffer.putInt(contentSize);
          buffer.putLong(_mgmtId);
          buffer.putLong(_agentId);
          buffer.putLong(_via);
          buffer.flip();
  
          return buffer;
      }
  
      public static ByteBuffer doDecompress(ByteBuffer buffer, int length) {
          byte[] byteArrayIn = new byte[1024];
          ByteArrayInputStream byteIn;
          if (buffer.hasArray()) {
              byteIn = new ByteArrayInputStream(buffer.array(), buffer.position() + buffer.arrayOffset(), buffer.remaining());
          } else {
              byte[] array = new byte[buffer.limit() - buffer.position()];
              buffer.get(array);
              byteIn = new ByteArrayInputStream(array);
          }
          ByteBuffer retBuff = ByteBuffer.allocate(length);
          int len = 0;
          try {
              GZIPInputStream in = new GZIPInputStream(byteIn);
              while ((len = in.read(byteArrayIn)) > 0) {
                  retBuff.put(byteArrayIn, 0, len);
              }
              in.close();
          } catch (IOException e) {
              LOGGER.error("Fail to decompress the request!", e);
          }
          retBuff.flip();
          return retBuff;
      }
  
      public static ByteBuffer doCompress(ByteBuffer buffer, int length) {
          ByteArrayOutputStream byteOut = new ByteArrayOutputStream(length);
          byte[] array;
          if (buffer.hasArray()) {
              array = buffer.array();
          } else {
              array = new byte[buffer.capacity()];
              buffer.get(array);
          }
          try {
              GZIPOutputStream out = new GZIPOutputStream(byteOut, length);
              out.write(array);
              out.finish();
              out.close();
          } catch (IOException e) {
              LOGGER.error("Fail to compress the request!", e);
          }
          return ByteBuffer.wrap(byteOut.toByteArray());
      }
  
      public ByteBuffer[] toBytes() {
          final ByteBuffer[] buffers = new ByteBuffer[2];
          ByteBuffer tmp;
  
          if (_content == null) {
              _content = s_gson.toJson(_cmds, _cmds.getClass());
          }
          tmp = ByteBuffer.wrap(_content.getBytes());
          int capacity = tmp.capacity();
          /* Check if we need to compress the data */
          if (capacity >= 8192) {
              tmp = doCompress(tmp, capacity);
              _flags |= FLAG_COMPRESSED;
          }
          buffers[1] = tmp;
          buffers[0] = serializeHeader(capacity);
  
          return buffers;
      }
  
      public byte[] getBytes() {
          final ByteBuffer[] buffers = toBytes();
          final int len1 = buffers[0].remaining();
          final int len2 = buffers[1].remaining();
          final byte[] bytes = new byte[len1 + len2];
          buffers[0].get(bytes, 0, len1);
          buffers[1].get(bytes, len1, len2);
          return bytes;
      }
  
      protected byte getVersionInByte() {
          return (byte)_ver.ordinal();
      }
  
      protected short getFlags() {
          return (short)(((this instanceof Response) ? FLAG_RESPONSE : FLAG_REQUEST) | _flags);
      }
  
      public void logD(String msg) {
          logD(msg, true);
      }
  
      public void logD(String msg, boolean logContent) {
          if (LOGGER.isDebugEnabled()) {
              String log = log(msg, logContent, Level.DEBUG);
              if (log != null) {
                  LOGGER.debug(log);
              }
          }
      }
  
      public void logT(String msg, boolean logD) {
          if (LOGGER.isTraceEnabled()) {
              String log = log(msg, true, Level.TRACE);
              if (log != null) {
                  LOGGER.trace(log);
              }
          } else if (logD && LOGGER.isDebugEnabled()) {
              String log = log(msg, false, Level.DEBUG);
              if (log != null) {
                  LOGGER.debug(log);
              }
          }
      }
  
      @Override
      public String toString() {
          return log("", true, Level.DEBUG);
      }
  
      protected String log(String msg, boolean logContent, Level level) {
          StringBuilder content = new StringBuilder();
          if (logContent) {
              if (_cmds == null) {
                  try {
                      _cmds = s_gson.fromJson(_content, this instanceof Response ? Answer[].class : Command[].class);
                  } catch (RuntimeException e) {
                      LOGGER.error("Unable to deserialize from json: " + _content);
                      throw e;
                  }
              }
              try {
                  s_gogger.toJson(_cmds, content);
              } catch (Throwable e) {
                  StringBuilder buff = new StringBuilder();
                  for (Command cmd : _cmds) {
                      buff.append(cmd.getClass().getSimpleName()).append("/");
                  }
                  LOGGER.error("Gson serialization error " + buff.toString(), e);
                  assert false : "More gson errors on " + buff.toString();
                  return "";
              }
              content = new StringBuilder(HumanReadableJson.getHumanReadableBytesJson(content.toString()));
              if (content.length() <= (1 + _cmds.length * 3)) {
                  return null;
              }
          } else {
              if (_cmds == null) {
                  _cmds = s_gson.fromJson(_content, this instanceof Response ? Answer[].class : Command[].class);
              }
              content.append("{ ");
              for (Command cmd : _cmds) {
                  content.append(cmd.getClass().getSimpleName()).append(", ");
              }
              content.replace(content.length() - 2, content.length(), " }");
  
          }
  
          StringBuilder buf = new StringBuilder("Seq ");
  
          buf.append(_agentId).append("-").append(_seq).append(": ");
  
          buf.append(msg);
          buf.append(" { ").append(getType());
          if (_agentName != null) {
              buf.append(", MgmtId: ").append(_mgmtId).append(", via: ").append(_via).append("(" + _agentName + ")");
          } else {
              buf.append(", MgmtId: ").append(_mgmtId).append(", via: ").append(_via);
          }
          buf.append(", Ver: ").append(_ver.toString());
          buf.append(", Flags: ").append(Integer.toBinaryString(getFlags())).append(", ");
          String cleanContent = content.toString();
          if(cleanContent.contains("password")) {
              buf.append(cleanPassword(cleanContent));
          } else {
              buf.append(content);
          }
          buf.append(" }");
          return buf.toString();
      }
  
      public static String cleanPassword(String logString) {
          String cleanLogString = null;
          if (logString != null) {
              cleanLogString = logString;
              String[] temp = logString.split(",");
              int i = 0;
              if (temp != null) {
                  while (i < temp.length) {
                      temp[i] = com.cloud.utils.StringUtils.cleanString(temp[i]);
                      i++;
                  }
                  List<String> stringList = new ArrayList<String>();
                  Collections.addAll(stringList, temp);
                  cleanLogString = StringUtils.join(stringList, ",");
              }
          }
          return cleanLogString;
      }
  
      /**
       * Factory method for Request and Response. It expects the bytes to be
       * correctly formed so it's possible that it throws underflow exceptions
       * but you shouldn't be concerned about that since that all bytes sent in
       * should already be formatted correctly.
       *
       * @param bytes bytes to be converted.
       * @return Request or Response depending on the data.
       * @throws ClassNotFoundException if the Command or Answer can not be formed.
       * @throws
       */
      public static Request parse(final byte[] bytes) throws ClassNotFoundException, UnsupportedVersionException {
          ByteBuffer buff = ByteBuffer.wrap(bytes);
          final byte ver = buff.get();
          final Version version = Version.get(ver);
          if (version.ordinal() != Version.v1.ordinal() && version.ordinal() != Version.v3.ordinal()) {
              throw new UnsupportedVersionException("This version is no longer supported: " + version.toString(), UnsupportedVersionException.IncompatibleVersion);
          }
          buff.get();
          final short flags = buff.getShort();
          final boolean isRequest = (flags & FLAG_REQUEST) > 0;
  
          final long seq = buff.getLong();
          // The size here is uncompressed size, if the data is compressed.
          final int size = buff.getInt();
          final long mgmtId = buff.getLong();
          final long agentId = buff.getLong();
  
          long via;
          if (version.ordinal() == Version.v1.ordinal()) {
              via = buff.getLong();
          } else {
              via = agentId;
          }
  
          if ((flags & FLAG_COMPRESSED) != 0) {
              buff = doDecompress(buff, size);
          }
  
          byte[] command = null;
          int offset = 0;
          if (buff.hasArray()) {
              command = buff.array();
              offset = buff.arrayOffset() + buff.position();
          } else {
              command = new byte[buff.remaining()];
              buff.get(command);
              offset = 0;
          }
  
          final String content = new String(command, offset, command.length - offset);
  
          if (isRequest) {
              return new Request(version, seq, agentId, mgmtId, via, flags, content);
          } else {
              return new Response(Version.get(ver), seq, agentId, mgmtId, via, flags, content);
          }
      }
  
      public long getAgentId() {
          return _agentId;
      }
  
      public long getViaAgentId() {
          return _via;
      }
  
      public static boolean requiresSequentialExecution(final byte[] bytes) {
          return (bytes[3] & FLAG_IN_SEQUENCE) > 0;
      }
  
      public static Version getVersion(final byte[] bytes) throws UnsupportedVersionException {
          try {
              return Version.get(bytes[0]);
          } catch (UnsupportedVersionException e) {
              throw new CloudRuntimeException("Unsupported version: " + bytes[0]);
          }
      }
  
      public static long getManagementServerId(final byte[] bytes) {
          return NumbersUtil.bytesToLong(bytes, 16);
      }
  
      public static long getAgentId(final byte[] bytes) {
          return NumbersUtil.bytesToLong(bytes, 24);
      }
  
      public static long getViaAgentId(final byte[] bytes) {
          return NumbersUtil.bytesToLong(bytes, 32);
      }
  
      public static boolean fromServer(final byte[] bytes) {
          return (bytes[3] & FLAG_FROM_SERVER) > 0;
      }
  
      public static boolean isRequest(final byte[] bytes) {
          return (bytes[3] & FLAG_REQUEST) > 0;
      }
  
      public static long getSequence(final byte[] bytes) {
          return NumbersUtil.bytesToLong(bytes, 4);
      }
  
      public static boolean isControl(final byte[] bytes) {
          return (bytes[3] & FLAG_CONTROL) > 0;
      }
  
      public static class NwGroupsCommandTypeAdaptor implements JsonDeserializer<Pair<Long, Long>>, JsonSerializer<Pair<Long, Long>> {
  
          public NwGroupsCommandTypeAdaptor() {
          }
  
          @Override
          public JsonElement serialize(Pair<Long, Long> src, java.lang.reflect.Type typeOfSrc, JsonSerializationContext context) {
              JsonArray array = new JsonArray();
              if (src.first() != null) {
                  array.add(s_gson.toJsonTree(src.first()));
              } else {
                  array.add(new JsonNull());
              }
  
              if (src.second() != null) {
                  array.add(s_gson.toJsonTree(src.second()));
              } else {
                  array.add(new JsonNull());
              }
  
              return array;
          }
  
          @Override
          public Pair<Long, Long> deserialize(JsonElement json, java.lang.reflect.Type type, JsonDeserializationContext context) throws JsonParseException {
              Pair<Long, Long> pairs = new Pair<Long, Long>(null, null);
              JsonArray array = json.getAsJsonArray();
              if (array.size() != 2) {
                  return pairs;
              }
              JsonElement element = array.get(0);
              if (!element.isJsonNull()) {
                  pairs.first(element.getAsLong());
              }
  
              element = array.get(1);
              if (!element.isJsonNull()) {
                  pairs.second(element.getAsLong());
              }
  
              return pairs;
          }
  
      }
  
      public static class PortConfigListTypeAdaptor implements JsonDeserializer<List<PortConfig>>, JsonSerializer<List<PortConfig>> {
  
          public PortConfigListTypeAdaptor() {
          }
  
          @Override
          public JsonElement serialize(List<PortConfig> src, Type typeOfSrc, JsonSerializationContext context) {
              if (src.size() == 0) {
                  return new JsonNull();
              }
              JsonArray array = new JsonArray();
              for (PortConfig pc : src) {
                  array.add(s_gson.toJsonTree(pc));
              }
  
              return array;
          }
  
          @Override
          public List<PortConfig> deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) throws JsonParseException {
              if (json.isJsonNull()) {
                  return new ArrayList<PortConfig>();
              }
              List<PortConfig> pcs = new ArrayList<PortConfig>();
              JsonArray array = json.getAsJsonArray();
              Iterator<JsonElement> it = array.iterator();
              while (it.hasNext()) {
                  JsonElement element = it.next();
                  pcs.add(s_gson.fromJson(element, PortConfig.class));
              }
              return pcs;
          }
      }
  }
  