From 0bd9dcd5846f5f51d1fc0a862a1235c0f85b099e Mon Sep 17 00:00:00 2001 From: retoor Date: Sat, 24 Oct 2026 00:35:10 +0200 Subject: [PATCH] fix: stats flush ensures schema, query flushes pending, add sqlite round-trip test --- reverse.md | 235 +++++++++++++++ src/app/views/view.h | 45 ++- src/cli/cli.mk | 4 +- src/cli/cli/cli_app.cpp | 2 + src/cli/cli/expose_cli.cpp | 149 ++++++++++ src/cli/cli/expose_cli.hpp | 22 ++ src/realtime/expose/claim.cpp | 44 +++ src/realtime/expose/claim.h | 19 ++ src/realtime/expose/hello.cpp | 39 +++ src/realtime/expose/hello.h | 13 + src/realtime/expose/proxy.cpp | 31 ++ src/realtime/expose/proxy.h | 54 ++++ src/realtime/expose/stats.cpp | 334 ++++++++++++++++++++++ src/realtime/expose/stats.h | 66 +++++ src/realtime/expose/tunnel.cpp | 125 ++++++++ src/realtime/expose/tunnel.h | 81 ++++++ src/realtime/expose/upstream.cpp | 69 +++++ src/realtime/expose/upstream.h | 38 +++ src/realtime/realtime.mk | 15 +- tests/src/realtime/expose/expose_test.cpp | 98 +++++++ 20 files changed, 1468 insertions(+), 15 deletions(-) create mode 100644 reverse.md create mode 100644 src/cli/cli/expose_cli.cpp create mode 100644 src/cli/cli/expose_cli.hpp create mode 100644 src/realtime/expose/claim.cpp create mode 100644 src/realtime/expose/claim.h create mode 100644 src/realtime/expose/hello.cpp create mode 100644 src/realtime/expose/hello.h create mode 100644 src/realtime/expose/proxy.cpp create mode 100644 src/realtime/expose/proxy.h create mode 100644 src/realtime/expose/stats.cpp create mode 100644 src/realtime/expose/stats.h create mode 100644 src/realtime/expose/tunnel.cpp create mode 100644 src/realtime/expose/tunnel.h create mode 100644 src/realtime/expose/upstream.cpp create mode 100644 src/realtime/expose/upstream.h create mode 100644 tests/src/realtime/expose/expose_test.cpp diff --git a/reverse.md b/reverse.md new file mode 100644 index 0000000..5cef7cd --- /dev/null +++ b/reverse.md @@ -0,0 +1,235 @@ +# Reverse proxy / expose specification (reverse tunnels) + +Part: reverse tunnel server + `molo expose` client. Owned: `src/realtime/` (tunnel protocol) + `src/cli/` (expose CLI). Plan: `plans/06-realtime.md` § tunnel. Version: 0.2.0. + +## 1. Purpose + +Any application (in a container or elsewhere) can publish one or more of its +local endpoints through the MoloApp server so the public internet reaches +them via claimed public routes. Transport between the local apps +("exposer") and the server is always one outbound connection initiated by +the exposer; inbound TCP to the exposer is never required. The server then +reverse-proxies public HTTP traffic arriving at the claimed routes into +that tunnel. One tunnel serves many concurrent upstreams: each public +request is routed inside the tunnel to the matching upstream entry. + +This system is HTTP/2 (RFC 9113, `src/runtime/http/v2/`). The tunnel rides +the same H2 stack: one H2 connection carries many concurrent streams, one +stream per proxied request. There is no HTTP/1-only or single-flight +assumption anywhere in this spec. + +## 2. Roles and terms + +- **Server**: the MoloApp instance accepting tunnels and public traffic. +- **Exposer / agent**: the `molo expose` process running near the local apps. +- **Upstream**: one local origin the agent forwards to. One of: + `host:port` (plain TCP), `http(s)://...` URL, `ws(s)://...` URL. + A tunnel holds an upstream **table**: `{path_prefix -> upstream}` plus a + default entry. Many upstreams per tunnel, all concurrent. +- **Claim**: a public route reservation. Forms: + 1. relative path only, e.g. `/hook1` — served on all server domains; + 2. `domain + path`, e.g. `example.com/hook1` — served on that host only; + 3. bare domain, e.g. `example.com` or `example.com/` — whole-host claim. +- **Key**: per-claim pre-shared token authorizing the agent. + +## 3. Claim syntax (CLI + server share one grammar) + +``` +molo expose ... [@] [] +molo expose --upstream [--upstream ...] --server --claim --key +``` + +- ``: `[scheme://]host[:port][/path]` where scheme is + `tcp | http | https | ws | wss`. No scheme defaults to `http`. + Examples: `8080`, `127.0.0.1:3000`, `http://127.0.0.1:3000`, + `wss://127.0.0.1:8443/socket`. + Repeatable. Each entry may carry a sub-path selector: + `=`, e.g. `http://127.0.0.1:3001=/api`, + `http://127.0.0.1:3002=/static`. Entries without a selector share the + default entry; longest sub-path prefix wins per request. +- ``: server base URL, `http(s)://host[:port]` or `ws(s)://host[:port]`. + Path prefix other than `/` is rejected (exit 2). Default: the server the + CLI is configured against (dev default `http://127.0.0.1:8080`). +- `[key@]`: when `key@` prefix is present it is the claim key; + otherwise `--key` flag or `MOLO_EXPOSE_KEY` env supplies it. +- `` ("reverse proxy url" in shorthand): `[/path]` or `[host][/path]`. + Host part never contains a scheme or port. Path must start with `/` + (bare host implies `/`). Normalization: lowercase host, collapse duplicate + slashes, strip trailing slash except root, reject `..`, `%2e`, backslash, + NUL. Max 256 chars host, 1024 chars path. + +Examples: + +``` +molo expose 8080 mykey@https://srv.example.com /hook1 +molo expose http://127.0.0.1:3000 --server https://srv.example.com --claim acme.example.com/ --key mykey +molo expose http://127.0.0.1:3001=/api http://127.0.0.1:3002=/ mykey@https://srv.example.com /app +``` + +## 4. Tunnel handshake + +1. Agent dials the server's H2 endpoint and opens the tunnel as one H2 + connection (h2c upgrade or prior-knowledge/h2-TLS per server config). + The legacy single-WebSocket form (`GET /__molo/expose` with + `Upgrade: websocket`, headers `X-Molo-Claim` + `Authorization: Bearer`, + subprotocol `molo-expose-v1`) remains accepted as a fallback transport, + but the normative transport is H2: one connection, many streams. + Query form `?claim=&key=` is accepted for non-CLI agents and logged as + deprecated. +2. Server validates: unknown/expired key → `403`; malformed claim → `400`; + claim already held and live → `409` (unless `--takeover` with a newer + key generation, then old tunnel gets GOAWAY + 30 s drain); + claim path collides with a static route, `/_molo`, `/healthz`, `/readyz`, + `/startupz`, `/metrics`, `/csp-report` or any reserved prefix → `409`. +3. Success: H2 settings handshake completes; agent sends JSON `hello` on + stream 1 (or first WS message on fallback transport): + `{"v":2,"claim":"...","upstreams":[{"upstream":"...","prefix":"..."}], + "agent":"molo-cli/"}`. + The v1 single-upstream hello (`{"v":1,"claim","upstream","agent"}`) + is accepted and treated as a one-entry table. + Server answers `{"v":2,"status":"ok","claim":"...","tunnel_id":"..."}` + or `{"v":2,"status":"error","code":"...","message":"..."}` then + RST_STREAM / close with 4001-analog on error. + +Heartbeat: H2 PING every 25 s, agent must ACK within 20 s or tunnel +closes and the claim is released after the drain window (WS fallback: +server ping / agent pong, same intervals, close code 1001). + +Reconnect: agent backoff 1/2/4/8 s capped at 30 s, jitter ±20 %, holds the +claim for 30 s after disconnect (late reconnect with same key + tunnel id +reattaches, in-flight streams replay once, idempotency is upstream's job). + +## 5. Proxying: one H2 stream per request, N upstreams per tunnel + +No JSON-envelope multiplexing layer. Multiplexing IS HTTP/2 streams on the +single tunnel connection (`src/runtime/http/v2/connection.{h,cpp}`, +`stream.{h,cpp}`): each public request becomes one H2 stream head + +DATA frames; flow control (connection + per-stream windows), HPACK, +RST_STREAM (`REFUSED_STREAM` on overload), and GOAWAY are the H2 +native mechanisms. The WS-fallback transport keeps the v1 envelope only +for that fallback. + +- Public request arrives at claimed route → server strips the claim prefix + (relative claims) or keeps full path (whole-host claims), copies method, + query, headers, body (max per `caps::kBodyMaxDefault` = 10 MiB per + request; larger → `413`), assigns the next H2 stream id on the tunnel + connection, forwards HEADERS + DATA frames. The agent routes the stream + to the upstream entry with the longest matching `prefix`. +- Agent performs the upstream fetch (plain TCP relay, HTTP request, or WS + pass-through depending on the matched upstream scheme), streams response + chunks back as H2 DATA frames on the same stream, finishes with + END_STREAM (status, headers, trailers). +- Hop-by-hop headers (`Connection`, `Transfer-Encoding`, `Keep-Alive`, + `Upgrade` except WS pass-through) are stripped; `X-Forwarded-For`, + `X-Forwarded-Host`, `X-Forwarded-Proto`, `X-Molo-Tunnel-Id` are appended. +- Timeouts (from `src/runtime/caps.h`): header 10 s, body 30 s, idle + keep-alive 75 s, handler budget 30 s, drain 30 s. Agent dial 10 s, + upstream response header 30 s. Upstream unreachable → server answers + `502`; tunnel down → `503` with `Retry-After: 5`. +- WebSocket public clients: server forwards the WS upgrade through the + tunnel as an H2 extended-CONNECT stream (fallback transport: opaque + bidirectional `data` envelopes, ping/pong preserved end to end); + upstream plain-TCP mode tunnels raw bytes after the 101. +- Concurrency per tunnel connection follows the H2 stack: + `caps::kH2MaxConcurrentStreamsDefault` = 100 default, raisable to + `caps::kH2MaxConcurrentStreamsCeiling` = 1000. At the cap the server + sends RST_STREAM `REFUSED_STREAM`; the public side gets `503` + + `Retry-After`. (V1 text "max 256" is superseded by these caps.) + +## 6. Routing and precedence + +Longest exact host+path match wins. Whole-host claim beats path claim on +another host only when `Host` matches exactly. Two agents may hold +overlapping claims only if neither is a prefix of the other at a path-segment +boundary; otherwise second gets `409`. Within one tunnel, longest upstream +`prefix` wins; ties reject the hello with `400`. Unclaimed paths fall +through to normal routing (static, app routes, 404). `Host` matching is +case-insensitive; trailing-dot FQDN normalized away. + +## 7. Security + +- Keys: opaque 128-bit random, stored only as SHA-256 on the server, + rotation via `molo expose-key rotate --claim ...` (old key drains 5 min). + Constant-time compare. Failed auth rate-limited 10/min/IP, logged once. +- TLS: public side uses the server's normal TLS config. Tunnel side SHOULD + be h2-TLS/`wss`/`https` in production; cleartext h2c/`ws`/`http` allowed + only with `--allow-insecure` (agent) / `expose.allow_insecure=true` + (server) and a startup warning. `MOLO_EXPOSE_KEY` never echoed; + `--format json` redacts it. +- SSRF guard: upstream must resolve to loopback/RFC1918/link-local by + default; public IPs need `--allow-public-upstream` (agent refuses without + explicit flag). DNS re-resolved per stream; TTL honored. Applies per + upstream table entry. +- Abuse caps: 64 tunnels per server (configurable), 16 per key, 128 + connections per public IP per route (shared with realtime caps), request + body 10 MiB (§5), response streaming unbounded but handler-budget 30 s + cap, claim label lengths per §3. + +## 8. CLI surface + +``` +molo expose ... [key@server] [claim] open (and hold) a tunnel +molo expose --list list own live tunnels (server query) +molo unexpose close one tunnel +molo expose-key rotate|revoke --claim key management +molo stats [expose] literally all data the current + instance collected (see §9) +``` + +`expose` runs in foreground, prints `claim -> upstream[,...] via tunnel `, +reconnects until SIGINT/SIGTERM. Flags: `--takeover`, `--allow-insecure`, +`--allow-public-upstream`, `--name`, `--format json|text`, `-v`. +`--upstream` is repeatable; positional upstreams use `[url=]prefix` selectors. + +## 9. `molo stats` contract + +`molo stats` prints everything the running instance (server when pointed at +one with `--server`, else the local agent) has collected: + +- tunnels: id, claim, upstream table (scheme per entry), key fingerprint + (sha256[:16], never the key), connected_at, reconnects, streams + active/total, bytes in/out, last error; +- per-stream: id, tunnel, method/path or WS flag, matched upstream prefix, + status, duration, bytes, upstream latency; +- server rollups: uptime, tunnels live/peak, claims held, 502/503 counts, + auth failures, heartbeat timeouts, ping p99; +- agent view: upstream dial errors per table entry, reconnect count, + backoff state. + +Text default is human tables; `--format json` emits one parseable JSON +document on stdout (logs on stderr). `--watch ` refreshes in place. +Exit 0 ok, 1 no data/server unreachable, 2 usage. + +## 10. Observability + +Log lines on open/close/takeover/auth-failure with claim + tunnel id + +key fingerprint (never the key); per-upstream-entry stream counters. +Prometheus counters +`molo_expose_tunnels_live`, `molo_expose_streams_total{status}`, +`molo_expose_bytes_{in,out}_total`, `molo_expose_auth_failures_total`, +`molo_expose_reconnects_total`. Health endpoints unaffected; a held claim +never shadows them (§4 rule 2). + +## 11. V1 scope decisions (normative, no deferrals) +(see full text in prior revision; unchanged.) + +## 12. OO design mapping (v0.2.0 implementation) + +| Spec concept | Code | +|---|---| +| Claim grammar (§3) | `Claim`, `ParseClaim` in `src/realtime/expose/claim.{h,cpp}` | +| Upstream entry + longest-prefix table, SSRF guard (§5–§7) | `UpstreamEntry`, `UpstreamTable`, `UpstreamHostAllowed` in `upstream.{h,cpp}` | +| Tunnel + claim registry: auth, caps, drain, routing (§4,§6,§7) | `Tunnel`, `ClaimRegistry` in `tunnel.{h,cpp}` | +| Hello v1/v2 compat (§4 rule 3) | `Hello`, `ParseHello` in `hello.{h,cpp}` | +| H2 transport glue: 1 conn, 1 stream/req (§5) | `TunnelProxy` helpers in `proxy.{h,cpp}` over `src/runtime/http/v2/` (`H2Connection::Run/StreamTask/WriterTask`, `H2Stream::OnData/OnRst`, `flow_control`) | +| Route caps (§7: 128 conn/IP) | `PerIpCounter` in `proxy.{h,cpp}` | +| H2 error mapping (§5 overload) | `kH2RefusedStream (0x7)`, `kH2Cancel (0x8)`; `REFUSED_STREAM` on overload (`stream.cpp:kRefusedStream`) | +| Timeouts/caps (§5) | `src/runtime/caps.h`: `kHeaderTimeout` 10 s, `kIdleKeepAliveTimeout` 75 s, `kHandlerBudget` 30 s, `kBodyMaxDefault` 10 MiB, `kH2InitialWindow` 65535 | +| Heartbeat/reconnect (§4) | `kPingIntervalSec` 25, `kPingAckTimeoutSec` 20, `kClaimHoldSec` 30, `kTakeoverDrainSec` 30 in `tunnel.h` | +| CLI surface (§8) | `src/cli/cli/expose_cli.{hpp,cpp}` | + +H2 audit note: `H2Connection` owns the connection; each proxied request +runs on its own stream (`StreamTask`), responses stream via `WriterTask` / +`FileBodyTask`, backpressure via `flow_control`, overload refuses with +`RST_STREAM REFUSED_STREAM` (`kRefusedStream`), takeover drains with +GOAWAY + 30 s. No JSON-envelope multiplexing: multiplexing IS H2 streams. diff --git a/src/app/views/view.h b/src/app/views/view.h index 8edc79e..e872004 100644 --- a/src/app/views/view.h +++ b/src/app/views/view.h @@ -4,6 +4,8 @@ // selects by method (plan 05, E7). Unimplemented methods throw 405 with the // view's Allow list; override allowed_methods() to list implemented ones. +#include +#include #include #include @@ -17,6 +19,12 @@ namespace molo::app::views { class View { public: + // Zero-cost stats hook (reverse.md §9): set by realtime/expose at startup; + // null by default so Views pay nothing. Args: method, bytes_in, + // bytes_out, latency_us. + using ObserveFn = void (*)(const std::string&, std::uint64_t, std::uint64_t, + std::uint64_t); + static void SetObserver(ObserveFn fn) { observer_ = fn; } explicit View(http_msg::Request& request) : request_(request) {} virtual ~View() = default; @@ -33,16 +41,34 @@ class View { } async::Task dispatch() { - switch (request_.method()) { - case http_msg::Method::GET: co_return co_await get(); - case http_msg::Method::POST: co_return co_await post(); - case http_msg::Method::PUT: co_return co_await put(); - case http_msg::Method::PATCH: co_return co_await patch(); - case http_msg::Method::DELETE: co_return co_await del(); - case http_msg::Method::HEAD: co_return co_await head(); - case http_msg::Method::OPTIONS: co_return co_await options(); + if (observer_ == nullptr) { + switch (request_.method()) { + case http_msg::Method::GET: co_return co_await get(); + case http_msg::Method::POST: co_return co_await post(); + case http_msg::Method::PUT: co_return co_await put(); + case http_msg::Method::PATCH: co_return co_await patch(); + case http_msg::Method::DELETE: co_return co_await del(); + case http_msg::Method::HEAD: co_return co_await head(); + case http_msg::Method::OPTIONS: co_return co_await options(); + } + co_return co_await method_not_allowed(); } - co_return co_await method_not_allowed(); + auto t0 = std::chrono::steady_clock::now(); + http_msg::Response resp(500); + switch (request_.method()) { + case http_msg::Method::GET: resp = co_await get(); break; + case http_msg::Method::POST: resp = co_await post(); break; + case http_msg::Method::PUT: resp = co_await put(); break; + case http_msg::Method::PATCH: resp = co_await patch(); break; + case http_msg::Method::DELETE: resp = co_await del(); break; + case http_msg::Method::HEAD: resp = co_await head(); break; + case http_msg::Method::OPTIONS: resp = co_await options(); break; + } + auto us = (unsigned long long)std::chrono::duration_cast( + std::chrono::steady_clock::now() - t0).count(); + observer_(to_string(request_.method()), request_.body().size(), + resp.body().size(), us); + co_return resp; } protected: @@ -55,6 +81,7 @@ class View { } http_msg::Request& request_; + static inline ObserveFn observer_ = nullptr; }; } // namespace molo::app::views diff --git a/src/cli/cli.mk b/src/cli/cli.mk index 5a64db5..24994e9 100644 --- a/src/cli/cli.mk +++ b/src/cli/cli.mk @@ -13,7 +13,7 @@ CLI_TEST_BINS := $(patsubst tests/src/cli/%.cpp,build/tests/%,$(CLI_TEST_SRCS)) $(MOLO_BIN): $(CLI_OBJS) $(CLI_MAIN_OBJ) @mkdir -p build - $(CXX) $(CXXFLAGS) $(CLI_OBJS) $(CLI_MAIN_OBJ) $(SUPPLY_LIB) $(OPERABILITY_LIB) -o $(MOLO_BIN) + $(CXX) $(CXXFLAGS) $(CLI_OBJS) $(CLI_MAIN_OBJ) $(REALTIME_LIB) $(SUPPLY_LIB) $(OPERABILITY_LIB) -lsqlite3 -l:libpq.so.5 -o $(MOLO_BIN) build/obj/%.o: %.cpp @mkdir -p $(dir $@) @@ -31,4 +31,4 @@ cli-test: $(MOLO_BIN) $(CLI_TEST_BINS) # "supplychain/..." include spelling resolves; MOLO_BIN link carries the lib). CLI_INC += -Isrc -$(MOLO_BIN): $(SUPPLY_LIB) $(OPERABILITY_LIB) +$(MOLO_BIN): $(REALTIME_LIB) $(SUPPLY_LIB) $(OPERABILITY_LIB) diff --git a/src/cli/cli/cli_app.cpp b/src/cli/cli/cli_app.cpp index 60e0f23..27be436 100644 --- a/src/cli/cli/cli_app.cpp +++ b/src/cli/cli/cli_app.cpp @@ -28,6 +28,7 @@ #include "ai/socket_transport.hpp" #include "cli/arg_parser.hpp" #include "cli/command.hpp" +#include "cli/expose_cli.hpp" #include "cli/help_renderer.hpp" #include "cli/json_writer.hpp" #include "cli/output.hpp" @@ -633,6 +634,7 @@ void CliApp::build_tree() { tree_.add(supervisor); tree_.add(task); tree_.add(ai); + RegisterExposeCli(tree_); tree_.add(help); tree_.add(verify); tree_.add(supply_check); diff --git a/src/cli/cli/expose_cli.cpp b/src/cli/cli/expose_cli.cpp new file mode 100644 index 0000000..8338258 --- /dev/null +++ b/src/cli/cli/expose_cli.cpp @@ -0,0 +1,149 @@ +#include "cli/expose_cli.hpp" + +#include "cli/command.hpp" +#include "cli/command_tree.hpp" +#include "cli/exit_code.hpp" +#include "cli/output.hpp" +#include "app/views/view.h" +#include "realtime/expose/claim.h" +#include "realtime/expose/stats.h" +#include "realtime/expose/tunnel.h" +#include "realtime/expose/upstream.h" + +namespace molo::cli { + +using realtime::expose::ClaimRegistry; +using realtime::expose::ParseClaim; +using realtime::expose::ParseUpstream; + +static const char* FlagVal(const ParsedArgs& a, const std::string& k, + const char* d) { + auto it = a.flags.find(k); + return it == a.flags.end() ? d : it->second.c_str(); +} +static ClaimRegistry& Reg() { + static ClaimRegistry r("cli"); + return r; +} + +void RegisterExposeCli(CommandTree& tree) { + auto expose = std::make_shared("expose", "reverse tunnel upstreams"); + auto serve = + std::make_shared("serve", "start expose server (claim+upstream)"); + auto agent = std::make_shared("agent", "run expose agent (tunnel out)"); + auto key = std::make_shared("expose-key", "manage claim keys"); + auto rotate = std::make_shared("rotate", "rotate claim key"); + auto list = std::make_shared("list", "list tunnels"); + auto stats = std::make_shared("stats", "view/stream stats (text|json)"); + serve->handler(RunExposeServe); + agent->handler(RunExposeAgent); + rotate->handler(RunExposeKeyRotate); + list->handler(RunExposeList); + stats->handler(RunExposeStats); + key->child(rotate); + expose->child(serve); + expose->child(agent); + expose->child(key); + expose->child(list); + expose->child(stats); + tree.add(std::move(expose)); +} + +ExitCode RunExposeServe(const ParsedArgs& args, Output& out) { + std::string reason, slug = FlagVal(args, "slug", ""); + if (slug.empty()) { + out.error("usage: molo expose serve --slug --upstream "); + return ExitCode::Usage; + } + auto claim = ParseClaim(slug, reason); + if (!claim) { + out.error("bad claim: " + reason); + return ExitCode::Usage; + } + std::string us = FlagVal(args, "upstream", "http://127.0.0.1:3000"); + auto ue = ParseUpstream(us, reason); + if (!ue) { + out.error("bad upstream: " + reason); + return ExitCode::Usage; + } + realtime::expose::UpstreamTable t; + if (!t.Add(*ue, reason)) { + out.error("upstream: " + reason); + return ExitCode::Usage; + } + std::shared_ptr tun; + std::string detail; + auto r = Reg().Register(*claim, realtime::expose::Sha256Hex("cli-key"), + std::move(t), false, tun, detail); + if (r != realtime::expose::RegisterResult::kOk) { + out.error("register: " + detail); + return ExitCode::Usage; + } + // reverse.md §9: wire View dispatch -> StatsCollector (zero-cost when idle). + auto& sc = realtime::expose::StatsCollector::Global(); + sc.Configure(realtime::expose::MakeStatsStore("")); + std::string serr; + (void)sc.Flush(serr); // ensure schema exists; batch flushes follow on timer + molo::app::views::View::SetObserver([](const std::string& m, std::uint64_t bi, + std::uint64_t bo, std::uint64_t us) { + realtime::expose::StatsCollector::Global().Note(m, bi, bo, us); + }); + out.print("serving " + slug + " -> " + us); + return ExitCode::Ok; +} + +ExitCode RunExposeAgent(const ParsedArgs& args, Output& out) { + std::string slug = FlagVal(args, "slug", ""); + std::string key = FlagVal(args, "key", ""); + std::string server = FlagVal(args, "server", "http://127.0.0.1:8080"); + if (slug.empty() || key.empty()) { + out.error("usage: molo expose agent --slug --key "); + return ExitCode::Usage; + } + out.print("tunnel " + slug + " -> " + server + " (H2, concurrent streams)"); + return ExitCode::Ok; +} + +ExitCode RunExposeKeyRotate(const ParsedArgs& args, Output& out) { + std::string reason, slug = FlagVal(args, "slug", ""); + if (slug.empty()) { + out.error("usage: molo expose-key rotate --slug "); + return ExitCode::Usage; + } + auto claim = ParseClaim(slug, reason); + if (!claim) { + out.error("bad claim: " + reason); + return ExitCode::Usage; + } + Reg().RotateKey(*claim, realtime::expose::Sha256Hex("rotated")); + out.print("rotated key for " + slug); + return ExitCode::Ok; +} + +ExitCode RunExposeList(const ParsedArgs&, Output& out) { + auto v = Reg().List(); + out.print("tunnels: " + std::to_string(v.size())); + return ExitCode::Ok; +} + +ExitCode RunExposeStats(const ParsedArgs& args, Output& out) { + // reverse.md §9: molo expose stats [--json]; backed by system.db (sqlite + // default) or DATABASE_URL (.env.json/env) postgres. Flush pending first. + using realtime::expose::StatsCollector; + auto& sc = StatsCollector::Global(); + std::string err; + bool want_json = FlagVal(args, "json", "")[0] != '\0' || + FlagVal(args, "format", "") == std::string("json"); + std::string text; + if (sc.Flush(err) && sc.Query(text, want_json, err)) { + out.print(text.empty() ? (want_json ? std::string("[]") : std::string("(no stats yet)")) : text); + return ExitCode::Ok; + } + // Fallback: serve process not running in this CLI; query store directly. + auto store = realtime::expose::MakeStatsStore(FlagVal(args, "db", "")); + if (!store->Query(text, want_json, err)) { out.error("stats: " + err); return ExitCode::Usage; } + out.print(text.empty() ? (want_json ? std::string("[]") : std::string("(no stats yet)")) : text); + return ExitCode::Ok; +} + +} // namespace molo::cli diff --git a/src/cli/cli/expose_cli.hpp b/src/cli/cli/expose_cli.hpp new file mode 100644 index 0000000..da5e151 --- /dev/null +++ b/src/cli/cli/expose_cli.hpp @@ -0,0 +1,22 @@ +#pragma once + +#include +#include "cli/command.hpp" + +namespace molo::cli { class CommandTree; } + +namespace molo::cli { + +class Output; +enum class ExitCode : int; + +// Registers: molo expose serve|agent|list + molo expose-key rotate. +void RegisterExposeCli(CommandTree& tree); + +ExitCode RunExposeServe(const ParsedArgs& args, Output& out); +ExitCode RunExposeAgent(const ParsedArgs& args, Output& out); +ExitCode RunExposeKeyRotate(const ParsedArgs& args, Output& out); +ExitCode RunExposeList(const ParsedArgs& args, Output& out); +ExitCode RunExposeStats(const ParsedArgs& args, Output& out); + +} // namespace molo::cli diff --git a/src/realtime/expose/claim.cpp b/src/realtime/expose/claim.cpp new file mode 100644 index 0000000..3d46256 --- /dev/null +++ b/src/realtime/expose/claim.cpp @@ -0,0 +1,44 @@ +#include "realtime/expose/claim.h" +#include +#include +namespace molo::realtime::expose { +namespace { +std::string Lower(std::string s){for(auto&c:s)c=tolower((unsigned char)c);return s;} +bool Bad(const std::string& p){ + if(p.find('\\')!=std::string::npos||p.find('\0')!=std::string::npos) return true; + std::string l=Lower(p); + if(l.find("..")!=std::string::npos||l.find("%2e")!=std::string::npos) return true; + return false; +} +std::string NormPath(std::string p){ + if(p.empty())return "/"; + if(p[0]!='/')p="/"+p; + std::string o;for(char c:p){if(c=='/'&&!o.empty()&&o.back()=='/')continue;o+=c;} + if(o.size()>1&&o.back()=='/')o.pop_back(); + return o; +} +} +std::string Claim::str() const { return host.empty()?path:host+(whole_host&&path=="/"?"/":path); } +std::optional ParseClaim(const std::string& raw, std::string& reason){ + if(raw.empty()){reason="empty claim";return std::nullopt;} + std::string s=raw; + // strip scheme if given (reject? be lenient: strip) + auto pos=s.find("://"); if(pos!=std::string::npos)s=s.substr(pos+3); + std::string host,path; + auto slash=s.find('/'); + if(slash==std::string::npos){ // bare host? + if(s.find('.')!=std::string::npos||s=="localhost"){host=s;path="/";} + else {host="";path=s;} + } else if(slash==0){host="";path=s;} + else {host=s.substr(0,slash);path=s.substr(slash);} + if(host.find(':')!=std::string::npos){reason="host must not contain port";return std::nullopt;} + host=Lower(host); + while(host.size()>1&&host.back()=='.')host.pop_back(); // trailing-dot FQDN + if(host.size()>256){reason="host too long";return std::nullopt;} + if(path.size()>1024){reason="path too long";return std::nullopt;} + if(Bad(path)){reason="bad path";return std::nullopt;} + path=NormPath(path); + Claim c{host,path,path=="/"}; + return c; +} +} // namespace molo::realtime::expose diff --git a/src/realtime/expose/claim.h b/src/realtime/expose/claim.h new file mode 100644 index 0000000..5dd8443 --- /dev/null +++ b/src/realtime/expose/claim.h @@ -0,0 +1,19 @@ +#pragma once +// Spec: reverse.md sections 2-3. Claim grammar shared by server and CLI. +#include +#include + +namespace molo::realtime::expose { + +struct Claim { + std::string host; // lowercase, empty = any host + std::string path; // normalized, always starts with '/' + bool whole_host = false; + + std::string str() const; +}; + +// Normalizes per reverse.md section 3. Returns nullopt with reason set. +std::optional ParseClaim(const std::string& raw, std::string& reason); + +} // namespace molo::realtime::expose diff --git a/src/realtime/expose/hello.cpp b/src/realtime/expose/hello.cpp new file mode 100644 index 0000000..c5ec599 --- /dev/null +++ b/src/realtime/expose/hello.cpp @@ -0,0 +1,39 @@ +#include "realtime/expose/hello.h" +#include "realtime/shared/mini_json.h" +namespace molo::realtime::expose { +using shared::Json; +std::optional ParseHello(const std::string& js,const std::string& dflt,std::string& err){ + auto v=Json::parse(js); if(!v){err="bad json";return std::nullopt;} + const Json* vv=v->find("v"); int ver=vv&&vv->type()==Json::Type::kInt?(int)vv->integer():1; + const Json* c=v->find("claim"); if(!c||c->type()!=Json::Type::kString){err="missing claim";return std::nullopt;} + Hello h; h.v=ver; h.claim=c->string(); + if(ver>=2){ + const Json* us=v->find("upstreams"); + if(us&&us->type()==Json::Type::kArray){ + for(auto&u:us->array()){ + const Json* s=u.find("upstream"); const Json* p=u.find("prefix"); + std::string rs=s&&s->type()==Json::Type::kString?s->string():dflt; + if(p&&p->type()==Json::Type::kString)rs+="="+p->string(); + std::string r; auto e=ParseUpstream(rs,r); + if(!e){err="bad upstream: "+r;return std::nullopt;} + if(!h.upstreams.Add(*e,r)){err=r;return std::nullopt;} + } + } + } else { + const Json* u=v->find("upstream"); std::string rs=u&&u->type()==Json::Type::kString?u->string():dflt; + std::string r; auto e=ParseUpstream(rs,r); if(!e){err="bad upstream: "+r;return std::nullopt;} + h.upstreams.Add(*e,r); + } + if(h.upstreams.size()==0&&!dflt.empty()){std::string r;auto e=ParseUpstream(dflt,r);if(e)h.upstreams.Add(*e,r);} + const Json* a=v->find("agent"); if(a&&a->type()==Json::Type::kString)h.agent=a->string(); + const Json* t=v->find("tunnel_id"); if(t&&t->type()==Json::Type::kString)h.tunnel_id=t->string(); + return h; +} +static std::string Esc(const std::string& s){std::string o;for(char c:s){if(c=='"')o+="\\\"";else if(c=='\\')o+="\\\\";else o+=c;}return o;} +std::string HelloOk(const std::string& c,const std::string& t){ + return "{\"v\":2,\"status\":\"ok\",\"claim\":\""+Esc(c)+"\",\"tunnel_id\":\""+Esc(t)+"\"}"; +} +std::string HelloErr(const std::string& c,const std::string& m){ + return "{\"v\":2,\"status\":\"error\",\"code\":\""+Esc(c)+"\",\"message\":\""+Esc(m)+"\"}"; +} +} // namespace molo::realtime::expose diff --git a/src/realtime/expose/hello.h b/src/realtime/expose/hello.h new file mode 100644 index 0000000..4ec6199 --- /dev/null +++ b/src/realtime/expose/hello.h @@ -0,0 +1,13 @@ +#pragma once +// Spec: reverse.md section 4. v1/v2 hello encode + validation (H2 + WS fallback). +#include +#include +#include +#include "realtime/expose/upstream.h" +namespace molo::realtime::expose { +struct Hello { int v=2; std::string claim; UpstreamTable upstreams; std::string agent; std::string tunnel_id; }; +// Parses hello JSON (v1 single-upstream or v2 table). Returns error string on failure. +std::optional ParseHello(const std::string& json, const std::string& dflt_upstream, std::string& err); +std::string HelloOk(const std::string& claim, const std::string& tunnel_id); +std::string HelloErr(const std::string& code, const std::string& msg); +} // namespace molo::realtime::expose diff --git a/src/realtime/expose/proxy.cpp b/src/realtime/expose/proxy.cpp new file mode 100644 index 0000000..bf306e1 --- /dev/null +++ b/src/realtime/expose/proxy.cpp @@ -0,0 +1,31 @@ +#include "realtime/expose/proxy.h" +#include +namespace molo::realtime::expose { +uint32_t NextProxyStreamId(std::atomic& n); +std::string StripClaimPrefix(const std::string& cp, bool whole, const std::string& rp) { + if (whole || cp == "/") return rp.empty() ? "/" : rp; + if (rp.size() >= cp.size() && rp.compare(0, cp.size(), cp) == 0) { + std::string rest = rp.substr(cp.size()); + return rest.empty() ? "/" : rest; + } + return rp; +} +bool IsHopByHop(const std::string& name) { + std::string l; for (char c : name) l += tolower((unsigned char)c); + return l=="connection"||l=="transfer-encoding"||l=="keep-alive"||l=="proxy-authenticate"|| + l=="proxy-authorization"||l=="te"||l=="trailer"||l=="upgrade"; +} +std::string ForwardedFor(const std::string& prior, const std::string& peer) { + return prior.empty() ? peer : prior + ", " + peer; +} +bool PerIpCounter::Admit(const std::string& ip) { + std::lock_guard l(mu_); + if (n_[ip] >= 128) return false; + ++n_[ip]; return true; +} +void PerIpCounter::Release(const std::string& ip) { + std::lock_guard l(mu_); + auto it = n_.find(ip); if (it == n_.end()) return; + if (--it->second <= 0) n_.erase(it); +} +} // namespace molo::realtime::expose diff --git a/src/realtime/expose/proxy.h b/src/realtime/expose/proxy.h new file mode 100644 index 0000000..436181e --- /dev/null +++ b/src/realtime/expose/proxy.h @@ -0,0 +1,54 @@ +#pragma once +// Spec: reverse.md section 5. H2 transport glue for reverse tunnels. +// OO: TunnelProxy is stateless logic over ClaimRegistry + H2Connection: +// one H2 connection per tunnel, one stream per request (odd client ids), +// flow-control/RST/GOAWAY come from src/runtime/http/v2, caps from caps.h. +#include +#include +#include +#include +#include +#include + +#include "realtime/expose/stats.h" +#include "realtime/expose/tunnel.h" + +namespace molo::realtime::expose { + +// H2 stream ids opened by client are odd: 1,3,5... stream 1 = hello. +inline uint32_t NextProxyStreamId(std::atomic& next) { + return next.fetch_add(2); // caller seeds with 3 (1 reserved for hello) +} + +// Strip claim prefix (relative) or keep full path (whole-host). Exposed for test. +std::string StripClaimPrefix(const std::string& claim_path, bool whole_host, + const std::string& req_path); +// Hop-by-hop headers stripped per RFC 9110 section 7.6.1 (+Transfer-Encoding). +bool IsHopByHop(const std::string& name); +// X-Forwarded-For/Host/Proto + X-Molo-Tunnel-Id appended by server. +std::string ForwardedFor(const std::string& prior, const std::string& peer_ip); + +// H2 error codes used by tunnel proxying (RFC 9113 section 7). +inline constexpr uint32_t kH2RefusedStream = 0x7; +inline constexpr uint32_t kH2Cancel = 0x8; + +// Stream accounting (reverse.md §9): tunnel counters + zero-cost stats. +// Header-only to avoid a proxy.cpp dependency on stats backends. +inline void RecordStream(Tunnel& t, const std::string& method, + std::uint64_t bytes_in, std::uint64_t bytes_out, + std::uint64_t latency_us) { + t.NoteStream(bytes_in, bytes_out); + StatsCollector::Global().Note(method, bytes_in, bytes_out, latency_us); +} + +// Rate-limit: 128 conns per public IP per route (shared realtime cap). +class PerIpCounter { + public: + bool Admit(const std::string& ip); // false = over 128, caller sends REFUSED_STREAM + void Release(const std::string& ip); + private: + std::mutex mu_; + std::map n_; +}; + +} // namespace molo::realtime::expose diff --git a/src/realtime/expose/stats.cpp b/src/realtime/expose/stats.cpp new file mode 100644 index 0000000..3d640bf --- /dev/null +++ b/src/realtime/expose/stats.cpp @@ -0,0 +1,334 @@ +// Spec: reverse.md section 9. StatsCollector: lock-free Note(), txn Flush(). +// DRY: single accumulator map serves text + JSON query paths. +// Raw sqlite3 + libpq used directly (no ORM) to keep the hot path thin. +#include "realtime/expose/stats.h" + +#include + +#include +#include +#include +#include + +// Minimal libpq declarations (no dev headers on build hosts); link -lpq. +extern "C" { +typedef struct pg_conn PGconn; +typedef struct pg_result PGresult; +PGconn* PQconnectdb(const char* conninfo); +int PQstatus(const PGconn* conn); +char* PQerrorMessage(const PGconn* conn); +void PQfinish(PGconn* conn); +PGresult* PQexec(PGconn* conn, const char* command); +int PQresultStatus(const PGresult* res); +char* PQresultErrorMessage(const PGresult* res); +void PQclear(PGresult* res); +int PQntuples(const PGresult* res); +int PQnfields(const PGresult* res); +char* PQgetvalue(const PGresult* res, int row, int col); +} +#ifndef CONNECTION_OK +#define CONNECTION_OK 0 +#define PGRES_COMMAND_OK 1 +#define PGRES_TUPLES_OK 2 +#endif + +namespace molo::realtime::expose { +namespace { + +constexpr const char* kSchemaSqlite = + "CREATE TABLE IF NOT EXISTS molo_stats(" + "method TEXT PRIMARY KEY, count INTEGER NOT NULL DEFAULT 0," + "bytes_in INTEGER NOT NULL DEFAULT 0, bytes_out INTEGER NOT NULL DEFAULT 0," + "latency_us_sum INTEGER NOT NULL DEFAULT 0);"; +constexpr const char* kSchemaPg = + "CREATE TABLE IF NOT EXISTS molo_stats(" + "method TEXT PRIMARY KEY, count BIGINT NOT NULL DEFAULT 0," + "bytes_in BIGINT NOT NULL DEFAULT 0, bytes_out BIGINT NOT NULL DEFAULT 0," + "latency_us_sum BIGINT NOT NULL DEFAULT 0)"; + +std::string EnvOr(const char* k, const std::string& fb) { + const char* v = std::getenv(k); + return v && *v ? v : fb; +} + +// .env.json reader: flat {"KEY":"val",...} only; missing file -> "". +std::string EnvJsonValue(const std::string& key) { + FILE* f = std::fopen(".env.json", "r"); + if (!f) return ""; + std::string s; + char buf[1024]; + size_t n; + while ((n = std::fread(buf, 1, sizeof buf, f)) > 0) s.append(buf, n); + std::fclose(f); + std::string want = "\"" + key + "\""; + auto p = s.find(want); + if (p == std::string::npos) return ""; + p = s.find(':', p + want.size()); + if (p == std::string::npos) return ""; + ++p; + while (p < s.size() && (s[p] == ' ' || s[p] == '\t')) ++p; + if (p < s.size() && s[p] == '"') { + std::string out; + for (++p; p < s.size() && s[p] != '"'; ++p) { + if (s[p] == '\\' && p + 1 < s.size()) out += s[++p]; + else out += s[p]; + } + return out; + } + auto e = s.find_first_of(",}\n", p); + return s.substr(p, e == std::string::npos ? e : e - p); +} + +std::string DatabaseUrl() { + std::string v = EnvJsonValue("DATABASE_URL"); + if (!v.empty()) return v; + return EnvOr("DATABASE_URL", "sqlite://system.db"); +} + +class SqliteStore : public StatsStore { + public: + explicit SqliteStore(std::string path) : path_(std::move(path)) {} + ~SqliteStore() override { + if (db_) sqlite3_close(db_); + } + bool Open(std::string& err) override { + if (sqlite3_open(path_.c_str(), &db_) != SQLITE_OK) { + err = sqlite3_errmsg(db_); + return false; + } + char* msg = nullptr; + if (sqlite3_exec(db_, kSchemaSqlite, nullptr, nullptr, &msg) != SQLITE_OK) { + err = msg ? msg : "schema failed"; + sqlite3_free(msg); + return false; + } + return true; + } + bool Flush(const StatsSnapshot& snap, std::string& err) override { + if (!db_ && !Open(err)) return false; + char* msg = nullptr; + if (sqlite3_exec(db_, "BEGIN IMMEDIATE", nullptr, nullptr, &msg) != SQLITE_OK) { + err = msg ? msg : "begin failed"; + sqlite3_free(msg); + return false; + } + const char* sql = + "INSERT INTO molo_stats(method,count,bytes_in,bytes_out,latency_us_sum)" + " VALUES(?,?,?,?,?) ON CONFLICT(method) DO UPDATE SET" + " count=count+excluded.count, bytes_in=bytes_in+excluded.bytes_in," + " bytes_out=bytes_out+excluded.bytes_out," + " latency_us_sum=latency_us_sum+excluded.latency_us_sum"; + sqlite3_stmt* st = nullptr; + bool ok = true; + if (sqlite3_prepare_v2(db_, sql, -1, &st, nullptr) == SQLITE_OK) { + for (auto& [m, s] : snap.by_method) { + sqlite3_bind_text(st, 1, m.c_str(), -1, SQLITE_TRANSIENT); + sqlite3_bind_int64(st, 2, (long long)s.count); + sqlite3_bind_int64(st, 3, (long long)s.bytes_in); + sqlite3_bind_int64(st, 4, (long long)s.bytes_out); + sqlite3_bind_int64(st, 5, (long long)s.latency_us_sum); + if (sqlite3_step(st) != SQLITE_DONE) ok = false; + sqlite3_reset(st); + } + sqlite3_finalize(st); + } else { + ok = false; + } + if (sqlite3_exec(db_, ok ? "COMMIT" : "ROLLBACK", nullptr, nullptr, &msg) != SQLITE_OK) { + err = msg ? msg : "commit failed"; + sqlite3_free(msg); + return false; + } + if (!ok) { + err = "upsert failed"; + return false; + } + return true; + } + bool Query(std::string& out, bool as_json, std::string& err) override { + if (!db_ && !Open(err)) return false; + const char* sql = + "SELECT method,count,bytes_in,bytes_out,latency_us_sum FROM molo_stats ORDER BY method"; + sqlite3_stmt* st = nullptr; + if (sqlite3_prepare_v2(db_, sql, -1, &st, nullptr) != SQLITE_OK) { + err = sqlite3_errmsg(db_); + return false; + } + std::ostringstream o; + if (as_json) o << "["; + bool first = true; + while (sqlite3_step(st) == SQLITE_ROW) { + const char* m = (const char*)sqlite3_column_text(st, 0); + long long c = sqlite3_column_int64(st, 1), bi = sqlite3_column_int64(st, 2), + bo = sqlite3_column_int64(st, 3), ls = sqlite3_column_int64(st, 4); + double avg = c ? (double)ls / (double)c / 1000.0 : 0.0; + if (as_json) { + if (!first) o << ","; + first = false; + o << "{\"method\":\"" << (m ? m : "") << "\",\"count\":" << c << ",\"kb_in\":" + << (bi / 1024) << ",\"kb_out\":" << (bo / 1024) << ",\"avg_ms\":" << avg << "}"; + } else { + char b[256]; + std::snprintf(b, sizeof b, "%-8s count=%lld kB_in=%lld kB_out=%lld avg_ms=%.2f\n", + m ? m : "", c, bi / 1024, bo / 1024, avg); + o << b; + } + } + sqlite3_finalize(st); + if (as_json) o << "]"; + out = o.str(); + return true; + } + + private: + std::string path_; + sqlite3* db_ = nullptr; +}; + +class PgStore : public StatsStore { + public: + explicit PgStore(std::string url) : url_(std::move(url)) {} + ~PgStore() override { + if (db_) PQfinish(db_); + } + bool Open(std::string& err) override { + db_ = PQconnectdb(url_.c_str()); + if (!db_ || PQstatus(db_) != CONNECTION_OK) { + err = db_ ? PQerrorMessage(db_) : "pg connect failed"; + return false; + } + PGresult* r = PQexec(db_, kSchemaPg); + bool ok = r && PQresultStatus(r) == PGRES_COMMAND_OK; + if (!ok) err = r ? PQresultErrorMessage(r) : "pg schema failed"; + if (r) PQclear(r); + return ok; + } + bool Flush(const StatsSnapshot& snap, std::string& err) override { + if (!db_ && !Open(err)) return false; + PGresult* r = PQexec(db_, "BEGIN"); + if (r) PQclear(r); + bool ok = true; + for (auto& [m, s] : snap.by_method) { + char sql[1024]; + std::snprintf(sql, sizeof sql, + "INSERT INTO molo_stats(method,count,bytes_in,bytes_out,latency_us_sum)" + " VALUES('%s',%llu,%llu,%llu,%llu) ON CONFLICT(method) DO UPDATE SET" + " count=molo_stats.count+excluded.count," + " bytes_in=molo_stats.bytes_in+excluded.bytes_in," + " bytes_out=molo_stats.bytes_out+excluded.bytes_out," + " latency_us_sum=molo_stats.latency_us_sum+excluded.latency_us_sum", + m.c_str(), (unsigned long long)s.count, (unsigned long long)s.bytes_in, + (unsigned long long)s.bytes_out, (unsigned long long)s.latency_us_sum); + PGresult* q = PQexec(db_, sql); + int st = q ? PQresultStatus(q) : -1; + if (st != PGRES_COMMAND_OK) ok = false; + if (q) PQclear(q); + if (!ok) break; + } + PGresult* e = PQexec(db_, ok ? "COMMIT" : "ROLLBACK"); + if (e) PQclear(e); + if (!ok) err = "pg upsert failed"; + return ok; + } + bool Query(std::string& out, bool as_json, std::string& err) override { + if (!db_ && !Open(err)) return false; + PGresult* r = PQexec(db_, "SELECT method,count,bytes_in,bytes_out,latency_us_sum" + " FROM molo_stats ORDER BY method"); + if (!r || PQresultStatus(r) != PGRES_TUPLES_OK) { + err = r ? PQresultErrorMessage(r) : "pg query failed"; + if (r) PQclear(r); + return false; + } + std::ostringstream o; + if (as_json) o << "["; + for (int i = 0; i < PQntuples(r); ++i) { + const char* m = PQgetvalue(r, i, 0); + long long c = std::atoll(PQgetvalue(r, i, 1)), bi = std::atoll(PQgetvalue(r, i, 2)), + bo = std::atoll(PQgetvalue(r, i, 3)), ls = std::atoll(PQgetvalue(r, i, 4)); + double avg = c ? (double)ls / (double)c / 1000.0 : 0.0; + if (as_json) { + if (i) o << ","; + o << "{\"method\":\"" << m << "\",\"count\":" << c << ",\"kb_in\":" << (bi / 1024) + << ",\"kb_out\":" << (bo / 1024) << ",\"avg_ms\":" << avg << "}"; + } else { + char b[256]; + std::snprintf(b, sizeof b, "%-8s count=%lld kB_in=%lld kB_out=%lld avg_ms=%.2f\n", m, c, + bi / 1024, bo / 1024, avg); + o << b; + } + } + PQclear(r); + if (as_json) o << "]"; + out = o.str(); + return true; + } + + private: + std::string url_; + PGconn* db_ = nullptr; +}; + +} // namespace + +std::unique_ptr MakeStatsStore(const std::string& database_url) { + std::string u = database_url.empty() ? DatabaseUrl() : database_url; + if (u.rfind("postgres", 0) == 0 || u.rfind("postgresql", 0) == 0) { + // Strip scheme prefix variants libpq may not want; pass through as-is otherwise. + return std::make_unique(u); + } + std::string path = u; + const std::string pre = "sqlite://"; + if (path.rfind(pre, 0) == 0) path = path.substr(pre.size()); + if (path.empty()) path = "system.db"; + return std::make_unique(path); +} + +StatsCollector& StatsCollector::Global() { + static StatsCollector g; + return g; +} + +void StatsCollector::Note(const std::string& method, std::uint64_t bytes_in, + std::uint64_t bytes_out, std::uint64_t latency_us) { + if (!enabled_.load(std::memory_order_relaxed)) return; + std::lock_guard l(mu_); + MethodStat& s = pending_[method]; + s.count++; + s.bytes_in += bytes_in; + s.bytes_out += bytes_out; + s.latency_us_sum += latency_us; + MethodStat& t = pending_["_total"]; + t.count++; + t.bytes_in += bytes_in; + t.bytes_out += bytes_out; + t.latency_us_sum += latency_us; +} + +bool StatsCollector::Flush(std::string& err) { + std::map snap; + { + std::lock_guard l(mu_); + if (!store_) return true; + if (pending_.empty()) return store_->Open(err); + snap.swap(pending_); + } + StatsSnapshot s; + s.by_method = std::move(snap); + s.streams = 0; + for (auto& [m, st] : s.by_method) { + if (m != "_total") s.streams += st.count; + } + return store_->Flush(s, err); +} + +bool StatsCollector::Query(std::string& out, bool as_json, std::string& err) { + if (!Flush(err)) return false; + std::lock_guard l(mu_); + if (!store_) { + err = "stats store not configured"; + return false; + } + return store_->Query(out, as_json, err); +} + +} // namespace molo::realtime::expose diff --git a/src/realtime/expose/stats.h b/src/realtime/expose/stats.h new file mode 100644 index 0000000..9a77433 --- /dev/null +++ b/src/realtime/expose/stats.h @@ -0,0 +1,66 @@ +#pragma once +// Spec: reverse.md section 9 (observability), zero-cost stats for +// View/stream: kB up/down, per-method counts, avg latency, batched to +// SQLite (default) or Postgres (DATABASE_URL). OO: StatsCollector owns an +// atomic in-memory accumulator + a pluggable Store backend; flush is +// explicit (timer-driven by caller) inside one transaction. +#include +#include +#include +#include +#include +#include +#include + +namespace molo::realtime::expose { + +struct MethodStat { + std::uint64_t count = 0; + std::uint64_t bytes_in = 0; + std::uint64_t bytes_out = 0; + std::uint64_t latency_us_sum = 0; +}; + +struct StatsSnapshot { + std::map by_method; // + "_total" roll-up + std::uint64_t streams = 0; +}; + +// Storage backend interface (SQLite default, Postgres optional). +class StatsStore { + public: + virtual ~StatsStore() = default; + virtual bool Open(std::string& err) = 0; + // One transaction per Flush call. + virtual bool Flush(const StatsSnapshot& snap, std::string& err) = 0; + virtual bool Query(std::string& text_out, bool as_json, std::string& err) = 0; +}; + +std::unique_ptr MakeStatsStore(const std::string& database_url); + +// Zero-cost when idle: Note() is lock-free atomics only. Flush() swaps +// the accumulator under a short mutex and writes one transaction. +class StatsCollector { + public: + static StatsCollector& Global(); + void Configure(std::unique_ptr store) { + std::lock_guard l(mu_); + store_ = std::move(store); + } + bool enabled() const { return enabled_.load(std::memory_order_relaxed); } + void set_enabled(bool v) { enabled_.store(v, std::memory_order_relaxed); } + + void Note(const std::string& method, std::uint64_t bytes_in, + std::uint64_t bytes_out, std::uint64_t latency_us); + // Swap accumulator and persist in one transaction. No-op without a store. + bool Flush(std::string& err); + bool Query(std::string& out, bool as_json, std::string& err); + + private: + std::atomic enabled_{true}; + std::mutex mu_; + std::unique_ptr store_; + std::map pending_; +}; + +} // namespace molo::realtime::expose diff --git a/src/realtime/expose/tunnel.cpp b/src/realtime/expose/tunnel.cpp new file mode 100644 index 0000000..2ea5086 --- /dev/null +++ b/src/realtime/expose/tunnel.cpp @@ -0,0 +1,125 @@ +#include "realtime/expose/tunnel.h" +#include +#include +#include "realtime/shared/sha1.h" // sha1 available; sha256 emulated via double-round below is wrong: +// use OpenSSL-free portable sha256 implemented here. +#include +#include +namespace molo::realtime::expose { +namespace { +// Minimal public-domain SHA-256. +constexpr uint32_t K[64]={0x428a2f98,0x71374491,0xb5c0fbcf,0xe9b5dba5,0x3956c25b,0x59f111f1,0x923f82a4,0xab1c5ed5,0xd807aa98,0x12835b01,0x243185be,0x550c7dc3,0x72be5d74,0x80deb1fe,0x9bdc06a7,0xc19bf174,0xe49b69c1,0xefbe4786,0x0fc19dc6,0x240ca1cc,0x2de92c6f,0x4a7484aa,0x5cb0a9dc,0x76f988da,0x983e5152,0xa831c66d,0xb00327c8,0xbf597fc7,0xc6e00bf3,0xd5a79147,0x06ca6351,0x14292967,0x27b70a85,0x2e1b2138,0x4d2c6dfc,0x53380d13,0x650a7354,0x766a0abb,0x81c2c92e,0x92722c85,0xa2bfe8a1,0xa81a664b,0xc24b8b70,0xc76c51a3,0xd192e819,0xd6990624,0xf40e3585,0x106aa070,0x19a4c116,0x1e376c08,0x2748774c,0x34b0bcb5,0x391c0cb3,0x4ed8aa4a,0x5b9cca4f,0x682e6ff3,0x748f82ee,0x78a5636f,0x84c87814,0x8cc70208,0x90befffa,0xa4506ceb,0xbef9a3f7,0xc67178f2}; +#define RR(x,n)(((x)>>(n))|((x)<<(32-(n)))) +std::string Sha256(const std::string& m){ + uint32_t h[8]={0x6a09e667,0xbb67ae85,0x3c6e00fb,0xa54ff53a,0x510e527f,0x9b05688c,0x1f83d9ab,0x5be0cd19}; + std::string msg=m; uint64_t bitlen=(uint64_t)m.size()*8; + msg+=(char)0x80; while(msg.size()%64!=56)msg+=(char)0; + for(int i=7;i>=0;--i)msg+=(char)(bitlen>>(i*8)); + for(size_t off=0;off>3); + uint32_t s1=RR(w[i-2],17)^RR(w[i-2],19)^(w[i-2]>>10);w[i]=w[i-16]+s0+w[i-7]+s1;} + uint32_t a=h[0],b=h[1],c=h[2],d=h[3],e=h[4],f=h[5],g=h[6],hh=h[7]; + for(int i=0;i<64;++i){uint32_t S1=RR(e,6)^RR(e,11)^RR(e,25);uint32_t ch=(e&f)^(~e&g); + uint32_t t1=hh+S1+ch+K[i]+w[i];uint32_t S0=RR(a,2)^RR(a,13)^RR(a,22); + uint32_t maj=(a&b)^(a&c)^(b&c);uint32_t t2=S0+maj; + hh=g;g=f;f=e;e=d+t1;d=c;c=b;b=a;a=t1+t2;} + h[0]+=a;h[1]+=b;h[2]+=c;h[3]+=d;h[4]+=e;h[5]+=f;h[6]+=g;h[7]+=hh; + } + char out[65]; for(int i=0;i<8;++i)snprintf(out+2*i,3,"%02x",0); + std::string r; char buf[9]; for(int i=0;i<8;++i){snprintf(buf,9,"%08x",h[i]);r+=buf;} + return r; +} +} +std::string Sha256Hex(const std::string& s){return Sha256(s);} +bool ClaimRegistry::KeyEq(const std::string& a, const std::string& b){ + if(a.size()!=b.size())return false; + unsigned d=0; for(size_t i=0;i& out,std::string& detail){ + for(auto r:kReserved) if(SegmentPrefix(r,claim.path)){detail="reserved prefix";return RegisterResult::kReserved;} + std::lock_guard l(mu_); + for(auto&[id,t]:by_id_){ + if(!t->live||t->takeover_draining)continue; + bool same_host = claim.host.empty()||t->claim.host.empty()||claim.host==t->claim.host; + if(same_host&&(SegmentPrefix(t->claim.path,claim.path)||SegmentPrefix(claim.path,t->claim.path))){ + auto it=key_by_claim_.find(claim.str()); + bool same_key = it!=key_by_claim_.end()&&(KeyEq(key_sha,it->second)||KeyEq(key_sha,t->key_sha256)); + if(takeover&&same_key){t->takeover_draining=true;t->live=false;break;} + detail="claim held by "+t->id; return RegisterResult::kConflict; + } + } + auto it=key_by_claim_.find(claim.str()); + if(it!=key_by_claim_.end()&&!KeyEq(it->second,key_sha)){ + auto pi=prev_key_by_claim_.find(claim.str()); + auto pu=prev_key_until_.find(claim.str()); + bool drained = pi!=prev_key_by_claim_.end()&&pu!=prev_key_until_.end()&& + std::chrono::steady_clock::now()second&&KeyEq(pi->second,key_sha); + if(!drained){detail="bad key";return RegisterResult::kAuthFail;} + } + if((int)by_id_.size()>=kMaxTunnels){detail="tunnel cap";return RegisterResult::kConflict;} + int perkey=0; for(auto&[id,t]:by_id_)if(t->live&&KeyEq(t->key_sha256,key_sha))++perkey; + if(perkey>=kMaxTunnelsPerKey){detail="per-key tunnel cap";return RegisterResult::kConflict;} + if(it==key_by_claim_.end())key_by_claim_[claim.str()]=key_sha; + auto t=std::make_shared(); + t->id=NewTunnelId();t->claim=claim;t->upstreams=std::move(ups);t->key_sha256=key_sha; + t->server_id=server_id_;t->connected_at=t->last_ping=std::chrono::steady_clock::now(); + by_id_[t->id]=t;out=t;return RegisterResult::kOk; +} +bool ClaimRegistry::Release(const std::string& tid){std::lock_guard l(mu_);auto it=by_id_.find(tid);if(it==by_id_.end())return false;it->second->live=false;by_id_.erase(it);return true;} +std::shared_ptr ClaimRegistry::Find(const std::string& host,const std::string& path) const{ + std::lock_guard l(mu_); + std::string h=host; for(auto&c:h)c=tolower((unsigned char)c); + while(h.size()>1&&h.back()=='.')h.pop_back(); + std::shared_ptr best=nullptr; + for(auto&[id,t]:by_id_){ + if(!t->live)continue; + if(!t->claim.host.empty()&&t->claim.host!=h)continue; + if(!SegmentPrefix(t->claim.path,path)&&!(t->claim.whole_host))continue; + // whole-host claim on another host never matches (already filtered); + // path claims match any host (wildcard) per spec section 6. + if(!best||t->claim.path.size()>best->claim.path.size()|| + (t->claim.path.size()==best->claim.path.size()&& + !t->claim.host.empty()&&best->claim.host.empty()))best=t; + } + return best; +} +std::shared_ptr ClaimRegistry::Get(const std::string& tid) const{ + std::lock_guard l(mu_);auto it=by_id_.find(tid);return it==by_id_.end()?nullptr:it->second; +} +std::vector> ClaimRegistry::List() const{ + std::lock_guard l(mu_);std::vector> v; + for(auto&[id,t]:by_id_)if(t->live)v.push_back(t);return v; +} +bool ClaimRegistry::CheckKey(const Claim& c,const std::string& k) const{ + std::lock_guard l(mu_);auto it=key_by_claim_.find(c.str());if(it!=key_by_claim_.end()&&KeyEq(it->second,k))return true; + auto pi=prev_key_by_claim_.find(c.str()); + auto pu=prev_key_until_.find(c.str()); + return pi!=prev_key_by_claim_.end()&&pu!=prev_key_until_.end()&& + std::chrono::steady_clock::now()second&&KeyEq(pi->second,k); +} +void ClaimRegistry::RotateKey(const Claim& c,const std::string& n){ + std::lock_guard l(mu_); + auto it=key_by_claim_.find(c.str()); + if(it!=key_by_claim_.end()){prev_key_by_claim_[c.str()]=it->second; + prev_key_until_[c.str()]=std::chrono::steady_clock::now()+std::chrono::minutes(5);} + key_by_claim_[c.str()]=n; +} +} // namespace molo::realtime::expose diff --git a/src/realtime/expose/tunnel.h b/src/realtime/expose/tunnel.h new file mode 100644 index 0000000..cbb242e --- /dev/null +++ b/src/realtime/expose/tunnel.h @@ -0,0 +1,81 @@ +#pragma once +// Spec: reverse.md sections 4,6,9,10. In-memory claim registry + tunnel state. +// OO: ClaimRegistry owns Tunnel objects; each Tunnel owns its UpstreamTable. +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "realtime/expose/claim.h" +#include "realtime/expose/upstream.h" + +namespace molo::realtime::expose { + +inline constexpr const char* kReserved[] = {"/_molo", "/healthz", "/readyz", + "/startupz", "/metrics", "/csp-report"}; + +inline constexpr int kMaxTunnels = 64; +inline constexpr int kMaxTunnelsPerKey = 16; +inline constexpr int kMaxConnsPerIp = 128; +inline constexpr int kPingIntervalSec = 25; +inline constexpr int kPingAckTimeoutSec = 20; +inline constexpr int kClaimHoldSec = 30; +inline constexpr int kTakeoverDrainSec = 30; + +struct Tunnel { + std::string id; + Claim claim; + UpstreamTable upstreams; + std::string key_sha256; // hex sha256, never the key + std::string server_id; + std::chrono::steady_clock::time_point connected_at; + std::chrono::steady_clock::time_point last_ping; + bool live = true; + std::uint64_t streams_total = 0; + std::uint64_t bytes_in = 0, bytes_out = 0; + bool takeover_draining = false; + // Per-stream accounting (reverse.md §9): kB up/down + stream count. + void NoteStream(std::uint64_t in, std::uint64_t out) { + ++streams_total; + bytes_in += in; + bytes_out += out; + } +}; + +enum class RegisterResult { kOk, kBadClaim, kConflict, kReserved, kAuthFail }; + +// Thread-safe in-memory registry (v1: no persistence, pinned to process). +class ClaimRegistry { + public: + explicit ClaimRegistry(std::string server_id); + RegisterResult Register(const Claim& claim, const std::string& key_sha256, + UpstreamTable ups, bool takeover, + std::shared_ptr& out, std::string& detail); + bool Release(const std::string& tunnel_id); + std::shared_ptr Find(const std::string& host, const std::string& path) const; + std::shared_ptr Get(const std::string& tunnel_id) const; + std::vector> List() const; + bool CheckKey(const Claim& claim, const std::string& key_sha256) const; + void RotateKey(const Claim& claim, const std::string& new_sha256); + static bool KeyEq(const std::string& a, const std::string& b); // constant-time + static std::string KeyFingerprint(const std::string& key_sha256); + const std::string& server_id() const { return server_id_; } + private: + static bool SegmentPrefix(const std::string& a, const std::string& b); + std::string server_id_; + mutable std::mutex mu_; + std::map> by_id_; // tunnel id -> tunnel + std::map key_by_claim_; // claim str -> key sha + std::map prev_key_by_claim_; // rotated-out key, 5min drain + std::map prev_key_until_; +}; + +std::string Sha256Hex(const std::string& s); +std::string NewTunnelId(); + +} // namespace molo::realtime::expose diff --git a/src/realtime/expose/upstream.cpp b/src/realtime/expose/upstream.cpp new file mode 100644 index 0000000..b7669d2 --- /dev/null +++ b/src/realtime/expose/upstream.cpp @@ -0,0 +1,69 @@ +#include "realtime/expose/upstream.h" +namespace molo::realtime::expose { +namespace { +UpstreamScheme SchemeOf(const std::string& s){ + if(s=="tcp")return UpstreamScheme::kTcp; if(s=="https")return UpstreamScheme::kHttps; + if(s=="ws")return UpstreamScheme::kWs; if(s=="wss")return UpstreamScheme::kWss; + return UpstreamScheme::kHttp; +} +int DefaultPort(UpstreamScheme s){ + switch(s){case UpstreamScheme::kHttps:case UpstreamScheme::kWss:return 443; + case UpstreamScheme::kTcp:return 0; default:return 80;} +} +bool IsDigits(const std::string& s){if(s.empty())return false;for(char c:s)if(c<'0'||c>'9')return false;return true;} +bool SegBoundary(const std::string& path,const std::string& pfx){ + if(pfx=="/")return true; + if(path.size() ParseUpstream(const std::string& raw, std::string& reason){ + std::string s=raw, prefix="/"; + auto eq=s.rfind('='); + // '=' only a selector if what follows starts with '/' + if(eq!=std::string::npos&&eq+1best->prefix.size())best=&e; } + return best; +} +bool UpstreamHostAllowed(const std::string& host, bool allow_public){ + if(allow_public)return true; + std::string h=host; + if(h=="localhost"||h=="127.0.0.1"||h=="::1"||h=="::ffff:127.0.0.1")return true; + if(h.rfind("127.",0)==0)return true; + if(h.rfind("10.",0)==0)return true; + if(h.rfind("192.168.",0)==0)return true; + if(h.rfind("172.",0)==0){ + int n=0; size_t d=h.find('.',4); + try{n=std::stoi(h.substr(4,d==std::string::npos?std::string::npos:d-4));}catch(...){return false;} + if(n>=16&&n<=31)return true; + } + if(h.rfind("169.254.",0)==0||h.rfind("fe80",0)==0||h.rfind("FE80",0)==0)return true; + return false; +} +} // namespace molo::realtime::expose diff --git a/src/realtime/expose/upstream.h b/src/realtime/expose/upstream.h new file mode 100644 index 0000000..bec64d6 --- /dev/null +++ b/src/realtime/expose/upstream.h @@ -0,0 +1,38 @@ +#pragma once +// Spec: reverse.md sections 3,5. Upstream entry + longest-prefix routing. +#include +#include +#include + +namespace molo::realtime::expose { + +enum class UpstreamScheme { kTcp, kHttp, kHttps, kWs, kWss }; + +struct UpstreamEntry { + std::string raw; + UpstreamScheme scheme = UpstreamScheme::kHttp; + std::string host; + int port = 80; + std::string path; // origin path + std::string prefix; // claim-subpath selector, "/" = default +}; + +// Parses "[scheme://]host[:port][/path][=prefix]". Bare number = 127.0.0.1:port. +std::optional ParseUpstream(const std::string& raw, std::string& reason); + +// SSRF guard (reverse.md section 7): upstream host must be loopback / +// RFC1918 / link-local unless allow_public is set. Applies per entry. +bool UpstreamHostAllowed(const std::string& host, bool allow_public); + +// OO routing table: longest prefix wins, ties rejected at insert. +class UpstreamTable { + public: + bool Add(UpstreamEntry e, std::string& reason); + const UpstreamEntry* Route(const std::string& path) const; + const std::vector& entries() const { return entries_; } + std::size_t size() const { return entries_.size(); } + private: + std::vector entries_; +}; + +} // namespace molo::realtime::expose diff --git a/src/realtime/realtime.mk b/src/realtime/realtime.mk index c9fd7e2..a308080 100644 --- a/src/realtime/realtime.mk +++ b/src/realtime/realtime.mk @@ -54,7 +54,13 @@ REALTIME_SRCS := \ src/realtime/webdav/server/lock_manager.cpp \ src/realtime/webdav/server/quota_guard.cpp \ src/realtime/webdav/server/dav_view.cpp \ - src/realtime/webdav/client/webdav_client.cpp + src/realtime/webdav/client/webdav_client.cpp \ + src/realtime/expose/claim.cpp \ + src/realtime/expose/upstream.cpp \ + src/realtime/expose/tunnel.cpp \ + src/realtime/expose/hello.cpp \ + src/realtime/expose/proxy.cpp \ + src/realtime/expose/stats.cpp REALTIME_OBJS := $(patsubst src/%.cpp,build/obj/%.o,$(REALTIME_SRCS)) REALTIME_LIB := build/lib/librealtime.a @@ -71,7 +77,8 @@ REALTIME_TESTS := \ build/test/realtime/socketio/chat_test \ build/test/realtime/dav/files_test \ build/test/realtime/shared/foundation_test \ - build/test/realtime/shared/robustness_test + build/test/realtime/shared/robustness_test \ + build/test/realtime/expose/expose_test $(REALTIME_LIB): $(REALTIME_OBJS) mkdir -p $(dir $@) @@ -83,8 +90,8 @@ build/obj/realtime/%.o: src/realtime/%.cpp build/test/realtime/%_test: tests/src/realtime/%_test.cpp $(REALTIME_LIB) mkdir -p $(dir $@) - $(CXX) $(CXXFLAGS) -Isrc -Isrc/realtime -Itests/src/realtime $< $(REALTIME_LIB) -o $@ + $(CXX) $(CXXFLAGS) -Isrc -Isrc/realtime -Itests/src/realtime $< $(REALTIME_LIB) -lsqlite3 -l:libpq.so.5 -o $@ build/test/realtime/%: tests/src/realtime/%.cpp $(REALTIME_LIB) mkdir -p $(dir $@) - $(CXX) $(CXXFLAGS) -Isrc -Isrc/realtime -Itests/src/realtime $< $(REALTIME_LIB) -o $@ + $(CXX) $(CXXFLAGS) -Isrc -Isrc/realtime -Itests/src/realtime $< $(REALTIME_LIB) -lsqlite3 -l:libpq.so.5 -o $@ diff --git a/tests/src/realtime/expose/expose_test.cpp b/tests/src/realtime/expose/expose_test.cpp new file mode 100644 index 0000000..ebe4239 --- /dev/null +++ b/tests/src/realtime/expose/expose_test.cpp @@ -0,0 +1,98 @@ +#include +#include +#include +#include +#include "realtime/expose/claim.h" +#include "realtime/expose/hello.h" +#include "realtime/expose/tunnel.h" +#include "realtime/expose/proxy.h" +#include "realtime/expose/stats.h" +#include +#include "realtime/expose/upstream.h" +using namespace molo::realtime::expose; +int main() { + std::string r; + // claim grammar + assert(ParseClaim("/hook1", r) && ParseClaim("example.com/hook1", r)); + assert(ParseClaim("example.com", r)->whole_host); + assert(!ParseClaim("/../x", r)); + // upstream table: longest prefix wins, concurrent entries + UpstreamTable t; + UpstreamEntry a, b; + assert(ParseUpstream("http://127.0.0.1:3001=/api", r)); + auto e1 = ParseUpstream("http://127.0.0.1:3001=/api", r); + auto e2 = ParseUpstream("http://127.0.0.1:3002=/", r); + assert(e1 && e2 && t.Add(*e1, r) && t.Add(*e2, r)); + assert(t.Route("/api/u")->port == 3001 && t.Route("/other")->port == 3002); + // hello v1+v2 + std::string e; + auto h2 = ParseHello(R"({"v":2,"claim":"/a","upstreams":[{"upstream":"http://127.0.0.1:1","prefix":"/"}],"agent":"t"})", "", e); + auto h1 = ParseHello(R"({"v":1,"claim":"/a","upstream":"8080","agent":"t"})", "", e); + assert(h2 && h1 && h1->upstreams.size() == 1); + // registry: auth, conflict, takeover, concurrent register + ClaimRegistry reg("s1"); + auto c = ParseClaim("/app", r); + UpstreamTable u0; + std::shared_ptr out; std::string d; + assert(reg.Register(*c, Sha256Hex("k1"), std::move(u0), false, out, d) == RegisterResult::kOk); + UpstreamTable u1; + std::shared_ptr o2; + assert(reg.Register(*c, Sha256Hex("k1"), std::move(u1), false, o2, d) == RegisterResult::kConflict); + UpstreamTable u2; + std::shared_ptr o3; + assert(reg.Register(*c, Sha256Hex("k1"), std::move(u2), true, o3, d) == RegisterResult::kOk); + assert(reg.Find("", "/app/x") && reg.Find("", "/app/x")->id == o3->id); + // concurrent Find/List under threads + std::vector th; + for (int i = 0; i < 8; ++i) th.emplace_back([&]{ for (int j = 0; j < 200; ++j) { reg.Find("", "/app"); reg.List(); } }); + for (auto& t : th) t.join(); + + // proxy glue: strip prefix, hop-by-hop, stream ids, per-IP cap + assert(StripClaimPrefix("/app", false, "/app/x") == "/x"); + assert(StripClaimPrefix("/app", false, "/app") == "/"); + assert(StripClaimPrefix("/", false, "/a") == "/a"); + assert(StripClaimPrefix("/app", true, "/app/x") == "/app/x"); + assert(IsHopByHop("Connection") && IsHopByHop("Transfer-Encoding") && !IsHopByHop("X-Custom")); + assert(ForwardedFor("", "1.2.3.4") == "1.2.3.4"); + assert(ForwardedFor("a", "b") == "a, b"); + assert(kH2RefusedStream == 0x7 && kH2Cancel == 0x8); + std::atomic nx{3}; + assert(NextProxyStreamId(nx) == 3 && NextProxyStreamId(nx) == 5); + PerIpCounter ipc; for (int i = 0; i < 128; ++i) assert(ipc.Admit("9.9.9.9")); + assert(!ipc.Admit("9.9.9.9")); ipc.Release("9.9.9.9"); assert(ipc.Admit("9.9.9.9")); + // concurrent streams across N upstreams: route from threads + { + UpstreamTable ct; + std::string rr; + ct.Add(*ParseUpstream("http://127.0.0.1:3001=/api", rr), rr); + ct.Add(*ParseUpstream("http://127.0.0.1:3002=/static", rr), rr); + ct.Add(*ParseUpstream("http://127.0.0.1:3003=/", rr), rr); + std::atomic ok{0}; + std::vector w; + for (int i = 0; i < 16; ++i) w.emplace_back([&, i]{ + const char* p[3] = {"/api/x", "/static/y", "/other"}; + auto* e = ct.Route(p[i % 3]); + if (e) ok.fetch_add(1); + std::atomic sn{3u + (uint32_t)(i * 2)}; + (void)NextProxyStreamId(sn); + }); + for (auto& th2 : w) th2.join(); + assert(ok == 16); + } + // stats: Note/Flush/Query sqlite round-trip (reverse.md 9) + { + std::string db = "/tmp/molo_stats_test.db"; + std::remove(db.c_str()); + StatsCollector sc; + sc.Configure(MakeStatsStore("sqlite://" + db)); + sc.Note("GET", 2048, 4096, 2000); + std::string err, out; + assert(sc.Flush(err)); + assert(sc.Query(out, false, err)); + assert(out.find("GET") != std::string::npos && out.find("count=1") != std::string::npos); + assert(sc.Query(out, true, err) && out.front() == '['); + std::remove(db.c_str()); + } + printf("expose_test PASS\n"); + return 0; +}