hatchery_node/
m4a_attach.rs

1//! `POST /m4a/attach`: on-demand m4a account for a provider session the node did not spawn
2//! (spec live-session-m4a-attach).
3//!
4//! Caller: `m4a-agent mcp` running inside that provider session. Checks, all required:
5//! 1. owner: the ops listener's owner-only control token (a file only the node's OS user can
6//!    read) -- checked by the router before this module runs;
7//! 2. policy: `HATCHERY_M4A_ATTACH=auto` (default `off`: nothing is adopted) and the provider in
8//!    `HATCHERY_M4A_ATTACH_PROVIDERS` (default: all built-in providers);
9//! 3. process: the given pid is alive and descends (within a few generations) from a live
10//!    process of that provider.
11//! Then the node provisions a per-session account (own registry entry, own store root and key)
12//! keyed by the provider session id (or the provider's pid when none is known) and returns
13//! nick, session id and a fresh capability. The capability is returned once, never logged.
14
15use serde_json::{json, Value};
16
17pub const POLICY_ENV: &str = "HATCHERY_M4A_ATTACH";
18pub const PROVIDERS_ENV: &str = "HATCHERY_M4A_ATTACH_PROVIDERS";
19const MAX_GENERATIONS: usize = 12;
20
21#[derive(Debug, Clone, PartialEq, Eq)]
22pub struct Proc {
23    pub pid: u32,
24    pub ppid: u32,
25    pub name: String,
26}
27
28/// Process image base names that count as `provider`.
29pub fn provider_images(provider: &str) -> &'static [&'static str] {
30    match provider {
31        "grok" => &["grok"],
32        "codex" => &["codex"],
33        "kimi" => &["kimi", "kimi-code"],
34        "claude" => &["claude"],
35        _ => &[],
36    }
37}
38
39fn base_name(name: &str) -> String {
40    let n = name.rsplit(['/', '\\']).next().unwrap_or(name).to_ascii_lowercase();
41    n.strip_suffix(".exe").map(str::to_owned).unwrap_or(n)
42}
43
44/// Walks from `pid` up the parent chain; returns the nearest process whose image is the provider.
45pub fn provider_ancestor(table: &[Proc], pid: u32, provider: &str) -> Option<u32> {
46    let images = provider_images(provider);
47    let mut current = pid;
48    for _ in 0..MAX_GENERATIONS {
49        let p = table.iter().find(|p| p.pid == current)?;
50        let name = base_name(&p.name);
51        if images.iter().any(|i| name == *i || name.starts_with(&format!("{i}-"))) {
52            return Some(p.pid);
53        }
54        if p.ppid == 0 || p.ppid == p.pid {
55            return None;
56        }
57        current = p.ppid;
58    }
59    None
60}
61
62pub fn policy_allows(provider: &str) -> Result<(), &'static str> {
63    let mode = std::env::var(POLICY_ENV).unwrap_or_default().trim().to_ascii_lowercase();
64    if mode != "auto" {
65        return Err("adoption policy is off on this node");
66    }
67    if let Ok(list) = std::env::var(PROVIDERS_ENV) {
68        if !list.split(',').map(str::trim).any(|p| p == provider) {
69            return Err("provider is not allowed by the adoption policy");
70        }
71    }
72    Ok(())
73}
74
75pub fn query_get(query: &str, key: &str) -> Option<String> {
76    query.split('&').find_map(|kv| {
77        let (k, v) = kv.split_once('=')?;
78        (k == key).then(|| percent_decode(v))
79    })
80}
81
82fn percent_decode(v: &str) -> String {
83    let b = v.as_bytes();
84    let mut out = Vec::with_capacity(b.len());
85    let mut i = 0;
86    while i < b.len() {
87        if b[i] == b'%' && i + 2 < b.len() {
88            if let Ok(x) = u8::from_str_radix(std::str::from_utf8(&b[i + 1..i + 3]).unwrap_or(""), 16) {
89                out.push(x);
90                i += 3;
91                continue;
92            }
93        }
94        out.push(if b[i] == b'+' { b' ' } else { b[i] });
95        i += 1;
96    }
97    String::from_utf8_lossy(&out).into_owned()
98}
99
100pub fn valid_session(s: &str) -> bool {
101    !s.is_empty() && s.len() <= 128 && s.bytes().all(|b| b.is_ascii_alphanumeric() || b"-_.:".contains(&b))
102}
103
104/// Handles the request; `table` is the current process table (injected for tests).
105pub fn handle(
106    query: &str,
107    accounts: Option<&crate::m4a_accounts::M4aSessionAccounts>,
108    table: &[Proc],
109) -> (u16, Value) {
110    let Some(provider) = query_get(query, "provider") else {
111        return (400, json!({"error": "provider is required"}));
112    };
113    let Some(m4a_provider) = crate::m4a_accounts::m4a_provider(&provider) else {
114        return (400, json!({"error": "unknown provider"}));
115    };
116    if let Err(reason) = policy_allows(&provider) {
117        tracing::info!(%provider, "m4a attach refused by policy");
118        return (403, json!({"error": reason, "state": "pending"}));
119    }
120    let Some(pid) = query_get(query, "pid").and_then(|p| p.parse::<u32>().ok()) else {
121        return (400, json!({"error": "pid is required"}));
122    };
123    let Some(root) = provider_ancestor(table, pid, &provider) else {
124        tracing::warn!(%provider, pid, "m4a attach refused: caller is not inside a live provider process");
125        return (403, json!({"error": "caller is not inside a live process of that provider"}));
126    };
127    let session = query_get(query, "session");
128    if session.as_deref().is_some_and(|s| !valid_session(s)) {
129        return (400, json!({"error": "invalid session id"}));
130    }
131    let Some(accounts) = accounts else {
132        return (503, json!({"error": "m4a accounts are not configured on this node"}));
133    };
134    // Per-session binding: an MCP child of a shared backend has no session of its own, so the
135    // node reads it from the provider's command line or takes a pending adoption.
136    // An adopted provider binds only to its adopted sessions (saved in the accounts root, so
137    // this holds across node restarts); it never gets a fresh per-pid account.
138    let adopted: Vec<String> = crate::m4a_wake::load_adoptions(accounts.base()).into_iter().filter(|a| a.provider == provider).map(|a| a.session).collect();
139    let key = if adopted.is_empty() {
140        let session = crate::m4a_adopt::resolve_session(session, &provider, root, crate::m4a_adopt::process_cmdline).filter(|s| valid_session(s));
141        match &session {
142            Some(s) => format!("attach-{provider}-{s}"),
143            None => format!("attach-{provider}-pid{root}"),
144        }
145    } else {
146        let from_cmdline = crate::m4a_adopt::session_from_cmdline(&crate::m4a_adopt::process_cmdline(root));
147        match pick_adopted(session, from_cmdline, &adopted) {
148            Ok(s) => format!("attach-{provider}-{s}"),
149            Err(why) => {
150                tracing::warn!(%provider, adopted = adopted.len(), "m4a attach refused: {why}");
151                return (409, json!({"error": why}));
152            }
153        }
154    };
155    match accounts.provision(m4a_provider, &key) {
156        Ok(account) => {
157            tracing::info!(%provider, provider_pid = root, nick = %account.nick, "m4a account attached to a live session");
158            (200, account.attach_reply())
159        }
160        Err(error) => {
161            tracing::warn!(%provider, %error, "m4a attach provision failed");
162            (500, json!({"error": "account provision failed"}))
163        }
164    }
165}
166
167/// The adopted session an attach belongs to: the session it names, else the one on the
168/// provider's command line when that one is adopted, else the only adopted session. Several
169/// adopted sessions and nothing to tell them apart is an error, never a new account.
170pub fn pick_adopted(explicit: Option<String>, from_cmdline: Option<String>, adopted: &[String]) -> Result<String, &'static str> {
171    if let Some(s) = explicit {
172        return Ok(s);
173    }
174    if let Some(s) = from_cmdline.filter(|s| adopted.contains(s)) {
175        return Ok(s);
176    }
177    match adopted {
178        [only] => Ok(only.clone()),
179        [] => Err("no adopted session"),
180        _ => Err("several adopted sessions of this provider and this MCP names none; start it with --provider-session"),
181    }
182}
183
184/// The live process table.
185#[cfg(target_os = "linux")]
186pub fn process_table() -> Vec<Proc> {
187    let mut out = Vec::new();
188    let Ok(dir) = std::fs::read_dir("/proc") else { return out };
189    for e in dir.flatten() {
190        let Some(pid) = e.file_name().to_str().and_then(|s| s.parse::<u32>().ok()) else { continue };
191        let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else { continue };
192        let (Some(l), Some(r)) = (stat.find('('), stat.rfind(')')) else { continue };
193        let ppid = stat[r + 1..].split_whitespace().nth(1).and_then(|s| s.parse().ok()).unwrap_or(0);
194        // comm is truncated to 15 bytes; prefer the executable name.
195        let name = std::fs::read_link(format!("/proc/{pid}/exe"))
196            .ok()
197            .and_then(|p| p.file_name().map(|n| n.to_string_lossy().into_owned()))
198            .unwrap_or_else(|| stat[l + 1..r].to_owned());
199        out.push(Proc { pid, ppid, name });
200    }
201    out
202}
203
204#[cfg(windows)]
205pub fn process_table() -> Vec<Proc> {
206    use windows_sys::Win32::Foundation::{CloseHandle, INVALID_HANDLE_VALUE};
207    use windows_sys::Win32::System::Diagnostics::ToolHelp::{
208        CreateToolhelp32Snapshot, Process32FirstW, Process32NextW, PROCESSENTRY32W, TH32CS_SNAPPROCESS,
209    };
210    let mut out = Vec::new();
211    unsafe {
212        let snap = CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0);
213        if snap == INVALID_HANDLE_VALUE {
214            return out;
215        }
216        let mut entry: PROCESSENTRY32W = std::mem::zeroed();
217        entry.dwSize = std::mem::size_of::<PROCESSENTRY32W>() as u32;
218        let mut ok = Process32FirstW(snap, &mut entry) != 0;
219        while ok {
220            let len = entry.szExeFile.iter().position(|c| *c == 0).unwrap_or(entry.szExeFile.len());
221            out.push(Proc {
222                pid: entry.th32ProcessID,
223                ppid: entry.th32ParentProcessID,
224                name: String::from_utf16_lossy(&entry.szExeFile[..len]),
225            });
226            ok = Process32NextW(snap, &mut entry) != 0;
227        }
228        CloseHandle(snap);
229    }
230    out
231}
232
233#[cfg(not(any(target_os = "linux", windows)))]
234pub fn process_table() -> Vec<Proc> {
235    Vec::new()
236}
237
238#[cfg(test)]
239mod tests {
240    use super::*;
241
242    fn table() -> Vec<Proc> {
243        vec![
244            Proc { pid: 1, ppid: 0, name: "init".into() },
245            Proc { pid: 10, ppid: 1, name: "Grok.EXE".into() },
246            Proc { pid: 11, ppid: 10, name: "cmd.exe".into() },
247            Proc { pid: 12, ppid: 11, name: "m4a-agent.exe".into() },
248            Proc { pid: 20, ppid: 1, name: "bash".into() },
249            Proc { pid: 21, ppid: 20, name: "m4a-agent".into() },
250        ]
251    }
252
253    #[test]
254    fn attach_without_session_takes_the_adopted_account_across_restarts() {
255        let dir = std::env::temp_dir().join(format!("m4a-attach-adopt-{}", std::process::id()));
256        let _ = std::fs::remove_dir_all(&dir);
257        std::fs::create_dir_all(&dir).unwrap();
258        let a = crate::m4a_wake::Adoption { provider: "grok".into(), session: "s-1".into(), cwd: Some("w".into()), nick: "grok-1".into(), account_ref: "r".into(), store_root: None };
259        crate::m4a_wake::save_adoption(&dir, &a).unwrap();
260        // "Restart": nothing in memory, only the saved file.
261        let adopted = |d: &std::path::Path| crate::m4a_wake::load_adoptions(d).into_iter().map(|a| a.session).collect::<Vec<_>>();
262        assert_eq!(pick_adopted(None, None, &adopted(&dir)).as_deref(), Ok("s-1"));
263        assert_eq!(pick_adopted(None, Some("not-adopted".into()), &adopted(&dir)).as_deref(), Ok("s-1"));
264        crate::m4a_wake::save_adoption(&dir, &crate::m4a_wake::Adoption { session: "s-2".into(), ..a.clone() }).unwrap();
265        assert!(pick_adopted(None, None, &adopted(&dir)).is_err(), "ambiguous: error, not a pid account");
266        assert_eq!(pick_adopted(None, Some("s-2".into()), &adopted(&dir)).as_deref(), Ok("s-2"));
267        assert_eq!(pick_adopted(Some("s-1".into()), None, &adopted(&dir)).as_deref(), Ok("s-1"));
268        std::fs::remove_dir_all(&dir).unwrap();
269    }
270
271    #[test]
272    fn ancestry_finds_the_provider_and_refuses_strangers() {
273        let t = table();
274        assert_eq!(provider_ancestor(&t, 12, "grok"), Some(10));
275        assert_eq!(provider_ancestor(&t, 12, "codex"), None);
276        assert_eq!(provider_ancestor(&t, 21, "grok"), None);
277        assert_eq!(provider_ancestor(&t, 999, "grok"), None);
278    }
279
280    #[test]
281    fn query_parsing_decodes_and_validates() {
282        assert_eq!(query_get("provider=grok&session=a%3Ab", "session").as_deref(), Some("a:b"));
283        assert!(valid_session("0aa-b_c.d:e"));
284        assert!(!valid_session("../x"));
285        assert!(!valid_session(""));
286    }
287
288    // One test owns the policy env var (tests in this binary run in parallel).
289    #[test]
290    fn policy_and_process_gates_come_before_any_account() {
291        std::env::remove_var(POLICY_ENV);
292        let (s, v) = handle("provider=grok&pid=12", None, &table());
293        assert_eq!((s, v["state"].as_str()), (403, Some("pending")));
294        std::env::set_var(POLICY_ENV, "auto");
295        std::env::set_var(PROVIDERS_ENV, "codex");
296        assert_eq!(handle("provider=grok&pid=12", None, &table()).0, 403);
297        std::env::remove_var(PROVIDERS_ENV);
298        assert_eq!(handle("provider=grok&pid=21", None, &table()).0, 403);
299        assert_eq!(handle("provider=grok&pid=12&session=../x", None, &table()).0, 400);
300        assert_eq!(handle("provider=grok&pid=12", None, &table()).0, 503);
301        assert_eq!(handle("provider=nope&pid=12", None, &table()).0, 400);
302        std::env::remove_var(POLICY_ENV);
303    }
304
305    #[cfg(target_os = "linux")]
306    #[test]
307    fn live_table_contains_this_process() {
308        let me = std::process::id();
309        assert!(process_table().iter().any(|p| p.pid == me));
310    }
311}
hatchery_node/m4a_attach.rs

hatchery_node/
m4a_attach.rs
