Compare commits
2 Commits
d6a93506e8
...
c526c55c36
| Author | SHA1 | Date | |
|---|---|---|---|
| c526c55c36 | |||
| e2a58f3455 |
71
README.md
Normal file
71
README.md
Normal file
@@ -0,0 +1,71 @@
|
|||||||
|
# rust-link
|
||||||
|
|
||||||
|
The **sidecar** half of the Runic Gateway bridge for [Rust](https://rust.facepunch.com/). It
|
||||||
|
terminates the loopback link from a Rust server's Oxide bridge plugin and exposes the WebSocket +
|
||||||
|
REST surface the website consumes.
|
||||||
|
|
||||||
|
It is the mirror of [`RunicGateway/link`](https://gitea.whitlocktech.com/RunicGateway/link), which
|
||||||
|
does the same job for Ultima Online, and it keeps that bridge's central invariant unchanged:
|
||||||
|
|
||||||
|
> **The game server is never reachable from the website.** The plugin dials *out* to this process;
|
||||||
|
> this process owns the listener. Only the sidecar is exposed, and only the website's backend talks
|
||||||
|
> to it.
|
||||||
|
|
||||||
|
```
|
||||||
|
Rust server + Oxide (RunicGateway/Rust-Plugins, C#)
|
||||||
|
│ loopback TCP 127.0.0.1:7799, newline-delimited JSON, bidirectional
|
||||||
|
│ the PLUGIN dials out (the game opens no listening port for us)
|
||||||
|
▼
|
||||||
|
rust-link sidecar (this repo, Rust) ← the only network-facing bridge component
|
||||||
|
│ WebSocket (live feed) + REST (point-in-time reads), bearer-token auth
|
||||||
|
▼
|
||||||
|
website backend + module-rust (RunicGateway/Module-Rust, Node)
|
||||||
|
```
|
||||||
|
|
||||||
|
## One server, one sidecar
|
||||||
|
|
||||||
|
This binary serves **exactly one** Rust game server. A community running six servers runs six
|
||||||
|
pairs, each with its own port, database and token; `module-rust` holds six clients and the website
|
||||||
|
core never learns there is more than one. Nothing here is multiplexed, and nothing here should
|
||||||
|
become multiplexed.
|
||||||
|
|
||||||
|
## Build and run
|
||||||
|
|
||||||
|
```bash
|
||||||
|
cd sidecar
|
||||||
|
cargo build --release # → target/release/rust-link-sidecar
|
||||||
|
cargo run # info logging; writes sidecar.toml (with a generated token) on first run
|
||||||
|
RUST_LOG=debug cargo run # verbose, including heartbeats
|
||||||
|
```
|
||||||
|
|
||||||
|
Everything is configured from `sidecar.toml` — nothing is compiled into the binary. **Auth is
|
||||||
|
always on**: a blank token is generated and written back on first start, so there is no state in
|
||||||
|
which this process listens without one. `--print-config` resolves the configuration and prints it
|
||||||
|
as JSON, which is how an installer reads the token back without scraping a log.
|
||||||
|
|
||||||
|
See [`sidecar/README.md`](sidecar/README.md) for the configuration reference and the endpoint list.
|
||||||
|
|
||||||
|
## The protocol is a contract
|
||||||
|
|
||||||
|
The loopback JSON protocol (plugin ↔ sidecar) and this sidecar's HTTP/WS API (sidecar ↔ website)
|
||||||
|
are **versioned compatibility contracts**, not build dependencies. `PROTOCOL_VERSION` lives in
|
||||||
|
[`sidecar/src/main.rs`](sidecar/src/main.rs); every response carries `X-RustLink-Version`, and a
|
||||||
|
client that declares a different one is refused `409` rather than served something it will
|
||||||
|
mis-parse.
|
||||||
|
|
||||||
|
A version is declared in **three** places and they must agree:
|
||||||
|
|
||||||
|
| Where | Repo |
|
||||||
|
|---|---|
|
||||||
|
| `PROTOCOL_VERSION` | this repo |
|
||||||
|
| `overlay.toml` | [`RunicGateway/Rust-Plugins`](https://gitea.whitlocktech.com/RunicGateway/Rust-Plugins) |
|
||||||
|
| `module.json` | [`RunicGateway/Module-Rust`](https://gitea.whitlocktech.com/RunicGateway/Module-Rust) |
|
||||||
|
|
||||||
|
The canonical spec is
|
||||||
|
[`docs/rust-link/`](https://gitea.whitlocktech.com/RunicGateway/docs/src/branch/main/rust-link). If
|
||||||
|
you add or change an event or a command, update all three repos **and** the spec in the same
|
||||||
|
change.
|
||||||
|
|
||||||
|
## Licence
|
||||||
|
|
||||||
|
GPL-3.0-or-later. See [LICENSE.md](LICENSE.md).
|
||||||
2277
sidecar/Cargo.lock
generated
Normal file
2277
sidecar/Cargo.lock
generated
Normal file
File diff suppressed because it is too large
Load Diff
22
sidecar/Cargo.toml
Normal file
22
sidecar/Cargo.toml
Normal file
@@ -0,0 +1,22 @@
|
|||||||
|
[package]
|
||||||
|
name = "rust-link-sidecar"
|
||||||
|
version = "0.1.0"
|
||||||
|
edition = "2021"
|
||||||
|
license = "GPL-3.0-or-later"
|
||||||
|
description = "Rust sidecar for the rust-link bridge: terminates the loopback link to a Rust/Oxide game server and exposes WebSocket + REST to the website."
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "io-util", "sync", "time", "signal"] }
|
||||||
|
serde = { version = "1", features = ["derive"] }
|
||||||
|
serde_json = "1"
|
||||||
|
tracing = "0.1"
|
||||||
|
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
|
||||||
|
anyhow = "1"
|
||||||
|
axum = { version = "0.7", features = ["ws"] }
|
||||||
|
sqlx = { version = "0.8", features = ["runtime-tokio", "sqlite"] }
|
||||||
|
toml = "0.8"
|
||||||
|
getrandom = "0.2"
|
||||||
|
chrono = { version = "0.4", default-features = false, features = ["std", "clock"] }
|
||||||
|
|
||||||
|
[profile.release]
|
||||||
|
opt-level = 2
|
||||||
92
sidecar/README.md
Normal file
92
sidecar/README.md
Normal file
@@ -0,0 +1,92 @@
|
|||||||
|
# rust-link-sidecar
|
||||||
|
|
||||||
|
Configuration reference and endpoint list. For what this component *is*, see the
|
||||||
|
[repo README](../README.md).
|
||||||
|
|
||||||
|
## Configuration
|
||||||
|
|
||||||
|
`sidecar.toml`, resolved in this order: `--config <PATH>`, else `$RUSTLINK_CONFIG`, else
|
||||||
|
`./sidecar.toml`. Environment variables override the file; the file overrides the defaults.
|
||||||
|
|
||||||
|
| Key | Env | Default | What it is |
|
||||||
|
|---|---|---|---|
|
||||||
|
| `[game].bind` | `RUSTLINK_GAME_BIND` | `127.0.0.1:7799` | Where the Oxide plugin dials in |
|
||||||
|
| `[game].server_id` | `RUSTLINK_SERVER_ID` | *(empty)* | Optional cross-check against the plugin's own `serverId` |
|
||||||
|
| `[web].bind` | `RUSTLINK_WEB_BIND` | `127.0.0.1:8090` | Where the website reaches this sidecar |
|
||||||
|
| `[web].auth_token` | `RUSTLINK_WEB_TOKEN` | *(generated)* | The shared secret the website presents |
|
||||||
|
| `[store].path` | `RUSTLINK_DB_PATH` | `rust-link.db` | SQLite file |
|
||||||
|
|
||||||
|
Two things about those defaults are load-bearing:
|
||||||
|
|
||||||
|
- **`[game].bind` is loopback, and there is no token on that link.** The plugin and the sidecar
|
||||||
|
share a host; `127.0.0.1` *is* the authentication. Binding it to a routable address puts an
|
||||||
|
unauthenticated command channel on the network.
|
||||||
|
- **A relative `[store].path` resolves against the directory holding `sidecar.toml`**, not the
|
||||||
|
working directory. A service manager's working directory must not decide where the database
|
||||||
|
lands — on Windows that can be `%SystemRoot%\System32`, or a silently redirected VirtualStore
|
||||||
|
copy.
|
||||||
|
|
||||||
|
`[game].server_id` is a **cross-check, not a second source of truth**. The plugin announces its own
|
||||||
|
`serverId` and that is the authority; when both are set and they disagree, the sidecar logs the
|
||||||
|
disagreement loudly and keeps the plugin's. Two game servers pointed at one sidecar by a copied
|
||||||
|
config is the mistake this catches, and it is silent in every other design.
|
||||||
|
|
||||||
|
### Reading the token back
|
||||||
|
|
||||||
|
```bash
|
||||||
|
rust-link-sidecar --print-config
|
||||||
|
```
|
||||||
|
|
||||||
|
Resolves the configuration exactly as a normal start would — writing the file and generating the
|
||||||
|
token if they are missing — and prints it as JSON on stdout, **including the token in clear text**.
|
||||||
|
That is the supported way for an installer to obtain it; the alternative is scraping a log.
|
||||||
|
|
||||||
|
## Endpoints
|
||||||
|
|
||||||
|
Everything except `/health` requires the token, as `Authorization: Bearer <t>`, `X-Api-Key: <t>`,
|
||||||
|
or `?token=<t>` (the last so browser WebSocket clients, which cannot set handshake headers, can
|
||||||
|
still authenticate). Every response carries `X-RustLink-Version`.
|
||||||
|
|
||||||
|
| Route | Backed by | Notes |
|
||||||
|
|---|---|---|
|
||||||
|
| `GET /health` | — | Unauthenticated, so monitoring can reach it |
|
||||||
|
| `GET /server` | store | The last `server.hello`. **`204` when the game has never connected** |
|
||||||
|
| `GET /events?kind=&limit=` | store | Newest first; `limit` clamped to 1–1000 |
|
||||||
|
| `GET /status` | plugin (RPC) | A live round trip. `503` with no plugin, `504` on no reply |
|
||||||
|
| `GET /ws` | broadcast | The live feed. Sends `ws.hello` on connect |
|
||||||
|
|
||||||
|
The split is the point: the store-backed reads answer while the game is off, which is what lets the
|
||||||
|
website render a server list during a wipe or a restart. `/status` is the one route that fails when
|
||||||
|
the game is down, because "what is it doing right now" has no stale answer worth giving.
|
||||||
|
|
||||||
|
`/server` answers `204`, not `200` with a null, when the game has never connected. "We have never
|
||||||
|
heard from this server" and "this server reports nothing" are different answers, and a client that
|
||||||
|
cannot tell them apart renders a server that does not exist.
|
||||||
|
|
||||||
|
## Protocol 1
|
||||||
|
|
||||||
|
Newline-delimited JSON over TCP, both directions. Outbound frames (plugin → sidecar) carry `kind`;
|
||||||
|
inbound frames (sidecar → plugin) carry `cmd`. Lines are capped at 1 MiB; an over-long line is
|
||||||
|
discarded and the connection stays up.
|
||||||
|
|
||||||
|
| Frame | Direction | Purpose |
|
||||||
|
|---|---|---|
|
||||||
|
| `server.hello` | plugin → sidecar | Sent on **every connect**, not once at server start — this process restarts independently of the game. Carries `serverId` and `bootId` |
|
||||||
|
| `ping` / `pong` | sidecar → plugin → sidecar | The heartbeat, every 30s. A `pong` is never persisted; it only moves `last_event` |
|
||||||
|
| `server.status` | sidecar → plugin → sidecar | The one request/reply verb, correlated by `reqId` |
|
||||||
|
|
||||||
|
`bootId` is how a game restart is told apart from a sidecar reconnect — the distinction the event
|
||||||
|
system's `reconcile` hangs off later.
|
||||||
|
|
||||||
|
**The RPC reply timeout (`rpc::REPLY_TIMEOUT`, 10s) is a ceiling every later command budget sits
|
||||||
|
under.** Core classifies a budget overrun as retryable unconditionally, because it cannot ask the
|
||||||
|
game while the action is still awaiting a socket. An action whose `budgetMs` exceeds this can never
|
||||||
|
report `retry: false`.
|
||||||
|
|
||||||
|
## Checks
|
||||||
|
|
||||||
|
```bash
|
||||||
|
cargo fmt --all -- --check
|
||||||
|
cargo clippy --all-targets -- -D warnings
|
||||||
|
cargo test
|
||||||
|
```
|
||||||
219
sidecar/src/app.rs
Normal file
219
sidecar/src/app.rs
Normal file
@@ -0,0 +1,219 @@
|
|||||||
|
//! The sidecar itself: everything that happens between "we have a config path" and "we were told
|
||||||
|
//! to stop".
|
||||||
|
//!
|
||||||
|
//! [`run`] is parameterised on the two things a supervisor cares about:
|
||||||
|
//!
|
||||||
|
//! - `ready` is called once the sidecar is actually up (listener bound, store open). A service
|
||||||
|
//! wrapper reports `Running` to its host there, so a config or bind failure surfaces as a *start*
|
||||||
|
//! failure rather than a service that reports Running and then dies.
|
||||||
|
//! - `shutdown` is whatever "stop" means on this host.
|
||||||
|
//!
|
||||||
|
//! Phase 1 runs in the foreground only; the parameters exist now so that adding a host's own
|
||||||
|
//! supervision later is a new module rather than a restructuring of this one.
|
||||||
|
|
||||||
|
use std::future::Future;
|
||||||
|
use std::sync::atomic::AtomicI64;
|
||||||
|
use std::sync::Arc;
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
|
use tokio::sync::{broadcast, mpsc};
|
||||||
|
use tracing::{info, warn};
|
||||||
|
|
||||||
|
use crate::{config, game, rpc, store, web};
|
||||||
|
|
||||||
|
/// How often the sidecar pings the plugin.
|
||||||
|
///
|
||||||
|
/// This is the only thing that moves `last_event` on a quiet server, and a Rust server with nobody
|
||||||
|
/// on it is very quiet. Without it, "the game has said nothing for six hours" would be
|
||||||
|
/// indistinguishable from "the link died six hours ago".
|
||||||
|
const HEARTBEAT: Duration = Duration::from_secs(30);
|
||||||
|
|
||||||
|
/// Runs the sidecar until `shutdown` resolves.
|
||||||
|
///
|
||||||
|
/// `config_path` is the `--config` argument, or `None` to resolve `$RUSTLINK_CONFIG` and the
|
||||||
|
/// default as usual.
|
||||||
|
pub async fn run<R, S>(config_path: Option<&str>, ready: R, shutdown: S) -> anyhow::Result<()>
|
||||||
|
where
|
||||||
|
R: FnOnce(),
|
||||||
|
S: Future<Output = ()>,
|
||||||
|
{
|
||||||
|
info!("rust-link sidecar starting");
|
||||||
|
|
||||||
|
let loaded = config::Config::load(config_path)?;
|
||||||
|
let cfg = loaded.cfg;
|
||||||
|
info!(
|
||||||
|
config = %loaded.path.display(),
|
||||||
|
game = %cfg.game.bind,
|
||||||
|
web = %cfg.web.bind,
|
||||||
|
db = %cfg.store.path,
|
||||||
|
auth = cfg.auth_required(),
|
||||||
|
"configuration loaded"
|
||||||
|
);
|
||||||
|
|
||||||
|
// Game link: events in, commands out.
|
||||||
|
let (event_tx, mut event_rx) = mpsc::unbounded_channel::<game::GameEvent>();
|
||||||
|
let (handle, _bound) = game::serve(&cfg.game.bind, event_tx).await?;
|
||||||
|
|
||||||
|
// Live feed: every game event fans out to all connected website WebSocket clients.
|
||||||
|
let (bcast_tx, _) = broadcast::channel::<String>(1024);
|
||||||
|
|
||||||
|
// Request/reply correlation for REST queries.
|
||||||
|
let rpc = rpc::Rpc::new();
|
||||||
|
|
||||||
|
// Durable store: event history and the server board.
|
||||||
|
let store = store::Store::open(&cfg.store.path).await?;
|
||||||
|
|
||||||
|
// Health/observability state.
|
||||||
|
let started = Instant::now();
|
||||||
|
let last_event = Arc::new(AtomicI64::new(0));
|
||||||
|
|
||||||
|
// Website-facing HTTP server.
|
||||||
|
let web_state = web::AppState {
|
||||||
|
events: bcast_tx.clone(),
|
||||||
|
game: handle.clone(),
|
||||||
|
rpc: rpc.clone(),
|
||||||
|
store: store.clone(),
|
||||||
|
token: Arc::new(cfg.web.auth_token.clone()),
|
||||||
|
started,
|
||||||
|
last_event: last_event.clone(),
|
||||||
|
};
|
||||||
|
let web_bind = cfg.web.bind.clone();
|
||||||
|
tokio::spawn(async move {
|
||||||
|
if let Err(e) = web::serve(&web_bind, web_state).await {
|
||||||
|
tracing::error!(error = %e, "web server exited");
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
// The heartbeat. Commands to a disconnected plugin are dropped rather than queued, so this is
|
||||||
|
// safe to fire whether or not anything is connected.
|
||||||
|
let ping_handle = handle.clone();
|
||||||
|
tokio::spawn(async move {
|
||||||
|
let mut tick = tokio::time::interval(HEARTBEAT);
|
||||||
|
// The first tick fires immediately, which would ping before the plugin has had a chance to
|
||||||
|
// dial in and log a dropped command on every start.
|
||||||
|
tick.tick().await;
|
||||||
|
loop {
|
||||||
|
tick.tick().await;
|
||||||
|
if ping_handle.is_connected().await {
|
||||||
|
ping_handle.send(r#"{"cmd":"ping"}"#.to_string()).await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
// The event loop. A line that correlates to a pending REST call is a reply — route it to the
|
||||||
|
// waiting caller and stop. Everything else is a live event: persist it, then broadcast it.
|
||||||
|
let configured_server_id = cfg.game.server_id.clone();
|
||||||
|
let feed_tx = bcast_tx.clone();
|
||||||
|
let route_rpc = rpc.clone();
|
||||||
|
let event_store = store.clone();
|
||||||
|
let last_event_ts = last_event.clone();
|
||||||
|
tokio::spawn(async move {
|
||||||
|
let mut total: u64 = 0;
|
||||||
|
|
||||||
|
while let Some(ev) = event_rx.recv().await {
|
||||||
|
// Any line from the plugin — a pong included — is a sign of life.
|
||||||
|
last_event_ts.store(now_ms(), std::sync::atomic::Ordering::Relaxed);
|
||||||
|
|
||||||
|
if route_rpc.try_route(&ev.value).await {
|
||||||
|
continue; // consumed as a reply
|
||||||
|
}
|
||||||
|
|
||||||
|
total += 1;
|
||||||
|
let line = ev.value.to_string();
|
||||||
|
|
||||||
|
match ev.kind.as_str() {
|
||||||
|
"server.hello" => {
|
||||||
|
check_server_id(&configured_server_id, &ev.value);
|
||||||
|
info!(kind = %ev.kind, n = total, "{}", line);
|
||||||
|
if let Err(e) = event_store
|
||||||
|
.put_server_state(frame_t(&ev.value), &line)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
warn!(error = %e, "server board write failed");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
"link.down" => info!("game link down"),
|
||||||
|
_ => tracing::debug!(kind = %ev.kind, n = total, "{}", line),
|
||||||
|
}
|
||||||
|
|
||||||
|
// `pong` and `link.down` are ephemeral: one is heartbeat chatter and the other is this
|
||||||
|
// process's own observation, not something the game said. Neither is history.
|
||||||
|
if ev.kind != "pong" && ev.kind != "link.down" {
|
||||||
|
if let Err(e) = event_store
|
||||||
|
.insert_event(frame_t(&ev.value), &ev.kind, &line)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
warn!(error = %e, "event persist failed");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Broadcast after persisting, so a client that reacts by reading the store cannot
|
||||||
|
// beat its own event there. `send` fails only when nobody is subscribed.
|
||||||
|
let _ = feed_tx.send(line);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
ready();
|
||||||
|
info!("rust-link sidecar ready");
|
||||||
|
|
||||||
|
shutdown.await;
|
||||||
|
info!("shutting down");
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The plugin's `serverId` is the authority; `[game].server_id` is a cross-check. A disagreement is
|
||||||
|
/// almost always two game servers pointed at one sidecar by a copied config, which is silent in
|
||||||
|
/// every other design and produces one server's history under another's name.
|
||||||
|
fn check_server_id(configured: &str, hello: &serde_json::Value) {
|
||||||
|
if configured.is_empty() {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
let announced = hello.get("serverId").and_then(|v| v.as_str()).unwrap_or("");
|
||||||
|
if !announced.is_empty() && announced != configured {
|
||||||
|
warn!(
|
||||||
|
configured,
|
||||||
|
announced,
|
||||||
|
"the connected plugin announces a different serverId than this sidecar is configured \
|
||||||
|
for; keeping the plugin's. Two servers sharing one sidecar is the usual cause."
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A frame's own timestamp, falling back to ours. The plugin stamps `t` at the moment the world was
|
||||||
|
/// read, which is earlier than the moment we saw the line and is the one worth keeping.
|
||||||
|
fn frame_t(value: &serde_json::Value) -> i64 {
|
||||||
|
value
|
||||||
|
.get("t")
|
||||||
|
.and_then(|v| v.as_i64())
|
||||||
|
.unwrap_or_else(now_ms)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn now_ms() -> i64 {
|
||||||
|
std::time::SystemTime::now()
|
||||||
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
|
.map(|d| d.as_millis() as i64)
|
||||||
|
.unwrap_or(0)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use serde_json::json;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn a_frames_own_timestamp_wins_over_ours() {
|
||||||
|
assert_eq!(
|
||||||
|
frame_t(&json!({"t": 1_700_000_000_000i64})),
|
||||||
|
1_700_000_000_000
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A frame with no `t` must still land with a usable timestamp rather than at the epoch, or
|
||||||
|
/// every un-stamped event sorts to the beginning of history forever.
|
||||||
|
#[test]
|
||||||
|
fn a_frame_without_a_timestamp_gets_one() {
|
||||||
|
assert!(frame_t(&json!({"kind": "x"})) > 1_600_000_000_000);
|
||||||
|
// A non-numeric `t` is a malformed frame, not a zero.
|
||||||
|
assert!(frame_t(&json!({"t": "nope"})) > 1_600_000_000_000);
|
||||||
|
}
|
||||||
|
}
|
||||||
170
sidecar/src/cli.rs
Normal file
170
sidecar/src/cli.rs
Normal file
@@ -0,0 +1,170 @@
|
|||||||
|
//! Command-line surface.
|
||||||
|
//!
|
||||||
|
//! The sidecar is configured by file and environment (see [`crate::config`]); this is deliberately
|
||||||
|
//! not a second configuration mechanism. It exists so the binary can be *driven by an installer*
|
||||||
|
//! rather than only by a human reading its logs:
|
||||||
|
//!
|
||||||
|
//! - `--print-config` resolves the configuration exactly as a normal start would — including
|
||||||
|
//! generating the auth token on first run — and prints it as JSON on stdout. That is the
|
||||||
|
//! supported way to obtain the token for the website's Admin -> Modules -> Rust form.
|
||||||
|
//! - `--config <PATH>` names the config file without having to export `$RUSTLINK_CONFIG`, so a
|
||||||
|
//! diagnostic run can point at an installed config from any working directory.
|
||||||
|
//!
|
||||||
|
//! Hand-rolled rather than pulled from a crate: four flags, no subcommands, no completions.
|
||||||
|
|
||||||
|
/// What this invocation should do. Everything except `Run` prints and exits.
|
||||||
|
#[derive(Debug, PartialEq, Eq)]
|
||||||
|
pub enum Mode {
|
||||||
|
/// Normal operation: bind the game listener and the web server.
|
||||||
|
Run,
|
||||||
|
/// Resolve config, print it as JSON, exit.
|
||||||
|
PrintConfig,
|
||||||
|
Help,
|
||||||
|
Version,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, PartialEq, Eq)]
|
||||||
|
pub struct Cli {
|
||||||
|
pub mode: Mode,
|
||||||
|
/// `--config <PATH>`, which outranks `$RUSTLINK_CONFIG`.
|
||||||
|
pub config: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
pub const USAGE: &str = "\
|
||||||
|
rust-link sidecar — bridges one Rust game server to the Runic Gateway website.
|
||||||
|
|
||||||
|
Usage: rust-link-sidecar [OPTIONS]
|
||||||
|
|
||||||
|
Options:
|
||||||
|
--print-config Resolve the configuration, print it as JSON, and exit.
|
||||||
|
Runs first-run setup like a normal start does: if the
|
||||||
|
config file is missing it is written, and a blank auth
|
||||||
|
token is generated and saved. The JSON CONTAINS THE
|
||||||
|
AUTH TOKEN in clear text.
|
||||||
|
--config <PATH> Path to sidecar.toml. Overrides $RUSTLINK_CONFIG;
|
||||||
|
defaults to ./sidecar.toml.
|
||||||
|
-V, --version Print the sidecar and protocol versions and exit.
|
||||||
|
-h, --help Print this help and exit.
|
||||||
|
|
||||||
|
Configuration lives in sidecar.toml; environment variables override the file:
|
||||||
|
RUSTLINK_CONFIG, RUSTLINK_GAME_BIND, RUSTLINK_WEB_BIND, RUSTLINK_WEB_TOKEN,
|
||||||
|
RUSTLINK_DB_PATH, RUSTLINK_SERVER_ID
|
||||||
|
";
|
||||||
|
|
||||||
|
/// Parses arguments **without** the program name.
|
||||||
|
///
|
||||||
|
/// Returns the message to print on stderr when the arguments are unusable; the caller exits `2`.
|
||||||
|
pub fn parse<I: IntoIterator<Item = String>>(args: I) -> Result<Cli, String> {
|
||||||
|
let mut mode = Mode::Run;
|
||||||
|
let mut config = None;
|
||||||
|
let mut it = args.into_iter();
|
||||||
|
|
||||||
|
while let Some(arg) = it.next() {
|
||||||
|
match arg.as_str() {
|
||||||
|
"--print-config" => mode = Mode::PrintConfig,
|
||||||
|
"-h" | "--help" => {
|
||||||
|
return Ok(Cli {
|
||||||
|
mode: Mode::Help,
|
||||||
|
config,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
"-V" | "--version" => {
|
||||||
|
return Ok(Cli {
|
||||||
|
mode: Mode::Version,
|
||||||
|
config,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
"--config" => {
|
||||||
|
// `--config` with nothing after it would otherwise silently fall through and start
|
||||||
|
// the sidecar against the default config — the opposite of what was asked for.
|
||||||
|
let path = it
|
||||||
|
.next()
|
||||||
|
.ok_or_else(|| "--config requires a path".to_string())?;
|
||||||
|
config = Some(path);
|
||||||
|
}
|
||||||
|
_ => match arg.strip_prefix("--config=") {
|
||||||
|
Some("") => return Err("--config requires a path".into()),
|
||||||
|
Some(path) => config = Some(path.to_string()),
|
||||||
|
None => return Err(format!("unrecognized argument: {arg}")),
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(Cli { mode, config })
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
fn parse_str(args: &[&str]) -> Result<Cli, String> {
|
||||||
|
parse(args.iter().map(|s| s.to_string()))
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn no_arguments_runs_the_sidecar() {
|
||||||
|
let cli = parse_str(&[]).unwrap();
|
||||||
|
assert_eq!(cli.mode, Mode::Run);
|
||||||
|
assert_eq!(cli.config, None);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn print_config_is_recognized() {
|
||||||
|
assert_eq!(
|
||||||
|
parse_str(&["--print-config"]).unwrap().mode,
|
||||||
|
Mode::PrintConfig
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn config_accepts_both_spellings() {
|
||||||
|
let spaced = parse_str(&["--config", "/etc/runicgateway/sidecar.toml"]).unwrap();
|
||||||
|
let equals = parse_str(&["--config=/etc/runicgateway/sidecar.toml"]).unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
spaced.config.as_deref(),
|
||||||
|
Some("/etc/runicgateway/sidecar.toml")
|
||||||
|
);
|
||||||
|
assert_eq!(spaced, equals);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn config_combines_with_print_config() {
|
||||||
|
let cli = parse_str(&["--config", "c.toml", "--print-config"]).unwrap();
|
||||||
|
assert_eq!(cli.mode, Mode::PrintConfig);
|
||||||
|
assert_eq!(cli.config.as_deref(), Some("c.toml"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn a_path_is_never_swallowed_as_a_flag() {
|
||||||
|
// `--config --print-config` takes the next token as the path, wrong as that path is. The
|
||||||
|
// alternative — treating it as a missing value — guesses at intent.
|
||||||
|
let cli = parse_str(&["--config", "--print-config"]).unwrap();
|
||||||
|
assert_eq!(cli.mode, Mode::Run);
|
||||||
|
assert_eq!(cli.config.as_deref(), Some("--print-config"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn config_without_a_value_is_an_error() {
|
||||||
|
assert!(parse_str(&["--config"]).is_err());
|
||||||
|
assert!(parse_str(&["--config="]).is_err());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn unknown_arguments_are_rejected() {
|
||||||
|
// Silently ignoring a typo'd flag would start a sidecar that is not what was asked for.
|
||||||
|
let err = parse_str(&["--pirnt-config"]).unwrap_err();
|
||||||
|
assert!(err.contains("--pirnt-config"), "{err}");
|
||||||
|
assert!(parse_str(&["/etc/runicgateway/sidecar.toml"]).is_err());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn help_and_version_win_immediately() {
|
||||||
|
assert_eq!(parse_str(&["--help", "--bogus"]).unwrap().mode, Mode::Help);
|
||||||
|
assert_eq!(parse_str(&["-h"]).unwrap().mode, Mode::Help);
|
||||||
|
assert_eq!(
|
||||||
|
parse_str(&["--version", "--bogus"]).unwrap().mode,
|
||||||
|
Mode::Version
|
||||||
|
);
|
||||||
|
assert_eq!(parse_str(&["-V"]).unwrap().mode, Mode::Version);
|
||||||
|
}
|
||||||
|
}
|
||||||
423
sidecar/src/config.rs
Normal file
423
sidecar/src/config.rs
Normal file
@@ -0,0 +1,423 @@
|
|||||||
|
//! Runtime configuration, loaded from an external file — nothing here is compiled into the binary.
|
||||||
|
//!
|
||||||
|
//! Precedence: environment variables override the file, the file overrides built-in defaults. On
|
||||||
|
//! first run, if the file is absent, a default one is written with a freshly generated auth token,
|
||||||
|
//! so the sidecar is secured out of the box and the operator just copies the token to the website.
|
||||||
|
//!
|
||||||
|
//! File path: `--config <PATH>`, else `$RUSTLINK_CONFIG`, else `sidecar.toml` in the working
|
||||||
|
//! directory.
|
||||||
|
//!
|
||||||
|
//! **Paths are anchored to the config file, not the working directory.** A relative `[store].path`
|
||||||
|
//! resolves against the directory holding `sidecar.toml`, so a service started with
|
||||||
|
//! `RUSTLINK_CONFIG=/etc/runicgateway/rust-main.toml` keeps its database beside its config instead
|
||||||
|
//! of wherever the service manager happened to set the working directory.
|
||||||
|
//!
|
||||||
|
//! # `server_id`, and why it is here rather than only in the plugin
|
||||||
|
//!
|
||||||
|
//! R8 makes the platform multi-server: one sidecar per game server, and every row the module
|
||||||
|
//! stores carries the server it came from. The plugin declares its own `serverId` in `server.hello`
|
||||||
|
//! and that is the authority. This setting is a **cross-check**, not a second source of truth: when
|
||||||
|
//! both are set and they disagree, the sidecar logs the disagreement loudly and keeps the plugin's.
|
||||||
|
//! Two servers pointed at one sidecar by a copy-pasted config is the mistake this catches, and it
|
||||||
|
//! is silent in every other design.
|
||||||
|
|
||||||
|
use std::env;
|
||||||
|
use std::fs;
|
||||||
|
use std::path::{Component, Path, PathBuf};
|
||||||
|
|
||||||
|
use serde::Deserialize;
|
||||||
|
use serde_json::json;
|
||||||
|
use tracing::info;
|
||||||
|
|
||||||
|
use crate::PROTOCOL_VERSION;
|
||||||
|
|
||||||
|
#[derive(Debug, Default, Deserialize)]
|
||||||
|
pub struct Config {
|
||||||
|
#[serde(default)]
|
||||||
|
pub game: GameCfg,
|
||||||
|
#[serde(default)]
|
||||||
|
pub web: WebCfg,
|
||||||
|
#[serde(default)]
|
||||||
|
pub store: StoreCfg,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Deserialize)]
|
||||||
|
pub struct GameCfg {
|
||||||
|
#[serde(default = "default_game_bind")]
|
||||||
|
pub bind: String,
|
||||||
|
/// Optional cross-check against the `serverId` the plugin announces. See the module docs.
|
||||||
|
#[serde(default)]
|
||||||
|
pub server_id: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Deserialize)]
|
||||||
|
pub struct WebCfg {
|
||||||
|
#[serde(default = "default_web_bind")]
|
||||||
|
pub bind: String,
|
||||||
|
/// Shared secret the website must present. Never empty in practice — `Config::load` generates
|
||||||
|
/// and persists one when it finds none, so the web surface is authenticated from first boot.
|
||||||
|
#[serde(default)]
|
||||||
|
pub auth_token: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Deserialize)]
|
||||||
|
pub struct StoreCfg {
|
||||||
|
#[serde(default = "default_db_path")]
|
||||||
|
pub path: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A loaded configuration plus what loading it *did* — an installer re-running the binary needs to
|
||||||
|
/// distinguish "read an existing install" from "provisioned a new one", and it cannot tell from the
|
||||||
|
/// values alone.
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub struct Loaded {
|
||||||
|
pub cfg: Config,
|
||||||
|
/// Absolute path of the config file that was read or written.
|
||||||
|
pub path: PathBuf,
|
||||||
|
/// The config file did not exist and was created by this run.
|
||||||
|
pub config_created: bool,
|
||||||
|
/// No usable token was configured, so one was generated and saved.
|
||||||
|
pub token_generated: bool,
|
||||||
|
}
|
||||||
|
|
||||||
|
fn default_game_bind() -> String {
|
||||||
|
"127.0.0.1:7799".into()
|
||||||
|
}
|
||||||
|
fn default_web_bind() -> String {
|
||||||
|
"127.0.0.1:8090".into()
|
||||||
|
}
|
||||||
|
fn default_db_path() -> String {
|
||||||
|
"rust-link.db".into()
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for GameCfg {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self {
|
||||||
|
bind: default_game_bind(),
|
||||||
|
server_id: String::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
impl Default for WebCfg {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self {
|
||||||
|
bind: default_web_bind(),
|
||||||
|
auth_token: String::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
impl Default for StoreCfg {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self {
|
||||||
|
path: default_db_path(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Config {
|
||||||
|
/// Which config file this invocation will use: `--config`, else `$RUSTLINK_CONFIG`, else
|
||||||
|
/// `sidecar.toml` beside the working directory. Always returned absolute, so every later
|
||||||
|
/// message names a path the operator can act on.
|
||||||
|
pub fn resolve_path(cli_override: Option<&str>) -> PathBuf {
|
||||||
|
let raw = cli_override
|
||||||
|
.map(str::to_string)
|
||||||
|
.or_else(|| env::var("RUSTLINK_CONFIG").ok())
|
||||||
|
.unwrap_or_else(|| "sidecar.toml".into());
|
||||||
|
absolutize(PathBuf::from(raw))
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn load(cli_override: Option<&str>) -> anyhow::Result<Loaded> {
|
||||||
|
let path = Self::resolve_path(cli_override);
|
||||||
|
let existed = path.exists();
|
||||||
|
|
||||||
|
let mut cfg: Config = if existed {
|
||||||
|
let text = fs::read_to_string(&path)?;
|
||||||
|
toml::from_str(&text)?
|
||||||
|
} else {
|
||||||
|
Config::default()
|
||||||
|
};
|
||||||
|
|
||||||
|
cfg.apply_env();
|
||||||
|
|
||||||
|
// Authentication is always on. A blank token is never allowed — if none is set (fresh
|
||||||
|
// install, or someone cleared it), generate one, save it, and continue. This keeps setup
|
||||||
|
// effortless while making it impossible to accidentally run with auth off.
|
||||||
|
let token_generated = cfg.web.auth_token.trim().is_empty();
|
||||||
|
if token_generated {
|
||||||
|
let token = generate_token();
|
||||||
|
|
||||||
|
if existed {
|
||||||
|
persist_token(&path, &token)?;
|
||||||
|
} else {
|
||||||
|
// The parent may not exist yet when an installer points at a fresh
|
||||||
|
// /etc/runicgateway; failing here would mean "run me again after mkdir".
|
||||||
|
create_parent_dir(&path)?;
|
||||||
|
fs::write(&path, default_file(&token))?;
|
||||||
|
}
|
||||||
|
|
||||||
|
cfg.web.auth_token = token.clone();
|
||||||
|
|
||||||
|
info!("No auth token configured.");
|
||||||
|
info!("Generated new token: {}", token);
|
||||||
|
info!("Saved to {}. Authentication is on.", path.display());
|
||||||
|
}
|
||||||
|
|
||||||
|
cfg.anchor_store_path(&path);
|
||||||
|
|
||||||
|
Ok(Loaded {
|
||||||
|
cfg,
|
||||||
|
path,
|
||||||
|
config_created: !existed,
|
||||||
|
token_generated,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Environment overrides, so a deployment can set secrets without editing the file.
|
||||||
|
fn apply_env(&mut self) {
|
||||||
|
if let Ok(v) = env::var("RUSTLINK_GAME_BIND") {
|
||||||
|
self.game.bind = v;
|
||||||
|
}
|
||||||
|
if let Ok(v) = env::var("RUSTLINK_SERVER_ID") {
|
||||||
|
self.game.server_id = v;
|
||||||
|
}
|
||||||
|
if let Ok(v) = env::var("RUSTLINK_WEB_BIND") {
|
||||||
|
self.web.bind = v;
|
||||||
|
}
|
||||||
|
if let Ok(v) = env::var("RUSTLINK_WEB_TOKEN") {
|
||||||
|
self.web.auth_token = v;
|
||||||
|
}
|
||||||
|
if let Ok(v) = env::var("RUSTLINK_DB_PATH") {
|
||||||
|
self.store.path = v;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Resolves `[store].path` against the config file's directory (see the module docs). Absolute
|
||||||
|
/// paths and SQLite's non-filesystem spellings are left exactly as written.
|
||||||
|
fn anchor_store_path(&mut self, config_path: &Path) {
|
||||||
|
if is_sqlite_special(&self.store.path) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
let raw = PathBuf::from(&self.store.path);
|
||||||
|
let anchored = if raw.is_absolute() {
|
||||||
|
raw
|
||||||
|
} else {
|
||||||
|
config_dir(config_path).join(raw)
|
||||||
|
};
|
||||||
|
self.store.path = absolutize(anchored).to_string_lossy().into_owned();
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn auth_required(&self) -> bool {
|
||||||
|
// Always true now — load() guarantees a non-empty token.
|
||||||
|
!self.web.auth_token.is_empty()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The `--print-config` document: everything an installer needs to register this sidecar with a
|
||||||
|
/// website, in one non-interactive read.
|
||||||
|
///
|
||||||
|
/// **This includes the auth token in clear text**, which is the point: the manual token hunt is the
|
||||||
|
/// largest "I installed it and nothing happened" failure mode. The caller prints it to stdout and
|
||||||
|
/// starts no log subscriber, so the document is the whole output.
|
||||||
|
pub fn describe(loaded: &Loaded) -> serde_json::Value {
|
||||||
|
json!({
|
||||||
|
"component": "rust-link-sidecar",
|
||||||
|
"version": env!("CARGO_PKG_VERSION"),
|
||||||
|
"protocol": PROTOCOL_VERSION,
|
||||||
|
"config_path": loaded.path.to_string_lossy(),
|
||||||
|
"config_created": loaded.config_created,
|
||||||
|
"token_generated": loaded.token_generated,
|
||||||
|
"game": {
|
||||||
|
"bind": loaded.cfg.game.bind,
|
||||||
|
"server_id": loaded.cfg.game.server_id,
|
||||||
|
},
|
||||||
|
"web": {
|
||||||
|
"bind": loaded.cfg.web.bind,
|
||||||
|
"ws_path": crate::web::WS_PATH,
|
||||||
|
"auth_required": loaded.cfg.auth_required(),
|
||||||
|
"auth_token": loaded.cfg.web.auth_token,
|
||||||
|
},
|
||||||
|
"store": { "path": loaded.cfg.store.path },
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Directory holding the config file. A bare `sidecar.toml` has no parent component, which would
|
||||||
|
/// join into an empty base — treat it as the current directory.
|
||||||
|
fn config_dir(config_path: &Path) -> PathBuf {
|
||||||
|
match config_path.parent() {
|
||||||
|
Some(p) if !p.as_os_str().is_empty() => p.to_path_buf(),
|
||||||
|
_ => PathBuf::from("."),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Prefixes the working directory onto a relative path, then drops the `.` components that
|
||||||
|
/// joining leaves behind — cosmetic, but these paths are printed and pasted into service units.
|
||||||
|
fn absolutize(p: PathBuf) -> PathBuf {
|
||||||
|
let joined = if p.is_absolute() {
|
||||||
|
p
|
||||||
|
} else {
|
||||||
|
match env::current_dir() {
|
||||||
|
Ok(cwd) => cwd.join(p),
|
||||||
|
Err(_) => p,
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let cleaned: PathBuf = joined
|
||||||
|
.components()
|
||||||
|
.filter(|c| !matches!(c, Component::CurDir))
|
||||||
|
.collect();
|
||||||
|
if cleaned.as_os_str().is_empty() {
|
||||||
|
joined
|
||||||
|
} else {
|
||||||
|
cleaned
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `:memory:` and `file:` URIs are instructions to SQLite, not paths on disk. Anchoring them to a
|
||||||
|
/// directory would turn a working in-memory store into an attempt to create a file called
|
||||||
|
/// `:memory:` — which Windows cannot even name.
|
||||||
|
fn is_sqlite_special(path: &str) -> bool {
|
||||||
|
path == ":memory:" || path.starts_with("file:")
|
||||||
|
}
|
||||||
|
|
||||||
|
fn create_parent_dir(path: &Path) -> anyhow::Result<()> {
|
||||||
|
if let Some(dir) = path.parent() {
|
||||||
|
if !dir.as_os_str().is_empty() && !dir.exists() {
|
||||||
|
fs::create_dir_all(dir)?;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Rewrites the `auth_token` line in an existing config file, preserving everything else. Falls
|
||||||
|
/// back to inserting it under `[web]`, or appending a `[web]` section, if the key is absent.
|
||||||
|
fn persist_token(path: &Path, token: &str) -> anyhow::Result<()> {
|
||||||
|
let text = fs::read_to_string(path)?;
|
||||||
|
let line = format!("auth_token = \"{token}\"");
|
||||||
|
|
||||||
|
if text
|
||||||
|
.lines()
|
||||||
|
.any(|l| l.trim_start().starts_with("auth_token"))
|
||||||
|
{
|
||||||
|
let out: String = text
|
||||||
|
.lines()
|
||||||
|
.map(|l| {
|
||||||
|
if l.trim_start().starts_with("auth_token") {
|
||||||
|
line.clone()
|
||||||
|
} else {
|
||||||
|
l.to_string()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
.collect::<Vec<_>>()
|
||||||
|
.join("\n");
|
||||||
|
fs::write(path, out + "\n")?;
|
||||||
|
} else if text.lines().any(|l| l.trim() == "[web]") {
|
||||||
|
let out: String = text
|
||||||
|
.lines()
|
||||||
|
.flat_map(|l| {
|
||||||
|
if l.trim() == "[web]" {
|
||||||
|
vec![l.to_string(), line.clone()]
|
||||||
|
} else {
|
||||||
|
vec![l.to_string()]
|
||||||
|
}
|
||||||
|
})
|
||||||
|
.collect::<Vec<_>>()
|
||||||
|
.join("\n");
|
||||||
|
fs::write(path, out + "\n")?;
|
||||||
|
} else {
|
||||||
|
fs::write(path, format!("{text}\n[web]\n{line}\n"))?;
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn generate_token() -> String {
|
||||||
|
let mut buf = [0u8; 24];
|
||||||
|
// OS randomness; falls back to a time-seeded token only if the OS RNG is unavailable.
|
||||||
|
if getrandom::getrandom(&mut buf).is_err() {
|
||||||
|
let nanos = std::time::SystemTime::now()
|
||||||
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
|
.map(|d| d.as_nanos())
|
||||||
|
.unwrap_or(0);
|
||||||
|
return format!("insecure-fallback-{nanos:x}");
|
||||||
|
}
|
||||||
|
buf.iter().map(|b| format!("{b:02x}")).collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn default_file(token: &str) -> String {
|
||||||
|
format!(
|
||||||
|
r#"# rust-link sidecar configuration.
|
||||||
|
#
|
||||||
|
# One sidecar serves one Rust game server. A community running six servers runs
|
||||||
|
# six of these, each with its own port, its own database and its own token.
|
||||||
|
#
|
||||||
|
# Environment variables override every value here.
|
||||||
|
|
||||||
|
[game]
|
||||||
|
# Where the Oxide bridge plugin dials in. The plugin is the client; this is the
|
||||||
|
# listener, which is why the game server itself opens no extra port.
|
||||||
|
bind = "127.0.0.1:7799"
|
||||||
|
|
||||||
|
# Optional. If set, it is cross-checked against the serverId the plugin
|
||||||
|
# announces in server.hello; a disagreement is logged and the plugin wins.
|
||||||
|
server_id = ""
|
||||||
|
|
||||||
|
[web]
|
||||||
|
# Where the website reaches this sidecar. Bind to a LAN or public address only
|
||||||
|
# behind TLS and a firewall — the token below is the only thing guarding it.
|
||||||
|
bind = "127.0.0.1:8090"
|
||||||
|
|
||||||
|
# Generated on first run. Paste it into the website's Rust server form. It is
|
||||||
|
# write-only there: the site never shows it back.
|
||||||
|
auth_token = "{token}"
|
||||||
|
|
||||||
|
[store]
|
||||||
|
# Relative paths resolve against the directory holding THIS FILE, not the
|
||||||
|
# working directory of whatever started the process.
|
||||||
|
path = "rust-link.db"
|
||||||
|
"#
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn a_generated_token_is_48_hex_characters() {
|
||||||
|
let t = generate_token();
|
||||||
|
assert!(!t.starts_with("insecure-fallback-"), "OS RNG unavailable");
|
||||||
|
assert_eq!(t.len(), 48);
|
||||||
|
assert!(t.chars().all(|c| c.is_ascii_hexdigit()));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn sqlite_special_paths_are_not_anchored() {
|
||||||
|
let mut cfg = Config::default();
|
||||||
|
cfg.store.path = ":memory:".into();
|
||||||
|
cfg.anchor_store_path(Path::new("/etc/runicgateway/sidecar.toml"));
|
||||||
|
assert_eq!(cfg.store.path, ":memory:");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The whole point of anchoring: a service manager's working directory must not decide where
|
||||||
|
/// the database lands.
|
||||||
|
#[test]
|
||||||
|
fn a_relative_store_path_anchors_to_the_config_directory() {
|
||||||
|
let mut cfg = Config::default();
|
||||||
|
cfg.store.path = "rust-link.db".into();
|
||||||
|
let config_path = absolutize(PathBuf::from("cfgdir/sidecar.toml"));
|
||||||
|
cfg.anchor_store_path(&config_path);
|
||||||
|
|
||||||
|
let expected = config_path.parent().unwrap().join("rust-link.db");
|
||||||
|
assert_eq!(PathBuf::from(&cfg.store.path), expected);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The written default must be loadable by the loader that wrote it — a template with a typo
|
||||||
|
/// in it fails on the second start, not the first.
|
||||||
|
#[test]
|
||||||
|
fn the_default_file_round_trips() {
|
||||||
|
let cfg: Config = toml::from_str(&default_file("deadbeef")).unwrap();
|
||||||
|
assert_eq!(cfg.web.auth_token, "deadbeef");
|
||||||
|
assert_eq!(cfg.game.bind, default_game_bind());
|
||||||
|
assert_eq!(cfg.store.path, default_db_path());
|
||||||
|
assert_eq!(cfg.game.server_id, "");
|
||||||
|
}
|
||||||
|
}
|
||||||
415
sidecar/src/game.rs
Normal file
415
sidecar/src/game.rs
Normal file
@@ -0,0 +1,415 @@
|
|||||||
|
//! The loopback link to the Rust server's Oxide bridge plugin.
|
||||||
|
//!
|
||||||
|
//! The plugin is the TCP *client*: it dials out to us. So the sidecar owns the listener, and the
|
||||||
|
//! plugin's outbound socket is the only thing that ever connects. This is the whole reason the game
|
||||||
|
//! is never directly reachable from the website — it exposes no port for us.
|
||||||
|
//!
|
||||||
|
//! Framing is newline-delimited JSON, bidirectional: the plugin sends events (`kind`), we send
|
||||||
|
//! commands (`cmd`). We accept one plugin connection at a time and re-accept when it drops (the
|
||||||
|
//! plugin reconnects on its own, with a bounded backoff).
|
||||||
|
//!
|
||||||
|
//! **Loopback is the trust boundary.** There is no token on this link, exactly as on the ServUO
|
||||||
|
//! bridge: the plugin and the sidecar share a host, and the address to bind is `127.0.0.1`. Binding
|
||||||
|
//! `[game].bind` to anything routable puts an unauthenticated command channel on the network, and
|
||||||
|
//! the operator documentation says so in as many words.
|
||||||
|
//!
|
||||||
|
//! Inbound lines are **capped** (see [`MAX_INBOUND_LINE_BYTES`]) from the start rather than after
|
||||||
|
//! the first large frame arrives: an unbounded `read_line` facing a peer that will one day send a
|
||||||
|
//! map image is a memory-exhaustion shape we would be inventing ourselves.
|
||||||
|
|
||||||
|
use std::sync::Arc;
|
||||||
|
|
||||||
|
use serde_json::Value;
|
||||||
|
use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt, BufReader};
|
||||||
|
use tokio::net::TcpListener;
|
||||||
|
use tokio::sync::{mpsc, Mutex};
|
||||||
|
use tracing::{info, warn};
|
||||||
|
|
||||||
|
/// The longest line the sidecar will accept from the plugin, in bytes.
|
||||||
|
///
|
||||||
|
/// Over-long lines are **discarded, not buffered**, and the connection stays up: a single malformed
|
||||||
|
/// frame is not a reason to tear down a link that live events are flowing over. A dropped reply
|
||||||
|
/// simply times out on the caller's side and is re-requested.
|
||||||
|
pub const MAX_INBOUND_LINE_BYTES: usize = 1024 * 1024;
|
||||||
|
|
||||||
|
/// What one read off the plugin socket produced.
|
||||||
|
#[derive(Debug)]
|
||||||
|
enum Line {
|
||||||
|
/// A complete line, within the cap.
|
||||||
|
Complete(String),
|
||||||
|
/// A line that ran past the cap. Carries how many bytes were thrown away, for the log.
|
||||||
|
TooLong(usize),
|
||||||
|
/// The plugin closed the connection.
|
||||||
|
Eof,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A cancel-safe, capped, newline-delimited reader.
|
||||||
|
///
|
||||||
|
/// Every piece of state that must survive a partial read lives here rather than in a local, because
|
||||||
|
/// this is polled inside a `tokio::select!`: the loop below drops the future whenever a command
|
||||||
|
/// wins the race, and a `discarding` flag or a half-filled buffer held in a local would be lost
|
||||||
|
/// with it. Losing the buffer corrupts the *next* line; losing `discarding` turns the tail of an
|
||||||
|
/// over-long line into a line of its own. Both are silent.
|
||||||
|
///
|
||||||
|
/// The only await point is `fill_buf`, and nothing is consumed until after it returns, so a
|
||||||
|
/// cancellation between the two can lose at most the wakeup.
|
||||||
|
#[derive(Default)]
|
||||||
|
struct LineReader {
|
||||||
|
buf: Vec<u8>,
|
||||||
|
discarding: bool,
|
||||||
|
discarded: usize,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl LineReader {
|
||||||
|
async fn next<R: AsyncBufRead + Unpin>(&mut self, reader: &mut R) -> std::io::Result<Line> {
|
||||||
|
loop {
|
||||||
|
let consumed;
|
||||||
|
let outcome;
|
||||||
|
|
||||||
|
{
|
||||||
|
let available = reader.fill_buf().await?;
|
||||||
|
|
||||||
|
if available.is_empty() {
|
||||||
|
return Ok(Line::Eof);
|
||||||
|
}
|
||||||
|
|
||||||
|
match available.iter().position(|&b| b == b'\n') {
|
||||||
|
Some(at) => {
|
||||||
|
consumed = at + 1;
|
||||||
|
|
||||||
|
if self.discarding {
|
||||||
|
// The tail of a line we already gave up on. Swallow it, terminator
|
||||||
|
// included, and report the size once.
|
||||||
|
self.discarded += at;
|
||||||
|
let total = self.discarded;
|
||||||
|
self.discarding = false;
|
||||||
|
self.discarded = 0;
|
||||||
|
outcome = Some(Line::TooLong(total));
|
||||||
|
} else if self.buf.len() + at > MAX_INBOUND_LINE_BYTES {
|
||||||
|
// The cap is reached only now, on the chunk that also holds the
|
||||||
|
// terminator — so there is nothing left to discard.
|
||||||
|
let total = self.buf.len() + at;
|
||||||
|
self.buf.clear();
|
||||||
|
outcome = Some(Line::TooLong(total));
|
||||||
|
} else {
|
||||||
|
self.buf.extend_from_slice(&available[..at]);
|
||||||
|
let line = String::from_utf8_lossy(&self.buf).into_owned();
|
||||||
|
self.buf.clear();
|
||||||
|
outcome = Some(Line::Complete(line));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
None => {
|
||||||
|
consumed = available.len();
|
||||||
|
|
||||||
|
if self.discarding {
|
||||||
|
self.discarded += consumed;
|
||||||
|
} else if self.buf.len() + consumed > MAX_INBOUND_LINE_BYTES {
|
||||||
|
// Refuse rather than buffer: this is the whole point of the cap.
|
||||||
|
// Everything up to the next newline is now dropped on the floor.
|
||||||
|
self.discarded = self.buf.len() + consumed;
|
||||||
|
self.buf.clear();
|
||||||
|
self.discarding = true;
|
||||||
|
} else {
|
||||||
|
self.buf.extend_from_slice(available);
|
||||||
|
}
|
||||||
|
|
||||||
|
outcome = None;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
reader.consume(consumed);
|
||||||
|
|
||||||
|
if let Some(line) = outcome {
|
||||||
|
return Ok(line);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// An event line received from the plugin, parsed. `kind` is lifted out for routing.
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
pub struct GameEvent {
|
||||||
|
pub kind: String,
|
||||||
|
pub value: Value,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A handle for sending command lines to the plugin. Cloneable and cheap.
|
||||||
|
///
|
||||||
|
/// Commands are dropped (with a warning) when no plugin is connected, rather than buffered: a
|
||||||
|
/// website query that arrives during an outage should fail fast and be retried, not silently queue
|
||||||
|
/// behind a reconnect. Live *events* are what must survive an outage, and those the plugin buffers
|
||||||
|
/// on its side.
|
||||||
|
#[derive(Clone)]
|
||||||
|
pub struct GameHandle {
|
||||||
|
tx: Arc<Mutex<Option<mpsc::UnboundedSender<String>>>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl GameHandle {
|
||||||
|
fn new() -> Self {
|
||||||
|
Self {
|
||||||
|
tx: Arc::new(Mutex::new(None)),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn set(&self, sender: Option<mpsc::UnboundedSender<String>>) {
|
||||||
|
*self.tx.lock().await = sender;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Send one command line (a complete JSON object, no newline — we add the frame delimiter).
|
||||||
|
/// Returns false if no plugin is currently connected.
|
||||||
|
pub async fn send(&self, line: String) -> bool {
|
||||||
|
let guard = self.tx.lock().await;
|
||||||
|
match guard.as_ref() {
|
||||||
|
Some(sender) => sender.send(line).is_ok(),
|
||||||
|
None => {
|
||||||
|
warn!("dropping command; no plugin connected");
|
||||||
|
false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn is_connected(&self) -> bool {
|
||||||
|
self.tx.lock().await.is_some()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Binds the loopback listener and accepts plugin connections forever. Each accepted connection
|
||||||
|
/// runs until it drops, then we loop back to accept the next one. Events are forwarded to
|
||||||
|
/// `event_tx`; the returned handle sends commands to whichever plugin is currently connected.
|
||||||
|
///
|
||||||
|
/// The **bound** address is returned alongside the handle rather than assumed to be the one asked
|
||||||
|
/// for: `127.0.0.1:0` is a legitimate thing to configure (and what the tests use), and a log line
|
||||||
|
/// echoing the request rather than the result is the kind that is wrong exactly when it matters.
|
||||||
|
pub async fn serve(
|
||||||
|
addr: &str,
|
||||||
|
event_tx: mpsc::UnboundedSender<GameEvent>,
|
||||||
|
) -> std::io::Result<(GameHandle, std::net::SocketAddr)> {
|
||||||
|
let listener = TcpListener::bind(addr).await?;
|
||||||
|
let bound = listener.local_addr()?;
|
||||||
|
info!(addr = %bound, "game link listening");
|
||||||
|
|
||||||
|
let handle = GameHandle::new();
|
||||||
|
let accept_handle = handle.clone();
|
||||||
|
|
||||||
|
tokio::spawn(async move {
|
||||||
|
loop {
|
||||||
|
match listener.accept().await {
|
||||||
|
Ok((stream, peer)) => {
|
||||||
|
info!(%peer, "plugin connected");
|
||||||
|
if let Err(e) = handle_connection(stream, &event_tx, &accept_handle).await {
|
||||||
|
warn!(error = %e, "plugin connection ended");
|
||||||
|
} else {
|
||||||
|
info!("plugin disconnected");
|
||||||
|
}
|
||||||
|
accept_handle.set(None).await;
|
||||||
|
// A disconnect is a fact the website should see without polling, so it rides
|
||||||
|
// the same channel every other fact does. Nothing persists it.
|
||||||
|
let _ = event_tx.send(GameEvent {
|
||||||
|
kind: "link.down".to_string(),
|
||||||
|
value: serde_json::json!({ "kind": "link.down" }),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
warn!(error = %e, "accept failed");
|
||||||
|
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
Ok((handle, bound))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn handle_connection(
|
||||||
|
stream: tokio::net::TcpStream,
|
||||||
|
event_tx: &mpsc::UnboundedSender<GameEvent>,
|
||||||
|
handle: &GameHandle,
|
||||||
|
) -> std::io::Result<()> {
|
||||||
|
stream.set_nodelay(true).ok();
|
||||||
|
let (read_half, mut write_half) = stream.into_split();
|
||||||
|
|
||||||
|
// Install the outbound command channel for this connection.
|
||||||
|
let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel::<String>();
|
||||||
|
handle.set(Some(cmd_tx)).await;
|
||||||
|
|
||||||
|
let mut reader = BufReader::new(read_half);
|
||||||
|
let mut lines = LineReader::default();
|
||||||
|
|
||||||
|
loop {
|
||||||
|
tokio::select! {
|
||||||
|
// Inbound: a line from the plugin.
|
||||||
|
result = lines.next(&mut reader) => {
|
||||||
|
match result? {
|
||||||
|
Line::Eof => return Ok(()), // clean EOF: plugin closed
|
||||||
|
Line::TooLong(bytes) => {
|
||||||
|
// Deliberately not a disconnect. See MAX_INBOUND_LINE_BYTES.
|
||||||
|
warn!(
|
||||||
|
bytes,
|
||||||
|
cap = MAX_INBOUND_LINE_BYTES,
|
||||||
|
"inbound line over the cap; discarded"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
Line::Complete(line) => {
|
||||||
|
let trimmed = line.trim_end();
|
||||||
|
if !trimmed.is_empty() {
|
||||||
|
match serde_json::from_str::<Value>(trimmed) {
|
||||||
|
Ok(value) => {
|
||||||
|
let kind = value
|
||||||
|
.get("kind")
|
||||||
|
.and_then(|k| k.as_str())
|
||||||
|
.unwrap_or("")
|
||||||
|
.to_string();
|
||||||
|
let _ = event_tx.send(GameEvent { kind, value });
|
||||||
|
}
|
||||||
|
Err(e) => warn!(error = %e, line = %trimmed, "unparseable event"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// Outbound: a command to write to the plugin.
|
||||||
|
cmd = cmd_rx.recv() => {
|
||||||
|
match cmd {
|
||||||
|
Some(mut c) => {
|
||||||
|
c.push('\n');
|
||||||
|
write_half.write_all(c.as_bytes()).await?;
|
||||||
|
write_half.flush().await?;
|
||||||
|
}
|
||||||
|
None => return Ok(()), // channel closed
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
/// Drives `LineReader` over a byte slice, returning every outcome up to EOF.
|
||||||
|
async fn read_all(input: &[u8]) -> Vec<Line> {
|
||||||
|
let mut reader = BufReader::with_capacity(64, input);
|
||||||
|
let mut lines = LineReader::default();
|
||||||
|
let mut out = Vec::new();
|
||||||
|
|
||||||
|
loop {
|
||||||
|
match lines.next(&mut reader).await.unwrap() {
|
||||||
|
Line::Eof => break,
|
||||||
|
other => out.push(other),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
out
|
||||||
|
}
|
||||||
|
|
||||||
|
fn complete(lines: &[Line]) -> Vec<&str> {
|
||||||
|
lines
|
||||||
|
.iter()
|
||||||
|
.filter_map(|l| match l {
|
||||||
|
Line::Complete(s) => Some(s.as_str()),
|
||||||
|
_ => None,
|
||||||
|
})
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn splits_on_newlines() {
|
||||||
|
let lines = read_all(b"{\"a\":1}\n{\"b\":2}\n").await;
|
||||||
|
assert_eq!(complete(&lines), vec!["{\"a\":1}", "{\"b\":2}"]);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The reader's buffer is 64 bytes here, so every one of these lines spans several `fill_buf`
|
||||||
|
/// chunks. Reassembly across chunks is the thing `read_line` would have done for us.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn reassembles_across_chunks() {
|
||||||
|
let long = "x".repeat(500);
|
||||||
|
let input = format!("{}\n{}\n", long, long);
|
||||||
|
let lines = read_all(input.as_bytes()).await;
|
||||||
|
|
||||||
|
assert_eq!(complete(&lines), vec![long.as_str(), long.as_str()]);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The cap itself. The over-long line must be reported and thrown away, and — the part that
|
||||||
|
/// actually matters — the line *after* it must still arrive intact.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn refuses_an_over_long_line_and_recovers() {
|
||||||
|
let mut input = Vec::new();
|
||||||
|
input.extend_from_slice(&b"a".repeat(MAX_INBOUND_LINE_BYTES + 10));
|
||||||
|
input.push(b'\n');
|
||||||
|
input.extend_from_slice(b"{\"kind\":\"pong\"}\n");
|
||||||
|
|
||||||
|
let lines = read_all(&input).await;
|
||||||
|
|
||||||
|
assert_eq!(lines.len(), 2);
|
||||||
|
assert!(
|
||||||
|
matches!(lines[0], Line::TooLong(n) if n >= MAX_INBOUND_LINE_BYTES),
|
||||||
|
"expected TooLong, got {:?}",
|
||||||
|
lines[0]
|
||||||
|
);
|
||||||
|
assert_eq!(complete(&lines), vec!["{\"kind\":\"pong\"}"]);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A line of exactly the cap is legal; one byte more is not. Checking both sides is what says
|
||||||
|
/// the comparison is `>` rather than `>=`, which would silently cost a byte of the budget.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn the_cap_is_inclusive() {
|
||||||
|
let at_cap = "b".repeat(MAX_INBOUND_LINE_BYTES);
|
||||||
|
let lines = read_all(format!("{}\n", at_cap).as_bytes()).await;
|
||||||
|
assert_eq!(complete(&lines).len(), 1);
|
||||||
|
|
||||||
|
let over = "b".repeat(MAX_INBOUND_LINE_BYTES + 1);
|
||||||
|
let lines = read_all(format!("{}\n", over).as_bytes()).await;
|
||||||
|
assert!(complete(&lines).is_empty());
|
||||||
|
assert!(matches!(lines[0], Line::TooLong(_)));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// An over-long line whose terminator lands in the very chunk that crosses the cap: the reader
|
||||||
|
/// must not leave itself in `discarding` and eat the next line as well.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn over_long_line_terminating_in_the_crossing_chunk() {
|
||||||
|
let mut input = Vec::new();
|
||||||
|
input.extend_from_slice(&b"c".repeat(MAX_INBOUND_LINE_BYTES + 1));
|
||||||
|
input.extend_from_slice(b"\n{\"kind\":\"pong\"}\n");
|
||||||
|
|
||||||
|
let lines = read_all(&input).await;
|
||||||
|
|
||||||
|
assert!(matches!(lines[0], Line::TooLong(_)));
|
||||||
|
assert_eq!(complete(&lines), vec!["{\"kind\":\"pong\"}"]);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A partial line at EOF is dropped rather than delivered half-parsed. The plugin reconnects
|
||||||
|
/// and re-sends; half a JSON object is not something to hand to the event fan-out.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn trailing_partial_line_at_eof_is_dropped() {
|
||||||
|
let lines = read_all(b"{\"a\":1}\n{\"b\":").await;
|
||||||
|
assert_eq!(complete(&lines), vec!["{\"a\":1}"]);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The end-to-end shape of the link, over a real socket: the plugin dials in, sends a hello,
|
||||||
|
/// and receives a command. This is the criterion phase 1 is judged on, minus the two peers.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn a_dialling_plugin_is_accepted_and_can_be_commanded() {
|
||||||
|
use tokio::io::AsyncReadExt;
|
||||||
|
|
||||||
|
let (tx, mut rx) = mpsc::unbounded_channel();
|
||||||
|
let (handle, addr) = serve("127.0.0.1:0", tx).await.unwrap();
|
||||||
|
|
||||||
|
let mut client = tokio::net::TcpStream::connect(addr).await.unwrap();
|
||||||
|
client
|
||||||
|
.write_all(b"{\"kind\":\"server.hello\",\"serverId\":\"main\"}\n")
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let ev = rx.recv().await.unwrap();
|
||||||
|
assert_eq!(ev.kind, "server.hello");
|
||||||
|
assert_eq!(ev.value["serverId"], "main");
|
||||||
|
|
||||||
|
assert!(handle.is_connected().await);
|
||||||
|
assert!(handle.send("{\"cmd\":\"ping\"}".to_string()).await);
|
||||||
|
|
||||||
|
let mut buf = [0u8; 64];
|
||||||
|
let n = client.read(&mut buf).await.unwrap();
|
||||||
|
assert_eq!(&buf[..n], b"{\"cmd\":\"ping\"}\n");
|
||||||
|
}
|
||||||
|
}
|
||||||
134
sidecar/src/main.rs
Normal file
134
sidecar/src/main.rs
Normal file
@@ -0,0 +1,134 @@
|
|||||||
|
//! rust-link sidecar.
|
||||||
|
//!
|
||||||
|
//! Terminates the loopback link to a Rust game server's Oxide bridge plugin and exposes a
|
||||||
|
//! website-facing HTTP surface: a WebSocket live feed and REST queries backed by SQLite.
|
||||||
|
//!
|
||||||
|
//! # Why the game does not listen
|
||||||
|
//!
|
||||||
|
//! The plugin is the TCP *client*; this process owns the listener. A Rust server therefore opens
|
||||||
|
//! no extra port, and the only component the website can reach is this one. That invariant is
|
||||||
|
//! inherited wholesale from the ServUO bridge — the footing changed (Oxide hooks instead of shard
|
||||||
|
//! source) and the shape did not.
|
||||||
|
//!
|
||||||
|
//! # One server, one sidecar
|
||||||
|
//!
|
||||||
|
//! This binary serves exactly one game server. A community running six servers runs six pairs, and
|
||||||
|
//! `module-rust` holds six clients; core never learns there is more than one. Nothing here is
|
||||||
|
//! multiplexed, and nothing here should become multiplexed — the `serverId` on every frame exists
|
||||||
|
//! so the *module* can tell its own clients apart, not so this process can.
|
||||||
|
//!
|
||||||
|
//! # Layout
|
||||||
|
//!
|
||||||
|
//! `main` does argument handling and nothing else; the sidecar proper lives in [`app`], which is
|
||||||
|
//! parameterised on `ready`/`shutdown` so that a future service wrapper (the installer's phase)
|
||||||
|
//! can supply the host's own start and stop without restructuring anything.
|
||||||
|
|
||||||
|
mod app;
|
||||||
|
mod cli;
|
||||||
|
mod config;
|
||||||
|
mod game;
|
||||||
|
mod rpc;
|
||||||
|
mod store;
|
||||||
|
mod web;
|
||||||
|
|
||||||
|
use tracing_subscriber::EnvFilter;
|
||||||
|
|
||||||
|
/// Wire-protocol version between the website and this sidecar, and between this sidecar and the
|
||||||
|
/// bridge plugin. Bump it whenever a frame's shape changes, so a mismatched peer is detected
|
||||||
|
/// immediately (`409` on HTTP, a refusal in the log on the game link) rather than mis-parsed.
|
||||||
|
///
|
||||||
|
/// It is declared in **three** places and they must agree: here, in `Module-Rust`'s
|
||||||
|
/// `module.json`, and in `Rust-Plugins`' `overlay.toml`. The installer refuses to pair a sidecar
|
||||||
|
/// and an overlay that disagree, so a bump lands in the same change as the emitters it describes.
|
||||||
|
///
|
||||||
|
/// # Protocol 1 — the transport
|
||||||
|
///
|
||||||
|
/// Everything phase 1 defines, and deliberately nothing more:
|
||||||
|
///
|
||||||
|
/// * **`server.hello`** — the one event. The plugin sends it on every successful connect, not
|
||||||
|
/// once at server start: this process restarts independently of the game, so anything the
|
||||||
|
/// sidecar needs up front has to be re-sent per connection. It carries the `bootId`, which is
|
||||||
|
/// how a game restart is told apart from a sidecar reconnect — the distinction the event
|
||||||
|
/// system's `reconcile` will later hang off.
|
||||||
|
/// * **`ping` / `pong`** — the heartbeat. The sidecar asks, the plugin answers. A `pong` is
|
||||||
|
/// ephemeral chatter and is never persisted; it only moves `last_event`.
|
||||||
|
/// * **`server.status`** — the one request/reply verb, correlated by `reqId`. It exists in
|
||||||
|
/// phase 1 so the correlation path is exercised by something before anything depends on it.
|
||||||
|
///
|
||||||
|
/// Both directions are newline-delimited JSON over TCP. Outbound frames (plugin -> sidecar) carry
|
||||||
|
/// `kind`; inbound frames (sidecar -> plugin) carry `cmd`.
|
||||||
|
pub const PROTOCOL_VERSION: u32 = 1;
|
||||||
|
|
||||||
|
fn main() -> anyhow::Result<()> {
|
||||||
|
let args = match cli::parse(std::env::args().skip(1)) {
|
||||||
|
Ok(args) => args,
|
||||||
|
Err(msg) => {
|
||||||
|
eprintln!("rust-link-sidecar: {msg}\n\n{}", cli::USAGE);
|
||||||
|
std::process::exit(2);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
match args.mode {
|
||||||
|
cli::Mode::Help => {
|
||||||
|
print!("{}", cli::USAGE);
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
cli::Mode::Version => {
|
||||||
|
println!(
|
||||||
|
"rust-link-sidecar {} (protocol {})",
|
||||||
|
env!("CARGO_PKG_VERSION"),
|
||||||
|
PROTOCOL_VERSION
|
||||||
|
);
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
// Tracing stays uninitialized here on purpose: the subscriber writes to stdout, and on this
|
||||||
|
// path stdout is the document. An installer parses it.
|
||||||
|
cli::Mode::PrintConfig => {
|
||||||
|
let loaded = config::Config::load(args.config.as_deref())?;
|
||||||
|
println!("{:#}", config::describe(&loaded));
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
cli::Mode::Run => {}
|
||||||
|
}
|
||||||
|
|
||||||
|
init_console_tracing();
|
||||||
|
|
||||||
|
let runtime = tokio::runtime::Builder::new_multi_thread()
|
||||||
|
.enable_all()
|
||||||
|
.build()?;
|
||||||
|
|
||||||
|
runtime.block_on(app::run(args.config.as_deref(), || {}, shutdown_signal()))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Logging for a foreground run: human-readable, on stdout.
|
||||||
|
pub fn init_console_tracing() {
|
||||||
|
tracing_subscriber::fmt()
|
||||||
|
.with_env_filter(
|
||||||
|
EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
|
||||||
|
)
|
||||||
|
.init();
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Resolves on Ctrl-C, and on `SIGTERM` where there is one.
|
||||||
|
async fn shutdown_signal() {
|
||||||
|
#[cfg(unix)]
|
||||||
|
{
|
||||||
|
use tokio::signal::unix::{signal, SignalKind};
|
||||||
|
let mut term = match signal(SignalKind::terminate()) {
|
||||||
|
Ok(s) => s,
|
||||||
|
Err(_) => {
|
||||||
|
let _ = tokio::signal::ctrl_c().await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
tokio::select! {
|
||||||
|
_ = tokio::signal::ctrl_c() => {}
|
||||||
|
_ = term.recv() => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(not(unix))]
|
||||||
|
{
|
||||||
|
let _ = tokio::signal::ctrl_c().await;
|
||||||
|
}
|
||||||
|
}
|
||||||
135
sidecar/src/rpc.rs
Normal file
135
sidecar/src/rpc.rs
Normal file
@@ -0,0 +1,135 @@
|
|||||||
|
//! Request/reply correlation over the one game socket.
|
||||||
|
//!
|
||||||
|
//! REST is synchronous ("what is the server doing right now"); the game link is an async stream of
|
||||||
|
//! lines. This bridges them: a call registers a pending entry under a correlation id, sends the
|
||||||
|
//! command, and awaits a reply carrying that id. The event loop routes any incoming line whose
|
||||||
|
//! correlation id is pending back to the waiting caller; everything else flows on as a normal
|
||||||
|
//! event.
|
||||||
|
//!
|
||||||
|
//! One correlation field is recognised in protocol 1 — **`reqId`** — and it is a process-unique
|
||||||
|
//! counter. The UO bridge grew two more over eight protocol versions because callers there supply
|
||||||
|
//! their own ids for some verbs; that is a reason to keep the routing table keyed by string, not a
|
||||||
|
//! reason to invent the extra keys now.
|
||||||
|
//!
|
||||||
|
//! **The timeout here is a ceiling every later command budget sits under.** Core classifies a
|
||||||
|
//! budget overrun as retryable unconditionally — it cannot ask the game, because the action is
|
||||||
|
//! still awaiting a socket — so an action whose `budgetMs` exceeds [`REPLY_TIMEOUT`] can never
|
||||||
|
//! report `retry: false`. Derive one from the other rather than writing both down.
|
||||||
|
|
||||||
|
use std::collections::HashMap;
|
||||||
|
use std::sync::atomic::{AtomicU64, Ordering};
|
||||||
|
use std::sync::Arc;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
use serde_json::Value;
|
||||||
|
use tokio::sync::{oneshot, Mutex};
|
||||||
|
|
||||||
|
use crate::game::GameHandle;
|
||||||
|
|
||||||
|
/// How long a correlated call waits for its reply before giving up.
|
||||||
|
pub const REPLY_TIMEOUT: Duration = Duration::from_secs(10);
|
||||||
|
|
||||||
|
#[derive(Clone)]
|
||||||
|
pub struct Rpc {
|
||||||
|
pending: Arc<Mutex<HashMap<String, oneshot::Sender<Value>>>>,
|
||||||
|
counter: Arc<AtomicU64>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub enum RpcError {
|
||||||
|
NoPlugin,
|
||||||
|
Timeout,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Rpc {
|
||||||
|
pub fn new() -> Self {
|
||||||
|
Self {
|
||||||
|
pending: Arc::new(Mutex::new(HashMap::new())),
|
||||||
|
counter: Arc::new(AtomicU64::new(1)),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn next_req_id(&self) -> String {
|
||||||
|
format!("r-{}", self.counter.fetch_add(1, Ordering::Relaxed))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Sends `command` to the plugin and awaits the reply correlated by `corr_val`. The command
|
||||||
|
/// must already contain the correlation field (`reqId`) set to `corr_val`.
|
||||||
|
pub async fn call(
|
||||||
|
&self,
|
||||||
|
game: &GameHandle,
|
||||||
|
command: Value,
|
||||||
|
corr_val: &str,
|
||||||
|
) -> Result<Value, RpcError> {
|
||||||
|
let (tx, rx) = oneshot::channel();
|
||||||
|
|
||||||
|
self.pending.lock().await.insert(corr_val.to_string(), tx);
|
||||||
|
|
||||||
|
if !game.send(command.to_string()).await {
|
||||||
|
self.pending.lock().await.remove(corr_val);
|
||||||
|
return Err(RpcError::NoPlugin);
|
||||||
|
}
|
||||||
|
|
||||||
|
match tokio::time::timeout(REPLY_TIMEOUT, rx).await {
|
||||||
|
Ok(Ok(value)) => Ok(value),
|
||||||
|
_ => {
|
||||||
|
// Both the timeout and a dropped sender land here, and both must clear the entry:
|
||||||
|
// a pending map that only ever grows is a leak keyed by a counter.
|
||||||
|
self.pending.lock().await.remove(corr_val);
|
||||||
|
Err(RpcError::Timeout)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// If this incoming value correlates to a pending call, complete it and return true (the value
|
||||||
|
/// was a reply, not a broadcast event). Otherwise return false.
|
||||||
|
pub async fn try_route(&self, value: &Value) -> bool {
|
||||||
|
let corr = match value.get("reqId").and_then(|v| v.as_str()) {
|
||||||
|
Some(c) => c.to_string(),
|
||||||
|
None => return false,
|
||||||
|
};
|
||||||
|
|
||||||
|
let sender = self.pending.lock().await.remove(&corr);
|
||||||
|
match sender {
|
||||||
|
Some(tx) => {
|
||||||
|
let _ = tx.send(value.clone());
|
||||||
|
true
|
||||||
|
}
|
||||||
|
None => false,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use serde_json::json;
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn an_unknown_req_id_is_not_a_reply() {
|
||||||
|
let rpc = Rpc::new();
|
||||||
|
assert!(!rpc.try_route(&json!({"kind": "server.hello"})).await);
|
||||||
|
assert!(!rpc.try_route(&json!({"reqId": "r-999"})).await);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn req_ids_are_unique_within_a_process() {
|
||||||
|
let rpc = Rpc::new();
|
||||||
|
let a = rpc.next_req_id();
|
||||||
|
let b = rpc.next_req_id();
|
||||||
|
assert_ne!(a, b);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A reply routed while nothing is waiting must flow on as an ordinary event rather than be
|
||||||
|
/// swallowed — that is the difference between a late reply and a lost one.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn a_late_reply_falls_through_to_the_event_path() {
|
||||||
|
let rpc = Rpc::new();
|
||||||
|
let (tx, _rx) = oneshot::channel();
|
||||||
|
rpc.pending.lock().await.insert("r-1".into(), tx);
|
||||||
|
|
||||||
|
assert!(rpc.try_route(&json!({"reqId": "r-1"})).await);
|
||||||
|
// Second delivery of the same id: the entry is gone, so it is an event now.
|
||||||
|
assert!(!rpc.try_route(&json!({"reqId": "r-1"})).await);
|
||||||
|
}
|
||||||
|
}
|
||||||
271
sidecar/src/store.rs
Normal file
271
sidecar/src/store.rs
Normal file
@@ -0,0 +1,271 @@
|
|||||||
|
//! SQLite persistence: the event history, and the last thing the game said about itself.
|
||||||
|
//!
|
||||||
|
//! This is what lets the website read the past without asking the game, and what survives a sidecar
|
||||||
|
//! restart. The event loop writes every live event here as it broadcasts it; REST reads query here
|
||||||
|
//! instead of round-tripping the plugin.
|
||||||
|
//!
|
||||||
|
//! **The sidecar defines no schema for a frame's contents.** Events are persisted whole, as the
|
||||||
|
//! JSON text that arrived, with only `t` and `kind` lifted out for indexing. That is the
|
||||||
|
//! dumb-forwarder property doing real work: a protocol version that adds fields to an event needs
|
||||||
|
//! no change here, and only a version that adds a *new indexed column* ever needs a migration.
|
||||||
|
|
||||||
|
use std::path::Path;
|
||||||
|
|
||||||
|
use serde_json::Value;
|
||||||
|
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
|
||||||
|
use sqlx::{Row, SqlitePool};
|
||||||
|
use tracing::info;
|
||||||
|
|
||||||
|
const SCHEMA: &str = "
|
||||||
|
CREATE TABLE IF NOT EXISTS events (
|
||||||
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||||
|
t INTEGER NOT NULL,
|
||||||
|
kind TEXT NOT NULL,
|
||||||
|
json TEXT NOT NULL
|
||||||
|
);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_events_kind_id ON events (kind, id DESC);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_events_t ON events (t);
|
||||||
|
|
||||||
|
-- Exactly one row, id 1: the most recent server.hello. A board, in the sense chapter 4 uses the
|
||||||
|
-- word — current state with one producer, re-sent on every connect — rather than a history.
|
||||||
|
CREATE TABLE IF NOT EXISTS server_state (
|
||||||
|
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||||
|
t INTEGER NOT NULL,
|
||||||
|
json TEXT NOT NULL
|
||||||
|
);
|
||||||
|
";
|
||||||
|
|
||||||
|
#[derive(Clone)]
|
||||||
|
pub struct Store {
|
||||||
|
pool: SqlitePool,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Store {
|
||||||
|
/// Opens (creating if absent) the SQLite database and ensures the schema exists.
|
||||||
|
///
|
||||||
|
/// `path` is a filesystem path, handed to sqlx as one. It is deliberately **not** formatted
|
||||||
|
/// into a `sqlite://` URL first: that spelling is parsed as a URL, so it percent-decodes the
|
||||||
|
/// path and splits it on `?`. Under an installed layout the path is absolute and chosen by the
|
||||||
|
/// operator — `C:\ProgramData\RunicGateway\rust-link.db`, or something under a home directory
|
||||||
|
/// with a `%` or `#` in it — and a URL round-trip silently opens a *different* file.
|
||||||
|
pub async fn open(path: &str) -> anyhow::Result<Self> {
|
||||||
|
// A service unit can name a data directory that does not exist yet; creating it here means
|
||||||
|
// one less way for a fresh install to fail on first start.
|
||||||
|
if let Some(dir) = Path::new(path).parent() {
|
||||||
|
if !dir.as_os_str().is_empty() && !dir.exists() {
|
||||||
|
std::fs::create_dir_all(dir)?;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let opts = SqliteConnectOptions::new()
|
||||||
|
.filename(path)
|
||||||
|
.create_if_missing(true);
|
||||||
|
|
||||||
|
// An in-memory database is **per connection**, not per process: every connection the pool
|
||||||
|
// opens gets its own empty one, so a second pooled connection finds none of the schema the
|
||||||
|
// first created. It presents as `no such table` from a random subset of queries, which is
|
||||||
|
// as confusing a failure as this file has. A single connection is the only coherent
|
||||||
|
// reading of `:memory:`, and it is what makes it usable at all.
|
||||||
|
let max = if is_in_memory(path) { 1 } else { 4 };
|
||||||
|
|
||||||
|
let pool = SqlitePoolOptions::new()
|
||||||
|
.max_connections(max)
|
||||||
|
.connect_with(opts)
|
||||||
|
.await?;
|
||||||
|
|
||||||
|
sqlx::query(SCHEMA).execute(&pool).await?;
|
||||||
|
info!(%path, "store ready");
|
||||||
|
Ok(Self { pool })
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Cheap liveness check for the health endpoint.
|
||||||
|
pub async fn ping(&self) -> anyhow::Result<()> {
|
||||||
|
sqlx::query("SELECT 1").execute(&self.pool).await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Appends one live event. Failures are logged by the caller; persistence must never block the
|
||||||
|
/// live feed.
|
||||||
|
pub async fn insert_event(&self, t: i64, kind: &str, json: &str) -> anyhow::Result<()> {
|
||||||
|
sqlx::query("INSERT INTO events (t, kind, json) VALUES (?, ?, ?)")
|
||||||
|
.bind(t)
|
||||||
|
.bind(kind)
|
||||||
|
.bind(json)
|
||||||
|
.execute(&self.pool)
|
||||||
|
.await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Most-recent events, newest first, optionally filtered by kind.
|
||||||
|
pub async fn recent(&self, kind: Option<&str>, limit: i64) -> anyhow::Result<Vec<Value>> {
|
||||||
|
let limit = limit.clamp(1, 1000);
|
||||||
|
let rows = match kind {
|
||||||
|
Some(k) => {
|
||||||
|
sqlx::query("SELECT json FROM events WHERE kind = ? ORDER BY id DESC LIMIT ?")
|
||||||
|
.bind(k)
|
||||||
|
.bind(limit)
|
||||||
|
.fetch_all(&self.pool)
|
||||||
|
.await?
|
||||||
|
}
|
||||||
|
None => {
|
||||||
|
sqlx::query("SELECT json FROM events ORDER BY id DESC LIMIT ?")
|
||||||
|
.bind(limit)
|
||||||
|
.fetch_all(&self.pool)
|
||||||
|
.await?
|
||||||
|
}
|
||||||
|
};
|
||||||
|
Ok(parse_json_column(rows))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Replaces the one `server_state` row. Called for every `server.hello`, which the plugin sends
|
||||||
|
/// on every connect — so this is an upsert by construction, not by accident.
|
||||||
|
pub async fn put_server_state(&self, t: i64, json: &str) -> anyhow::Result<()> {
|
||||||
|
sqlx::query(
|
||||||
|
"INSERT INTO server_state (id, t, json) VALUES (1, ?, ?)
|
||||||
|
ON CONFLICT(id) DO UPDATE SET t = excluded.t, json = excluded.json",
|
||||||
|
)
|
||||||
|
.bind(t)
|
||||||
|
.bind(json)
|
||||||
|
.execute(&self.pool)
|
||||||
|
.await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The last thing the game said about itself, or `None` if it has never connected.
|
||||||
|
///
|
||||||
|
/// This is the read that makes the website render while the game is off, which is the whole
|
||||||
|
/// reason the sidecar holds a database at all.
|
||||||
|
pub async fn server_state(&self) -> anyhow::Result<Option<Value>> {
|
||||||
|
let row = sqlx::query("SELECT json FROM server_state WHERE id = 1")
|
||||||
|
.fetch_optional(&self.pool)
|
||||||
|
.await?;
|
||||||
|
Ok(row.and_then(|r| serde_json::from_str(&r.get::<String, _>("json")).ok()))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Whether this path names an in-memory database rather than a file. Covers the bare `:memory:`
|
||||||
|
/// spelling and the `file:` URI form that carries `mode=memory`.
|
||||||
|
fn is_in_memory(path: &str) -> bool {
|
||||||
|
path == ":memory:" || (path.starts_with("file:") && path.contains("mode=memory"))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn parse_json_column(rows: Vec<sqlx::sqlite::SqliteRow>) -> Vec<Value> {
|
||||||
|
rows.into_iter()
|
||||||
|
.filter_map(|r| serde_json::from_str(&r.get::<String, _>("json")).ok())
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use serde_json::json;
|
||||||
|
|
||||||
|
async fn store() -> Store {
|
||||||
|
Store::open(":memory:").await.unwrap()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The trap this file's pool sizing exists for: a multi-connection pool over `:memory:` hands
|
||||||
|
/// out empty databases. Asserting the *pool* is what makes the reason visible; asserting only
|
||||||
|
/// that a query works would pass again the moment someone "tidied" the sizing back.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn an_in_memory_store_uses_exactly_one_connection() {
|
||||||
|
assert!(is_in_memory(":memory:"));
|
||||||
|
assert!(is_in_memory("file:x?mode=memory&cache=shared"));
|
||||||
|
assert!(!is_in_memory("rust-link.db"));
|
||||||
|
assert!(!is_in_memory("file:/var/lib/rg/rust-link.db"));
|
||||||
|
|
||||||
|
let s = store().await;
|
||||||
|
assert_eq!(s.pool.options().get_max_connections(), 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `sqlx::query` over a multi-statement string is the kind of thing that quietly runs only the
|
||||||
|
/// first statement. Both tables and both reads have to work on a real file, under the pool
|
||||||
|
/// size production uses.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn the_whole_schema_is_created_on_a_pooled_file_store() {
|
||||||
|
let dir = std::env::temp_dir().join(format!("rust-link-test-{}", std::process::id()));
|
||||||
|
let path = dir.join("schema.db");
|
||||||
|
let _ = std::fs::remove_dir_all(&dir);
|
||||||
|
|
||||||
|
let s = Store::open(path.to_str().unwrap()).await.unwrap();
|
||||||
|
assert_eq!(s.pool.options().get_max_connections(), 4);
|
||||||
|
|
||||||
|
s.insert_event(1, "k", "{}").await.unwrap();
|
||||||
|
s.put_server_state(1, "{}").await.unwrap();
|
||||||
|
assert_eq!(s.recent(None, 10).await.unwrap().len(), 1);
|
||||||
|
assert!(s.server_state().await.unwrap().is_some());
|
||||||
|
|
||||||
|
drop(s);
|
||||||
|
let _ = std::fs::remove_dir_all(&dir);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn events_come_back_newest_first_and_filter_by_kind() {
|
||||||
|
let s = store().await;
|
||||||
|
s.insert_event(1, "server.hello", &json!({"n": 1}).to_string())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
s.insert_event(2, "other", &json!({"n": 2}).to_string())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
s.insert_event(3, "server.hello", &json!({"n": 3}).to_string())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let all = s.recent(None, 10).await.unwrap();
|
||||||
|
assert_eq!(all.len(), 3);
|
||||||
|
assert_eq!(all[0]["n"], 3);
|
||||||
|
|
||||||
|
let hellos = s.recent(Some("server.hello"), 10).await.unwrap();
|
||||||
|
assert_eq!(hellos.len(), 2);
|
||||||
|
assert_eq!(hellos[0]["n"], 3);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// An absent board reads as `None`, not as an empty object. A caller must be able to tell
|
||||||
|
/// "the game has never connected" from "the game connected and said nothing" — collapsing the
|
||||||
|
/// two is how a site ends up rendering a server that does not exist.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn server_state_is_absent_until_a_hello_arrives() {
|
||||||
|
let s = store().await;
|
||||||
|
assert!(s.server_state().await.unwrap().is_none());
|
||||||
|
|
||||||
|
s.put_server_state(1, &json!({"serverId": "main", "players": 0}).to_string())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(s.server_state().await.unwrap().unwrap()["serverId"], "main");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The board holds exactly one row however many times the plugin reconnects.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn a_second_hello_replaces_the_first() {
|
||||||
|
let s = store().await;
|
||||||
|
s.put_server_state(1, &json!({"bootId": "a"}).to_string())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
s.put_server_state(2, &json!({"bootId": "b"}).to_string())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
assert_eq!(s.server_state().await.unwrap().unwrap()["bootId"], "b");
|
||||||
|
let count: i64 = sqlx::query("SELECT COUNT(*) AS c FROM server_state")
|
||||||
|
.fetch_one(&s.pool)
|
||||||
|
.await
|
||||||
|
.unwrap()
|
||||||
|
.get("c");
|
||||||
|
assert_eq!(count, 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn the_limit_is_clamped_rather_than_trusted() {
|
||||||
|
let s = store().await;
|
||||||
|
for i in 0..5 {
|
||||||
|
s.insert_event(i, "k", &json!({"i": i}).to_string())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
}
|
||||||
|
// 0 and negatives would otherwise mean "no rows" and "SQLite's unlimited" respectively.
|
||||||
|
assert_eq!(s.recent(None, 0).await.unwrap().len(), 1);
|
||||||
|
assert_eq!(s.recent(None, -1).await.unwrap().len(), 1);
|
||||||
|
assert_eq!(s.recent(None, 100_000).await.unwrap().len(), 5);
|
||||||
|
}
|
||||||
|
}
|
||||||
422
sidecar/src/web.rs
Normal file
422
sidecar/src/web.rs
Normal file
@@ -0,0 +1,422 @@
|
|||||||
|
//! The website-facing HTTP surface: a WebSocket live feed, and REST reads backed by the store.
|
||||||
|
//!
|
||||||
|
//! Two rules shape everything here, and both are inherited rather than invented:
|
||||||
|
//!
|
||||||
|
//! 1. **Authentication is always on.** `/health` is the only unauthenticated route, because
|
||||||
|
//! monitoring must be able to reach it. Everything else is behind a shared token that the config
|
||||||
|
//! generates on first run, so there is no state in which this process is listening without one.
|
||||||
|
//! 2. **Every response advertises the protocol version**, and a client that declares a different
|
||||||
|
//! one is refused `409` rather than served something it will mis-parse. A version mismatch is a
|
||||||
|
//! deployment fault, and it should look like one.
|
||||||
|
//!
|
||||||
|
//! The reads are store-backed on purpose. `/server` answers the last thing the game said about
|
||||||
|
//! itself even while the game is off, which is what lets the website render a server list during a
|
||||||
|
//! wipe or a restart. `/status` is the one route that round-trips the plugin, and it is the one
|
||||||
|
//! route that fails when the game is down — which is the honest answer to "what is it doing *right
|
||||||
|
//! now*".
|
||||||
|
|
||||||
|
use std::sync::atomic::{AtomicI64, Ordering};
|
||||||
|
use std::sync::Arc;
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
|
use axum::{
|
||||||
|
extract::ws::{Message, WebSocket, WebSocketUpgrade},
|
||||||
|
extract::{Query, Request, State},
|
||||||
|
http::{HeaderValue, StatusCode},
|
||||||
|
middleware::{self, Next},
|
||||||
|
response::{IntoResponse, Response},
|
||||||
|
routing::get,
|
||||||
|
Json, Router,
|
||||||
|
};
|
||||||
|
use serde::Deserialize;
|
||||||
|
use serde_json::{json, Value};
|
||||||
|
use tokio::sync::broadcast;
|
||||||
|
use tracing::{debug, info, warn};
|
||||||
|
|
||||||
|
use crate::game::GameHandle;
|
||||||
|
use crate::rpc::{Rpc, RpcError};
|
||||||
|
use crate::store::Store;
|
||||||
|
use crate::PROTOCOL_VERSION;
|
||||||
|
|
||||||
|
/// The header every response carries, and the one a client may send to declare its own version.
|
||||||
|
pub const VERSION_HEADER: &str = "X-RustLink-Version";
|
||||||
|
|
||||||
|
/// Where the live feed lives. Named rather than spelled inline because `--print-config` reports it
|
||||||
|
/// to the installer, and the two must not drift.
|
||||||
|
pub const WS_PATH: &str = "/ws";
|
||||||
|
|
||||||
|
/// Shared state handed to each request handler.
|
||||||
|
#[derive(Clone)]
|
||||||
|
pub struct AppState {
|
||||||
|
pub events: broadcast::Sender<String>,
|
||||||
|
pub game: GameHandle,
|
||||||
|
pub rpc: Rpc,
|
||||||
|
pub store: Store,
|
||||||
|
/// Shared secret the website must present. Always set (config guarantees non-empty).
|
||||||
|
pub token: Arc<String>,
|
||||||
|
pub started: Instant,
|
||||||
|
/// Epoch ms of the last line received from the plugin, 0 if none yet.
|
||||||
|
pub last_event: Arc<AtomicI64>,
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn serve(addr: &str, state: AppState) -> anyhow::Result<()> {
|
||||||
|
let protected = Router::new()
|
||||||
|
.route(WS_PATH, get(ws_upgrade))
|
||||||
|
// The board: the last `server.hello`. Store-backed, so it answers while the game is off.
|
||||||
|
.route("/server", get(server_board))
|
||||||
|
// Event history, newest first, optionally filtered by kind.
|
||||||
|
.route("/events", get(events))
|
||||||
|
// Live: a correlated round trip to the plugin. Fails when the game is down, by design.
|
||||||
|
.route("/status", get(status))
|
||||||
|
.route_layer(middleware::from_fn_with_state(state.clone(), gate));
|
||||||
|
|
||||||
|
let app = Router::new()
|
||||||
|
.route("/health", get(health))
|
||||||
|
.merge(protected)
|
||||||
|
// Every response advertises the protocol version, so a client can notice a mismatch even
|
||||||
|
// on /health or on an error response.
|
||||||
|
.layer(middleware::from_fn(version_header))
|
||||||
|
.with_state(state);
|
||||||
|
|
||||||
|
let listener = tokio::net::TcpListener::bind(addr).await?;
|
||||||
|
info!(addr = %listener.local_addr()?, "web server listening");
|
||||||
|
axum::serve(listener, app).await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- health ----
|
||||||
|
|
||||||
|
/// Protocol version, whether the plugin is connected, database reachability, uptime, and when the
|
||||||
|
/// plugin last sent anything. Unauthenticated, so monitoring can reach it.
|
||||||
|
async fn health(State(st): State<AppState>) -> impl IntoResponse {
|
||||||
|
let plugin = st.game.is_connected().await;
|
||||||
|
let db_ok = st.store.ping().await.is_ok();
|
||||||
|
let last_ms = st.last_event.load(Ordering::Relaxed);
|
||||||
|
|
||||||
|
let status = if plugin && db_ok { "ok" } else { "degraded" };
|
||||||
|
|
||||||
|
Json(json!({
|
||||||
|
"status": status,
|
||||||
|
"protocol": PROTOCOL_VERSION,
|
||||||
|
"plugin_connected": plugin,
|
||||||
|
"database": if db_ok { "ok" } else { "error" },
|
||||||
|
"uptime": format_uptime(st.started.elapsed()),
|
||||||
|
"last_event": iso_ms(last_ms),
|
||||||
|
}))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn format_uptime(d: Duration) -> String {
|
||||||
|
let secs = d.as_secs();
|
||||||
|
let (days, hours, mins) = (secs / 86400, (secs % 86400) / 3600, (secs % 3600) / 60);
|
||||||
|
if days > 0 {
|
||||||
|
format!("{days}d {hours}h")
|
||||||
|
} else if hours > 0 {
|
||||||
|
format!("{hours}h {mins}m")
|
||||||
|
} else {
|
||||||
|
format!("{mins}m")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn iso_ms(ms: i64) -> Option<String> {
|
||||||
|
if ms <= 0 {
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
chrono::DateTime::from_timestamp_millis(ms)
|
||||||
|
.map(|dt| dt.format("%Y-%m-%dT%H:%M:%SZ").to_string())
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- reads ----
|
||||||
|
|
||||||
|
/// The last `server.hello`, or `204` if the game has never connected.
|
||||||
|
///
|
||||||
|
/// `204` rather than `200` with a null: "we have never heard from this server" and "this server
|
||||||
|
/// reports nothing" are different answers, and a client that cannot tell them apart renders a
|
||||||
|
/// server that does not exist.
|
||||||
|
async fn server_board(State(st): State<AppState>) -> Response {
|
||||||
|
match st.store.server_state().await {
|
||||||
|
Ok(Some(v)) => Json(v).into_response(),
|
||||||
|
Ok(None) => StatusCode::NO_CONTENT.into_response(),
|
||||||
|
Err(e) => {
|
||||||
|
warn!(error = %e, "server board read failed");
|
||||||
|
store_error()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Deserialize)]
|
||||||
|
struct EventsQuery {
|
||||||
|
kind: Option<String>,
|
||||||
|
limit: Option<i64>,
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn events(State(st): State<AppState>, Query(q): Query<EventsQuery>) -> Response {
|
||||||
|
match st
|
||||||
|
.store
|
||||||
|
.recent(q.kind.as_deref(), q.limit.unwrap_or(50))
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
Ok(rows) => Json(json!({ "events": rows })).into_response(),
|
||||||
|
Err(e) => {
|
||||||
|
warn!(error = %e, "event read failed");
|
||||||
|
store_error()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn store_error() -> Response {
|
||||||
|
(
|
||||||
|
StatusCode::INTERNAL_SERVER_ERROR,
|
||||||
|
Json(json!({"error": "store read failed"})),
|
||||||
|
)
|
||||||
|
.into_response()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A correlated round trip to the plugin: what the server is doing right now.
|
||||||
|
async fn status(State(st): State<AppState>) -> Response {
|
||||||
|
let req_id = st.rpc.next_req_id();
|
||||||
|
let command = json!({ "cmd": "server.status", "reqId": req_id });
|
||||||
|
respond(st.rpc.call(&st.game, command, &req_id).await)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Maps an RPC outcome onto a status code.
|
||||||
|
///
|
||||||
|
/// The two failures are deliberately different codes. `NoPlugin` is `503`: the game is down and the
|
||||||
|
/// caller should stop asking for a moment. `Timeout` is `504`: the game is up and did not answer in
|
||||||
|
/// time, which is a different operational problem with a different fix.
|
||||||
|
fn respond(result: Result<Value, RpcError>) -> Response {
|
||||||
|
match result {
|
||||||
|
Ok(v) => Json(v).into_response(),
|
||||||
|
Err(RpcError::NoPlugin) => (
|
||||||
|
StatusCode::SERVICE_UNAVAILABLE,
|
||||||
|
Json(json!({"error": "no plugin connected"})),
|
||||||
|
)
|
||||||
|
.into_response(),
|
||||||
|
Err(RpcError::Timeout) => (
|
||||||
|
StatusCode::GATEWAY_TIMEOUT,
|
||||||
|
Json(json!({"error": "the plugin did not reply in time"})),
|
||||||
|
)
|
||||||
|
.into_response(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- gate: protocol check + auth ----
|
||||||
|
|
||||||
|
/// Adds the version header to every response.
|
||||||
|
async fn version_header(req: Request, next: Next) -> Response {
|
||||||
|
let mut resp = next.run(req).await;
|
||||||
|
if let Ok(v) = HeaderValue::from_str(&PROTOCOL_VERSION.to_string()) {
|
||||||
|
resp.headers_mut().insert(VERSION_HEADER, v);
|
||||||
|
}
|
||||||
|
resp
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Guards every non-health route: first a protocol-version check (if the client declares one), then
|
||||||
|
/// authentication. The token may arrive as `Authorization: Bearer <t>`, `X-Api-Key: <t>`, or
|
||||||
|
/// `?token=<t>` (the last so browser WebSocket clients, which cannot set handshake headers, still
|
||||||
|
/// authenticate). The token compare is constant-time.
|
||||||
|
async fn gate(State(st): State<AppState>, req: Request, next: Next) -> Response {
|
||||||
|
// Protocol version: if the client states one and it disagrees, fail loudly and specifically.
|
||||||
|
if let Some(v) = req
|
||||||
|
.headers()
|
||||||
|
.get(VERSION_HEADER.to_ascii_lowercase().as_str())
|
||||||
|
.and_then(|h| h.to_str().ok())
|
||||||
|
{
|
||||||
|
if v.trim() != PROTOCOL_VERSION.to_string() {
|
||||||
|
return (
|
||||||
|
StatusCode::CONFLICT,
|
||||||
|
Json(json!({
|
||||||
|
"error": "protocol version mismatch",
|
||||||
|
"sidecar_protocol": PROTOCOL_VERSION,
|
||||||
|
"client_protocol": v.trim(),
|
||||||
|
})),
|
||||||
|
)
|
||||||
|
.into_response();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
match extract_token(&req) {
|
||||||
|
Some(t) if constant_time_eq(t.as_bytes(), st.token.as_bytes()) => next.run(req).await,
|
||||||
|
_ => (
|
||||||
|
StatusCode::UNAUTHORIZED,
|
||||||
|
Json(json!({"error": "missing or invalid auth token"})),
|
||||||
|
)
|
||||||
|
.into_response(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn extract_token(req: &Request) -> Option<String> {
|
||||||
|
// Authorization: Bearer <token>
|
||||||
|
if let Some(v) = req
|
||||||
|
.headers()
|
||||||
|
.get("authorization")
|
||||||
|
.and_then(|h| h.to_str().ok())
|
||||||
|
{
|
||||||
|
if let Some(rest) = v
|
||||||
|
.strip_prefix("Bearer ")
|
||||||
|
.or_else(|| v.strip_prefix("bearer "))
|
||||||
|
{
|
||||||
|
return Some(rest.trim().to_string());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// X-Api-Key: <token>
|
||||||
|
if let Some(v) = req.headers().get("x-api-key").and_then(|h| h.to_str().ok()) {
|
||||||
|
return Some(v.trim().to_string());
|
||||||
|
}
|
||||||
|
// ?token=<token>
|
||||||
|
if let Some(q) = req.uri().query() {
|
||||||
|
for pair in q.split('&') {
|
||||||
|
if let Some(v) = pair.strip_prefix("token=") {
|
||||||
|
return Some(v.to_string());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
None
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Length-independent, early-return-free comparison, so a wrong token leaks no timing signal.
|
||||||
|
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
|
||||||
|
if a.len() != b.len() {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
let mut diff = 0u8;
|
||||||
|
for (x, y) in a.iter().zip(b.iter()) {
|
||||||
|
diff |= x ^ y;
|
||||||
|
}
|
||||||
|
diff == 0
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- live feed ----
|
||||||
|
|
||||||
|
async fn ws_upgrade(ws: WebSocketUpgrade, State(state): State<AppState>) -> impl IntoResponse {
|
||||||
|
ws.on_upgrade(move |socket| ws_client(socket, state))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn ws_client(mut socket: WebSocket, state: AppState) {
|
||||||
|
let mut rx = state.events.subscribe();
|
||||||
|
info!("ws client connected");
|
||||||
|
|
||||||
|
let hello = json!({"kind": "ws.hello", "protocol": PROTOCOL_VERSION}).to_string();
|
||||||
|
if socket.send(Message::Text(hello)).await.is_err() {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
loop {
|
||||||
|
tokio::select! {
|
||||||
|
recv = rx.recv() => {
|
||||||
|
match recv {
|
||||||
|
Ok(line) => {
|
||||||
|
if socket.send(Message::Text(line)).await.is_err() {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(broadcast::error::RecvError::Lagged(n)) => {
|
||||||
|
warn!(skipped = n, "ws client lagged; dropping missed events");
|
||||||
|
}
|
||||||
|
Err(broadcast::error::RecvError::Closed) => break,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
msg = socket.recv() => {
|
||||||
|
match msg {
|
||||||
|
Some(Ok(Message::Close(_))) | None => break,
|
||||||
|
Some(Ok(Message::Ping(p))) => { let _ = socket.send(Message::Pong(p)).await; }
|
||||||
|
Some(Ok(other)) => debug!(?other, "ws client message ignored"),
|
||||||
|
Some(Err(e)) => { debug!(error = %e, "ws client error"); break; }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
_ = tokio::time::sleep(Duration::from_secs(30)) => {
|
||||||
|
if socket.send(Message::Ping(Vec::new())).await.is_err() {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
info!("ws client disconnected");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use axum::body::Body;
|
||||||
|
|
||||||
|
fn req_with(headers: &[(&str, &str)], uri: &str) -> Request {
|
||||||
|
let mut b = Request::builder().uri(uri);
|
||||||
|
for (k, v) in headers {
|
||||||
|
b = b.header(*k, *v);
|
||||||
|
}
|
||||||
|
b.body(Body::empty()).unwrap()
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn the_token_is_read_from_all_three_places() {
|
||||||
|
assert_eq!(
|
||||||
|
extract_token(&req_with(&[("authorization", "Bearer abc")], "/events")).as_deref(),
|
||||||
|
Some("abc")
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
extract_token(&req_with(&[("authorization", "bearer abc")], "/events")).as_deref(),
|
||||||
|
Some("abc")
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
extract_token(&req_with(&[("x-api-key", "abc")], "/events")).as_deref(),
|
||||||
|
Some("abc")
|
||||||
|
);
|
||||||
|
// The query form exists for browser WebSocket clients, which cannot set a handshake header.
|
||||||
|
assert_eq!(
|
||||||
|
extract_token(&req_with(&[], "/ws?token=abc")).as_deref(),
|
||||||
|
Some("abc")
|
||||||
|
);
|
||||||
|
assert_eq!(extract_token(&req_with(&[], "/events")), None);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `Authorization: abc` with no scheme is not a token. Accepting it would make the header's
|
||||||
|
/// grammar optional, and a caller that got it wrong would work here and nowhere else.
|
||||||
|
#[test]
|
||||||
|
fn a_bare_authorization_value_is_not_a_token() {
|
||||||
|
assert_eq!(
|
||||||
|
extract_token(&req_with(&[("authorization", "abc")], "/x")),
|
||||||
|
None
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn constant_time_eq_still_compares_correctly() {
|
||||||
|
assert!(constant_time_eq(b"abc", b"abc"));
|
||||||
|
assert!(!constant_time_eq(b"abc", b"abd"));
|
||||||
|
assert!(!constant_time_eq(b"abc", b"abcd"));
|
||||||
|
assert!(constant_time_eq(b"", b""));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The two RPC failures must not collapse into one code: "the game is down" and "the game is up
|
||||||
|
/// and slow" have different fixes, and the website's client branches on the status.
|
||||||
|
#[test]
|
||||||
|
fn rpc_failures_map_to_distinct_codes() {
|
||||||
|
assert_eq!(
|
||||||
|
respond(Err(RpcError::NoPlugin)).status(),
|
||||||
|
StatusCode::SERVICE_UNAVAILABLE
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
respond(Err(RpcError::Timeout)).status(),
|
||||||
|
StatusCode::GATEWAY_TIMEOUT
|
||||||
|
);
|
||||||
|
assert_eq!(respond(Ok(json!({"ok": true}))).status(), StatusCode::OK);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn uptime_reads_as_a_human_would_write_it() {
|
||||||
|
assert_eq!(format_uptime(Duration::from_secs(90)), "1m");
|
||||||
|
assert_eq!(format_uptime(Duration::from_secs(3 * 3600 + 120)), "3h 2m");
|
||||||
|
assert_eq!(
|
||||||
|
format_uptime(Duration::from_secs(2 * 86400 + 3600)),
|
||||||
|
"2d 1h"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A zero timestamp means "never", and must not render as 1970.
|
||||||
|
#[test]
|
||||||
|
fn a_never_timestamp_is_none_rather_than_the_epoch() {
|
||||||
|
assert_eq!(iso_ms(0), None);
|
||||||
|
assert_eq!(iso_ms(-1), None);
|
||||||
|
assert!(iso_ms(1_757_000_000_000).unwrap().starts_with("2025-"));
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user