From 58bc4bbd52ab22f7d02972fda84836fe39479e20 Mon Sep 17 00:00:00 2001 From: wudgexyz Date: Wed, 14 Oct 2026 18:10:36 +0300 Subject: [PATCH] Motion events: tell the cloud when majestic sees movement, with frames - Installs /usr/sbin/motion.sh, which only wakes the agent (SIGUSR1, pid checked against /proc//cmdline); a hook that was already there is kept and still called - Hook calls become episodes, posted signed to POST /nodes/{id}/events with a key derived from the device key and the coordinator's eventKey - When the cloud asks for frames, sends one a second from majestic's local /image.jpg until told to stop; `cerberus snapshot` and `cerberus motion` to check by hand - uninstall/clean put the original hook back; READMEs and the test suite cover it --- README.md | 28 +- README.ru.md | 30 +- README.zh-CN.md | 4 + S99cerberus | 3 + main.c | 691 +++++++++++++++++++++++++++++++++++--- tests/fake_coordinator.py | 136 +++++++- tests/suite.sh | 53 ++- uninstall.sh | 11 +- 8 files changed, 909 insertions(+), 47 deletions(-) diff --git a/README.md b/README.md index a4e85b3..471b650 100644 --- a/README.md +++ b/README.md @@ -122,6 +122,8 @@ Almost all of that is mbedTLS built with its default configuration; the agent it | `cerberus_ping_interval` | `CERBERUS_PING_INTERVAL` | `60` | seconds between probes; `0` disables | | `cerberus_ping_timeout` | `CERBERUS_PING_TIMEOUT` | `3` | seconds one probe waits | | `cerberus_ping_fails` | `CERBERUS_PING_FAILS` | `3` | failed probes in a row before the interface is rebuilt | +| `cerberus_events` | `CERBERUS_EVENTS` | `1` | motion events; `0` installs no hook and sends nothing | +| `cerberus_snapshot_url` | `CERBERUS_SNAPSHOT_URL` | `http://127.0.0.1/image.jpg?width=640&height=360` | majestic's frame; `http://user:pass@127.0.0.1/…` when it wants a password | `cerberus_ca` matters more than it looks: a firmware may carry no CA bundle at all, and then enrolment fails on the coordinator's certificate with no way to fix it short of a reflash — point it @@ -131,7 +133,8 @@ Variables only the agent reads: `CERBERUS_STORE` (`auto` \| `uboot` \| `file:/pa `CERBERUS_STALE_AFTER` (`180`, seconds without a handshake before the peer is re-poked), `CERBERUS_RETRY_MIN` / `CERBERUS_RETRY_MAX` (`5` / `300`, enrolment backoff), `CERBERUS_INSECURE=1` (skip TLS verification, debugging only), `CERBERUS_DRY_RUN=1` (print the `ip`/`wg` commands instead -of running them). +of running them), `CERBERUS_MOTION_HOOK` (`/usr/sbin/motion.sh`), `CERBERUS_MOTION_QUIET` (`10`, +seconds without motion that end an episode), `CERBERUS_PIDFILE` (`/var/run/cerberus.pid`). ``` cerberus run # enroll + up + watch (the default; what S99cerberus starts) @@ -140,6 +143,8 @@ cerberus up # bring the interface up from the stored material cerberus status # what is stored, handshake age, whether the tunnel answers cerberus keygen # print the keypair (stdout is data, diagnostics go to stderr) cerberus clean # remove the agent: interface, init script, stored material, the binary +cerberus motion # tell the running agent there is motion, as majestic's hook does +cerberus snapshot # fetch one frame from CERBERUS_SNAPSHOT_URL into /tmp/cerberus-snapshot.jpg ``` ### Formerly `streq_*` @@ -214,6 +219,27 @@ Video does not pass through the agent: the relay pulls `rtsp://{overlayIp}:554/s the camera through the tunnel. There is no peer-to-peer on this agent — the camera stays on the hub path, which is the path that always works. +### Motion events + +majestic notices motion itself once its motion detection is switched on (web UI → Majestic → +Motion detection), and runs `/usr/sbin/motion.sh`. cerberus installs that script — keeping one that +was already there as `motion.sh.orig` and still calling it — and the script does nothing but wake +the agent. The agent turns the calls into episodes (motion, then `CERBERUS_MOTION_QUIET` seconds of +quiet) and reports each one twice, `started` and `ended`, to `POST /nodes/{id}/events`. + +When the owner has switched object detection on for the camera, the answer to `started` asks for +frames: one a second from majestic's own `/image.jpg`, until the cloud says it has seen enough. The +cloud looks for people, cars and animals in them and tells the owner — the camera only says "look +now", from the moment it happens, without the cloud pulling the stream or knowing the RTSP password. + +Events are signed. The heartbeat's answer carries the coordinator's `eventKey`; the key is +`HKDF-SHA256(X25519(device key, eventKey), salt = nodeId, info = "streq-events-v1")` and each request +carries `X-Streq-Ts` and `X-Streq-Sig = base64(HMAC-SHA256(key, ts \n nodeId \n scope \n body))`. No new +secret is stored on the camera: it lives exactly as long as the device key. + +To check it on a camera: `cerberus snapshot` (does majestic hand out a frame as configured?) and +`cerberus motion` (an episode, as if somebody walked by). + ## Building ```sh diff --git a/README.ru.md b/README.ru.md index 1f950ba..8893716 100644 --- a/README.ru.md +++ b/README.ru.md @@ -119,6 +119,8 @@ fw_setenv cerberus_enrollment_token tok_... | `cerberus_ping_interval` | `CERBERUS_PING_INTERVAL` | `60` | секунд между пробами; `0` — выключить | | `cerberus_ping_timeout` | `CERBERUS_PING_TIMEOUT` | `3` | сколько секунд ждать ответа на пробу | | `cerberus_ping_fails` | `CERBERUS_PING_FAILS` | `3` | неотвеченных проб подряд до пересборки интерфейса | +| `cerberus_events` | `CERBERUS_EVENTS` | `1` | события движения; `0` — без хука и без событий | +| `cerberus_snapshot_url` | `CERBERUS_SNAPSHOT_URL` | `http://127.0.0.1/image.jpg?width=640&height=360` | кадр majestic; `http://user:pass@127.0.0.1/…`, если он просит пароль | `cerberus_ca` важнее, чем кажется: в прошивке может не быть бандла CA вовсе, и тогда энролл падает на сертификате координатора, а поправить это без перепрошивки нечем — укажите путь к корневому @@ -127,7 +129,9 @@ fw_setenv cerberus_enrollment_token tok_... Переменные, которые читает только агент: `CERBERUS_STORE` (`auto` \| `uboot` \| `file:/path`), `CERBERUS_STALE_AFTER` (`180`, секунд без хендшейка до перетыкания пира), `CERBERUS_RETRY_MIN` / `CERBERUS_RETRY_MAX` (`5` / `300`, отступ повторов энролла), `CERBERUS_INSECURE=1` (без проверки TLS, -только для отладки), `CERBERUS_DRY_RUN=1` (печатать команды `ip`/`wg`, а не выполнять). +только для отладки), `CERBERUS_DRY_RUN=1` (печатать команды `ip`/`wg`, а не выполнять), +`CERBERUS_MOTION_HOOK` (`/usr/sbin/motion.sh`), `CERBERUS_MOTION_QUIET` (`10`, секунд тишины, после +которых эпизод кончился), `CERBERUS_PIDFILE` (`/var/run/cerberus.pid`). ``` cerberus run # enroll + up + watch (по умолчанию; это запускает S99cerberus) @@ -136,6 +140,8 @@ cerberus up # поднять интерфейс из сохранённо cerberus status # что сохранено, возраст хендшейка, отвечает ли туннель cerberus keygen # напечатать пару ключей (stdout — данные, диагностика в stderr) cerberus clean # убрать агента: интерфейс, автозапуск, сохранённое и сам бинарь +cerberus motion # сказать работающему агенту «движение», как это делает хук majestic +cerberus snapshot # взять один кадр с CERBERUS_SNAPSHOT_URL в /tmp/cerberus-snapshot.jpg ``` ### Раньше — `streq_*` @@ -206,6 +212,28 @@ cerberus clean # убрать агента: интерфейс, автоза От недоступного координатора `cerberus run` тоже не умирает: init на камере запускает его один раз и больше не смотрит, поэтому энролл повторяется с удваивающейся паузой сколько понадобится. +### События движения + +Движение majestic замечает сам, когда в нём включена детекция движения (веб-интерфейс → Majestic → +Детекция движения), и запускает `/usr/sbin/motion.sh`. Этот скрипт ставит cerberus — если там уже +лежал чужой, он сохраняется как `motion.sh.orig` и по-прежнему вызывается, — и скрипт ничего не +делает, кроме как будит агента. Агент собирает вызовы в эпизоды (движение, затем +`CERBERUS_MOTION_QUIET` секунд тишины) и сообщает о каждом дважды, `started` и `ended`, в +`POST /nodes/{id}/events`. + +Если владелец включил для камеры детекцию объектов, ответ на `started` просит кадры: по одному в +секунду с собственного `/image.jpg` majestic, пока облако не скажет, что увидело достаточно. Облако +ищет в них людей, машины и животных и сообщает владельцу. Камера говорит только «смотри сейчас», с +самого момента движения: облаку не нужно ни тянуть поток, ни знать пароль RTSP. + +События подписаны. В ответе на heartbeat едет `eventKey` координатора; ключ — +`HKDF-SHA256(X25519(ключ устройства, eventKey), salt = nodeId, info = "streq-events-v1")`, у каждого +запроса `X-Streq-Ts` и `X-Streq-Sig = base64(HMAC-SHA256(ключ, ts \n nodeId \n scope \n тело))`. Нового +секрета на камере не появляется: подпись живёт ровно столько, сколько ключ устройства. + +Проверить на камере: `cerberus snapshot` (отдаёт ли majestic кадр по настроенному адресу) и +`cerberus motion` (эпизод, как будто кто-то прошёл мимо). + Видео через агента не идёт: релей тянет `rtsp://{overlayIp}:554/stream0` прямо с камеры по туннелю. P2P этот агент не делает — камера остаётся на пути через хаб, который работает всегда. diff --git a/README.zh-CN.md b/README.zh-CN.md index 9f83c76..690edbf 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -114,6 +114,8 @@ fw_setenv cerberus_enrollment_token tok_... | `cerberus_ping_interval` | `CERBERUS_PING_INTERVAL` | `60` | 探测间隔（秒），`0` 表示关闭 | | `cerberus_ping_timeout` | `CERBERUS_PING_TIMEOUT` | `3` | 单次探测等待的秒数 | | `cerberus_ping_fails` | `CERBERUS_PING_FAILS` | `3` | 连续多少次探测无响应后重建接口 | +| `cerberus_events` | `CERBERUS_EVENTS` | `1` | 移动事件；`0` 表示不安装钩子、不发送事件 | +| `cerberus_snapshot_url` | `CERBERUS_SNAPSHOT_URL` | `http://127.0.0.1/image.jpg?width=640&height=360` | majestic 的截图；需要密码时写成 `http://user:pass@127.0.0.1/…` | `cerberus_ca` 比看上去更重要：固件可能根本不带 CA 证书包，此时注册会因协调器证书校验失败而中断， 除了重新刷写固件之外无法修复——请将其指向根证书（Let's Encrypt 使用 ISRG Root X1）。 @@ -130,6 +132,8 @@ cerberus up # 用已保存的材料启动接口 cerberus status # 已保存的内容、握手时间、隧道是否有响应 cerberus keygen # 打印密钥对（stdout 为数据，诊断信息输出到 stderr） cerberus clean # 移除代理：接口、启动脚本、已保存的材料以及二进制文件本身 +cerberus motion # 像 majestic 的钩子一样，通知正在运行的代理“有移动” +cerberus snapshot # 从 CERBERUS_SNAPSHOT_URL 取一帧到 /tmp/cerberus-snapshot.jpg ``` ### 以前的 `streq_*` diff --git a/S99cerberus b/S99cerberus index 5fe9fa1..cbf7450 100755 --- a/S99cerberus +++ b/S99cerberus @@ -25,6 +25,9 @@ export CERBERUS_PING_TARGET="$(uenv ping_target)" export CERBERUS_PING_INTERVAL="$(uenv ping_interval)" export CERBERUS_PING_TIMEOUT="$(uenv ping_timeout)" export CERBERUS_PING_FAILS="$(uenv ping_fails)" +# Motion events: on by default. cerberus_snapshot_url carries majestic's password when it has one. +export CERBERUS_EVENTS="$(uenv events)" +export CERBERUS_SNAPSHOT_URL="$(uenv snapshot_url)" CERBERUS_IFACE=$(uenv iface) export CERBERUS_IFACE=${CERBERUS_IFACE:-wg0} diff --git a/main.c b/main.c index 5d92766..c0fdb20 100644 --- a/main.c +++ b/main.c @@ -14,6 +14,10 @@ * cannot hold a SignalR connection (the coordinator's "thin node" REST twins) — and applies the * NodeDirective that comes back: rotated join material (the hub moved or re-keyed) and one-shot * commands (Restart / Reboot). That answer is the only back-channel a thin node has. + * 6. tells the cloud about motion: majestic runs a hook, the hook wakes us, and we post a signed + * episode (started / ended) plus, while it lasts and only if the owner switched detection on, + * a frame a second from majestic's own /image.jpg. What is in the frame — a person, a car — is + * decided by the cloud's object detector; the camera only says "look now". * * It is deliberately NOT the .NET agent: no SignalR, no map polling, no go2rtc republishing. A camera * that is itself on the overlay needs none of that — the relay pulls rtsp://{overlayIp}:554/stream0 @@ -44,6 +48,7 @@ #include #include #include +#include #include #include #include @@ -63,7 +68,7 @@ /* Release builds pass the tag in (`make VERSION=...`); a plain `make` reports the source version. */ #ifndef CERBERUS_VERSION -#define CERBERUS_VERSION "0.1.0" +#define CERBERUS_VERSION "0.2.0" #endif /* Where the errors are decoded. Printed after every fatal one: the reader is on a camera @@ -140,6 +145,13 @@ struct cfg { int ping_fails; /* consecutive failed probes before the interface is rebuilt */ int retry_min; /* first pause before a failed enrolment is retried */ int retry_max; /* ceiling that pause backs off to */ + int events; /* report motion episodes; 0 disables the hook and the events */ + const char *motion_hook; /* the script majestic runs on motion */ + const char *pidfile; /* where the hook finds us */ + const char *snapshot_url; /* majestic's JPEG, user:pass@ allowed */ + int snapshot_max; /* largest frame we accept from majestic, bytes */ + int motion_quiet; /* seconds without motion that end an episode */ + int motion_max; /* an episode is cut after this many seconds, whatever happens */ }; static struct cfg g_cfg; @@ -197,6 +209,20 @@ static void cfg_load(void) g_cfg.retry_max = atoi(cfg_env("RETRY_MAX", "300")); if (g_cfg.retry_min < 1) g_cfg.retry_min = 1; if (g_cfg.retry_max < g_cfg.retry_min) g_cfg.retry_max = g_cfg.retry_min; + + /* Motion. The frame is majestic's: taken on the camera at the moment of motion, so the cloud + * needs neither the stream nor the RTSP password — and gets it without the seconds a relay + * spends bringing a stream up. 640 px wide is plenty for a person at the gate. */ + g_cfg.events = atoi(cfg_env("EVENTS", "1")); + g_cfg.motion_hook = cfg_env("MOTION_HOOK", "/usr/sbin/motion.sh"); + g_cfg.pidfile = cfg_env("PIDFILE", "/var/run/cerberus.pid"); + g_cfg.snapshot_url = cfg_env("SNAPSHOT_URL", "http://127.0.0.1/image.jpg?width=640&height=360"); + g_cfg.snapshot_max = atoi(cfg_env("SNAPSHOT_MAX", "524288")); + g_cfg.motion_quiet = atoi(cfg_env("MOTION_QUIET", "10")); + g_cfg.motion_max = atoi(cfg_env("MOTION_MAX", "600")); + if (g_cfg.snapshot_max < 16384) g_cfg.snapshot_max = 16384; + if (g_cfg.motion_quiet < 1) g_cfg.motion_quiet = 1; + if (g_cfg.motion_max < g_cfg.motion_quiet) g_cfg.motion_max = g_cfg.motion_quiet; } /* ------------------------------------------------------------------ crypto */ @@ -273,6 +299,56 @@ static int x25519_pub(const unsigned char priv[32], unsigned char pub[32]) return rc; } +/* + * X25519(priv, peer): the shared secret, little-endian, as RFC 7748 and the coordinator compute it. + * Same scalar multiplication as x25519_pub, with the peer's point in place of the base point. + */ +static int x25519_shared(const unsigned char priv[32], const unsigned char peer[32], unsigned char out[32]) +{ + mbedtls_ecp_group grp; + mbedtls_ecp_point P, R; + mbedtls_mpi d; + size_t olen = 0; + int rc; + + mbedtls_ecp_group_init(&grp); + mbedtls_ecp_point_init(&P); + mbedtls_ecp_point_init(&R); + mbedtls_mpi_init(&d); + + rc = mbedtls_ecp_group_load(&grp, MBEDTLS_ECP_DP_CURVE25519); + if (rc == 0) rc = mbedtls_mpi_read_binary_le(&d, priv, 32); + if (rc == 0) rc = mbedtls_ecp_point_read_binary(&grp, &P, peer, 32); + if (rc == 0) rc = mbedtls_ecp_mul(&grp, &R, &d, &P, mbedtls_ctr_drbg_random, &g_drbg); + if (rc == 0) rc = mbedtls_ecp_point_write_binary(&grp, &R, MBEDTLS_ECP_PF_UNCOMPRESSED, + &olen, out, 32); + if (rc == 0 && olen != 32) rc = -1; + + mbedtls_mpi_free(&d); + mbedtls_ecp_point_free(&R); + mbedtls_ecp_point_free(&P); + mbedtls_ecp_group_free(&grp); + return rc; +} + +/* HMAC-SHA256 over two parts, either of which may be empty. The md module is in every TLS build. */ +static int hmac256(const unsigned char *key, size_t key_len, + const unsigned char *a, size_t a_len, + const unsigned char *b, size_t b_len, unsigned char out[32]) +{ + mbedtls_md_context_t ctx; + int rc; + + mbedtls_md_init(&ctx); + rc = mbedtls_md_setup(&ctx, mbedtls_md_info_from_type(MBEDTLS_MD_SHA256), 1); + if (rc == 0) rc = mbedtls_md_hmac_starts(&ctx, key, key_len); + if (rc == 0 && a_len) rc = mbedtls_md_hmac_update(&ctx, a, a_len); + if (rc == 0 && b_len) rc = mbedtls_md_hmac_update(&ctx, b, b_len); + if (rc == 0) rc = mbedtls_md_hmac_finish(&ctx, out); + mbedtls_md_free(&ctx); + return rc; +} + static void wg_clamp(unsigned char k[32]) { k[0] &= 248; @@ -651,11 +727,16 @@ static void conn_close(struct conn *c) mbedtls_net_free(&c->net); } -/* https://host[:port]/path — the only URL shape the coordinator is ever configured with. */ +/* + * scheme://[user:pass@]host[:port]/path. The coordinator is https; majestic on the camera itself is + * plain http on loopback, and may want a password — which then rides in the URL, the one place + * CERBERUS_SNAPSHOT_URL has to put it. `auth` gets "user:pass" base64'd, or stays empty. + */ static int url_split(const char *url, int *tls, char *host, size_t host_sz, - char *port, size_t port_sz, char *path, size_t path_sz) + char *port, size_t port_sz, char *path, size_t path_sz, + char *auth, size_t auth_sz) { - const char *p, *slash; + const char *p, *slash, *at; char *colon; if (!strncmp(url, "https://", 8)) { *tls = 1; p = url + 8; snprintf(port, port_sz, "443"); } @@ -664,6 +745,13 @@ static int url_split(const char *url, int *tls, char *host, size_t host_sz, slash = strchr(p, '/'); snprintf(path, path_sz, "%s", slash ? slash : "/"); + + auth[0] = 0; + at = memchr(p, '@', slash ? (size_t)(slash - p) : strlen(p)); + if (at) { + b64((const unsigned char *)p, (size_t)(at - p), auth, auth_sz); + p = at + 1; + } snprintf(host, host_sz, "%.*s", slash ? (int)(slash - p) : (int)strlen(p), p); colon = strrchr(host, ':'); @@ -674,23 +762,72 @@ static int url_split(const char *url, int *tls, char *host, size_t host_sz, return 0; } +/* Case-insensitive search inside the response head only — never into a JPEG body. */ +static const char *head_find(const char *head, size_t head_len, const char *needle) +{ + size_t n = strlen(needle), i, j; + for (i = 0; i + n <= head_len; i++) { + for (j = 0; j < n; j++) { + char a = head[i + j], b = needle[j]; + if (a >= 'A' && a <= 'Z') a = (char)(a - 'A' + 'a'); + if (a != b) break; + } + if (j == n) return head + i; + } + return NULL; +} + +/* Undoes Transfer-Encoding: chunked in place. Returns the payload length, or -1 if malformed. */ +static long dechunk(unsigned char *buf, size_t len) +{ + size_t in = 0, out = 0; + while (in < len) { + char *end; + unsigned long size = strtoul((const char *)buf + in, &end, 16); + size_t hdr = (size_t)((unsigned char *)end - buf); + while (hdr + 1 < len && !(buf[hdr] == '\r' && buf[hdr + 1] == '\n')) hdr++; /* chunk extensions */ + if (hdr + 1 >= len) return -1; + in = hdr + 2; + if (size == 0) return (long)out; + if (in + size > len) return -1; + memmove(buf + out, buf + in, size); + out += size; + in += size + 2; /* the CRLF after the chunk */ + } + return -1; +} + +struct http_req { + const char *method; + const char *url; + const char *ctype; /* NULL when there is no body */ + const unsigned char *body; + size_t body_len; + const char *headers; /* extra header lines, each ending in \r\n; may be NULL */ +}; + /* - * One request, one connection, `Connection: close` — so the body is simply "everything until EOF" - * and neither Content-Length nor chunked framing has to be parsed. The agent makes at most one HTTP - * call per boot; keep-alive would buy nothing and cost a de-chunker. + * One request, one connection, `Connection: close` — so the body is "everything until EOF" and + * keep-alive bookkeeping is not needed. Binary-safe: the answer may be a JPEG. Chunked answers are + * de-chunked (majestic's are not, but an HTTP server is allowed to). Returns the body length, + * NUL-terminated in `resp`, with the status in *status; -1 on a transport failure, -2 when the + * answer does not fit in resp_sz. */ -static int http_post_json(const char *url, const char *body, char *resp, size_t resp_sz, int *status) +static long http_do(const struct http_req *rq, unsigned char *resp, size_t resp_sz, int *status) { + static int tls_said; struct conn c; - char host[256], port[16], path[512], head[1024]; + char host[256], port[16], path[512], auth[192], head[1536]; unsigned char buf[4096]; - char *hdr_end; - size_t total = 0; - int rc, n; + unsigned char *hdr_end = NULL; + size_t total = 0, i, head_len; + long body_len; + int rc, n, overflow = 0, chunked; memset(&c, 0, sizeof c); - if (url_split(url, &c.tls, host, sizeof host, port, sizeof port, path, sizeof path) != 0) - die("unsupported URL: %s", url); + if (url_split(rq->url, &c.tls, host, sizeof host, port, sizeof port, path, sizeof path, + auth, sizeof auth) != 0) + die("unsupported URL: %s", rq->url); mbedtls_net_init(&c.net); if ((rc = mbedtls_net_connect(&c.net, host, port, MBEDTLS_NET_PROTO_TCP)) != 0) { @@ -711,7 +848,7 @@ static int http_post_json(const char *url, const char *body, char *resp, size_t if (g_cfg.insecure) { /* Dev only. Loud, because an enrollment token in the clear is a device someone else * can enroll. */ - warn_("TLS verification disabled (CERBERUS_INSECURE=1)"); + if (!tls_said) warn_("TLS verification disabled (CERBERUS_INSECURE=1)"); mbedtls_ssl_conf_authmode(&c.conf, MBEDTLS_SSL_VERIFY_NONE); } else { struct stat st; @@ -738,44 +875,81 @@ static int http_post_json(const char *url, const char *body, char *resp, size_t conn_close(&c); return -1; } - note("TLS ok: %s, %s", mbedtls_ssl_get_version(&c.ssl), mbedtls_ssl_get_ciphersuite(&c.ssl)); + /* Once per process: with motion frames going out every second, a line per request would + * be most of the log. */ + if (!tls_said++) + note("TLS ok: %s, %s", mbedtls_ssl_get_version(&c.ssl), mbedtls_ssl_get_ciphersuite(&c.ssl)); } - snprintf(head, sizeof head, - "POST %s HTTP/1.1\r\n" - "Host: %s\r\n" - "User-Agent: cerberus/" CERBERUS_VERSION "\r\n" - "Content-Type: application/json\r\n" - "Content-Length: %u\r\n" - "Connection: close\r\n\r\n", - path, host, (unsigned)strlen(body)); - - if (conn_send(&c, head, strlen(head)) != 0 || conn_send(&c, body, strlen(body)) != 0) { + n = snprintf(head, sizeof head, + "%s %s HTTP/1.1\r\n" + "Host: %s\r\n" + "User-Agent: cerberus/" CERBERUS_VERSION "\r\n" + "%s%s%s" + "%s", + rq->method, path, host, + auth[0] ? "Authorization: Basic " : "", auth, auth[0] ? "\r\n" : "", + rq->headers ? rq->headers : ""); + if (rq->ctype && n > 0 && (size_t)n < sizeof head) + n += snprintf(head + n, sizeof head - (size_t)n, "Content-Type: %s\r\nContent-Length: %u\r\n", + rq->ctype, (unsigned)rq->body_len); + if (n > 0 && (size_t)n < sizeof head) + n += snprintf(head + n, sizeof head - (size_t)n, "Connection: close\r\n\r\n"); + if (n <= 0 || (size_t)n >= sizeof head) + die("request head too long for %s", rq->url); + + if (conn_send(&c, head, (size_t)n) != 0 || + (rq->body_len && conn_send(&c, (const char *)rq->body, rq->body_len) != 0)) { warn_("sending the request failed"); conn_close(&c); return -1; } - while (total + 1 < resp_sz) { + for (;;) { n = conn_recv(&c, buf, sizeof buf); - if (n == MBEDTLS_ERR_SSL_PEER_CLOSE_NOTIFY || n == 0) break; - if (n < 0) break; - if (total + (size_t)n + 1 > resp_sz) n = (int)(resp_sz - total - 1); + if (n == MBEDTLS_ERR_SSL_PEER_CLOSE_NOTIFY || n <= 0) break; + if (total + (size_t)n + 1 > resp_sz) { overflow = 1; break; } memcpy(resp + total, buf, (size_t)n); total += (size_t)n; } - resp[total] = 0; conn_close(&c); + resp[total] = 0; - if (total == 0 || strncmp(resp, "HTTP/1.", 7) != 0) { - warn_("no HTTP response"); + if (total < 12 || strncmp((const char *)resp, "HTTP/1.", 7) != 0) { + warn_("no HTTP response from %s", host); return -1; } - *status = atoi(resp + 9); + *status = atoi((const char *)resp + 9); - hdr_end = strstr(resp, "\r\n\r\n"); - if (hdr_end) memmove(resp, hdr_end + 4, strlen(hdr_end + 4) + 1); - return 0; + for (i = 0; i + 3 < total; i++) + if (resp[i] == '\r' && resp[i + 1] == '\n' && resp[i + 2] == '\r' && resp[i + 3] == '\n') { + hdr_end = resp + i; + break; + } + if (!hdr_end) { resp[0] = 0; return overflow ? -2 : 0; } + + /* Look at the head before the body is moved over it. */ + head_len = (size_t)(hdr_end - resp); + chunked = head_find((const char *)resp, head_len, "transfer-encoding: chunked") != NULL; + body_len = (long)(total - head_len - 4); + memmove(resp, hdr_end + 4, (size_t)body_len); + resp[body_len] = 0; + if (overflow) + return -2; /* what fits is in resp — enough for a JSON caller, never for a frame */ + if (chunked && (body_len = dechunk(resp, (size_t)body_len)) < 0) { + warn_("malformed chunked answer from %s", host); + return -1; + } + resp[body_len] = 0; + return body_len; +} + +/* The coordinator's JSON calls: text in, text out. Returns 0 when an HTTP answer came back — an + * answer longer than resp_sz included, cut to fit, as it always was. */ +static int http_post_json(const char *url, const char *body, char *resp, size_t resp_sz, int *status) +{ + struct http_req rq = { "POST", url, "application/json", (const unsigned char *)body, strlen(body), NULL }; + return http_do(&rq, (unsigned char *)resp, resp_sz, status) == -1 ? -1 : 0; } /* --------------------------------------------------------------- tiny JSON */ @@ -1164,6 +1338,415 @@ static int ping_ok(void) return rc == 0; } +/* ------------------------------------------------------------------ events */ + +/* + * Motion, end to end on the camera's side. + * + * majestic runs a hook script when it sees motion (its own detector — free, always on once enabled + * in its settings). Our hook does nothing but wake us with SIGUSR1: signing, TLS, retries and the + * coordinator's address all live here, and a second HTTP client in a shell script would be a second + * set of bugs. One episode is motion → quiet for CERBERUS_MOTION_QUIET seconds → over; it goes out + * twice, `started` at once and `ended` with its length, however many times the hook fired between. + * + * The answer to `started` says how many frames the cloud wants. Zero unless the owner switched + * object detection on for this camera; then a frame a second, taken by majestic itself + * (CERBERUS_SNAPSHOT_URL), until the cloud says it has seen enough. So the cloud gets the picture + * from the moment of motion, without pulling the stream and without the RTSP password. + * + * Every request is signed: the heartbeat is anonymous, an event is not — it ends up as a message + * with a picture in somebody's Telegram. The key is derived, not stored: + * shared = X25519(device key, coordinator's eventKey from the directive) + * k = HKDF-SHA256(ikm = shared, salt = nodeId, info = "streq-events-v1") + * sig = base64(HMAC-SHA256(k, ts "\n" nodeId "\n" scope "\n" body)) + * and lives exactly as long as the device key does. + */ + +static volatile sig_atomic_t g_motion; +static void on_motion(int sig) { (void)sig; g_motion = 1; } + +static char g_node[128]; +static char g_event_pub[64]; +static unsigned char g_event_k[32]; +static int g_event_ready; + +/* Flat JSON number field; `fallback` when absent. Same contract as json_str. */ +static long json_int(const char *json, const char *key, long fallback) +{ + char pat[64]; + const char *p; + + snprintf(pat, sizeof pat, "\"%s\"", key); + p = strstr(json, pat); + if (!p || !(p = strchr(p + strlen(pat), ':'))) return fallback; + while (*++p == ' ' || *p == '\t' || *p == '\n' || *p == '\r') { } + return (*p == '-' || (*p >= '0' && *p <= '9')) ? strtol(p, NULL, 10) : fallback; +} + +static int json_true(const char *json, const char *key) +{ + char pat[64]; + const char *p; + + snprintf(pat, sizeof pat, "\"%s\"", key); + p = strstr(json, pat); + if (!p || !(p = strchr(p + strlen(pat), ':'))) return 0; + while (*++p == ' ' || *p == '\t' || *p == '\n' || *p == '\r') { } + return !strncmp(p, "true", 4); +} + +/* The directive hands the coordinator's eventKey on every beat; the key is derived once per value. */ +static void event_key_update(const char *coord_pub_b64) +{ + static const char info[] = "streq-events-v1"; + static const unsigned char one = 1; + unsigned char coord[48], priv[48], shared[32], prk[32]; + char priv_b64[64], pub_b64[64]; + size_t clen = 0, plen = 0; + + if (!g_node[0] || (g_event_ready && !strcmp(coord_pub_b64, g_event_pub))) + return; + if (b64_decode(coord_pub_b64, coord, sizeof coord, &clen) != 0 || clen != 32) { + warn_("the coordinator's eventKey is not a 32-byte key — motion events stay off"); + return; + } + + device_key(priv_b64, pub_b64); + if (b64_decode(priv_b64, priv, sizeof priv, &plen) != 0 || plen != 32 || + x25519_shared(priv, coord, shared) != 0 || + /* HKDF (RFC 5869) with one output block: PRK = HMAC(salt, IKM), OKM = HMAC(PRK, info | 0x01). */ + hmac256((const unsigned char *)g_node, strlen(g_node), shared, sizeof shared, NULL, 0, prk) != 0 || + hmac256(prk, sizeof prk, (const unsigned char *)info, strlen(info), &one, 1, g_event_k) != 0) { + warn_("could not derive the event signing key — motion events stay off"); + g_event_ready = 0; + } else { + snprintf(g_event_pub, sizeof g_event_pub, "%s", coord_pub_b64); + if (!g_event_ready) note("motion events: the coordinator takes them from this camera"); + g_event_ready = 1; + } + memset(priv, 0, sizeof priv); + memset(shared, 0, sizeof shared); + memset(prk, 0, sizeof prk); + memset(priv_b64, 0, sizeof priv_b64); +} + +/* POST , signed for `scope`. Returns the HTTP status, or -1 when there was no answer. */ +static int event_post(const char *path, const char *scope, const char *ctype, + const unsigned char *body, size_t len, char *resp, size_t resp_sz) +{ + char url[512], ts[24], signed_head[256], sig[64], hdr[256]; + unsigned char mac[32]; + struct http_req rq; + int n, status = 0; + + snprintf(ts, sizeof ts, "%ld", (long)time(NULL)); + n = snprintf(signed_head, sizeof signed_head, "%s\n%s\n%s\n", ts, g_node, scope); + if (n <= 0 || (size_t)n >= sizeof signed_head || + hmac256(g_event_k, sizeof g_event_k, (const unsigned char *)signed_head, (size_t)n, body, len, mac) != 0) + return -1; + b64(mac, sizeof mac, sig, sizeof sig); + snprintf(hdr, sizeof hdr, "X-Streq-Ts: %s\r\nX-Streq-Sig: %s\r\n", ts, sig); + snprintf(url, sizeof url, "%s%s", g_cfg.api, path); + + rq.method = "POST"; + rq.url = url; + rq.ctype = ctype; + rq.body = body; + rq.body_len = len; + rq.headers = hdr; + return http_do(&rq, (unsigned char *)resp, resp_sz, &status) == -1 ? -1 : status; +} + +static unsigned char *g_frame; + +/* One JPEG from majestic into g_frame. Returns its length, or -1. */ +static long snapshot_take(void) +{ + static int complaints; + struct http_req rq = { "GET", NULL, NULL, NULL, 0, NULL }; + int status = 0; + long n; + + if (!g_frame && !(g_frame = malloc((size_t)g_cfg.snapshot_max + 1))) + return -1; + rq.url = g_cfg.snapshot_url; + n = http_do(&rq, g_frame, (size_t)g_cfg.snapshot_max + 1, &status); + if (n == -1) + return -1; + if (status == 401 || status == 403) { + if (complaints++ < 3) + warn_("majestic wants a password for %s — put it in the address: " + "fw_setenv cerberus_snapshot_url 'http://user:pass@127.0.0.1/image.jpg?width=640&height=360'", + g_cfg.snapshot_url); + return -1; + } + if (n == -2) { + if (complaints++ < 3) + warn_("the snapshot is larger than %d bytes — ask majestic for a smaller one (width=640)", g_cfg.snapshot_max); + return -1; + } + if (status != 200 || n < 4 || g_frame[0] != 0xFF || g_frame[1] != 0xD8) { + if (complaints++ < 3) + warn_("the snapshot is not a JPEG (HTTP %d, %ld bytes) — check CERBERUS_SNAPSHOT_URL", status, n); + return -1; + } + return n; +} + +static struct { + int active; + char id[32]; + time_t started, last_motion, next_frame; + int frames_left, interval, seq, misses; +} g_ep; + +static void episode_post(const char *phase, long duration, char *resp, size_t resp_sz, int *status) +{ + char path[256], body[192]; + + if (duration >= 0) + snprintf(body, sizeof body, "{\"eventId\":\"%s\",\"kind\":\"motion\",\"phase\":\"%s\",\"durationSec\":%ld}", + g_ep.id, phase, duration); + else + snprintf(body, sizeof body, "{\"eventId\":\"%s\",\"kind\":\"motion\",\"phase\":\"%s\"}", g_ep.id, phase); + snprintf(path, sizeof path, "/nodes/%s/events", g_node); + *status = event_post(path, "event", "application/json", (const unsigned char *)body, strlen(body), + resp, resp_sz); +} + +static void episode_start(time_t now) +{ + static int said_clock, said_refused; + unsigned char r[12]; + char resp[1024]; + int status, i; + + /* Signatures carry the time; a camera still at 1970 would be refused on every episode. */ + if (now < 1600000000) { + if (!said_clock++) warn_("motion: the clock is not set (no NTP yet?) — events wait for it"); + return; + } + + memset(&g_ep, 0, sizeof g_ep); + if (mbedtls_ctr_drbg_random(&g_drbg, r, sizeof r) != 0) return; + g_ep.id[0] = 'm'; + for (i = 0; i < 12; i++) + sprintf(g_ep.id + 1 + i * 2, "%02x", r[i]); + g_ep.active = 1; + g_ep.started = g_ep.last_motion = g_ep.next_frame = now; + g_ep.interval = 1; + + episode_post("started", -1, resp, sizeof resp, &status); + if (status == 200) { + g_ep.frames_left = (int)json_int(resp, "frames", 0); + g_ep.interval = (int)json_int(resp, "intervalSeconds", 1); + if (g_ep.interval < 1) g_ep.interval = 1; + note("motion: episode %s%s", g_ep.id, + g_ep.frames_left > 0 ? ", the cloud looks for people and cars in it" : ""); + } else if (status == 401) { + if (!said_refused++) + warn_("motion: the coordinator refused the signature — check the camera's clock"); + } else if (status == 429) { + warn_("motion: too many episodes this hour — narrow majestic's motion zone or sensitivity"); + } else { + warn_("motion: the event was not taken (HTTP %d)", status); + } +} + +static void episode_frame(time_t now) +{ + char path[256], scope[96], resp[1024]; + long n; + int status; + + g_ep.next_frame = now + g_ep.interval; + if ((n = snapshot_take()) < 0) { + if (++g_ep.misses >= 3) g_ep.frames_left = 0; + return; + } + + snprintf(path, sizeof path, "/nodes/%s/events/%s/frames/%d", g_node, g_ep.id, g_ep.seq); + snprintf(scope, sizeof scope, "frame/%s/%d", g_ep.id, g_ep.seq); + status = event_post(path, scope, "image/jpeg", g_frame, (size_t)n, resp, sizeof resp); + g_ep.seq++; + g_ep.frames_left--; + if (status != 200) { + warn_("motion: frame %d was not taken (HTTP %d)", g_ep.seq - 1, status); + g_ep.frames_left = 0; + } else if (!json_true(resp, "more")) { + g_ep.frames_left = 0; + if (strstr(resp, "\"labels\":[\"")) + note("motion: the cloud has seen enough: %s", resp); + } +} + +static void episode_end(time_t now) +{ + char resp[512]; + int status; + long duration = (long)(g_ep.last_motion - g_ep.started); + + episode_post("ended", duration, resp, sizeof resp, &status); + if (status != 200) + warn_("motion: the end of episode %s was not taken (HTTP %d)", g_ep.id, status); + note("motion: episode %s over after %lds, %d frame(s) sent", g_ep.id, duration, g_ep.seq); + g_ep.active = 0; + (void)now; +} + +/* One step of the motion machinery; called every loop turn, which is every second in an episode. */ +static void events_tick(time_t now) +{ + static int said_no_key; + + if (g_motion) { + g_motion = 0; + if (g_ep.active) + g_ep.last_motion = now; + else if (g_event_ready) + episode_start(now); + else if (!said_no_key++) + note("motion seen, but the coordinator has not handed this camera an event key (older cloud?)"); + } + if (!g_ep.active) + return; + if (g_ep.frames_left > 0 && now >= g_ep.next_frame) + episode_frame(now); + if (now - g_ep.last_motion >= g_cfg.motion_quiet || now - g_ep.started >= g_cfg.motion_max) + episode_end(time(NULL)); +} + +/* + * The hook majestic runs. Ours only wakes the agent — and checks that the pid in the pidfile is + * still us before signalling it, because a pid outlives the process it named, and SIGUSR1 kills + * whatever does not expect it. Someone else's hook is not overwritten: it is kept next to ours as + * .orig and still called, with the same arguments. + */ +#define HOOK_MARK "CERBERUS_HOOK=1" + +static int file_has(const char *path, const char *needle) +{ + char buf[2048]; + size_t n; + FILE *f = fopen(path, "r"); + if (!f) return 0; + n = fread(buf, 1, sizeof buf - 1, f); + fclose(f); + buf[n] = 0; + return strstr(buf, needle) != NULL; +} + +static void motion_hook_install(void) +{ + char orig[320], tmp[320], body[1024]; + struct stat st; + FILE *f; + + snprintf(orig, sizeof orig, "%s.orig", g_cfg.motion_hook); + snprintf(tmp, sizeof tmp, "%s.cerberus-new", g_cfg.motion_hook); + snprintf(body, sizeof body, + "#!/bin/sh\n" + HOOK_MARK "\n" + "[ -x %s ] && %s \"$@\" &\n" + "p=$(cat %s 2>/dev/null)\n" + "[ -n \"$p\" ] && grep -q cerberus /proc/\"$p\"/cmdline 2>/dev/null && kill -USR1 \"$p\"\n" + "exit 0\n", + orig, orig, g_cfg.pidfile); + + if (stat(g_cfg.motion_hook, &st) == 0) { + if (file_has(g_cfg.motion_hook, body)) + return; /* ours, current */ + if (!file_has(g_cfg.motion_hook, HOOK_MARK)) { + if (stat(orig, &st) == 0) { + warn_("%s is not ours and %s exists already — leaving both alone; motion events are off", + g_cfg.motion_hook, orig); + return; + } + if (rename(g_cfg.motion_hook, orig) != 0) { + warn_("cannot keep %s aside: %s — motion events are off", g_cfg.motion_hook, strerror(errno)); + return; + } + note("kept the existing %s as %s; ours calls it", g_cfg.motion_hook, orig); + } + } + + mkparent(g_cfg.motion_hook); + if (!(f = fopen(tmp, "w"))) { + warn_("cannot write %s: %s — motion events are off", tmp, strerror(errno)); + return; + } + fputs(body, f); + fclose(f); + chmod(tmp, 0755); + if (rename(tmp, g_cfg.motion_hook) != 0) { + warn_("cannot install %s: %s — motion events are off", g_cfg.motion_hook, strerror(errno)); + unlink(tmp); + return; + } + note("motion hook installed at %s (majestic runs it when motion detection is on)", g_cfg.motion_hook); +} + +/* `clean`: our hook goes, and whatever it replaced comes back. */ +static void motion_hook_remove(void) +{ + char orig[320]; + struct stat st; + + if (!file_has(g_cfg.motion_hook, HOOK_MARK)) + return; + snprintf(orig, sizeof orig, "%s.orig", g_cfg.motion_hook); + if (stat(orig, &st) == 0 ? rename(orig, g_cfg.motion_hook) : unlink(g_cfg.motion_hook)) + warn_("cannot remove the motion hook %s: %s", g_cfg.motion_hook, strerror(errno)); +} + +static void pidfile_write(void) +{ + FILE *f; + mkparent(g_cfg.pidfile); + if ((f = fopen(g_cfg.pidfile, "w")) != NULL) { + fprintf(f, "%ld\n", (long)getpid()); + fclose(f); + } +} + +/* `cerberus motion`: what the hook does, by hand — the way to test without walking past the camera. */ +static int cmd_motion(void) +{ + char buf[32]; + long pid = 0; + FILE *f = fopen(g_cfg.pidfile, "r"); + + if (f) { + if (fgets(buf, sizeof buf, f)) pid = atol(buf); + fclose(f); + } + if (pid <= 0 || kill((pid_t)pid, SIGUSR1) != 0) { + fprintf(stderr, "cerberus: no running agent in %s\n", g_cfg.pidfile); + return 1; + } + printf("motion signalled to %ld\n", pid); + return 0; +} + +/* `cerberus snapshot`: is majestic's frame reachable as configured? Saves it for a look. */ +static int cmd_snapshot(void) +{ + const char *out = "/tmp/cerberus-snapshot.jpg"; + long n = snapshot_take(); + FILE *f; + + if (n < 0) { + fprintf(stderr, "cerberus: no snapshot from %s\n", g_cfg.snapshot_url); + return 1; + } + if ((f = fopen(out, "wb")) != NULL) { + fwrite(g_frame, 1, (size_t)n, f); + fclose(f); + } + printf("snapshot: %ld bytes -> %s\n", n, out); + return 0; +} + /* ------------------------------------------------------------- directives */ static char **g_argv; @@ -1240,6 +1823,9 @@ static void apply_directive(const char *json) if (rotated && link_apply(1) != 0) warn_("could not apply the rotated join material; the stale-handshake path will retry"); + if (g_cfg.events && json_str(json, "eventKey", fresh, sizeof fresh) && fresh[0]) + event_key_update(fresh); + if (!json_str(json, "command", command, sizeof command)) return; if (!strcmp(command, "Reboot")) @@ -1264,6 +1850,8 @@ static void catch_signals(void) { signal(SIGINT, on_signal); signal(SIGTERM, on_signal); + /* The motion hook's wake-up. SIGUSR1 kills by default, so this is set before anything slow. */ + signal(SIGUSR1, on_motion); } /* @@ -1293,6 +1881,12 @@ static int watch(void) return 1; store_get("cerberus_node_id", node, sizeof node); snprintf(beat_url, sizeof beat_url, "%s/nodes/%s/heartbeat", g_cfg.api, node); + snprintf(g_node, sizeof g_node, "%s", node); + + if (g_cfg.events) { + pidfile_write(); + motion_hook_install(); + } tick = g_cfg.watch_interval; if (g_cfg.heartbeat > 0 && g_cfg.heartbeat < tick) tick = g_cfg.heartbeat; @@ -1371,8 +1965,15 @@ static int watch(void) next_beat = time(NULL) + g_cfg.heartbeat; } - if (!g_stop) sleep((unsigned)tick); + if (g_cfg.events) + events_tick(time(NULL)); + + /* A second at a time while an episode runs: frames go out every second, and the hook's + * signal cuts any sleep short anyway. */ + if (!g_stop && !g_motion) sleep(g_ep.active ? 1u : (unsigned)tick); } + if (g_ep.active) + episode_end(time(NULL)); note("stopping"); return 0; } @@ -1469,6 +2070,7 @@ static int cmd_clean(void) "node_id", "overlay_ip", "hub_endpoint", "hub_pubkey", "allowed_ips", "network_subnet", "hub_ip", "claim_code", "device_key", "api", "enrollment_token", "hostname", "kind", "iface", "subnet", "ca", "heartbeat", "ping_target", "ping_interval", "ping_timeout", "ping_fails", + "events", "snapshot_url", NULL }; char self[256]; @@ -1485,6 +2087,7 @@ static int cmd_clean(void) self[n] = 0; runf("ip link del dev %s", g_cfg.iface); + motion_hook_remove(); if (unlink(INIT_SCRIPT) != 0 && errno != ENOENT) warn_("cannot remove %s: %s", INIT_SCRIPT, strerror(errno)); @@ -1511,13 +2114,15 @@ static void usage(void) fputs( "cerberus " CERBERUS_VERSION " — on-camera cloud agent for OpenIPC\n" "\n" - "usage: cerberus [enroll|up|run|status|keygen|clean]\n" + "usage: cerberus [enroll|up|run|status|keygen|clean|motion|snapshot]\n" " enroll enroll into the coordinator and persist the join material (idempotent)\n" " up program the WireGuard interface from the stored material\n" " run enroll + up + watch the tunnel (the default; what the init script calls)\n" " status print what is stored, how old the handshake is, whether the tunnel answers\n" " keygen print the device keypair (private, public), creating it if needed\n" " clean remove the agent: interface, init script, stored material, the binary\n" + " motion tell the running agent there is motion, as majestic's hook does\n" + " snapshot fetch one frame from CERBERUS_SNAPSHOT_URL into /tmp/cerberus-snapshot.jpg\n" "\n" "config (env; the old STREQ_* names are accepted too):\n" " CERBERUS_API coordinator base URL incl. /v1 [https://api.localhost/v1]\n" @@ -1534,6 +2139,12 @@ static void usage(void) " CERBERUS_PING_INTERVAL seconds between probes; 0 disables [60]\n" " CERBERUS_PING_TIMEOUT seconds one probe waits [3]\n" " CERBERUS_PING_FAILS failed probes before the interface is rebuilt [3]\n" + " CERBERUS_EVENTS=0 no motion hook, no events [1]\n" + " CERBERUS_SNAPSHOT_URL majestic's JPEG, user:pass@ allowed\n" + " [http://127.0.0.1/image.jpg?width=640&height=360]\n" + " CERBERUS_MOTION_HOOK the script majestic runs on motion [/usr/sbin/motion.sh]\n" + " CERBERUS_MOTION_QUIET seconds without motion that end an episode [10]\n" + " CERBERUS_PIDFILE where the hook finds the agent [/var/run/cerberus.pid]\n" " CERBERUS_DRY_RUN=1 print ip/wg commands instead of running them\n", stderr); } @@ -1559,6 +2170,8 @@ int main(int argc, char **argv) if (!strcmp(cmd, "clean")) return cmd_clean(); if (!strcmp(cmd, "enroll")) return enroll(); if (!strcmp(cmd, "up")) return link_up(); + if (!strcmp(cmd, "motion")) return cmd_motion(); + if (!strcmp(cmd, "snapshot")) return cmd_snapshot(); if (!strcmp(cmd, "run")) { catch_signals(); /* Ненулевой ответ здесь бывает ровно один: попросили остановиться. Это не отказ. */ diff --git a/tests/fake_coordinator.py b/tests/fake_coordinator.py index 6851532..46a74a4 100755 --- a/tests/fake_coordinator.py +++ b/tests/fake_coordinator.py @@ -11,17 +11,29 @@ POST /_test/join {"hubEndpoint": "..."} rewrites fields of the join material POST /_test/forget {} the node is revoked: heartbeats answer 404 POST /_test/fail {"status": 503} heartbeats answer this until set to 0 + POST /_test/frames {"frames": 2} how many frames an episode is asked for GET /_test/requests every request so far, as a JSON list +Events are checked the way the coordinator checks them: the heartbeat hands out an eventKey, and +every event and frame must carry an HMAC under the key both sides derive from it and the node's +device key (X25519 + HKDF). Each recorded request says whether its signature held (`verified`). + +With --majestic-port it also plays majestic on plain HTTP: /image.jpg behind Basic auth, answered +chunked, because an HTTP server is allowed to and the agent has to cope. + Stdlib only: the test containers have python3 and nothing more. """ import argparse +import base64 +import hashlib +import hmac import json import re import secrets import ssl import threading +import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer LOCK = threading.Lock() @@ -33,9 +45,63 @@ "join": {}, "command": None, "fail_status": 0, + "frames": 0, "requests": [], + "event_priv": secrets.token_bytes(32), } +# RFC 7748 X25519, because the stdlib has none and the test containers have nothing else. +P = 2 ** 255 - 19 + + +def x25519(k, u): + k = bytearray(k) + k[0] &= 248 + k[31] &= 127 + k[31] |= 64 + k = int.from_bytes(k, "little") + u = bytearray(u) + u[31] &= 127 + x1 = int.from_bytes(u, "little") + x2, z2, x3, z3, swap = 1, 0, x1, 1, 0 + for t in reversed(range(255)): + bit = (k >> t) & 1 + swap ^= bit + if swap: + x2, x3, z2, z3 = x3, x2, z3, z2 + swap = bit + a, b = (x2 + z2) % P, (x2 - z2) % P + aa, bb = a * a % P, b * b % P + e = (aa - bb) % P + c, d = (x3 + z3) % P, (x3 - z3) % P + da, cb = d * a % P, c * b % P + x3, z3 = (da + cb) ** 2 % P, x1 * (da - cb) ** 2 % P + x2, z2 = aa * bb % P, e * (aa + 121665 * e) % P + if swap: + x2, z2 = x3, z3 + return (x2 * pow(z2, P - 2, P) % P).to_bytes(32, "little") + + +def event_pub(): + return base64.b64encode(x25519(STATE["event_priv"], (9).to_bytes(32, "little"))).decode() + + +def verified(node, scope, headers, raw): + """The coordinator's EventSigning, recomputed: X25519 → HKDF(salt=nodeId) → HMAC.""" + try: + pub = base64.b64decode(STATE["nodes"][node]["pub"]) + ts = headers.get("X-Streq-Ts") or "" + sig = base64.b64decode(headers.get("X-Streq-Sig") or "") + except (KeyError, ValueError): + return False + if not ts.isdigit() or abs(time.time() - int(ts)) > 120: + return False + shared = x25519(STATE["event_priv"], pub) + prk = hmac.new(node.encode(), shared, hashlib.sha256).digest() + key = hmac.new(prk, b"streq-events-v1\x01", hashlib.sha256).digest() + want = hmac.new(key, f"{ts}\n{node}\n{scope}\n".encode() + raw, hashlib.sha256).digest() + return hmac.compare_digest(want, sig) + def join_material(): return dict(STATE["join"], kind="Wireguard") @@ -73,12 +139,32 @@ def do_GET(self): def do_POST(self): raw, body = self._body() with LOCK: - if not self.path.startswith("/_test/"): + m = re.fullmatch(r"/v1/nodes/([^/]+)/events/([^/]+)/frames/(\d+)", self.path) + if m: + ok = verified(m.group(1), f"frame/{m.group(2)}/{m.group(3)}", self.headers, raw) STATE["requests"].append({ + "path": self.path, "verified": ok, "bytes": len(raw), + "jpeg": raw[:2] == b"\xff\xd8", + "contentType": self.headers.get("Content-Type"), + }) + seq = int(m.group(3)) + answer = (200, {"more": seq + 1 < STATE["frames"], "labels": []}) if ok else (401, {"code": "bad_signature"}) + return self._send(*answer) + + entry = None + if not self.path.startswith("/_test/"): + entry = { "path": self.path, "userAgent": self.headers.get("User-Agent"), "body": body if body is not None else raw.decode(errors="replace"), - }) + } + STATE["requests"].append(entry) + m = re.fullmatch(r"/v1/nodes/([^/]+)/events", self.path) + if m and entry is not None: + entry["verified"] = verified(m.group(1), "event", self.headers, raw) + if not entry["verified"]: + return self._send(401, {"code": "bad_signature"}) + return self._send(200, {"id": "nev_test", "frames": STATE["frames"], "intervalSeconds": 1}) status, answer = self.route(body) self._send(status, answer) @@ -86,6 +172,10 @@ def route(self, body): if body is None: return 400, {"error": "bad_json"} + if self.path == "/_test/frames": + STATE["frames"] = int(body.get("frames", 0)) + return 200, {} + if self.path == "/_test/command": STATE["command"] = body.get("command") return 202, {} @@ -136,7 +226,42 @@ def heartbeat(self, node): return 404, {"error": "unknown_node"} command = STATE["command"] or "None" STATE["command"] = None - return 200, {"command": command, "join": join_material(), "mapVersion": 1} + return 200, {"command": command, "join": join_material(), "mapVersion": 1, "eventKey": event_pub()} + + +class Majestic(BaseHTTPRequestHandler): + """majestic's /image.jpg: Basic auth root:12345, a JPEG-looking body, sent chunked.""" + protocol_version = "HTTP/1.1" + FRAME = b"\xff\xd8\xff\xe0" + bytes(range(256)) * 40 + b"\xff\xd9" + + def log_message(self, fmt, *args): + pass + + def do_GET(self): + want = "Basic " + base64.b64encode(b"root:12345").decode() + with LOCK: + STATE["requests"].append({"path": "majestic:" + self.path, + "authorized": self.headers.get("Authorization") == want}) + if not self.path.startswith("/image.jpg"): + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + return + if self.headers.get("Authorization") != want: + self.send_response(401) + self.send_header("WWW-Authenticate", 'Basic realm="majestic"') + self.send_header("Content-Length", "0") + self.end_headers() + return + self.send_response(200) + self.send_header("Content-Type", "image/jpeg") + self.send_header("Transfer-Encoding", "chunked") + self.send_header("Connection", "close") + self.end_headers() + for i in range(0, len(self.FRAME), 3000): + part = self.FRAME[i:i + 3000] + self.wfile.write(b"%x\r\n" % len(part) + part + b"\r\n") + self.wfile.write(b"0\r\n\r\n") def main(): @@ -151,8 +276,13 @@ def main(): ap.add_argument("--hub-overlay-ip", required=True) ap.add_argument("--hub-endpoint", required=True) ap.add_argument("--hub-public-key", required=True) + ap.add_argument("--majestic-port", type=int, default=0) a = ap.parse_args() + if a.majestic_port: + cam = ThreadingHTTPServer(("127.0.0.1", a.majestic_port), Majestic) + threading.Thread(target=cam.serve_forever, daemon=True).start() + STATE["token"] = a.token STATE["token_max_uses"] = a.token_max_uses STATE["join"] = { diff --git a/tests/suite.sh b/tests/suite.sh index 08bdaf7..f6bd8c6 100755 --- a/tests/suite.sh +++ b/tests/suite.sh @@ -120,7 +120,7 @@ TOKEN=tok_cerberus_test python3 /src/tests/fake_coordinator.py --port 8443 --cert srv.crt --key srv.key \ --token "$TOKEN" --token-max-uses 2 --overlay-ip $OVERLAY --network-subnet $SUBNET --hub-overlay-ip $HUB_IP \ - --hub-endpoint $HUB_EP --hub-public-key "$HUB_PUB" >coordinator.log 2>&1 & + --hub-endpoint $HUB_EP --hub-public-key "$HUB_PUB" --majestic-port 8081 >coordinator.log 2>&1 & COORD=$! trap 'kill $COORD 2>/dev/null' EXIT @@ -456,6 +456,54 @@ env PATH="$T/fw:$PATH" FAKE_UBOOT=$INST sh /src/uninstall.sh >uninst.log 2>&1 check "uninstall.sh removes it" "$(yes_if test -e /usr/sbin/cerberus)$(yes_if test -e /etc/init.d/S99cerberus)" "nono" check "and every key it wrote" "$(grep -c -E '^(cerberus|streq)_' $INST)" "0" +# ------------------------------------------------------------------ motion +section "motion events" +HOOK=$T/hooks/motion.sh +SNAP=http://root:12345@127.0.0.1:8081/image.jpg +mkdir -p $T/hooks +printf '#!/bin/sh\necho "theirs $*" >>%s/theirs.log\n' $T >$HOOK && chmod +x $HOOK +M="CERBERUS_MOTION_HOOK=$HOOK CERBERUS_PIDFILE=$T/cerberus.pid CERBERUS_SNAPSHOT_URL=$SNAP CERBERUS_MOTION_QUIET=2" + +check "snapshot: majestic's frame, with its password, de-chunked" \ + "$(env $M $A snapshot 2>/dev/null | sed -n 's/^snapshot: \([0-9]*\) bytes.*/\1/p')" "10246" +check "without the password: refused, and says how to fix it" \ + "$(CERBERUS_SNAPSHOT_URL=http://127.0.0.1:8081/image.jpg $A snapshot 2>&1 | grep -c 'cerberus_snapshot_url')" "1" + +cp enrolled.env agent.env +ctl frames '{"frames":2}' +( sleep 3; sh $HOOK 1 2 3; sleep 1; sh $HOOK; ) & +runfor 12 motion.log $M +check "someone else's hook is kept aside" "$(yes_if test -x $HOOK.orig)" "yes" +check "and still called, with its arguments" "$(yes_if grep -q 'theirs 1 2 3' theirs.log)" "yes" +check "the coordinator's eventKey is taken" "$(yes_if grep -q 'motion events: the coordinator takes them' motion.log)" "yes" +check "one episode for several hook calls" \ + "$(q "print(sum(1 for r in R if r['path'].endswith('/events') and r['body'].get('phase') == 'started'))")" "1" +check "signed by the device key" \ + "$(q "print(all(r['verified'] for r in R if r['path'].endswith('/events')))")" "True" +check "frames follow, as many as asked" \ + "$(q "print(sum(1 for r in R if '/frames/' in r['path']))")" "2" +check "each a signed JPEG" \ + "$(q "print(all(r['verified'] and r['jpeg'] and r['contentType'] == 'image/jpeg' for r in R if '/frames/' in r['path']))")" "True" +check "and the episode ends on its own" \ + "$(q "print(any(r['path'].endswith('/events') and r['body'].get('phase') == 'ended' for r in R))")" "True" + +ctl frames '{"frames":0}' +B0=$(q "print(sum(1 for r in R if '/frames/' in r['path']))") +( sleep 3; env CERBERUS_PIDFILE=$T/cerberus.pid $A motion >/dev/null; ) & +runfor 8 motion2.log $M +check "cerberus motion: the hook's signal, by hand" \ + "$(q "print(sum(1 for r in R if r['path'].endswith('/events') and r['body'].get('phase') == 'started'))")" "2" +check "no frames when the cloud asks for none" "$(q "print(sum(1 for r in R if '/frames/' in r['path']))")" "$B0" +# A pid outlives its process: whatever holds it now must not get a SIGUSR1, which kills by default. +sleep 30 & +STRANGER=$! +sed "s#$T/cerberus.pid#$T/stale.pid#" $HOOK >$T/hooks/stale.sh +echo $STRANGER >$T/stale.pid +sh $T/hooks/stale.sh +sleep 1 +check "a stale pid is not signalled" "$(yes_if kill -0 $STRANGER)" "yes" +kill $STRANGER 2>/dev/null + # ------------------------------------------------------------------ revocation section "revocation and removal" cp enrolled.env agent.env @@ -478,7 +526,8 @@ check "with the device key kept" "$(ub $OLD cerberus_device_key)" "$PRIV" mkdir -p /etc/init.d $T/victim cp $A $T/victim/cerberus && touch /etc/init.d/S99cerberus cp enrolled.env clean.env -CERBERUS_STORE=file:$T/clean.env $T/victim/cerberus clean >clean.log 2>&1 +CERBERUS_STORE=file:$T/clean.env CERBERUS_MOTION_HOOK=$HOOK $T/victim/cerberus clean >clean.log 2>&1 +check "clean: their motion hook is back" "$(yes_if grep -q theirs $HOOK)$(yes_if test -e $HOOK.orig)" "yesno" check "clean: removes the init script" "$(yes_if test -e /etc/init.d/S99cerberus)" "no" check "the stored material" "$(yes_if test -e $T/clean.env)" "no" sleep 4 diff --git a/uninstall.sh b/uninstall.sh index 866c08b..de05a4b 100755 --- a/uninstall.sh +++ b/uninstall.sh @@ -20,7 +20,7 @@ ip link del dev "${iface:-wg0}" >/dev/null 2>&1 || true for name in node_id overlay_ip hub_endpoint hub_pubkey allowed_ips network_subnet hub_ip \ claim_code device_key api enrollment_token hostname kind iface subnet ca heartbeat \ - ping_target ping_interval ping_timeout ping_fails; do + ping_target ping_interval ping_timeout ping_fails events snapshot_url; do for prefix in cerberus streq; do if fw_printenv -n "${prefix}_$name" >/dev/null 2>&1; then fw_setenv "${prefix}_$name" >/dev/null 2>&1 || die 30 @@ -28,6 +28,15 @@ for name in node_id overlay_ip hub_endpoint hub_pubkey allowed_ips network_subne done done +hook=/usr/sbin/motion.sh +if grep -q CERBERUS_HOOK=1 "$hook" 2>/dev/null; then + if [ -e "$hook.orig" ]; then + mv -f "$hook.orig" "$hook" || die 30 + else + rm -f "$hook" || die 30 + fi +fi + rm -f "$INIT" || die 30 rm -f "$AGENT" || die 30 rm -rf /etc/cerberus /etc/streq || die 30