orchestral_cli/remote/
state.rs

1use std::path::{Path, PathBuf};
2use std::sync::Arc;
3
4use anyhow::{bail, Context};
5use base64::engine::general_purpose::URL_SAFE_NO_PAD;
6use base64::Engine;
7use chrono::Utc;
8use orchestral_core::agent_connector::AgentSessionExecutionProfile;
9use serde::{Deserialize, Serialize};
10use sha2::{Digest as _, Sha256};
11use tokio::sync::Mutex;
12
13const STATE_VERSION: u32 = 1;
14const DEVICE_TOKEN_PREFIX: &str = "orch_device_";
15
16#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
17pub struct DeviceView {
18    pub id: String,
19    pub name: String,
20    pub created_at_unix_ms: i64,
21    pub last_seen_at_unix_ms: i64,
22    pub current: bool,
23}
24
25#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
26pub struct SessionView {
27    pub id: String,
28    pub created_at_unix_ms: i64,
29    pub updated_at_unix_ms: i64,
30    #[serde(default)]
31    pub run_ids: Vec<String>,
32    #[serde(default)]
33    pub cwd: Option<String>,
34    #[serde(default)]
35    pub execution_profile: AgentSessionExecutionProfile,
36}
37
38/// Execution metadata inherited by native Sessions from the composed Host.
39///
40/// This has the same provider-neutral profile shape as connector-backed Agent
41/// Sessions. It describes where a new turn will execute; immutable per-Run
42/// provenance remains the responsibility of the Run journal.
43#[derive(Debug, Clone, Default, PartialEq, Eq)]
44pub struct NativeSessionDefaults {
45    pub cwd: Option<String>,
46    pub execution_profile: AgentSessionExecutionProfile,
47}
48
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub struct DevicePrincipal {
51    pub device_id: String,
52}
53
54#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
55pub struct PairingClaim {
56    pub token: String,
57    pub device: DeviceView,
58}
59
60#[derive(Debug, Clone)]
61pub struct PairingTicket {
62    secret: String,
63    secret_digest: String,
64    pub expires_at_unix_ms: i64,
65}
66
67impl PairingTicket {
68    pub fn issue(ttl_ms: i64) -> anyhow::Result<Self> {
69        if ttl_ms <= 0 {
70            bail!("pairing ticket TTL must be positive");
71        }
72        let secret = random_secret(32)?;
73        Ok(Self {
74            secret_digest: secret_digest(&secret),
75            secret,
76            expires_at_unix_ms: now_unix_ms().saturating_add(ttl_ms),
77        })
78    }
79
80    pub fn secret(&self) -> &str {
81        &self.secret
82    }
83}
84
85#[derive(Debug, Clone, Serialize, Deserialize)]
86struct DeviceRecord {
87    id: String,
88    name: String,
89    token_digest: String,
90    created_at_unix_ms: i64,
91    last_seen_at_unix_ms: i64,
92    #[serde(default)]
93    revoked_at_unix_ms: Option<i64>,
94}
95
96#[derive(Debug, Clone, Serialize, Deserialize)]
97struct RemoteStateFile {
98    version: u32,
99    owner_id: String,
100    #[serde(default)]
101    devices: Vec<DeviceRecord>,
102}
103
104impl RemoteStateFile {
105    fn fresh() -> Self {
106        Self {
107            version: STATE_VERSION,
108            owner_id: format!("owner-{}", uuid::Uuid::new_v4()),
109            devices: Vec::new(),
110        }
111    }
112
113    fn validate(&self) -> anyhow::Result<()> {
114        if self.version != STATE_VERSION {
115            bail!(
116                "unsupported remote-control state version {}; expected {STATE_VERSION}",
117                self.version
118            );
119        }
120        if self.owner_id.trim().is_empty() {
121            bail!("remote-control state owner is empty");
122        }
123        for device in &self.devices {
124            if device.id.trim().is_empty()
125                || device.name.trim().is_empty()
126                || device.token_digest.len() != 64
127            {
128                bail!("remote-control state contains an invalid device");
129            }
130        }
131        Ok(())
132    }
133}
134
135struct RegistryState {
136    durable: RemoteStateFile,
137    pairing: Option<PairingTicket>,
138}
139
140#[derive(Clone)]
141pub struct RemoteRegistry {
142    path: Option<PathBuf>,
143    state: Arc<Mutex<RegistryState>>,
144}
145
146impl RemoteRegistry {
147    pub fn in_memory(pairing: Option<PairingTicket>) -> Self {
148        Self {
149            path: None,
150            state: Arc::new(Mutex::new(RegistryState {
151                durable: RemoteStateFile::fresh(),
152                pairing,
153            })),
154        }
155    }
156
157    pub fn open(path: impl Into<PathBuf>, pairing: Option<PairingTicket>) -> anyhow::Result<Self> {
158        let path = path.into();
159        let existed = path.exists();
160        let durable = if existed {
161            let bytes = std::fs::read(&path)
162                .with_context(|| format!("read remote-control state '{}'", path.display()))?;
163            let state: RemoteStateFile = serde_json::from_slice(&bytes)
164                .with_context(|| format!("decode remote-control state '{}'", path.display()))?;
165            state.validate()?;
166            state
167        } else {
168            RemoteStateFile::fresh()
169        };
170        // Version 1 previously mixed Session/Run indexes into the device
171        // credential file. Unknown legacy fields are intentionally discarded
172        // and the sanitized authentication-only state is persisted here.
173        if existed {
174            persist_state(&path, &durable)?;
175        }
176        Ok(Self {
177            path: Some(path),
178            state: Arc::new(Mutex::new(RegistryState { durable, pairing })),
179        })
180    }
181
182    pub async fn claim_pairing(
183        &self,
184        secret: &str,
185        device_name: &str,
186    ) -> anyhow::Result<PairingClaim> {
187        let device_name = normalize_device_name(device_name)?;
188        let mut state = self.state.lock().await;
189        let ticket = state
190            .pairing
191            .as_ref()
192            .context("pairing ticket is unavailable or was already claimed")?;
193        if ticket.expires_at_unix_ms < now_unix_ms() {
194            bail!("pairing ticket has expired");
195        }
196        if !constant_time_eq(
197            ticket.secret_digest.as_bytes(),
198            secret_digest(secret).as_bytes(),
199        ) {
200            bail!("pairing secret is invalid");
201        }
202
203        let raw_token = random_secret(32)?;
204        let device_id = format!("device-{}", uuid::Uuid::new_v4());
205        let token = format!("{DEVICE_TOKEN_PREFIX}{device_id}.{raw_token}");
206        let timestamp = now_unix_ms();
207        let record = DeviceRecord {
208            id: device_id.clone(),
209            name: device_name,
210            token_digest: secret_digest(&token),
211            created_at_unix_ms: timestamp,
212            last_seen_at_unix_ms: timestamp,
213            revoked_at_unix_ms: None,
214        };
215        state.durable.devices.push(record.clone());
216        state.pairing = None;
217        self.persist_locked(&state.durable)?;
218        Ok(PairingClaim {
219            token,
220            device: device_view(&record, Some(&device_id)),
221        })
222    }
223
224    pub async fn authenticate(&self, token: &str) -> anyhow::Result<DevicePrincipal> {
225        let Some((device_id, _)) = parse_device_token(token) else {
226            bail!("device token is malformed");
227        };
228        let token_digest = secret_digest(token);
229        let mut state = self.state.lock().await;
230        let record = state
231            .durable
232            .devices
233            .iter_mut()
234            .find(|record| record.id == device_id && record.revoked_at_unix_ms.is_none())
235            .context("device token is unknown or revoked")?;
236        if !constant_time_eq(record.token_digest.as_bytes(), token_digest.as_bytes()) {
237            bail!("device token is invalid");
238        }
239        let timestamp = now_unix_ms();
240        if timestamp.saturating_sub(record.last_seen_at_unix_ms) >= 60_000 {
241            record.last_seen_at_unix_ms = timestamp;
242            self.persist_locked(&state.durable)?;
243        }
244        Ok(DevicePrincipal {
245            device_id: device_id.to_owned(),
246        })
247    }
248
249    pub async fn devices(&self, current_device_id: &str) -> Vec<DeviceView> {
250        let state = self.state.lock().await;
251        state
252            .durable
253            .devices
254            .iter()
255            .filter(|record| record.revoked_at_unix_ms.is_none())
256            .map(|record| device_view(record, Some(current_device_id)))
257            .collect()
258    }
259
260    pub async fn active_device_count(&self) -> usize {
261        self.state
262            .lock()
263            .await
264            .durable
265            .devices
266            .iter()
267            .filter(|record| record.revoked_at_unix_ms.is_none())
268            .count()
269    }
270
271    pub async fn revoke_device(&self, device_id: &str) -> anyhow::Result<()> {
272        let mut state = self.state.lock().await;
273        let record = state
274            .durable
275            .devices
276            .iter_mut()
277            .find(|record| record.id == device_id && record.revoked_at_unix_ms.is_none())
278            .context("device was not found")?;
279        record.revoked_at_unix_ms = Some(now_unix_ms());
280        self.persist_locked(&state.durable)
281    }
282
283    fn persist_locked(&self, state: &RemoteStateFile) -> anyhow::Result<()> {
284        let Some(path) = &self.path else {
285            return Ok(());
286        };
287        persist_state(path, state)
288    }
289}
290
291fn normalize_device_name(name: &str) -> anyhow::Result<String> {
292    let name = name.trim();
293    if name.is_empty() || name.chars().count() > 80 || name.chars().any(char::is_control) {
294        bail!("device name must contain 1 to 80 printable characters");
295    }
296    Ok(name.to_owned())
297}
298
299fn device_view(record: &DeviceRecord, current_device_id: Option<&str>) -> DeviceView {
300    DeviceView {
301        id: record.id.clone(),
302        name: record.name.clone(),
303        created_at_unix_ms: record.created_at_unix_ms,
304        last_seen_at_unix_ms: record.last_seen_at_unix_ms,
305        current: current_device_id.is_some_and(|current| current == record.id),
306    }
307}
308
309fn parse_device_token(token: &str) -> Option<(&str, &str)> {
310    let token = token.strip_prefix(DEVICE_TOKEN_PREFIX)?;
311    let (device_id, secret) = token.split_once('.')?;
312    if device_id.is_empty() || secret.len() < 32 || secret.contains('.') {
313        return None;
314    }
315    Some((device_id, secret))
316}
317
318fn random_secret(bytes: usize) -> anyhow::Result<String> {
319    let mut value = vec![0_u8; bytes];
320    getrandom::fill(&mut value)
321        .map_err(|error| anyhow::anyhow!("generate remote-control secret: {error}"))?;
322    Ok(URL_SAFE_NO_PAD.encode(value))
323}
324
325fn secret_digest(value: &str) -> String {
326    let mut hasher = Sha256::new();
327    hasher.update(value.as_bytes());
328    hex::encode(hasher.finalize())
329}
330
331fn constant_time_eq(left: &[u8], right: &[u8]) -> bool {
332    if left.len() != right.len() {
333        return false;
334    }
335    left.iter()
336        .zip(right)
337        .fold(0_u8, |difference, (left, right)| {
338            difference | (left ^ right)
339        })
340        == 0
341}
342
343fn now_unix_ms() -> i64 {
344    Utc::now().timestamp_millis()
345}
346
347fn persist_state(path: &Path, state: &RemoteStateFile) -> anyhow::Result<()> {
348    state.validate()?;
349    let parent = path
350        .parent()
351        .filter(|parent| !parent.as_os_str().is_empty())
352        .context("remote-control state path has no parent")?;
353    std::fs::create_dir_all(parent)
354        .with_context(|| format!("create remote-control directory '{}'", parent.display()))?;
355    set_private_directory_permissions(parent)?;
356    let temporary = parent.join(format!(".remote-control-{}.tmp", uuid::Uuid::new_v4()));
357    let bytes = serde_json::to_vec_pretty(state).context("encode remote-control state")?;
358    let write_result = (|| {
359        use std::io::Write;
360        let mut options = std::fs::OpenOptions::new();
361        options.create_new(true).write(true);
362        #[cfg(unix)]
363        {
364            use std::os::unix::fs::OpenOptionsExt;
365            options.mode(0o600);
366        }
367        let mut file = options.open(&temporary)?;
368        file.write_all(&bytes)?;
369        file.sync_all()?;
370        std::fs::rename(&temporary, path)?;
371        // Windows does not support opening a directory through File::open.
372        #[cfg(unix)]
373        std::fs::File::open(parent)?.sync_all()?;
374        Ok::<(), std::io::Error>(())
375    })();
376    if let Err(error) = write_result {
377        let _ = std::fs::remove_file(&temporary);
378        return Err(error)
379            .with_context(|| format!("persist remote-control state '{}'", path.display()));
380    }
381    Ok(())
382}
383
384fn set_private_directory_permissions(_path: &Path) -> anyhow::Result<()> {
385    #[cfg(unix)]
386    {
387        use std::os::unix::fs::PermissionsExt;
388        std::fs::set_permissions(_path, std::fs::Permissions::from_mode(0o700))
389            .with_context(|| format!("secure remote-control directory '{}'", _path.display()))?;
390    }
391    Ok(())
392}
393
394#[cfg(test)]
395mod tests {
396    use super::*;
397
398    #[tokio::test]
399    async fn pairing_is_one_time_and_revocation_is_immediate() {
400        let ticket = PairingTicket::issue(60_000).unwrap();
401        let secret = ticket.secret().to_owned();
402        let registry = RemoteRegistry::in_memory(Some(ticket));
403
404        let claim = registry
405            .claim_pairing(&secret, "Alice's phone")
406            .await
407            .unwrap();
408        let principal = registry.authenticate(&claim.token).await.unwrap();
409        assert_eq!(principal.device_id, claim.device.id);
410        assert!(registry
411            .claim_pairing(&secret, "second phone")
412            .await
413            .is_err());
414
415        registry.revoke_device(&principal.device_id).await.unwrap();
416        assert!(registry.authenticate(&claim.token).await.is_err());
417    }
418
419    #[tokio::test]
420    async fn persisted_remote_state_contains_only_authentication_data() {
421        let root = std::env::temp_dir().join(format!(
422            "orchestral-remote-auth-state-test-{}",
423            uuid::Uuid::new_v4()
424        ));
425        let path = root.join("state.json");
426        let ticket = PairingTicket::issue(60_000).unwrap();
427        let secret = ticket.secret().to_owned();
428        let registry = RemoteRegistry::open(&path, Some(ticket)).unwrap();
429        registry.claim_pairing(&secret, "Phone").await.unwrap();
430
431        let persisted: serde_json::Value =
432            serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
433        assert!(persisted.get("devices").is_some());
434        assert!(persisted.get("sessions").is_none());
435
436        std::fs::remove_dir_all(root).unwrap();
437    }
438
439    #[test]
440    fn legacy_session_index_is_discarded_when_auth_state_opens() {
441        let root = std::env::temp_dir().join(format!(
442            "orchestral-remote-legacy-state-test-{}",
443            uuid::Uuid::new_v4()
444        ));
445        std::fs::create_dir_all(&root).unwrap();
446        let path = root.join("state.json");
447        std::fs::write(
448            &path,
449            serde_json::json!({
450                "version": 1,
451                "owner_id": "owner-legacy",
452                "devices": [],
453                "sessions": [{
454                    "id": "dangling-session",
455                    "created_at_unix_ms": 1,
456                    "updated_at_unix_ms": 1,
457                    "run_ids": ["missing-run"]
458                }]
459            })
460            .to_string(),
461        )
462        .unwrap();
463
464        RemoteRegistry::open(&path, None).unwrap();
465        let sanitized: serde_json::Value =
466            serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
467        assert!(sanitized.get("sessions").is_none());
468
469        std::fs::remove_dir_all(root).unwrap();
470    }
471
472    #[tokio::test]
473    async fn persisted_tokens_are_hashed_and_reloadable() {
474        let root = std::env::temp_dir().join(format!(
475            "orchestral-remote-state-test-{}",
476            uuid::Uuid::new_v4()
477        ));
478        let path = root.join("state.json");
479        let ticket = PairingTicket::issue(60_000).unwrap();
480        let secret = ticket.secret().to_owned();
481        let registry = RemoteRegistry::open(&path, Some(ticket)).unwrap();
482        let claim = registry.claim_pairing(&secret, "Phone").await.unwrap();
483
484        let persisted = std::fs::read_to_string(&path).unwrap();
485        assert!(!persisted.contains(&claim.token));
486        let reloaded = RemoteRegistry::open(&path, None).unwrap();
487        assert!(reloaded.authenticate(&claim.token).await.is_ok());
488
489        std::fs::remove_dir_all(root).unwrap();
490    }
491}
orchestral_cli/remote/state.rs

orchestral_cli/remote/
state.rs
