// Define _GNU_SOURCE to ensure all POSIX features are available #define _GNU_SOURCE // Includes #include #include "debug.h" #include #include #include #include #include #include // Include libwebsockets after pthread.h to ensure pthread_rwlock_t is defined #include #include #include #include #include // Include nostr_core_lib for Nostr functionality #include "../nostr_core_lib/cjson/cJSON.h" #include "../nostr_core_lib/nostr_core/nostr_core.h" #include "../nostr_core_lib/nostr_core/nip013.h" // NIP-13: Proof of Work #include "config.h" // Configuration management system #include "sql_schema.h" // Embedded database schema #include "websockets.h" // WebSocket structures and constants #include "subscriptions.h" // Subscription structures and functions #include "embedded_web_content.h" // Embedded web content #include "api.h" // API for embedded files #include "dm_admin.h" // DM admin functions including NIP-17 #include "ip_ban.h" // IP auth failure ban system #include "thread_pool.h" // Thread pool scaffold #include "db_ops.h" // DB abstraction wrappers #include "main.h" // Async REQ completion integration #include "caching_inbox_poller.h" // Caching inbox poller (PostgreSQL-only) // Forward declarations for logging functions // Forward declarations for configuration functions const char* get_config_value(const char* key); int get_config_int(const char* key, int default_value); int get_config_bool(const char* key, int default_value); // Forward declarations for NIP-42 authentication functions int is_nip42_auth_globally_required(void); int is_nip42_auth_required_for_kind(int kind); void send_nip42_auth_challenge(struct lws* wsi, struct per_session_data* pss); void handle_nip42_auth_signed_event(struct lws* wsi, struct per_session_data* pss, cJSON* auth_event); void handle_nip42_auth_challenge_response(struct lws* wsi, struct per_session_data* pss, const char* challenge); // Forward declaration for status posts int generate_and_post_status_event(void); // Forward declarations for NIP-11 relay information handling int handle_nip11_http_request(struct lws* wsi, const char* accept_header); // Forward declarations for embedded file handling int handle_embedded_file_writeable(struct lws* wsi); // Forward declarations for database functions int store_event(cJSON* event); // Forward declarations for subscription management int broadcast_event_to_subscriptions(cJSON* event); int add_subscription_to_manager(struct subscription* sub); int remove_subscription_from_manager(const char* sub_id, struct lws* wsi); // Forward declarations for event handling int handle_event_message(cJSON* event, char* error_message, size_t error_size); int nostr_validate_unified_request(const char* json_string, size_t json_length); int event_id_exists_in_db(const char* event_id); // Forward declarations for admin event processing int process_admin_event_in_config(cJSON* event, char* error_message, size_t error_size, struct lws* wsi); int is_authorized_admin_event(cJSON* event, char* error_message, size_t error_size); // Forward declarations for DM stats command handling int process_dm_stats_command(cJSON* dm_event, char* error_message, size_t error_size, struct lws* wsi); // Forward declarations for NIP-09 deletion request handling int handle_deletion_request(cJSON* event, char* error_message, size_t error_size); // Forward declarations for NIP-13 PoW handling int validate_event_pow(cJSON* event, char* error_message, size_t error_size); // Forward declarations for NIP-40 expiration handling int is_event_expired(cJSON* event, time_t current_time); // Forward declarations for subscription handling int handle_req_message(const char* sub_id, cJSON* filters, struct lws *wsi, struct per_session_data *pss); int handle_count_message(const char* sub_id, cJSON* filters, struct lws *wsi, struct per_session_data *pss); // Forward declaration for query logging (defined in main.c) extern void log_query_execution(const char* query_type, const char* sub_id, const char* client_ip, const char* sql, long elapsed_us, int rows_returned); // Forward declarations for rate limiting int is_client_rate_limited_for_malformed_requests(struct per_session_data *pss); void record_malformed_request(struct per_session_data *pss); // Forward declarations for filter validation int validate_filter_array(cJSON* filters, char* error_message, size_t error_size); // Forward declarations for NOTICE message support void send_notice_message(struct lws* wsi, struct per_session_data* pss, const char* message); // Configuration functions from config.c extern int get_config_bool(const char* key, int default_value); // Forward declarations for global state extern int g_server_running; extern volatile sig_atomic_t g_shutdown_flag; extern int g_restart_requested; extern struct lws_context *ws_context; // Global subscription manager struct subscription_manager g_subscription_manager; // Actual relay port bound by libwebsockets at runtime (after fallback/retry logic) int g_relay_port = DEFAULT_PORT; // Global connection list for idle connection tracking // Tracks ALL WebSocket connections (not just subscribed ones) // so the periodic timer can find and close idle connections #define MAX_TRACKED_CONNECTIONS 4096 typedef struct { struct lws* wsi; struct per_session_data* pss; } tracked_connection_t; static tracked_connection_t g_connections[MAX_TRACKED_CONNECTIONS]; static int g_connection_count = 0; static pthread_mutex_t g_connections_lock = PTHREAD_MUTEX_INITIALIZER; int get_active_connection_count(void) { return g_connection_count; } int websocket_get_live_pss(struct lws* wsi, struct per_session_data** out_pss) { if (!wsi || !out_pss) { return 0; } *out_pss = NULL; pthread_mutex_lock(&g_connections_lock); for (int i = 0; i < MAX_TRACKED_CONNECTIONS; i++) { if (g_connections[i].wsi == wsi && g_connections[i].pss != NULL) { *out_pss = g_connections[i].pss; pthread_mutex_unlock(&g_connections_lock); return 1; } } pthread_mutex_unlock(&g_connections_lock); return 0; } static void connection_list_add(struct lws* wsi, struct per_session_data* pss) { pthread_mutex_lock(&g_connections_lock); for (int i = 0; i < MAX_TRACKED_CONNECTIONS; i++) { if (g_connections[i].wsi == NULL) { g_connections[i].wsi = wsi; g_connections[i].pss = pss; g_connection_count++; break; } } pthread_mutex_unlock(&g_connections_lock); } static void connection_list_remove(struct lws* wsi) { pthread_mutex_lock(&g_connections_lock); for (int i = 0; i < MAX_TRACKED_CONNECTIONS; i++) { if (g_connections[i].wsi == wsi) { g_connections[i].wsi = NULL; g_connections[i].pss = NULL; g_connection_count--; break; } } pthread_mutex_unlock(&g_connections_lock); } // Check all tracked connections for idle timeout and close them // Called from the periodic maintenance timer (every 60 seconds) static void check_idle_connections(int idle_timeout_sec) { if (idle_timeout_sec <= 0) return; time_t now = time(NULL); // Collect WSIs to close outside the lock to avoid deadlock struct lws* to_close[MAX_TRACKED_CONNECTIONS]; int close_count = 0; pthread_mutex_lock(&g_connections_lock); for (int i = 0; i < MAX_TRACKED_CONNECTIONS; i++) { if (g_connections[i].wsi == NULL || g_connections[i].pss == NULL) continue; struct per_session_data* pss = g_connections[i].pss; if (pss->session_active) continue; // Already active — skip if (pss->connection_established <= 0) continue; time_t age = now - pss->connection_established; if (age >= idle_timeout_sec) { to_close[close_count++] = g_connections[i].wsi; } } pthread_mutex_unlock(&g_connections_lock); for (int i = 0; i < close_count; i++) { // Get pss again safely — it may have been freed if connection closed between lock release and here struct per_session_data* pss = (struct per_session_data*)lws_wsi_user(to_close[i]); if (!pss) continue; DEBUG_LOG("Closing idle connection from %s (no REQ/EVENT after %d seconds)", pss->client_ip, idle_timeout_sec); lws_close_reason(to_close[i], LWS_CLOSE_STATUS_POLICY_VIOLATION, (unsigned char*)"Idle connection timeout", 23); } } // Hot-path config cache (Phase 4): reduce per-message SQLite config lookups. typedef struct { time_t last_refresh; int ttl_sec; int nip70_protected_events_enabled; int nip17_admin_enabled; int wot_enabled; int nip42_auth_timeout_sec; int idle_connection_timeout_sec; int max_connection_seconds; int kind_1_status_posts_hours; int debug_level; } hot_config_cache_t; static hot_config_cache_t g_hot_config = { .last_refresh = 0, .ttl_sec = 5, .nip70_protected_events_enabled = 0, .nip17_admin_enabled = 0, .wot_enabled = 0, .nip42_auth_timeout_sec = 10, .idle_connection_timeout_sec = 30, .max_connection_seconds = 86400, .kind_1_status_posts_hours = 0, .debug_level = -1, }; static pthread_mutex_t g_hot_config_mutex = PTHREAD_MUTEX_INITIALIZER; static void refresh_hot_config_if_needed(void) { time_t now = time(NULL); pthread_mutex_lock(&g_hot_config_mutex); if (g_hot_config.last_refresh != 0 && (now - g_hot_config.last_refresh) < g_hot_config.ttl_sec) { pthread_mutex_unlock(&g_hot_config_mutex); return; } g_hot_config.nip70_protected_events_enabled = get_config_bool("nip70_protected_events_enabled", 0); g_hot_config.nip17_admin_enabled = get_config_bool("nip17_admin_enabled", 0); g_hot_config.wot_enabled = get_config_int("wot_enabled", 0); g_hot_config.nip42_auth_timeout_sec = get_config_int("nip42_auth_timeout_sec", 10); g_hot_config.idle_connection_timeout_sec = get_config_int("idle_connection_timeout_sec", 30); g_hot_config.max_connection_seconds = get_config_int("max_connection_seconds", 86400); g_hot_config.kind_1_status_posts_hours = get_config_int("kind_1_status_posts_hours", 0); g_hot_config.debug_level = get_config_int("debug_level", -1); g_hot_config.last_refresh = now; pthread_mutex_unlock(&g_hot_config_mutex); } static int hot_cfg_nip70_protected_events_enabled(void) { refresh_hot_config_if_needed(); pthread_mutex_lock(&g_hot_config_mutex); int v = g_hot_config.nip70_protected_events_enabled; pthread_mutex_unlock(&g_hot_config_mutex); return v; } static int hot_cfg_nip17_admin_enabled(void) { refresh_hot_config_if_needed(); pthread_mutex_lock(&g_hot_config_mutex); int v = g_hot_config.nip17_admin_enabled; pthread_mutex_unlock(&g_hot_config_mutex); return v; } static int hot_cfg_wot_enabled(void) { refresh_hot_config_if_needed(); pthread_mutex_lock(&g_hot_config_mutex); int v = g_hot_config.wot_enabled; pthread_mutex_unlock(&g_hot_config_mutex); return v; } static int hot_cfg_nip42_auth_timeout_sec(void) { refresh_hot_config_if_needed(); pthread_mutex_lock(&g_hot_config_mutex); int v = g_hot_config.nip42_auth_timeout_sec; pthread_mutex_unlock(&g_hot_config_mutex); return v; } static int hot_cfg_idle_connection_timeout_sec(void) { refresh_hot_config_if_needed(); pthread_mutex_lock(&g_hot_config_mutex); int v = g_hot_config.idle_connection_timeout_sec; pthread_mutex_unlock(&g_hot_config_mutex); return v; } static int hot_cfg_max_connection_seconds(void) { refresh_hot_config_if_needed(); pthread_mutex_lock(&g_hot_config_mutex); int v = g_hot_config.max_connection_seconds; pthread_mutex_unlock(&g_hot_config_mutex); return v; } static int hot_cfg_kind_1_status_posts_hours(void) { refresh_hot_config_if_needed(); pthread_mutex_lock(&g_hot_config_mutex); int v = g_hot_config.kind_1_status_posts_hours; pthread_mutex_unlock(&g_hot_config_mutex); return v; } static int hot_cfg_debug_level(void) { refresh_hot_config_if_needed(); pthread_mutex_lock(&g_hot_config_mutex); int v = g_hot_config.debug_level; pthread_mutex_unlock(&g_hot_config_mutex); return v; } // Async EVENT processing (Phase 1): offload signature validation + store_event from main lws thread. typedef struct async_event_job { char* event_json; char event_id[65]; int event_kind; struct lws* wsi; struct per_session_data* pss_token; struct async_event_job* next; } async_event_job_t; typedef struct async_event_completion { struct lws* wsi; struct per_session_data* pss_token; char* event_json; char event_id[65]; int success; int should_broadcast; int run_post_actions; char ok_message[128]; char error_message[512]; struct async_event_completion* next; } async_event_completion_t; static pthread_mutex_t g_async_event_job_mutex = PTHREAD_MUTEX_INITIALIZER; static pthread_cond_t g_async_event_job_cond = PTHREAD_COND_INITIALIZER; static async_event_job_t* g_async_event_job_head = NULL; static async_event_job_t* g_async_event_job_tail = NULL; static pthread_mutex_t g_async_event_completion_mutex = PTHREAD_MUTEX_INITIALIZER; static async_event_completion_t* g_async_event_completion_head = NULL; static async_event_completion_t* g_async_event_completion_tail = NULL; static pthread_t g_async_event_worker_thread; static int g_async_event_worker_running = 0; typedef struct count_async_state { char sub_id[SUBSCRIPTION_ID_MAX_LENGTH]; struct lws* wsi; struct per_session_data* pss_token; int pending_jobs; int total_count; int had_error; char error_message[256]; } count_async_state_t; typedef struct count_async_completion { count_async_state_t* state; thread_pool_status_t status; int count; struct count_async_completion* next; } count_async_completion_t; static pthread_mutex_t g_count_async_completion_mutex = PTHREAD_MUTEX_INITIALIZER; static count_async_completion_t* g_count_async_completion_head = NULL; static count_async_completion_t* g_count_async_completion_tail = NULL; static void map_validation_error_message(int validation_result, char* out, size_t out_size) { if (!out || out_size == 0) return; switch (validation_result) { case NOSTR_ERROR_INVALID_INPUT: strncpy(out, "invalid: malformed event structure", out_size - 1); break; case NOSTR_ERROR_EVENT_INVALID_SIGNATURE: strncpy(out, "invalid: signature verification failed", out_size - 1); break; case NOSTR_ERROR_EVENT_INVALID_ID: strncpy(out, "invalid: event id verification failed", out_size - 1); break; case NOSTR_ERROR_EVENT_INVALID_PUBKEY: strncpy(out, "invalid: invalid pubkey format", out_size - 1); break; case -103: // NOSTR_ERROR_EVENT_EXPIRED strncpy(out, "rejected: event expired", out_size - 1); break; case -102: // NOSTR_ERROR_NIP42_DISABLED strncpy(out, "auth-required: NIP-42 authentication required", out_size - 1); break; case -101: // NOSTR_ERROR_AUTH_REQUIRED strncpy(out, "blocked: pubkey not authorized", out_size - 1); break; default: strncpy(out, "error: validation failed", out_size - 1); break; } out[out_size - 1] = '\0'; } static void send_ok_response(struct lws* wsi, struct per_session_data* pss, const char* event_id, int accepted, const char* message) { if (!wsi || !pss || !event_id) { return; } cJSON* response = cJSON_CreateArray(); if (!response) { return; } cJSON_AddItemToArray(response, cJSON_CreateString("OK")); cJSON_AddItemToArray(response, cJSON_CreateString(event_id)); cJSON_AddItemToArray(response, cJSON_CreateBool(accepted ? 1 : 0)); cJSON_AddItemToArray(response, cJSON_CreateString(message ? message : "")); char* response_str = cJSON_Print(response); if (response_str) { size_t response_len = strlen(response_str); if (queue_message(wsi, pss, response_str, response_len, LWS_WRITE_TEXT) != 0) { DEBUG_ERROR("Failed to queue OK response message"); } free(response_str); } cJSON_Delete(response); } static void async_event_completion_push(async_event_completion_t* completion) { if (!completion) return; pthread_mutex_lock(&g_async_event_completion_mutex); completion->next = NULL; if (!g_async_event_completion_tail) { g_async_event_completion_head = completion; g_async_event_completion_tail = completion; } else { g_async_event_completion_tail->next = completion; g_async_event_completion_tail = completion; } pthread_mutex_unlock(&g_async_event_completion_mutex); } static async_event_completion_t* async_event_completion_pop(void) { pthread_mutex_lock(&g_async_event_completion_mutex); async_event_completion_t* completion = g_async_event_completion_head; if (completion) { g_async_event_completion_head = completion->next; if (!g_async_event_completion_head) { g_async_event_completion_tail = NULL; } } pthread_mutex_unlock(&g_async_event_completion_mutex); return completion; } static int event_is_async_eligible(cJSON* event, int* out_kind, char event_id_out[65]) { if (!event || !out_kind || !event_id_out) { return 0; } cJSON* id_obj = cJSON_GetObjectItemCaseSensitive(event, "id"); cJSON* kind_obj = cJSON_GetObjectItemCaseSensitive(event, "kind"); if (!id_obj || !cJSON_IsString(id_obj) || !kind_obj || !cJSON_IsNumber(kind_obj)) { return 0; } const char* event_id = cJSON_GetStringValue(id_obj); if (!event_id || strlen(event_id) >= 65) { return 0; } int kind = (int)cJSON_GetNumberValue(kind_obj); // Keep special command/DM/protected paths on main thread for existing behavior. if (kind == 23456 || kind == 1059 || kind == 14) { return 0; } // NIP-70 protected events must stay on sync path so auth checks run. cJSON* tags_obj = cJSON_GetObjectItemCaseSensitive(event, "tags"); if (tags_obj && cJSON_IsArray(tags_obj)) { cJSON* tag = NULL; cJSON_ArrayForEach(tag, tags_obj) { if (cJSON_IsArray(tag) && cJSON_GetArraySize(tag) >= 1) { cJSON* tag_name = cJSON_GetArrayItem(tag, 0); if (tag_name && cJSON_IsString(tag_name) && strcmp(cJSON_GetStringValue(tag_name), "-") == 0) { return 0; } } } } strncpy(event_id_out, event_id, 64); event_id_out[64] = '\0'; *out_kind = kind; return 1; } static async_event_job_t* async_event_job_pop_blocking(void) { pthread_mutex_lock(&g_async_event_job_mutex); while (g_async_event_worker_running && !g_async_event_job_head) { pthread_cond_wait(&g_async_event_job_cond, &g_async_event_job_mutex); } async_event_job_t* job = g_async_event_job_head; if (job) { g_async_event_job_head = job->next; if (!g_async_event_job_head) { g_async_event_job_tail = NULL; } } pthread_mutex_unlock(&g_async_event_job_mutex); return job; } static void* async_event_worker_main(void* arg) { (void)arg; pthread_setname_np(pthread_self(), "event-worker"); void* worker_db = NULL; const char* db_path = db_get_database_path(); if (db_open_worker_connection(db_path, &worker_db) != 0) { DEBUG_WARN("event-worker: failed to open dedicated DB connection; using default connection context"); } while (g_async_event_worker_running) { async_event_job_t* job = async_event_job_pop_blocking(); if (!job) { continue; } async_event_completion_t* completion = calloc(1, sizeof(*completion)); if (!completion) { free(job->event_json); free(job); continue; } completion->wsi = job->wsi; completion->pss_token = job->pss_token; completion->event_json = job->event_json; // transfer ownership strncpy(completion->event_id, job->event_id, sizeof(completion->event_id) - 1); completion->event_id[sizeof(completion->event_id) - 1] = '\0'; if (job->event_id[0] != '\0' && event_id_exists_in_db(job->event_id)) { completion->success = 1; completion->should_broadcast = 0; completion->run_post_actions = 0; snprintf(completion->ok_message, sizeof(completion->ok_message), "duplicate: already have this event"); } else { int validation_result = nostr_validate_unified_request(completion->event_json, strlen(completion->event_json)); if (validation_result != NOSTR_SUCCESS) { completion->success = 0; completion->should_broadcast = 0; map_validation_error_message(validation_result, completion->error_message, sizeof(completion->error_message)); } else { cJSON* event_obj = cJSON_Parse(completion->event_json); if (!event_obj || !cJSON_IsObject(event_obj)) { completion->success = 0; completion->should_broadcast = 0; strncpy(completion->error_message, "error: failed to parse event", sizeof(completion->error_message) - 1); completion->error_message[sizeof(completion->error_message) - 1] = '\0'; } else { if (job->event_kind >= 20000 && job->event_kind < 30000) { completion->success = 1; completion->should_broadcast = 1; completion->run_post_actions = 0; } else { int core_rc = store_event_core(event_obj); if (core_rc < 0) { completion->success = 0; completion->should_broadcast = 0; completion->run_post_actions = 0; strncpy(completion->error_message, "error: failed to store event", sizeof(completion->error_message) - 1); completion->error_message[sizeof(completion->error_message) - 1] = '\0'; } else { completion->success = 1; completion->should_broadcast = 1; completion->run_post_actions = (core_rc == 0); } } cJSON_Delete(event_obj); } } } async_event_completion_push(completion); if (ws_context) { lws_cancel_service(ws_context); } free(job); } if (worker_db) { db_close_worker_connection(worker_db); } return NULL; } static int start_async_event_worker(void) { if (g_async_event_worker_running) { return 0; } g_async_event_worker_running = 1; if (pthread_create(&g_async_event_worker_thread, NULL, async_event_worker_main, NULL) != 0) { g_async_event_worker_running = 0; return -1; } return 0; } static void stop_async_event_worker(void) { if (!g_async_event_worker_running) { return; } pthread_mutex_lock(&g_async_event_job_mutex); g_async_event_worker_running = 0; pthread_cond_broadcast(&g_async_event_job_cond); pthread_mutex_unlock(&g_async_event_job_mutex); pthread_join(g_async_event_worker_thread, NULL); pthread_mutex_lock(&g_async_event_job_mutex); async_event_job_t* job = g_async_event_job_head; while (job) { async_event_job_t* next = job->next; free(job->event_json); free(job); job = next; } g_async_event_job_head = g_async_event_job_tail = NULL; pthread_mutex_unlock(&g_async_event_job_mutex); async_event_completion_t* completion = NULL; while ((completion = async_event_completion_pop()) != NULL) { free(completion->event_json); free(completion); } } // Returns: // 0 => accepted for async handling // 1 => not eligible (safe to continue with special synchronous path) // -1 => submit/allocation failure // -2 => async worker unavailable (do not fall back to synchronous store on lws-main) static int try_submit_async_event(cJSON* event, const char* event_json, struct lws* wsi, struct per_session_data* pss) { if (!event || !event_json || !wsi || !pss) { return 1; } if (!g_async_event_worker_running) { return -2; } int event_kind = 0; char event_id[65] = {0}; if (!event_is_async_eligible(event, &event_kind, event_id)) { return 1; } async_event_job_t* job = calloc(1, sizeof(*job)); if (!job) { return -1; } job->event_json = strdup(event_json); if (!job->event_json) { free(job); return -1; } strncpy(job->event_id, event_id, sizeof(job->event_id) - 1); job->event_id[sizeof(job->event_id) - 1] = '\0'; job->event_kind = event_kind; job->wsi = wsi; job->pss_token = pss; pthread_mutex_lock(&g_async_event_job_mutex); job->next = NULL; if (!g_async_event_job_tail) { g_async_event_job_head = job; g_async_event_job_tail = job; } else { g_async_event_job_tail->next = job; g_async_event_job_tail = job; } pthread_cond_signal(&g_async_event_job_cond); pthread_mutex_unlock(&g_async_event_job_mutex); return 0; } static void process_async_event_completions(void) { async_event_completion_t* completion = NULL; while ((completion = async_event_completion_pop()) != NULL) { struct per_session_data* current_pss = NULL; int target_alive = websocket_get_live_pss(completion->wsi, ¤t_pss) && current_pss == completion->pss_token; int needs_event_obj = completion->run_post_actions || (target_alive && completion->success && completion->should_broadcast); if (completion->success && needs_event_obj) { cJSON* event_obj = cJSON_Parse(completion->event_json); if (event_obj && cJSON_IsObject(event_obj)) { // Must run on main lws thread due config/monitoring DB access. if (completion->run_post_actions) { store_event_post_actions(event_obj); } if (target_alive && completion->should_broadcast) { broadcast_event_to_subscriptions(event_obj); } cJSON_Delete(event_obj); } else { completion->success = 0; strncpy(completion->error_message, "error: failed to process event", sizeof(completion->error_message) - 1); completion->error_message[sizeof(completion->error_message) - 1] = '\0'; if (event_obj) { cJSON_Delete(event_obj); } } } if (target_alive) { send_ok_response(completion->wsi, current_pss, completion->event_id, completion->success, completion->success ? (completion->ok_message[0] ? completion->ok_message : "") : completion->error_message); } free(completion->event_json); free(completion); } } static void free_count_payload_ws(void* payload_void) { thread_pool_count_payload_t* payload = (thread_pool_count_payload_t*)payload_void; if (!payload) return; free(payload->sql); if (payload->bind_params) { for (int i = 0; i < payload->bind_param_count; i++) { free(payload->bind_params[i]); } free(payload->bind_params); } free(payload); } static void count_async_completion_push(count_async_completion_t* completion) { if (!completion) return; pthread_mutex_lock(&g_count_async_completion_mutex); completion->next = NULL; if (!g_count_async_completion_tail) { g_count_async_completion_head = completion; g_count_async_completion_tail = completion; } else { g_count_async_completion_tail->next = completion; g_count_async_completion_tail = completion; } pthread_mutex_unlock(&g_count_async_completion_mutex); } static count_async_completion_t* count_async_completion_pop(void) { pthread_mutex_lock(&g_count_async_completion_mutex); count_async_completion_t* completion = g_count_async_completion_head; if (completion) { g_count_async_completion_head = completion->next; if (!g_count_async_completion_head) { g_count_async_completion_tail = NULL; } } pthread_mutex_unlock(&g_count_async_completion_mutex); return completion; } static void count_async_result_cb(const thread_pool_result_t* result, void* user_ctx) { count_async_state_t* state = (count_async_state_t*)user_ctx; if (!result || !state) { if (result && result->result_data) { free(result->result_data); } return; } count_async_completion_t* completion = calloc(1, sizeof(*completion)); if (!completion) { if (result->result_data) { free(result->result_data); } return; } completion->state = state; completion->status = result->status; if (result->status == THREAD_POOL_STATUS_OK && result->result_data) { thread_pool_count_result_t* count_result = (thread_pool_count_result_t*)result->result_data; completion->count = count_result->count; } if (result->result_data) { free(result->result_data); } count_async_completion_push(completion); if (ws_context) { lws_cancel_service(ws_context); } } static int submit_count_query_async(count_async_state_t* state, const char* sql, const char** bind_params, int bind_param_count) { if (!state || !sql) { return -1; } thread_pool_count_payload_t* payload = calloc(1, sizeof(*payload)); if (!payload) { return -1; } payload->sql = strdup(sql); payload->bind_param_count = bind_param_count; if (!payload->sql) { free_count_payload_ws(payload); return -1; } if (bind_param_count > 0) { payload->bind_params = calloc((size_t)bind_param_count, sizeof(char*)); if (!payload->bind_params) { free_count_payload_ws(payload); return -1; } for (int i = 0; i < bind_param_count; i++) { const char* value = (bind_params && bind_params[i]) ? bind_params[i] : ""; payload->bind_params[i] = strdup(value); if (!payload->bind_params[i]) { free_count_payload_ws(payload); return -1; } } } thread_pool_job_t job; memset(&job, 0, sizeof(job)); job.type = THREAD_POOL_JOB_COUNT_QUERY; job.payload = payload; job.payload_free = free_count_payload_ws; job.result_cb = count_async_result_cb; job.result_cb_ctx = state; thread_pool_status_t submit_rc = thread_pool_submit_read(&job, NULL); if (submit_rc != THREAD_POOL_STATUS_OK) { free_count_payload_ws(payload); return -1; } state->pending_jobs++; return 0; } static void send_count_response(struct lws* wsi, struct per_session_data* pss, const char* sub_id, int total_count) { cJSON* count_response = cJSON_CreateArray(); if (!count_response) { return; } cJSON_AddItemToArray(count_response, cJSON_CreateString("COUNT")); cJSON_AddItemToArray(count_response, cJSON_CreateString(sub_id)); cJSON* count_obj = cJSON_CreateObject(); cJSON_AddNumberToObject(count_obj, "count", total_count); cJSON_AddItemToArray(count_response, count_obj); char* count_str = cJSON_Print(count_response); if (count_str) { size_t count_len = strlen(count_str); if (queue_message(wsi, pss, count_str, count_len, LWS_WRITE_TEXT) != 0) { DEBUG_ERROR("Failed to queue COUNT message"); } free(count_str); } cJSON_Delete(count_response); } static void process_count_async_completions(void) { count_async_completion_t* completion = NULL; while ((completion = count_async_completion_pop()) != NULL) { count_async_state_t* state = completion->state; if (state) { if (completion->status == THREAD_POOL_STATUS_OK) { state->total_count += completion->count; } else { state->had_error = 1; if (state->error_message[0] == '\0') { strncpy(state->error_message, "error: failed to execute count query", sizeof(state->error_message) - 1); state->error_message[sizeof(state->error_message) - 1] = '\0'; } } state->pending_jobs--; if (state->pending_jobs <= 0) { struct per_session_data* current_pss = NULL; int target_alive = websocket_get_live_pss(state->wsi, ¤t_pss) && current_pss == state->pss_token; if (target_alive) { if (state->had_error) { send_notice_message(state->wsi, current_pss, state->error_message[0] ? state->error_message : "error: count query failed"); } else { send_count_response(state->wsi, current_pss, state->sub_id, state->total_count); } } free(state); } } free(completion); } } // Message queue functions for proper libwebsockets pattern /** * Queue a message for WebSocket writing following libwebsockets' proper pattern. * This function adds messages to a per-session queue and requests writeable callback. * * @param wsi WebSocket instance * @param pss Per-session data containing message queue * @param message Message string to write * @param length Length of message string * @param type LWS_WRITE_* type (LWS_WRITE_TEXT, etc.) * @return 0 on success, -1 on error */ int queue_message(struct lws* wsi, struct per_session_data* pss, const char* message, size_t length, enum lws_write_protocol type) { if (!wsi || !pss || !message || length == 0) { DEBUG_ERROR("queue_message: invalid parameters"); return -1; } // Drop message if queue is full to prevent unbounded memory growth under load. // Slow or disconnected clients should not cause the relay to OOM. if (pss->message_queue_count >= MAX_MESSAGE_QUEUE_SIZE) { DEBUG_WARN("queue_message: queue full (%d), dropping message for slow/disconnected client", pss->message_queue_count); return -1; } // Allocate message queue node struct message_queue_node* node = malloc(sizeof(struct message_queue_node)); if (!node) { DEBUG_ERROR("queue_message: failed to allocate queue node"); return -1; } // Allocate buffer with LWS_PRE space size_t buffer_size = LWS_PRE + length; unsigned char* buffer = malloc(buffer_size); if (!buffer) { DEBUG_ERROR("queue_message: failed to allocate message buffer"); free(node); return -1; } // Copy message to buffer with LWS_PRE offset memcpy(buffer + LWS_PRE, message, length); // Initialize node node->data = buffer; node->length = length; node->type = type; node->next = NULL; // Add to queue (thread-safe) pthread_mutex_lock(&pss->session_lock); if (!pss->message_queue_head) { // Queue was empty pss->message_queue_head = node; pss->message_queue_tail = node; } else { // Add to end of queue pss->message_queue_tail->next = node; pss->message_queue_tail = node; } pss->message_queue_count++; pthread_mutex_unlock(&pss->session_lock); // Request writeable callback (only if not already requested) if (!pss->writeable_requested) { pss->writeable_requested = 1; lws_callback_on_writable(wsi); } DEBUG_TRACE("Queued message: len=%zu, queue_count=%d", length, pss->message_queue_count); return 0; } /** * Zero-copy variant of queue_message. The caller allocates a buffer of * (LWS_PRE + length) bytes, writes the message at (buf + LWS_PRE), then * passes ownership to the queue. The queue will free buf when done. * No memcpy is performed — eliminates one copy per queued message. * * @param wsi WebSocket instance * @param pss Per-session data containing message queue * @param buf Pre-allocated buffer of size (LWS_PRE + length); ownership transferred * @param length Length of message (NOT including LWS_PRE) * @param type LWS_WRITE_* type * @return 0 on success, -1 on error (buf is freed on error) */ int queue_message_take_ownership(struct lws* wsi, struct per_session_data* pss, unsigned char* buf, size_t length, enum lws_write_protocol type) { if (!wsi || !pss || !buf || length == 0) { DEBUG_ERROR("queue_message_take_ownership: invalid parameters"); free(buf); return -1; } // Drop message if queue is full if (pss->message_queue_count >= MAX_MESSAGE_QUEUE_SIZE) { DEBUG_WARN("queue_message_take_ownership: queue full (%d), dropping message", pss->message_queue_count); free(buf); return -1; } struct message_queue_node* node = malloc(sizeof(struct message_queue_node)); if (!node) { DEBUG_ERROR("queue_message_take_ownership: failed to allocate queue node"); free(buf); return -1; } node->data = buf; // buf already has LWS_PRE prefix — no copy needed node->length = length; node->type = type; node->next = NULL; pthread_mutex_lock(&pss->session_lock); if (!pss->message_queue_head) { pss->message_queue_head = node; pss->message_queue_tail = node; } else { pss->message_queue_tail->next = node; pss->message_queue_tail = node; } pss->message_queue_count++; pthread_mutex_unlock(&pss->session_lock); if (!pss->writeable_requested) { pss->writeable_requested = 1; lws_callback_on_writable(wsi); } DEBUG_TRACE("Queued message (zero-copy): len=%zu, queue_count=%d", length, pss->message_queue_count); return 0; } /** * Process message queue when the socket becomes writeable. * This function is called from LWS_CALLBACK_SERVER_WRITEABLE. * * @param wsi WebSocket instance * @param pss Per-session data containing message queue * @return 0 on success, -1 on error */ int process_message_queue(struct lws* wsi, struct per_session_data* pss) { if (!wsi || !pss) { DEBUG_ERROR("process_message_queue: invalid parameters"); return -1; } // Drain as many queued messages as possible in one writeable callback. // This avoids short-lived clients timing out before receiving all EVENT/EOSE frames. int processed = 0; while (1) { pthread_mutex_lock(&pss->session_lock); struct message_queue_node* node = pss->message_queue_head; if (!node) { // Queue is empty pss->writeable_requested = 0; pthread_mutex_unlock(&pss->session_lock); break; } // Remove from queue pss->message_queue_head = node->next; if (!pss->message_queue_head) { pss->message_queue_tail = NULL; } pss->message_queue_count--; pthread_mutex_unlock(&pss->session_lock); // Write message (libwebsockets handles partial writes internally) int write_result = lws_write(wsi, node->data + LWS_PRE, node->length, node->type); // Free node resources free(node->data); free(node); if (write_result < 0) { DEBUG_ERROR("process_message_queue: write failed, result=%d", write_result); return -1; } processed++; // If socket is currently choked, stop and continue later. if (lws_send_pipe_choked(wsi)) { break; } } DEBUG_TRACE("Processed %d queued messages", processed); // If queue still has pending messages, ensure we request another callback. pthread_mutex_lock(&pss->session_lock); if (pss->message_queue_head) { pss->writeable_requested = 1; lws_callback_on_writable(wsi); } else { pss->writeable_requested = 0; } pthread_mutex_unlock(&pss->session_lock); return 0; } ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // WEBSOCKET PROTOCOL ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // WebSocket callback function for Nostr relay protocol static int nostr_relay_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len) { struct per_session_data *pss = (struct per_session_data *)user; switch (reason) { case LWS_CALLBACK_HTTP: // Handle HTTP requests // Mark session as active so HTTP requests don't trigger idle ban if (pss) { pthread_mutex_lock(&pss->session_lock); pss->session_active = 1; pthread_mutex_unlock(&pss->session_lock); } { char *requested_uri = (char *)in; // Check if this is an OPTIONS request char method[16] = {0}; int method_len = lws_hdr_copy(wsi, method, sizeof(method) - 1, WSI_TOKEN_GET_URI); if (method_len > 0) { method[method_len] = '\0'; if (strcmp(method, "OPTIONS") == 0) { // Handle OPTIONS request with CORS headers unsigned char buf[LWS_PRE + 1024]; unsigned char *p = &buf[LWS_PRE]; unsigned char *start = p; unsigned char *end = &buf[sizeof(buf) - 1]; if (lws_add_http_header_status(wsi, HTTP_STATUS_OK, &p, end)) return -1; if (lws_add_http_header_by_name(wsi, (unsigned char*)"access-control-allow-origin:", (unsigned char*)"*", 1, &p, end)) return -1; if (lws_add_http_header_by_name(wsi, (unsigned char*)"access-control-allow-headers:", (unsigned char*)"content-type, accept", 20, &p, end)) return -1; if (lws_add_http_header_by_name(wsi, (unsigned char*)"access-control-allow-methods:", (unsigned char*)"GET, OPTIONS", 12, &p, end)) return -1; if (lws_add_http_header_by_name(wsi, (unsigned char*)"connection:", (unsigned char*)"close", 5, &p, end)) return -1; if (lws_finalize_http_header(wsi, &p, end)) return -1; if (lws_write(wsi, start, p - start, LWS_WRITE_HTTP_HEADERS) < 0) return -1; return 0; } } // Check if this is a GET request to the root path if (strcmp(requested_uri, "/") == 0) { // Check if this is a WebSocket upgrade request char upgrade_header[64] = {0}; int upgrade_len = lws_hdr_copy(wsi, upgrade_header, sizeof(upgrade_header) - 1, WSI_TOKEN_UPGRADE); if (upgrade_len > 0) { upgrade_header[upgrade_len] = '\0'; if (strstr(upgrade_header, "websocket") != NULL) { DEBUG_LOG("WebSocket upgrade request - allowing connection"); return 0; } } // Not a WebSocket upgrade: treat root path as NIP-11 endpoint. // NIP-11 handler will return 200 for proper Accept header and 406 otherwise. char accept_header[256] = {0}; int header_len = lws_hdr_copy(wsi, accept_header, sizeof(accept_header) - 1, WSI_TOKEN_HTTP_ACCEPT); const char* accept_ptr = NULL; if (header_len > 0) { accept_header[header_len] = '\0'; accept_ptr = accept_header; } // Mark session as active so HTTP requests don't trigger idle ban if (pss) { pthread_mutex_lock(&pss->session_lock); pss->session_active = 1; pthread_mutex_unlock(&pss->session_lock); } if (handle_nip11_http_request(wsi, accept_ptr) == 0) { return 0; // Successfully handled } return -1; } // Check for embedded API files if (handle_embedded_file_request(wsi, requested_uri) == 0) { return 0; // Successfully handled } // Return 404 for other paths lws_return_http_status(wsi, HTTP_STATUS_NOT_FOUND, NULL); return -1; } case LWS_CALLBACK_HTTP_WRITEABLE: // Handle HTTP body transmission for NIP-11 or embedded files { void* user_data = lws_wsi_user(wsi); if (user_data) { int type = *(int*)user_data; if (type == 0) { // NIP-11 struct nip11_session_data* session_data = (struct nip11_session_data*)user_data; if (session_data->headers_sent && !session_data->body_sent) { // Allocate buffer for JSON body transmission (no LWS_PRE needed for body) unsigned char *json_buf = malloc(session_data->json_length); if (!json_buf) { DEBUG_ERROR("Failed to allocate buffer for NIP-11 body transmission"); // Clean up session data free(session_data->json_buffer); free(session_data); lws_set_wsi_user(wsi, NULL); return -1; } // Copy JSON data to buffer memcpy(json_buf, session_data->json_buffer, session_data->json_length); // Write JSON body int write_result = lws_write(wsi, json_buf, session_data->json_length, LWS_WRITE_HTTP); // Free the transmission buffer immediately (it's been copied by libwebsockets) free(json_buf); if (write_result < 0) { DEBUG_ERROR("Failed to write NIP-11 JSON body"); // Clean up session data free(session_data->json_buffer); free(session_data); lws_set_wsi_user(wsi, NULL); return -1; } // Mark body as sent and clean up session data session_data->body_sent = 1; free(session_data->json_buffer); free(session_data); lws_set_wsi_user(wsi, NULL); return 0; // Close connection after successful transmission } } else if (type == 1) { // Embedded file return handle_embedded_file_writeable(wsi); } } } break; case LWS_CALLBACK_ESTABLISHED: DEBUG_TRACE("WebSocket connection established"); memset(pss, 0, sizeof(*pss)); pthread_mutex_init(&pss->session_lock, NULL); // Initialize database query tracking pss->db_queries_executed = 0; pss->db_rows_returned = 0; pss->query_tracking_start = time(NULL); // Get real client IP address char client_ip[CLIENT_IP_MAX_LENGTH]; memset(client_ip, 0, sizeof(client_ip)); // Check if we should trust proxy headers int trust_proxy = get_config_bool("trust_proxy_headers", 0); if (trust_proxy) { // Try to get IP from X-Forwarded-For header first char x_forwarded_for[CLIENT_IP_MAX_LENGTH]; int header_len = lws_hdr_copy(wsi, x_forwarded_for, sizeof(x_forwarded_for) - 1, WSI_TOKEN_X_FORWARDED_FOR); if (header_len > 0) { x_forwarded_for[header_len] = '\0'; // X-Forwarded-For can contain multiple IPs (client, proxy1, proxy2, ...) // We want the first (leftmost) IP which is the original client char* comma = strchr(x_forwarded_for, ','); if (comma) { *comma = '\0'; // Truncate at first comma } // Trim leading/trailing whitespace char* ip_start = x_forwarded_for; while (*ip_start == ' ' || *ip_start == '\t') ip_start++; size_t ip_len = strlen(ip_start); while (ip_len > 0 && (ip_start[ip_len-1] == ' ' || ip_start[ip_len-1] == '\t')) { ip_start[--ip_len] = '\0'; } if (ip_len > 0 && ip_len < CLIENT_IP_MAX_LENGTH) { strncpy(client_ip, ip_start, CLIENT_IP_MAX_LENGTH - 1); client_ip[CLIENT_IP_MAX_LENGTH - 1] = '\0'; DEBUG_TRACE("Using X-Forwarded-For IP: %s", client_ip); } } // If X-Forwarded-For didn't work, try X-Real-IP if (client_ip[0] == '\0') { char x_real_ip[CLIENT_IP_MAX_LENGTH]; header_len = lws_hdr_copy(wsi, x_real_ip, sizeof(x_real_ip) - 1, WSI_TOKEN_HTTP_X_REAL_IP); if (header_len > 0) { x_real_ip[header_len] = '\0'; strncpy(client_ip, x_real_ip, CLIENT_IP_MAX_LENGTH - 1); client_ip[CLIENT_IP_MAX_LENGTH - 1] = '\0'; DEBUG_TRACE("Using X-Real-IP: %s", client_ip); } } } // Fall back to direct connection IP if proxy headers not available or not trusted if (client_ip[0] == '\0') { lws_get_peer_simple(wsi, client_ip, sizeof(client_ip)); DEBUG_TRACE("Using direct connection IP: %s", client_ip); } // Ensure client_ip is null-terminated and copy safely client_ip[CLIENT_IP_MAX_LENGTH - 1] = '\0'; size_t ip_len = strlen(client_ip); size_t copy_len = (ip_len < CLIENT_IP_MAX_LENGTH - 1) ? ip_len : CLIENT_IP_MAX_LENGTH - 1; memcpy(pss->client_ip, client_ip, copy_len); pss->client_ip[copy_len] = '\0'; // Record connection establishment time for duration tracking pss->connection_established = time(NULL); DEBUG_LOG("WebSocket connection established from %s", pss->client_ip); // Initialize NIP-42 authentication state pss->authenticated = 0; pss->nip42_auth_required_events = get_config_bool("nip42_auth_required_events", 0); pss->nip42_auth_required_subscriptions = get_config_bool("nip42_auth_required_subscriptions", 0); pss->auth_challenge_sent = 0; memset(pss->authenticated_pubkey, 0, sizeof(pss->authenticated_pubkey)); memset(pss->active_challenge, 0, sizeof(pss->active_challenge)); pss->challenge_created = 0; pss->challenge_expires = 0; // Mark as WebSocket connection (not HTTP) pss->is_websocket = 1; // Register in global connection list for idle tracking connection_list_add(wsi, pss); // Record connection for stats tracking ip_ban_record_connection(pss->client_ip); // Check IP ban using the resolved client IP (which may be from X-Forwarded-For). // This must happen AFTER pss->client_ip is populated so the same IP string // is used for both ban recording (at CLOSED) and ban checking (here). if (ip_ban_is_banned(pss->client_ip)) { DEBUG_LOG("Rejecting banned IP %s at connection establishment", pss->client_ip); return -1; // Close connection immediately — no challenge, no processing } // Initialize session activity tracking pss->session_active = 0; pss->idle_timeout_sec = hot_cfg_idle_connection_timeout_sec(); // Set idle timeout for ALL connections (not just auth-required) // This catches bots that connect and do nothing if (pss->idle_timeout_sec > 0) { lws_set_timeout(wsi, PENDING_TIMEOUT_AWAITING_PING, pss->idle_timeout_sec); DEBUG_TRACE("Idle timeout set: %d seconds for connection from %s", pss->idle_timeout_sec, pss->client_ip); } // Also set auth timeout if auth is required (separate concern) if (pss->nip42_auth_required_events || pss->nip42_auth_required_subscriptions) { int auth_timeout = hot_cfg_nip42_auth_timeout_sec(); // Use the shorter of the two timeouts int effective_timeout = (pss->idle_timeout_sec > 0 && pss->idle_timeout_sec < auth_timeout) ? pss->idle_timeout_sec : auth_timeout; if (effective_timeout > 0) { lws_set_timeout(wsi, PENDING_TIMEOUT_AWAITING_PING, effective_timeout); DEBUG_TRACE("Auth timeout set: %d seconds for unauthenticated connection from %s", effective_timeout, pss->client_ip); } } DEBUG_TRACE("WebSocket connection initialization complete"); break; case LWS_CALLBACK_RECEIVE: if (len > 0) { DEBUG_TRACE("LWS_CALLBACK_RECEIVE: received %zu bytes", len); // Check if client is rate limited for malformed requests if (is_client_rate_limited_for_malformed_requests(pss)) { send_notice_message(wsi, pss, "error: too many malformed requests - temporarily blocked"); return 0; } // Check if this is a fragmented message int is_first_fragment = lws_is_first_fragment(wsi); int is_final_fragment = lws_is_final_fragment(wsi); size_t remaining_payload = lws_remaining_packet_payload(wsi); DEBUG_TRACE("Fragment info: first=%d, final=%d, remaining=%zu, reassembly_active=%d", is_first_fragment, is_final_fragment, remaining_payload, pss->reassembly_active); // Handle message reassembly for fragmented messages // Only use reassembly if message is actually fragmented (not both first and final) int is_fragmented = (is_first_fragment && !is_final_fragment) || pss->reassembly_active; if (is_fragmented) { // Start or continue reassembly if (is_first_fragment) { // First fragment - initialize reassembly buffer if (pss->reassembly_buffer) { DEBUG_WARN("Starting new reassembly but buffer already exists - cleaning up"); free(pss->reassembly_buffer); pss->reassembly_buffer = NULL; // Ensure buffer is NULL after cleanup } pss->reassembly_size = 0; pss->reassembly_capacity = 0; pss->reassembly_active = 1; DEBUG_TRACE("Starting message reassembly"); } // Ensure buffer has enough capacity size_t needed_capacity = pss->reassembly_size + len + 1; // +1 for null terminator if (needed_capacity > pss->reassembly_capacity) { size_t new_capacity = pss->reassembly_capacity == 0 ? 8192 : pss->reassembly_capacity * 2; while (new_capacity < needed_capacity) { new_capacity *= 2; } char* new_buffer = realloc(pss->reassembly_buffer, new_capacity); if (!new_buffer) { DEBUG_ERROR("Failed to allocate reassembly buffer (capacity %zu)", new_capacity); // Clean up and abort reassembly free(pss->reassembly_buffer); pss->reassembly_buffer = NULL; pss->reassembly_size = 0; pss->reassembly_capacity = 0; pss->reassembly_active = 0; send_notice_message(wsi, pss, "error: message too large - memory allocation failed"); return 0; } pss->reassembly_buffer = new_buffer; pss->reassembly_capacity = new_capacity; DEBUG_TRACE("Expanded reassembly buffer to %zu bytes", new_capacity); } // Append fragment to buffer memcpy(pss->reassembly_buffer + pss->reassembly_size, in, len); pss->reassembly_size += len; // Check if this is the final fragment if (is_final_fragment) { // Message complete - process it pss->reassembly_buffer[pss->reassembly_size] = '\0'; pss->reassembly_active = 0; DEBUG_TRACE("Message reassembly complete: total size %zu bytes", pss->reassembly_size); // Process the complete message char* complete_message = pss->reassembly_buffer; size_t message_len = pss->reassembly_size; // Reset reassembly state (keep buffer pointer for processing) pss->reassembly_size = 0; pss->reassembly_active = 0; // Parse JSON message DEBUG_TRACE("Parsing reassembled JSON message of length %zu", message_len); cJSON* json = cJSON_Parse(complete_message); // Process the message (same logic as before) if (json && cJSON_IsArray(json)) { // Get message type cJSON* type = cJSON_GetArrayItem(json, 0); if (type && cJSON_IsString(type)) { const char* msg_type = cJSON_GetStringValue(type); if (strcmp(msg_type, "EVENT") == 0) { // Mark session as active - client sent EVENT pthread_mutex_lock(&pss->session_lock); pss->session_active = 1; pthread_mutex_unlock(&pss->session_lock); // Cancel idle timeout - this is legitimate usage lws_set_timeout(wsi, NO_PENDING_TIMEOUT, 0); // Handle EVENT message cJSON* event = cJSON_GetArrayItem(json, 1); if (event && cJSON_IsObject(event)) { // Extract event JSON string for unified validator char *event_json_str = cJSON_Print(event); if (!event_json_str) { DEBUG_ERROR("Failed to serialize event JSON for validation"); cJSON* error_response = cJSON_CreateArray(); cJSON_AddItemToArray(error_response, cJSON_CreateString("OK")); cJSON_AddItemToArray(error_response, cJSON_CreateString("unknown")); cJSON_AddItemToArray(error_response, cJSON_CreateBool(0)); cJSON_AddItemToArray(error_response, cJSON_CreateString("error: failed to process event")); char *error_str = cJSON_Print(error_response); if (error_str) { size_t error_len = strlen(error_str); // Use proper message queue system instead of direct lws_write if (queue_message(wsi, pss, error_str, error_len, LWS_WRITE_TEXT) != 0) { DEBUG_ERROR("Failed to queue error response message"); } free(error_str); } cJSON_Delete(error_response); cJSON_Delete(json); // Note: complete_message points to reassembly_buffer, which is managed separately // and should not be freed here - it will be cleaned up in LWS_CALLBACK_CLOSED return 0; } // Try async EVENT offload first (Phase 1). int async_submit_rc = try_submit_async_event(event, event_json_str, wsi, pss); if (async_submit_rc == 0) { free(event_json_str); cJSON_Delete(json); return 0; } if (async_submit_rc < 0) { cJSON* event_id = cJSON_GetObjectItemCaseSensitive(event, "id"); if (event_id && cJSON_IsString(event_id)) { const char* async_err = (async_submit_rc == -2) ? "error: async event worker unavailable" : "error: async event queue unavailable"; send_ok_response(wsi, pss, cJSON_GetStringValue(event_id), 0, async_err); } free(event_json_str); cJSON_Delete(json); return 0; } // Call unified validator with JSON string (sync path for special ineligible events) size_t event_json_len = strlen(event_json_str); int validation_result = nostr_validate_unified_request(event_json_str, event_json_len); // Map validation result to old result format (0 = success, -1 = failure) int result = (validation_result == NOSTR_SUCCESS) ? 0 : -1; // Generate error message based on validation result char error_message[512] = {0}; if (result != 0) { switch (validation_result) { case NOSTR_ERROR_INVALID_INPUT: strncpy(error_message, "invalid: malformed event structure", sizeof(error_message) - 1); break; case NOSTR_ERROR_EVENT_INVALID_SIGNATURE: strncpy(error_message, "invalid: signature verification failed", sizeof(error_message) - 1); break; case NOSTR_ERROR_EVENT_INVALID_ID: strncpy(error_message, "invalid: event id verification failed", sizeof(error_message) - 1); break; case NOSTR_ERROR_EVENT_INVALID_PUBKEY: strncpy(error_message, "invalid: invalid pubkey format", sizeof(error_message) - 1); break; case -103: // NOSTR_ERROR_EVENT_EXPIRED strncpy(error_message, "rejected: event expired", sizeof(error_message) - 1); break; case -102: // NOSTR_ERROR_NIP42_DISABLED strncpy(error_message, "auth-required: NIP-42 authentication required", sizeof(error_message) - 1); break; case -101: // NOSTR_ERROR_AUTH_REQUIRED strncpy(error_message, "blocked: pubkey not authorized", sizeof(error_message) - 1); break; default: strncpy(error_message, "error: validation failed", sizeof(error_message) - 1); break; } } // Cleanup event JSON string free(event_json_str); // Check for NIP-70 protected events if (result == 0) { // Check if event has protected tag ["-"] int is_protected_event = 0; cJSON* tags = cJSON_GetObjectItemCaseSensitive(event, "tags"); if (tags && cJSON_IsArray(tags)) { cJSON* tag = NULL; cJSON_ArrayForEach(tag, tags) { if (cJSON_IsArray(tag) && cJSON_GetArraySize(tag) >= 1) { cJSON* tag_name = cJSON_GetArrayItem(tag, 0); if (tag_name && cJSON_IsString(tag_name) && strcmp(cJSON_GetStringValue(tag_name), "-") == 0) { is_protected_event = 1; break; } } } } if (is_protected_event) { // Check if protected events are enabled using hot-path cache int protected_events_enabled = hot_cfg_nip70_protected_events_enabled(); if (!protected_events_enabled) { // Protected events not supported result = -1; strncpy(error_message, "blocked: protected events not supported", sizeof(error_message) - 1); error_message[sizeof(error_message) - 1] = '\0'; DEBUG_WARN("Protected event rejected: protected events not enabled"); } else { // Protected events enabled - check authentication cJSON* pubkey_obj = cJSON_GetObjectItemCaseSensitive(event, "pubkey"); const char* event_pubkey = pubkey_obj ? cJSON_GetStringValue(pubkey_obj) : NULL; if (!pss || !pss->authenticated || !event_pubkey || strcmp(pss->authenticated_pubkey, event_pubkey) != 0) { // Not authenticated or pubkey mismatch result = -1; strncpy(error_message, "auth-required: protected event requires authentication", sizeof(error_message) - 1); error_message[sizeof(error_message) - 1] = '\0'; DEBUG_WARN("Protected event rejected: authentication required"); } } } } // Check for admin events (kind 23456) and intercept them if (result == 0) { cJSON* kind_obj = cJSON_GetObjectItemCaseSensitive(event, "kind"); if (kind_obj && cJSON_IsNumber(kind_obj)) { int event_kind = (int)cJSON_GetNumberValue(kind_obj); DEBUG_TRACE("Processing event kind %d, message length: %zu", event_kind, message_len); // Log reception of Kind 23456 events if (event_kind == 23456) { DEBUG_LOG("Admin event (kind 23456) received"); } if (event_kind == 23456) { // Enhanced admin event security - check authorization first char auth_error[512] = {0}; int auth_result = is_authorized_admin_event(event, auth_error, sizeof(auth_error)); if (auth_result != 0) { // Authorization failed - log and reject DEBUG_WARN("Admin event authorization failed"); result = -1; size_t error_len = strlen(auth_error); size_t copy_len = (error_len < sizeof(error_message) - 1) ? error_len : sizeof(error_message) - 1; memcpy(error_message, auth_error, copy_len); error_message[copy_len] = '\0'; } else { // Authorization successful - process through admin API char admin_error[512] = {0}; int admin_result = process_admin_event_in_config(event, admin_error, sizeof(admin_error), wsi); // Log results for Kind 23456 events if (event_kind == 23456) { if (admin_result != 0) { char error_result_msg[512]; if (strlen(admin_error) > 0) { // Safely truncate admin_error if too long size_t max_error_len = sizeof(error_result_msg) - 50; // Leave room for prefix size_t error_len = strlen(admin_error); if (error_len > max_error_len) { error_len = max_error_len; } char truncated_error[512]; memcpy(truncated_error, admin_error, error_len); truncated_error[error_len] = '\0'; // Use a safer approach to avoid truncation warning size_t prefix_len = snprintf(error_result_msg, sizeof(error_result_msg), "ERROR: Kind %d event processing failed: ", event_kind); if (prefix_len < sizeof(error_result_msg)) { size_t remaining = sizeof(error_result_msg) - prefix_len; size_t copy_len = strlen(truncated_error); if (copy_len >= remaining) { copy_len = remaining - 1; } memcpy(error_result_msg + prefix_len, truncated_error, copy_len); error_result_msg[prefix_len + copy_len] = '\0'; } } else { snprintf(error_result_msg, sizeof(error_result_msg), "ERROR: Kind %d event processing failed", event_kind); } DEBUG_ERROR(error_result_msg); } } if (admin_result != 0) { DEBUG_ERROR("Failed to process admin event"); result = -1; size_t error_len = strlen(admin_error); size_t copy_len = (error_len < sizeof(error_message) - 1) ? error_len : sizeof(error_message) - 1; memcpy(error_message, admin_error, copy_len); error_message[copy_len] = '\0'; } else { // Admin events are processed by the admin API, not broadcast to subscriptions } } } else if (event_kind == 1059) { // NIP-17 gift wrap events // Admin DM processing is opt-in via nip17_admin_enabled config (default: off) // to prevent expensive decryption on every incoming gift wrap event int nip17_admin_enabled = hot_cfg_nip17_admin_enabled(); if (nip17_admin_enabled) { char nip17_error[512] = {0}; cJSON* response_event = process_nip17_admin_message(event, nip17_error, sizeof(nip17_error), wsi); if (!response_event) { // Check if this is an error or if the command was already handled if (strlen(nip17_error) > 0) { // There was an actual error DEBUG_ERROR("NIP-17 admin message processing failed"); result = -1; size_t error_len = strlen(nip17_error); size_t copy_len = (error_len < sizeof(error_message) - 1) ? error_len : sizeof(error_message) - 1; memcpy(error_message, nip17_error, copy_len); error_message[copy_len] = '\0'; } else { // No error message means the command was already handled (plain text commands) // Store the original gift wrap event in database if (store_event(event) != 0) { DEBUG_ERROR("Failed to store gift wrap event in database"); result = -1; strncpy(error_message, "error: failed to store gift wrap event", sizeof(error_message) - 1); } } } else { // Store the original gift wrap event in database (unlike kind 23456) if (store_event(event) != 0) { DEBUG_ERROR("Failed to store gift wrap event in database"); result = -1; strncpy(error_message, "error: failed to store gift wrap event", sizeof(error_message) - 1); cJSON_Delete(response_event); } else { // Broadcast RESPONSE event to matching persistent subscriptions broadcast_event_to_subscriptions(response_event); // Clean up response event cJSON_Delete(response_event); } } } else { // NIP-17 admin DMs disabled: store gift wrap as a regular event DEBUG_TRACE("NIP-17 admin DMs disabled - storing kind 1059 event without decryption"); if (store_event(event) != 0) { DEBUG_ERROR("Failed to store gift wrap event in database"); result = -1; strncpy(error_message, "error: failed to store event", sizeof(error_message) - 1); } else { broadcast_event_to_subscriptions(event); } } } else if (event_kind == 14) { // Check for DM stats commands addressed to relay char dm_error[512] = {0}; int dm_result = process_dm_stats_command(event, dm_error, sizeof(dm_error), wsi); if (dm_result != 0) { DEBUG_ERROR("DM stats command processing failed"); result = -1; size_t error_len = strlen(dm_error); size_t copy_len = (error_len < sizeof(error_message) - 1) ? error_len : sizeof(error_message) - 1; memcpy(error_message, dm_error, copy_len); error_message[copy_len] = '\0'; } else { // Store the DM event in database if (store_event(event) != 0) { DEBUG_ERROR("Failed to store DM event in database"); result = -1; strncpy(error_message, "error: failed to store DM event", sizeof(error_message) - 1); } else { // Broadcast DM event to matching persistent subscriptions broadcast_event_to_subscriptions(event); } } } else { // Check if this is an ephemeral event (kinds 20000-29999) // Per NIP-01: ephemeral events are broadcast but never stored if (event_kind >= 20000 && event_kind < 30000) { DEBUG_TRACE("Ephemeral event (kind %d) - broadcasting without storage", event_kind); // Broadcast directly to subscriptions without database storage broadcast_event_to_subscriptions(event); } else { DEBUG_TRACE("Storing regular event in database"); // Regular event - store in database and broadcast if (store_event(event) != 0) { DEBUG_ERROR("Failed to store event in database"); result = -1; strncpy(error_message, "error: failed to store event", sizeof(error_message) - 1); } else { DEBUG_LOG("Event stored and broadcast (kind %d)", event_kind); // Broadcast event to matching persistent subscriptions broadcast_event_to_subscriptions(event); } } } } else { // Event without valid kind - try normal storage DEBUG_WARN("Event without valid kind - trying normal storage"); if (store_event(event) != 0) { DEBUG_ERROR("Failed to store event without kind in database"); result = -1; strncpy(error_message, "error: failed to store event", sizeof(error_message) - 1); } else { broadcast_event_to_subscriptions(event); } } } // Send OK response cJSON* event_id = cJSON_GetObjectItemCaseSensitive(event, "id"); if (event_id && cJSON_IsString(event_id)) { cJSON* response = cJSON_CreateArray(); cJSON_AddItemToArray(response, cJSON_CreateString("OK")); cJSON_AddItemToArray(response, cJSON_CreateString(cJSON_GetStringValue(event_id))); cJSON_AddItemToArray(response, cJSON_CreateBool(result == 0)); cJSON_AddItemToArray(response, cJSON_CreateString(strlen(error_message) > 0 ? error_message : "")); char *response_str = cJSON_Print(response); if (response_str) { size_t response_len = strlen(response_str); // DEBUG: Log WebSocket frame details before sending DEBUG_TRACE("WS_FRAME_SEND: type=OK len=%zu data=%.100s%s", response_len, response_str, response_len > 100 ? "..." : ""); // Queue message for proper libwebsockets pattern if (queue_message(wsi, pss, response_str, response_len, LWS_WRITE_TEXT) != 0) { DEBUG_ERROR("Failed to queue OK response message"); } free(response_str); } cJSON_Delete(response); } } } else if (strcmp(msg_type, "REQ") == 0) { DEBUG_TRACE("REQ message received, starting processing"); // Check NIP-42 authentication for REQ subscriptions if required if (pss && pss->nip42_auth_required_subscriptions && !pss->authenticated) { DEBUG_TRACE("REQ rejected: NIP-42 authentication required"); if (!pss->auth_challenge_sent) { send_nip42_auth_challenge(wsi, pss); } else { send_notice_message(wsi, pss, "NIP-42 authentication required for subscriptions"); DEBUG_WARN("REQ rejected: NIP-42 authentication required"); // Auth timeout: close connection if challenge was sent but client // hasn't authenticated within nip42_auth_timeout_sec seconds int auth_timeout = hot_cfg_nip42_auth_timeout_sec(); if (auth_timeout > 0 && pss->connection_established > 0) { time_t connection_age = time(NULL) - pss->connection_established; if (connection_age >= auth_timeout) { DEBUG_LOG("Closing unauthenticated connection from %s after %ld seconds (timeout=%d)", pss->client_ip, connection_age, auth_timeout); lws_close_reason(wsi, LWS_CLOSE_STATUS_POLICY_VIOLATION, (unsigned char*)"Authentication timeout", 22); cJSON_Delete(json); return -1; } } } cJSON_Delete(json); // Note: complete_message points to reassembly_buffer, which is managed separately // and should not be freed here - it will be cleaned up in LWS_CALLBACK_CLOSED return 0; } // WoT read restriction check (wot_enabled == 2) // After NIP-42 auth, check if authenticated pubkey is in WoT whitelist if (pss && pss->authenticated && hot_cfg_wot_enabled() == 2) { // Client is authenticated - check if their pubkey is in the WoT whitelist extern int check_database_auth_rules(const char* pubkey, const char* operation, const char* resource_hash); int wot_result = check_database_auth_rules(pss->authenticated_pubkey, "subscription", NULL); if (wot_result != 0) { send_notice_message(wsi, pss, "restricted: your pubkey is not in this relay's web of trust"); DEBUG_INFO("REQ rejected: pubkey %s not in WoT whitelist", pss->authenticated_pubkey); cJSON_Delete(json); return 0; } } // Mark session as active - client sent REQ pthread_mutex_lock(&pss->session_lock); pss->session_active = 1; pthread_mutex_unlock(&pss->session_lock); // Cancel idle timeout - this is legitimate usage lws_set_timeout(wsi, NO_PENDING_TIMEOUT, 0); DEBUG_TRACE("REQ message passed authentication check"); // Handle REQ message cJSON* sub_id = cJSON_GetArrayItem(json, 1); if (sub_id && cJSON_IsString(sub_id)) { const char* subscription_id = cJSON_GetStringValue(sub_id); DEBUG_TRACE("Processing REQ message for subscription %s", subscription_id); // Validate subscription ID before processing if (!subscription_id) { DEBUG_TRACE("REQ rejected: NULL subscription ID"); send_notice_message(wsi, pss, "error: invalid subscription ID"); DEBUG_WARN("REQ rejected: NULL subscription ID"); record_malformed_request(pss); cJSON_Delete(json); // Note: complete_message points to reassembly_buffer, which is managed separately // and should not be freed here - it will be cleaned up in LWS_CALLBACK_CLOSED return 0; } // Validate subscription ID if (!validate_subscription_id(subscription_id)) { DEBUG_TRACE("REQ rejected: invalid subscription ID format"); send_notice_message(wsi, pss, "error: invalid subscription ID"); DEBUG_WARN("REQ rejected: invalid subscription ID"); cJSON_Delete(json); // Note: complete_message points to reassembly_buffer, which is managed separately // and should not be freed here - it will be cleaned up in LWS_CALLBACK_CLOSED return 0; } DEBUG_TRACE("REQ subscription ID validated: %s", subscription_id); // Create array of filter objects from position 2 onwards cJSON* filters = cJSON_CreateArray(); if (!filters) { DEBUG_TRACE("REQ failed: could not create filters array"); send_notice_message(wsi, pss, "error: failed to process filters"); DEBUG_ERROR("REQ failed: could not create filters array"); cJSON_Delete(json); // Note: complete_message points to reassembly_buffer, which is managed separately // and should not be freed here - it will be cleaned up in LWS_CALLBACK_CLOSED return 0; } int json_size = cJSON_GetArraySize(json); int filter_count = 0; for (int i = 2; i < json_size; i++) { cJSON* filter = cJSON_GetArrayItem(json, i); if (filter) { cJSON_AddItemToArray(filters, cJSON_Duplicate(filter, 1)); filter_count++; } } DEBUG_TRACE("REQ created %d filters from message", filter_count); // Validate filters before processing char filter_error[512] = {0}; int validation_result = validate_filter_array(filters, filter_error, sizeof(filter_error)); if (validation_result <= 0) { DEBUG_TRACE("REQ rejected: filter validation failed - %s", filter_error); send_notice_message(wsi, pss, filter_error); DEBUG_WARN("REQ rejected: invalid filters"); // Only record as malformed if it's a true error (0), not benign error (-1) if (validation_result == 0) { record_malformed_request(pss); } cJSON_Delete(filters); cJSON_Delete(json); // Note: complete_message points to reassembly_buffer, which is managed separately // and should not be freed here - it will be cleaned up in LWS_CALLBACK_CLOSED return 0; } DEBUG_TRACE("REQ filters validated successfully"); DEBUG_TRACE("About to call handle_req_message for subscription %s", subscription_id); int req_result = handle_req_message(subscription_id, filters, wsi, pss); DEBUG_TRACE("handle_req_message completed for subscription %s", subscription_id); // Clean up the filters array we created cJSON_Delete(filters); // Async path will send EOSE from completion handler. if (req_result != HANDLE_REQ_ASYNC_PENDING) { DEBUG_LOG("REQ subscription %s processed, sending EOSE", subscription_id); // Send EOSE (End of Stored Events) cJSON* eose_response = cJSON_CreateArray(); if (eose_response) { cJSON_AddItemToArray(eose_response, cJSON_CreateString("EOSE")); cJSON_AddItemToArray(eose_response, cJSON_CreateString(subscription_id)); char *eose_str = cJSON_Print(eose_response); if (eose_str) { size_t eose_len = strlen(eose_str); // DEBUG: Log WebSocket frame details before sending DEBUG_TRACE("WS_FRAME_SEND: type=EOSE len=%zu data=%.100s%s", eose_len, eose_str, eose_len > 100 ? "..." : ""); // Queue message for proper libwebsockets pattern if (queue_message(wsi, pss, eose_str, eose_len, LWS_WRITE_TEXT) != 0) { DEBUG_ERROR("Failed to queue EOSE message"); } free(eose_str); } cJSON_Delete(eose_response); } } } else { send_notice_message(wsi, pss, "error: missing or invalid subscription ID in REQ"); DEBUG_WARN("REQ rejected: missing or invalid subscription ID"); } } else if (strcmp(msg_type, "COUNT") == 0) { // Check NIP-42 authentication for COUNT requests if required if (pss && pss->nip42_auth_required_subscriptions && !pss->authenticated) { if (!pss->auth_challenge_sent) { send_nip42_auth_challenge(wsi, pss); } else { send_notice_message(wsi, pss, "NIP-42 authentication required for count requests"); DEBUG_WARN("COUNT rejected: NIP-42 authentication required"); } cJSON_Delete(json); // Note: complete_message points to reassembly_buffer, which is managed separately // and should not be freed here - it will be cleaned up in LWS_CALLBACK_CLOSED return 0; } // Handle COUNT message cJSON* sub_id = cJSON_GetArrayItem(json, 1); if (sub_id && cJSON_IsString(sub_id)) { const char* subscription_id = cJSON_GetStringValue(sub_id); // Create array of filter objects from position 2 onwards cJSON* filters = cJSON_CreateArray(); int json_size = cJSON_GetArraySize(json); for (int i = 2; i < json_size; i++) { cJSON* filter = cJSON_GetArrayItem(json, i); if (filter) { cJSON_AddItemToArray(filters, cJSON_Duplicate(filter, 1)); } } // Validate filters before processing char filter_error[512] = {0}; int validation_result = validate_filter_array(filters, filter_error, sizeof(filter_error)); if (validation_result <= 0) { send_notice_message(wsi, pss, filter_error); DEBUG_WARN("COUNT rejected: invalid filters"); // Only record as malformed if it's a true error (0), not benign error (-1) if (validation_result == 0) { record_malformed_request(pss); } cJSON_Delete(filters); cJSON_Delete(json); // Note: complete_message points to reassembly_buffer, which is managed separately // and should not be freed here - it will be cleaned up in LWS_CALLBACK_CLOSED return 0; } int count_result = handle_count_message(subscription_id, filters, wsi, pss); // Clean up the filters array we created cJSON_Delete(filters); if (count_result == HANDLE_REQ_ASYNC_PENDING) { cJSON_Delete(json); return 0; } } } else if (strcmp(msg_type, "CLOSE") == 0) { // Handle CLOSE message cJSON* sub_id = cJSON_GetArrayItem(json, 1); if (sub_id && cJSON_IsString(sub_id)) { const char* subscription_id = cJSON_GetStringValue(sub_id); // Validate subscription ID before processing if (!subscription_id) { send_notice_message(wsi, pss, "error: invalid subscription ID in CLOSE"); DEBUG_WARN("CLOSE rejected: NULL subscription ID"); cJSON_Delete(json); // Note: complete_message points to reassembly_buffer, which is managed separately // and should not be freed here - it will be cleaned up in LWS_CALLBACK_CLOSED return 0; } // Validate subscription ID if (!validate_subscription_id(subscription_id)) { send_notice_message(wsi, pss, "error: invalid subscription ID in CLOSE"); DEBUG_WARN("CLOSE rejected: invalid subscription ID"); cJSON_Delete(json); // Note: complete_message points to reassembly_buffer, which is managed separately // and should not be freed here - it will be cleaned up in LWS_CALLBACK_CLOSED return 0; } // CRITICAL FIX: Mark subscription as inactive in global manager FIRST // This prevents other threads from accessing it during removal pthread_mutex_lock(&g_subscription_manager.subscriptions_lock); subscription_t* target_sub = g_subscription_manager.active_subscriptions; while (target_sub) { if (strcmp(target_sub->id, subscription_id) == 0 && target_sub->wsi == wsi) { target_sub->active = 0; // Mark as inactive immediately break; } target_sub = target_sub->next; } pthread_mutex_unlock(&g_subscription_manager.subscriptions_lock); // Now safe to remove from session list if (pss) { pthread_mutex_lock(&pss->session_lock); struct subscription** current = &pss->subscriptions; while (*current) { if (strcmp((*current)->id, subscription_id) == 0) { struct subscription* to_remove = *current; *current = to_remove->session_next; pss->subscription_count--; break; } current = &((*current)->session_next); } pthread_mutex_unlock(&pss->session_lock); } // Finally remove from global manager (which will free it) remove_subscription_from_manager(subscription_id, wsi); // Subscription closed } else { send_notice_message(wsi, pss, "error: missing or invalid subscription ID in CLOSE"); DEBUG_WARN("CLOSE rejected: missing or invalid subscription ID"); } } else if (strcmp(msg_type, "AUTH") == 0) { // Handle NIP-42 AUTH message if (cJSON_GetArraySize(json) >= 2) { cJSON* auth_payload = cJSON_GetArrayItem(json, 1); if (cJSON_IsString(auth_payload)) { // AUTH challenge response: ["AUTH", ] (unusual) handle_nip42_auth_challenge_response(wsi, pss, cJSON_GetStringValue(auth_payload)); } else if (cJSON_IsObject(auth_payload)) { // AUTH signed event: ["AUTH", ] (standard NIP-42) handle_nip42_auth_signed_event(wsi, pss, auth_payload); } else { send_notice_message(wsi, pss, "Invalid AUTH message format"); DEBUG_WARN("Received AUTH message with invalid payload type"); } } else { send_notice_message(wsi, pss, "AUTH message requires payload"); DEBUG_WARN("Received AUTH message without payload"); } } else { // Unknown message type char unknown_msg[128]; snprintf(unknown_msg, sizeof(unknown_msg), "Unknown message type: %.32s", msg_type); DEBUG_WARN(unknown_msg); send_notice_message(wsi, pss, "Unknown message type"); } } } // Clean up the reassembled message if (json) cJSON_Delete(json); // Free the reassembly buffer now that processing is complete free(complete_message); pss->reassembly_buffer = NULL; pss->reassembly_capacity = 0; return 0; // Fragmented message processed } else { // Not the final fragment - continue accumulating DEBUG_TRACE("Accumulated %zu bytes so far, waiting for more fragments", pss->reassembly_size); return 0; } } // Handle non-fragmented messages (original code path) char *message = malloc(len + 1); if (message) { memcpy(message, in, len); message[len] = '\0'; // Parse JSON message (this is the normal program flow) DEBUG_TRACE("Parsing JSON message of length %zu", strlen(message)); cJSON* json = cJSON_Parse(message); if (json && cJSON_IsArray(json)) { // Get message type cJSON* type = cJSON_GetArrayItem(json, 0); if (type && cJSON_IsString(type)) { const char* msg_type = cJSON_GetStringValue(type); if (strcmp(msg_type, "EVENT") == 0) { // Mark session as active - client sent EVENT pthread_mutex_lock(&pss->session_lock); pss->session_active = 1; pthread_mutex_unlock(&pss->session_lock); // Cancel idle timeout - this is legitimate usage lws_set_timeout(wsi, NO_PENDING_TIMEOUT, 0); // Extract event for kind-specific NIP-42 authentication check cJSON* event_obj = cJSON_GetArrayItem(json, 1); if (event_obj && cJSON_IsObject(event_obj)) { // Extract event kind for kind-specific NIP-42 authentication check cJSON* kind_obj = cJSON_GetObjectItemCaseSensitive(event_obj, "kind"); int event_kind = kind_obj && cJSON_IsNumber(kind_obj) ? (int)cJSON_GetNumberValue(kind_obj) : -1; // Extract pubkey for debugging cJSON* pubkey_obj = cJSON_GetObjectItemCaseSensitive(event_obj, "pubkey"); const char* event_pubkey = pubkey_obj ? cJSON_GetStringValue(pubkey_obj) : "unknown"; // Check if NIP-42 authentication is required for this event kind or globally int auth_required = is_nip42_auth_globally_required() || is_nip42_auth_required_for_kind(event_kind); // Special case: allow kind 14 DMs addressed to relay to bypass auth (admin commands) int bypass_auth = 0; if (event_kind == 14 && event_obj && cJSON_IsObject(event_obj)) { cJSON* tags = cJSON_GetObjectItemCaseSensitive(event_obj, "tags"); if (tags && cJSON_IsArray(tags)) { const char* relay_pubkey = get_config_value("relay_pubkey"); if (relay_pubkey) { cJSON* tag = NULL; cJSON_ArrayForEach(tag, tags) { if (cJSON_IsArray(tag) && cJSON_GetArraySize(tag) >= 2) { cJSON* tag_name = cJSON_GetArrayItem(tag, 0); cJSON* tag_value = cJSON_GetArrayItem(tag, 1); if (tag_name && cJSON_IsString(tag_name) && strcmp(cJSON_GetStringValue(tag_name), "p") == 0 && tag_value && cJSON_IsString(tag_value) && strcmp(cJSON_GetStringValue(tag_value), relay_pubkey) == 0) { bypass_auth = 1; break; } } } free((char*)relay_pubkey); } } } // Special case: allow kind 23456 admin events from authorized admin to bypass auth if (event_kind == 23456 && event_pubkey) { if (is_admin_pubkey(event_pubkey)) { bypass_auth = 1; } else { DEBUG_INFO("DEBUG: Kind 23456 event but pubkey mismatch or no admin pubkey"); } } if (pss && auth_required && !pss->authenticated && !bypass_auth) { if (!pss->auth_challenge_sent) { send_nip42_auth_challenge(wsi, pss); } else { char auth_msg[256]; if (event_kind == 4 || event_kind == 14) { snprintf(auth_msg, sizeof(auth_msg), "NIP-42 authentication required for direct message events (kind %d)", event_kind); } else { snprintf(auth_msg, sizeof(auth_msg), "NIP-42 authentication required for event kind %d", event_kind); } send_notice_message(wsi, pss, auth_msg); DEBUG_WARN("Event rejected: NIP-42 authentication required for kind"); } cJSON_Delete(json); free(message); return 0; } } // Handle EVENT message cJSON* event = cJSON_GetArrayItem(json, 1); if (event && cJSON_IsObject(event)) { // Extract event JSON string for unified validator char *event_json_str = cJSON_Print(event); if (!event_json_str) { DEBUG_ERROR("Failed to serialize event JSON for validation"); cJSON* error_response = cJSON_CreateArray(); cJSON_AddItemToArray(error_response, cJSON_CreateString("OK")); cJSON_AddItemToArray(error_response, cJSON_CreateString("unknown")); cJSON_AddItemToArray(error_response, cJSON_CreateBool(0)); cJSON_AddItemToArray(error_response, cJSON_CreateString("error: failed to process event")); char *error_str = cJSON_Print(error_response); if (error_str) { size_t error_len = strlen(error_str); // Use proper message queue system instead of direct lws_write if (queue_message(wsi, pss, error_str, error_len, LWS_WRITE_TEXT) != 0) { DEBUG_ERROR("Failed to queue error response message"); } free(error_str); } cJSON_Delete(error_response); return 0; } // Try async EVENT offload first (Phase 1). int async_submit_rc = try_submit_async_event(event, event_json_str, wsi, pss); if (async_submit_rc == 0) { free(event_json_str); cJSON_Delete(json); free(message); return 0; } if (async_submit_rc < 0) { cJSON* event_id = cJSON_GetObjectItemCaseSensitive(event, "id"); if (event_id && cJSON_IsString(event_id)) { const char* async_err = (async_submit_rc == -2) ? "error: async event worker unavailable" : "error: async event queue unavailable"; send_ok_response(wsi, pss, cJSON_GetStringValue(event_id), 0, async_err); } free(event_json_str); cJSON_Delete(json); free(message); return 0; } // Call unified validator with JSON string (sync path for special ineligible events) size_t event_json_len = strlen(event_json_str); int validation_result = nostr_validate_unified_request(event_json_str, event_json_len); // Map validation result to old result format (0 = success, -1 = failure) int result = (validation_result == NOSTR_SUCCESS) ? 0 : -1; // Generate error message based on validation result char error_message[512] = {0}; if (result != 0) { switch (validation_result) { case NOSTR_ERROR_INVALID_INPUT: strncpy(error_message, "invalid: malformed event structure", sizeof(error_message) - 1); break; case NOSTR_ERROR_EVENT_INVALID_SIGNATURE: strncpy(error_message, "invalid: signature verification failed", sizeof(error_message) - 1); break; case NOSTR_ERROR_EVENT_INVALID_ID: strncpy(error_message, "invalid: event id verification failed", sizeof(error_message) - 1); break; case NOSTR_ERROR_EVENT_INVALID_PUBKEY: strncpy(error_message, "invalid: invalid pubkey format", sizeof(error_message) - 1); break; case -103: // NOSTR_ERROR_EVENT_EXPIRED strncpy(error_message, "rejected: event expired", sizeof(error_message) - 1); break; case -102: // NOSTR_ERROR_NIP42_DISABLED strncpy(error_message, "auth-required: NIP-42 authentication required", sizeof(error_message) - 1); break; case -101: // NOSTR_ERROR_AUTH_REQUIRED strncpy(error_message, "blocked: pubkey not authorized", sizeof(error_message) - 1); break; default: strncpy(error_message, "error: validation failed", sizeof(error_message) - 1); break; } } // Cleanup event JSON string free(event_json_str); // Check for NIP-70 protected events if (result == 0) { // Check if event has protected tag ["-"] int is_protected_event = 0; cJSON* tags = cJSON_GetObjectItemCaseSensitive(event, "tags"); if (tags && cJSON_IsArray(tags)) { cJSON* tag = NULL; cJSON_ArrayForEach(tag, tags) { if (cJSON_IsArray(tag) && cJSON_GetArraySize(tag) >= 1) { cJSON* tag_name = cJSON_GetArrayItem(tag, 0); if (tag_name && cJSON_IsString(tag_name) && strcmp(cJSON_GetStringValue(tag_name), "-") == 0) { is_protected_event = 1; break; } } } } if (is_protected_event) { // Check if protected events are enabled using hot-path cache int protected_events_enabled = hot_cfg_nip70_protected_events_enabled(); if (!protected_events_enabled) { // Protected events not supported result = -1; strncpy(error_message, "blocked: protected events not supported", sizeof(error_message) - 1); error_message[sizeof(error_message) - 1] = '\0'; DEBUG_WARN("Protected event rejected: protected events not enabled"); } else { // Protected events enabled - check authentication cJSON* pubkey_obj = cJSON_GetObjectItemCaseSensitive(event, "pubkey"); const char* event_pubkey = pubkey_obj ? cJSON_GetStringValue(pubkey_obj) : NULL; if (!pss || !pss->authenticated || !event_pubkey || strcmp(pss->authenticated_pubkey, event_pubkey) != 0) { // Not authenticated or pubkey mismatch result = -1; strncpy(error_message, "auth-required: protected event requires authentication", sizeof(error_message) - 1); error_message[sizeof(error_message) - 1] = '\0'; DEBUG_WARN("Protected event rejected: authentication required"); } } } } // Check for admin events (kind 23456) and intercept them if (result == 0) { cJSON* kind_obj = cJSON_GetObjectItemCaseSensitive(event, "kind"); if (kind_obj && cJSON_IsNumber(kind_obj)) { int event_kind = (int)cJSON_GetNumberValue(kind_obj); DEBUG_TRACE("Processing event kind %d, message length: %zu", event_kind, strlen(message)); // Log reception of Kind 23456 events if (event_kind == 23456) { DEBUG_LOG("Admin event (kind 23456) received"); } if (event_kind == 23456) { // Enhanced admin event security - check authorization first char auth_error[512] = {0}; int auth_result = is_authorized_admin_event(event, auth_error, sizeof(auth_error)); if (auth_result != 0) { // Authorization failed - log and reject DEBUG_WARN("Admin event authorization failed"); result = -1; size_t error_len = strlen(auth_error); size_t copy_len = (error_len < sizeof(error_message) - 1) ? error_len : sizeof(error_message) - 1; memcpy(error_message, auth_error, copy_len); error_message[copy_len] = '\0'; } else { // Authorization successful - process through admin API char admin_error[512] = {0}; int admin_result = process_admin_event_in_config(event, admin_error, sizeof(admin_error), wsi); // Log results for Kind 23456 events if (event_kind == 23456) { if (admin_result != 0) { char error_result_msg[512]; if (strlen(admin_error) > 0) { // Safely truncate admin_error if too long size_t max_error_len = sizeof(error_result_msg) - 50; // Leave room for prefix size_t error_len = strlen(admin_error); if (error_len > max_error_len) { error_len = max_error_len; } char truncated_error[512]; memcpy(truncated_error, admin_error, error_len); truncated_error[error_len] = '\0'; // Use a safer approach to avoid truncation warning size_t prefix_len = snprintf(error_result_msg, sizeof(error_result_msg), "ERROR: Kind %d event processing failed: ", event_kind); if (prefix_len < sizeof(error_result_msg)) { size_t remaining = sizeof(error_result_msg) - prefix_len; size_t copy_len = strlen(truncated_error); if (copy_len >= remaining) { copy_len = remaining - 1; } memcpy(error_result_msg + prefix_len, truncated_error, copy_len); error_result_msg[prefix_len + copy_len] = '\0'; } } else { snprintf(error_result_msg, sizeof(error_result_msg), "ERROR: Kind %d event processing failed", event_kind); } DEBUG_ERROR(error_result_msg); } } if (admin_result != 0) { DEBUG_ERROR("Failed to process admin event"); result = -1; size_t error_len = strlen(admin_error); size_t copy_len = (error_len < sizeof(error_message) - 1) ? error_len : sizeof(error_message) - 1; memcpy(error_message, admin_error, copy_len); error_message[copy_len] = '\0'; } else { // Admin events are processed by the admin API, not broadcast to subscriptions } } } else if (event_kind == 1059) { // NIP-17 gift wrap events // Admin DM processing is opt-in via nip17_admin_enabled config (default: off) // to prevent expensive decryption on every incoming gift wrap event int nip17_admin_enabled = hot_cfg_nip17_admin_enabled(); if (nip17_admin_enabled) { char nip17_error[512] = {0}; cJSON* response_event = process_nip17_admin_message(event, nip17_error, sizeof(nip17_error), wsi); if (!response_event) { // Check if this is an error or if the command was already handled if (strlen(nip17_error) > 0) { // There was an actual error DEBUG_ERROR("NIP-17 admin message processing failed"); result = -1; size_t error_len = strlen(nip17_error); size_t copy_len = (error_len < sizeof(error_message) - 1) ? error_len : sizeof(error_message) - 1; memcpy(error_message, nip17_error, copy_len); error_message[copy_len] = '\0'; } else { // No error message means the command was already handled (plain text commands) // Store the original gift wrap event in database if (store_event(event) != 0) { DEBUG_ERROR("Failed to store gift wrap event in database"); result = -1; strncpy(error_message, "error: failed to store gift wrap event", sizeof(error_message) - 1); } } } else { // Store the original gift wrap event in database (unlike kind 23456) if (store_event(event) != 0) { DEBUG_ERROR("Failed to store gift wrap event in database"); result = -1; strncpy(error_message, "error: failed to store gift wrap event", sizeof(error_message) - 1); cJSON_Delete(response_event); } else { // Broadcast RESPONSE event to matching persistent subscriptions broadcast_event_to_subscriptions(response_event); // Clean up response event cJSON_Delete(response_event); } } } else { // NIP-17 admin DMs disabled: store gift wrap as a regular event DEBUG_TRACE("NIP-17 admin DMs disabled - storing kind 1059 event without decryption"); if (store_event(event) != 0) { DEBUG_ERROR("Failed to store gift wrap event in database"); result = -1; strncpy(error_message, "error: failed to store event", sizeof(error_message) - 1); } else { broadcast_event_to_subscriptions(event); } } } else if (event_kind == 14) { // Check for DM stats commands addressed to relay char dm_error[512] = {0}; int dm_result = process_dm_stats_command(event, dm_error, sizeof(dm_error), wsi); if (dm_result != 0) { DEBUG_ERROR("DM stats command processing failed"); result = -1; size_t error_len = strlen(dm_error); size_t copy_len = (error_len < sizeof(error_message) - 1) ? error_len : sizeof(error_message) - 1; memcpy(error_message, dm_error, copy_len); error_message[copy_len] = '\0'; } else { // Store the DM event in database if (store_event(event) != 0) { DEBUG_ERROR("Failed to store DM event in database"); result = -1; strncpy(error_message, "error: failed to store DM event", sizeof(error_message) - 1); } else { // Broadcast DM event to matching persistent subscriptions broadcast_event_to_subscriptions(event); } } } else { // Check if this is an ephemeral event (kinds 20000-29999) // Per NIP-01: ephemeral events are broadcast but never stored if (event_kind >= 20000 && event_kind < 30000) { DEBUG_TRACE("Ephemeral event (kind %d) - broadcasting without storage", event_kind); // Broadcast directly to subscriptions without database storage broadcast_event_to_subscriptions(event); } else { DEBUG_TRACE("Storing regular event in database"); // Regular event - store in database and broadcast if (store_event(event) != 0) { DEBUG_ERROR("Failed to store event in database"); result = -1; strncpy(error_message, "error: failed to store event", sizeof(error_message) - 1); } else { DEBUG_LOG("Event stored and broadcast (kind %d)", event_kind); // Broadcast event to matching persistent subscriptions broadcast_event_to_subscriptions(event); } } } } else { // Event without valid kind - try normal storage DEBUG_WARN("Event without valid kind - trying normal storage"); if (store_event(event) != 0) { DEBUG_ERROR("Failed to store event without kind in database"); result = -1; strncpy(error_message, "error: failed to store event", sizeof(error_message) - 1); } else { broadcast_event_to_subscriptions(event); } } } // Send OK response cJSON* event_id = cJSON_GetObjectItemCaseSensitive(event, "id"); if (event_id && cJSON_IsString(event_id)) { cJSON* response = cJSON_CreateArray(); cJSON_AddItemToArray(response, cJSON_CreateString("OK")); cJSON_AddItemToArray(response, cJSON_CreateString(cJSON_GetStringValue(event_id))); cJSON_AddItemToArray(response, cJSON_CreateBool(result == 0)); cJSON_AddItemToArray(response, cJSON_CreateString(strlen(error_message) > 0 ? error_message : "")); char *response_str = cJSON_Print(response); if (response_str) { size_t response_len = strlen(response_str); // DEBUG: Log WebSocket frame details before sending DEBUG_TRACE("WS_FRAME_SEND: type=OK len=%zu data=%.100s%s", response_len, response_str, response_len > 100 ? "..." : ""); // Queue message for proper libwebsockets pattern if (queue_message(wsi, pss, response_str, response_len, LWS_WRITE_TEXT) != 0) { DEBUG_ERROR("Failed to queue OK response message"); } free(response_str); } cJSON_Delete(response); } } } else if (strcmp(msg_type, "REQ") == 0) { DEBUG_TRACE("REQ message received, starting processing"); // Log the full REQ message for debugging // Check NIP-42 authentication for REQ subscriptions if required if (pss && pss->nip42_auth_required_subscriptions && !pss->authenticated) { DEBUG_TRACE("REQ rejected: NIP-42 authentication required"); if (!pss->auth_challenge_sent) { send_nip42_auth_challenge(wsi, pss); } else { send_notice_message(wsi, pss, "NIP-42 authentication required for subscriptions"); DEBUG_WARN("REQ rejected: NIP-42 authentication required"); // Auth timeout: close connection if challenge was sent but client // hasn't authenticated within nip42_auth_timeout_sec seconds int auth_timeout = hot_cfg_nip42_auth_timeout_sec(); if (auth_timeout > 0 && pss->connection_established > 0) { time_t connection_age = time(NULL) - pss->connection_established; if (connection_age >= auth_timeout) { DEBUG_LOG("Closing unauthenticated connection from %s after %ld seconds (timeout=%d)", pss->client_ip, connection_age, auth_timeout); lws_close_reason(wsi, LWS_CLOSE_STATUS_POLICY_VIOLATION, (unsigned char*)"Authentication timeout", 22); cJSON_Delete(json); free(message); return -1; } } } cJSON_Delete(json); free(message); return 0; } DEBUG_TRACE("REQ message passed authentication check"); // Handle REQ message cJSON* sub_id = cJSON_GetArrayItem(json, 1); if (sub_id && cJSON_IsString(sub_id)) { const char* subscription_id = cJSON_GetStringValue(sub_id); // Mark session as active - client sent REQ pthread_mutex_lock(&pss->session_lock); pss->session_active = 1; pthread_mutex_unlock(&pss->session_lock); // Cancel idle timeout - this is legitimate usage lws_set_timeout(wsi, NO_PENDING_TIMEOUT, 0); DEBUG_TRACE("Processing REQ message for subscription %s", subscription_id); // Validate subscription ID before processing if (!subscription_id) { DEBUG_TRACE("REQ rejected: NULL subscription ID"); send_notice_message(wsi, pss, "error: invalid subscription ID"); DEBUG_WARN("REQ rejected: NULL subscription ID"); record_malformed_request(pss); cJSON_Delete(json); free(message); return 0; } // Validate subscription ID if (!validate_subscription_id(subscription_id)) { DEBUG_TRACE("REQ rejected: invalid subscription ID format"); send_notice_message(wsi, pss, "error: invalid subscription ID"); DEBUG_WARN("REQ rejected: invalid subscription ID"); cJSON_Delete(json); free(message); return 0; } DEBUG_TRACE("REQ subscription ID validated: %s", subscription_id); // Create array of filter objects from position 2 onwards cJSON* filters = cJSON_CreateArray(); if (!filters) { DEBUG_TRACE("REQ failed: could not create filters array"); send_notice_message(wsi, pss, "error: failed to process filters"); DEBUG_ERROR("REQ failed: could not create filters array"); cJSON_Delete(json); free(message); return 0; } int json_size = cJSON_GetArraySize(json); int filter_count = 0; for (int i = 2; i < json_size; i++) { cJSON* filter = cJSON_GetArrayItem(json, i); if (filter) { cJSON_AddItemToArray(filters, cJSON_Duplicate(filter, 1)); filter_count++; } } DEBUG_TRACE("REQ created %d filters from message", filter_count); // Validate filters before processing char filter_error[512] = {0}; int validation_result = validate_filter_array(filters, filter_error, sizeof(filter_error)); if (validation_result <= 0) { DEBUG_TRACE("REQ rejected: filter validation failed - %s", filter_error); send_notice_message(wsi, pss, filter_error); DEBUG_WARN("REQ rejected: invalid filters"); // Only record as malformed if it's a true error (0), not benign error (-1) if (validation_result == 0) { record_malformed_request(pss); } cJSON_Delete(filters); cJSON_Delete(json); free(message); return 0; } DEBUG_TRACE("REQ filters validated successfully"); DEBUG_TRACE("About to call handle_req_message for subscription %s", subscription_id); int req_result = handle_req_message(subscription_id, filters, wsi, pss); DEBUG_TRACE("handle_req_message completed for subscription %s", subscription_id); // Clean up the filters array we created cJSON_Delete(filters); // Async path will send EOSE from completion handler. if (req_result != HANDLE_REQ_ASYNC_PENDING) { DEBUG_LOG("REQ subscription %s processed, sending EOSE", subscription_id); // Send EOSE (End of Stored Events) cJSON* eose_response = cJSON_CreateArray(); if (eose_response) { cJSON_AddItemToArray(eose_response, cJSON_CreateString("EOSE")); cJSON_AddItemToArray(eose_response, cJSON_CreateString(subscription_id)); char *eose_str = cJSON_Print(eose_response); if (eose_str) { size_t eose_len = strlen(eose_str); // DEBUG: Log WebSocket frame details before sending DEBUG_TRACE("WS_FRAME_SEND: type=EOSE len=%zu data=%.100s%s", eose_len, eose_str, eose_len > 100 ? "..." : ""); // Queue message for proper libwebsockets pattern if (queue_message(wsi, pss, eose_str, eose_len, LWS_WRITE_TEXT) != 0) { DEBUG_ERROR("Failed to queue EOSE message"); } free(eose_str); } cJSON_Delete(eose_response); } } } else { send_notice_message(wsi, pss, "error: missing or invalid subscription ID in REQ"); DEBUG_WARN("REQ rejected: missing or invalid subscription ID"); } } else if (strcmp(msg_type, "COUNT") == 0) { // Check NIP-42 authentication for COUNT requests if required if (pss && pss->nip42_auth_required_subscriptions && !pss->authenticated) { if (!pss->auth_challenge_sent) { send_nip42_auth_challenge(wsi, pss); } else { send_notice_message(wsi, pss, "NIP-42 authentication required for count requests"); DEBUG_WARN("COUNT rejected: NIP-42 authentication required"); } cJSON_Delete(json); free(message); return 0; } // Handle COUNT message cJSON* sub_id = cJSON_GetArrayItem(json, 1); if (sub_id && cJSON_IsString(sub_id)) { const char* subscription_id = cJSON_GetStringValue(sub_id); // Create array of filter objects from position 2 onwards cJSON* filters = cJSON_CreateArray(); int json_size = cJSON_GetArraySize(json); for (int i = 2; i < json_size; i++) { cJSON* filter = cJSON_GetArrayItem(json, i); if (filter) { cJSON_AddItemToArray(filters, cJSON_Duplicate(filter, 1)); } } // Validate filters before processing char filter_error[512] = {0}; int validation_result = validate_filter_array(filters, filter_error, sizeof(filter_error)); if (validation_result <= 0) { send_notice_message(wsi, pss, filter_error); DEBUG_WARN("COUNT rejected: invalid filters"); // Only record as malformed if it's a true error (0), not benign error (-1) if (validation_result == 0) { record_malformed_request(pss); } cJSON_Delete(filters); cJSON_Delete(json); free(message); return 0; } int count_result = handle_count_message(subscription_id, filters, wsi, pss); // Clean up the filters array we created cJSON_Delete(filters); if (count_result == HANDLE_REQ_ASYNC_PENDING) { cJSON_Delete(json); free(message); return 0; } } } else if (strcmp(msg_type, "CLOSE") == 0) { // Handle CLOSE message cJSON* sub_id = cJSON_GetArrayItem(json, 1); if (sub_id && cJSON_IsString(sub_id)) { const char* subscription_id = cJSON_GetStringValue(sub_id); // Validate subscription ID before processing if (!subscription_id) { send_notice_message(wsi, pss, "error: invalid subscription ID in CLOSE"); DEBUG_WARN("CLOSE rejected: NULL subscription ID"); cJSON_Delete(json); free(message); return 0; } // Validate subscription ID if (!validate_subscription_id(subscription_id)) { send_notice_message(wsi, pss, "error: invalid subscription ID in CLOSE"); DEBUG_WARN("CLOSE rejected: invalid subscription ID"); cJSON_Delete(json); free(message); return 0; } // CRITICAL FIX: Mark subscription as inactive in global manager FIRST // This prevents other threads from accessing it during removal pthread_mutex_lock(&g_subscription_manager.subscriptions_lock); subscription_t* target_sub = g_subscription_manager.active_subscriptions; while (target_sub) { if (strcmp(target_sub->id, subscription_id) == 0 && target_sub->wsi == wsi) { target_sub->active = 0; // Mark as inactive immediately break; } target_sub = target_sub->next; } pthread_mutex_unlock(&g_subscription_manager.subscriptions_lock); // Now safe to remove from session list if (pss) { pthread_mutex_lock(&pss->session_lock); struct subscription** current = &pss->subscriptions; while (*current) { if (strcmp((*current)->id, subscription_id) == 0) { struct subscription* to_remove = *current; *current = to_remove->session_next; pss->subscription_count--; break; } current = &((*current)->session_next); } pthread_mutex_unlock(&pss->session_lock); } // Finally remove from global manager (which will free it) remove_subscription_from_manager(subscription_id, wsi); // Subscription closed } else { send_notice_message(wsi, pss, "error: missing or invalid subscription ID in CLOSE"); DEBUG_WARN("CLOSE rejected: missing or invalid subscription ID"); } } else if (strcmp(msg_type, "AUTH") == 0) { // Handle NIP-42 AUTH message if (cJSON_GetArraySize(json) >= 2) { cJSON* auth_payload = cJSON_GetArrayItem(json, 1); if (cJSON_IsString(auth_payload)) { // AUTH challenge response: ["AUTH", ] (unusual) handle_nip42_auth_challenge_response(wsi, pss, cJSON_GetStringValue(auth_payload)); } else if (cJSON_IsObject(auth_payload)) { // AUTH signed event: ["AUTH", ] (standard NIP-42) handle_nip42_auth_signed_event(wsi, pss, auth_payload); } else { send_notice_message(wsi, pss, "Invalid AUTH message format"); DEBUG_WARN("Received AUTH message with invalid payload type"); } } else { send_notice_message(wsi, pss, "AUTH message requires payload"); DEBUG_WARN("Received AUTH message without payload"); } } else { // Unknown message type char unknown_msg[128]; snprintf(unknown_msg, sizeof(unknown_msg), "Unknown message type: %.32s", msg_type); DEBUG_WARN(unknown_msg); send_notice_message(wsi, pss, "Unknown message type"); } } } if (json) cJSON_Delete(json); free(message); } } break; case LWS_CALLBACK_SERVER_WRITEABLE: // Handle message queue when socket becomes writeable if (pss) { process_message_queue(wsi, pss); } break; case LWS_CALLBACK_CLOSED: DEBUG_TRACE("WebSocket connection closed"); // Remove from global connection list (must happen before pss cleanup) connection_list_remove(wsi); // Enhanced closure logging with detailed diagnostics if (pss) { // Calculate connection duration time_t now = time(NULL); long duration = (pss->connection_established > 0) ? (long)(now - pss->connection_established) : 0; // Determine closure reason const char* reason = "client_disconnect"; if (g_shutdown_flag || !g_server_running) { reason = "server_shutdown"; } // Record failure if connection closed without ever becoming active // This catches: // 1. Idle connections that timed out (never sent REQ/EVENT/AUTH) // 2. Early disconnects (client closed before sending REQ/EVENT/AUTH) // Only record for WebSocket connections, not HTTP requests (NIP-11, embedded files) if (!pss->session_active && pss->is_websocket && strlen(pss->client_ip) > 0) { // Use separate idle failure recording (has its own threshold/duration) ip_ban_record_idle_failure(pss->client_ip); DEBUG_LOG("Recording idle/early-disconnect failure for IP %s (connected %ld seconds)", pss->client_ip, time(NULL) - pss->connection_established); } // Legacy: record auth failure if auth was required but not completed else if (!pss->authenticated && (pss->nip42_auth_required_events || pss->nip42_auth_required_subscriptions) && pss->auth_challenge_sent && strlen(pss->client_ip) > 0) { ip_ban_record_failure(pss->client_ip); } // Format authentication status char auth_status[80]; if (pss->authenticated && strlen(pss->authenticated_pubkey) > 0) { // Show first 8 chars of pubkey for identification snprintf(auth_status, sizeof(auth_status), "yes(%.8s...)", pss->authenticated_pubkey); } else { snprintf(auth_status, sizeof(auth_status), "no"); } // Log comprehensive closure information DEBUG_LOG("WebSocket CLOSED: ip=%s duration=%lds subscriptions=%d authenticated=%s reason=%s", pss->client_ip, duration, pss->subscription_count, auth_status, reason); // Clean up message queue to prevent memory leaks while (pss->message_queue_head) { struct message_queue_node* node = pss->message_queue_head; pss->message_queue_head = node->next; free(node->data); free(node); } pss->message_queue_tail = NULL; pss->message_queue_count = 0; pss->writeable_requested = 0; // Clean up message reassembly buffer if (pss->reassembly_buffer) { free(pss->reassembly_buffer); pss->reassembly_buffer = NULL; } pss->reassembly_size = 0; pss->reassembly_capacity = 0; pss->reassembly_active = 0; // Clean up session subscriptions - copy IDs first to avoid use-after-free pthread_mutex_lock(&pss->session_lock); // First pass: collect subscription IDs safely typedef struct temp_sub_id { char id[SUBSCRIPTION_ID_MAX_LENGTH]; struct temp_sub_id* next; } temp_sub_id_t; temp_sub_id_t* temp_ids = NULL; temp_sub_id_t* temp_tail = NULL; int temp_count = 0; struct subscription* sub = pss->subscriptions; while (sub) { if (sub->active) { // Only process active subscriptions temp_sub_id_t* temp = malloc(sizeof(temp_sub_id_t)); if (temp) { memcpy(temp->id, sub->id, SUBSCRIPTION_ID_MAX_LENGTH); temp->id[SUBSCRIPTION_ID_MAX_LENGTH - 1] = '\0'; temp->next = NULL; if (!temp_ids) { temp_ids = temp; temp_tail = temp; } else { temp_tail->next = temp; temp_tail = temp; } temp_count++; } } sub = sub->session_next; } // Clear session list immediately pss->subscriptions = NULL; pss->subscription_count = 0; pthread_mutex_unlock(&pss->session_lock); // Second pass: remove from global manager using copied IDs temp_sub_id_t* current_temp = temp_ids; while (current_temp) { temp_sub_id_t* next_temp = current_temp->next; remove_subscription_from_manager(current_temp->id, wsi); free(current_temp); current_temp = next_temp; } pthread_mutex_destroy(&pss->session_lock); } else { DEBUG_LOG("WebSocket CLOSED: ip=unknown duration=0s subscriptions=0 authenticated=no reason=unknown"); } DEBUG_TRACE("WebSocket connection cleanup complete"); break; default: break; } return 0; } // Check and disconnect connections that have exceeded max age // This function works by checking connection age through the subscription system static void check_connection_age(int max_connection_seconds) { if (max_connection_seconds <= 0) { return; // Feature disabled } time_t current_time = time(NULL); // Lock the subscription manager to safely iterate through subscriptions pthread_mutex_lock(&g_subscription_manager.subscriptions_lock); // Track unique WSI pointers we've already checked to avoid duplicate checks struct lws** checked_wsis = NULL; int checked_count = 0; int checked_capacity = 0; subscription_t* sub = g_subscription_manager.active_subscriptions; while (sub) { if (!sub->active || !sub->wsi) { sub = sub->next; continue; } // Check if we've already processed this WSI int already_checked = 0; for (int i = 0; i < checked_count; i++) { if (checked_wsis[i] == sub->wsi) { already_checked = 1; break; } } if (!already_checked) { // Get per-session data to check connection age struct per_session_data *pss = (struct per_session_data *)lws_wsi_user(sub->wsi); if (pss && pss->connection_established > 0) { time_t connection_age = current_time - pss->connection_established; if (connection_age >= max_connection_seconds) { DEBUG_LOG("Disconnecting client %s: connection age %ld seconds exceeds limit %d seconds", pss->client_ip, connection_age, max_connection_seconds); // Close connection with normal closure status lws_close_reason(sub->wsi, LWS_CLOSE_STATUS_NORMAL, (unsigned char*)"Connection age limit reached", 28); } } // Add to checked list if (checked_count >= checked_capacity) { checked_capacity = checked_capacity == 0 ? 64 : checked_capacity * 2; checked_wsis = realloc(checked_wsis, checked_capacity * sizeof(struct lws*)); } checked_wsis[checked_count++] = sub->wsi; } sub = sub->next; } pthread_mutex_unlock(&g_subscription_manager.subscriptions_lock); // Cleanup free(checked_wsis); // Periodic IP ban maintenance: cleanup expired entries, log stats, save to DB ip_ban_cleanup(); ip_ban_log_stats(); } // WebSocket protocol definition static struct lws_protocols protocols[] = { { "nostr-relay-protocol", nostr_relay_callback, sizeof(struct per_session_data), 65536, // rx buffer size 0, NULL, 0 }, { NULL, NULL, 0, 0, 0, NULL, 0 } // terminator }; // Check if a port is available for binding int check_port_available(int port) { int sockfd; struct sockaddr_in addr; int result; int reuse = 1; // Create a socket sockfd = socket(AF_INET, SOCK_STREAM, 0); if (sockfd < 0) { return 0; // Cannot create socket, assume port unavailable } // Set SO_REUSEADDR to allow binding to ports in TIME_WAIT state // This matches libwebsockets behavior and prevents false unavailability if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) < 0) { close(sockfd); return 0; // Failed to set socket option } // Set up the address structure memset(&addr, 0, sizeof(addr)); addr.sin_family = AF_INET; addr.sin_addr.s_addr = INADDR_ANY; addr.sin_port = htons(port); // Try to bind to the port result = bind(sockfd, (struct sockaddr*)&addr, sizeof(addr)); // Close the socket close(sockfd); // Return 1 if bind succeeded (port available), 0 if failed (port in use) return (result == 0) ? 1 : 0; } // Start libwebsockets-based WebSocket Nostr relay server int start_websocket_relay(int port_override, int strict_port) { struct lws_context_creation_info info; // Starting libwebsockets-based Nostr relay server // Set libwebsockets log level to errors only lws_set_log_level(LLL_USER | LLL_ERR, NULL); memset(&info, 0, sizeof(info)); // Use port override if provided, otherwise use configuration int configured_port = (port_override > 0) ? port_override : get_config_int("relay_port", DEFAULT_PORT); int actual_port = configured_port; int port_attempts = 0; const int max_port_attempts = 10; // Increased from 5 to 10 // Minimal libwebsockets configuration info.protocols = protocols; info.gid = -1; info.uid = -1; info.options = LWS_SERVER_OPTION_VALIDATE_UTF8; // Remove interface restrictions - let system choose // info.vhost_name = NULL; // info.iface = NULL; // Increase max connections for relay usage info.max_http_header_pool = 16; info.timeout_secs = 10; // Max payload size for Nostr events info.max_http_header_data = 4096; // Find an available port with pre-checking (or fail immediately in strict mode) while (port_attempts < (strict_port ? 1 : max_port_attempts)) { // Checking port availability // Pre-check if port is available if (!check_port_available(actual_port)) { port_attempts++; if (strict_port) { char error_msg[256]; snprintf(error_msg, sizeof(error_msg), "Strict port mode: port %d is not available", actual_port); DEBUG_ERROR(error_msg); return -1; } else if (port_attempts < max_port_attempts) { char retry_msg[256]; snprintf(retry_msg, sizeof(retry_msg), "Port %d is in use, trying port %d (attempt %d/%d)", actual_port, actual_port + 1, port_attempts + 1, max_port_attempts); DEBUG_WARN(retry_msg); actual_port++; continue; } else { char error_msg[512]; snprintf(error_msg, sizeof(error_msg), "Failed to find available port after %d attempts (tried ports %d-%d)", max_port_attempts, configured_port, actual_port); DEBUG_ERROR(error_msg); return -1; } } // Port appears available, try creating libwebsockets context info.port = actual_port; // Attempting to bind libwebsockets ws_context = lws_create_context(&info); if (ws_context) { // Success! Port binding worked break; } // libwebsockets failed even though port check passed // This could be due to timing or different socket options int errno_saved = errno; char lws_error_msg[256]; snprintf(lws_error_msg, sizeof(lws_error_msg), "libwebsockets failed to bind to port %d (errno: %d)", actual_port, errno_saved); DEBUG_WARN(lws_error_msg); port_attempts++; if (strict_port) { char error_msg[256]; snprintf(error_msg, sizeof(error_msg), "Strict port mode: failed to bind to port %d", actual_port); DEBUG_ERROR(error_msg); break; } else if (port_attempts < max_port_attempts) { actual_port++; continue; } // If we get here, we've exhausted attempts break; } if (!ws_context) { char error_msg[512]; snprintf(error_msg, sizeof(error_msg), "Failed to create libwebsockets context after %d attempts. Last attempted port: %d", port_attempts, actual_port); DEBUG_ERROR(error_msg); perror("libwebsockets creation error"); return -1; } // Persist the actual bound port for components that need the runtime relay URL g_relay_port = actual_port; char startup_msg[256]; if (actual_port != configured_port) { snprintf(startup_msg, sizeof(startup_msg), "WebSocket relay started on ws://127.0.0.1:%d (configured port %d was unavailable)", actual_port, configured_port); DEBUG_WARN(startup_msg); } else { snprintf(startup_msg, sizeof(startup_msg), "WebSocket relay started on ws://127.0.0.1:%d", actual_port); } // Static variable for status post timing (initialize to 0 for immediate first post) static time_t last_status_post_time = 0; // Static variable for connection age check timing static time_t last_connection_age_check = 0; if (start_async_event_worker() != 0) { DEBUG_WARN("Async event worker failed to start; EVENT path will remain synchronous"); } if (start_api_worker() != 0) { DEBUG_WARN("API worker failed to start; monitoring/status work will run on lws-main"); } pthread_setname_np(pthread_self(), "lws-main"); // Main event loop with proper signal handling while (g_server_running && !g_shutdown_flag) { int result = lws_service(ws_context, 1000); if (result < 0) { DEBUG_ERROR("libwebsockets service error"); break; } // Drain completed async REQ jobs and emit EVENT/EOSE on service thread. process_req_async_completions(); // Drain completed async EVENT jobs and emit OK/broadcast on service thread. process_async_event_completions(); // Drain completed caching-inbox ingestion jobs and run post-actions/broadcast // on the service thread (PostgreSQL-only; no-op otherwise). process_inbox_event_completions(); // Poll the caching_event_inbox table and feed dequeued events through // ingest_event() with EVENT_SOURCE_CACHING_INBOX (PostgreSQL-only; no-op // otherwise). Runs once per lws service loop iteration on the main thread. caching_inbox_poller_tick(); // Drain completed async COUNT jobs and emit COUNT/NOTICE on service thread. process_count_async_completions(); // Drain completed api-worker tasks. api_worker_process_completions(); // Check if it's time to post status update time_t current_time = time(NULL); int status_post_hours = hot_cfg_kind_1_status_posts_hours(); if (status_post_hours > 0) { int seconds_interval = status_post_hours * 3600; // Convert hours to seconds if (current_time - last_status_post_time >= seconds_interval) { last_status_post_time = current_time; if (api_worker_enqueue_status_post() != 0) { generate_and_post_status_event(); } } } // Check connection age limits and run IP ban maintenance (every 60 seconds) int max_connection_seconds = hot_cfg_max_connection_seconds(); if (current_time - last_connection_age_check >= 60) { last_connection_age_check = current_time; // Live debug level update: read from config table so it can be changed // without restarting the relay (via config_set admin command or direct SQL) int config_debug_level = hot_cfg_debug_level(); if (config_debug_level >= 0 && config_debug_level != g_debug_level) { DEBUG_WARN("Debug level changed: %d -> %d", g_debug_level, config_debug_level); g_debug_level = config_debug_level; } // Check and close idle connections (no REQ/EVENT sent within timeout) int idle_timeout_sec = hot_cfg_idle_connection_timeout_sec(); check_idle_connections(idle_timeout_sec); if (thread_pool_submit_wal_checkpoint() != 0) { DEBUG_TRACE("WAL checkpoint submit skipped or failed"); } if (max_connection_seconds > 0) { check_connection_age(max_connection_seconds); } else { // Even when connection age limit is disabled, run IP ban maintenance ip_ban_cleanup(); ip_ban_log_stats(); } } } stop_api_worker(); stop_async_event_worker(); lws_context_destroy(ws_context); ws_context = NULL; return 0; } // Process DM stats command int process_dm_stats_command(cJSON* dm_event, char* error_message, size_t error_size, struct lws* wsi) { // Suppress unused parameter warning (void)wsi; if (!dm_event || !error_message) { return -1; } // Check if DM is addressed to relay cJSON* tags = cJSON_GetObjectItemCaseSensitive(dm_event, "tags"); if (!tags || !cJSON_IsArray(tags)) { strncpy(error_message, "DM missing or invalid tags", error_size - 1); return -1; } const char* relay_pubkey = get_config_value("relay_pubkey"); if (!relay_pubkey) { strncpy(error_message, "Could not get relay pubkey", error_size - 1); return -1; } // Look for "p" tag with relay pubkey int addressed_to_relay = 0; cJSON* tag = NULL; cJSON_ArrayForEach(tag, tags) { if (cJSON_IsArray(tag) && cJSON_GetArraySize(tag) >= 2) { cJSON* tag_name = cJSON_GetArrayItem(tag, 0); cJSON* tag_value = cJSON_GetArrayItem(tag, 1); if (tag_name && cJSON_IsString(tag_name) && strcmp(cJSON_GetStringValue(tag_name), "p") == 0 && tag_value && cJSON_IsString(tag_value) && strcmp(cJSON_GetStringValue(tag_value), relay_pubkey) == 0) { addressed_to_relay = 1; break; } } } free((char*)relay_pubkey); if (!addressed_to_relay) { // Not addressed to relay, allow normal processing return 0; } // Get sender pubkey cJSON* pubkey_obj = cJSON_GetObjectItemCaseSensitive(dm_event, "pubkey"); if (!pubkey_obj || !cJSON_IsString(pubkey_obj)) { strncpy(error_message, "DM missing sender pubkey", error_size - 1); return -1; } const char* sender_pubkey = cJSON_GetStringValue(pubkey_obj); // Check if sender is admin if (!is_admin_pubkey(sender_pubkey)) { strncpy(error_message, "Unauthorized: not admin", error_size - 1); return -1; } // Get relay private key for decryption char* relay_privkey_hex = get_relay_private_key(); if (!relay_privkey_hex) { strncpy(error_message, "Could not get relay private key", error_size - 1); return -1; } // Convert relay private key to bytes unsigned char relay_privkey[32]; if (nostr_hex_to_bytes(relay_privkey_hex, relay_privkey, sizeof(relay_privkey)) != 0) { free(relay_privkey_hex); strncpy(error_message, "Failed to convert relay private key", error_size - 1); return -1; } free(relay_privkey_hex); // Convert sender pubkey to bytes unsigned char sender_pubkey_bytes[32]; if (nostr_hex_to_bytes(sender_pubkey, sender_pubkey_bytes, sizeof(sender_pubkey_bytes)) != 0) { strncpy(error_message, "Failed to convert sender pubkey", error_size - 1); return -1; } // Get encrypted content cJSON* content_obj = cJSON_GetObjectItemCaseSensitive(dm_event, "content"); if (!content_obj || !cJSON_IsString(content_obj)) { strncpy(error_message, "DM missing content", error_size - 1); return -1; } const char* encrypted_content = cJSON_GetStringValue(content_obj); // Decrypt content char decrypted_content[16384]; int decrypt_result = nostr_nip44_decrypt(relay_privkey, sender_pubkey_bytes, encrypted_content, decrypted_content, sizeof(decrypted_content)); if (decrypt_result != NOSTR_SUCCESS) { char decrypt_error[256]; snprintf(decrypt_error, sizeof(decrypt_error), "NIP-44 decryption failed: %d", decrypt_result); strncpy(error_message, decrypt_error, error_size - 1); return -1; } // Check if content is "stats" if (strcmp(decrypted_content, "stats") != 0) { // Not a stats command, allow normal processing return 0; } // Processing DM stats command from admin // Generate stats JSON char* stats_json = generate_stats_json(); if (!stats_json) { strncpy(error_message, "Failed to generate stats", error_size - 1); return -1; } // Encrypt stats for response char encrypted_response[4096]; int encrypt_result = nostr_nip44_encrypt(relay_privkey, sender_pubkey_bytes, stats_json, encrypted_response, sizeof(encrypted_response)); free(stats_json); if (encrypt_result != NOSTR_SUCCESS) { char encrypt_error[256]; snprintf(encrypt_error, sizeof(encrypt_error), "NIP-44 encryption failed: %d", encrypt_result); strncpy(error_message, encrypt_error, error_size - 1); return -1; } // Create DM response event cJSON* dm_response = cJSON_CreateObject(); cJSON_AddStringToObject(dm_response, "id", ""); // Will be set by event creation cJSON_AddStringToObject(dm_response, "pubkey", relay_pubkey); cJSON_AddNumberToObject(dm_response, "created_at", (double)time(NULL)); cJSON_AddNumberToObject(dm_response, "kind", 14); cJSON_AddStringToObject(dm_response, "content", encrypted_response); // Add tags: p tag for recipient (admin) cJSON* response_tags = cJSON_CreateArray(); cJSON* p_tag = cJSON_CreateArray(); cJSON_AddItemToArray(p_tag, cJSON_CreateString("p")); cJSON_AddItemToArray(p_tag, cJSON_CreateString(sender_pubkey)); cJSON_AddItemToArray(response_tags, p_tag); cJSON_AddItemToObject(dm_response, "tags", response_tags); // Add signature placeholder cJSON_AddStringToObject(dm_response, "sig", ""); // Will be set by event creation/signing // Store and broadcast the DM response int store_result = store_event(dm_response); if (store_result != 0) { cJSON_Delete(dm_response); strncpy(error_message, "Failed to store DM response", error_size - 1); return -1; } // Broadcast to subscriptions broadcast_event_to_subscriptions(dm_response); cJSON_Delete(dm_response); return 0; } // Handle NIP-45 COUNT message int handle_count_message(const char* sub_id, cJSON* filters, struct lws *wsi, struct per_session_data *pss) { // pss is now used for query tracking, so remove unused warning suppression if (!cJSON_IsArray(filters)) { DEBUG_ERROR("COUNT filters is not an array"); return 0; } // Parameter binding helpers char** bind_params = NULL; int bind_param_count = 0; int bind_param_capacity = 0; count_async_state_t* async_state = calloc(1, sizeof(*async_state)); if (!async_state) { DEBUG_ERROR("Failed to allocate COUNT async state"); return 0; } strncpy(async_state->sub_id, sub_id ? sub_id : "", sizeof(async_state->sub_id) - 1); async_state->sub_id[sizeof(async_state->sub_id) - 1] = '\0'; async_state->wsi = wsi; async_state->pss_token = pss; // Process each filter in the array for (int i = 0; i < cJSON_GetArraySize(filters); i++) { cJSON* filter = cJSON_GetArrayItem(filters, i); if (!filter || !cJSON_IsObject(filter)) { DEBUG_WARN("Invalid filter object in COUNT"); continue; } // Reset bind params for this filter for (int j = 0; j < bind_param_count; j++) { free(bind_params[j]); } free(bind_params); bind_params = NULL; bind_param_count = 0; bind_param_capacity = 0; // Build SQL COUNT query based on filter - exclude ephemeral events (kinds 20000-29999) from historical queries char sql[8192] = "SELECT COUNT(*) FROM events WHERE 1=1 AND (kind < 20000 OR kind >= 30000)"; char* sql_ptr = sql + strlen(sql); int remaining = sizeof(sql) - strlen(sql); // Note: Expiration filtering will be done at application level // after retrieving events to ensure compatibility with all SQLite versions // Handle kinds filter cJSON* kinds = cJSON_GetObjectItemCaseSensitive(filter, "kinds"); if (kinds && cJSON_IsArray(kinds)) { int kind_count = cJSON_GetArraySize(kinds); if (kind_count > 0) { snprintf(sql_ptr, remaining, " AND kind IN ("); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); for (int k = 0; k < kind_count; k++) { cJSON* kind = cJSON_GetArrayItem(kinds, k); if (cJSON_IsNumber(kind)) { if (k > 0) { snprintf(sql_ptr, remaining, ","); sql_ptr++; remaining--; } snprintf(sql_ptr, remaining, "%d", (int)cJSON_GetNumberValue(kind)); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } } snprintf(sql_ptr, remaining, ")"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } } // Handle authors filter cJSON* authors = cJSON_GetObjectItemCaseSensitive(filter, "authors"); if (authors && cJSON_IsArray(authors)) { int author_count = 0; // Count valid authors for (int a = 0; a < cJSON_GetArraySize(authors); a++) { cJSON* author = cJSON_GetArrayItem(authors, a); if (cJSON_IsString(author)) { author_count++; } } if (author_count > 0) { snprintf(sql_ptr, remaining, " AND pubkey IN ("); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); for (int a = 0; a < author_count; a++) { if (a > 0) { snprintf(sql_ptr, remaining, ","); sql_ptr++; remaining--; } snprintf(sql_ptr, remaining, "?"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } snprintf(sql_ptr, remaining, ")"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); // Add author values to bind params for (int a = 0; a < cJSON_GetArraySize(authors); a++) { cJSON* author = cJSON_GetArrayItem(authors, a); if (cJSON_IsString(author)) { if (bind_param_count >= bind_param_capacity) { bind_param_capacity = bind_param_capacity == 0 ? 16 : bind_param_capacity * 2; bind_params = realloc(bind_params, bind_param_capacity * sizeof(char*)); } bind_params[bind_param_count++] = strdup(cJSON_GetStringValue(author)); } } } } // Handle ids filter cJSON* ids = cJSON_GetObjectItemCaseSensitive(filter, "ids"); if (ids && cJSON_IsArray(ids)) { int id_count = 0; // Count valid ids for (int i = 0; i < cJSON_GetArraySize(ids); i++) { cJSON* id = cJSON_GetArrayItem(ids, i); if (cJSON_IsString(id)) { id_count++; } } if (id_count > 0) { snprintf(sql_ptr, remaining, " AND id IN ("); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); for (int i = 0; i < id_count; i++) { if (i > 0) { snprintf(sql_ptr, remaining, ","); sql_ptr++; remaining--; } snprintf(sql_ptr, remaining, "?"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } snprintf(sql_ptr, remaining, ")"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); // Add id values to bind params for (int i = 0; i < cJSON_GetArraySize(ids); i++) { cJSON* id = cJSON_GetArrayItem(ids, i); if (cJSON_IsString(id)) { if (bind_param_count >= bind_param_capacity) { bind_param_capacity = bind_param_capacity == 0 ? 16 : bind_param_capacity * 2; bind_params = realloc(bind_params, bind_param_capacity * sizeof(char*)); } bind_params[bind_param_count++] = strdup(cJSON_GetStringValue(id)); } } } } // Handle tag filters (#e, #p, #t, etc.) cJSON* filter_item = NULL; cJSON_ArrayForEach(filter_item, filter) { const char* filter_key = filter_item->string; if (filter_key && filter_key[0] == '#' && strlen(filter_key) > 1) { // This is a tag filter like "#e", "#p", etc. const char* tag_name = filter_key + 1; // e, p, t, type, ... if (cJSON_IsArray(filter_item)) { int tag_value_count = 0; // Count valid tag values for (int i = 0; i < cJSON_GetArraySize(filter_item); i++) { cJSON* tag_value = cJSON_GetArrayItem(filter_item, i); if (cJSON_IsString(tag_value)) { tag_value_count++; } } if (tag_value_count > 0) { #ifdef DB_BACKEND_POSTGRES // PostgreSQL path: JSONB containment against events.tags using GIN index. // Match Nostr semantics: tag name + first value only, allowing extra trailing elements. snprintf(sql_ptr, remaining, " AND ("); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); int clause_idx = 0; for (int i = 0; i < cJSON_GetArraySize(filter_item); i++) { cJSON* tag_value = cJSON_GetArrayItem(filter_item, i); if (!cJSON_IsString(tag_value)) { continue; } if (clause_idx > 0) { snprintf(sql_ptr, remaining, " OR "); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } snprintf(sql_ptr, remaining, "tags @> jsonb_build_array(jsonb_build_array(?::text, ?::text))"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); if (bind_param_count >= bind_param_capacity) { bind_param_capacity = bind_param_capacity == 0 ? 16 : bind_param_capacity * 2; bind_params = realloc(bind_params, bind_param_capacity * sizeof(char*)); } bind_params[bind_param_count++] = strdup(tag_name); if (bind_param_count >= bind_param_capacity) { bind_param_capacity = bind_param_capacity == 0 ? 16 : bind_param_capacity * 2; bind_params = realloc(bind_params, bind_param_capacity * sizeof(char*)); } bind_params[bind_param_count++] = strdup(cJSON_GetStringValue(tag_value)); clause_idx++; } snprintf(sql_ptr, remaining, ")"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); #else // SQLite path: keep json_each/json_extract behavior. snprintf(sql_ptr, remaining, " AND EXISTS (SELECT 1 FROM json_each(json(tags)) WHERE json_extract(value, '$[0]') = ? AND json_extract(value, '$[1]') IN ("); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); for (int i = 0; i < tag_value_count; i++) { if (i > 0) { snprintf(sql_ptr, remaining, ","); sql_ptr++; remaining--; } snprintf(sql_ptr, remaining, "?"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } snprintf(sql_ptr, remaining, "))"); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); // Add tag name and values to bind params if (bind_param_count >= bind_param_capacity) { bind_param_capacity = bind_param_capacity == 0 ? 16 : bind_param_capacity * 2; bind_params = realloc(bind_params, bind_param_capacity * sizeof(char*)); } bind_params[bind_param_count++] = strdup(tag_name); for (int i = 0; i < cJSON_GetArraySize(filter_item); i++) { cJSON* tag_value = cJSON_GetArrayItem(filter_item, i); if (cJSON_IsString(tag_value)) { if (bind_param_count >= bind_param_capacity) { bind_param_capacity = bind_param_capacity == 0 ? 16 : bind_param_capacity * 2; bind_params = realloc(bind_params, bind_param_capacity * sizeof(char*)); } bind_params[bind_param_count++] = strdup(cJSON_GetStringValue(tag_value)); } } #endif } } } } // Handle search filter (NIP-50) cJSON* search = cJSON_GetObjectItemCaseSensitive(filter, "search"); if (search && cJSON_IsString(search)) { const char* search_term = cJSON_GetStringValue(search); if (search_term && strlen(search_term) > 0) { // Search in both content and tag values using LIKE // Escape single quotes in search term for SQL safety char escaped_search[256]; size_t escaped_len = 0; for (size_t i = 0; search_term[i] && escaped_len < sizeof(escaped_search) - 1; i++) { if (search_term[i] == '\'') { escaped_search[escaped_len++] = '\''; escaped_search[escaped_len++] = '\''; } else { escaped_search[escaped_len++] = search_term[i]; } } escaped_search[escaped_len] = '\0'; // Add search conditions for content and tags #ifdef DB_BACKEND_POSTGRES snprintf(sql_ptr, remaining, " AND (content ILIKE '%%%s%%' OR tags::text ILIKE '%%\"%s\"%%')", escaped_search, escaped_search); #else // Use tags LIKE to search within the JSON string representation of tags snprintf(sql_ptr, remaining, " AND (content LIKE '%%%s%%' OR tags LIKE '%%\"%s\"%%')", escaped_search, escaped_search); #endif sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } } // Handle since filter cJSON* since = cJSON_GetObjectItemCaseSensitive(filter, "since"); if (since && cJSON_IsNumber(since)) { snprintf(sql_ptr, remaining, " AND created_at >= %ld", (long)cJSON_GetNumberValue(since)); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } // Handle until filter cJSON* until = cJSON_GetObjectItemCaseSensitive(filter, "until"); if (until && cJSON_IsNumber(until)) { snprintf(sql_ptr, remaining, " AND created_at <= %ld", (long)cJSON_GetNumberValue(until)); sql_ptr += strlen(sql_ptr); remaining = sizeof(sql) - strlen(sql); } if (remaining <= 1) { DEBUG_ERROR("COUNT SQL query exceeded internal buffer capacity; skipping filter"); continue; } // Submit COUNT query asynchronously to DB reader pool. if (submit_count_query_async(async_state, sql, (const char**)bind_params, bind_param_count) != 0) { DEBUG_ERROR("Failed to submit COUNT query async"); strncpy(async_state->error_message, "error: failed to submit count query", sizeof(async_state->error_message) - 1); async_state->error_message[sizeof(async_state->error_message) - 1] = '\0'; async_state->had_error = 1; break; } } // Cleanup bind params from the final filter iteration. for (int i = 0; i < bind_param_count; i++) { free(bind_params[i]); } free(bind_params); if (async_state->pending_jobs <= 0) { if (async_state->had_error) { send_notice_message(wsi, pss, async_state->error_message[0] ? async_state->error_message : "error: count query failed"); } else { send_count_response(wsi, pss, sub_id, 0); } free(async_state); return 0; } return HANDLE_REQ_ASYNC_PENDING; } ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// // RATE LIMITING FUNCTIONS ///////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////// /** * Check if a client is currently rate limited for malformed requests */ int is_client_rate_limited_for_malformed_requests(struct per_session_data *pss) { if (!pss) { return 0; } time_t now = time(NULL); // Check if currently blocked if (pss->malformed_request_blocked_until > now) { return 1; } // Reset block if expired if (pss->malformed_request_blocked_until > 0 && pss->malformed_request_blocked_until <= now) { pss->malformed_request_blocked_until = 0; pss->malformed_request_count = 0; pss->malformed_request_window_start = now; } // Check if within current hour window if (pss->malformed_request_window_start == 0 || (now - pss->malformed_request_window_start) >= 3600) { // 1 hour // Start new window pss->malformed_request_window_start = now; pss->malformed_request_count = 0; } // Check if exceeded limit if (pss->malformed_request_count >= MAX_MALFORMED_REQUESTS_PER_HOUR) { // Block for the specified duration pss->malformed_request_blocked_until = now + MALFORMED_REQUEST_BLOCK_DURATION; DEBUG_WARN("Client rate limited for malformed requests"); return 1; } return 0; } /** * Record a malformed request for rate limiting purposes */ void record_malformed_request(struct per_session_data *pss) { if (!pss) { return; } time_t now = time(NULL); // Initialize window if needed if (pss->malformed_request_window_start == 0) { pss->malformed_request_window_start = now; pss->malformed_request_count = 0; } // Reset window if hour has passed if ((now - pss->malformed_request_window_start) >= 3600) { pss->malformed_request_window_start = now; pss->malformed_request_count = 0; } // Increment count pss->malformed_request_count++; } /** * Validate if a string is valid hexadecimal of specified length */ int is_valid_hex_string(const char* str, size_t expected_len) { if (!str || strlen(str) != expected_len) { return 0; } for (size_t i = 0; i < expected_len; i++) { char c = str[i]; if (!((c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F'))) { return 0; } } return 1; } /** * Validate a filter array for REQ and COUNT messages * Returns: * 1 = valid * 0 = invalid (malformed, should count toward rate limit) * -1 = invalid but benign (e.g., kind 99999 from NDK ping, should not count toward rate limit) */ int validate_filter_array(cJSON* filters, char* error_message, size_t error_size) { if (!filters || !cJSON_IsArray(filters)) { snprintf(error_message, error_size, "error: filters must be an array"); return 0; } int filter_count = cJSON_GetArraySize(filters); if (filter_count > MAX_FILTERS_PER_REQUEST) { snprintf(error_message, error_size, "error: too many filters (max %d)", MAX_FILTERS_PER_REQUEST); return 0; } int has_kind_99999 = 0; // Track if we encounter kind 99999 (NDK ping) // Validate each filter object for (int i = 0; i < filter_count; i++) { cJSON* filter = cJSON_GetArrayItem(filters, i); if (!filter || !cJSON_IsObject(filter)) { snprintf(error_message, error_size, "error: filter %d is not an object", i); return 0; } // Validate filter fields cJSON* filter_item = NULL; cJSON_ArrayForEach(filter_item, filter) { const char* key = filter_item->string; if (!key) continue; // Validate authors array if (strcmp(key, "authors") == 0) { if (!cJSON_IsArray(filter_item)) { snprintf(error_message, error_size, "error: authors must be an array"); return 0; } int author_count = cJSON_GetArraySize(filter_item); if (author_count > MAX_AUTHORS_PER_FILTER) { snprintf(error_message, error_size, "error: too many authors (max %d)", MAX_AUTHORS_PER_FILTER); return 0; } for (int j = 0; j < author_count; j++) { cJSON* author = cJSON_GetArrayItem(filter_item, j); if (!cJSON_IsString(author)) { snprintf(error_message, error_size, "error: author %d is not a string", j); return 0; } const char* author_str = cJSON_GetStringValue(author); if (!is_valid_hex_string(author_str, 64)) { snprintf(error_message, error_size, "error: invalid author hex string"); return 0; } } } // Validate ids array else if (strcmp(key, "ids") == 0) { if (!cJSON_IsArray(filter_item)) { snprintf(error_message, error_size, "error: ids must be an array"); return 0; } int id_count = cJSON_GetArraySize(filter_item); if (id_count > MAX_IDS_PER_FILTER) { snprintf(error_message, error_size, "error: too many ids (max %d)", MAX_IDS_PER_FILTER); return 0; } for (int j = 0; j < id_count; j++) { cJSON* id = cJSON_GetArrayItem(filter_item, j); if (!cJSON_IsString(id)) { snprintf(error_message, error_size, "error: id %d is not a string", j); return 0; } const char* id_str = cJSON_GetStringValue(id); if (!is_valid_hex_string(id_str, 64)) { snprintf(error_message, error_size, "error: invalid id hex string"); return 0; } } } // Validate kinds array else if (strcmp(key, "kinds") == 0) { if (!cJSON_IsArray(filter_item)) { snprintf(error_message, error_size, "error: kinds must be an array"); return 0; } int kind_count = cJSON_GetArraySize(filter_item); if (kind_count > MAX_KINDS_PER_FILTER) { snprintf(error_message, error_size, "error: too many kinds (max %d)", MAX_KINDS_PER_FILTER); return 0; } for (int j = 0; j < kind_count; j++) { cJSON* kind = cJSON_GetArrayItem(filter_item, j); if (!cJSON_IsNumber(kind)) { snprintf(error_message, error_size, "error: kind %d is not a number", j); return 0; } int kind_val = (int)cJSON_GetNumberValue(kind); // Special case: kind 99999 is used by NDK (Nostr Development Kit) for ping/connectivity checks // We reject it but don't count it as a malformed request to avoid rate limiting NDK clients if (kind_val == 99999) { has_kind_99999 = 1; snprintf(error_message, error_size, "error: invalid kind value %d (NDK ping)", kind_val); continue; // Continue checking other kinds } if (kind_val < 0 || kind_val > MAX_KIND_VALUE) { snprintf(error_message, error_size, "error: invalid kind value %d", kind_val); return 0; } } } // Validate since/until timestamps else if (strcmp(key, "since") == 0 || strcmp(key, "until") == 0) { if (!cJSON_IsNumber(filter_item)) { snprintf(error_message, error_size, "error: %s must be a number", key); return 0; } double timestamp = cJSON_GetNumberValue(filter_item); if (timestamp < 0 || timestamp > MAX_TIMESTAMP_VALUE) { snprintf(error_message, error_size, "error: invalid %s timestamp", key); return 0; } } // Validate limit else if (strcmp(key, "limit") == 0) { if (!cJSON_IsNumber(filter_item)) { snprintf(error_message, error_size, "error: limit must be a number"); return 0; } int limit_val = (int)cJSON_GetNumberValue(filter_item); if (limit_val < 0 || limit_val > MAX_LIMIT_VALUE) { snprintf(error_message, error_size, "error: invalid limit value %d", limit_val); return 0; } } // Validate search term else if (strcmp(key, "search") == 0) { if (!cJSON_IsString(filter_item)) { snprintf(error_message, error_size, "error: search must be a string"); return 0; } const char* search_str = cJSON_GetStringValue(filter_item); size_t search_len = strlen(search_str); if (search_len > MAX_SEARCH_LENGTH) { snprintf(error_message, error_size, "error: search term too long (max %d)", MAX_SEARCH_LENGTH); return 0; } // Check for SQL injection characters if (strchr(search_str, ';') || strstr(search_str, "--") || strstr(search_str, "/*") || strstr(search_str, "*/")) { snprintf(error_message, error_size, "error: invalid characters in search term"); return 0; } } // Validate tag filters (#e, #p, #t, etc.) else if (key[0] == '#' && strlen(key) > 1) { if (!cJSON_IsArray(filter_item)) { snprintf(error_message, error_size, "error: %s must be an array", key); return 0; } int tag_count = cJSON_GetArraySize(filter_item); if (tag_count > MAX_TAG_VALUES_PER_FILTER) { snprintf(error_message, error_size, "error: too many %s values (max %d)", key, MAX_TAG_VALUES_PER_FILTER); return 0; } for (int j = 0; j < tag_count; j++) { cJSON* tag_value = cJSON_GetArrayItem(filter_item, j); if (!cJSON_IsString(tag_value)) { snprintf(error_message, error_size, "error: %s[%d] is not a string", key, j); return 0; } const char* tag_str = cJSON_GetStringValue(tag_value); size_t tag_len = strlen(tag_str); if (tag_len > MAX_TAG_VALUE_LENGTH) { snprintf(error_message, error_size, "error: %s value too long (max %d)", key, MAX_TAG_VALUE_LENGTH); return 0; } } } // Unknown filter keys are allowed but ignored } } // If we found kind 99999 (NDK ping), return -1 to indicate benign error if (has_kind_99999) { return -1; } return 1; // All filters valid }