osquery监控系统解密

$ ps -ef | grep osqueryd                     
root      28232      1  0 16:17 ?        00:00:00 /usr/bin/osqueryd --flagfile /etc/osquery/osquery.flags --config_path /etc/osquery/osquery.conf
root      28259  28232  2 16:17 ?        00:00:00 /usr/bin/osqueryd

using WatchdogLimitMap = std::map<WatchdogLimitType, LimitDefinition>;

struct LimitDefinition {
  size_t normal;
  size_t restrictive;
  size_t disabled;
};

const WatchdogLimitMap kWatchdogLimits = {
    // Maximum MB worker can privately allocate.
    {WatchdogLimitType::MEMORY_LIMIT, {200, 100, 10000}},
    // % of (User + System + Idle) CPU time worker can utilize
    // for LATENCY_LIMIT seconds.
    {WatchdogLimitType::UTILIZATION_LIMIT, {10, 5, 100}},
    // Number of seconds the worker should run, else consider the exit fatal.
    {WatchdogLimitType::RESPAWN_LIMIT, {4, 4, 1000}},
    // If the worker respawns too quickly, backoff on creating additional.
    {WatchdogLimitType::RESPAWN_DELAY, {5, 5, 1}},
    // Seconds of tolerable UTILIZATION_LIMIT sustained latency.
    {WatchdogLimitType::LATENCY_LIMIT, {12, 6, 1000}},
    // How often to poll for performance limit violations.
    {WatchdogLimitType::INTERVAL, {3, 3, 3}},
};

size_t getWorkerLimit(WatchdogLimitType name) {
    if (kWatchdogLimits.count(name) == 0) {
        return 0;
    }

    if (name == WatchdogLimitType::MEMORY_LIMIT &&
        FLAGS_watchdog_memory_limit > 0) {
        return FLAGS_watchdog_memory_limit;
    }

    if (name == WatchdogLimitType::UTILIZATION_LIMIT &&
        FLAGS_watchdog_utilization_limit > 0) {
        return FLAGS_watchdog_utilization_limit;
    }

    auto level = FLAGS_watchdog_level;
    // If no level was provided then use the default (config/switch).
    if (level == -1) {
        return kWatchdogLimits.at(name).disabled;
    }

    if (level == 1) {
        return kWatchdogLimits.at(name).restrictive;
    }
    return kWatchdogLimits.at(name).normal;
}

void WatcherRunner::start() {
    // Hold the current process (watcher) for inspection too.
    auto &watcher = Watcher::get();
    auto self = PlatformProcess::getCurrentProcess();

    // Set worker performance counters to an initial state.
    watcher.resetWorkerCounters(0);
    PerformanceState watcher_state;

    // Enter the watch loop.
    do {
        if (use_worker_ && !watch(watcher.getWorker())) {
            if (watcher.fatesBound()) {
                // A signal has interrupted the watcher.
                break;
            }

            auto status = watcher.getWorkerStatus();
            if (status == EXIT_CATASTROPHIC) {
                Initializer::requestShutdown(EXIT_CATASTROPHIC);
                break;
            }

            if (watcher.workerRestartCount() ==
                getWorkerLimit(WatchdogLimitType::RESPAWN_LIMIT)) {
                // Too many worker restarts.
                Initializer::requestShutdown(EXIT_FAILURE, "Too many worker restarts");
                break;
            }

            // The watcher failed, create a worker.
            createWorker();
        }

        // After inspecting the worker, check the extensions.
        // Extensions may be active even if a worker/watcher is not used.
        watchExtensions();

        if (use_worker_) {
            auto status = isWatcherHealthy(*self, watcher_state);
            if (!status.ok()) {
                Initializer::requestShutdown(
                        EXIT_CATASTROPHIC,
                        "Watcher has become unhealthy: " + status.getMessage());
                break;
            }
        }

        if (run_once_) {
            // A test harness can end the thread immediately.
            break;
        }
        pause(std::chrono::seconds(getWorkerLimit(WatchdogLimitType::INTERVAL)));
    } while (!interrupted() && ok());
}

