proxy.cc

#include "proxy.h"

#include <cstdint>
#include <functional>
#include <memory>
#include <optional>
#include <string>
#include <utility>
#include <vector>

#include "absl/log/check.h"
#include "absl/log/log.h"
#include "absl/status/status.h"
#include "absl/status/statusor.h"
#include "absl/strings/str_cat.h"
#include "absl/strings/string_view.h"
#include "absl/synchronization/mutex.h"
#include "absl/time/time.h"
#include "absl/types/span.h"
#include "redfish_query_engine/file/uds.h"
#include "grpc/grpc.h"
#include "grpcpp/impl/service_type.h"
#include "grpcpp/security/auth_context.h"
#include "grpcpp/security/authorization_policy_provider.h"
#include "grpcpp/server_builder.h"
#include "grpcpp/support/status.h"
#include "metrics.h"
#include "proxy_config.pb.h"
#include "redfish_plugin.h"
#include "request_response.h"
#include "resource_authz.h"
#include "utils/status_macros.h"
#include "voyager/deferrable_priority_queue.hpp"
#include "voyager/priority_queue.hpp"

constexpr absl::string_view kUdsUrlBase = "unix://";

// Channel arguments to be added to the gRPC server builder.
// These are the default values in G3 go/keepalive-ping-grpc-core.
constexpr uint32_t kKeepAliveTimeMs = 10 * 60 * 1000 /*10 min*/;
constexpr uint32_t kKeepAliveTimeoutMs = 20 * 1000 /*20 sec*/;
constexpr uint32_t kKeepAlivePermitWithoutCalls = 1;
constexpr uint32_t kHttp2MaxPingWithoutData = 2;
constexpr uint32_t kHttp2MinRecvPingIntervalWithoutDataMs =
    10 * 1000 /*10 sec*/;
constexpr uint32_t kHttp2MaxPingStrikes = 2;

