feat(qbzd): tiny_http control plane, Origin shield + opt-in token, ping/info/status and CLI diagnostics

cc
This commit is contained in:
vicrodh committed 2026-07-14 15:59:17 -06:00
1 parent 9597573a61
commit 286cb51957
9 files changed
+1441 -65

No files matched your search

+382 -3
View File
@@ -1,4 +1,383 @@
// crates/qbzd/src/api/mod.rs — HTTP surface (stub). Route handlers and the
// server itself land in T6/T7/T8; this module currently only carries the
// `/api/status` response contract (T2, 02-cli-and-api.md §3.3.3).
// crates/qbzd/src/api/mod.rs — the HTTP control plane (02-cli-and-api.md §3).
//
// tiny_http 0.12, single listener, thread-per-connection collapsed to ONE
// serving thread (requests are handled inline, serialized — a control plane
// sees a handful of clients, never a load). Two mechanisms qualify the
// otherwise-open surface (§3.1.2):
// - Origin shield (ALWAYS on): any request carrying an `Origin` header is
// refused `403 origin_forbidden` before routing — CLI/curl/scripts send no
// `Origin`, browsers do (CSRF / DNS-rebinding guard at zero UX cost).
// - Opt-in `[server] token`: when set, `Authorization: Bearer <token>` is
// required on every route EXCEPT `GET /api/ping`; a mismatch is
// `401 invalid_token`. Unset = no auth machinery exists at runtime.
//
// The two-call split — `bind` at boot step 5 (stateless, so the foreign-occupant
// diagnosis runs BEFORE the stores/runtime exist), `serve` at boot step 11 — is
// what keeps the 01-architecture.md §8.1 boot order honest.
pub mod status;
use std::io::Cursor;
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use std::time::Instant;
use tiny_http::{Header, Request, Response};
use crate::adapter::DaemonAdapter;
use crate::paths::ProfileRoots;
use crate::state::DaemonShared;
use qbz_app::shell::AppRuntime;
use qbz_audio::settings::AudioSettingsStore;
/// The counted P0 route table (02 §3.2). A route exists only iff a shipped
/// client calls it (§3.1.4). This task lands **3**; T7 adds the 9 playback +
/// now-playing routes, T8 the 4 queue routes, T11 `/api/settings/reload` — the
/// inline `route_table_matches_spec_count` test pins the number so the
/// 68-routes failure shape (a route with no caller) cannot creep back in.
pub const P0_ROUTES: &[(&str, &str)] = &[
("GET", "/api/ping"),
("GET", "/api/info"),
("GET", "/api/status"),
];
/// A socket bound at boot step 5, not yet serving. Wraps the tiny_http server
/// in an `Arc` so the serving thread and the shutdown handle can both hold it
/// (`unblock` from the handle terminates the thread's `incoming_requests`).
pub struct BoundServer {
server: Arc<tiny_http::Server>,
}
/// Everything the route handlers read. Owned by the single serving thread
/// (moved into it by [`serve`]), so it only needs `Send`, never `Sync` — which
/// is why a plain `AudioSettingsStore` (rusqlite `Connection`: Send, not Sync)
/// can live here directly. `token` is the opt-in `[server] token`, read once at
/// boot (`None` = open).
pub struct ApiState {
pub runtime: Arc<AppRuntime<DaemonAdapter>>,
pub shared: Arc<Mutex<DaemonShared>>,
pub roots: ProfileRoots,
pub token: Option<String>,
/// The bound address, echoed verbatim by `/api/info`.
pub bind: String,
/// Handle to the daemon's tokio runtime — the serving thread is a plain
/// `std::thread`, so async core calls (`get_queue_state`) run via
/// `Handle::block_on` (never called from a runtime worker → no panic).
pub rt: tokio::runtime::Handle,
/// Second read-only connection to the daemon-root audio settings DB (WAL
/// allows it alongside the Player's). Supplies `configured_device`/`backend`.
pub audio: AudioSettingsStore,
/// Cached device enumeration for `device_present` (refreshed on a TTL so a
/// `status` poll never re-enumerates CPAL on every call).
pub devices: Mutex<DeviceCache>,
}
/// TTL-cached output-device names for the `device_present` check.
#[derive(Default)]
pub struct DeviceCache {
pub at: Option<Instant>,
pub names: Vec<String>,
}
/// Live serving handle. [`ApiHandle::shutdown`] unblocks the serving thread and
/// joins it — dropping the thread's `ApiState` (and with it the `Arc<AppRuntime>`
/// clone) BEFORE the daemon drops the runtime, preserving the §8.2 audio
/// clock-release ordering (the API thread is one more `Arc<AppRuntime>` holder,
/// exactly like the driver and auth-retry tasks).
pub struct ApiHandle {
server: Arc<tiny_http::Server>,
thread: Option<std::thread::JoinHandle<()>>,
}
impl ApiHandle {
pub fn shutdown(mut self) {
self.server.unblock();
if let Some(t) = self.thread.take() {
let _ = t.join();
}
}
}
/// Why a bind failed — `AddrInUse` is the case the boot step-5 diagnosis probes
/// (foreign qbzd vs another process); everything else is a generic fatal.
#[derive(Debug)]
pub enum BindError {
AddrInUse(SocketAddr),
Other(String),
}
/// Boot step 5 (01 §8.1): bind only — stateless, so the foreign-occupant
/// diagnosis (in `daemon.rs`) runs BEFORE stores (6) and runtime composition (7).
pub fn bind(addr: SocketAddr) -> Result<BoundServer, BindError> {
match tiny_http::Server::http(addr) {
Ok(server) => Ok(BoundServer {
server: Arc::new(server),
}),
Err(e) => Err(classify_bind_error(e, addr)),
}
}
fn classify_bind_error(
e: Box<dyn std::error::Error + Send + Sync + 'static>,
addr: SocketAddr,
) -> BindError {
if let Some(io) = e.downcast_ref::<std::io::Error>() {
if io.kind() == std::io::ErrorKind::AddrInUse {
return BindError::AddrInUse(addr);
}
}
BindError::Other(e.to_string())
}
/// Best-effort occupant probe for the step-5 diagnosis: `GET /api/ping` and
/// check the response identifies as qbzd (`"app":"qbzd"`). Loopback, short
/// timeout, no dependency on reqwest (the CLI's async client is not built here).
pub fn probe_is_qbzd(addr: SocketAddr) -> bool {
use std::io::{Read, Write};
use std::net::TcpStream;
use std::time::Duration;
let mut stream = match TcpStream::connect_timeout(&addr, Duration::from_millis(500)) {
Ok(s) => s,
Err(_) => return false,
};
let _ = stream.set_read_timeout(Some(Duration::from_secs(1)));
let _ = stream.set_write_timeout(Some(Duration::from_secs(1)));
let req = format!(
"GET /api/ping HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n\r\n"
);
if stream.write_all(req.as_bytes()).is_err() {
return false;
}
let _ = stream.flush();
let mut buf = Vec::new();
let _ = stream.take(4096).read_to_end(&mut buf);
let text = String::from_utf8_lossy(&buf);
text.contains("\"app\":\"qbzd\"")
}
/// Boot step 11 (01 §8.1): start serving on the already-bound socket. Requests
/// are handled inline on one thread (serialized). `unblock` (from the handle)
/// ends `incoming_requests` for graceful shutdown.
pub fn serve(server: BoundServer, state: ApiState) -> ApiHandle {
// The counted P0 surface — logged at boot so an operator sees exactly how
// many routes this build exposes (and it anchors the const to production).
log::info!("control plane serving {} P0 route(s)", P0_ROUTES.len());
let srv = server.server;
let srv_handle = srv.clone();
let thread = std::thread::Builder::new()
.name("qbzd-api".into())
.spawn(move || {
for mut req in srv.incoming_requests() {
let resp = route(&state, &mut req);
let _ = req.respond(resp);
}
})
.expect("failed to spawn qbzd api thread");
ApiHandle {
server: srv_handle,
thread: Some(thread),
}
}
fn route(state: &ApiState, req: &mut Request) -> Response<Cursor<Vec<u8>>> {
let method = req.method().as_str().to_owned();
let path = req.url().split('?').next().unwrap_or("").to_owned();
let has_origin = req.headers().iter().any(|h| h.field.equiv("Origin"));
let auth_header = req
.headers()
.iter()
.find(|h| h.field.equiv("Authorization"))
.map(|h| h.value.as_str());
// Origin shield (ALWAYS on) + opt-in [server] token — one pre-routing gate,
// /api/ping exempt from the token only (§3.1.2).
if let Some(reject) = access_gate(has_origin, &method, &path, auth_header, state.token.as_deref())
{
return reject.response();
}
match (method.as_str(), path.as_str()) {
("GET", "/api/ping") => json(
200,
serde_json::json!({"ok": true, "app": "qbzd", "api_version": crate::API_VERSION}),
),
("GET", "/api/info") => status::info(state),
("GET", "/api/status") => status::status(state),
// T7: /api/now-playing + 8 playback routes · T8: 4 queue routes ·
// T11: /api/settings/reload.
_ => err_json(404, "not_found", "unknown route", "see qbzd --help"),
}
}
/// A pre-routing rejection. Carried as a small enum so [`access_gate`] stays a
/// pure decision (unit-testable without a tiny_http `Request`, which has no
/// public constructor) while `route` renders the normative envelope.
enum GateReject {
OriginForbidden,
InvalidToken,
}
impl GateReject {
fn response(&self) -> Response<Cursor<Vec<u8>>> {
match self {
GateReject::OriginForbidden => err_json(
403,
"origin_forbidden",
"requests with an Origin header are refused",
"the control plane is not a browser API",
),
GateReject::InvalidToken => err_json(
401,
"invalid_token",
"missing or wrong bearer token",
"set QBZD_TOKEN or check [server] token in qbzd.toml",
),
}
}
}
/// The pre-routing access decision (02 §3.1.2): Origin shield always on; the
/// opt-in Bearer required on every route except `GET /api/ping` when `token`
/// is `Some`. `None` = open (no auth machinery). Returns `Some(_)` to reject.
fn access_gate(
has_origin: bool,
method: &str,
path: &str,
auth_header: Option<&str>,
token: Option<&str>,
) -> Option<GateReject> {
if has_origin {
return Some(GateReject::OriginForbidden);
}
if let Some(secret) = token {
let is_ping = method == "GET" && path == "/api/ping";
let expected = format!("Bearer {secret}");
let ok = auth_header
.map(|v| constant_time_eq(v.as_bytes(), expected.as_bytes()))
.unwrap_or(false);
if !is_ping && !ok {
return Some(GateReject::InvalidToken);
}
}
None
}
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
}
/// A 2xx JSON response. `pub(crate)` so the per-route handlers in `status.rs`
/// share the exact same envelope framing.
pub(crate) fn json(status: u16, body: serde_json::Value) -> Response<Cursor<Vec<u8>>> {
let bytes = serde_json::to_vec(&body).unwrap_or_else(|_| b"{}".to_vec());
let header = Header::from_bytes(&b"Content-Type"[..], &b"application/json"[..])
.expect("static content-type header");
Response::from_data(bytes)
.with_status_code(status)
.with_header(header)
}
/// The normative error envelope (02 §3.1.3): `{"error":{"code","message","hint"}}`.
/// The CLI keys its exit code off `code` (never raw HTTP status), and every hint
/// names the fix (§1.4 error voice). The G0 addendum's shorthand
/// `{"error":"origin_forbidden"}` is this same nested envelope — the uniform
/// §3.1.3 shape the CLI's `error_from_envelope` reads via `error.code`.
pub(crate) fn err_json(
status: u16,
code: &str,
message: &str,
hint: &str,
) -> Response<Cursor<Vec<u8>>> {
json(status, error_body(code, message, hint))
}
pub(crate) fn error_body(code: &str, message: &str, hint: &str) -> serde_json::Value {
serde_json::json!({"error": {"code": code, "message": message, "hint": hint}})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn route_table_matches_spec_count() {
// 02-cli-and-api.md §3.2 — P0 = exactly 17 routes; grows ONLY with a
// shipped client. This task lands 3; T7 +9; T8 +4; T11 +1.
assert_eq!(P0_ROUTES.len(), 3);
assert!(P0_ROUTES.contains(&("GET", "/api/ping")));
assert!(P0_ROUTES.contains(&("GET", "/api/info")));
assert!(P0_ROUTES.contains(&("GET", "/api/status")));
}
#[test]
fn constant_time_eq_matches_only_identical_slices() {
assert!(constant_time_eq(b"Bearer s3cret", b"Bearer s3cret"));
assert!(!constant_time_eq(b"Bearer s3cret", b"Bearer wrong"));
assert!(!constant_time_eq(b"Bearer s3cret", b"Bearer s3cret-extra"));
assert!(!constant_time_eq(b"", b"x"));
}
fn code(r: Option<GateReject>) -> Option<&'static str> {
r.map(|r| match r {
GateReject::OriginForbidden => "origin_forbidden",
GateReject::InvalidToken => "invalid_token",
})
}
#[test]
fn origin_header_is_refused_on_every_route_including_ping() {
// Step 4(a): an Origin header → 403 origin_forbidden everywhere, even
// /api/ping, and even in the open (token=None) default.
for (m, p) in P0_ROUTES {
assert_eq!(
code(access_gate(true, m, p, None, None)),
Some("origin_forbidden"),
"{m} {p} with Origin must be refused"
);
}
// ...and the Origin shield wins even when a valid Bearer is present.
assert_eq!(
code(access_gate(true, "GET", "/api/ping", Some("Bearer s3cret"), Some("s3cret"))),
Some("origin_forbidden")
);
}
#[test]
fn open_mode_answers_every_route_without_auth() {
// Step 4(b): token=None → no auth machinery; nothing is rejected.
for (m, p) in P0_ROUTES {
assert!(access_gate(false, m, p, None, None).is_none(), "{m} {p} must be open");
}
}
#[test]
fn opt_in_token_rejects_missing_or_wrong_bearer_but_never_ping() {
// Step 4(c): token=Some → missing/wrong bearer is 401 on non-ping routes;
// /api/ping stays 200 (exempt); the correct bearer passes.
let tok = Some("s3cret");
assert_eq!(code(access_gate(false, "GET", "/api/status", None, tok)), Some("invalid_token"));
assert_eq!(
code(access_gate(false, "GET", "/api/status", Some("Bearer nope"), tok)),
Some("invalid_token")
);
assert!(access_gate(false, "GET", "/api/status", Some("Bearer s3cret"), tok).is_none());
// /api/ping answers even with no/ wrong bearer.
assert!(access_gate(false, "GET", "/api/ping", None, tok).is_none());
assert!(access_gate(false, "GET", "/api/ping", Some("Bearer nope"), tok).is_none());
}
#[test]
fn error_envelope_is_the_nested_shape_with_code() {
// 02 §3.1.3 — the on-the-wire shape the CLI reads via `error.code`.
let body = error_body("origin_forbidden", "refused", "not a browser API");
assert_eq!(body["error"]["code"], "origin_forbidden");
assert_eq!(body["error"]["message"], "refused");
assert_eq!(body["error"]["hint"], "not a browser API");
}
}
+247 -55
View File
@@ -4,9 +4,12 @@
// `audio`/`playback`/`network` are placeholders wired by T3 (audio/playback
// driver), T6 (HTTP server + network reachability) and T10 (qconnect report
// tick, already flowing through `DaemonShared::qconnect`).
use serde::Serialize;
use std::io::Cursor;
use crate::state::{AuthState, DaemonShared, LatchedErrors, QconnectStatus};
use serde::Serialize;
use tiny_http::Response;
use crate::state::{AuthState, LatchedErrors, QconnectStatus};
#[derive(Debug, Clone, Serialize)]
pub struct StatusDoc {
@@ -62,77 +65,256 @@ pub struct NetworkStatus {
pub online: bool,
}
/// Stub assembler: fills what T2 has (process facts + the `DaemonShared`
/// snapshot); `audio`/`playback`/`network` stay placeholders until their
/// producing tasks land.
pub fn assemble(shared: &DaemonShared, version: &str, data_root: &str) -> StatusDoc {
// ============================ live handlers (T6) ============================
/// `GET /api/info` (02 §3.3.2) — identity for the CLI's version-skew diagnosis
/// (§1.6, the single sanctioned second request). Deliberately minimal.
pub fn info(state: &super::ApiState) -> Response<Cursor<Vec<u8>>> {
let uptime = state
.shared
.lock()
.ok()
.map(|s| s.started_at.elapsed().as_secs())
.unwrap_or(0);
super::json(
200,
serde_json::json!({
"app": "qbzd",
"version": env!("CARGO_PKG_VERSION"),
"api_version": crate::API_VERSION,
"bind": state.bind,
"uptime_secs": uptime,
"data_root": state.roots.data.display().to_string(),
}),
)
}
/// `GET /api/status` (02 §3.3.3) — the composite daemon status. ALWAYS 200;
/// the CLI maps degradation (needs_auth, missing device) to exit codes.
pub fn status(state: &super::ApiState) -> Response<Cursor<Vec<u8>>> {
let doc = assemble_live(state);
super::json(
200,
serde_json::to_value(&doc).unwrap_or_else(|_| serde_json::json!({})),
)
}
/// Compose [`StatusDoc`] from live sources: `DaemonShared` (auth/qconnect/latched
/// errors/tick age), the Player's sync getters via `get_playback_event`, the
/// queue via an async core call (`block_on` on the daemon runtime — this is a
/// plain serving thread, never a tokio worker, so no panic), and the audio
/// store + TTL device cache for the audio block.
fn assemble_live(state: &super::ApiState) -> StatusDoc {
// 1. snapshot DaemonShared, then DROP the guard before any block_on so the
// mutex is never held across an await point.
let (auth, user_id, subscription, last_errors, qconnect, tick_age, muted, uptime) =
match state.shared.lock() {
Ok(s) => (
s.auth,
s.user_id,
s.subscription.clone(),
s.last_errors.clone(),
s.qconnect.clone(),
s.driver_last_tick.map(|t| t.elapsed().as_millis() as u64),
s.muted,
s.started_at.elapsed().as_secs(),
),
Err(_) => (
AuthState::Restoring,
None,
None,
LatchedErrors::default(),
QconnectStatus::default(),
None,
false,
0,
),
};
// 2. live player snapshot (all sync atomics, folded into one PlaybackEvent).
let player = state.runtime.core().player();
let ev = player.get_playback_event();
let device_open = player.state.current_device().is_some();
// 3. queue — async core read, driven from this non-worker thread.
let queue = state.rt.block_on(state.runtime.core().get_queue_state());
// 4. audio config from the store; device_present from the TTL cache.
let settings = state.audio.get_settings().ok();
let backend = settings.as_ref().and_then(|s| backend_label(s.backend_type));
let configured_device = settings.as_ref().and_then(|s| s.output_device.clone());
let device_present = match &configured_device {
None => true, // system default is always "present"
Some(dev) => device_is_present(state, dev),
};
// 5. playback block. `stopped` when nothing is loaded and the queue has no
// current track; otherwise `playing`/`paused`.
let has_track = queue.current_track.is_some();
let pstate = if ev.is_playing {
"playing"
} else if has_track || player.has_loaded_audio() {
"paused"
} else {
"stopped"
};
let stopped = pstate == "stopped";
let (title, artist, track_id) = match &queue.current_track {
Some(t) => (Some(t.title.clone()), Some(t.artist.clone()), Some(t.id)),
None if ev.track_id != 0 => (None, None, Some(ev.track_id)),
None => (None, None, None),
};
StatusDoc {
version: version.to_string(),
version: env!("CARGO_PKG_VERSION").to_string(),
api_version: crate::API_VERSION,
uptime_secs: shared.started_at.elapsed().as_secs(),
data_root: data_root.to_string(),
driver_tick_age_ms: shared
.driver_last_tick
.map(|t| t.elapsed().as_millis() as u64),
uptime_secs: uptime,
data_root: state.roots.data.display().to_string(),
driver_tick_age_ms: tick_age,
auth: AuthStatus {
state: shared.auth,
user_id: shared.user_id,
subscription: shared.subscription.clone(),
state: auth,
user_id,
subscription,
},
audio: AudioStatus {
backend: None,
configured_device: None,
device_present: false,
device_open: false,
bit_perfect: None,
sample_rate: None,
bit_depth: None,
backend,
configured_device,
device_present,
device_open,
bit_perfect: bitperfect_label(ev.bit_perfect_mode),
sample_rate: ev.sample_rate,
bit_depth: ev.bit_depth,
},
playback: PlaybackStatus {
state: "stopped".to_string(),
track_id: None,
title: None,
artist: None,
position: None,
duration: None,
volume: 0.0,
muted: shared.muted,
queue_len: 0,
state: pstate.to_string(),
track_id: if stopped { None } else { track_id },
title: if stopped { None } else { title },
artist: if stopped { None } else { artist },
position: if stopped { None } else { Some(ev.position) },
duration: if stopped { None } else { Some(ev.duration) },
volume: ev.volume,
muted,
queue_len: queue.total_tracks,
},
qconnect: shared.qconnect.clone(),
qconnect,
network: NetworkStatus { online: true },
last_errors: shared.last_errors.clone(),
last_errors,
}
}
/// `BitPerfectMode` → its serde variant string (02 §3.3.3:
/// `"DirectHardware"|"PluginFallback"|"Disabled"`). `None` = no active stream.
fn bitperfect_label(m: Option<qbz_audio::BitPerfectMode>) -> Option<String> {
use qbz_audio::BitPerfectMode as M;
m.map(|m| {
match m {
M::DirectHardware => "DirectHardware",
M::PluginFallback => "PluginFallback",
M::Disabled => "Disabled",
}
.to_string()
})
}
/// Configured backend → the lowercase label the status block shows. `None`
/// (auto-detect) stays `null` until a stream picks a concrete backend.
fn backend_label(b: Option<qbz_audio::AudioBackendType>) -> Option<String> {
use qbz_audio::AudioBackendType as B;
b.map(|b| {
match b {
B::PipeWire => "pipewire",
B::Alsa => "alsa",
B::Pulse => "pulse",
B::Jack => "jack",
B::SystemDefault => "system",
}
.to_string()
})
}
/// Best-effort presence check against the TTL-cached device enumeration. Exact
/// device identity is refined in T10; here a substring match on either side
/// tolerates the CPAL-name vs `hw:` mismatch without false negatives on a match.
fn device_is_present(state: &super::ApiState, dev: &str) -> bool {
cached_device_names(state)
.iter()
.any(|n| n == dev || n.contains(dev) || dev.contains(n.as_str()))
}
/// Device names, re-enumerated at most every 5 s (a `status` poll must not
/// re-scan CPAL on every call). On enumeration failure the timestamp is still
/// bumped so a broken audio stack is not hammered.
fn cached_device_names(state: &super::ApiState) -> Vec<String> {
use std::time::{Duration, Instant};
let mut cache = match state.devices.lock() {
Ok(c) => c,
Err(_) => return Vec::new(),
};
let fresh = cache
.at
.map(|t| t.elapsed() < Duration::from_secs(5))
.unwrap_or(false);
if !fresh {
if let Ok(sinks) = qbz_audio::output_sinks::list_output_sinks() {
cache.names = sinks.into_iter().map(|s| s.name).collect();
}
cache.at = Some(Instant::now());
}
cache.names.clone()
}
#[cfg(test)]
mod tests {
use super::*;
fn shared_fixture() -> DaemonShared {
DaemonShared {
auth: AuthState::NeedsAuth,
user_id: None,
subscription: None,
/// A NeedsAuth [`StatusDoc`] built directly (no live runtime), matching the
/// 02 §3.3.3 NeedsAuth fragment — the serde-shape contract these tests pin.
fn needs_auth_doc() -> StatusDoc {
StatusDoc {
version: "2.1.0".into(),
api_version: crate::API_VERSION,
uptime_secs: 261_360,
data_root: "/home/pi/.local/share/qbzd".into(),
driver_tick_age_ms: Some(210),
auth: AuthStatus {
state: AuthState::NeedsAuth,
user_id: None,
subscription: None,
},
audio: AudioStatus {
backend: Some("alsa".into()),
configured_device: None,
device_present: true,
device_open: false,
bit_perfect: None,
sample_rate: None,
bit_depth: None,
},
playback: PlaybackStatus {
state: "stopped".into(),
track_id: None,
title: None,
artist: None,
position: None,
duration: None,
volume: 0.0,
muted: false,
queue_len: 0,
},
qconnect: QconnectStatus::default(),
network: NetworkStatus { online: true },
last_errors: LatchedErrors {
stream: None,
auth: Some("token rejected by Qobuz (401) — cleared".into()),
transport: None,
},
driver_last_tick: None,
muted: false,
premute_volume: 1.0,
started_at: std::time::Instant::now(),
startup_warnings: 0,
qconnect: QconnectStatus::default(),
}
}
#[test]
fn assemble_mirrors_the_top_level_status_contract_keys() {
fn status_doc_mirrors_the_top_level_contract_keys() {
// 02-cli-and-api.md §3.3.3 top-level keys, exactly.
let doc = assemble(&shared_fixture(), "2.1.0", "/home/pi/.local/share/qbzd");
let json = serde_json::to_value(&doc).unwrap();
let json = serde_json::to_value(needs_auth_doc()).unwrap();
let obj = json.as_object().unwrap();
for key in [
"version",
@@ -152,9 +334,9 @@ mod tests {
}
#[test]
fn assemble_needs_auth_fragment_matches_spec_example() {
// 02 §3.3.3 NeedsAuth example fragment.
let doc = assemble(&shared_fixture(), "2.1.0", "/home/pi/.local/share/qbzd");
fn needs_auth_fragment_matches_spec_example() {
// 02 §3.3.3 NeedsAuth example fragment + auth.state serde string.
let doc = needs_auth_doc();
assert_eq!(doc.auth.state, AuthState::NeedsAuth);
assert!(doc.auth.user_id.is_none());
assert!(doc.auth.subscription.is_none());
@@ -167,10 +349,20 @@ mod tests {
}
#[test]
fn assemble_carries_data_root_and_version_through_verbatim() {
let doc = assemble(&shared_fixture(), "2.1.0", "/home/pi/.local/share/qbzd");
assert_eq!(doc.version, "2.1.0");
assert_eq!(doc.data_root, "/home/pi/.local/share/qbzd");
assert_eq!(doc.api_version, crate::API_VERSION);
fn audio_block_serializes_the_documented_keys() {
// 02 §3.3.3 audio object — the shape the live assembler fills.
let json = serde_json::to_value(needs_auth_doc()).unwrap();
let audio = json["audio"].as_object().unwrap();
for key in [
"backend",
"configured_device",
"device_present",
"device_open",
"bit_perfect",
"sample_rate",
"bit_depth",
] {
assert!(audio.contains_key(key), "missing audio key: {key}");
}
}
}
+282
View File
@@ -0,0 +1,282 @@
// crates/qbzd/src/cli/client.rs — the stateless HTTP client behind every
// networked CLI verb (02-cli-and-api.md §1.1). One verb = one request; the
// client holds no daemon state beyond target discovery (§1.5).
//
// Target discovery precedence (§1.5): `--host` > `QBZD_HOST` > local default
// `127.0.0.1:8182`. The Bearer is sent ONLY in opt-in mode: when `QBZD_TOKEN`
// is set (remote or local) or, when targeting the LOCAL daemon, the local
// `qbzd.toml` carries `[server] token`. Default: no token anywhere.
use std::time::Duration;
use serde_json::Value;
use crate::config::QbzdConfig;
use crate::paths::ProfileRoots;
/// The frozen exit-code taxonomy (02 §1.3). `exit_code()` is the ONLY source of
/// a networked verb's process exit code; scripts encode these numbers forever.
#[derive(Debug)]
pub enum CliError {
/// exit 3 — connect refused / timeout on the target (carries `host` for the
/// §1.4 daemon-down copy).
Unreachable(String),
/// exit 4 — daemon in NeedsAuth; a Qobuz session is required.
NeedsAuth,
/// exit 5 — audio/device error (device unopenable, volume/seek fixed in DSD).
Device(String),
/// exit 6 — unknown id / index out of range.
NotFound(String),
/// exit 1 — breaking `api_version` skew: the verb refuses politely (§1.6).
ApiSkew { daemon: u32, cli: u32 },
/// exit 1 — any other runtime error (daemon said no / local failure).
Runtime(String),
}
impl CliError {
pub fn exit_code(&self) -> i32 {
match self {
CliError::Runtime(_) | CliError::ApiSkew { .. } => 1,
CliError::Unreachable(_) => 3,
CliError::NeedsAuth => 4,
CliError::Device(_) => 5,
CliError::NotFound(_) => 6,
}
}
}
impl std::fmt::Display for CliError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
use crate::cli::copy;
match self {
CliError::Unreachable(host) => write!(f, "{}", copy::daemon_down(host)),
CliError::NeedsAuth => write!(f, "{}", copy::daemon_up_needs_auth()),
CliError::Device(m) => write!(f, "error: {m}"),
CliError::NotFound(m) => write!(f, "error: {m}"),
CliError::ApiSkew { daemon, cli } => write!(f, "{}", copy::api_version_skew(*daemon, *cli)),
CliError::Runtime(m) => write!(f, "error: {m}"),
}
}
}
impl std::error::Error for CliError {}
/// A resolved target host + whether it is the local daemon (governs whether the
/// local `qbzd.toml` token and the linger check apply).
pub struct Target {
pub addr: String,
pub is_local: bool,
}
/// §1.5 target discovery: `--host` > `QBZD_HOST` > local `127.0.0.1:8182`. An
/// explicit override (flag or env) is treated as remote — only the local
/// default reads the local token / runs the linger check.
pub fn resolve_host(flag: Option<String>) -> Target {
if let Some(h) = flag.filter(|h| !h.is_empty()) {
return Target { addr: normalize_hostport(&h), is_local: false };
}
if let Ok(h) = std::env::var("QBZD_HOST") {
if !h.is_empty() {
return Target { addr: normalize_hostport(&h), is_local: false };
}
}
Target { addr: "127.0.0.1:8182".into(), is_local: true }
}
/// Append the default port when the operator gave a bare host.
fn normalize_hostport(h: &str) -> String {
if h.contains(':') {
h.to_string()
} else {
format!("{h}:8182")
}
}
/// §1.5 token discovery: `QBZD_TOKEN` (remote or local), else — only when
/// targeting the local daemon — the local `qbzd.toml` `[server] token`.
fn resolve_token(target: &Target, roots: &ProfileRoots) -> Option<String> {
if let Ok(t) = std::env::var("QBZD_TOKEN") {
if !t.is_empty() {
return Some(t);
}
}
if target.is_local {
if let Ok((cfg, _)) = QbzdConfig::load(&roots.config.join("qbzd.toml")) {
return cfg.server.token;
}
}
None
}
/// A thin skin over one HTTP request to the daemon.
pub struct ApiClient {
base: String,
host: String,
is_local: bool,
token: Option<String>,
client: reqwest::Client,
}
impl ApiClient {
/// Discover the target + token per §1.5 and build the client.
pub fn new(host: Option<String>, roots: &ProfileRoots) -> Self {
let target = resolve_host(host);
let token = resolve_token(&target, roots);
let client = reqwest::Client::builder()
.connect_timeout(Duration::from_secs(2))
.timeout(Duration::from_secs(10))
.build()
.expect("failed to build reqwest client");
ApiClient {
base: format!("http://{}", target.addr),
host: target.addr,
is_local: target.is_local,
token,
client,
}
}
/// The target `ip:port` (for the daemon-down copy + the status header line).
pub fn host(&self) -> &str {
&self.host
}
/// Whether the target is the local daemon (governs the linger check).
pub fn is_local(&self) -> bool {
self.is_local
}
pub async fn get(&self, path: &str) -> Result<Value, CliError> {
let req = self.bearer(self.client.get(format!("{}{}", self.base, path)));
self.send(req).await
}
/// P0 mutation transport — first consumer is the T7 transport verbs.
#[allow(dead_code)] // wired by the T7 play/pause/… verbs
pub async fn post(&self, path: &str, body: Value) -> Result<Value, CliError> {
let req = self.bearer(self.client.post(format!("{}{}", self.base, path)).json(&body));
self.send(req).await
}
fn bearer(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
match &self.token {
Some(t) => req.bearer_auth(t),
None => req,
}
}
async fn send(&self, req: reqwest::RequestBuilder) -> Result<Value, CliError> {
let resp = req.send().await.map_err(|e| self.classify_transport(e))?;
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
if status.is_success() {
match serde_json::from_str::<Value>(&text) {
Ok(v) => Ok(v),
// Unparseable 2xx body → the §1.6 sanctioned second request.
Err(_) => Err(self.diagnose_skew().await),
}
} else {
match serde_json::from_str::<Value>(&text) {
Ok(v) => Err(error_from_envelope(&v)),
Err(_) => Err(self.diagnose_skew().await),
}
}
}
fn classify_transport(&self, e: reqwest::Error) -> CliError {
if e.is_connect() || e.is_timeout() {
CliError::Unreachable(self.host.clone())
} else {
CliError::Runtime(format!("request failed: {e}"))
}
}
/// §1.6: an unreadable body / unknown envelope is the only trigger for the
/// single sanctioned second request, `GET /api/info` — the stable identity
/// route. `api_version` mismatch → refuse politely; otherwise a plain error.
async fn diagnose_skew(&self) -> CliError {
if let Some(api) = self.info_api_version().await {
if api != crate::API_VERSION {
return CliError::ApiSkew {
daemon: api,
cli: crate::API_VERSION,
};
}
}
CliError::Runtime("daemon returned an unreadable response".to_string())
}
async fn info_api_version(&self) -> Option<u32> {
let req = self.bearer(self.client.get(format!("{}/api/info", self.base)));
let resp = req.send().await.ok()?;
let v: Value = resp.json().await.ok()?;
v.get("api_version").and_then(|a| a.as_u64()).map(|a| a as u32)
}
}
/// Map an error envelope's `code` (02 §3.1.3) to the frozen exit taxonomy. The
/// CLI keys off `code`, never raw HTTP status. `origin_forbidden`/`invalid_token`
/// and anything unrecognized → exit 1.
fn error_from_envelope(v: &Value) -> CliError {
let err = v.get("error");
let code = err
.and_then(|e| e.get("code"))
.and_then(|c| c.as_str())
.unwrap_or("");
let message = err
.and_then(|e| e.get("message"))
.and_then(|m| m.as_str())
.unwrap_or("daemon returned an error")
.to_string();
match code {
"needs_auth" => CliError::NeedsAuth,
"not_found" => CliError::NotFound(message),
"audio_unavailable" | "volume_fixed_dsd" | "seek_unsupported_dsd" | "device_error" => {
CliError::Device(message)
}
_ => CliError::Runtime(message),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn exit_codes_match_the_frozen_table() {
// 02-cli-and-api.md §1.3.
assert_eq!(CliError::Unreachable("x".into()).exit_code(), 3);
assert_eq!(CliError::NeedsAuth.exit_code(), 4);
assert_eq!(CliError::Device("x".into()).exit_code(), 5);
assert_eq!(CliError::NotFound("x".into()).exit_code(), 6);
assert_eq!(CliError::ApiSkew { daemon: 2, cli: 1 }.exit_code(), 1);
assert_eq!(CliError::Runtime("x".into()).exit_code(), 1);
}
#[test]
fn error_envelope_maps_code_to_exit() {
let needs = serde_json::json!({"error": {"code": "needs_auth", "message": "no"}});
assert_eq!(error_from_envelope(&needs).exit_code(), 4);
let nf = serde_json::json!({"error": {"code": "not_found", "message": "no"}});
assert_eq!(error_from_envelope(&nf).exit_code(), 6);
let dev = serde_json::json!({"error": {"code": "volume_fixed_dsd", "message": "no"}});
assert_eq!(error_from_envelope(&dev).exit_code(), 5);
// origin_forbidden / invalid_token / unknown all fall to 1.
let origin = serde_json::json!({"error": {"code": "origin_forbidden", "message": "no"}});
assert_eq!(error_from_envelope(&origin).exit_code(), 1);
let tok = serde_json::json!({"error": {"code": "invalid_token", "message": "no"}});
assert_eq!(error_from_envelope(&tok).exit_code(), 1);
}
#[test]
fn host_discovery_appends_default_port_and_flags_local() {
assert_eq!(normalize_hostport("192.168.0.40"), "192.168.0.40:8182");
assert_eq!(normalize_hostport("192.168.0.40:9000"), "192.168.0.40:9000");
// An explicit flag is remote; the bare default is local.
assert!(!resolve_host(Some("192.168.0.40".into())).is_local);
// (Env-free path.) The default target is local.
let t = resolve_host(None);
if std::env::var("QBZD_HOST").is_err() {
assert!(t.is_local);
assert_eq!(t.addr, "127.0.0.1:8182");
}
}
}
+89
View File
@@ -39,3 +39,92 @@ pub fn logout_success(daemon_nudged: bool) -> String {
"logged out".to_string()
}
}
// ============================ error voice (02 §1.4) ============================
// Verbatim §1.4 / §2.2 / §6.3 copy — "modulo interpolated values". Every message
// ends in one to three `→` fix lines; changing the wording is a spec violation.
/// Daemon unreachable — exit 3 (02 §1.4). `host` is the target `ip:port`.
pub fn daemon_down(host: &str) -> String {
format!(
"error: daemon not reachable at {host}
→ is it running? systemctl --user status qbzd
→ just installed? systemctl --user enable --now qbzd
→ different host? qbzd --host <ip>:<port> ... or export QBZD_HOST=<ip>:<port>"
)
}
/// Daemon up but not logged in — exit 4 (02 §1.4). The down-vs-unhealthy
/// distinction: the daemon answered, the Qobuz session is what's missing.
/// Consumed by the T7 transport verbs that hit a NeedsAuth daemon; `status`
/// renders the composite block instead.
#[allow(dead_code)] // wired by the T7 verbs (now/play/…) that refuse on NeedsAuth
pub fn daemon_up_needs_auth() -> String {
"error: daemon is running but not logged in to Qobuz
→ log in: qbzd login
→ have a bundle? qbzd settings import qbz-settings-20260714.qbzb --include-auth"
.to_string()
}
/// Linger-off warning (02 §1.4; NOT an error) — printed by `qbzd status` on the
/// daemon box when `loginctl show-user $USER -p Linger` reports `Linger=no`.
pub fn linger_off(user: &str) -> String {
format!(
"warning: linger is off for user '{user}' — the daemon stops when you log out
→ keep it running: sudo loginctl enable-linger {user}"
)
}
/// Volume fixed under DSD-direct — exit 5 (02 §1.4). Consumed by the T7
/// `volume`/`mute` verbs; defined here so all four §1.4 blocks live in one place.
#[allow(dead_code)] // wired by the T7 volume/mute verbs
pub fn volume_fixed_dsd() -> String {
"error: volume is fixed in DSD-direct mode (bit-perfect passthrough)
→ to get software volume, set DSD mode to \"convert\": qbzd setup (Audio screen)"
.to_string()
}
/// Foreign occupant on the control port that is NOT qbzd — printed by the daemon
/// at boot step 5 (02 §2.2, verbatim). `port` is interpolated.
pub fn port_in_use(port: u16) -> String {
format!(
"error: port {port} is in use by another process (not qbzd)
→ change the port: edit [server].port in ~/.config/qbzd/qbzd.toml"
)
}
/// A DIFFERENT qbzd already answering on the port while our instance lock is on
/// another data root (a stale foreign root). Boot step 5 (02 §8.1-5).
pub fn foreign_qbzd(addr: &str) -> String {
format!(
"error: another qbzd is already answering on {addr} (the instance lock said this root is free — stale foreign root?)
→ find it: ss -ltnp | grep {addr}
→ or change the port: edit [server].port in ~/.config/qbzd/qbzd.toml"
)
}
/// LAN-exposure warning (02 §6.3, verbatim) — printed by the daemon at boot when
/// bound to a non-loopback address, and by the TUI Network screen (T13).
pub fn lan_bind_warning(addr: &str) -> String {
format!(
"warning: control API is bound to {addr} — reachable from your network
anyone on your LAN can control playback (like Sonos or Chromecast)
→ set [server] token in qbzd.toml to require a shared secret"
)
}
/// Version skew — daemon and CLI run different bin semvers (02 §1.6). A warning:
/// `status` still renders. `daemon`/`cli` are the two `version` strings.
pub fn version_skew(daemon: &str, cli: &str) -> String {
format!(
"warning: daemon runs {daemon}, this CLI is {cli}
→ restart the daemon: systemctl --user restart qbzd"
)
}
/// Breaking api_version skew (02 §1.6) — the verb refuses politely, exit 1.
pub fn api_version_skew(daemon: u32, cli: u32) -> String {
format!(
"error: daemon speaks api v{daemon}, this CLI speaks v{cli} — update so both ends run the same package version"
)
}
+2
View File
@@ -3,4 +3,6 @@
// The CLI is a stateless renderer (02-cli-and-api.md §1.1); the copy strings it
// prints are normative (§1.4 error voice, §2.2 per-verb output). Keeping them in
// one place lets the spec and the code diff cleanly.
pub mod client;
pub mod copy;
pub mod status;
+332
View File
@@ -0,0 +1,332 @@
// crates/qbzd/src/cli/status.rs — the `status` and `ping` verbs (02 §2.2).
//
// Both render an already-parsed API payload; neither holds state. `status` also
// runs the version-skew check (§1.6, from the /api/status payload — it carries
// `version` + `api_version`, so it needs no /api/info fallback) and, on the
// daemon box, the linger check (§1.4). Exit codes come from the frozen table
// (§1.3): 0 healthy · 3 unreachable · 4 needs_auth · 5 device unopenable.
use serde_json::Value;
use crate::cli::client::ApiClient;
use crate::cli::copy;
use crate::paths::ProfileRoots;
/// `qbzd ping` — liveness. Human `pong`; `--json` the raw body. Exit 0 · 3.
pub async fn ping(host: Option<String>, json: bool, roots: &ProfileRoots) -> i32 {
let client = ApiClient::new(host, roots);
match client.get("/api/ping").await {
Ok(v) => {
if json {
println!("{}", serde_json::to_string(&v).unwrap_or_default());
} else {
println!("pong");
}
0
}
Err(e) => {
eprintln!("{e}");
e.exit_code()
}
}
}
/// `qbzd status` — THE diagnostic. Human composite block; `--json` raw payload.
/// Exit 0 healthy · 3 unreachable · 4 needs_auth · 5 device unopenable.
pub async fn status(host: Option<String>, json: bool, roots: &ProfileRoots) -> i32 {
let client = ApiClient::new(host, roots);
let payload = match client.get("/api/status").await {
Ok(v) => v,
Err(e) => {
eprintln!("{e}");
return e.exit_code();
}
};
// Version skew (§1.6): breaking api_version mismatch refuses; a semver-only
// mismatch is a warning that does not stop the render.
let daemon_api = payload.get("api_version").and_then(|a| a.as_u64()).unwrap_or(0) as u32;
if daemon_api != crate::API_VERSION {
eprintln!("{}", copy::api_version_skew(daemon_api, crate::API_VERSION));
return 1;
}
let cli_ver = env!("CARGO_PKG_VERSION");
if let Some(daemon_ver) = payload.get("version").and_then(|v| v.as_str()) {
if !daemon_ver.is_empty() && daemon_ver != cli_ver {
eprintln!("{}", copy::version_skew(daemon_ver, cli_ver));
}
}
if json {
println!("{}", serde_json::to_string(&payload).unwrap_or_default());
} else {
print!("{}", render(&payload, client.host()));
}
// Linger check on the daemon box only (§1.4) — a warning, never fatal.
if client.is_local() {
if let Some(w) = linger_warning() {
eprintln!("{w}");
}
}
exit_from_state(&payload)
}
/// 4 needs_auth · 5 configured device not present · else 0 (§1.3). Auth gates
/// before device: a login is the more common fix.
fn exit_from_state(p: &Value) -> i32 {
let auth = str_at(p, &["auth", "state"]);
if auth == "needs_auth" {
return 4;
}
let configured = p
.pointer("/audio/configured_device")
.map(|v| !v.is_null())
.unwrap_or(false);
let present = p
.pointer("/audio/device_present")
.and_then(|v| v.as_bool())
.unwrap_or(true);
if configured && !present {
return 5;
}
0
}
/// The §2.2 composite block. `host` is the target (the payload has no `bind`).
fn render(p: &Value, host: &str) -> String {
let version = str_at(p, &["version"]);
let api = p.get("api_version").and_then(|a| a.as_u64()).unwrap_or(0);
let uptime = fmt_uptime(p.get("uptime_secs").and_then(|u| u.as_u64()).unwrap_or(0));
let data_root = str_at(p, &["data_root"]);
let mut out = String::new();
out.push_str(&format!(
"qbzd {version} · api v{api} · up {uptime} · {host} · data {data_root}\n"
));
out.push_str(&format!("auth : {}\n", render_auth(p)));
out.push_str(&format!("audio : {}\n", render_audio(p)));
out.push_str(&format!("playback : {}\n", render_playback(p)));
out.push_str(&format!("qconnect : {}\n", render_qconnect(p)));
out.push_str(&format!(
"network : {}\n",
if p.pointer("/network/online").and_then(|v| v.as_bool()).unwrap_or(false) {
"online"
} else {
"offline"
}
));
out.push_str(&format!("last error: {}\n", render_last_error(p)));
out
}
fn render_auth(p: &Value) -> String {
match str_at(p, &["auth", "state"]).as_str() {
"logged_in" => {
let user = p.pointer("/auth/user_id").and_then(|v| v.as_u64());
let sub = p.pointer("/auth/subscription").and_then(|v| v.as_str());
match (user, sub) {
(Some(u), Some(s)) => format!("logged in (user {u}, {s})"),
(Some(u), None) => format!("logged in (user {u})"),
_ => "logged in".to_string(),
}
}
"restoring" => "restoring session…".to_string(),
_ => "not logged in".to_string(),
}
}
fn render_audio(p: &Value) -> String {
let backend = p.pointer("/audio/backend").and_then(|v| v.as_str());
let device = p.pointer("/audio/configured_device").and_then(|v| v.as_str());
let present = p.pointer("/audio/device_present").and_then(|v| v.as_bool()).unwrap_or(false);
let bit_perfect = p.pointer("/audio/bit_perfect").and_then(|v| v.as_str());
let sr = p.pointer("/audio/sample_rate").and_then(|v| v.as_u64());
let bd = p.pointer("/audio/bit_depth").and_then(|v| v.as_u64());
let mut parts: Vec<String> = Vec::new();
let head = match (backend, device) {
(Some(b), Some(d)) => format!("{b} {d}"),
(Some(b), None) => format!("{b} (system default)"),
(None, Some(d)) => d.to_string(),
(None, None) => "system default".to_string(),
};
parts.push(head);
parts.push(if present { "present".into() } else { "not present".into() });
if let Some(bp) = bit_perfect {
parts.push(format!("bit-perfect: {bp}"));
}
if let (Some(sr), Some(bd)) = (sr, bd) {
parts.push(format!("{sr} Hz / {bd}-bit"));
}
parts.join(" · ")
}
fn render_playback(p: &Value) -> String {
let state = str_at(p, &["playback", "state"]);
let queue = p.pointer("/playback/queue_len").and_then(|v| v.as_u64()).unwrap_or(0);
if state == "stopped" {
return format!("stopped · queue {queue}");
}
let title = p.pointer("/playback/title").and_then(|v| v.as_str());
let artist = p.pointer("/playback/artist").and_then(|v| v.as_str());
let pos = p.pointer("/playback/position").and_then(|v| v.as_u64()).unwrap_or(0);
let dur = p.pointer("/playback/duration").and_then(|v| v.as_u64()).unwrap_or(0);
let vol = p.pointer("/playback/volume").and_then(|v| v.as_f64()).unwrap_or(0.0);
let muted = p.pointer("/playback/muted").and_then(|v| v.as_bool()).unwrap_or(false);
let track = match (title, artist) {
(Some(t), Some(a)) => format!("\"{t}\" — {a}"),
(Some(t), None) => format!("\"{t}\""),
_ => "(unknown track)".to_string(),
};
let vol_str = if muted {
"muted".to_string()
} else {
format!("vol {}%", (vol * 100.0).round() as i64)
};
format!(
"{state} · {track} · {} / {} · {vol_str} · queue {queue}",
fmt_mmss(pos),
fmt_mmss(dur)
)
}
fn render_qconnect(p: &Value) -> String {
let enabled = p.pointer("/qconnect/enabled").and_then(|v| v.as_bool()).unwrap_or(false);
if !enabled {
return "off".to_string();
}
let state = p.pointer("/qconnect/state").and_then(|v| v.as_str()).unwrap_or("");
let session = p.pointer("/qconnect/session_active").and_then(|v| v.as_bool()).unwrap_or(false);
let name = p.pointer("/qconnect/device_name").and_then(|v| v.as_str()).unwrap_or("");
let mut parts = vec![if state.is_empty() { "enabled".to_string() } else { state.to_string() }];
if session {
parts.push("session active".to_string());
}
if !name.is_empty() {
parts.push(format!("name \"{name}\""));
}
parts.join(" · ")
}
fn render_last_error(p: &Value) -> String {
for key in ["stream", "auth", "transport"] {
if let Some(m) = p.pointer(&format!("/last_errors/{key}")).and_then(|v| v.as_str()) {
if !m.is_empty() {
return format!("{key}: {m}");
}
}
}
"none".to_string()
}
/// `loginctl show-user $USER -p Linger` → the §1.4 linger warning on `Linger=no`.
/// Any failure (no loginctl, no session) → no warning.
fn linger_warning() -> Option<String> {
let user = std::env::var("USER").ok().filter(|u| !u.is_empty())?;
let out = std::process::Command::new("loginctl")
.args(["show-user", &user, "-p", "Linger"])
.output()
.ok()?;
let text = String::from_utf8_lossy(&out.stdout);
if text.trim() == "Linger=no" {
Some(copy::linger_off(&user))
} else {
None
}
}
fn str_at(p: &Value, path: &[&str]) -> String {
let mut cur = p;
for k in path {
match cur.get(k) {
Some(v) => cur = v,
None => return String::new(),
}
}
cur.as_str().unwrap_or("").to_string()
}
fn fmt_mmss(secs: u64) -> String {
format!("{}:{:02}", secs / 60, secs % 60)
}
fn fmt_uptime(secs: u64) -> String {
let days = secs / 86_400;
let hours = (secs % 86_400) / 3_600;
let mins = (secs % 3_600) / 60;
if days > 0 {
format!("{days}d {hours}h")
} else if hours > 0 {
format!("{hours}h {mins}m")
} else {
format!("{mins}m")
}
}
#[cfg(test)]
mod tests {
use super::*;
fn logged_in_payload() -> Value {
serde_json::json!({
"version": "2.1.0", "api_version": 1, "uptime_secs": 259_200,
"data_root": "/home/pi/.local/share/qbzd", "driver_tick_age_ms": 210,
"auth": {"state": "logged_in", "user_id": 1234567, "subscription": "studio"},
"audio": {"backend": "alsa", "configured_device": "hw:CARD=D30,DEV=0",
"device_present": true, "device_open": true,
"bit_perfect": "DirectHardware", "sample_rate": 192000, "bit_depth": 24},
"playback": {"state": "playing", "track_id": 176544871, "title": "Spain",
"artist": "Chick Corea", "position": 192, "duration": 581,
"volume": 0.8, "muted": false, "queue_len": 14},
"qconnect": {"enabled": true, "state": "connected", "device_name": "QBZ (kitchen-pi)",
"session_active": true, "last_transport_reconnect": null},
"network": {"online": true},
"last_errors": {"stream": null, "auth": null, "transport": null}
})
}
#[test]
fn healthy_status_exits_zero() {
assert_eq!(exit_from_state(&logged_in_payload()), 0);
}
#[test]
fn needs_auth_exits_four() {
let mut p = logged_in_payload();
p["auth"]["state"] = serde_json::json!("needs_auth");
assert_eq!(exit_from_state(&p), 4);
}
#[test]
fn configured_but_absent_device_exits_five() {
let mut p = logged_in_payload();
p["audio"]["device_present"] = serde_json::json!(false);
assert_eq!(exit_from_state(&p), 5);
// system default (no configured device) never trips exit 5.
let mut sysdef = logged_in_payload();
sysdef["audio"]["configured_device"] = serde_json::Value::Null;
sysdef["audio"]["device_present"] = serde_json::json!(false);
assert_eq!(exit_from_state(&sysdef), 0);
}
#[test]
fn render_covers_the_composite_block() {
let block = render(&logged_in_payload(), "127.0.0.1:8182");
assert!(block.contains("qbzd 2.1.0 · api v1 · up 3d 0h · 127.0.0.1:8182"), "{block}");
assert!(block.contains("auth : logged in (user 1234567, studio)"), "{block}");
assert!(block.contains("alsa hw:CARD=D30,DEV=0 · present · bit-perfect: DirectHardware · 192000 Hz / 24-bit"), "{block}");
assert!(block.contains("playback : playing · \"Spain\" — Chick Corea · 3:12 / 9:41 · vol 80% · queue 14"), "{block}");
assert!(block.contains("qconnect : connected · session active · name \"QBZ (kitchen-pi)\""), "{block}");
assert!(block.contains("last error: none"), "{block}");
}
#[test]
fn stopped_playback_renders_queue_only() {
let mut p = logged_in_payload();
p["playback"]["state"] = serde_json::json!("stopped");
let line = render_playback(&p);
assert_eq!(line, "stopped · queue 14");
}
}
+21
View File
@@ -18,6 +18,12 @@ pub struct QbzdConfig {
pub struct ServerCfg {
pub bind: String,
pub port: u16,
/// Opt-in shared secret (02 §3.1.2). Default `None` = the control plane is
/// UNAUTHENTICATED (loopback and LAN alike). When set, every route except
/// `GET /api/ping` requires `Authorization: Bearer <token>`; a mismatch is
/// `401 invalid_token`. A plain config value the user writes — there is no
/// generated file and no rotation verb (rotate = edit this + restart).
pub token: Option<String>,
}
#[derive(Debug, Deserialize, Serialize)]
#[serde(default)]
@@ -35,6 +41,7 @@ impl Default for ServerCfg {
Self {
bind: "127.0.0.1".into(),
port: 8182,
token: None, // open by default (02 §3.1.2)
}
}
}
@@ -69,6 +76,7 @@ const KNOWN: &[(&str, &str)] = &[
("", "data_root"),
("server", "bind"),
("server", "port"),
("server", "token"),
("log", "level"),
("mpris", "enabled"),
];
@@ -130,4 +138,17 @@ mod tests {
let (_c, warns) = QbzdConfig::from_str("[server]\nbindd = \"0.0.0.0\"\n").unwrap();
assert_eq!(warns, vec!["[server].bindd".to_string()]);
}
#[test]
fn server_token_defaults_none_and_parses_when_set() {
// 02-cli-and-api.md §3.1.2: `[server] token` is opt-in — absent = None
// (open control plane); present = the shared secret, no warning.
let (open, warns) = QbzdConfig::from_str("").unwrap();
assert_eq!(open.server.token, None);
assert!(warns.is_empty());
let (secured, warns) =
QbzdConfig::from_str("[server]\ntoken = \"s3cret\"\n").unwrap();
assert_eq!(secured.server.token.as_deref(), Some("s3cret"));
assert!(warns.is_empty(), "known key must not warn: {warns:?}");
}
}
+73 -3
View File
@@ -50,9 +50,27 @@ pub async fn run(roots: ProfileRoots, cfg: QbzdConfig, warns: Vec<String>) -> Re
// not the port, protects the single-device_uuid / single-session.db
// invariants. A second daemon on the same root is diagnosed → exit 3.
let _lock = InstanceLock::acquire(&roots.data).map_err(diagnose_lock)?;
// 5. port bind + foreign-occupant diagnosis (probe with GET /api/ping) is a
// stateless step that lands in T6 (api::bind); serving starts at step 11.
// Until T6 the daemon runs API-less.
// 5. port bind + foreign-occupant diagnosis — STATELESS, so it runs BEFORE
// stores (6) and runtime composition (7) per the §8.1 order. On a bind
// conflict the occupant is probed with GET /api/ping: a qbzd answer means
// a stale foreign root (the lock said this root was free), anything else
// the §2.2 "another process" copy. The socket is bound here but not served
// until step 11 — connections queue in the listen backlog through boot.
let bind_addr = resolve_bind_addr(&cfg)?;
let bound = match crate::api::bind(bind_addr) {
Ok(b) => b,
Err(crate::api::BindError::AddrInUse(addr)) => return Err(diagnose_port_conflict(addr)),
Err(crate::api::BindError::Other(msg)) => {
return Err(format!(
"error: could not bind the control API on {bind_addr}: {msg}\n → check [server] bind/port in ~/.config/qbzd/qbzd.toml"
));
}
};
if !bind_addr.ip().is_loopback() {
// §6.3 verbatim — reachable-from-LAN warning (also shown by the TUI).
eprintln!("{}", crate::cli::copy::lan_bind_warning(&bind_addr.to_string()));
log::warn!("control API bound to non-loopback {bind_addr}");
}
// 6.-9. compose stores + runtime + restore credentials + restore session.
let mut booted = boot(&roots, &cfg, warns.len()).await?;
@@ -74,6 +92,27 @@ pub async fn run(roots: ProfileRoots, cfg: QbzdConfig, warns: Vec<String>) -> Re
shutdown_rx,
));
// 11. HTTP serve (02 §3) on the already-bound socket. `ApiState` carries a
// second read-only audio-store connection (WAL) for the status audio
// block, the tokio handle for the async queue read, and the opt-in
// [server] token (None = open). 12. QConnect (T9/T10) splices after this.
let api_audio = qbz_audio::settings::AudioSettingsStore::new_at(&roots.data)
.map_err(|e| format!("error: could not open the audio settings store for the API: {e}"))?;
let api = crate::api::serve(
bound,
crate::api::ApiState {
runtime: booted.runtime.clone(),
shared: booted.shared.clone(),
roots: roots.clone(),
token: cfg.server.token.clone(),
bind: bind_addr.to_string(),
rt: tokio::runtime::Handle::current(),
audio: api_audio,
devices: std::sync::Mutex::new(crate::api::DeviceCache::default()),
},
);
log::info!("control API listening on {bind_addr}");
// 13. park on SIGTERM/SIGINT. NO startup audio "hygiene": both candidate
// fns are verified no-ops from a fresh process and re-adding them is the
// documented skeptic-correction #1 trap (§8.1).
@@ -97,6 +136,11 @@ pub async fn run(roots: ProfileRoots, cfg: QbzdConfig, warns: Vec<String>) -> Re
retry.abort();
let _ = retry.await;
}
// Stop the API thread and JOIN it: its `ApiState` holds an `Arc<AppRuntime>`
// clone, which must drop before `drop(booted)` releases the audio device
// ahead of the #521 pair — the same ordering constraint as the driver and
// auth-retry tasks (§8.2).
api.shutdown();
// Release the audio device by dropping the runtime (its Player) BEFORE the
// #521 pair (§8.2 step 3 precedes step 4).
drop(booted);
@@ -370,6 +414,32 @@ fn spawn_auth_retry(
})
}
/// Resolve `[server] bind:port` to a `SocketAddr` (01 §10.1). A malformed value
/// is a fatal boot error that names the fix (exit 1).
fn resolve_bind_addr(cfg: &QbzdConfig) -> Result<std::net::SocketAddr, String> {
use std::net::ToSocketAddrs;
let hostport = format!("{}:{}", cfg.server.bind, cfg.server.port);
hostport
.to_socket_addrs()
.map_err(|e| {
format!("error: invalid [server] bind/port '{hostport}': {e}\n → set a valid ip and port in ~/.config/qbzd/qbzd.toml")
})?
.next()
.ok_or_else(|| {
format!("error: [server] '{hostport}' resolved to no address\n → set a valid ip and port in ~/.config/qbzd/qbzd.toml")
})
}
/// The step-5 bind-conflict diagnosis (02 §8.1-5 / §2.2): a qbzd occupant on a
/// different data root vs. an unrelated process on the port.
fn diagnose_port_conflict(addr: std::net::SocketAddr) -> String {
if crate::api::probe_is_qbzd(addr) {
crate::cli::copy::foreign_qbzd(&addr.to_string())
} else {
crate::cli::copy::port_in_use(addr.port())
}
}
/// Render an [`InstanceLock`] failure. For the already-running case this prints
/// the frozen exit-3 error voice (02 §1.3/§1.4) and exits 3 directly — the new
/// process must never clobber the running one. An I/O failure returns a String
+13 -4
View File
@@ -19,9 +19,6 @@ struct Cli {
/// Target daemon (default 127.0.0.1:8182; env QBZD_HOST)
#[arg(long, global = true)]
host: Option<String>,
/// API token (default: auto-read ~/.config/qbzd/api_token; env QBZD_TOKEN)
#[arg(long, global = true)]
token: Option<String>,
#[arg(short, long, global = true)]
quiet: bool,
#[command(subcommand)]
@@ -92,8 +89,10 @@ enum SettingsCmd {
#[derive(Subcommand)]
enum QconnectCmd { Enable, Disable, Name { name: String } }
// The tokenless default has no rotation verb (02 §3.1.2): `config` is just
// path|show. Rotating the opt-in [server] token = edit qbzd.toml + restart.
#[derive(Subcommand)]
enum ConfigCmd { Path, Show { #[arg(long)] json: bool }, RegenToken }
enum ConfigCmd { Path, Show { #[arg(long)] json: bool } }
#[tokio::main]
async fn main() {
@@ -179,6 +178,16 @@ async fn main() {
}
}
}
Cmd::Status { json } => {
// The CLI reads only local qbzd.toml (for the opt-in token); the
// config root is always at its XDG default.
let roots = paths::ProfileRoots::resolve(None, None);
cli::status::status(cli.host, json, &roots).await
}
Cmd::Ping { json } => {
let roots = paths::ProfileRoots::resolve(None, None);
cli::status::ping(cli.host, json, &roots).await
}
_ => { eprintln!("not implemented yet"); 1 } // burned down task by task
};
std::process::exit(code);