diff --git a/apps/nautd/main.c b/apps/nautd/main.c index e25195d..b8c8e95 100644 --- a/apps/nautd/main.c +++ b/apps/nautd/main.c @@ -957,6 +957,11 @@ int main(int argc, char **argv) { close(listener); unlink(socket_path); + /* Tear down plugins first: a plugin like webui runs its own threads that + * call back into the daemon via RPC, so it must be stopped (and its + * threads joined) before we free the torrent tasks those calls touch. */ + naut_plugin_manager_destroy(state.plugins); + state.plugins = NULL; naut_script_destroy(state.script); state.script = NULL; stop_torrents(&state); @@ -966,7 +971,6 @@ int main(int argc, char **argv) { for (size_t i = 0; i < state.subscriber_count; i++) close(state.subscribers[i]); pthread_mutex_unlock(&state.subscriber_lock); - naut_plugin_manager_destroy(state.plugins); naut_rpc_registry_destroy(state.rpc); naut_event_bus_destroy(state.events); pthread_mutex_destroy(&state.subscriber_lock); diff --git a/plugins/webui/webui.c b/plugins/webui/webui.c index 6c3363f..d8d4f32 100644 --- a/plugins/webui/webui.c +++ b/plugins/webui/webui.c @@ -15,6 +15,7 @@ #include #include #include +#include #include #include #include @@ -27,6 +28,9 @@ #define SESSION_COOKIE "naut_session" #define SESSION_TTL_SECONDS (60 * 60 * 24 * 7) #define MAX_SESSIONS 64 +#define MAX_CONNECTIONS 128 +#define SPEED_SLOTS 256 +#define ETA_INFINITY 8640000 /* torrent-ui renders >= this as the infinity glyph */ typedef struct { char token[96]; @@ -34,6 +38,16 @@ typedef struct { bool used; } webui_session; +/* Single-writer (sampler thread) running estimate of a torrent's download + * rate, derived from successive byte counts. */ +typedef struct { + uint64_t id; + uint64_t last_bytes; + double last_time; + double dlspeed; + bool used; +} speed_slot; + typedef struct { naut_host_api host; char root[PATH_MAX]; @@ -46,10 +60,27 @@ typedef struct { atomic_bool stopping; bool thread_started; pthread_t thread; + + bool sampler_started; + pthread_t sampler; + pthread_mutex_t auth_lock; + pthread_mutex_t conn_lock; pthread_cond_t conn_cond; size_t active_connections; + + /* Latest snapshot, published once per second by the sampler thread and + * shared by /api/snapshot, /api/torrents and every SSE stream. */ + pthread_mutex_t snap_lock; + pthread_cond_t snap_cond; + char *snapshot_str; + char *torrents_str; + uint64_t snap_seq; + + pthread_mutex_t speed_lock; + speed_slot speeds[SPEED_SLOTS]; + webui_session sessions[MAX_SESSIONS]; } webui_state; @@ -59,6 +90,12 @@ typedef struct { static webui_state g_webui; +static double monotonic_seconds(void) { + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return (double)ts.tv_sec + (double)ts.tv_nsec / 1e9; +} + static void log_msg(int level, const char *message) { if (g_webui.host.log) g_webui.host.log(g_webui.host.host_context, level, message); @@ -124,6 +161,13 @@ static void http_json_extra(int fd, int code, json_t *json, const char *extra) { free(txt); } +/* Serve an already-serialized JSON string under one lock copy. */ +static void http_json_str(int fd, const char *json, const char *fallback) { + const char *body = json ? json : fallback; + http_head(fd, 200, "OK", "application/json; charset=utf-8", strlen(body)); + send_all_fd(fd, body, strlen(body)); +} + static const char *mime_type(const char *path) { const char *dot = strrchr(path, '.'); if (!dot) return "application/octet-stream"; @@ -143,20 +187,37 @@ static void strip_query(char *path) { if (hash) *hash = 0; } +/* Constant-time equality so credential checks don't leak length/content via + * timing. Returns true when both NUL-terminated strings match exactly. */ +static bool constant_time_equal(const char *a, const char *b) { + if (!a || !b) return false; + size_t la = strlen(a), lb = strlen(b); + size_t n = la > lb ? la : lb; + unsigned diff = (unsigned)(la ^ lb); + for (size_t i = 0; i < n; i++) { + unsigned char ca = i < la ? (unsigned char)a[i] : 0; + unsigned char cb = i < lb ? (unsigned char)b[i] : 0; + diff |= (unsigned)(ca ^ cb); + } + return diff == 0; +} + +/* Cryptographically strong hex. Fails closed: if the kernel CSPRNG is + * unavailable we refuse rather than fall back to predictable bytes (these + * feed session tokens). */ static bool random_hex(char *out, size_t out_size, size_t bytes) { static const char hex[] = "0123456789abcdef"; if (out_size < bytes * 2 + 1) return false; unsigned char buf[64]; if (bytes > sizeof buf) return false; - int fd = open("/dev/urandom", O_RDONLY); - ssize_t got = fd >= 0 ? read(fd, buf, bytes) : -1; - if (fd >= 0) close(fd); - if (got != (ssize_t)bytes) { - unsigned seed = (unsigned)time(NULL) ^ (unsigned)getpid(); - for (size_t i = 0; i < bytes; i++) { - seed = seed * 1103515245u + 12345u; - buf[i] = (unsigned char)(seed >> 16); + size_t got = 0; + while (got < bytes) { + ssize_t n = getrandom(buf + got, bytes - got, 0); + if (n < 0) { + if (errno == EINTR) continue; + return false; } + got += (size_t)n; } for (size_t i = 0; i < bytes; i++) { out[i * 2] = hex[buf[i] >> 4]; @@ -180,7 +241,11 @@ static void init_auth(void) { g_webui.generated_password = false; return; } - random_hex(g_webui.auth_password, sizeof g_webui.auth_password, 9); + if (!random_hex(g_webui.auth_password, sizeof g_webui.auth_password, 9)) { + /* No CSPRNG: leave the password empty so login is impossible rather + * than guessable. The operator must set NAUT_AUTH_PASSWORD. */ + g_webui.auth_password[0] = 0; + } g_webui.generated_password = true; } @@ -252,16 +317,17 @@ static bool current_user(const char *headers, const char *end) { static bool create_session(char *out, size_t out_size) { char token[96]; if (!random_hex(token, sizeof token, 24)) return false; - time_t expires = time(NULL) + SESSION_TTL_SECONDS; + time_t now = time(NULL); + time_t expires = now + SESSION_TTL_SECONDS; pthread_mutex_lock(&g_webui.auth_lock); webui_session *slot = NULL; for (size_t i = 0; i < MAX_SESSIONS; i++) { - if (!g_webui.sessions[i].used) { - slot = &g_webui.sessions[i]; - break; - } + webui_session *s = &g_webui.sessions[i]; + if (!s->used || s->expires < now) { slot = s; break; } + /* Otherwise track the session that expires soonest, so a full table + * evicts the oldest rather than always clobbering slot 0. */ + if (!slot || s->expires < slot->expires) slot = s; } - if (!slot) slot = &g_webui.sessions[0]; snprintf(slot->token, sizeof slot->token, "%s", token); slot->expires = expires; slot->used = true; @@ -403,6 +469,79 @@ static const char *ui_state(const char *state, double progress) { return "downloading"; } +/* ---- single-writer download-rate estimate keyed by torrent id ---- */ + +static double speed_sample(uint64_t id, uint64_t bytes) { + double now = monotonic_seconds(); + double result = 0.0; + pthread_mutex_lock(&g_webui.speed_lock); + speed_slot *slot = NULL, *spare = NULL; + for (size_t i = 0; i < SPEED_SLOTS; i++) { + speed_slot *s = &g_webui.speeds[i]; + if (s->used && s->id == id) { slot = s; break; } + if (!s->used && !spare) spare = s; + } + if (!slot) { + if (!spare) { + /* table full: evict least-recently-updated */ + spare = &g_webui.speeds[0]; + for (size_t i = 1; i < SPEED_SLOTS; i++) + if (g_webui.speeds[i].last_time < spare->last_time) + spare = &g_webui.speeds[i]; + } + slot = spare; + slot->used = true; + slot->id = id; + slot->last_bytes = bytes; + slot->last_time = now; + slot->dlspeed = 0.0; + pthread_mutex_unlock(&g_webui.speed_lock); + return 0.0; + } + double dt = now - slot->last_time; + if (dt > 0.0) { + double delta = bytes >= slot->last_bytes + ? (double)(bytes - slot->last_bytes) : 0.0; + double inst = delta / dt; + slot->dlspeed = slot->dlspeed * 0.6 + inst * 0.4; + if (slot->dlspeed < 0.0) slot->dlspeed = 0.0; + slot->last_bytes = bytes; + slot->last_time = now; + } + result = slot->dlspeed; + pthread_mutex_unlock(&g_webui.speed_lock); + return result; +} + +static double speed_peek(uint64_t id) { + double result = 0.0; + pthread_mutex_lock(&g_webui.speed_lock); + for (size_t i = 0; i < SPEED_SLOTS; i++) + if (g_webui.speeds[i].used && g_webui.speeds[i].id == id) { + result = g_webui.speeds[i].dlspeed; + break; + } + pthread_mutex_unlock(&g_webui.speed_lock); + return result; +} + +/* Drop slots for ids no longer present so a long-lived server doesn't hand a + * stale rate to a recycled id. */ +static void speed_retain(json_t *torrents) { + pthread_mutex_lock(&g_webui.speed_lock); + for (size_t i = 0; i < SPEED_SLOTS; i++) { + speed_slot *s = &g_webui.speeds[i]; + if (!s->used) continue; + bool found = false; + size_t index; + json_t *torrent; + json_array_foreach(torrents, index, torrent) + if (json_u64(torrent, "torrent_id") == s->id) { found = true; break; } + if (!found) s->used = false; + } + pthread_mutex_unlock(&g_webui.speed_lock); +} + static json_t *tracker_hosts(json_t *trackers) { json_t *hosts = json_array(); if (!hosts || !json_is_array(trackers)) return hosts; @@ -421,8 +560,8 @@ static json_t *tracker_hosts(json_t *trackers) { return hosts; } -static json_t *map_torrent(json_t *torrent, bool detail) { - uint64_t id = json_u64(torrent, "torrent_id"); +static double torrent_progress_ratio(json_t *torrent, uint64_t *done_out, + uint64_t *total_out) { uint64_t done = json_u64(torrent, "bytes_done"); uint64_t total = json_u64(torrent, "total_bytes"); uint64_t pieces = json_u64(torrent, "total_pieces"); @@ -430,6 +569,23 @@ static json_t *map_torrent(json_t *torrent, bool detail) { double progress = total ? (double)done / (double)total : (pieces ? (double)pieces_done / (double)pieces : 0.0); if (progress > 1.0) progress = 1.0; + if (done_out) *done_out = done; + if (total_out) *total_out = total; + return progress; +} + +static json_int_t compute_eta(uint64_t done, uint64_t total, double dlspeed) { + if (total > done && dlspeed >= 1.0) + return (json_int_t)((double)(total - done) / dlspeed); + return ETA_INFINITY; +} + +static json_t *map_torrent(json_t *torrent, bool detail, double dlspeed) { + uint64_t id = json_u64(torrent, "torrent_id"); + uint64_t done = 0, total = 0; + double progress = torrent_progress_ratio(torrent, &done, &total); + uint64_t pieces = json_u64(torrent, "total_pieces"); + uint64_t pieces_done = json_u64(torrent, "pieces_done"); char hash[32]; snprintf(hash, sizeof hash, "%llu", (unsigned long long)id); @@ -439,43 +595,46 @@ static json_t *map_torrent(json_t *torrent, bool detail) { "state")), progress); - json_t *trackers = json_array(); - if (trackers) { - json_array_append_new(trackers, json_pack( - "{s:s,s:i,s:s,s:i,s:i,s:i,s:i,s:s}", - "url", "** [DHT] **", "tier", -1, "status", "working", - "seeds", (int)json_u64(torrent, "peers_discovered"), - "peers", (int)json_u64(torrent, "peers"), - "leeches", -1, "downloaded", -1, "message", "")); + json_t *trackers = NULL; + json_t *files = NULL; + json_t *peers_list = NULL; + json_t *hosts = NULL; + if (detail) { + trackers = json_array(); + if (trackers) + json_array_append_new(trackers, json_pack( + "{s:s,s:i,s:s,s:i,s:i,s:i,s:i,s:s}", + "url", "** [DHT] **", "tier", -1, "status", "working", + "seeds", (int)json_u64(torrent, "peers_discovered"), + "peers", (int)json_u64(torrent, "peers"), + "leeches", -1, "downloaded", -1, "message", "")); + files = json_array(); + if (files) + json_array_append_new(files, json_pack( + "{s:s,s:I,s:f,s:i,s:f}", "name", name, + "size", (json_int_t)total, "progress", progress, + "priority", 1, "availability", 1.0)); + peers_list = json_array(); + hosts = tracker_hosts(trackers); } - json_t *files = json_array(); - if (files) { - json_array_append_new(files, json_pack( - "{s:s,s:I,s:f,s:i,s:f}", "name", name, - "size", (json_int_t)total, "progress", progress, - "priority", 1, "availability", 1.0)); - } - json_t *peers_list = json_array(); - json_t *tags = json_array(); - json_t *hosts = tracker_hosts(trackers); json_t *out = json_pack( - "{s:s,s:s,s:I,s:f,s:i,s:i,s:i,s:i,s:i,s:i,s:f,s:s,s:o,s:s," + "{s:s,s:s,s:I,s:f,s:I,s:i,s:I,s:i,s:i,s:i,s:f,s:s,s:o,s:s," "s:I,s:I,s:I,s:I,s:f,s:i,s:o,s:b,s:b,s:b,s:I,s:I,s:s,s:s}", "hash", hash, "name", name, "size", (json_int_t)total, "progress", progress, - "dlspeed", 0, + "dlspeed", (json_int_t)dlspeed, "upspeed", 0, - "eta", progress > 0.0 && progress < 1.0 ? 8640000 : 0, + "eta", compute_eta(done, total, dlspeed), "seeds", (int)json_u64(torrent, "peers"), "seedsTotal", (int)json_u64(torrent, "peers_discovered"), "peers", (int)json_u64(torrent, "peers_connecting"), "peersTotal", (int)json_u64(torrent, "peers_discovered"), "ratio", 0.0, "category", "", - "tags", tags ? tags : json_array(), + "tags", json_array(), "savePath", json_string_or(torrent, "output", ""), "addedOn", (json_int_t)0, "completionOn", progress >= 1.0 ? (json_int_t)0 : (json_int_t)-1, @@ -504,16 +663,18 @@ static json_t *map_torrent(json_t *torrent, bool detail) { json_object_set_new(out, "trackers", trackers ? trackers : json_array()); json_object_set_new(out, "peersList", peers_list ? peers_list : json_array()); json_object_set_new(out, "files", files ? files : json_array()); - } else { + } else if (detail) { json_decref(trackers); json_decref(files); json_decref(peers_list); + json_decref(hosts); } free(name); return out; } -static json_t *snapshot_json(void) { +/* Build a fresh snapshot (grid + global stats) with live download rates. */ +static json_t *build_snapshot(void) { json_t *params = json_object(); json_t *torrents = rpc_call_json("torrents", params); json_decref(params); @@ -521,24 +682,32 @@ static json_t *snapshot_json(void) { json_decref(torrents); torrents = json_array(); } + speed_retain(torrents); json_t *items = json_array(); uint64_t active = 0; + uint64_t total_rate = 0; + uint64_t total_data = 0; size_t index; json_t *torrent; json_array_foreach(torrents, index, torrent) { - json_t *mapped = map_torrent(torrent, false); + uint64_t id = json_u64(torrent, "torrent_id"); + uint64_t done = json_u64(torrent, "bytes_done"); + double dlspeed = speed_sample(id, done); + json_t *mapped = map_torrent(torrent, false, dlspeed); if (!mapped) continue; const char *state = json_string_value(json_object_get(mapped, "state")); if (state && strcmp(state, "downloading") == 0) active++; + total_rate += (uint64_t)dlspeed; + total_data += done; json_array_append_new(items, mapped); } json_decref(torrents); json_t *server = json_pack( - "{s:i,s:i,s:i,s:i,s:i,s:i,s:f,s:i,s:s,s:i,s:I,s:i,s:i,s:s,s:i}", - "dl_info_speed", 0, + "{s:I,s:i,s:I,s:i,s:i,s:i,s:f,s:i,s:s,s:i,s:I,s:i,s:i,s:s,s:i}", + "dl_info_speed", (json_int_t)total_rate, "up_info_speed", 0, - "dl_info_data", 0, + "dl_info_data", (json_int_t)total_data, "up_info_data", 0, "dl_rate_limit", 0, "up_rate_limit", 0, @@ -555,6 +724,43 @@ static json_t *snapshot_json(void) { "server", server, "torrents", items); } +/* Publish a newly built snapshot for all readers; wakes SSE waiters. */ +static void publish_snapshot(void) { + json_t *snapshot = build_snapshot(); + if (!snapshot) return; + char *full = json_dumps(snapshot, JSON_COMPACT | JSON_ENCODE_ANY); + json_t *torrents = json_object_get(snapshot, "torrents"); + char *list = json_dumps(torrents ? torrents : json_array(), + JSON_COMPACT | JSON_ENCODE_ANY); + json_decref(snapshot); + if (!full || !list) { + free(full); + free(list); + return; + } + pthread_mutex_lock(&g_webui.snap_lock); + free(g_webui.snapshot_str); + free(g_webui.torrents_str); + g_webui.snapshot_str = full; + g_webui.torrents_str = list; + g_webui.snap_seq++; + pthread_cond_broadcast(&g_webui.snap_cond); + pthread_mutex_unlock(&g_webui.snap_lock); +} + +static void *sampler_thread(void *arg) { + (void)arg; + while (!atomic_load(&g_webui.stopping)) { + publish_snapshot(); + /* sleep ~1s but stay responsive to shutdown */ + for (int i = 0; i < 10 && !atomic_load(&g_webui.stopping); i++) { + struct timespec ts = { .tv_sec = 0, .tv_nsec = 100 * 1000 * 1000 }; + nanosleep(&ts, NULL); + } + } + return NULL; +} + static bool parse_id(const char *text, uint64_t *id) { if (!text || !*text) return false; char *end = NULL; @@ -571,7 +777,7 @@ static json_t *full_torrent_by_hash(const char *hash) { json_t *torrent = rpc_call_json("torrent", params); json_decref(params); if (!torrent) return NULL; - json_t *mapped = map_torrent(torrent, true); + json_t *mapped = map_torrent(torrent, true, speed_peek(id)); json_decref(torrent); return mapped; } @@ -714,6 +920,8 @@ static void api_add(int fd, const char *body, size_t len) { http_json(fd, 200, reply); json_decref(reply); json_decref(result); + /* refresh the shared snapshot so the new torrent shows up immediately */ + publish_snapshot(); } static void api_delete(int fd, const char *body, size_t len) { @@ -739,12 +947,23 @@ static void api_delete(int fd, const char *body, size_t len) { json_t *reply = json_pack("{s:b,s:i}", "ok", 1, "removed", (int)removed); http_json(fd, 200, reply); json_decref(reply); + if (removed) publish_snapshot(); } -static void api_noop(int fd) { - json_t *reply = json_pack("{s:b,s:i}", "ok", 1, "affected", 0); - http_json(fd, 200, reply); - json_decref(reply); +/* The engine has no pause/resume/recheck/queue/category/limit verbs yet, so + * rather than claim success we tell the UI the action is unsupported. The + * front end surfaces a non-2xx as an honest "Action failed" toast. */ +static void api_action(int fd, const char *body, size_t len) { + json_t *req = read_body_json(body, len); + const char *action = json_string_value(json_object_get(req, "action")); + char message[128]; + snprintf(message, sizeof message, + "action '%s' is not supported by the engine", + action ? action : ""); + json_decref(req); + json_t *json = json_pack("{s:b,s:s}", "ok", 0, "error", message); + http_json(fd, 501, json); + json_decref(json); } static void api_stream(int fd) { @@ -753,20 +972,47 @@ static void api_stream(int fd) { "Cache-Control: no-cache\r\nConnection: keep-alive\r\n\r\n" "retry: 2000\n\n"; if (!send_all_fd(fd, head, strlen(head))) return; + uint64_t seen = 0; while (!atomic_load(&g_webui.stopping)) { - json_t *snapshot = snapshot_json(); - char *text = json_dumps(snapshot, JSON_COMPACT | JSON_ENCODE_ANY); - json_decref(snapshot); - if (!text) break; + char *payload = NULL; + pthread_mutex_lock(&g_webui.snap_lock); + while (!atomic_load(&g_webui.stopping) && g_webui.snap_seq == seen) { + struct timespec ts; + clock_gettime(CLOCK_REALTIME, &ts); + ts.tv_nsec += 250 * 1000 * 1000; + if (ts.tv_nsec >= 1000000000) { ts.tv_sec++; ts.tv_nsec -= 1000000000; } + pthread_cond_timedwait(&g_webui.snap_cond, &g_webui.snap_lock, &ts); + } + if (!atomic_load(&g_webui.stopping) && g_webui.snapshot_str) { + payload = strdup(g_webui.snapshot_str); + seen = g_webui.snap_seq; + } + pthread_mutex_unlock(&g_webui.snap_lock); + if (!payload) break; bool ok = send_all_fd(fd, "event: snapshot\ndata: ", 22) && - send_all_fd(fd, text, strlen(text)) && + send_all_fd(fd, payload, strlen(payload)) && send_all_fd(fd, "\n\n", 2); - free(text); + free(payload); if (!ok) break; - sleep(1); } } +static void serve_cached_snapshot(int fd) { + pthread_mutex_lock(&g_webui.snap_lock); + char *copy = g_webui.snapshot_str ? strdup(g_webui.snapshot_str) : NULL; + pthread_mutex_unlock(&g_webui.snap_lock); + http_json_str(fd, copy, "{\"server\":{},\"torrents\":[]}"); + free(copy); +} + +static void serve_cached_torrents(int fd) { + pthread_mutex_lock(&g_webui.snap_lock); + char *copy = g_webui.torrents_str ? strdup(g_webui.torrents_str) : NULL; + pthread_mutex_unlock(&g_webui.snap_lock); + http_json_str(fd, copy, "[]"); + free(copy); +} + static void handle_api(int fd, const char *method, char *path, const char *headers, const char *headers_end, const char *body, size_t body_len) { @@ -785,8 +1031,10 @@ static void handle_api(int fd, const char *method, char *path, const char *user = json_string_value(json_object_get(req, "username")); const char *password = json_string_value(json_object_get(req, "password")); - if (!user || !password || strcmp(user, g_webui.auth_user) != 0 || - strcmp(password, g_webui.auth_password) != 0) { + bool user_ok = user && constant_time_equal(user, g_webui.auth_user); + bool pass_ok = password && g_webui.auth_password[0] && + constant_time_equal(password, g_webui.auth_password); + if (!user_ok || !pass_ok) { json_t *json = json_pack("{s:b,s:s}", "ok", 0, "error", "invalid credentials"); http_json(fd, 401, json); @@ -827,9 +1075,7 @@ static void handle_api(int fd, const char *method, char *path, } else if (strcmp(path, "/api/stream") == 0 && strcmp(method, "GET") == 0) { api_stream(fd); } else if (strcmp(path, "/api/snapshot") == 0 && strcmp(method, "GET") == 0) { - json_t *json = snapshot_json(); - http_json(fd, 200, json); - json_decref(json); + serve_cached_snapshot(fd); } else if (strcmp(path, "/api/meta") == 0 && strcmp(method, "GET") == 0) { api_meta(fd); } else if (strcmp(path, "/api/preferences") == 0) { @@ -851,11 +1097,7 @@ static void handle_api(int fd, const char *method, char *path, http_json(fd, 200, json); json_decref(json); } else if (strcmp(path, "/api/torrents") == 0 && strcmp(method, "GET") == 0) { - json_t *snapshot = snapshot_json(); - json_t *torrents = json_incref(json_object_get(snapshot, "torrents")); - http_json(fd, 200, torrents); - json_decref(torrents); - json_decref(snapshot); + serve_cached_torrents(fd); } else if (path_after(path, "/api/torrents/") && strcmp(method, "GET") == 0) { api_torrent_detail(fd, path_after(path, "/api/torrents/")); } else if (strcmp(path, "/api/add") == 0 && strcmp(method, "POST") == 0) { @@ -863,7 +1105,7 @@ static void handle_api(int fd, const char *method, char *path, } else if (strcmp(path, "/api/delete") == 0 && strcmp(method, "POST") == 0) { api_delete(fd, body, body_len); } else if (strcmp(path, "/api/action") == 0 && strcmp(method, "POST") == 0) { - api_noop(fd); + api_action(fd, body, body_len); } else if (strncmp(path, "/api/rss", 8) == 0 && strcmp(method, "GET") == 0) { json_t *json = json_array(); http_json(fd, 200, json); @@ -914,7 +1156,14 @@ static void handle_conn(int fd) { size_t content_length = 0; char *cl = strcasestr(request, "content-length:"); if (cl && cl < hdrend) content_length = strtoull(cl + 15, NULL, 10); - if (content_length > READ_LIMIT - header_len) content_length = READ_LIMIT - header_len; + /* Reject bodies we can't buffer instead of silently truncating an upload + * into a corrupt torrent. */ + if (content_length > READ_LIMIT - header_len) { + free(request); + http_text(fd, 413, "Payload Too Large", + "request body exceeds the 8 MiB limit"); + return; + } while (len - header_len < content_length && len < READ_LIMIT) { ssize_t n = recv(fd, request + len, READ_LIMIT - len, 0); if (n < 0) { @@ -962,15 +1211,26 @@ static void *server_thread(void *arg) { struct timeval timeout = { .tv_sec = 5, .tv_usec = 0 }; setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof timeout); setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof timeout); - conn_arg *conn = malloc(sizeof(*conn)); - if (!conn) { + + /* Bound concurrent connections so a client can't spawn unlimited + * threads (each SSE stream parks one). */ + pthread_mutex_lock(&g_webui.conn_lock); + bool full = g_webui.active_connections >= MAX_CONNECTIONS; + if (!full) g_webui.active_connections++; + pthread_mutex_unlock(&g_webui.conn_lock); + if (full) { + http_text(fd, 503, "Service Unavailable", "too many connections"); close(fd); continue; } + + conn_arg *conn = malloc(sizeof(*conn)); + if (!conn) { + close(fd); + finish_connection(); + continue; + } conn->fd = fd; - pthread_mutex_lock(&g_webui.conn_lock); - g_webui.active_connections++; - pthread_mutex_unlock(&g_webui.conn_lock); pthread_t thread; if (pthread_create(&thread, NULL, conn_thread, conn) != 0) { close(fd); @@ -1040,7 +1300,19 @@ static naut_err start_server(void) { return NAUT_ERR_IO; } g_webui.listener = fd; + + /* Prime the cache so the first request doesn't see an empty snapshot. */ + publish_snapshot(); + if (pthread_create(&g_webui.sampler, NULL, sampler_thread, NULL) != 0) { + close(fd); + g_webui.listener = -1; + return NAUT_ERR_NOMEM; + } + g_webui.sampler_started = true; if (pthread_create(&g_webui.thread, NULL, server_thread, NULL) != 0) { + atomic_store(&g_webui.stopping, true); + pthread_join(g_webui.sampler, NULL); + g_webui.sampler_started = false; close(fd); g_webui.listener = -1; return NAUT_ERR_NOMEM; @@ -1050,12 +1322,18 @@ static naut_err start_server(void) { snprintf(msg, sizeof msg, "webui: serving http://%s:%d from %s", g_webui.host_name, g_webui.port, g_webui.root); log_msg(2, msg); + if (strcmp(g_webui.host_name, DEFAULT_HOST) != 0) + log_msg(1, "webui: bound to a non-loopback address; credentials cross " + "the network in plaintext (set NAUT_AUTH_PASSWORD)"); snprintf(msg, sizeof msg, "webui: auth user %s", g_webui.auth_user); log_msg(2, msg); - if (g_webui.generated_password) { + if (g_webui.generated_password && g_webui.auth_password[0]) { snprintf(msg, sizeof msg, "webui: generated password %s", g_webui.auth_password); log_msg(1, msg); + } else if (!g_webui.auth_password[0]) { + log_msg(0, "webui: no password available (CSPRNG unavailable); set " + "NAUT_AUTH_PASSWORD to enable login"); } return NAUT_OK; } @@ -1064,40 +1342,51 @@ naut_err naut_plugin_register(const naut_host_api *host) { if (!host || host->abi_version != NAUT_PLUGIN_ABI_VERSION || host->struct_size < sizeof(*host) || !host->call_rpc) return NAUT_ERR_INVAL; + naut_err error = NAUT_ERR_NOMEM; memset(&g_webui, 0, sizeof g_webui); g_webui.listener = -1; g_webui.host = *host; if (pthread_mutex_init(&g_webui.auth_lock, NULL) != 0) return NAUT_ERR_NOMEM; - if (pthread_mutex_init(&g_webui.conn_lock, NULL) != 0) { - pthread_mutex_destroy(&g_webui.auth_lock); - return NAUT_ERR_NOMEM; - } - if (pthread_cond_init(&g_webui.conn_cond, NULL) != 0) { - pthread_mutex_destroy(&g_webui.conn_lock); - pthread_mutex_destroy(&g_webui.auth_lock); - return NAUT_ERR_NOMEM; - } + if (pthread_mutex_init(&g_webui.conn_lock, NULL) != 0) + goto fail_conn_lock; + if (pthread_cond_init(&g_webui.conn_cond, NULL) != 0) + goto fail_conn_cond; + if (pthread_mutex_init(&g_webui.snap_lock, NULL) != 0) + goto fail_snap_lock; + if (pthread_cond_init(&g_webui.snap_cond, NULL) != 0) + goto fail_snap_cond; + if (pthread_mutex_init(&g_webui.speed_lock, NULL) != 0) + goto fail_speed_lock; init_auth(); - naut_err error = g_webui.host.set_plugin_name(g_webui.host.host_context, + error = g_webui.host.set_plugin_name(g_webui.host.host_context, "webui"); - if (error != NAUT_OK) { - pthread_cond_destroy(&g_webui.conn_cond); - pthread_mutex_destroy(&g_webui.conn_lock); - pthread_mutex_destroy(&g_webui.auth_lock); - return error; - } + if (error != NAUT_OK) goto fail_named; error = start_server(); - if (error != NAUT_OK) { - pthread_cond_destroy(&g_webui.conn_cond); - pthread_mutex_destroy(&g_webui.conn_lock); - pthread_mutex_destroy(&g_webui.auth_lock); - } + if (error != NAUT_OK) goto fail_named; + return NAUT_OK; + +fail_named: + pthread_mutex_destroy(&g_webui.speed_lock); +fail_speed_lock: + pthread_cond_destroy(&g_webui.snap_cond); +fail_snap_cond: + pthread_mutex_destroy(&g_webui.snap_lock); +fail_snap_lock: + pthread_cond_destroy(&g_webui.conn_cond); +fail_conn_cond: + pthread_mutex_destroy(&g_webui.conn_lock); +fail_conn_lock: + pthread_mutex_destroy(&g_webui.auth_lock); return error; } naut_err naut_plugin_shutdown(void) { atomic_store(&g_webui.stopping, true); + /* wake any SSE streams parked on the snapshot condition */ + pthread_mutex_lock(&g_webui.snap_lock); + pthread_cond_broadcast(&g_webui.snap_cond); + pthread_mutex_unlock(&g_webui.snap_lock); if (g_webui.listener >= 0) { shutdown(g_webui.listener, SHUT_RDWR); close(g_webui.listener); @@ -1106,10 +1395,22 @@ naut_err naut_plugin_shutdown(void) { if (g_webui.thread_started) pthread_join(g_webui.thread, NULL); g_webui.thread_started = false; + if (g_webui.sampler_started) + pthread_join(g_webui.sampler, NULL); + g_webui.sampler_started = false; pthread_mutex_lock(&g_webui.conn_lock); while (g_webui.active_connections > 0) pthread_cond_wait(&g_webui.conn_cond, &g_webui.conn_lock); pthread_mutex_unlock(&g_webui.conn_lock); + + free(g_webui.snapshot_str); + free(g_webui.torrents_str); + g_webui.snapshot_str = NULL; + g_webui.torrents_str = NULL; + + pthread_mutex_destroy(&g_webui.speed_lock); + pthread_cond_destroy(&g_webui.snap_cond); + pthread_mutex_destroy(&g_webui.snap_lock); pthread_cond_destroy(&g_webui.conn_cond); pthread_mutex_destroy(&g_webui.conn_lock); pthread_mutex_destroy(&g_webui.auth_lock);