namespace milotic {

// Only intended for use with numeric types whose default value is 0
// Returns override_value if config_value is 0
template <typename T>
T GetConfigValue(T config_value, T override_value) {
  if (config_value == 0) return override_value;
  return config_value;
}

Proxy::Proxy(std::vector<Host> hosts,
             absl::Span<std::unique_ptr<RedfishPlugin>> handlers,
             voyager::DeferrablePriorityQueue* queue, Resources resources,
             milotic_grpc_proxy::AuthorizationPolicy authorization_policy,
             std::unique_ptr<PermissionChecker> permission_checker,
             std::string name)
    : hosts_(std::move(hosts)),
      queue_(queue),
      resources_(std::move(resources)),
      authorization_policy_(std::move(authorization_policy)),
      permission_checker_(std::move(permission_checker)) {
  CHECK(!hosts_.empty()) << "No hosts provided";
  name_ = name.empty() ? absl::StrCat("@", hosts_[0].GetHostname())
                       : std::move(name);
  handlers_.reserve(handlers.size());
  for (std::unique_ptr<RedfishPlugin>& handler : handlers) {
    absl::Status init_status = handler->Initialize(this);
    if (!init_status.ok()) {
      LOG(FATAL) << name_ << " - Handler init failed for " << handler->Name()
                 << ": " << init_status.message();
    }
    LOG(INFO) << name_ << " - Handler init succeeded: " << handler->Name();
    handlers_.push_back(std::move(handler));
  }
  handler_initialized_ = true;
}

Proxy::~Proxy() {
  if (server_ != nullptr) {
    server_->Shutdown();
    server_->Wait();
  }
}

RedfishPlugin::RequestAction Proxy::GetRequestAction(
    RedfishPlugin::RequestVerb request_verb, const ProxyRequest& request,
    const std::vector<std::pair<RedfishPlugin::RequestVerb, absl::string_view>>&
        paths_to_handle) {
  std::string request_path = std::string(request.GetPath());
  RedfishPlugin::RequestAction action_if_no_match =
      RedfishPlugin::RequestAction::kNext;
  for (const auto& [verb, path] : paths_to_handle) {
    if (request_path == path) {
      action_if_no_match = RedfishPlugin::RequestAction::kDrop;
      // Only handle requests for the correct verb
      if (request_verb == verb) return RedfishPlugin::RequestAction::kHandle;
    }
  }
  return action_if_no_match;
}

std::unique_ptr<ProxyRequest> Proxy::CreateRequest(
    absl::string_view redfish_id) const {
  auto request =
      std::make_unique<ProxyRequest>(hosts_, std::string(redfish_id));
  // TODO(jainrish): Header should be added by the caller based on the host
  // being used.

  //  NO_CDC: The size of hosts_ is safety-checked in the constructor.
  request->AddHeader("Host", hosts_[0].GetHostHeader());
  return request;
}

absl::StatusOr<RedfishPlugin*> Proxy::GetRequestHandler(
    RedfishPlugin::RequestVerb verb, ProxyRequest& request) {
  if (!handler_initialized_)
    return absl::UnavailableError("Handlers not initialized");

  for (std::unique_ptr<RedfishPlugin>& handler : handlers_) {
    RedfishPlugin::RequestAction action =
        handler->PreprocessRequest(verb, request);
    if (action == RedfishPlugin::RequestAction::kDrop) {
      return absl::UnavailableError("Request dropped");
    }
    if (action == RedfishPlugin::RequestAction::kHandle) {
      return handler.get();
    }
    if (action == RedfishPlugin::RequestAction::kNext) {
      // do nothing
    } else {
      return absl::UnknownError("Bad plugin action");
    }
  }
  return absl::UnimplementedError("Request not handled.");
}

absl::StatusOr<ProxyResponse> Proxy::DispatchRequest(
    RedfishPlugin::RequestVerb verb,
    std::unique_ptr<ProxyRequest> http_request) {
  ASSIGN_OR_RETURN(RedfishPlugin * handler,
                   GetRequestHandler(verb, *http_request));
  return handler->HandleRequest(verb, std::move(http_request));
}

absl::Status Proxy::Subscribe(std::unique_ptr<ProxyRequest> http_request,
                              RedfishPlugin::EventHandler* event_handler) {
  ASSIGN_OR_RETURN(
      RedfishPlugin * handler,
      GetRequestHandler(RedfishPlugin::RequestVerb::kSubscribe, *http_request));
  return handler->Subscribe(std::move(http_request), event_handler);
}

absl::StatusOr<std::unique_ptr<RedfishPlugin::EventHandler>>
Proxy::BidirectionalStream(std::unique_ptr<ProxyRequest> http_request,
                           RedfishPlugin::EventHandler* event_handler) {
  ASSIGN_OR_RETURN(
      RedfishPlugin * handler,
      GetRequestHandler(RedfishPlugin::RequestVerb::kBidirectionalStream,
                        *http_request));
  return handler->BidirectionalStream(std::move(http_request), event_handler);
}

absl::StatusOr<std::unique_ptr<Proxy::RequestJob>>
Proxy::DispatchRequestToQueue(RedfishPlugin::RequestVerb verb,
                              std::unique_ptr<ProxyRequest> http_request,
                              Proxy::RequestPriority priority) {
  if (queue_ == nullptr) {
    return absl::UnavailableError("Queue not initialized");
  }
  auto job = std::make_unique<RequestJob>(*this, verb, std::move(http_request));
  absl::Status result = queue_->Enqueue(*job, RequestJob::Run, priority);
  if (!result.ok()) {
    return result;
  }
  return job;
}

absl::Status Proxy::DispatchRequestToQueue(
    std::unique_ptr<RequestJob> request_job, Proxy::RequestPriority priority,
    std::optional<absl::Time> after) {
  if (queue_ == nullptr) {
    return absl::UnavailableError("Queue not initialized");
  }
  if (after) {
    return queue_->EnqueueAfter(*after, std::move(request_job), RequestJob::Run,
                                priority);
  } else {
    return queue_->Enqueue(std::move(request_job), RequestJob::Run, priority);
  }
}

absl::Status Proxy::ConfigGrpcAndStart(
    const milotic_grpc_proxy::GrpcConfiguration& config,
    const std::function<bool(const std::string&)>& is_safe_uds_root,
    const GrpcCredentials& credentials,
    const std::shared_ptr<
        grpc::experimental::AuthorizationPolicyProviderInterface>& provider) {
  using ecclesia::DomainSocketOwners;
  using ecclesia::DomainSocketPermissions;
  using ecclesia::SetUpUnixDomainSocket;

  grpc::ServerBuilder builder;

  if (provider != nullptr) {
    builder.experimental().SetAuthorizationPolicyProvider(provider);
  }

  if (config.has_net_endpoint()) {
    if (credentials.net_creds == nullptr) {
      return absl::FailedPreconditionError(
          absl::StrCat(name_, " - No network credentials provided"));
    }
    std::string net_endpoint = absl::StrCat(config.net_endpoint().host(), ":",
                                            config.net_endpoint().port());
    LOG(INFO) << name_ << " - Adding net endpoint " << net_endpoint;
    builder.AddListeningPort(net_endpoint, credentials.net_creds);
  }

  if (config.has_uds_endpoint()) {
    if (credentials.uds_creds == nullptr) {
      return absl::FailedPreconditionError(
          absl::StrCat(name_, " - No UDS credentials provided"));
    }
    DomainSocketOwners owners;
    DomainSocketPermissions permissions =
        config.uds_endpoint().permission() ==
                milotic_grpc_proxy::Permission::PERM_USER_ONLY
            ? DomainSocketPermissions::kUserOnly
            : DomainSocketPermissions::kUserAndGroup;

    if (config.uds_endpoint().has_user()) {
      owners.uid = config.uds_endpoint().user().id();
    }

    if (config.uds_endpoint().has_group()) {
      owners.gid = config.uds_endpoint().group().id();
    }

    if (!SetUpUnixDomainSocket(config.uds_endpoint().path(), permissions,
                               owners, is_safe_uds_root)) {
      return absl::UnavailableError(
          absl::StrCat(name_, " - Failed to set up socket at ",
                       config.uds_endpoint().path()));
    }
    LOG(INFO) << "Created socket file at " << config.uds_endpoint().path();
    std::string server_uds_path(
        absl::StrCat(kUdsUrlBase, config.uds_endpoint().path()));
    LOG(INFO) << name_ << " - Adding UDS endpoint " << server_uds_path;
    builder.AddListeningPort(server_uds_path, credentials.uds_creds);
  }
  for (std::unique_ptr<grpc::Service>& service : services_) {
    builder.RegisterService(service.get());
  }

  // Adding explicit ChannelArguments to prevent gRPC from using the default
  // values in OpenSource version.
  builder.AddChannelArgument(
      GRPC_ARG_KEEPALIVE_TIME_MS,
      GetConfigValue(config.channel_arguments().keepalive_time_ms(),
                     kKeepAliveTimeMs));
  builder.AddChannelArgument(
      GRPC_ARG_KEEPALIVE_TIMEOUT_MS,
      GetConfigValue(config.channel_arguments().keepalive_timeout_ms(),
                     kKeepAliveTimeoutMs));
  builder.AddChannelArgument(
      GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS,
      GetConfigValue(
          config.channel_arguments().keepalive_permit_without_calls(),
          kKeepAlivePermitWithoutCalls));
  builder.AddChannelArgument(
      GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA,
      GetConfigValue(config.channel_arguments().http2_max_pings_without_data(),
                     kHttp2MaxPingWithoutData));
  builder.AddChannelArgument(
      GRPC_ARG_HTTP2_MIN_RECV_PING_INTERVAL_WITHOUT_DATA_MS,
      GetConfigValue(config.channel_arguments()
                         .http2_min_recv_ping_interval_without_data_ms(),
                     kHttp2MinRecvPingIntervalWithoutDataMs));
  builder.AddChannelArgument(
      GRPC_ARG_HTTP2_MAX_PING_STRIKES,
      GetConfigValue(config.channel_arguments().http2_max_ping_strikes(),
                     kHttp2MaxPingStrikes));

  server_ = builder.BuildAndStart();
  if (server_ == nullptr) {
    return absl::InternalError(
        absl::StrCat(name_, " - Failed to create server"));
  }
  LOG(INFO) << "Proxy " << name_ << " started.";
  return absl::OkStatus();
}

absl::Status Proxy::RequestJob::RunImpl() {
  response_ = proxy_.DispatchRequest(verb_, std::move(request_));
  Handler handler;
  {
    absl::MutexLock lock(handler_mutex_);
    has_run_ = true;
    handler = std::move(handler_);
  }
  if (handler) {
    handler(response_);
  }
  return response_.status();
}

void Proxy::RequestJob::Handle(Handler handler) {
  if (absl::MutexLock lock(handler_mutex_); !has_run_) {
    handler_ = std::move(handler);
    return;
  }
  handler(response_);
}

absl::StatusOr<std::unique_ptr<ProxyRequest>> Proxy::CreateAuthorizedRequest(
    RedfishPlugin::RequestVerb verb, absl::string_view redfish_id,
    const grpc::AuthContext& auth_context) const {
  AuthorizationContext context{.grpc_context = auth_context,
                               .permission_checker = permission_checker_.get()};
  if (!authorization_policy_.keep_query_params()) {
    absl::string_view::size_type pos = redfish_id.find('?');
    if (pos != absl::string_view::npos) {
      redfish_id.remove_suffix(redfish_id.length() - pos);
      CommonMetrics::Get().query_parameters_removed.Increment({});
    }
  }
  absl::Status status =
      AuthorizeRequest(authorization_policy_, verb, redfish_id, context);
  if (!status.ok()) {
    LOG_EVERY_N_SEC(ERROR, 1) << name_ << " - Unauthorized request: " << status;
    return status;
  }
  return CreateRequest(redfish_id);
}

grpc::Status ConvertStatus(const absl::Status& status) {
  if (status.ok()) return grpc::Status::OK;
  if (status.code() == absl::StatusCode::kInvalidArgument) {
    return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
                        std::string(status.message()));
  }
  if (status.code() == absl::StatusCode::kNotFound) {
    return grpc::Status(grpc::StatusCode::NOT_FOUND,
                        std::string(status.message()));
  }
  if (status.code() == absl::StatusCode::kPermissionDenied) {
    return grpc::Status(grpc::StatusCode::PERMISSION_DENIED,
                        std::string(status.message()));
  }
  return grpc::Status(grpc::StatusCode::INTERNAL,
                      std::string(status.message()));
}

absl::Status Proxy::AddService(const google::protobuf::Any& config) {
  std::unique_ptr<::grpc::Service> service = CreateService(config);
  if (service == nullptr) {
    return absl::InvalidArgumentError(
        absl::StrCat("Failed to create service with config: ", config));
  }
  services_.push_back(std::move(service));
  return absl::OkStatus();
}

std::unique_ptr<grpc::Service> Proxy::CreateService(
    const google::protobuf::Any& config) {
  for (const auto& creator : service_creators()) {
    std::unique_ptr<grpc::Service> service = creator(*this, config);
    if (service != nullptr) {
      return service;
    }
  }
  return nullptr;
}

}  // namespace milotic