From 2c54778116cf13346d015a00022cccfd33ddaf2d Mon Sep 17 00:00:00 2001 From: xavix-yo Date: Sat, 11 Jul 2026 11:18:35 +0100 Subject: [PATCH] feat(mobile): per-user device bindings, multi-user Inbox routing, and admin-mediated authorization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Device→user bindings persisted in config table (auth.rs), loaded at plugin start - RelayApp now routes Inbox responses per-user via UserChannelApi, never globally - New mobile_bind_device LLM tool for admin-mediated device→user assignment - Per-user event forwarders (events.rs) with per-user debounced notifiers - Config listener (auth::config_listener) refreshes bindings cache reactively - Reconcile loop catches users who unlock after boot - Hello/Logout treated as device-registry ops (no user resolution needed) - Unbound device payloads are silently dropped - RelayAgent::authorize_client → bind_device (atomic bind + authorize) - Approval rules seed mobile_bind_device/revoke_device as require --- CLAUDE.md | 4 +- Cargo.lock | 67 ++++ Cargo.toml | 2 + crates/core-api/src/user_channel.rs | 7 + crates/plugin-mobile-connector/src/agent.rs | 22 +- crates/plugin-mobile-connector/src/app.rs | 315 +++++++++++++----- crates/plugin-mobile-connector/src/auth.rs | 207 ++++++++++++ crates/plugin-mobile-connector/src/events.rs | 146 ++++++++ crates/plugin-mobile-connector/src/lib.rs | 183 +++++----- .../plugin-mobile-connector/src/notifier.rs | 41 ++- .../plugin-mobile-connector/src/payloads.rs | 5 +- crates/plugin-mobile-connector/src/tools.rs | 66 +++- crates/skald-core/src/approval/mod.rs | 4 + crates/skald-core/src/skald/user_context.rs | 7 + src/main.rs | 1 + 15 files changed, 856 insertions(+), 221 deletions(-) create mode 100644 crates/plugin-mobile-connector/src/auth.rs create mode 100644 crates/plugin-mobile-connector/src/events.rs diff --git a/CLAUDE.md b/CLAUDE.md index 1901d87..3c8b2ca 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -35,7 +35,7 @@ Domain words are allowed only in seed data, preset labels, UI copy and positioni ### Current state -`UserManager` (§11) exists and works — `crates/skald-core/src/users/mod.rs`, with real per-user SQLCipher encryption (§4). It is **not consumed yet**: there is no login, and `Runtime` still hands every call site the one shared `Arc` on `system.db`, so chats still land in that file's owner tables. The next step is migrating those call sites to `pool_of`, and only then deciding where the owner-without-a-user lives (see blueprint §19). +`UserManager` (§11) is now **consumed**. Login exists (`crates/skald-core/src/auth/`: `SessionStore` + the `guard.rs` deny-by-default middleware; first admin created by `skald-setup`), and the per-user owner-bound runtime is `UserContext` (`crates/skald-core/src/skald/user_context.rs`) — resolved by `Skald::user_context` / the frontend's `require_context`, keyed off `UserManager::pool_of`. The frontend owner call-sites (WS, sessions, inbox, approval-pending, projects, uploads, run-context, **cron**) route through the per-user pool; dev/stats read `llm_requests` — a *registry* table — from `system.db`, which is correct. The "owner-without-a-user" question resolved to **there isn't one**: every owner content belongs to a logged-in user (the admin included). The global owner-bound bundles (`Conversation`/`Tasks`: the "ownerless" `ChatSessionManager`, `ChatHub`, cron `TaskManager`, `TicManager`) are still constructed but **inert** — their loops never spawn and nothing consumes their accessors; removing them is pending follow-on work (kept for now because `RunContextManager` shares the `Conversation` bundle and *is* used, being registry-backed). See blueprint §19. Direction of travel, decided but not yet executed: strip the **power-user surface** (self-rewriting, arbitrary shell, dev-agent suite, ticket system) and move to a **binary-first** layout — the app is built once and run from a compiled binary, not executed from its own source tree. @@ -108,7 +108,7 @@ The schema is split into two buckets (§5.1), and the split is the point: **Memory injection into the prompt**: `MessageBuilder::load_inject_memory` routes each `meta.inject_memory` entry — `user-memory/…` → owner pool, `shared-memory/…` → the shared (`system.db`) pool, both via `memory_docs::get`; anything else (`data/…`, `$WD/…`) is a disk read. The shared pool is threaded `ChatSessionManager` → handler → `MessageBuilder`. `main` and `project-coordinator` inject `user-memory/index.md` + `shared-memory/index.md`. -`system.db` currently gets **both** bucket functions, because nothing has migrated to per-user pools yet. That is transitional. +`system.db` still gets **both** bucket functions — but no longer because the migration is unstarted. It gets the owner schema because it *is* the owner of **shared** memory (`memory_docs`) plus, for now, the globally-scoped `secrets` and `mcp_servers`/`mcp_events` (`SecretsStore` and `McpManager` are built on the system pool and shared by reference into every `UserContext`). Every *other* owner table is created there but never written to anymore — the global owner-bound managers that would write them (chat/jobs/etc.) are inert (see "Current state"). Fully dropping `create_owner_tables` from `system.db` is blocked on the §4/§7/§14 scope decisions for secrets and MCP, not on call-site migration. `users` (`crates/skald-core/src/db/users.rs`) holds the directory plus auth material. It lives in the system DB, which the box owner can read, so it must never store anything that derives a user's key. `Credentials` is an enum mirroring the table's `CHECK`: an encrypted user carries a **wrapped DEK** (whose AEAD tag *is* the password verifier — hence no `password_hash`); a cleartext user carries an ordinary verifier, or none. `User` is deliberately not `Serialize` and its `Debug` redacts key material — use `User::summary()` for anything leaving the process. `role_id` has no foreign key yet: sqlx enables `PRAGMA foreign_keys`, so referencing the not-yet-existing `roles` table would fail every insert. diff --git a/Cargo.lock b/Cargo.lock index 7d8cfb6..f4dad63 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -432,6 +432,12 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" +[[package]] +name = "byteorder-lite" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f1fe948ff07f4bd06c30984e69f5b4899c516a3ef74f34df92a2df2ab535495" + [[package]] name = "bytes" version = "1.11.1" @@ -2563,6 +2569,19 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "image" +version = "0.25.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85ab80394333c02fe689eaf900ab500fbd0c2213da414687ebf995a65d5a6104" +dependencies = [ + "bytemuck", + "byteorder-lite", + "moxcms", + "num-traits", + "png 0.18.1", +] + [[package]] name = "include_dir" version = "0.7.4" @@ -3162,6 +3181,16 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "moxcms" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb85c154ba489f01b25c0d36ae69a87e4a1c73a72631fc6c0eb6dde34a73e44b" +dependencies = [ + "num-traits", + "pxfm", +] + [[package]] name = "muda" version = "0.19.3" @@ -3986,6 +4015,28 @@ dependencies = [ "tracing", ] +[[package]] +name = "plugin-mobile-connector" +version = "0.1.0" +dependencies = [ + "anyhow", + "async-trait", + "axum", + "chrono", + "core-api", + "hex", + "image", + "qrcode", + "rand 0.9.4", + "serde", + "serde_json", + "skald-relay-client", + "skald-relay-common", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "plugin-tailscale-remote" version = "0.1.0" @@ -4392,6 +4443,21 @@ dependencies = [ "cc", ] +[[package]] +name = "pxfm" +version = "0.1.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d55d956fa96f5ec02be2e13af0e20391a5aa83d6a074e3ad368959d0fab299ea" + +[[package]] +name = "qrcode" +version = "0.14.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d68782463e408eb1e668cf6152704bd856c78c5b6417adaee3203d8f4c1fc9ec" +dependencies = [ + "image", +] + [[package]] name = "quick-xml" version = "0.41.0" @@ -5407,6 +5473,7 @@ dependencies = [ "notify", "plugin-comfyui", "plugin-elevenlabs", + "plugin-mobile-connector", "plugin-tailscale-remote", "plugin-telegram-bot", "plugin-transcribe-whisper-local", diff --git a/Cargo.toml b/Cargo.toml index 4de31c1..67c3139 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,7 @@ members = [ "crates/mcp-client", "crates/plugin-tailscale-remote", "crates/plugin-telegram-bot", + "crates/plugin-mobile-connector", "crates/plugin-transcribe-whisper-local", "crates/plugin-comfyui", "crates/plugin-tts-orpheus-3b", @@ -83,6 +84,7 @@ core-api = { path = "crates/core-api" } mcp-client = { path = "crates/mcp-client" } plugin-tailscale-remote = { path = "crates/plugin-tailscale-remote" } plugin-telegram-bot = { path = "crates/plugin-telegram-bot" } +plugin-mobile-connector = { path = "crates/plugin-mobile-connector" } plugin-transcribe-whisper-local = { path = "crates/plugin-transcribe-whisper-local", optional = true } plugin-comfyui = { path = "crates/plugin-comfyui" } plugin-tts-orpheus-3b = { path = "crates/plugin-tts-orpheus-3b" } diff --git a/crates/core-api/src/user_channel.rs b/crates/core-api/src/user_channel.rs index 5fbce09..7cb03a9 100644 --- a/crates/core-api/src/user_channel.rs +++ b/crates/core-api/src/user_channel.rs @@ -22,6 +22,7 @@ use tokio::sync::broadcast; use crate::approval::ApprovalApi; use crate::chat_hub::ChatHubApi; use crate::events::GlobalEvent; +use crate::inbox::InboxApi; /// Resolves an unlocked user's channel handle. /// @@ -51,6 +52,12 @@ pub trait UserChannelHandle: Send + Sync { /// The user's approval manager — resolve pending tool-call approvals. fn approval(&self) -> Arc; + /// The user's Inbox — the unified view over pending approvals, + /// clarifications and MCP elicitations. Channel adapters that bridge the + /// whole Inbox (e.g. the mobile connector) use this instead of wiring + /// `approval()`/clarification/elicitation separately. + fn inbox(&self) -> Arc; + /// Subscribe to the user's server→client event stream. /// Events are scoped to this user; no cross-user leakage. fn subscribe(&self) -> broadcast::Receiver; diff --git a/crates/plugin-mobile-connector/src/agent.rs b/crates/plugin-mobile-connector/src/agent.rs index bc30169..ade46d0 100644 --- a/crates/plugin-mobile-connector/src/agent.rs +++ b/crates/plugin-mobile-connector/src/agent.rs @@ -31,6 +31,8 @@ pub struct ClientInfo { pub platform: Option, /// Unix ms of last activity, if any. pub last_seen: Option, + /// The Skald user this device is bound to, if any (blueprint §13). + pub bound_user: Option, } /// The control API exposed by the plugin. Reachable via @@ -50,18 +52,18 @@ pub trait RelayAgent: Send + Sync { /// Derived namespace id (hex). fn namespace_id(&self) -> String; - /// Send the current Inbox snapshot to all authorized clients. - async fn broadcast_inbox(&self) -> anyhow::Result<()>; - - /// Generic push notification to all authorized clients. - async fn broadcast_notification(&self, title: &str, body: &str) -> anyhow::Result<()>; - - /// List all known devices. + /// List all known devices, each tagged with its bound user (if any). async fn list_clients(&self) -> Vec; - /// Authorize a Pending device by its ed25519 pubkey. - async fn authorize_client(&self, ed25519_pub: [u8; 32]) -> anyhow::Result<()>; + /// Bind a paired device to a Skald user and authorize it (blueprint §13, + /// admin-mediated — the mobile analogue of `telegram_pairing`). + async fn bind_device( + &self, + ed25519_pub: [u8; 32], + user_id: String, + display: Option, + ) -> anyhow::Result<()>; - /// Revoke a device (lost/stolen) by its ed25519 pubkey. + /// Revoke a device (lost/stolen) by its ed25519 pubkey and drop its binding. async fn revoke_client(&self, ed25519_pub: [u8; 32]) -> anyhow::Result<()>; } diff --git a/crates/plugin-mobile-connector/src/app.rs b/crates/plugin-mobile-connector/src/app.rs index aeec5a2..763a998 100644 --- a/crates/plugin-mobile-connector/src/app.rs +++ b/crates/plugin-mobile-connector/src/app.rs @@ -1,40 +1,80 @@ //! The application layer on top of the payload-agnostic [`RelayClient`]. //! -//! `RelayApp` owns the Skald-specific semantics that the transport crate -//! deliberately knows nothing about: the E2E JSON payload schemas (`payloads`), -//! the `InboxApi` dispatch, and the authorization policy. It consumes -//! `client.events()` and calls `client.send(...)`; the client handles the wire, -//! crypto, counters, and device registry. +//! `RelayApp` is the plugin's central shared hub (the mobile analogue of the +//! Telegram `TgShared`): it owns the E2E payload semantics, the device→user +//! bindings cache, and the per-user forwarder/notifier registries. It knows +//! nothing about the wire — the client handles transport, crypto, counters, and +//! the device registry. +//! +//! # Multi-user (blueprint §13) +//! +//! Every device is bound to one Skald user (`auth::Binding`). Inbound payloads +//! resolve the sending device's user, then apply to **that user's** Inbox via the +//! [`UserChannelApi`] seam. Outbound Inbox pushes go only to the authorized +//! devices bound to the target user — never a global broadcast. A per-user +//! forwarder (`events`) drives pushes from the user's event stream. +use std::collections::{HashMap, HashSet}; use std::sync::Arc; +use std::time::Duration; use anyhow::Result; -use tokio::sync::broadcast; +use tokio::sync::{broadcast, Mutex, RwLock}; use tokio_util::sync::CancellationToken; -use tracing::{debug, warn}; +use tracing::{debug, info, warn}; -use core_api::inbox::InboxApi; +use core_api::config_api::ConfigApi; +use core_api::user_channel::UserChannelApi; use skald_relay_client::{ClientState, RelayClient, RelayEvent}; use crate::PLUGIN_ID; +use crate::auth::{self, MobileConfig}; +use crate::notifier::DelayedNotifier; use crate::payloads::{self, ClientPayload}; -/// Glue between the relay transport ([`RelayClient`]) and Skald's Inbox. +/// The plugin's shared application state. pub struct RelayApp { client: Arc, - inbox: Arc, - /// When true, a freshly paired device stays Pending until a human confirms; - /// when false, the app auto-authorizes on `ClientPaired`. + /// Per-user runtime resolver (blueprint §13). `None` = user locked (§9). + pub(crate) user_channel: Arc, + /// Config store — used to persist binding removals (logout/revoke). + config: Arc, + /// Device→user bindings, cached in memory; kept in sync by `auth::config_listener`. + pub(crate) bindings: RwLock, + /// When true, a freshly paired device stays Pending until an admin binds it + /// (`mobile_bind_device`, which authorizes it). Binding *is* the confirmation. require_device_confirmation: bool, + /// Debounce before an unresolved Inbox item is pushed to the phone. + pub(crate) notify_delay: Duration, + /// Cancellation for every task spawned by this run (forwarders, listeners). + cancel: CancellationToken, + /// user_ids with an active per-user forwarder task. + pub(crate) forwarders: Mutex>, + /// Per-user debounced notifiers, created on demand by the forwarders. + pub(crate) notifiers: Mutex>>, } impl RelayApp { pub fn new( client: Arc, - inbox: Arc, + user_channel: Arc, + config: Arc, + bindings: MobileConfig, require_device_confirmation: bool, - ) -> Self { - Self { client, inbox, require_device_confirmation } + notify_delay: Duration, + cancel: CancellationToken, + ) -> Arc { + Arc::new(Self { + client, + user_channel, + config, + bindings: RwLock::new(bindings), + require_device_confirmation, + notify_delay, + cancel, + forwarders: Mutex::new(HashSet::new()), + notifiers: Mutex::new(HashMap::new()), + }) } /// The underlying transport client (used by the `RelayAgent` impl + router). @@ -42,128 +82,219 @@ impl RelayApp { &self.client } - // ── Inbox → clients ─────────────────────────────────────────────────────── + /// Cancellation token for this run's spawned tasks. + pub(crate) fn cancel(&self) -> CancellationToken { + self.cancel.clone() + } - /// Build the Inbox snapshot and send it (encrypted) to every Authorized - /// client. `live=false` so the relay stores-and-forwards + pushes to offline - /// phones. - pub async fn broadcast_inbox(&self) -> Result<()> { - let snapshot = self.inbox.list_pending().await; + /// Write guard on the bindings cache (used by the config listener). + pub(crate) async fn bindings_mut(&self) -> tokio::sync::RwLockWriteGuard<'_, MobileConfig> { + self.bindings.write().await + } + + /// The user bound to a device pubkey, if any. + pub(crate) async fn user_for_device(&self, pubkey: &[u8; 32]) -> Option { + self.bindings.read().await.user_for_pubkey(&hex::encode(pubkey)) + } + + // ── Bind / unbind (admin-mediated, via the control tools) ───────────────── + + /// Bind a paired device to a user and authorize it, then push that user's + /// current Inbox to it. Persists the binding (fires `ConfigKeyUpdated`, which + /// refreshes the cache and spawns the user's forwarder). + pub async fn bind_device( + &self, + pubkey: [u8; 32], + user_id: String, + display: Option, + ) -> Result<()> { + // Update the persisted config + local cache up front (avoids a race with + // the listener before its event arrives). + let snapshot = { + let mut cfg = self.bindings.write().await; + cfg.upsert(auth::Binding { pubkey_hex: hex::encode(pubkey), user_id: user_id.clone(), display }); + cfg.clone() + }; + auth::save_config(&*self.config, &snapshot).await?; + + // Authorize at the relay level so pushes/sends reach the device. + self.client.authorize(&pubkey).await?; + info!(plugin = PLUGIN_ID, user_id = %user_id, device = %hex::encode(pubkey), "device bound + authorized"); + + // Send the user's current Inbox to the freshly bound device. + if let Err(e) = self.push_inbox_to_user(&user_id).await { + warn!(plugin = PLUGIN_ID, error = %e, "failed to push inbox after bind"); + } + Ok(()) + } + + /// Revoke a device and drop its binding. + pub async fn revoke_device(&self, pubkey: [u8; 32]) -> Result<()> { + self.client.revoke(&pubkey).await?; + let snapshot = { + let mut cfg = self.bindings.write().await; + cfg.remove(&hex::encode(pubkey)); + cfg.clone() + }; + auth::save_config(&*self.config, &snapshot).await?; + Ok(()) + } + + // ── Inbox → user's devices ──────────────────────────────────────────────── + + /// Build the Inbox snapshot for `user_id` and send it to every Authorized + /// device bound to that user. `live=false` so the relay stores-and-forwards + + /// pushes to offline phones. No-op if the user is locked (§9). + pub async fn push_inbox_to_user(&self, user_id: &str) -> Result<()> { + let Some(handle) = self.user_channel.resolve_user(user_id).await else { + debug!(plugin = PLUGIN_ID, user_id, "inbox push skipped — user locked"); + return Ok(()); + }; + let snapshot = handle.inbox().list_pending().await; let plaintext = serde_json::to_vec(&payloads::build_inbox_update(&snapshot))?; - self.broadcast_plaintext(&plaintext).await; + self.send_to_user_devices(user_id, &plaintext, false).await; Ok(()) } - /// Build and send a generic notification to all Authorized clients. - pub async fn broadcast_notification(&self, title: &str, body: &str) -> Result<()> { - let plaintext = serde_json::to_vec(&payloads::build_notification(title, body))?; - self.broadcast_plaintext(&plaintext).await; - Ok(()) - } - - /// Send an opaque plaintext to every Authorized device (`live=false`). - async fn broadcast_plaintext(&self, plaintext: &[u8]) { - for c in self + /// Send an opaque plaintext to every Authorized device bound to `user_id`. + async fn send_to_user_devices(&self, user_id: &str, plaintext: &[u8], live: bool) { + let bound = self.bindings.read().await.pubkeys_for_user(user_id); + if bound.is_empty() { + return; + } + let authorized: HashSet<[u8; 32]> = self .client .list_clients() .await .into_iter() .filter(|c| c.state == ClientState::Authorized) - { - if let Err(e) = self.client.send(&c.ed25519_pub, plaintext, false).await { - warn!(plugin = PLUGIN_ID, error = %e, "failed to send to client"); + .map(|c| c.ed25519_pub) + .collect(); + for pk_hex in bound { + let Some(pk) = skald_relay_common::crypto::decode_hex::<32>(&pk_hex) else { continue }; + if !authorized.contains(&pk) { + continue; + } + if let Err(e) = self.client.send(&pk, plaintext, live).await { + warn!(plugin = PLUGIN_ID, error = %e, "failed to send to device"); } } } - /// Send the current Inbox snapshot to a single client (the targeted reply to - /// `inbox_request`). `live=true`: the requester is online by construction. - async fn send_inbox_to(&self, client_ed25519_pub: &[u8; 32]) -> Result<()> { - let snapshot = self.inbox.list_pending().await; + /// Send the current Inbox snapshot to a single requesting device (`live=true`: + /// the requester is online by construction). + async fn send_inbox_to_device(&self, user_id: &str, device: &[u8; 32]) -> Result<()> { + let Some(handle) = self.user_channel.resolve_user(user_id).await else { + return Ok(()); + }; + let snapshot = handle.inbox().list_pending().await; let plaintext = serde_json::to_vec(&payloads::build_inbox_update(&snapshot))?; - self.client.send(client_ed25519_pub, &plaintext, true).await + self.client.send(device, &plaintext, true).await } - // ── Clients → Inbox ─────────────────────────────────────────────────────── + // ── Devices → Inbox ─────────────────────────────────────────────────────── - /// Apply a decoded client payload to the Inbox. `payload` is the clean inner - /// JSON the client already decrypted + de-framed. + /// Apply a decoded client payload to the sending device's *user's* Inbox. + /// Unbound device or locked user → the request is ignored (no cross-user leak). async fn apply_client_payload(&self, from: &[u8; 32], payload: &[u8]) { - match payloads::parse_client_payload(payload) { - ClientPayload::ApprovalResponse { request_id, approved, reason } => { - if approved { - self.inbox.approve(request_id).await; - } else { - self.inbox.reject(request_id, reason.unwrap_or_default()).await; - } - let _ = self.broadcast_inbox().await; - } - ClientPayload::ClarificationResponse { request_id, answer } => { - self.inbox.answer(request_id, answer).await; - let _ = self.broadcast_inbox().await; - } - ClientPayload::ElicitationResponse { request_id, action, content } => { - // `content` may hold a secret (SSH/sudo password): hand it straight - // to the Inbox; never log/persist it in clear (payloads.md §3.1). - self.inbox.resolve_elicitation(request_id, action, content).await; - let _ = self.broadcast_inbox().await; - } + let parsed = payloads::parse_client_payload(payload); + + // Hello / Logout are device-registry ops that need no user resolution. + match &parsed { ClientPayload::Hello { device_info } => { if let Err(e) = self.client.set_device_info(from, &device_info.to_string()).await { warn!(plugin = PLUGIN_ID, error = %e, "failed to persist device_info"); } - } - ClientPayload::InboxRequest => { - if let Err(e) = self.send_inbox_to(from).await { - warn!(plugin = PLUGIN_ID, error = %e, "failed to send targeted inbox snapshot"); - } + return; } ClientPayload::Logout => { - if let Err(e) = self.client.revoke(from).await { + if let Err(e) = self.revoke_device(*from).await { warn!(plugin = PLUGIN_ID, error = %e, "logout revoke failed"); } + return; } ClientPayload::Unknown => { debug!(plugin = PLUGIN_ID, "unknown/ignored client payload"); + return; } + _ => {} + } + + // Everything else acts on a user's Inbox: resolve the device's user. + let Some(user_id) = self.user_for_device(from).await else { + warn!(plugin = PLUGIN_ID, device = %hex::encode(from), "payload from unbound device — ignored"); + return; + }; + let Some(handle) = self.user_channel.resolve_user(&user_id).await else { + debug!(plugin = PLUGIN_ID, user_id = %user_id, "payload dropped — user locked"); + return; + }; + let inbox = handle.inbox(); + + match parsed { + ClientPayload::ApprovalResponse { request_id, approved, reason } => { + if approved { + inbox.approve(request_id).await; + } else { + inbox.reject(request_id, reason.unwrap_or_default()).await; + } + let _ = self.push_inbox_to_user(&user_id).await; + } + ClientPayload::ClarificationResponse { request_id, answer } => { + inbox.answer(request_id, answer).await; + let _ = self.push_inbox_to_user(&user_id).await; + } + ClientPayload::ElicitationResponse { request_id, action, content } => { + // `content` may hold a secret (SSH/sudo password): hand it straight + // to the Inbox; never log/persist it in clear (payloads.md §3.1). + inbox.resolve_elicitation(request_id, action, content).await; + let _ = self.push_inbox_to_user(&user_id).await; + } + ClientPayload::InboxRequest => { + if let Err(e) = self.send_inbox_to_device(&user_id, from).await { + warn!(plugin = PLUGIN_ID, error = %e, "failed to send targeted inbox snapshot"); + } + } + // Handled above. + ClientPayload::Hello { .. } | ClientPayload::Logout | ClientPayload::Unknown => {} } } // ── Event loop ──────────────────────────────────────────────────────────── - /// Consume the client's [`RelayEvent`] stream until `cancel` fires. This is - /// where the authorization policy and Inbox application live. - pub async fn run_event_loop( - self: Arc, - mut rx: broadcast::Receiver, - cancel: CancellationToken, - ) { + /// Consume the client's [`RelayEvent`] stream until cancelled. Applies inbound + /// payloads and the pairing authorization policy, and lazily spawns a per-user + /// forwarder when a bound device becomes active. + pub async fn run_event_loop(self: Arc, mut rx: broadcast::Receiver) { + let cancel = self.cancel.clone(); loop { tokio::select! { _ = cancel.cancelled() => break, ev = rx.recv() => match ev { Ok(RelayEvent::Message { from, payload, .. }) => { + // A bound + unlocked device becoming active ⇒ ensure its + // user's forwarder is running so Inbox events reach the phone. + if let Some(user_id) = self.user_for_device(&from).await { + if let Some(handle) = self.user_channel.resolve_user(&user_id).await { + crate::events::ensure_forwarder(Arc::clone(&self), user_id, handle).await; + } + } self.apply_client_payload(&from, &payload).await; } Ok(RelayEvent::ClientPaired { ed25519_pub, .. }) => { - if self.require_device_confirmation { - debug!( - plugin = PLUGIN_ID, - device = %hex::encode(ed25519_pub), - "new device paired (pending manual confirmation)" - ); - let _ = self - .broadcast_notification( - "New device", - "A new device is pending confirmation", - ) - .await; - } else if let Err(e) = self.client.authorize(&ed25519_pub).await { - warn!(plugin = PLUGIN_ID, error = %e, "auto-authorize failed"); - } else { - // Send the newly-authorized device the current snapshot. - let _ = self.broadcast_inbox().await; + // The device is not bound to any user yet, so there is no + // one to push to. An admin binds it with `mobile_bind_device` + // (which authorizes it). We only optionally pre-authorize. + if !self.require_device_confirmation { + if let Err(e) = self.client.authorize(&ed25519_pub).await { + warn!(plugin = PLUGIN_ID, error = %e, "auto-authorize failed"); + } } + info!( + plugin = PLUGIN_ID, + device = %hex::encode(ed25519_pub), + "new device paired — awaiting admin binding (mobile_bind_device)" + ); } Ok(RelayEvent::ClientRevoked { .. }) | Ok(RelayEvent::Connected) diff --git a/crates/plugin-mobile-connector/src/auth.rs b/crates/plugin-mobile-connector/src/auth.rs new file mode 100644 index 0000000..5d36c58 --- /dev/null +++ b/crates/plugin-mobile-connector/src/auth.rs @@ -0,0 +1,207 @@ +//! Device→user binding (blueprint §13), the mobile analogue of the Telegram +//! `chat_id ↔ user_id` pairing. +//! +//! One relay identity serves many devices; each **authorized device** (identified +//! by its ed25519 public key) is bound to exactly one Skald user. The binding +//! lives in the `config` table (key `"mobile-connector"`) as JSON, so it survives +//! restarts and is admin-editable. Writing it emits a `ConfigKeyUpdated` event, +//! which [`config_listener`] uses to refresh the in-memory cache and (re)spawn +//! per-user forwarders — no polling, no restart. +//! +//! Unlike Telegram there is no "pairing code" concept here: the QR pairing and the +//! Pending/Authorized device lifecycle are handled by `skald-relay-client`. A +//! binding simply ties an already-paired pubkey to a user (admin-mediated, via the +//! `mobile_bind_device` tool — the mobile analogue of `telegram_pairing`). + +use std::sync::Arc; + +use serde::{Deserialize, Serialize}; +use tokio::sync::broadcast; +use tracing::{info, warn}; + +use core_api::config_api::ConfigApi; +use core_api::system_bus::SystemEvent; + +use crate::PLUGIN_ID; +use crate::app::RelayApp; + +/// Config-table key under which all mobile bindings are stored as JSON. +pub(crate) const CONFIG_KEY: &str = "mobile-connector"; + +// ── Bindings schema (stored as JSON in the `config` table) ──────────────────── + +#[derive(Debug, Serialize, Deserialize, Default, Clone)] +pub struct MobileConfig { + #[serde(default)] + pub bindings: Vec, +} + +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct Binding { + /// The device's ed25519 public key, hex-encoded (64 chars) — the stable, + /// opaque device identity the relay registry keys on. + pub pubkey_hex: String, + pub user_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub display: Option, +} + +impl MobileConfig { + /// The user bound to `pubkey_hex`, if any. + pub fn user_for_pubkey(&self, pubkey_hex: &str) -> Option { + self.bindings + .iter() + .find(|b| b.pubkey_hex.eq_ignore_ascii_case(pubkey_hex)) + .map(|b| b.user_id.clone()) + } + + /// Hex pubkeys of every device bound to `user_id`. + pub fn pubkeys_for_user(&self, user_id: &str) -> Vec { + self.bindings + .iter() + .filter(|b| b.user_id == user_id) + .map(|b| b.pubkey_hex.clone()) + .collect() + } + + /// The distinct user ids that have at least one bound device. + pub fn bound_user_ids(&self) -> Vec { + let mut ids: Vec = self.bindings.iter().map(|b| b.user_id.clone()).collect(); + ids.sort(); + ids.dedup(); + ids + } + + /// Insert or replace the binding for a pubkey (a device belongs to one user). + pub fn upsert(&mut self, binding: Binding) { + self.bindings + .retain(|b| !b.pubkey_hex.eq_ignore_ascii_case(&binding.pubkey_hex)); + self.bindings.push(binding); + } + + /// Remove the binding for `pubkey_hex`. Returns true if one was removed. + pub fn remove(&mut self, pubkey_hex: &str) -> bool { + let before = self.bindings.len(); + self.bindings + .retain(|b| !b.pubkey_hex.eq_ignore_ascii_case(pubkey_hex)); + self.bindings.len() != before + } +} + +// ── Config-table read/write ──────────────────────────────────────────────────── + +/// Reads the mobile config from the `config` table. Returns `Default` when the key +/// is absent or unparseable (never fails the caller). +pub(crate) async fn load_config(config: &dyn ConfigApi) -> anyhow::Result { + match config.get(CONFIG_KEY).await? { + Some(json) => Ok(serde_json::from_str(&json).unwrap_or_default()), + None => Ok(MobileConfig::default()), + } +} + +/// Writes the mobile config to the `config` table. `ConfigApi::set` emits a +/// `ConfigKeyUpdated` event when the value changes, so [`config_listener`] and the +/// in-memory cache pick it up automatically. +pub(crate) async fn save_config(config: &dyn ConfigApi, cfg: &MobileConfig) -> anyhow::Result<()> { + config + .set(CONFIG_KEY, &serde_json::to_string_pretty(cfg)?) + .await +} + +// ── Config listener ──────────────────────────────────────────────────────────── + +/// Subscribes to the system bus and reloads the in-memory bindings whenever the +/// `"mobile-connector"` config key changes, then (re)spawns forwarders for every +/// bound + unlocked user. The mobile analogue of Telegram's `config_listener`. +pub(crate) async fn config_listener( + app: Arc, + mut rx: broadcast::Receiver, +) { + let cancel = app.cancel(); + info!(plugin = PLUGIN_ID, "config listener started"); + loop { + tokio::select! { + _ = cancel.cancelled() => { + info!(plugin = PLUGIN_ID, "config listener stopped"); + return; + } + result = rx.recv() => match result { + Ok(SystemEvent::ConfigKeyUpdated { key, new_value, .. }) if key == CONFIG_KEY => { + match serde_json::from_str::(&new_value) { + Ok(cfg) => { + let n = cfg.bindings.len(); + *app.bindings_mut().await = cfg; + info!(plugin = PLUGIN_ID, bindings = n, "bindings reloaded from config event"); + crate::events::spawn_forwarders_for_bound_users(&app).await; + } + Err(e) => warn!(plugin = PLUGIN_ID, error = %e, "failed to parse config from event"), + } + } + Ok(_) => {} + Err(broadcast::error::RecvError::Lagged(n)) => { + warn!(plugin = PLUGIN_ID, skipped = n, "config listener lagged"); + } + Err(broadcast::error::RecvError::Closed) => return, + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn binding(pubkey_hex: &str, user_id: &str) -> Binding { + Binding { pubkey_hex: pubkey_hex.to_string(), user_id: user_id.to_string(), display: None } + } + + #[test] + fn upsert_replaces_by_pubkey_case_insensitive() { + let mut cfg = MobileConfig::default(); + cfg.upsert(binding("AABB", "user-a")); + // Re-binding the same device (different hex case) to another user replaces, + // never duplicates — a device belongs to exactly one user. + cfg.upsert(binding("aabb", "user-b")); + assert_eq!(cfg.bindings.len(), 1); + assert_eq!(cfg.user_for_pubkey("aabb").as_deref(), Some("user-b")); + } + + #[test] + fn user_for_pubkey_is_case_insensitive() { + let mut cfg = MobileConfig::default(); + cfg.upsert(binding("DEADbeef", "alice")); + assert_eq!(cfg.user_for_pubkey("deadBEEF").as_deref(), Some("alice")); + assert_eq!(cfg.user_for_pubkey("0000"), None); + } + + #[test] + fn pubkeys_for_user_scopes_to_owner() { + let mut cfg = MobileConfig::default(); + cfg.upsert(binding("aa", "alice")); + cfg.upsert(binding("bb", "alice")); + cfg.upsert(binding("cc", "bob")); + let mut alice = cfg.pubkeys_for_user("alice"); + alice.sort(); + assert_eq!(alice, vec!["aa".to_string(), "bb".to_string()]); + assert_eq!(cfg.pubkeys_for_user("bob"), vec!["cc".to_string()]); + assert!(cfg.pubkeys_for_user("carol").is_empty()); + } + + #[test] + fn bound_user_ids_dedup_sorted() { + let mut cfg = MobileConfig::default(); + cfg.upsert(binding("aa", "bob")); + cfg.upsert(binding("bb", "alice")); + cfg.upsert(binding("cc", "bob")); + assert_eq!(cfg.bound_user_ids(), vec!["alice".to_string(), "bob".to_string()]); + } + + #[test] + fn remove_reports_and_is_case_insensitive() { + let mut cfg = MobileConfig::default(); + cfg.upsert(binding("AbCd", "alice")); + assert!(cfg.remove("abcd")); + assert!(!cfg.remove("abcd")); + assert!(cfg.bindings.is_empty()); + } +} diff --git a/crates/plugin-mobile-connector/src/events.rs b/crates/plugin-mobile-connector/src/events.rs new file mode 100644 index 0000000..23825ca --- /dev/null +++ b/crates/plugin-mobile-connector/src/events.rs @@ -0,0 +1,146 @@ +//! Per-user event forwarders (blueprint §13), the mobile analogue of the Telegram +//! `ensure_forwarder`. +//! +//! One forwarder per unlocked user with bound devices subscribes to that user's +//! event stream ([`UserChannelHandle::subscribe`]) and routes the six Inbox +//! lifecycle events through the user's [`DelayedNotifier`], which decides whether +//! and when to push the Inbox to their phones. +//! +//! Unlike the Telegram forwarder there is **no `source` filter**: an approval +//! raised in the user's *web* session must still reach their phone. The relevant +//! events (`{Approval,Clarification,Elicitation}{Requested,Resolved}`) are +//! Inbox-scoped and carry request ids from the user's own pool, so no cross-user +//! collision is possible. + +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::broadcast; +use tokio_util::sync::CancellationToken; +use tracing::{info, warn}; + +use core_api::events::ServerEvent; +use core_api::user_channel::UserChannelHandle; + +use crate::PLUGIN_ID; +use crate::app::RelayApp; +use crate::notifier::{DelayedNotifier, Kind}; + +/// How often the reconcile loop picks up bound users who have unlocked since the +/// last pass. Pushes are best-effort, so a coarse cadence is fine. +const RECONCILE_INTERVAL: Duration = Duration::from_secs(60); + +/// Periodically (re)spawns forwarders for bound + unlocked users. +/// +/// This is load-bearing, not a nicety: at boot every pool is locked (§9), so the +/// eager start-time pass spawns nothing. Users unlock later via web/phone login, +/// and there is no "user unlocked" system event to hook. Without this loop a user +/// whose phone stays backgrounded would never get a forwarder — so no Inbox push +/// would ever be armed for them. `ensure_forwarder` dedups, so this is idempotent +/// and cheap (locked users resolve to `None` and are skipped without a build). +pub(crate) async fn reconcile_loop(app: Arc) { + let cancel = app.cancel(); + let mut ticker = tokio::time::interval(RECONCILE_INTERVAL); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = cancel.cancelled() => return, + _ = ticker.tick() => spawn_forwarders_for_bound_users(&app).await, + } + } +} + +/// Spawns forwarders for every bound user whose context is already unlocked. +/// Called at plugin start, on binding changes, and by the reconcile loop. Users +/// who unlock later also get their forwarder spawned lazily on device activity. +pub(crate) async fn spawn_forwarders_for_bound_users(app: &Arc) { + let user_ids = app.bindings.read().await.bound_user_ids(); + for user_id in user_ids { + if let Some(handle) = app.user_channel.resolve_user(&user_id).await { + ensure_forwarder(Arc::clone(app), user_id, handle).await; + } + } +} + +/// Spawns a per-user forwarder if one is not already running for `user_id`. +pub(crate) async fn ensure_forwarder( + app: Arc, + user_id: String, + handle: Arc, +) { + { + let mut forwarders = app.forwarders.lock().await; + if !forwarders.insert(user_id.clone()) { + return; + } + } + let cancel = app.cancel(); + info!(plugin = PLUGIN_ID, user_id = %user_id, "spawning per-user forwarder"); + tokio::spawn(user_forwarder(app, user_id, handle, cancel)); +} + +/// One forwarder per unlocked user. Subscribes to the user's event stream and +/// drives their `DelayedNotifier`. Exits when the stream closes (user context +/// dropped at restart / lock) or the plugin is cancelled — self-cleaning. +async fn user_forwarder( + app: Arc, + user_id: String, + handle: Arc, + cancel: CancellationToken, +) { + // Get or create this user's debounced notifier. + let notifier: Arc = { + let mut notifiers = app.notifiers.lock().await; + notifiers + .entry(user_id.clone()) + .or_insert_with(|| DelayedNotifier::new(Arc::downgrade(&app), user_id.clone(), app.notify_delay)) + .clone() + }; + + let mut rx = handle.subscribe(); + loop { + let event: ServerEvent = tokio::select! { + _ = cancel.cancelled() => break, + result = rx.recv() => match result { + Ok(ge) => ge.event, + Err(broadcast::error::RecvError::Lagged(n)) => { + warn!(plugin = PLUGIN_ID, user_id = %user_id, skipped = n, "forwarder lagged"); + continue; + } + Err(broadcast::error::RecvError::Closed) => { + info!(plugin = PLUGIN_ID, user_id = %user_id, "forwarder — user context closed, exiting"); + break; + } + }, + }; + + // `*Requested` arms a delayed push; `*Resolved` cancels it (or refreshes + // the phone if the push already went out). Any other event is ignored. + match event { + ServerEvent::ApprovalRequested { request_id, .. } => { + notifier.on_requested((Kind::Approval, request_id)).await; + } + ServerEvent::ApprovalResolved { request_id, .. } => { + notifier.on_resolved((Kind::Approval, request_id)).await; + } + ServerEvent::ClarificationRequested { request_id, .. } => { + notifier.on_requested((Kind::Clarification, request_id)).await; + } + ServerEvent::ClarificationResolved { request_id } => { + notifier.on_resolved((Kind::Clarification, request_id)).await; + } + ServerEvent::ElicitationRequested { request_id, .. } => { + notifier.on_requested((Kind::Elicitation, request_id)).await; + } + ServerEvent::ElicitationResolved { request_id } => { + notifier.on_resolved((Kind::Elicitation, request_id)).await; + } + _ => {} + } + } + + // Clean up so a later reconnect can respawn. + app.forwarders.lock().await.remove(&user_id); + app.notifiers.lock().await.remove(&user_id); + info!(plugin = PLUGIN_ID, user_id = %user_id, "forwarder exited"); +} diff --git a/crates/plugin-mobile-connector/src/lib.rs b/crates/plugin-mobile-connector/src/lib.rs index 6f0ebd4..7069cff 100644 --- a/crates/plugin-mobile-connector/src/lib.rs +++ b/crates/plugin-mobile-connector/src/lib.rs @@ -1,21 +1,36 @@ //! Mobile connector plugin (plugin id `mobile-connector`). //! -//! Bridges Skald's Inbox (approvals + clarifications) to mobile apps over the -//! relay. The **networking** (v2 WS transport, E2E crypto, anti-replay counters, -//! pairing, device authorization, SQLite persistence) lives in the standalone -//! `skald-relay-client` crate; this plugin is the thin **application** layer on -//! top of it. See `data/iOS-app/v2/relay-protocol.md` for the wire contract and -//! `docs/relay/` for the client/server split. +//! Bridges each Skald user's Inbox (approvals + clarifications + elicitations) to +//! their mobile devices over the relay, end-to-end encrypted. The **networking** +//! (v2 WS transport, E2E crypto, anti-replay counters, pairing, device +//! authorization, SQLite persistence) lives in the standalone `skald-relay-client` +//! crate; this plugin is the thin **application** layer on top of it. See +//! `data/iOS-app/v2/relay-protocol.md` for the wire contract. +//! +//! # Multi-user (blueprint §13) +//! +//! One relay identity serves many devices; each device is bound to one Skald user +//! (`auth`, admin-mediated via `mobile_bind_device`). Inbound payloads apply to +//! that user's Inbox via the [`UserChannelApi`] seam; per-user forwarders +//! (`events`) push Inbox changes only to that user's devices — never a global +//! broadcast. The HTTP reverse proxy (`proxy`) is user-agnostic: the phone renders +//! the authenticated web UI over the tunnel, which handles per-user auth itself. //! //! Module map: //! - `payloads` — E2E JSON payload schemas (inbox_update, responses, …) -//! - `app` — `RelayApp`: Inbox dispatch, auth policy, the events() loop +//! - `auth` — device→user bindings (config-table-backed) + config listener +//! - `app` — `RelayApp`: per-user Inbox dispatch, bindings, the events() loop +//! - `events` — per-user event forwarders (drive the notifiers) +//! - `notifier` — per-user debounced Inbox pushes +//! - `proxy` — HTTP reverse proxy to the local web UI (user-agnostic) //! - `router` — the QR-code HTTP endpoint //! - `agent` — the `RelayAgent` control trait //! - `tools` — `Tool` impls callable by the host (registered in the main crate) mod agent; mod app; +mod auth; +mod events; mod notifier; mod payloads; mod proxy; @@ -40,7 +55,6 @@ pub use agent::{ClientInfo, ClientState, PairingHandle, RelayAgent}; pub use tools::mobile_tools; use app::RelayApp; -use notifier::{DelayedNotifier, Kind}; pub(crate) const PLUGIN_ID: &str = "mobile-connector"; const DEFAULT_TTL: u32 = 300; @@ -60,8 +74,6 @@ pub struct MobileConnectorPlugin { inner: Arc>>>, cancel: Mutex>, handles: Mutex>>, - /// Debounces Inbox pushes to the phone; present only while running. - notifier: Mutex>>, } impl MobileConnectorPlugin { @@ -71,7 +83,6 @@ impl MobileConnectorPlugin { inner: Arc::new(Mutex::new(None)), cancel: Mutex::new(None), handles: Mutex::new(Vec::new()), - notifier: Mutex::new(None), } } @@ -119,69 +130,49 @@ impl MobileConnectorPlugin { ); client.start().await?; - let app = Arc::new(RelayApp::new( - Arc::clone(&client), - Arc::clone(&ctx.inbox), - require_device_confirmation, - )); - - let notifier = DelayedNotifier::new(Arc::clone(&app), notify_delay); - let cancel = CancellationToken::new(); + + // Load device→user bindings from the config table (or default if absent). + let bindings = auth::load_config(&*ctx.config).await.unwrap_or_default(); + info!(plugin = PLUGIN_ID, bindings = bindings.bindings.len(), "bindings loaded"); + + let app = RelayApp::new( + Arc::clone(&client), + Arc::clone(&ctx.user_channel), + Arc::clone(&ctx.config), + bindings, + require_device_confirmation, + notify_delay, + cancel.clone(), + ); + let mut handles = Vec::new(); - // Event loop: apply inbound payloads + authorization policy. + // Event loop: apply inbound payloads + pairing authorization policy, and + // lazily spawn per-user forwarders when a bound device becomes active. { let app2 = Arc::clone(&app); let rx = client.events(); - let c = cancel.clone(); handles.push(tokio::spawn(async move { - app2.run_event_loop(rx, c).await; + app2.run_event_loop(rx).await; })); } - // Bus subscriber: route the four Inbox events through the debouncer. - // `*Requested` arms a delayed push; `*Resolved` cancels it (or refreshes - // the phone if the push already went out). + // Config listener: reloads bindings when the "mobile-connector" config key + // changes (e.g. the bind tool writes a new binding) and (re)spawns forwarders. { - let notifier = Arc::clone(¬ifier); - let c = cancel.clone(); - let mut rx = ctx.chat_hub.events(PLUGIN_ID); - handles.push(tokio::spawn(async move { - use core_api::events::ServerEvent::*; - loop { - tokio::select! { - _ = c.cancelled() => break, - ev = rx.recv() => match ev { - Ok(ge) => match ge.event { - ApprovalRequested { request_id, .. } => { - notifier.on_requested((Kind::Approval, request_id)).await; - } - ApprovalResolved { request_id, .. } => { - notifier.on_resolved((Kind::Approval, request_id)).await; - } - ClarificationRequested { request_id, .. } => { - notifier.on_requested((Kind::Clarification, request_id)).await; - } - ClarificationResolved { request_id } => { - notifier.on_resolved((Kind::Clarification, request_id)).await; - } - ElicitationRequested { request_id, .. } => { - notifier.on_requested((Kind::Elicitation, request_id)).await; - } - ElicitationResolved { request_id } => { - notifier.on_resolved((Kind::Elicitation, request_id)).await; - } - _ => {} - }, - Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { - warn!(plugin = PLUGIN_ID, skipped = n, "event bus lagged"); - } - Err(_) => break, - } - } - } - })); + let app3 = Arc::clone(&app); + let bus_rx = ctx.system_bus.subscribe(); + handles.push(tokio::spawn(auth::config_listener(app3, bus_rx))); + } + + // Reconcile loop: (re)spawns forwarders for bound users as they unlock. Its + // first tick fires immediately (covering already-unlocked users at start), + // then it periodically catches users who log in later — there is no "user + // unlocked" event to hook, and at boot every pool is locked (§9). + { + let app4 = Arc::clone(&app); + handles.push(tokio::spawn(events::reconcile_loop(app4))); } // HTTP reverse proxy: bridge `http-local-proxy` pipes to the local web @@ -197,7 +188,6 @@ impl MobileConnectorPlugin { })); } - *self.notifier.lock().await = Some(notifier); *self.inner.lock().await = Some(app); *self.cancel.lock().await = Some(cancel); *self.handles.lock().await = handles; @@ -210,13 +200,12 @@ impl MobileConnectorPlugin { if let Some(c) = self.cancel.lock().await.take() { c.cancel(); } - // Cancel any armed (not-yet-fired) push timers. - if let Some(notifier) = self.notifier.lock().await.take() { - notifier.cancel_all().await; - } // Shut down the transport (cancels + joins the WS loop) before dropping - // the app. + // the app, cancelling any armed (not-yet-fired) per-user push timers first. if let Some(app) = self.inner.lock().await.take() { + for notifier in app.notifiers.lock().await.values() { + notifier.cancel_all().await; + } app.client().shutdown().await; } for h in self.handles.lock().await.drain(..) { @@ -363,47 +352,41 @@ impl RelayAgent for MobileConnectorPlugin { .unwrap_or_default() } - async fn broadcast_inbox(&self) -> Result<()> { - let app = self.app().await.ok_or_else(|| anyhow::anyhow!("plugin not running"))?; - app.broadcast_inbox().await - } - - async fn broadcast_notification(&self, title: &str, body: &str) -> Result<()> { - let app = self.app().await.ok_or_else(|| anyhow::anyhow!("plugin not running"))?; - app.broadcast_notification(title, body).await - } - async fn list_clients(&self) -> Vec { let Some(app) = self.app().await else { return Vec::new() }; - app.client() - .list_clients() - .await - .into_iter() - .map(|r| ClientInfo { - ed25519_pub: r.ed25519_pub, - x25519_pub: r.x25519_pub, - state: match r.state { - RelayClientState::Authorized => ClientState::Authorized, - RelayClientState::Pending => ClientState::Pending, - }, - device_info: r.device_info, - platform: r.platform, - last_seen: r.last_seen, + let rows = app.client().list_clients().await; + let bindings = app.bindings.read().await; + rows.into_iter() + .map(|r| { + let bound_user = bindings.user_for_pubkey(&hex::encode(r.ed25519_pub)); + ClientInfo { + ed25519_pub: r.ed25519_pub, + x25519_pub: r.x25519_pub, + state: match r.state { + RelayClientState::Authorized => ClientState::Authorized, + RelayClientState::Pending => ClientState::Pending, + }, + device_info: r.device_info, + platform: r.platform, + last_seen: r.last_seen, + bound_user, + } }) .collect() } - async fn authorize_client(&self, ed25519_pub: [u8; 32]) -> Result<()> { + async fn bind_device( + &self, + ed25519_pub: [u8; 32], + user_id: String, + display: Option, + ) -> Result<()> { let app = self.app().await.ok_or_else(|| anyhow::anyhow!("plugin not running"))?; - app.client().authorize(&ed25519_pub).await?; - // Send the current Inbox snapshot to the newly-authorized device - // (payload-agnostic client doesn't do this itself). - let _ = app.broadcast_inbox().await; - Ok(()) + app.bind_device(ed25519_pub, user_id, display).await } async fn revoke_client(&self, ed25519_pub: [u8; 32]) -> Result<()> { let app = self.app().await.ok_or_else(|| anyhow::anyhow!("plugin not running"))?; - app.client().revoke(&ed25519_pub).await + app.revoke_device(ed25519_pub).await } } diff --git a/crates/plugin-mobile-connector/src/notifier.rs b/crates/plugin-mobile-connector/src/notifier.rs index d39b894..6a5ddf5 100644 --- a/crates/plugin-mobile-connector/src/notifier.rs +++ b/crates/plugin-mobile-connector/src/notifier.rs @@ -14,9 +14,17 @@ //! Elicitations are the exception: they live only in the Inbox (never inline in //! the chat), so there is no computer-side answer to debounce against and they //! are pushed immediately regardless of `delay`. +//! +//! # Per-user (blueprint §13) +//! +//! There is one `DelayedNotifier` **per user** (owned by that user's forwarder), +//! so the `(kind, request_id)` keyspace is naturally scoped — request ids drawn +//! from different user pools never collide. On fire the notifier pushes only that +//! user's Inbox, via a `Weak` back-reference to the shared [`RelayApp`] (weak to +//! avoid an `Arc` cycle: `RelayApp` holds the notifiers). use std::collections::{HashMap, HashSet}; -use std::sync::Arc; +use std::sync::{Arc, Weak}; use std::time::Duration; use tokio::sync::Mutex; @@ -47,16 +55,27 @@ struct State { notified: HashSet, } -/// Debounces Inbox pushes to the phone. Cheap to clone (`Arc` inside). +/// Debounces Inbox pushes to one user's phones. pub struct DelayedNotifier { - app: Arc, + /// Weak to break the `RelayApp` → notifiers → `RelayApp` cycle. + app: Weak, + user_id: String, delay: Duration, state: Mutex, } impl DelayedNotifier { - pub fn new(app: Arc, delay: Duration) -> Arc { - Arc::new(Self { app, delay, state: Mutex::new(State::default()) }) + pub fn new(app: Weak, user_id: String, delay: Duration) -> Arc { + Arc::new(Self { app, user_id, delay, state: Mutex::new(State::default()) }) + } + + /// Push this user's Inbox to their phones, if the app is still alive. + async fn push(&self) { + if let Some(app) = self.app.upgrade() { + if let Err(e) = app.push_inbox_to_user(&self.user_id).await { + warn!(plugin = PLUGIN_ID, error = %e, "inbox push failed"); + } + } } /// A request entered the Inbox: arm a timer. If `delay` elapses before a @@ -78,9 +97,7 @@ impl DelayedNotifier { st.notified.insert(key) }; if newly_notified { - if let Err(e) = self.app.broadcast_inbox().await { - warn!(plugin = PLUGIN_ID, error = %e, "immediate elicitation push failed"); - } + self.push().await; } return; } @@ -111,9 +128,7 @@ impl DelayedNotifier { } }; if still_armed { - if let Err(e) = this.app.broadcast_inbox().await { - warn!(plugin = PLUGIN_ID, error = %e, "delayed inbox push failed"); - } + this.push().await; } } } @@ -135,9 +150,7 @@ impl DelayedNotifier { } }; if broadcast { - if let Err(e) = self.app.broadcast_inbox().await { - warn!(plugin = PLUGIN_ID, error = %e, "inbox broadcast failed"); - } + self.push().await; } } diff --git a/crates/plugin-mobile-connector/src/payloads.rs b/crates/plugin-mobile-connector/src/payloads.rs index 988fa72..77bb328 100644 --- a/crates/plugin-mobile-connector/src/payloads.rs +++ b/crates/plugin-mobile-connector/src/payloads.rs @@ -103,7 +103,10 @@ pub fn build_inbox_update(snapshot: &InboxSnapshot) -> Value { }) } -/// Build a generic `notification` payload (payloads.md §3.2). +/// Build a generic `notification` payload (payloads.md §3.2). Part of the wire +/// contract; retained for the protocol surface even though the current per-user +/// flow pushes Inbox updates rather than free-form notifications. +#[allow(dead_code)] pub fn build_notification(title: &str, body: &str) -> Value { serde_json::json!({ "v": 1, diff --git a/crates/plugin-mobile-connector/src/tools.rs b/crates/plugin-mobile-connector/src/tools.rs index 9d689a7..1ad2202 100644 --- a/crates/plugin-mobile-connector/src/tools.rs +++ b/crates/plugin-mobile-connector/src/tools.rs @@ -19,6 +19,7 @@ use crate::agent::{ClientState, RelayAgent}; /// Tool name constants (also the patterns the host uses for approval rules). pub const TOOL_START_PAIRING: &str = "mobile_start_pairing"; pub const TOOL_LIST_DEVICES: &str = "mobile_list_devices"; +pub const TOOL_BIND_DEVICE: &str = "mobile_bind_device"; pub const TOOL_REVOKE_DEVICE: &str = "mobile_revoke_device"; /// Build the plugin's LLM tools, bound to a `RelayAgent`. The host calls this @@ -27,6 +28,7 @@ pub fn mobile_tools(agent: Arc) -> Vec> { vec![ Arc::new(StartPairingTool { agent: Arc::clone(&agent) }), Arc::new(ListDevicesTool { agent: Arc::clone(&agent) }), + Arc::new(BindDeviceTool { agent: Arc::clone(&agent) }), Arc::new(RevokeDeviceTool { agent }), ] } @@ -82,7 +84,8 @@ struct ListDevicesTool { impl Tool for ListDevicesTool { fn name(&self) -> &str { TOOL_LIST_DEVICES } fn description(&self) -> &str { - "List paired mobile devices: state (pending/authorized), platform, device info, last seen." + "List paired mobile devices: state (pending/authorized), bound user, platform, device info, last seen. \ + Use the `ed25519_pub` of a pending device with mobile_bind_device to assign it to a user." } fn parameters_schema(&self) -> Value { json!({ "type": "object", "properties": {} }) @@ -109,6 +112,7 @@ impl Tool for ListDevicesTool { ClientState::Authorized => "authorized", ClientState::Pending => "pending", }, + "bound_user": c.bound_user, "platform": c.platform, "device_info": device_info, "last_seen": c.last_seen, @@ -120,6 +124,63 @@ impl Tool for ListDevicesTool { } } +// ── mobile_bind_device ──────────────────────────────────────────────────────── + +struct BindDeviceTool { + agent: Arc, +} + +impl Tool for BindDeviceTool { + fn name(&self) -> &str { TOOL_BIND_DEVICE } + fn description(&self) -> &str { + "Bind a paired mobile device to a Skald user and authorize it. The device then \ + receives that user's Inbox (approvals/clarifications/elicitations) and push \ + notifications — and only that user's. Requires user approval." + } + fn parameters_schema(&self) -> Value { + json!({ + "type": "object", + "properties": { + "pubkey": { + "type": "string", + "description": "The device's ed25519 public key, hex-encoded (64 chars), from mobile_list_devices." + }, + "user_id": { + "type": "string", + "description": "The Skald user id to bind this device to." + }, + "display": { + "type": "string", + "description": "Optional human-friendly label for the binding." + } + }, + "required": ["pubkey", "user_id"] + }) + } + fn category(&self) -> ToolCategory { ToolCategory::Config } + + fn execute_async<'a>( + &'a self, + args: Value, + ) -> std::pin::Pin> + Send + 'a>> { + Box::pin(async move { + let pubkey = args + .get("pubkey") + .and_then(Value::as_str) + .ok_or_else(|| anyhow::anyhow!("mobile_bind_device: missing `pubkey`"))?; + let user_id = args + .get("user_id") + .and_then(Value::as_str) + .ok_or_else(|| anyhow::anyhow!("mobile_bind_device: missing `user_id`"))?; + let display = args.get("display").and_then(Value::as_str).map(str::to_string); + let ed = skald_relay_common::crypto::decode_hex::<32>(pubkey) + .ok_or_else(|| anyhow::anyhow!("mobile_bind_device: `pubkey` is not 32-byte hex"))?; + self.agent.bind_device(ed, user_id.to_string(), display).await?; + Ok(format!("Device {pubkey} bound to user {user_id} and authorized.")) + }) + } +} + // ── mobile_revoke_device ────────────────────────────────────────────────────── struct RevokeDeviceTool { @@ -129,7 +190,8 @@ struct RevokeDeviceTool { impl Tool for RevokeDeviceTool { fn name(&self) -> &str { TOOL_REVOKE_DEVICE } fn description(&self) -> &str { - "Revoke a paired mobile device by its ed25519 public key (hex). The device loses access immediately." + "Revoke a paired mobile device by its ed25519 public key (hex) and drop its user binding. \ + The device loses access immediately." } fn parameters_schema(&self) -> Value { json!({ diff --git a/crates/skald-core/src/approval/mod.rs b/crates/skald-core/src/approval/mod.rs index 20dc7fa..58f5886 100644 --- a/crates/skald-core/src/approval/mod.rs +++ b/crates/skald-core/src/approval/mod.rs @@ -235,6 +235,10 @@ impl ApprovalManager { // Opening a mobile pairing window emits a secret (the QR) into chat: // it must be a deliberate human action, not LLM-triggerable (plugin.md §11). ("mobile_start_pairing", "require"), + // Binding/revoking a device assigns a phone to a user — a security + // decision (who receives whose Inbox), so it must be human-gated too. + ("mobile_bind_device", "require"), + ("mobile_revoke_device", "require"), ]; // NOTE: file-write tools are NOT seeded here as per-tool `require` rules. // Filesystem gating is owned by the "File System" category — path-scoped diff --git a/crates/skald-core/src/skald/user_context.rs b/crates/skald-core/src/skald/user_context.rs index d45df51..2747e91 100644 --- a/crates/skald-core/src/skald/user_context.rs +++ b/crates/skald-core/src/skald/user_context.rs @@ -32,6 +32,7 @@ use tokio_util::sync::CancellationToken; use core_api::approval::ApprovalApi; use core_api::chat_hub::ChatHubApi; use core_api::events::GlobalEvent; +use core_api::inbox::InboxApi; use core_api::system_bus::SystemEventBus; use core_api::user_channel::UserChannelHandle; @@ -292,6 +293,12 @@ impl UserChannelHandle for UserContextHandle { Arc::clone(&self.ctx.approval) as Arc } + fn inbox(&self) -> Arc { + // `Inbox` is a cheap facade over the interaction-stack `Arc`s (Clone); + // the clone shares the same pending state as the context's own inbox. + Arc::new(self.ctx.inbox.clone()) as Arc + } + fn subscribe(&self) -> broadcast::Receiver { self.ctx.global_tx.subscribe() } diff --git a/src/main.rs b/src/main.rs index 53453fa..392052c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -167,6 +167,7 @@ fn build_plugins() -> Vec> { let mut plugins: Vec> = vec![ Arc::new(plugin_tailscale_remote::RemotePlugin::new()), Arc::new(plugin_telegram_bot::TelegramPlugin::new()), + Arc::new(plugin_mobile_connector::MobileConnectorPlugin::new()), Arc::new(plugin_comfyui::ComfyUIPlugin::new()), Arc::new(plugin_tts_orpheus_3b::OrpheusTtsPlugin::new()), Arc::new(plugin_tts_kokoro::KokoroTtsPlugin::new()),