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
75fn 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
100fn 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    let key = match &session {
135        Some(s) => format!("attach-{provider}-{s}"),
136        None => format!("attach-{provider}-pid{root}"),
137    };
138    match accounts.provision(m4a_provider, &key) {
139        Ok(account) => {
140            tracing::info!(%provider, provider_pid = root, nick = %account.nick, "m4a account attached to a live session");
141            (200, account.attach_reply())
142        }
143        Err(error) => {
144            tracing::warn!(%provider, %error, "m4a attach provision failed");
145            (500, json!({"error": "account provision failed"}))
146        }
147    }
148}
149
150/// The live process table.
151#[cfg(target_os = "linux")]
152pub fn process_table() -> Vec<Proc> {
153    let mut out = Vec::new();
154    let Ok(dir) = std::fs::read_dir("/proc") else { return out };
155    for e in dir.flatten() {
156        let Some(pid) = e.file_name().to_str().and_then(|s| s.parse::<u32>().ok()) else { continue };
157        let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else { continue };
158        let (Some(l), Some(r)) = (stat.find('('), stat.rfind(')')) else { continue };
159        let ppid = stat[r + 1..].split_whitespace().nth(1).and_then(|s| s.parse().ok()).unwrap_or(0);
160        // comm is truncated to 15 bytes; prefer the executable name.
161        let name = std::fs::read_link(format!("/proc/{pid}/exe"))
162            .ok()
163            .and_then(|p| p.file_name().map(|n| n.to_string_lossy().into_owned()))
164            .unwrap_or_else(|| stat[l + 1..r].to_owned());
165        out.push(Proc { pid, ppid, name });
166    }
167    out
168}
169
170#[cfg(windows)]
171pub fn process_table() -> Vec<Proc> {
172    use windows_sys::Win32::Foundation::{CloseHandle, INVALID_HANDLE_VALUE};
173    use windows_sys::Win32::System::Diagnostics::ToolHelp::{
174        CreateToolhelp32Snapshot, Process32FirstW, Process32NextW, PROCESSENTRY32W, TH32CS_SNAPPROCESS,
175    };
176    let mut out = Vec::new();
177    unsafe {
178        let snap = CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0);
179        if snap == INVALID_HANDLE_VALUE {
180            return out;
181        }
182        let mut entry: PROCESSENTRY32W = std::mem::zeroed();
183        entry.dwSize = std::mem::size_of::<PROCESSENTRY32W>() as u32;
184        let mut ok = Process32FirstW(snap, &mut entry) != 0;
185        while ok {
186            let len = entry.szExeFile.iter().position(|c| *c == 0).unwrap_or(entry.szExeFile.len());
187            out.push(Proc {
188                pid: entry.th32ProcessID,
189                ppid: entry.th32ParentProcessID,
190                name: String::from_utf16_lossy(&entry.szExeFile[..len]),
191            });
192            ok = Process32NextW(snap, &mut entry) != 0;
193        }
194        CloseHandle(snap);
195    }
196    out
197}
198
199#[cfg(not(any(target_os = "linux", windows)))]
200pub fn process_table() -> Vec<Proc> {
201    Vec::new()
202}
203
204#[cfg(test)]
205mod tests {
206    use super::*;
207
208    fn table() -> Vec<Proc> {
209        vec![
210            Proc { pid: 1, ppid: 0, name: "init".into() },
211            Proc { pid: 10, ppid: 1, name: "Grok.EXE".into() },
212            Proc { pid: 11, ppid: 10, name: "cmd.exe".into() },
213            Proc { pid: 12, ppid: 11, name: "m4a-agent.exe".into() },
214            Proc { pid: 20, ppid: 1, name: "bash".into() },
215            Proc { pid: 21, ppid: 20, name: "m4a-agent".into() },
216        ]
217    }
218
219    #[test]
220    fn ancestry_finds_the_provider_and_refuses_strangers() {
221        let t = table();
222        assert_eq!(provider_ancestor(&t, 12, "grok"), Some(10));
223        assert_eq!(provider_ancestor(&t, 12, "codex"), None);
224        assert_eq!(provider_ancestor(&t, 21, "grok"), None);
225        assert_eq!(provider_ancestor(&t, 999, "grok"), None);
226    }
227
228    #[test]
229    fn query_parsing_decodes_and_validates() {
230        assert_eq!(query_get("provider=grok&session=a%3Ab", "session").as_deref(), Some("a:b"));
231        assert!(valid_session("0aa-b_c.d:e"));
232        assert!(!valid_session("../x"));
233        assert!(!valid_session(""));
234    }
235
236    // One test owns the policy env var (tests in this binary run in parallel).
237    #[test]
238    fn policy_and_process_gates_come_before_any_account() {
239        std::env::remove_var(POLICY_ENV);
240        let (s, v) = handle("provider=grok&pid=12", None, &table());
241        assert_eq!((s, v["state"].as_str()), (403, Some("pending")));
242        std::env::set_var(POLICY_ENV, "auto");
243        std::env::set_var(PROVIDERS_ENV, "codex");
244        assert_eq!(handle("provider=grok&pid=12", None, &table()).0, 403);
245        std::env::remove_var(PROVIDERS_ENV);
246        assert_eq!(handle("provider=grok&pid=21", None, &table()).0, 403);
247        assert_eq!(handle("provider=grok&pid=12&session=../x", None, &table()).0, 400);
248        assert_eq!(handle("provider=grok&pid=12", None, &table()).0, 503);
249        assert_eq!(handle("provider=nope&pid=12", None, &table()).0, 400);
250        std::env::remove_var(POLICY_ENV);
251    }
252
253    #[cfg(target_os = "linux")]
254    #[test]
255    fn live_table_contains_this_process() {
256        let me = std::process::id();
257        assert!(process_table().iter().any(|p| p.pid == me));
258    }
259}
hatchery_node/m4a_attach.rs

hatchery_node/
m4a_attach.rs
