// Copyright 2018 The Chromium OS Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "vm_tools/cicerone/tremplin_listener_impl.h"

#include <arpa/inet.h>
#include <inttypes.h>
#include <stdio.h>

#include <map>
#include <memory>
#include <string>
#include <utility>
#include <vector>

#include <base/bind.h>
#include <base/logging.h>
#include <base/strings/string_util.h>
#include <base/threading/thread_task_runner_handle.h>

#include "vm_tools/cicerone/service.h"

namespace vm_tools {
namespace cicerone {

TremplinListenerImpl::TremplinListenerImpl(
    base::WeakPtr<vm_tools::cicerone::Service> service)
    : service_(service), task_runner_(base::ThreadTaskRunnerHandle::Get()) {}

void TremplinListenerImpl::OverridePeerAddressForTesting(
    const std::string& testing_peer_address) {
  base::AutoLock lock_scope(testing_peer_address_lock_);
  testing_peer_address_ = testing_peer_address;
}

grpc::Status TremplinListenerImpl::TremplinReady(
    grpc::ServerContext* ctx,
    const vm_tools::tremplin::TremplinStartupInfo* request,
    vm_tools::tremplin::EmptyMessage* response) {
  uint32_t cid = ExtractCidFromPeerAddress(ctx);
  if (cid == 0) {
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Failed parsing vsock cid for TremplinListener");
  }

  bool result = false;
  base::WaitableEvent event(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);
  task_runner_->PostTask(
      FROM_HERE, base::Bind(&vm_tools::cicerone::Service::ConnectTremplin,
                            service_, cid, &result, &event));
  event.Wait();
  if (!result) {
    LOG(ERROR) << "Received TremplinReady but could not find matching VM: "
               << ctx->peer();
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Cannot find VM for TremplinListener");
  }

  return grpc::Status::OK;
}

grpc::Status TremplinListenerImpl::UpdateCreateStatus(
    grpc::ServerContext* ctx,
    const vm_tools::tremplin::ContainerCreationProgress* request,
    vm_tools::tremplin::EmptyMessage* response) {
  uint32_t cid = ExtractCidFromPeerAddress(ctx);
  if (cid == 0) {
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Failed parsing vsock cid for TremplinListener");
  }

  bool result = false;
  base::WaitableEvent event(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);
  if (request->status() == tremplin::ContainerCreationProgress::DOWNLOADING) {
    task_runner_->PostTask(
        FROM_HERE,
        base::Bind(&vm_tools::cicerone::Service::LxdContainerDownloading,
                   service_, cid, request->container_name(),
                   request->download_progress(), &result, &event));
  } else {
    vm_tools::cicerone::Service::CreateStatus status;
    switch (request->status()) {
      case tremplin::ContainerCreationProgress::CREATED:
        status = vm_tools::cicerone::Service::CreateStatus::CREATED;
        break;
      case tremplin::ContainerCreationProgress::DOWNLOAD_TIMED_OUT:
        status = vm_tools::cicerone::Service::CreateStatus::DOWNLOAD_TIMED_OUT;
        break;
      case tremplin::ContainerCreationProgress::CANCELLED:
        status = vm_tools::cicerone::Service::CreateStatus::CANCELLED;
        break;
      case tremplin::ContainerCreationProgress::FAILED:
        status = vm_tools::cicerone::Service::CreateStatus::FAILED;
        break;
      default:
        status = vm_tools::cicerone::Service::CreateStatus::UNKNOWN;
        break;
    }
    task_runner_->PostTask(
        FROM_HERE, base::Bind(&vm_tools::cicerone::Service::LxdContainerCreated,
                              service_, cid, request->container_name(), status,
                              request->failure_reason(), &result, &event));
  }

  event.Wait();
  if (!result) {
    LOG(ERROR)
        << "Received UpdateCreateStatus RPC but could not find matching VM: "
        << ctx->peer();
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Cannot find VM for TremplinListener");
  }

  return grpc::Status::OK;
}

grpc::Status TremplinListenerImpl::UpdateDeletionStatus(
    grpc::ServerContext* ctx,
    const vm_tools::tremplin::ContainerDeletionProgress* request,
    vm_tools::tremplin::EmptyMessage* response) {
  uint32_t cid = ExtractCidFromPeerAddress(ctx);
  if (cid == 0) {
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Failed parsing vsock cid for TremplinListener");
  }

  bool result = false;
  base::WaitableEvent event(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);

  task_runner_->PostTask(
      FROM_HERE,
      base::Bind(&vm_tools::cicerone::Service::LxdContainerDeleted, service_,
                 cid, request->container_name(), request->status(),
                 request->failure_reason(), &result, &event));

  event.Wait();
  if (!result) {
    LOG(ERROR)
        << "Received UpdateDeletionStatus RPC but could not find matching VM: "
        << ctx->peer();
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Cannot find VM for TremplinListener");
  }

  return grpc::Status::OK;
}

grpc::Status TremplinListenerImpl::UpdateStartStatus(
    grpc::ServerContext* ctx,
    const vm_tools::tremplin::ContainerStartProgress* request,
    vm_tools::tremplin::EmptyMessage* response) {
  uint32_t cid = ExtractCidFromPeerAddress(ctx);
  if (cid == 0) {
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Failed parsing vsock cid for TremplinListener");
  }

  bool result = false;
  base::WaitableEvent event(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);
  vm_tools::cicerone::Service::StartStatus status;
  switch (request->status()) {
    case tremplin::ContainerStartProgress::STARTED:
      status = vm_tools::cicerone::Service::StartStatus::STARTED;
      break;
    case tremplin::ContainerStartProgress::CANCELLED:
      status = vm_tools::cicerone::Service::StartStatus::CANCELLED;
      break;
    case tremplin::ContainerStartProgress::FAILED:
      status = vm_tools::cicerone::Service::StartStatus::FAILED;
      break;
    case tremplin::ContainerStartProgress::STARTING:
      status = vm_tools::cicerone::Service::StartStatus::STARTING;
      break;
    default:
      status = vm_tools::cicerone::Service::StartStatus::UNKNOWN;
      break;
  }
  task_runner_->PostTask(
      FROM_HERE, base::Bind(&vm_tools::cicerone::Service::LxdContainerStarting,
                            service_, cid, request->container_name(), status,
                            request->failure_reason(), &result, &event));

  event.Wait();
  if (!result) {
    LOG(ERROR)
        << "Received UpdateStartStatus RPC but could not find matching VM: "
        << ctx->peer();
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Cannot find VM for TremplinListener");
  }

  return grpc::Status::OK;
}

grpc::Status TremplinListenerImpl::UpdateExportStatus(
    grpc::ServerContext* ctx,
    const vm_tools::tremplin::ContainerExportProgress* request,
    vm_tools::tremplin::EmptyMessage* response) {
  uint32_t cid = ExtractCidFromPeerAddress(ctx);
  if (cid == 0) {
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Failed parsing vsock cid for TremplinListener");
  }

  ExportLxdContainerProgressSignal progress_signal;
  if (!ExportLxdContainerProgressSignal::Status_IsValid(
          static_cast<int>(request->status()))) {
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Invalid status field in protobuf request");
  }
  progress_signal.set_status(
      static_cast<ExportLxdContainerProgressSignal::Status>(request->status()));
  progress_signal.set_container_name(request->container_name());
  progress_signal.set_failure_reason(request->failure_reason());
  progress_signal.set_total_input_files(request->total_input_files());
  progress_signal.set_total_input_bytes(request->total_input_bytes());
  progress_signal.set_input_files_streamed(request->input_files_streamed());
  progress_signal.set_input_bytes_streamed(request->input_bytes_streamed());
  progress_signal.set_bytes_exported(request->bytes_exported());
  base::WaitableEvent event(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);
  bool result = false;
  task_runner_->PostTask(
      FROM_HERE,
      base::Bind(&vm_tools::cicerone::Service::ContainerExportProgress,
                 service_, cid, &progress_signal, &result, &event));
  event.Wait();
  if (!result) {
    LOG(ERROR) << "Failure updating container export progress";
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Failure in UpdateExportStatus");
  }

  return grpc::Status::OK;
}

grpc::Status TremplinListenerImpl::UpdateImportStatus(
    grpc::ServerContext* ctx,
    const vm_tools::tremplin::ContainerImportProgress* request,
    vm_tools::tremplin::EmptyMessage* response) {
  uint32_t cid = ExtractCidFromPeerAddress(ctx);
  if (cid == 0) {
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Failed parsing vsock cid for TremplinListener");
  }

  ImportLxdContainerProgressSignal progress_signal;
  if (!ImportLxdContainerProgressSignal::Status_IsValid(
          static_cast<int>(request->status()))) {
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Invalid status field in protobuf request");
  }
  progress_signal.set_status(
      static_cast<ImportLxdContainerProgressSignal::Status>(request->status()));
  progress_signal.set_container_name(request->container_name());
  progress_signal.set_progress_percent(request->progress_percent());
  progress_signal.set_progress_speed(request->progress_speed());
  progress_signal.set_failure_reason(request->failure_reason());
  progress_signal.set_architecture_device(request->architecture_device());
  progress_signal.set_architecture_container(request->architecture_container());
  progress_signal.set_available_space(request->disk_space_available_bytes());
  progress_signal.set_min_required_space(request->disk_space_required_bytes());

  base::WaitableEvent event(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);
  bool result = false;
  task_runner_->PostTask(
      FROM_HERE,
      base::Bind(&vm_tools::cicerone::Service::ContainerImportProgress,
                 service_, cid, &progress_signal, &result, &event));
  event.Wait();
  if (!result) {
    LOG(ERROR) << "Failure updating container import progress";
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Failure in UpdateImportStatus");
  }

  return grpc::Status::OK;
}

grpc::Status TremplinListenerImpl::ContainerShutdown(
    grpc::ServerContext* ctx,
    const vm_tools::tremplin::ContainerShutdownInfo* request,
    vm_tools::tremplin::EmptyMessage* response) {
  uint32_t cid = ExtractCidFromPeerAddress(ctx);
  if (cid == 0) {
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Failed parsing vsock cid for TremplinListener");
  }

  if (request->container_name().empty()) {
    return grpc::Status(grpc::INVALID_ARGUMENT,
                        "`container_name` cannot be empty");
  }

  // Calls coming from tremplin are trusted to use container_name rather than
  // container_token.
  std::string container_token = "";
  base::WaitableEvent event(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);

  bool result = false;
  task_runner_->PostTask(
      FROM_HERE, base::Bind(&vm_tools::cicerone::Service::ContainerShutdown,
                            service_, request->container_name(),
                            container_token, cid, &result, &event));
  event.Wait();
  if (!result) {
    LOG(ERROR) << "Error in tremplin listener ContainerShutdown for "
               << request->container_name();
  }
  return grpc::Status::OK;
}

grpc::Status TremplinListenerImpl::UpdateListeningPorts(
    grpc::ServerContext* ctx,
    const vm_tools::tremplin::ListeningPortInfo* request,
    vm_tools::tremplin::EmptyMessage* response) {
  uint32_t cid = ExtractCidFromPeerAddress(ctx);
  if (cid == 0) {
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Failed parsing vsock cid for TremplinListener");
  }

  std::map<std::string, std::vector<uint16_t>> listening_tcp4_ports;
  for (auto& pair : request->container_ports()) {
    std::vector<uint16_t> target_ports;
    for (int i = 0; i < pair.second.listening_tcp4_ports_size(); i++) {
      target_ports.push_back(pair.second.listening_tcp4_ports(i));
    }

    listening_tcp4_ports[pair.first] = target_ports;
  }

  base::WaitableEvent event(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);

  bool result = false;
  task_runner_->PostTask(
      FROM_HERE,
      base::Bind(&vm_tools::cicerone::Service::UpdateListeningPorts, service_,
                 std::move(listening_tcp4_ports), cid, &result, &event));
  event.Wait();
  if (!result) {
    LOG(ERROR) << "Error in tremplin listener UpdateListeningPorts";
  }
  return grpc::Status::OK;
}

grpc::Status TremplinListenerImpl::UpgradeContainerStatus(
    grpc::ServerContext* ctx,
    const vm_tools::tremplin::UpgradeContainerProgress* request,
    vm_tools::tremplin::EmptyMessage* response) {
  uint32_t cid = ExtractCidFromPeerAddress(ctx);
  if (cid == 0) {
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Failed parsing vsock cid for TremplinListener");
  }

  UpgradeContainerProgressSignal progress_signal;
  if (!UpgradeContainerProgressSignal::Status_IsValid(
          static_cast<int>(request->status()))) {
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Invalid status field in protobuf request");
  }
  progress_signal.set_status(
      static_cast<UpgradeContainerProgressSignal::Status>(request->status()));
  progress_signal.set_container_name(request->container_name());
  progress_signal.set_failure_reason(request->failure_reason());
  progress_signal.mutable_progress_messages()->CopyFrom(
      request->progress_messages());

  base::WaitableEvent event(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);
  bool result = false;
  task_runner_->PostTask(
      FROM_HERE,
      base::Bind(&vm_tools::cicerone::Service::ContainerUpgradeProgress,
                 service_, cid, &progress_signal, &result, &event));
  event.Wait();
  if (!result) {
    LOG(ERROR) << "Failure sending upgrade container progress";
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Failure in UgradeContainertatus");
  }

  return grpc::Status::OK;
}

grpc::Status TremplinListenerImpl::UpdateStartLxdStatus(
    grpc::ServerContext* ctx,
    const vm_tools::tremplin::StartLxdProgress* request,
    vm_tools::tremplin::EmptyMessage* response) {
  uint32_t cid = ExtractCidFromPeerAddress(ctx);
  if (cid == 0) {
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Failed parsing vsock cid for TremplinListener");
  }

  StartLxdProgressSignal progress_signal;
  if (!StartLxdProgressSignal::Status_IsValid(
          static_cast<int>(request->status()))) {
    return grpc::Status(grpc::FAILED_PRECONDITION,
                        "Invalid status field in protobuf request");
  }
  progress_signal.set_status(
      static_cast<StartLxdProgressSignal::Status>(request->status()));
  progress_signal.set_failure_reason(request->failure_reason());

  base::WaitableEvent event(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);
  bool result = false;
  task_runner_->PostTask(
      FROM_HERE, base::Bind(&vm_tools::cicerone::Service::StartLxdProgress,
                            service_, cid, &progress_signal, &result, &event));
  event.Wait();
  if (!result) {
    LOG(ERROR) << "Failure sending start lxd progress";
    return grpc::Status(grpc::FAILED_PRECONDITION, "Failure in StartLxdStatus");
  }

  return grpc::Status::OK;
}

// Returns 0 on failure, otherwise returns the 32-bit vsock cid.
uint32_t TremplinListenerImpl::ExtractCidFromPeerAddress(
    grpc::ServerContext* ctx) {
  uint32_t cid = 0;
  std::string peer_address = ctx->peer();
  {
    base::AutoLock lock_scope(testing_peer_address_lock_);
    if (!testing_peer_address_.empty()) {
      peer_address = testing_peer_address_;
    }
  }
  if (sscanf(peer_address.c_str(), "vsock:%" SCNu32, &cid) != 1) {
    LOG(WARNING) << "Failed to parse peer address " << peer_address;
    return 0;
  }
  return cid;
}

}  // namespace cicerone
}  // namespace vm_tools