From 6f661cac31eb271f93f195d81d4914daa83f732a Mon Sep 17 00:00:00 2001 From: "timmy.yuan" Date: Thu, 8 Oct 2026 11:28:05 +0800 Subject: [PATCH] fix(cpworker): write log lines atomically; take config errors from the caller's parse end log_log wrote each line with three stdio calls, so lines from concurrent threads interleaved, and localtime() shared one static struct tm. The line is now formatted into one buffer and written with a single fwrite, and the time comes from localtime_r. Config parse errors were read back with cJSON_GetErrorPtr(), whose process-wide position the control thread resets on every handshake or command it parses. The reload error could then read "before: (null)" or quote the control thread's buffer. cjson_parse() takes the position from its own parse end and reports "at line L, column C near ''", with the snippet kept on one line. cJSON itself still writes its global error position from both threads; nothing reads it. read_file_contents sized the buffer with fseek/ftell, which on a directory gives 0 or LONG_MAX, so "-c " failed with "empty file" or an overflowing malloc. It now checks the file with fstat and reports "not a regular file", and caps the size at 64 MiB. --- cpworker/src/cjson_utils.c | 37 +++++++ cpworker/src/cjson_utils.h | 5 + cpworker/src/config.c | 45 +++++--- cpworker/src/log.c | 60 +++++++++-- cpworker/tests/unit/CMakeLists.txt | 2 + cpworker/tests/unit/config_file.c | 166 +++++++++++++++++++++++++++++ cpworker/tests/unit/logging.c | 153 ++++++++++++++++++++++++++ 7 files changed, 446 insertions(+), 22 deletions(-) create mode 100644 cpworker/tests/unit/config_file.c create mode 100644 cpworker/tests/unit/logging.c diff --git a/cpworker/src/cjson_utils.c b/cpworker/src/cjson_utils.c index 5e2b1a38..c5c32382 100644 --- a/cpworker/src/cjson_utils.c +++ b/cpworker/src/cjson_utils.c @@ -7,6 +7,43 @@ #include "cjson_utils.h" +#define CJSON_ERROR_SNIPPET_LEN 32 + +cJSON *cjson_parse(const char *json, cJSONParseError *err) +{ + const char *end = NULL; + cJSON *item = cJSON_ParseWithOpts(json, &end, false); + if (item) + return item; + + if (!json || !end) + { + cjson_set_parse_error(err, "JSON parse error"); + return NULL; + } + + int line = 1, column = 1; + for (const char *p = json; p < end; p++) + { + if (*p == '\n') + { + line++; + column = 1; + } + else + column++; + } + + // The snippet stays on the error's line, so the message remains a single log line. + int snippet_len = 0; + while (snippet_len < CJSON_ERROR_SNIPPET_LEN && end[snippet_len] != '\0' && end[snippet_len] != '\n' && + end[snippet_len] != '\r') + snippet_len++; + + cjson_set_parse_error(err, "JSON parse error at line %d, column %d near '%.*s'", line, column, snippet_len, end); + return NULL; +} + void cjson_set_parse_error(cJSONParseError *err, const char *format, ...) { if (!err) diff --git a/cpworker/src/cjson_utils.h b/cpworker/src/cjson_utils.h index 97e420f5..fa392379 100644 --- a/cpworker/src/cjson_utils.h +++ b/cpworker/src/cjson_utils.h @@ -22,6 +22,11 @@ typedef struct char message[CJSON_ERRBUF_SIZE]; } cJSONParseError; +// Parses the NUL-terminated `json`. On a syntax error, returns NULL and sets err to +// "JSON parse error at line L, column C near ''". The position comes from this call's own parse end, not from +// cJSON_GetErrorPtr(), whose process-wide state another thread's parse can overwrite. +cJSON *cjson_parse(const char *json, cJSONParseError *err); + void cjson_set_parse_error(cJSONParseError *err, const char *format, ...); void cjson_wrap_parse_error(cJSONParseError *err, const char *format, ...); diff --git a/cpworker/src/config.c b/cpworker/src/config.c index b909479f..1f408c31 100644 --- a/cpworker/src/config.c +++ b/cpworker/src/config.c @@ -1,8 +1,10 @@ +#include #include #include #include #include #include +#include #include "cJSON/cJSON.h" @@ -20,6 +22,9 @@ // pcap_set_buffer_size() takes an int byte count; this is the largest whole-MB value that fits. #define MAX_LIBPCAP_BUFFER_SIZE_MB (INT_MAX / 1024 / 1024) +// A config file is a few KiB; the cap only keeps a wrong path from being read into memory whole. +#define CONFIG_FILE_MAX_SIZE (64 * 1024 * 1024) + // Parses a capturer's snaplen. Values above MAX_SNAPLEN become MAX_SNAPLEN. Values <= 0 also become MAX_SNAPLEN // ("no truncation") when `non_positive_means_max`, and are rejected otherwise. static int parse_snaplen(cJSON *obj, const char *name, bool non_positive_means_max, int *out, cJSONParseError *err) @@ -1218,16 +1223,35 @@ static char *read_file_contents(const char *filename, cJSONParseError *error) return NULL; } - fseek(fp, 0, SEEK_END); - long length = ftell(fp); - fseek(fp, 0, SEEK_SET); - - if (length <= 0) + // fopen succeeds on a directory, and seeking one does not yield a usable size, so the type and size come from + // fstat instead. + struct stat st; + if (fstat(fileno(fp), &st) != 0) + { + cjson_set_parse_error(error, "failed to stat file: %s: %s", filename, strerror(errno)); + fclose(fp); + return NULL; + } + if (!S_ISREG(st.st_mode)) + { + cjson_set_parse_error(error, "not a regular file: %s", filename); + fclose(fp); + return NULL; + } + if (st.st_size == 0) { cjson_set_parse_error(error, "empty file: %s", filename); fclose(fp); return NULL; } + if (st.st_size > CONFIG_FILE_MAX_SIZE) + { + cjson_set_parse_error(error, "file too large: %s (%lld bytes, limit %d)", filename, (long long)st.st_size, + CONFIG_FILE_MAX_SIZE); + fclose(fp); + return NULL; + } + size_t length = (size_t)st.st_size; char *buffer = (char *)malloc(length + 1); if (!buffer) @@ -1237,7 +1261,7 @@ static char *read_file_contents(const char *filename, cJSONParseError *error) return NULL; } - if (fread(buffer, 1, length, fp) != (size_t)length) + if (fread(buffer, 1, length, fp) != length) { cjson_set_parse_error(error, "partial read of file: %s", filename); free(buffer); @@ -1256,10 +1280,9 @@ TasksAllConfig *parse_tasks_file(const char *filename, cJSONParseError *err) if (!json_str) return NULL; - cJSON *json = cJSON_Parse(json_str); + cJSON *json = cjson_parse(json_str, err); if (!json) { - cjson_set_parse_error(err, "JSON parse error before: %s", cJSON_GetErrorPtr()); free(json_str); return NULL; } @@ -1473,10 +1496,9 @@ static Config *parse_config_json(cJSON *json, cJSONParseError *err) Config *parse_config_data(const char *json_str, cJSONParseError *err) { - cJSON *json = cJSON_Parse(json_str); + cJSON *json = cjson_parse(json_str, err); if (!json) { - cjson_set_parse_error(err, "JSON parse error before: %s", cJSON_GetErrorPtr()); return NULL; } @@ -1491,10 +1513,9 @@ Config *parse_config_file(const char *filename, cJSONParseError *err) if (!json_str) return NULL; - cJSON *json = cJSON_Parse(json_str); + cJSON *json = cjson_parse(json_str, err); if (!json) { - cjson_set_parse_error(err, "JSON parse error before: %s", cJSON_GetErrorPtr()); free(json_str); return NULL; } diff --git a/cpworker/src/log.c b/cpworker/src/log.c index 59725661..7a2d8d71 100644 --- a/cpworker/src/log.c +++ b/cpworker/src/log.c @@ -1,9 +1,13 @@ #include #include +#include #include #include "log.h" +// Lines that fit are formatted on the stack; longer ones go to the heap, or are truncated if that fails. +#define LOG_LINE_STACK_SIZE 1024 + static struct { int level; @@ -15,17 +19,52 @@ const char *log_level_string(int level) { return level_strings[level]; } void log_set_level(int level) { L.level = level; } +// Several threads log at once, and stdio locks each call rather than each line. The whole line is therefore +// formatted first and written with a single fwrite, so lines from different threads never interleave. static void stderr_callback(log_event_t *ev) { - char buf[64]; - if (ev->time) - buf[strftime(buf, sizeof(buf), "%Y-%m-%dT%H:%M:%S", ev->time)] = '\0'; - else - snprintf(buf, sizeof(buf), ""); - fprintf(stderr, "%s %-5s %s:%d: ", buf, log_level_string(ev->level), ev->file, ev->line); - vfprintf(stderr, ev->fmt, ev->ap); - fprintf(stderr, "\n"); + char stack_line[LOG_LINE_STACK_SIZE]; + char *line = stack_line; + size_t size = sizeof(stack_line); + + char ts[32]; + if (!ev->time || strftime(ts, sizeof(ts), "%Y-%m-%dT%H:%M:%S", ev->time) == 0) + snprintf(ts, sizeof(ts), ""); + int prefix_len = snprintf(line, size, "%s %-5s %s:%d: ", ts, log_level_string(ev->level), ev->file, ev->line); + if (prefix_len < 0) + return; + if ((size_t)prefix_len > size - 2) + prefix_len = (int)size - 2; + + va_list ap; + va_copy(ap, ev->ap); + int msg_len = vsnprintf(line + prefix_len, size - prefix_len, ev->fmt, ap); + va_end(ap); + if (msg_len < 0) + msg_len = 0; + + // Room for the message, '\n' and the terminating '\0'. + size_t need = (size_t)prefix_len + (size_t)msg_len + 2; + if (need > size) + { + char *heap_line = malloc(need); + if (heap_line) + { + memcpy(heap_line, line, prefix_len); + vsnprintf(heap_line + prefix_len, need - prefix_len, ev->fmt, ev->ap); + line = heap_line; + } + else + msg_len = (int)(size - prefix_len - 2); + } + + size_t len = (size_t)prefix_len + (size_t)msg_len; + line[len++] = '\n'; + fwrite(line, 1, len, stderr); fflush(stderr); + + if (line != stack_line) + free(line); } void log_log(int level, const char *file, int line, const char *fmt, ...) @@ -40,9 +79,10 @@ void log_log(int level, const char *file, int line, const char *fmt, ...) .level = level, }; time_t t = time(NULL); - ev.time = localtime(&t); + struct tm tm_buf; + ev.time = localtime_r(&t, &tm_buf); va_start(ev.ap, fmt); stderr_callback(&ev); va_end(ev.ap); -} \ No newline at end of file +} diff --git a/cpworker/tests/unit/CMakeLists.txt b/cpworker/tests/unit/CMakeLists.txt index 664cd8f1..986205f6 100644 --- a/cpworker/tests/unit/CMakeLists.txt +++ b/cpworker/tests/unit/CMakeLists.txt @@ -25,7 +25,9 @@ endif() set(MAIN_CONFIG_DIR "${CMAKE_BINARY_DIR}") set(unity_tests + config_file config_validation + logging misc output_file output_rotating_file diff --git a/cpworker/tests/unit/config_file.c b/cpworker/tests/unit/config_file.c new file mode 100644 index 00000000..2ab5420d --- /dev/null +++ b/cpworker/tests/unit/config_file.c @@ -0,0 +1,166 @@ +#include +#include +#include +#include +#include +#include + +#include "unity/src/unity.h" + +#include "atomic_util.h" +#include "cJSON/cJSON.h" +#include "config.h" + +void setUp(void) {} + +void tearDown(void) {} + +static char tmp_path[64]; + +static const char *write_tmp_file(const char *content) +{ + strcpy(tmp_path, "/tmp/cpworker_ut_cfg_XXXXXX"); + int fd = mkstemp(tmp_path); + TEST_ASSERT_TRUE(fd >= 0); + size_t len = strlen(content); + TEST_ASSERT_EQUAL(len, write(fd, content, len)); + close(fd); + return tmp_path; +} + +static void assert_message_contains(const cJSONParseError *err, const char *expected) +{ + TEST_ASSERT_NOT_NULL_MESSAGE(strstr(err->message, expected), err->message); +} + +// read_file_contents + +void test_config_path_naming_a_directory_is_rejected(void) +{ + char dir[] = "/tmp/cpworker_ut_cfgdir_XXXXXX"; + TEST_ASSERT_NOT_NULL(mkdtemp(dir)); + + cJSONParseError err = {0}; + TEST_ASSERT_NULL(parse_config_file(dir, &err)); + assert_message_contains(&err, "not a regular file"); + assert_message_contains(&err, dir); + + err = (cJSONParseError){0}; + TEST_ASSERT_NULL(parse_tasks_file(dir, &err)); + assert_message_contains(&err, "not a regular file"); + + rmdir(dir); +} + +void test_empty_config_file_is_rejected(void) +{ + const char *path = write_tmp_file(""); + cJSONParseError err = {0}; + TEST_ASSERT_NULL(parse_config_file(path, &err)); + assert_message_contains(&err, "empty file"); + unlink(path); +} + +void test_missing_config_file_is_rejected(void) +{ + cJSONParseError err = {0}; + TEST_ASSERT_NULL(parse_config_file("/nonexistent/cpworker.json", &err)); + assert_message_contains(&err, "failed to open file"); +} + +void test_regular_config_file_is_parsed(void) +{ + const char *path = write_tmp_file("{\"tasks\": [{\"capturer\": {\"type\": \"libpcap\", \"libpcap\": " + "{\"interface\": \"eth0\"}}, \"outputs\": [{\"type\": \"null\"}]}]}"); + cJSONParseError err = {0}; + Config *config = parse_config_file(path, &err); + TEST_ASSERT_NOT_NULL_MESSAGE(config, err.message); + TEST_ASSERT_EQUAL(1, config->tasks_cfg->num_tasks); + free_config(config); + unlink(path); +} + +// JSON syntax errors + +void test_syntax_error_reports_line_column_and_snippet(void) +{ + cJSONParseError err = {0}; + TEST_ASSERT_NULL(parse_config_data("{\"tasks\": oops}", &err)); + TEST_ASSERT_EQUAL_STRING("JSON parse error at line 1, column 11 near 'oops}'", err.message); +} + +void test_syntax_error_snippet_stops_at_end_of_line(void) +{ + cJSONParseError err = {0}; + TEST_ASSERT_NULL(parse_config_data("{\n \"tasks\": [\n oops,\n {}\n ]\n}\n", &err)); + TEST_ASSERT_EQUAL_STRING("JSON parse error at line 3, column 5 near 'oops,'", err.message); +} + +void test_syntax_error_in_config_file_names_the_position(void) +{ + const char *path = write_tmp_file("{\n \"tasks\": [}\n"); + cJSONParseError err = {0}; + TEST_ASSERT_NULL(parse_config_file(path, &err)); + TEST_ASSERT_EQUAL_STRING("JSON parse error at line 2, column 13 near '}'", err.message); + + err = (cJSONParseError){0}; + TEST_ASSERT_NULL(parse_tasks_file(path, &err)); + TEST_ASSERT_EQUAL_STRING("JSON parse error at line 2, column 13 near '}'", err.message); + unlink(path); +} + +// The control thread parses handshakes and commands while the reload thread parses the config. cJSON records the +// last error position in a process-wide variable; the config error must not depend on it. +static bool stop_parsing; + +static void *parse_like_control_thread(void *arg) +{ + (void)arg; + while (!atomic_load_acquire(&stop_parsing)) + { + const char *end = NULL; + cJSON *msg = cJSON_ParseWithOpts("{\"version\": \"v1\"}", &end, false); + cJSON_Delete(msg); + msg = cJSON_ParseWithOpts("{\"command\": ", &end, false); + cJSON_Delete(msg); + } + return NULL; +} + +void test_syntax_error_is_not_affected_by_concurrent_parsing(void) +{ + pthread_t control; + atomic_store_release(&stop_parsing, false); + TEST_ASSERT_EQUAL(0, pthread_create(&control, NULL, parse_like_control_thread, NULL)); + + int wrong = 0; + char first_wrong[CJSON_ERRBUF_SIZE] = ""; + for (int i = 0; i < 200000; i++) + { + cJSONParseError err = {0}; + TEST_ASSERT_NULL(parse_config_data("{\"tasks\": oops}", &err)); + if (strcmp(err.message, "JSON parse error at line 1, column 11 near 'oops}'") != 0 && wrong++ == 0) + strcpy(first_wrong, err.message); + } + + atomic_store_release(&stop_parsing, true); + pthread_join(control, NULL); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, wrong, first_wrong); +} + +int main(void) +{ + UNITY_BEGIN(); + + RUN_TEST(test_config_path_naming_a_directory_is_rejected); + RUN_TEST(test_empty_config_file_is_rejected); + RUN_TEST(test_missing_config_file_is_rejected); + RUN_TEST(test_regular_config_file_is_parsed); + + RUN_TEST(test_syntax_error_reports_line_column_and_snippet); + RUN_TEST(test_syntax_error_snippet_stops_at_end_of_line); + RUN_TEST(test_syntax_error_in_config_file_names_the_position); + RUN_TEST(test_syntax_error_is_not_affected_by_concurrent_parsing); + + return UNITY_END(); +} diff --git a/cpworker/tests/unit/logging.c b/cpworker/tests/unit/logging.c new file mode 100644 index 00000000..38abe8d5 --- /dev/null +++ b/cpworker/tests/unit/logging.c @@ -0,0 +1,153 @@ +#include +#include +#include +#include +#include + +#include "unity/src/unity.h" + +#include "log.h" + +#define NUM_THREADS 4 +#define LINES_PER_THREAD 20000 +#define PAYLOAD "abcdefghijklmnopqrstuvwxyz" + +static char capture_path[64]; +static int saved_stderr = -1; + +// Points fd 2 at a temporary file so the test can read back what log_log wrote. +static void capture_stderr_begin(void) +{ + strcpy(capture_path, "/tmp/cpworker_ut_log_XXXXXX"); + int fd = mkstemp(capture_path); + TEST_ASSERT_TRUE(fd >= 0); + fflush(stderr); + saved_stderr = dup(STDERR_FILENO); + TEST_ASSERT_TRUE(saved_stderr >= 0); + TEST_ASSERT_TRUE(dup2(fd, STDERR_FILENO) >= 0); + close(fd); +} + +// Restores fd 2 and returns the captured output; the caller frees it. +static char *capture_stderr_end(size_t *len) +{ + fflush(stderr); + dup2(saved_stderr, STDERR_FILENO); + close(saved_stderr); + saved_stderr = -1; + + FILE *fp = fopen(capture_path, "rb"); + TEST_ASSERT_NOT_NULL(fp); + fseek(fp, 0, SEEK_END); + long size = ftell(fp); + fseek(fp, 0, SEEK_SET); + char *buf = malloc(size + 1); + TEST_ASSERT_NOT_NULL(buf); + TEST_ASSERT_EQUAL(size, fread(buf, 1, size, fp)); + buf[size] = '\0'; + fclose(fp); + unlink(capture_path); + *len = (size_t)size; + return buf; +} + +static size_t count_occurrences(const char *s, size_t len, const char *needle) +{ + size_t n = 0, needle_len = strlen(needle); + for (const char *p = s; (p = memmem(p, len - (p - s), needle, needle_len)) != NULL; p += needle_len) + n++; + return n; +} + +void setUp(void) { log_set_level(LOG_TRACE); } + +void tearDown(void) {} + +static void *log_from_thread(void *arg) +{ + long id = (long)arg; + for (int i = 0; i < LINES_PER_THREAD; i++) + log_info("thread=%ld seq=%d payload=" PAYLOAD, id, i); + return NULL; +} + +// Each line must be "