//! The HTTP API. //! //! There is **no WebSocket and no fan-out hub**. The web UI polls //! `GET /api/state` every 5 s while its tab is visible, and that endpoint returns //! an `ETag` so an unchanged poll is a 304 with no body. For a handful of users //! that costs less than a broadcast hub, per-connection filter tasks, and //! lag/resync handling would — and it cannot desynchronise, because there is no //! second copy of the state to drift. //! //! Visibility is resolved in exactly one place, [`visible_user_ids`], used by //! every read path. Share expiry is enforced there, at query time, never by a //! background job: a stalled job must not be able to leak a position. use std::net::{IpAddr, SocketAddr}; use std::sync::Arc; use axum::extract::{ConnectInfo, Path, Query, Request, State}; use axum::http::{HeaderMap, StatusCode, header}; use axum::middleware::{self, Next}; use axum::response::{IntoResponse, Response}; use axum::routing::{delete, get, post}; use axum::{Json, Router}; use serde::{Deserialize, Serialize}; use sqlx::SqlitePool; use tower_sessions::Session; use tracing::warn; use crate::auth::{self, AuthError, LoginThrottle}; use crate::config::Config; use crate::db::{Db, now}; use crate::ingest::Ingest; use crate::keys::KeyVault; use crate::writer::{WriteHandle, WriteOp}; /// Session key holding the authenticated user id. const SESSION_USER: &str = "uid"; /// Required on every state-changing request. /// /// The value is never read. A browser cannot set a custom header on a /// cross-origin request without a CORS preflight, and this server grants no CORS /// at all, so its mere presence proves the request came from our own page. /// Together with `SameSite=Lax` on the session cookie that is the entire CSRF /// defence: no token to mint, store, rotate, or leak into a log. const CSRF_HEADER: &str = "x-ot-csrf"; /// Session key holding the moment the session was created, compared against /// `users.pw_changed_at` so a password change invalidates older sessions without /// having to enumerate them. const SESSION_ISSUED: &str = "iat"; pub struct AppState { pub db: Db, pub cfg: Config, pub vault: KeyVault, pub ingest: Arc, pub writer: WriteHandle, /// Shared with the periodic GC task, which sweeps its expired windows. pub throttle: Arc, } pub type Shared = Arc; // --------------------------------------------------------------------------- // Errors // --------------------------------------------------------------------------- #[derive(Debug)] pub enum ApiError { Unauthorized, Forbidden, NotFound, BadRequest(String), TooManyRequests(u64), /// Anything unexpected. The detail is logged, never returned: an internal /// error message is a free source of schema and path information. Internal(anyhow::Error), } impl IntoResponse for ApiError { fn into_response(self) -> Response { let (status, message) = match self { Self::Unauthorized => (StatusCode::UNAUTHORIZED, "not signed in".to_string()), Self::Forbidden => (StatusCode::FORBIDDEN, "forbidden".to_string()), Self::NotFound => (StatusCode::NOT_FOUND, "not found".to_string()), Self::BadRequest(m) => (StatusCode::BAD_REQUEST, m), Self::TooManyRequests(retry) => ( StatusCode::TOO_MANY_REQUESTS, format!("too many attempts; try again in {retry}s"), ), Self::Internal(e) => { warn!(error = ?e, "internal error"); ( StatusCode::INTERNAL_SERVER_ERROR, "internal error".to_string(), ) } }; (status, Json(ErrorBody { error: message })).into_response() } } impl From for ApiError { fn from(e: anyhow::Error) -> Self { Self::Internal(e) } } impl From for ApiError { fn from(e: sqlx::Error) -> Self { Self::Internal(e.into()) } } #[derive(Serialize)] struct ErrorBody { error: String, } type ApiResult = Result; // --------------------------------------------------------------------------- // Session helpers // --------------------------------------------------------------------------- /// The authenticated user, or [`ApiError::Unauthorized`]. /// /// Also checks the session against `pw_changed_at`, which is how a password /// change logs out every other browser without keeping a revocation list. pub(crate) async fn current_user(state: &Shared, session: &Session) -> ApiResult { let uid: i64 = session .get(SESSION_USER) .await .map_err(|e| ApiError::Internal(anyhow::anyhow!("session store: {e}")))? .ok_or(ApiError::Unauthorized)?; let issued: i64 = session .get(SESSION_ISSUED) .await .map_err(|e| ApiError::Internal(anyhow::anyhow!("session store: {e}")))? .unwrap_or(0); let row: Option<(i64, Option)> = sqlx::query_as("SELECT pw_changed_at, disabled_at FROM users WHERE id = ?") .bind(uid) .fetch_optional(&state.db.read) .await?; let Some((pw_changed_at, disabled_at)) = row else { let _ = session.flush().await; return Err(ApiError::Unauthorized); }; if disabled_at.is_some() || issued < pw_changed_at { let _ = session.flush().await; return Err(ApiError::Unauthorized); } Ok(uid) } async fn require_admin(state: &Shared, session: &Session) -> ApiResult { let uid = current_user(state, session).await?; let is_admin: i64 = sqlx::query_scalar("SELECT is_admin FROM users WHERE id = ?") .bind(uid) .fetch_one(&state.db.read) .await?; if is_admin == 0 { return Err(ApiError::Forbidden); } Ok(uid) } /// The set of `user_id`s `viewer` may see: themselves, plus anyone reachable /// through a live share. /// /// One helper, used by every read path. Per-account positions make this /// noticeably simpler than a per-device model would: there is no /// `device_id IS NULL` "all my devices" special case in the join. async fn visible_user_ids(pool: &SqlitePool, viewer: i64) -> ApiResult> { let now = now(); let ids: Vec = sqlx::query_scalar( "SELECT ? AS user_id \ UNION \ SELECT s.owner_user_id FROM shares s \ WHERE s.revoked_at IS NULL \ AND (s.expires_at IS NULL OR s.expires_at > ?) \ AND s.viewer_user_id = ?", ) .bind(viewer) .bind(now) .bind(viewer) .fetch_all(pool) .await?; Ok(ids) } /// Snap a coordinate to a `precision_m` grid, so a share can show a /// neighbourhood instead of a doorstep. /// /// Deliberately a pure rounding, with no jitter: a stationary person whose /// fuzzed dot wandered on every poll would leak their true position to anyone /// who averaged the samples. fn snap_e7(lat_e7: i64, lon_e7: i64, precision_m: i64) -> (i64, i64) { if precision_m <= 0 { return (lat_e7, lon_e7); } /// Metres per degree of latitude, and of longitude at the equator. const M_PER_DEG: f64 = 111_320.0; let grid = |value: i64, step: f64| ((value as f64 / step).round() * step) as i64; let lat_step = precision_m as f64 / M_PER_DEG * 1e7; let lat = grid(lat_e7, lat_step).clamp(-900_000_000, 900_000_000); // The longitude step is derived from the *snapped* latitude, not the real // one, so every point in a cell gets the same grid. Deriving it from the // input would give two neighbours slightly different grids and leak that // they are not in fact at the same place. // // cos(lat) goes to zero at the poles, where a metre of easting is an // unbounded number of degrees. Clamping the divisor keeps the step finite; // it only makes the cell smaller than asked for, never larger. let cos_lat = (lat as f64 / 1e7).to_radians().cos().abs().max(0.01); let lon_step = precision_m as f64 / (M_PER_DEG * cos_lat) * 1e7; (lat, grid(lon_e7, lon_step)) } /// `precision_m` per owner for everyone `viewer` can see. /// /// Several live shares could target the same viewer, so the most generous one /// wins: a second share must never be able to make an existing one stricter. /// Owners with no row (the viewer themselves) are exact. async fn share_precision( pool: &SqlitePool, viewer: i64, ids_json: &str, ) -> ApiResult> { let rows: Vec<(i64, i64)> = sqlx::query_as( "SELECT s.owner_user_id, MIN(s.precision_m) FROM shares s \ JOIN json_each(?) v ON v.value = s.owner_user_id \ WHERE s.viewer_user_id = ? AND s.revoked_at IS NULL \ AND (s.expires_at IS NULL OR s.expires_at > ?) \ GROUP BY s.owner_user_id", ) .bind(ids_json) .bind(viewer) .bind(now()) .fetch_all(pool) .await?; Ok(rows.into_iter().collect()) } // --------------------------------------------------------------------------- // Payloads // --------------------------------------------------------------------------- #[derive(Deserialize)] pub struct LoginRequest { pub username: String, pub password: String, /// `"browser"` (default) or `"device"`. A device login additionally mints an /// OTP/1 token. #[serde(default)] pub purpose: Purpose, #[serde(default)] pub device_name: Option, #[serde(default)] pub platform: Option, } #[derive(Deserialize, Default, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum Purpose { #[default] Browser, Device, } #[derive(Serialize)] pub struct LoginResponse { pub user: Me, /// Present only for a device login. #[serde(skip_serializing_if = "Option::is_none")] pub device: Option, } /// Everything a phone needs, returned exactly once. #[derive(Serialize)] pub struct DeviceCredentials { pub token_id: u64, /// base64 of 32 bytes. The only time the server emits this in the clear. pub token_key: String, /// base64 of 32 bytes: the key that seals a `REVOKED` notice for this token. /// /// Separate from `token_key` because it must outlive it. `K_up` and `K_down` /// derive from the token key and die with the token's row; this one is /// derived from a server master and the `token_id`, so the server can still /// speak to a device whose row is gone. /// /// It has to be issued at login and cannot be retrofitted: a device that /// never received one can never verify a notice, and the thing that would /// prompt it to log in again is exactly that notice. pub revoke_key: String, pub udp_host: String, pub udp_port: u16, #[serde(skip_serializing_if = "Option::is_none")] pub tls_url: Option, pub config: DeviceConfig, } #[derive(Serialize)] pub struct DeviceConfig { pub config_version: u16, pub profile: &'static str, } #[derive(Serialize)] pub struct Me { pub id: i64, pub username: String, pub display_name: String, pub is_admin: bool, pub server_time: i64, } #[derive(Serialize)] pub struct PersonState { pub user_id: i64, pub display_name: String, /// True for the viewer's own entry. pub is_self: bool, #[serde(skip_serializing_if = "Option::is_none")] pub position: Option, } #[derive(Serialize)] pub struct Position { pub ts: i64, /// Degrees × 1e7. Integers end to end, so there is no float-formatting drift /// between the wire, the database, this JSON, and the map. pub lat_e7: i64, pub lon_e7: i64, pub acc_dm: Option, pub alt_m: Option, pub spd_cms: Option, pub brg_cdeg: Option, pub bat_pct: Option, pub flags: i64, pub recv_at: i64, } #[derive(Serialize)] pub struct StateResponse { pub server_time: i64, pub people: Vec, } #[derive(Serialize)] pub struct TokenInfo { /// A decimal string, not a number. `token_id` is a full 64-bit random value /// and JavaScript's `number` is exact only to 2^53, so a JSON number would /// silently round — and a token id that does not round-trip cannot be /// revoked. The phone's `DeviceCredentials` keeps the numeric form because /// its parser has real 64-bit integers. pub token_id: String, pub name: String, pub platform: String, pub app_version: Option, pub os_api_level: Option, pub last_seen_at: Option, pub last_src_ip: Option, pub last_transport: Option, pub created_at: i64, } #[derive(Deserialize)] pub struct TrackQuery { pub from: Option, pub to: Option, #[serde(default = "default_max")] pub max: usize, } fn default_max() -> usize { 2000 } #[derive(Serialize)] pub struct TrackResponse { pub user_id: i64, pub from: i64, pub to: i64, /// Google-style encoded polyline at 1e5 precision. pub polyline: String, pub point_count: usize, } #[derive(Deserialize)] pub struct PasswordChange { pub current_password: String, pub new_password: String, } #[derive(Deserialize)] pub struct CreateUser { pub username: String, pub password: String, #[serde(default)] pub display_name: Option, #[serde(default)] pub is_admin: bool, } #[derive(Serialize)] pub struct ShareInfo { pub id: i64, pub viewer_user_id: i64, pub viewer_username: String, pub viewer_display_name: String, pub trail_visible: bool, pub precision_m: i64, pub expires_at: Option, pub created_at: i64, } #[derive(Deserialize)] pub struct CreateShare { /// Resolved server-side. There is deliberately no endpoint that lists or /// searches users: a share is granted to someone you already know the name /// of, and anything else is a user directory for anyone with an account. pub username: String, #[serde(default = "yes")] pub trail_visible: bool, #[serde(default)] pub precision_m: i64, /// Seconds from now. `None` means the share does not expire. #[serde(default)] pub expires_in_s: Option, } fn yes() -> bool { true } // --------------------------------------------------------------------------- // Router // --------------------------------------------------------------------------- /// Rejects a state-changing request that did not come from our own page. async fn require_csrf(req: Request, next: Next) -> Response { if req.method().is_safe() || req.headers().contains_key(CSRF_HEADER) { return next.run(req).await; } ( StatusCode::FORBIDDEN, Json(ErrorBody { error: format!("missing {CSRF_HEADER} header"), }), ) .into_response() } pub fn router(state: Shared) -> Router { Router::new() .route("/api/login", post(login)) .route("/api/logout", post(logout)) .route("/api/me", get(me)) .route("/api/me/password", post(change_password)) .route("/api/state", get(state_handler)) .route("/api/tokens", get(list_tokens)) .route("/api/tokens/{token_id}", delete(revoke_one_token)) .route("/api/tokens/revoke-others", post(revoke_others)) .route("/api/shares", get(list_shares).post(create_share)) .route("/api/shares/{share_id}", delete(revoke_share)) .route("/api/users/{user_id}/track", get(track)) .route("/api/users", post(create_user)) // Only the /api routes above; `route_layer` runs nothing when no route // matches, so the static fallback below is untouched. .route_layer(middleware::from_fn(require_csrf)) .route("/healthz", get(healthz)) .route("/metrics", get(metrics)) // Outside the CSRF layer above deliberately: it is a GET, and Leaflet // loads tiles as plain elements that cannot carry a header. .merge(crate::tiles::router()) .with_state(state) // Anything else is the web UI, including deep links it routes itself. .fallback(crate::web::serve) } // --------------------------------------------------------------------------- // Handlers // --------------------------------------------------------------------------- async fn healthz(State(state): State) -> ApiResult> { // A real query, not a constant: "healthy" has to mean the database answers. let _: i64 = sqlx::query_scalar("SELECT 1") .fetch_one(&state.db.read) .await?; Ok(Json(serde_json::json!({ "ok": true, "version": env!("CARGO_PKG_VERSION"), "server_time": now(), "tokens_loaded": state.ingest.active_token_count(), }))) } /// Aggregate counters, in Prometheus text format. /// /// Bound to loopback callers only: these numbers say how much abuse the UDP port /// is absorbing and how close the writer is to saturation, which is exactly the /// reconnaissance an attacker would want. Scrape it through the reverse proxy or /// over an SSH tunnel. async fn metrics( State(state): State, ConnectInfo(peer): ConnectInfo, ) -> ApiResult { if !peer.ip().is_loopback() { return Err(ApiError::Forbidden); } use std::sync::atomic::Ordering::Relaxed; let c = &state.ingest.counters; let mut out = String::new(); for (name, value) in [ ("otp_datagrams_received", c.received.load(Relaxed)), ("otp_datagrams_malformed", c.malformed.load(Relaxed)), ("otp_unknown_token", c.unknown_token.load(Relaxed)), ("otp_auth_failed", c.auth_failed.load(Relaxed)), ("otp_rate_limited", c.rate_limited.load(Relaxed)), ("otp_throttled", c.throttled.load(Relaxed)), ("otp_points_accepted", c.points_accepted.load(Relaxed)), ("otp_points_rejected", c.points_rejected.load(Relaxed)), ("otp_acks_sent", c.acks_sent.load(Relaxed)), ("otp_nacks_sent", c.nacks_sent.load(Relaxed)), ("otp_revoked_notices", c.revoked_notices_sent.load(Relaxed)), // The reflection budget actually spent. If this is nonzero and climbing, // someone is probing the port with forged token ids. ( "otp_unverified_notices", c.unverified_notices_sent.load(Relaxed), ), ("otp_notices_suppressed", c.notices_suppressed.load(Relaxed)), ("otp_silent_drops", c.silent_drops.load(Relaxed)), ] { out.push_str(&format!("# TYPE {name} counter\n{name} {value}\n")); } for (name, value) in [ ( "otp_tokens_loaded", state.ingest.active_token_count() as u64, ), ( "otp_limiter_tracked_ips", state.ingest.limits().tracked_ips() as u64, ), ("otp_writer_capacity_free", state.writer.capacity() as u64), ] { out.push_str(&format!("# TYPE {name} gauge\n{name} {value}\n")); } Ok(out) } async fn login( State(state): State, session: Session, ConnectInfo(peer): ConnectInfo, Json(req): Json, ) -> ApiResult> { let ip = peer.ip(); let account = match auth::authenticate( &state.db.read, &state.writer, &state.cfg, &state.throttle, ip, &req.username, req.password, ) .await { Ok(a) => a, Err(AuthError::LockedOut { retry_after_s }) => { return Err(ApiError::TooManyRequests(retry_after_s)); } Err(AuthError::Invalid) => { audit(&state, None, "login_failed", &req.username, ip).await; return Err(ApiError::Unauthorized); } }; // Rotate the session id on login, so a fixation attempt cannot survive it. session .cycle_id() .await .map_err(|e| ApiError::Internal(anyhow::anyhow!("session: {e}")))?; let issued = now(); session .insert(SESSION_USER, account.id) .await .map_err(|e| ApiError::Internal(anyhow::anyhow!("session: {e}")))?; session .insert(SESSION_ISSUED, issued) .await .map_err(|e| ApiError::Internal(anyhow::anyhow!("session: {e}")))?; let device = if req.purpose == Purpose::Device { let name = req.device_name.unwrap_or_else(|| "phone".to_string()); let platform = req.platform.unwrap_or_else(|| "android".to_string()); let minted = auth::mint_token( &state.db.write, &state.vault, &state.ingest, account.id, &name, &platform, ip, ) .await?; audit(&state, Some(account.id), "token_minted", &name, ip).await; use base64::Engine as _; Some(DeviceCredentials { token_id: minted.token_id, token_key: base64::engine::general_purpose::STANDARD.encode(minted.token_key), revoke_key: base64::engine::general_purpose::STANDARD .encode(state.vault.revocation_key(minted.token_id)), udp_host: state.cfg.public_udp_host.clone(), udp_port: state.cfg.public_udp_port, tls_url: state.cfg.public_tls_url.clone(), config: DeviceConfig { config_version: minted.config_version, profile: "balanced", }, }) } else { audit(&state, Some(account.id), "login", "browser", ip).await; None }; Ok(Json(LoginResponse { user: Me { id: account.id, username: account.username, display_name: account.display_name, is_admin: account.is_admin, server_time: issued, }, device, })) } async fn logout(session: Session) -> ApiResult { session .flush() .await .map_err(|e| ApiError::Internal(anyhow::anyhow!("session: {e}")))?; Ok(StatusCode::NO_CONTENT) } async fn me(State(state): State, session: Session) -> ApiResult> { let uid = current_user(&state, &session).await?; let (username, display_name, is_admin): (String, String, i64) = sqlx::query_as("SELECT username, display_name, is_admin FROM users WHERE id = ?") .bind(uid) .fetch_one(&state.db.read) .await?; Ok(Json(Me { id: uid, username, display_name, is_admin: is_admin != 0, server_time: now(), })) } async fn change_password( State(state): State, session: Session, ConnectInfo(peer): ConnectInfo, Json(req): Json, ) -> ApiResult { let uid = current_user(&state, &session).await?; if req.new_password.chars().count() < 10 { return Err(ApiError::BadRequest( "the new password must be at least 10 characters".into(), )); } let stored: String = sqlx::query_scalar("SELECT pw_hash FROM users WHERE id = ?") .bind(uid) .fetch_one(&state.db.read) .await?; if !auth::verify(&state.cfg, stored, req.current_password) .await? .ok { return Err(ApiError::Unauthorized); } let hash = auth::hash_password(&state.cfg, req.new_password).await?; let at = now(); sqlx::query("UPDATE users SET pw_hash = ?, pw_changed_at = ? WHERE id = ?") .bind(hash) .bind(at) .bind(uid) .execute(&state.db.write) .await?; // A password change logs out every phone and every other browser. Browsers // are handled by the pw_changed_at comparison in current_user; phones need // their tokens actually revoked, since they carry a key rather than a cookie. let revoked = auth::revoke_other_tokens(&state.db.write, &state.ingest, uid, None).await?; session .cycle_id() .await .map_err(|e| ApiError::Internal(anyhow::anyhow!("session: {e}")))?; session .insert(SESSION_ISSUED, at) .await .map_err(|e| ApiError::Internal(anyhow::anyhow!("session: {e}")))?; audit( &state, Some(uid), "password_changed", &format!("{revoked} tokens revoked"), peer.ip(), ) .await; Ok(StatusCode::NO_CONTENT) } /// Everything the live view needs, in one call. async fn state_handler( State(state): State, session: Session, headers: HeaderMap, ) -> ApiResult { let uid = current_user(&state, &session).await?; let visible = visible_user_ids(&state.db.read, uid).await?; // One query for everyone visible. A LEFT JOIN so a person with no position // yet still appears in the list — otherwise they would silently vanish from // the UI until their first fix, which reads as a bug. // // The id set is passed as a JSON array through `json_each` rather than by // building an `IN (?, ?, ?)` string: the SQL stays a literal, so there is no // interpolation to audit and the statement cache gets one entry instead of // one per group size. let ids_json = serde_json::to_string(&visible).map_err(|e| ApiError::Internal(e.into()))?; let rows = sqlx::query_as::<_, LatestRow>( "SELECT u.id, u.display_name, l.ts, l.lat, l.lon, l.acc_dm, l.alt_m, l.spd_cms, \ l.brg_cdeg, l.bat_pct, l.flags, l.recv_at \ FROM users u \ JOIN json_each(?) v ON v.value = u.id \ LEFT JOIN user_latest l ON l.user_id = u.id \ ORDER BY u.display_name", ) .bind(&ids_json) .fetch_all(&state.db.read) .await?; // Your own position is always exact; everyone else's is snapped to whatever // their share allows. let precision = share_precision(&state.db.read, uid, &ids_json).await?; let people: Vec = rows .into_iter() .map(|r| PersonState { user_id: r.id, is_self: r.id == uid, display_name: r.display_name, position: r.ts.map(|ts| { let (lat_e7, lon_e7) = snap_e7( r.lat.unwrap_or(0), r.lon.unwrap_or(0), precision.get(&r.id).copied().unwrap_or(0), ); Position { ts, lat_e7, lon_e7, acc_dm: r.acc_dm, alt_m: r.alt_m, spd_cms: r.spd_cms, brg_cdeg: r.brg_cdeg, bat_pct: r.bat_pct, flags: r.flags.unwrap_or(0), recv_at: r.recv_at.unwrap_or(ts), } }), }) .collect(); let body = StateResponse { server_time: now(), people, }; // ETag over the people list only — deliberately *not* including // `server_time`, which changes every second and would make every poll a 200. let etag = etag_of(&body.people); if headers .get(header::IF_NONE_MATCH) .and_then(|v| v.to_str().ok()) .is_some_and(|v| v == etag) { return Ok((StatusCode::NOT_MODIFIED, [(header::ETAG, etag)]).into_response()); } Ok(( [ (header::ETAG, etag), (header::CACHE_CONTROL, "no-store".to_string()), ], Json(body), ) .into_response()) } #[derive(sqlx::FromRow)] struct LatestRow { id: i64, display_name: String, ts: Option, lat: Option, lon: Option, acc_dm: Option, alt_m: Option, spd_cms: Option, brg_cdeg: Option, bat_pct: Option, flags: Option, recv_at: Option, } /// A weak ETag over the payload's meaningful content. fn etag_of(people: &[PersonState]) -> String { use std::hash::{Hash, Hasher}; let mut h = std::collections::hash_map::DefaultHasher::new(); for p in people { p.user_id.hash(&mut h); p.display_name.hash(&mut h); if let Some(pos) = &p.position { pos.ts.hash(&mut h); pos.lat_e7.hash(&mut h); pos.lon_e7.hash(&mut h); pos.acc_dm.hash(&mut h); pos.bat_pct.hash(&mut h); pos.flags.hash(&mut h); } else { 0u8.hash(&mut h); } } format!("W/\"{:x}\"", h.finish()) } async fn list_tokens( State(state): State, session: Session, ) -> ApiResult>> { let uid = current_user(&state, &session).await?; let rows: Vec = sqlx::query_as( "SELECT token_id, name, platform, app_version, os_api_level, last_seen_at, last_src_ip, \ last_transport, created_at \ FROM tokens WHERE user_id = ? AND revoked_at IS NULL ORDER BY created_at DESC", ) .bind(uid) .fetch_all(&state.db.read) .await?; Ok(Json( rows.into_iter() .map(|r| TokenInfo { token_id: (r.token_id as u64).to_string(), name: r.name, platform: r.platform, app_version: r.app_version, os_api_level: r.os_api_level, last_seen_at: r.last_seen_at, last_src_ip: r.last_src_ip, last_transport: r.last_transport, created_at: r.created_at, }) .collect(), )) } #[derive(sqlx::FromRow)] struct TokenRow { token_id: i64, name: String, platform: String, app_version: Option, os_api_level: Option, last_seen_at: Option, last_src_ip: Option, last_transport: Option, created_at: i64, } async fn revoke_one_token( State(state): State, session: Session, ConnectInfo(peer): ConnectInfo, Path(token_id): Path, ) -> ApiResult { let uid = current_user(&state, &session).await?; // Parsed as the u64 it is on the wire, then bit-cast: SQLite has no unsigned // integer type, so that cast is how every token id is stored. let token_id = token_id .parse::() .map_err(|_| ApiError::BadRequest("token id must be a u64".into()))? as i64; // Scope the ownership check into the query: fetching then comparing invites // the check being forgotten on some future path. let owner: Option = sqlx::query_scalar("SELECT user_id FROM tokens WHERE token_id = ?") .bind(token_id) .fetch_optional(&state.db.read) .await?; match owner { None => return Err(ApiError::NotFound), Some(o) if o != uid => return Err(ApiError::Forbidden), Some(_) => {} } if !auth::revoke_token(&state.db.write, &state.ingest, token_id).await? { return Err(ApiError::NotFound); } audit( &state, Some(uid), "token_revoked", &token_id.to_string(), peer.ip(), ) .await; Ok(StatusCode::NO_CONTENT) } async fn revoke_others( State(state): State, session: Session, ConnectInfo(peer): ConnectInfo, ) -> ApiResult> { let uid = current_user(&state, &session).await?; // From a browser there is no "current token" to keep, so this revokes every // phone. The browser's own session is unaffected. let revoked = auth::revoke_other_tokens(&state.db.write, &state.ingest, uid, None).await?; audit( &state, Some(uid), "tokens_revoked_all", &revoked.to_string(), peer.ip(), ) .await; Ok(Json(serde_json::json!({ "revoked": revoked }))) } async fn track( State(state): State, session: Session, Path(user_id): Path, Query(q): Query, ) -> ApiResult> { let viewer = current_user(&state, &session).await?; let visible = visible_user_ids(&state.db.read, viewer).await?; if !visible.contains(&user_id) { // 403 rather than 404: the caller already knows this user exists if they // saw them in /api/state, and pretending otherwise buys nothing. return Err(ApiError::Forbidden); } // Trails are only visible when the share says so. Your own trail is always // yours to see, and always exact. Where several live shares exist the most // generous one wins, so a second share cannot tighten an existing one. let mut precision_m = 0; if user_id != viewer { let (trail_visible, precision): (Option, Option) = sqlx::query_as( "SELECT MAX(s.trail_visible), MIN(s.precision_m) FROM shares s \ WHERE s.owner_user_id = ? AND s.revoked_at IS NULL \ AND (s.expires_at IS NULL OR s.expires_at > ?) \ AND s.viewer_user_id = ?", ) .bind(user_id) .bind(now()) .bind(viewer) .fetch_one(&state.db.read) .await?; if trail_visible.unwrap_or(0) == 0 { return Err(ApiError::Forbidden); } precision_m = precision.unwrap_or(0); } let to = q.to.unwrap_or_else(now); let from = q.from.unwrap_or(to - 24 * 3_600); if from >= to { return Err(ApiError::BadRequest("from must be before to".into())); } let max = q.max.clamp(2, 10_000); let mut rows: Vec<(i64, i64)> = sqlx::query_as( "SELECT lat, lon FROM points WHERE user_id = ? AND ts >= ? AND ts <= ? ORDER BY ts", ) .bind(user_id) .bind(from) .bind(to) .fetch_all(&state.db.read) .await?; // Snap before simplifying: simplifying first would let the exact geometry // decide which points survive, and the shape of a route is itself a hint. for p in &mut rows { (p.0, p.1) = snap_e7(p.0, p.1, precision_m); } // Decimate server-side. 2000 points as an encoded polyline is ~10 kB against // ~60 kB of JSON floats, and the browser has less to draw. let simplified = crate::polyline::simplify(&rows, max); let polyline = crate::polyline::encode(&simplified); Ok(Json(TrackResponse { user_id, from, to, point_count: simplified.len(), polyline, })) } async fn create_user( State(state): State, session: Session, ConnectInfo(peer): ConnectInfo, Json(req): Json, ) -> ApiResult> { let admin = require_admin(&state, &session).await?; let username = req.username.trim().to_string(); if username.is_empty() || username.chars().count() > 64 { return Err(ApiError::BadRequest( "username must be 1..=64 characters".into(), )); } if req.password.chars().count() < 10 { return Err(ApiError::BadRequest( "password must be at least 10 characters".into(), )); } let hash = auth::hash_password(&state.cfg, req.password).await?; let at = now(); let result = sqlx::query( "INSERT INTO users (username, pw_hash, display_name, is_admin, created_at, pw_changed_at) \ VALUES (?, ?, ?, ?, ?, ?)", ) .bind(&username) .bind(hash) .bind(req.display_name.unwrap_or_else(|| username.clone())) .bind(i64::from(req.is_admin)) .bind(at) .bind(at) .execute(&state.db.write) .await; let id = match result { Ok(r) => r.last_insert_rowid(), Err(sqlx::Error::Database(e)) if e.is_unique_violation() => { return Err(ApiError::BadRequest("that username is taken".into())); } Err(e) => return Err(e.into()), }; audit(&state, Some(admin), "user_created", &username, peer.ip()).await; Ok(Json(serde_json::json!({ "id": id, "username": username }))) } /// The signed-in user's *outgoing* shares: who can currently see them. async fn list_shares( State(state): State, session: Session, ) -> ApiResult>> { let uid = current_user(&state, &session).await?; let rows: Vec = sqlx::query_as( "SELECT s.id, s.viewer_user_id, u.username, u.display_name, s.trail_visible, \ s.precision_m, s.expires_at, s.created_at \ FROM shares s JOIN users u ON u.id = s.viewer_user_id \ WHERE s.owner_user_id = ? AND s.revoked_at IS NULL \ AND (s.expires_at IS NULL OR s.expires_at > ?) \ ORDER BY s.created_at DESC", ) .bind(uid) .bind(now()) .fetch_all(&state.db.read) .await?; Ok(Json(rows.into_iter().map(ShareRow::into_info).collect())) } #[derive(sqlx::FromRow)] struct ShareRow { id: i64, viewer_user_id: i64, username: String, display_name: String, trail_visible: i64, precision_m: i64, expires_at: Option, created_at: i64, } impl ShareRow { fn into_info(self) -> ShareInfo { ShareInfo { id: self.id, viewer_user_id: self.viewer_user_id, viewer_username: self.username, viewer_display_name: self.display_name, trail_visible: self.trail_visible != 0, precision_m: self.precision_m, expires_at: self.expires_at, created_at: self.created_at, } } } async fn create_share( State(state): State, session: Session, ConnectInfo(peer): ConnectInfo, Json(req): Json, ) -> ApiResult<(StatusCode, Json)> { let uid = current_user(&state, &session).await?; if !(0..=100_000).contains(&req.precision_m) { return Err(ApiError::BadRequest( "precision_m must be 0..=100000 metres".into(), )); } if req.expires_in_s.is_some_and(|s| s <= 0) { return Err(ApiError::BadRequest("expires_in_s must be positive".into())); } // `username` is COLLATE NOCASE, so this match is case-insensitive for free. let viewer: Option<(i64, String, String)> = sqlx::query_as( "SELECT id, username, display_name FROM users \ WHERE username = ? AND disabled_at IS NULL", ) .bind(req.username.trim()) .fetch_optional(&state.db.read) .await?; let Some((viewer_id, username, display_name)) = viewer else { return Err(ApiError::NotFound); }; if viewer_id == uid { return Err(ApiError::BadRequest("you can already see yourself".into())); } let at = now(); let expires_at = req.expires_in_s.map(|s| at + s); // One transaction, because the two statements below are one act. A revoke // that committed without its replacement would silently drop a share the // user was in the middle of editing. let mut tx = state.db.write.begin().await?; // At most one live row per (owner, viewer): re-sharing with new settings // replaces the old share rather than adding a more permissive one beside it. sqlx::query( "UPDATE shares SET revoked_at = ? \ WHERE owner_user_id = ? AND viewer_user_id = ? AND revoked_at IS NULL", ) .bind(at) .bind(uid) .bind(viewer_id) .execute(&mut *tx) .await?; let id = sqlx::query( "INSERT INTO shares (owner_user_id, viewer_user_id, trail_visible, precision_m, \ expires_at, created_at) \ VALUES (?, ?, ?, ?, ?, ?)", ) .bind(uid) .bind(viewer_id) .bind(i64::from(req.trail_visible)) .bind(req.precision_m) .bind(expires_at) .bind(at) .execute(&mut *tx) .await? .last_insert_rowid(); tx.commit().await?; audit(&state, Some(uid), "share_created", &username, peer.ip()).await; Ok(( StatusCode::CREATED, Json(ShareInfo { id, viewer_user_id: viewer_id, viewer_username: username, viewer_display_name: display_name, trail_visible: req.trail_visible, precision_m: req.precision_m, expires_at, created_at: at, }), )) } async fn revoke_share( State(state): State, session: Session, ConnectInfo(peer): ConnectInfo, Path(share_id): Path, ) -> ApiResult { let uid = current_user(&state, &session).await?; // Ownership is part of the UPDATE, not a fetch-then-compare: the check // cannot then be forgotten on some future path. let affected = sqlx::query( "UPDATE shares SET revoked_at = ? \ WHERE id = ? AND owner_user_id = ? AND revoked_at IS NULL", ) .bind(now()) .bind(share_id) .bind(uid) .execute(&state.db.write) .await? .rows_affected(); if affected == 0 { return Err(ApiError::NotFound); } audit( &state, Some(uid), "share_revoked", &share_id.to_string(), peer.ip(), ) .await; Ok(StatusCode::NO_CONTENT) } async fn audit(state: &Shared, user_id: Option, action: &str, detail: &str, ip: IpAddr) { let _ = state .writer .send(WriteOp::Audit { user_id, at: now(), action: action.to_string(), detail: detail.to_string(), src_ip: Some(ip.to_string()), }) .await; } #[cfg(test)] mod tests { use super::*; /// The web UI's `web/src/api.ts` types are hand-written. This is what stops /// them drifting: renaming a field here fails `cargo test` instead of /// producing `undefined` in a browser at runtime. /// /// Keys only, not values — the types carry no invariants worth asserting, /// and a value check would just restate the constructor above it. fn keys(value: &serde_json::Value) -> Vec<&str> { let mut k: Vec<&str> = value .as_object() .expect("expected a JSON object") .keys() .map(String::as_str) .collect(); k.sort_unstable(); k } #[test] fn the_json_shape_is_the_one_the_web_ui_expects() { let me = Me { id: 1, username: "a".into(), display_name: "A".into(), is_admin: false, server_time: 0, }; assert_eq!( keys(&serde_json::to_value(&me).expect("serialize")), ["display_name", "id", "is_admin", "server_time", "username"] ); let position = Position { ts: 0, lat_e7: 0, lon_e7: 0, acc_dm: None, alt_m: None, spd_cms: None, brg_cdeg: None, bat_pct: None, flags: 0, recv_at: 0, }; assert_eq!( keys(&serde_json::to_value(&position).expect("serialize")), [ "acc_dm", "alt_m", "bat_pct", "brg_cdeg", "flags", "lat_e7", "lon_e7", "recv_at", "spd_cms", "ts" ] ); let person = PersonState { user_id: 1, display_name: "A".into(), is_self: true, position: Some(position), }; assert_eq!( keys(&serde_json::to_value(&person).expect("serialize")), ["display_name", "is_self", "position", "user_id"] ); // `position` is skipped when absent, which is why the TypeScript field is // optional rather than nullable. let no_fix = PersonState { position: None, ..person }; assert_eq!( keys(&serde_json::to_value(&no_fix).expect("serialize")), ["display_name", "is_self", "user_id"] ); let state = StateResponse { server_time: 0, people: vec![], }; assert_eq!( keys(&serde_json::to_value(&state).expect("serialize")), ["people", "server_time"] ); let token = TokenInfo { token_id: "1".into(), name: "p".into(), platform: "android".into(), app_version: None, os_api_level: None, last_seen_at: None, last_src_ip: None, last_transport: None, created_at: 0, }; assert_eq!( keys(&serde_json::to_value(&token).expect("serialize")), [ "app_version", "created_at", "last_seen_at", "last_src_ip", "last_transport", "name", "os_api_level", "platform", "token_id" ] ); let track = TrackResponse { user_id: 1, from: 0, to: 1, polyline: String::new(), point_count: 0, }; assert_eq!( keys(&serde_json::to_value(&track).expect("serialize")), ["from", "point_count", "polyline", "to", "user_id"] ); let share = ShareInfo { id: 1, viewer_user_id: 2, viewer_username: "b".into(), viewer_display_name: "B".into(), trail_visible: true, precision_m: 0, expires_at: None, created_at: 0, }; assert_eq!( keys(&serde_json::to_value(&share).expect("serialize")), [ "created_at", "expires_at", "id", "precision_m", "trail_visible", "viewer_display_name", "viewer_user_id", "viewer_username" ] ); assert_eq!( keys(&serde_json::to_value(ErrorBody { error: "x".into() }).expect("serialize")), ["error"] ); } /// Metres between two coordinates, good enough to check a grid cell size. fn metres_between(a: (i64, i64), b: (i64, i64)) -> f64 { let dlat = (a.0 - b.0) as f64 / 1e7 * 111_320.0; let cos = (a.0 as f64 / 1e7).to_radians().cos(); let dlon = (a.1 - b.1) as f64 / 1e7 * 111_320.0 * cos; (dlat * dlat + dlon * dlon).sqrt() } #[test] fn a_precision_of_zero_or_less_leaves_the_position_untouched() { let exact = (521_234_567, 133_456_789); for precision in [0, -1, -100_000] { assert_eq!(snap_e7(exact.0, exact.1, precision), exact); } } /// The property that makes rounding safe where jitter would not be: an /// attacker polling a stationary person gets the same answer every time, so /// averaging the samples reveals nothing. #[test] fn snapping_is_deterministic() { let first = snap_e7(521_234_567, 133_456_789, 500); for _ in 0..10 { assert_eq!(snap_e7(521_234_567, 133_456_789, 500), first); } } #[test] fn a_snapped_point_stays_within_roughly_the_requested_precision() { for lat in [0, 100_000_000, 521_234_567, -335_000_000] { for lon in [0, 133_456_789, -740_000_000] { for precision in [10, 100, 1_000, 100_000] { let snapped = snap_e7(lat, lon, precision); let moved = metres_between((lat, lon), snapped); assert!( moved <= precision as f64, "moved {moved} m for a {precision} m grid" ); } } } } #[test] fn snapping_collapses_nearby_points_onto_one_cell() { // Two points ~11 m apart, on a 1 km grid. let a = snap_e7(521_234_567, 133_456_789, 1_000); let b = snap_e7(521_235_567, 133_456_789, 1_000); assert_eq!(a, b); // The same two points are still distinct when snapped finely. assert_ne!( snap_e7(521_234_567, 133_456_789, 1), snap_e7(521_235_567, 133_456_789, 1) ); } /// cos(lat) reaches zero at the poles, where the longitude divisor would be /// zero and every result a NaN cast to a garbage integer. #[test] fn a_point_at_the_pole_does_not_divide_by_zero() { for lat in [899_999_999, 900_000_000, -900_000_000] { let (lat_e7, lon_e7) = snap_e7(lat, 123_456_789, 1_000); assert!((-900_000_000..=900_000_000).contains(&lat_e7), "{lat_e7}"); assert!( (-1_800_000_000..=1_800_000_000).contains(&lon_e7), "{lon_e7}" ); } } /// Token ids routinely exceed 2^53, which is why [`TokenInfo::token_id`] is a /// string. This asserts the reason still holds rather than trusting the /// comment: if ids ever became small the string could go away. #[test] fn token_ids_are_too_large_for_a_javascript_number() { let big = (0..64) .map(|_| crate::keys::random_token_id().expect("rng")) .filter(|id| *id >= (1u64 << 53)) .count(); assert!( big > 32, "expected most token ids above 2^53, got {big} of 64" ); } /// The exact value a browser must be able to send back and have match. #[test] fn a_large_token_id_round_trips_through_its_string_form() { let id = u64::MAX - 3; let text = id.to_string(); assert_eq!(text.parse::().expect("parse") as i64, id as i64); } }