Skip to content

Commit

Permalink
Bug fix and minor changes
Browse files Browse the repository at this point in the history 
- Fixed bug in RawEventProcessor.cpp where it didn't handle ProcessTree null case
- Added auoms version field to all events
- Added SchemaVersion field to Text/Fluent/OMS output events
- Added support for inclusive (vs exclusive) field/record filtering
  • Loading branch information
@taglines
taglines committed Sep 2, 2021
1 parent 529db43 commit 1ea57d3
Show file tree
Hide file tree
Showing 9 changed files with 194 additions and 67 deletions.
58 changes: 36 additions & 22 deletions CollectionMonitor.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,22 +38,25 @@ void CollectionMonitor::run() {
Logger::Info("CollectionMonitor started");

if (_netlink.Open(nullptr) != 0) {
Logger::Error("AuditRulesMonitor: Could not open NETLINK connect, exiting");
Logger::Error("CollectionMonitor: Could not open NETLINK connect, exiting");
return;
}

do {
auto now = std::chrono::steady_clock::now();

uint32_t audit_pid = -1;
if (_pause_collector_check && now - _pause_time > std::chrono::seconds(3600)) {
_pause_collector_check = false;
}

uint32_t audit_pid = 0;
auto ret = NetlinkRetry([this,&audit_pid]() { return _netlink.AuditGetPid(audit_pid); });
if (ret != 0) {
// Treat NETLINK errors as unrecoverable.
if (!IsStopping()) {
Logger::Warn("CollectionMonitor: Failed to get audit pid from audit NETLINK: %s", std::strerror(-ret));
}
Logger::Info("CollectionMonitor stopping");
return;
audit_pid = 0;
}
if (!PathExists("/proc/"+std::to_string(audit_pid))) {
audit_pid = 0;
Expand All @@ -62,37 +65,45 @@ void CollectionMonitor::run() {
// Always get collector aliveness. This will ensure the child is reaped if it exits and won't be restarted.
bool is_alive = is_collector_alive();

if (!_disable_collector_check && !is_auditd_present() && !is_alive && audit_pid == 0) {
if (!_pause_collector_check && !is_auditd_present() && !is_alive && audit_pid == 0) {
start_collector();

while (audit_pid <= 0 && !_sleep(500) && std::chrono::steady_clock::now() - now < std::chrono::seconds(10)) {
int netlink_errno = 0;
while (!IsStopping() && audit_pid <= 0 && !_sleep(500) && std::chrono::steady_clock::now() - now < std::chrono::seconds(10)) {
auto ret = NetlinkRetry([this,&audit_pid]() { return _netlink.AuditGetPid(audit_pid); });
if (ret != 0) {
// Treat NETLINK errors as unrecoverable.
if (!IsStopping()) {
Logger::Warn("CollectionMonitor: Failed to get audit pid from audit NETLINK: %s", std::strerror(-ret));
netlink_errno = -ret;
}
Logger::Info("CollectionMonitor stopping");
return;
}
if (!PathExists("/proc/"+std::to_string(audit_pid))) {
audit_pid = 0;
} else {
netlink_errno = 0;
if (!PathExists("/proc/"+std::to_string(audit_pid))) {
audit_pid = 0;
}
}
}
if (IsStopping()) {
break;
}
if (audit_pid == 0) {
if (check_child(false)) {
Logger::Warn("CollectionMonitor: Collector has not set itself as the audit pid after 10 seconds");
if (netlink_errno != 0) {
Logger::Warn("CollectionMonitor: Failed to get audit pid from audit NETLINK: %s", std::strerror(netlink_errno));
} else {
if (audit_pid == 0) {
if (check_child(false)) {
Logger::Warn("CollectionMonitor: Collector has not set itself as the audit pid after 10 seconds");
}
}
}
}

if (audit_pid != _audit_pid || now - _last_audit_pid_report > std::chrono::seconds(3600)) {
_last_audit_pid_report = now;
_audit_pid = audit_pid;
send_audit_pid_report(audit_pid);
if (IsStopping()) {
if (audit_pid != _audit_pid || now - _last_audit_pid_report > std::chrono::seconds(3600)) {
_last_audit_pid_report = now;
_audit_pid = audit_pid;
send_audit_pid_report(audit_pid);
}
}
} while(!_sleep(10000));
Logger::Info("CollectionMonitor stopping");
Expand Down Expand Up @@ -135,7 +146,8 @@ bool CollectionMonitor::check_child(bool wait) {
auto ret = _collector.Wait(wait);
if (ret < 0) {
Logger::Warn("CollectionMonitor::check_child: waitpid() failed: %s", std::strerror(errno));
_disable_collector_check = true;
_pause_collector_check = true;
_pause_time = std::chrono::steady_clock::now();
return false;
} else if (ret == 1) {
report_proc_exit_status(_collector);
Expand All @@ -152,8 +164,9 @@ void CollectionMonitor::start_collector() {
}
// Disable collector start if num starts exceeds max allowed.
if (_collector_restarts.size() > MAX_COLLECTOR_RESTARTS) {
_disable_collector_check = true;
Logger::Warn("NETLINK collector started more than %d times in the last %d seconds. Collector will not be started again.", MAX_COLLECTOR_RESTARTS, COLLECTOR_RESTART_WINDOW);
_pause_collector_check = true;
_pause_time = std::chrono::steady_clock::now();
Logger::Warn("NETLINK collector started more than %d times in the last %d seconds. Collector will not be started again for one hour.", MAX_COLLECTOR_RESTARTS, COLLECTOR_RESTART_WINDOW);
return;
}
_collector_restarts.emplace(std::chrono::steady_clock::now());
Expand All @@ -172,7 +185,8 @@ void CollectionMonitor::signal_collector(int sig) {
// The child might have already died, so only report an error, if kill didn't return errno == ESRCH (process not found)
if(ret != 0 && ret != -ESRCH) {
Logger::Warn("CollectionMonitor: kill(%d, %d) failed: %s", _collector.Pid(), sig, std::strerror(errno));
_disable_collector_check = true;
_pause_collector_check = true;
_pause_time = std::chrono::steady_clock::now();
}
}
}
Expand Down
5 changes: 3 additions & 2 deletions CollectionMonitor.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ class CollectionMonitor: public RunBase {
const std::string& collector_config_path)
: _builder(std::make_shared<EventQueue>(std::move(queue)), nullptr),
_auditd_path(auditd_path), _collector_path(collector_path), _collector_config_path(collector_config_path),
_collector(collector_path, collector_args(collector_config_path), Cmd::PIPE_STDIN), _audit_pid(0), _disable_collector_check(false), _last_audit_pid_report(), _collector_restarts() {}
_collector(collector_path, collector_args(collector_config_path), Cmd::PIPE_STDIN), _audit_pid(0), _pause_collector_check(false), _pause_time(), _last_audit_pid_report(), _collector_restarts() {}

protected:
void run() override;
Expand Down Expand Up @@ -69,7 +69,8 @@ class CollectionMonitor: public RunBase {
std::string _collector_config_path;
Cmd _collector;
uint32_t _audit_pid;
bool _disable_collector_check;
bool _pause_collector_check;
std::chrono::steady_clock::time_point _pause_time;
std::chrono::steady_clock::time_point _last_audit_pid_report;
std::set<std::chrono::steady_clock::time_point> _collector_restarts;
};
Expand Down
51 changes: 39 additions & 12 deletions RawEventProcessor.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
#include "Translate.h"
#include "Interpret.h"
#include "StringUtils.h"
#include "auoms_version.h"

#include <climits>
#include <algorithm>
Expand Down Expand Up @@ -63,10 +64,14 @@ void RawEventProcessor::ProcessData(const void* data, size_t data_len) {

void RawEventProcessor::process_event(const Event& event) {

using namespace std::string_literals;
using namespace std::string_view_literals;

static auto S_PID = "pid"s;
static auto S_PPID = "ppid"s;
static auto SV_PID = "pid"sv;
static auto SV_PPID = "ppid"sv;
static auto SV_CONTAINERID = "containerid"sv;
static auto SV_AUOMSVERSION_NAME = "auoms_version"sv;
static std::string S_AUOMS_VERSION = AUOMS_VERSION;
static std::string_view SV_AUOMS_VERSION = S_AUOMS_VERSION;

if (!_builder->BeginEvent(event.Seconds(), event.Milliseconds(), event.Serial(), event.NumRecords())) {
throw std::runtime_error("Queue closed");
Expand All @@ -81,25 +86,31 @@ void RawEventProcessor::process_event(const Event& event) {
if (static_cast<RecordType>(rec.RecordType()) == RecordType::USER_CMD) {
process_user_cmd_record(event, rec);
} else {
auto pid_field = rec.FieldByName(S_PID);
int num_fields = rec.NumFields();
auto pid_field = rec.FieldByName(SV_PID);
int num_fields = rec.NumFields() + 1;
if (pid_field) {
num_fields++;
}
if (!_builder->BeginRecord(rec.RecordType(), rec.RecordTypeName(), rec.RecordText(), num_fields)) {
throw std::runtime_error("Queue closed");
}

if (!_builder->AddField(SV_AUOMSVERSION_NAME, SV_AUOMS_VERSION, nullptr, field_type_t::UNCLASSIFIED)) {
throw std::runtime_error("Queue closed");
}

std::string containerId = "";
if (pid_field) {
_pid = atoi(pid_field.RawValuePtr());
_builder->SetEventPid(_pid);
auto p = _processTree->GetInfoForPid(_pid);
if (p) {
containerId = p->containerid();
if (_processTree) {
auto p = _processTree->GetInfoForPid(_pid);
if (p) {
containerId = p->containerid();
}
}
}
auto ppid_field = rec.FieldByName(S_PPID);
auto ppid_field = rec.FieldByName(SV_PPID);
if (ppid_field) {
_ppid = atoi(ppid_field.RawValuePtr());
}
Expand All @@ -111,7 +122,7 @@ void RawEventProcessor::process_event(const Event& event) {
}
}
if (pid_field) {
if (!_builder->AddField("containerid", containerId, nullptr, field_type_t::UNCLASSIFIED)) {
if (!_builder->AddField(SV_CONTAINERID, containerId, nullptr, field_type_t::UNCLASSIFIED)) {
throw std::runtime_error("Queue closed");
}
}
Expand Down Expand Up @@ -165,6 +176,9 @@ bool RawEventProcessor::process_syscall_event(const Event& event) {
static auto auoms_syscall_name = RecordTypeToName(RecordType::AUOMS_SYSCALL);
static auto auoms_syscall_fragment_name = RecordTypeToName(RecordType::AUOMS_SYSCALL_FRAGMENT);
static auto auoms_execve_name = RecordTypeToName(RecordType::AUOMS_EXECVE);
static auto SV_AUOMSVERSION_NAME = "auoms_version"sv;
static std::string S_AUOMS_VERSION = AUOMS_VERSION;
static std::string_view SV_AUOMS_VERSION = S_AUOMS_VERSION;

int num_fields = 0;
int num_path = 0;
Expand Down Expand Up @@ -392,8 +406,8 @@ bool RawEventProcessor::process_syscall_event(const Event& event) {
return false;
}

// For containerid
num_fields += 1;
// For containerid and auoms_version
num_fields += 2;

if (!_builder->BeginEvent(event.Seconds(), event.Milliseconds(), event.Serial(), 1)) {
throw std::runtime_error("Queue closed");
Expand All @@ -404,6 +418,10 @@ bool RawEventProcessor::process_syscall_event(const Event& event) {
throw std::runtime_error("Queue closed");
}

if (!_builder->AddField(SV_AUOMSVERSION_NAME, SV_AUOMS_VERSION, nullptr, field_type_t::UNCLASSIFIED)) {
throw std::runtime_error("Queue closed");
}

if (syscall_rec) {
for (auto &f : syscall_rec) {
auto fname = f.FieldName();
Expand Down Expand Up @@ -745,17 +763,26 @@ void RawEventProcessor::process_user_cmd_record(const Event& event, const EventR
static auto S_PPID = "ppid"s;
static auto SV_CMD = "cmd"sv;
static auto SV_REDACTORS = "redactors"sv;
static auto SV_AUOMSVERSION_NAME = "auoms_version"sv;
static std::string S_AUOMS_VERSION = AUOMS_VERSION;
static std::string_view SV_AUOMS_VERSION = S_AUOMS_VERSION;

int num_fields = rec.NumFields();

if (rec.FieldByName(SV_CMD)) {
num_fields += 1;
}

num_fields += 1; // for auoms_version

if (!_builder->BeginRecord(rec.RecordType(), rec.RecordTypeName(), nullptr, num_fields)) {
throw std::runtime_error("Queue closed");
}

if (!_builder->AddField(SV_AUOMSVERSION_NAME, SV_AUOMS_VERSION, nullptr, field_type_t::UNCLASSIFIED)) {
throw std::runtime_error("Queue closed");
}

auto pid_field = rec.FieldByName(S_PID);
if (pid_field) {
_pid = atoi(pid_field.RawValuePtr());
Expand Down
Loading

0 comments on commit 1ea57d3

Please sign in to comment.