void WatcherRunner::createWorker() {
    auto &watcher = Watcher::get();
    /**
    * init check
    */
    
    // Get the complete path of the osquery process binary.
    boost::system::error_code ec;
    auto exec_path = fs::system_complete(fs::path(qd[0]["path"]), ec);
    if (!pathExists(exec_path).ok()) {
        LOG(WARNING) << "osqueryd doesn't exist in: " << exec_path.string();
        return;
    }
    if (!safePermissions(exec_path.parent_path().string(), exec_path.string(), true)) {
        // osqueryd binary has become unsafe.
        LOG(ERROR) << RLOG(1382) << "osqueryd has unsafe permissions: " << exec_path.string();
        Initializer::requestShutdown(EXIT_FAILURE);
        return;
    }

    auto worker = PlatformProcess::launchWorker(exec_path.string(), argc_, argv_);
    if (worker == nullptr) {
        // Unrecoverable error, cannot create a worker process.
        LOG(ERROR) << "osqueryd could not create a worker process";
        Initializer::shutdown(EXIT_FAILURE);
        return;
    }

    watcher.setWorker(worker);
    watcher.resetWorkerCounters(getUnixTime());
    VLOG(1) << "osqueryd watcher (" << PlatformProcess::getCurrentPid()
            << ") executing worker (" << worker->pid() << ")";
    watcher.worker_status_ = -1;
}

bool WatcherRunner::watch(const PlatformProcess &child) const {
    int process_status = 0;
    ProcessState result = child.checkStatus(process_status);
    if (Watcher::get().fatesBound()) {
        // A signal was handled while the watcher was watching.
        return false;
    }

    if (!child.isValid() || result == PROCESS_ERROR) {
        // Worker does not exist or never existed.
        return false;
    } else if (result == PROCESS_STILL_ALIVE) {
        // If the inspect finds problems it will stop/restart the worker.
        auto status = isChildSane(child);
        // A delayed watchdog does not stop the worker process.
        if (!status.ok() && getUnixTime() >= delayedTime()) {
            stopChild(child);
            return false;
        }
        return true;
    }

    if (result == PROCESS_EXITED) {
        // If the worker process existed, store the exit code.
        Watcher::get().worker_status_ = process_status;
        return false;
    }

    return true;
}

Status WatcherRunner::isChildSane(const PlatformProcess &child) const {
    auto rows = getProcessRow(child.pid());
    if (rows.size() == 0) {
        // Could not find worker process?
        return Status(1, "Cannot find process");
    }

    PerformanceChange change;
    {
        WatcherExtensionsLocker locker;
        auto &state = Watcher::get().getState(child);
        change = getChange(rows[0], state);
    }

    // Only make a decision about the child sanity if it is still the watcher's
    // child. It's possible for the child to die, and its pid reused.
    if (change.parent != PlatformProcess::getCurrentPid()) {
        // The child's parent is not the watcher.
        Watcher::get().reset(child);
        // Do not stop or call the child insane, since it is not our child.
        return Status(0);
    }

    if (exceededCyclesLimit(change)) {
        return Status(1,"Maximum sustainable CPU utilization limit exceeded: " + std::to_string(change.sustained_latency * change.iv));
    }

    // Check if the private memory exceeds a memory limit.
    if (exceededMemoryLimit(change)) {
        return Status(1, "Memory limits exceeded: " + std::to_string(change.footprint));
    }

    // The worker is sane, no action needed.
    // Attempt to flush status logs to the well-behaved worker.
    if (use_worker_ && child.pid() == Watcher::get().getWorker().pid()) {
        relayStatusLogs();
    }

    return Status(0);
}

