cmd_sub.cc

Go to the documentation of this file.

#ifndef __STDC_FORMAT_MACROS
// NOLINTNEXTLINE
#define __STDC_FORMAT_MACROS
#endif

#include "cmd_sub.h"

#include <inttypes.h>

#include "crypto/signature.h"
#include "manifest.h"
#include "manifest_fetch.h"
#include "network/download.h"
#include "notify/messages.h"
#include "options.h"
#include "subscriber_sse.h"
#include "subscriber_supervisor.h"
#include "supervisor.h"
#include "util/logging.h"
#include "util/pointer.h"
#include "util/posix.h"
#include "util/string.h"

namespace {

const LogFacilities& kLogInfo = DefaultLogging::info;
const LogFacilities& kLogError = DefaultLogging::error;

const int kMaxPoolHandles = 1;

class SwissknifeSubscriber : public notify::SubscriberSSE {
 public:
  SwissknifeSubscriber(const std::string& server_url,
                       const std::string& repository, uint64_t min_revision,
                       bool continuous, bool verbose)
      : notify::SubscriberSSE(server_url),
        repository_(repository),
        stats_(),
        dl_mgr_(new download::DownloadManager(kMaxPoolHandles,
                                perf::StatisticsTemplate("download", &stats_))),
        sig_mgr_(new signature::SignatureManager()),
        revision_(min_revision),
        continuous_(continuous),
        verbose_(verbose) {}
  virtual ~SwissknifeSubscriber() {
    sig_mgr_->Fini();
  }

  bool Init() {
    const std::string config_file =
        "/etc/cvmfs/repositories.d/" + repository_ + "/client.conf";
    SimpleOptionsParser options;
    if (!options.TryParsePath(config_file)) {
      LogCvmfs(kLogCvmfs, kLogError,
               "SwissknifeSubscriber - could not parse configuration file");
      return false;
    }

    std::string arg;
    if (options.GetValue("CVMFS_SERVER_URL", &arg)) {
      dl_mgr_->SetHostChain(arg);
    }

    sig_mgr_->Init();

    std::string public_keys =
        JoinStrings(FindFilesBySuffix("/etc/cvmfs/keys", ".pub"), ":");
    if (!sig_mgr_->LoadPublicRsaKeys(public_keys)) {
      LogCvmfs(kLogCvmfs, kLogError,
               "SwissknifeSubscriber - could not load public keys");
      return false;
    }

    return true;
  }

 private:
  virtual notify::Subscriber::Status Consume(const std::string& repo,
                                             const std::string& msg_text) {
    notify::msg::Activity msg;
    if (!msg.FromJSONString(msg_text)) {
      LogCvmfs(kLogCvmfs, kLogError,
               "SwissknifeSubscriber - could not decode message.");
      return notify::Subscriber::kError;
    }

    manifest::ManifestEnsemble ensemble;
    manifest::Failures res = manifest::Verify(
                          reinterpret_cast<unsigned char*>(&(msg.manifest_[0])),
                          msg.manifest_.size(), "", repo, 0, NULL,
                          sig_mgr_.weak_ref(), dl_mgr_.weak_ref(), &ensemble);

    if (res != manifest::kFailOk) {
      LogCvmfs(kLogCvmfs, kLogError,
               "SwissknifeSubscriber - manifest has invalid signature: %d",
               res);
      return notify::Subscriber::kError;
    }

    const UniquePtr<manifest::Manifest> manifest(manifest::Manifest::LoadMem(
        reinterpret_cast<const unsigned char*>(msg.manifest_.data()),
        msg.manifest_.size()));

    if (!manifest.IsValid()) {
      LogCvmfs(kLogCvmfs, kLogError,
               "SwissknifeSubscriber - could not parse manifest.");
      return notify::Subscriber::kError;
    }

    uint64_t new_revision = manifest->revision();
    bool triggered = false;
    if (new_revision > revision_) {
      LogCvmfs(kLogCvmfs, kLogInfo,
               "SwissknifeSubscriber - repository %s is now at revision %"
               PRIu64 ".", repo.c_str(), new_revision);
      if (verbose_) {
        LogCvmfs(kLogCvmfs, kLogInfo, "%s", msg_text.c_str());
      }
      revision_ = new_revision;
      triggered = true;
    }

    if (!continuous_ && triggered) {
      return notify::Subscriber::kFinish;
    }

    return notify::Subscriber::kContinue;
  }

  std::string repository_;

  perf::Statistics stats_;
  UniquePtr<download::DownloadManager> dl_mgr_;
  UniquePtr<signature::SignatureManager> sig_mgr_;

  uint64_t revision_;
  bool continuous_;
  bool verbose_;
};

}  // namespace

namespace notify {

int DoSubscribe(const std::string& server_url, const std::string& repo,
                uint64_t min_revision, bool continuous, bool verbose) {
  SwissknifeSubscriber subscriber(server_url, repo, min_revision, continuous,
                                  verbose);

  if (!subscriber.Init()) {
    LogCvmfs(kLogCvmfs, kLogError, "Could not initialize SwissknifeSubscriber");
    return 1;
  }

  // Retry settings: accept no more than 10 failures in the last minute
  const int num_retries = 10;
  const uint64_t interval = 60;
  SubscriberSupervisor supervisor(&subscriber, repo, num_retries, interval);
  supervisor.Run();

  return 0;
}

}  // namespace notify