PerformanceChange getChange(const Row &r, PerformanceState &state) {
    PerformanceChange change;

    // IV is the check interval in seconds, and utilization is set per-second.
    change.iv = std::max(getWorkerLimit(WatchdogLimitType::INTERVAL), 1_sz); // 3
    long long user_time = 0, system_time = 0;
    try {
        change.parent =
                static_cast<pid_t>(tryTo<long long>(r.at("parent")).takeOr(0LL));
        user_time = tryTo<long long>(r.at("user_time")).takeOr(0LL);
        system_time = tryTo<long long>(r.at("system_time")).takeOr(0LL);
        change.footprint = tryTo<long long>(r.at("resident_size")).takeOr(0LL);
    } catch (const std::exception & /* e */) {
        state.sustained_latency = 0;
    }

    // Check the difference of CPU time used since last check.
    auto percent_ul = getWorkerLimit(WatchdogLimitType::UTILIZATION_LIMIT);
    percent_ul = (percent_ul > 100) ? 100 : percent_ul;

    UNSIGNED_BIGINT_LITERAL iv_milliseconds = change.iv * 1000;  //3*1000
    // 此时cpu_ul = (10*3000*1)/100=300
    UNSIGNED_BIGINT_LITERAL cpu_ul =
            (percent_ul * iv_milliseconds * kNumOfCPUs) / 100;

    auto user_time_diff = user_time - state.user_time;
    auto sys_time_diff = system_time - state.system_time;
    auto cpu_utilization_time = user_time_diff + sys_time_diff;

    if (cpu_utilization_time > cpu_ul) {
        state.sustained_latency++;
    } else {
        state.sustained_latency = 0;
    }
    // Update the current CPU time.
    state.user_time = user_time;
    state.system_time = system_time;

    // Check if the sustained difference exceeded the acceptable latency limit.
    change.sustained_latency = state.sustained_latency;

    // Set the memory footprint as the amount of resident bytes allocated
    // since the process image was created (estimate).
    // A more-meaningful check would limit this to writable regions.
    if (state.initial_footprint == 0) {
        state.initial_footprint = change.footprint;
    }

    // Set the measured/limit-applied footprint to the post-launch allocations.
    if (change.footprint < state.initial_footprint) {
        change.footprint = 0;
    } else {
        change.footprint = change.footprint - state.initial_footprint;
    }

    return change;
}

change.footprint = tryTo<long long>(r.at("resident_size")).takeOr(0LL);
.....
if (state.initial_footprint == 0) {
    state.initial_footprint = change.footprint;
}

// Set the measured/limit-applied footprint to the post-launch allocations.
if (change.footprint < state.initial_footprint) {
    change.footprint = 0;
} else {
    change.footprint = change.footprint - state.initial_footprint;
}

// cpu_ul的默认设置
UNSIGNED_BIGINT_LITERAL cpu_ul = (percent_ul * iv_milliseconds * kNumOfCPUs) / 100;
// 得到当前状态的在用户态下消耗的时间和在内核态下消耗的时间
user_time = tryTo<long long>(r.at("user_time")).takeOr(0LL);
system_time = tryTo<long long>(r.at("system_time")).takeOr(0LL);

auto user_time_diff = user_time - state.user_time;
auto sys_time_diff = system_time - state.system_time;
// cpu_utilization_time 就是当前状态与上次状态的内核态时间和用户态时间的差值总和
auto cpu_utilization_time = user_time_diff + sys_time_diff;

// 如果超过了cpu_ul的默认设置，则sustained_latency的数量加1，为什么使用的是state.sustained_latency?因为state.sustained_latency可能之前就不为空，之前就有可能超过了预设的设置
if (cpu_utilization_time > cpu_ul) {
    state.sustained_latency++;
} else {
    state.sustained_latency = 0;
}
// 将state.sustained_latency赋值给change.sustained_latency，用于之后的CPU资源使用的分析判断。
change.sustained_latency = state.sustained_latency;、

auto status = isChildSane(child);
// A delayed watchdog does not stop the worker process.
if (!status.ok() && getUnixTime() >= delayedTime()) {
    // Since the watchdog cannot use the logger plugin the error message
    // should be logged to stderr and to the system log.
    std::stringstream error;
    error << "osqueryd worker (" << child.pid()<< ") stopping: " << status.getMessage();
    systemLog(error.str());
    LOG(WARNING) << error.str();
    stopChild(child);
    return false;
}
return true;

void WatcherRunner::stopChild(const PlatformProcess &child) const {
    child.killGracefully();

    // Clean up the defunct (zombie) process.
    if (!child.cleanup()) {
        auto child_pid = child.pid();

        LOG(WARNING) << "osqueryd worker (" << std::to_string(child_pid)
                        << ") could not be stopped. Sending kill signal.";

        child.kill();
        if (!child.cleanup()) {
            auto message = std::string("Watcher cannot stop worker process (") +
                            std::to_string(child_pid) + ").";
            Initializer::requestShutdown(EXIT_CATASTROPHIC, message);
        }
    }
}

`osquery/core/posix/process.cpp`
bool PlatformProcess::killGracefully() const {
  if (!isValid()) {
    return false;
  }

  // 关于kill进程的用法：http://man7.org/linux/man-pages/man2/kill.2.html
  // In the case of SIGCONT, it suffices when the sending and receiving processes belong to the same session
  int status = ::kill(nativeHandle(), SIGTERM);
  return (status == 0);
}

do {
    if (use_worker_ && !watch(watcher.getWorker())) {
        .......
        createWorker();
    }
    .....
} while (!interrupted() && ok());