netget/client/docker/
mod.rs

1pub mod actions;
2pub mod schema;
3use crate::client::{command_support, llm_budget::call_llm_for_client};
4use crate::logging::emit::Log;
5use crate::protocol::{ConnectContext, Event};
6use crate::state::{
7    client_handles::{ClientCommand, ClientSendOutcome},
8    AccessLogOwner, ClientStatus,
9};
10pub use actions::DockerClientProtocol;
11use anyhow::{ensure, Context, Result};
12use serde_json::{json, Value};
13use std::{net::SocketAddr, time::Duration};
14use tokio::sync::mpsc;
15pub const DEFAULT_API_VERSION: &str = "1.47";
16pub const DEFAULT_TIMEOUT_SECS: u64 = 15;
17pub const MAX_TIMEOUT_SECS: u64 = 30;
18pub const MAX_FOLLOWUPS: u8 = 4;
19pub const QUEUE_CAPACITY: usize = 8;
20pub fn api_version(v: &str) -> Result<(u16, u16)> {
21    let (major, minor) = v
22        .split_once('.')
23        .context("API version must be major.minor")?;
24    ensure!(
25        !major.is_empty()
26            && !minor.is_empty()
27            && major.len() <= 3
28            && minor.len() <= 3
29            && major
30                .bytes()
31                .chain(minor.bytes())
32                .all(|c| c.is_ascii_digit()),
33        "invalid API version"
34    );
35    Ok((major.parse()?, minor.parse()?))
36}
37pub struct Request {
38    pub operation: String,
39    pub path: String,
40}
41pub fn request(action: &Value, version: &str) -> Result<Request> {
42    api_version(version)?;
43    ensure!(
44        action["type"] == "docker_request",
45        "expected docker_request action"
46    );
47    let operation = action["operation"]
48        .as_str()
49        .context("operation must be string")?;
50    let (route, allowed) = match operation {
51        "ping" => ("/_ping".to_owned(), vec![]),
52        "version" => ("/version".into(), vec![]),
53        "info" => ("/info".into(), vec![]),
54        "containers" => (
55            "/containers/json".into(),
56            vec!["all", "limit", "size", "filters"],
57        ),
58        "container" => {
59            let id = action["container_id"]
60                .as_str()
61                .context("container_id required")?;
62            ensure!(
63                id.as_bytes().first().is_some_and(u8::is_ascii_alphanumeric)
64                    && id.len() <= 256
65                    && id
66                        .bytes()
67                        .all(|c| c.is_ascii_alphanumeric() || b"_-.".contains(&c)),
68                "container_id must be a bounded ID or name"
69            );
70            (
71                format!("/containers/{id}/json"),
72                vec!["container_id", "size"],
73            )
74        }
75        "images" => ("/images/json".into(), vec!["all", "digests", "filters"]),
76        "networks" => ("/networks".into(), vec!["filters"]),
77        "volumes" => ("/volumes".into(), vec!["filters"]),
78        _ => anyhow::bail!("unknown read-only Docker operation"),
79    };
80    let object = action.as_object().context("action must be object")?;
81    for field in object.keys() {
82        ensure!(
83            ["type", "operation"].contains(&field.as_str()) || allowed.contains(&field.as_str()),
84            "field {field} is not allowed for {operation}"
85        );
86    }
87    let mut query = Vec::new();
88    for field in ["all", "size", "digests"] {
89        if let Some(v) = action.get(field) {
90            let value = v
91                .as_bool()
92                .with_context(|| format!("{field} must be boolean"))?;
93            query.push(format!("{field}={value}"));
94        }
95    }
96    if let Some(v) = action.get("limit") {
97        let limit = v.as_u64().context("limit must be integer")?;
98        ensure!((1..=1000).contains(&limit), "limit must be 1..1000");
99        query.push(format!("limit={limit}"));
100    }
101    if let Some(v) = action.get("filters") {
102        let filters = v
103            .as_object()
104            .context("filters must be object of string arrays")?;
105        ensure!(filters.len() <= 64, "filter count limit");
106        for (key, values) in filters {
107            ensure!(
108                !key.is_empty()
109                    && key.len() <= 64
110                    && key
111                        .bytes()
112                        .all(|c| c.is_ascii_alphanumeric() || b"_-".contains(&c)),
113                "invalid filter name"
114            );
115            let values = values
116                .as_array()
117                .context("filter values must be string arrays")?;
118            ensure!(values.len() <= 64, "filter value count limit");
119            for v in values {
120                let text = v.as_str().context("filter value must be string")?;
121                ensure!(
122                    text.len() <= 1024 && !crate::utils::sanitize::has_controls(&text),
123                    "filter value limit/control character"
124                );
125            }
126        }
127        let encoded = serde_json::to_string(v)?;
128        ensure!(encoded.len() <= 8192, "encoded filter limit");
129        query.push(format!("filters={}", urlencoding::encode(&encoded)));
130    }
131    let mut path = if operation == "ping" {
132        route
133    } else {
134        format!("/v{version}{route}")
135    };
136    if !query.is_empty() {
137        path.push('?');
138        path.push_str(&query.join("&"));
139    }
140    Ok(Request {
141        operation: operation.into(),
142        path,
143    })
144}
145#[derive(Clone)]
146enum Endpoint {
147    Tcp {
148        origin: String,
149        host: String,
150        port: u16,
151        authority: String,
152    },
153    #[cfg(all(unix, not(target_arch = "wasm32")))]
154    Unix(String),
155}
156impl Endpoint {
157    fn parse(address: &str) -> Result<Self> {
158        if let Some(path) = address.strip_prefix("unix://") {
159            ensure!(
160                path.starts_with('/')
161                    && path.len() <= 4096
162                    && !crate::utils::sanitize::has_controls(&path)
163                    && !path.contains(['?', '#']),
164                "Unix endpoint must be an absolute socket path"
165            );
166            #[cfg(all(unix, not(target_arch = "wasm32")))]
167            return Ok(Self::Unix(path.into()));
168            #[cfg(not(all(unix, not(target_arch = "wasm32"))))]
169            anyhow::bail!("Unix Docker transport is unavailable on this platform");
170        }
171        let address = if address.contains("://") {
172            address.into()
173        } else {
174            format!("http://{address}")
175        };
176        let url = url::Url::parse(&address).context("invalid Docker endpoint")?;
177        ensure!(
178            url.scheme() == "http" && url.host_str().is_some(),
179            "Docker transport requires HTTP TCP or native unix://; HTTPS/TLS is not implemented"
180        );
181        ensure!(
182            url.username().is_empty()
183                && url.password().is_none()
184                && url.path() == "/"
185                && url.query().is_none()
186                && url.fragment().is_none(),
187            "Docker endpoint must be an origin without credentials/path/query"
188        );
189        let host = url.host_str().unwrap().trim_matches(['[', ']']).to_owned();
190        let port = url.port_or_known_default().unwrap();
191        let authority = url[url::Position::BeforeHost..url::Position::AfterPort].to_owned();
192        Ok(Self::Tcp {
193            origin: url.origin().ascii_serialization(),
194            host,
195            port,
196            authority,
197        })
198    }
199    fn name(&self) -> String {
200        match self {
201            Self::Tcp { origin, .. } => origin.clone(),
202            #[cfg(all(unix, not(target_arch = "wasm32")))]
203            Self::Unix(path) => format!("unix://{path}"),
204        }
205    }
206}
207trait Socket: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin {}
208impl<T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin> Socket for T {}
209async fn fetch(
210    endpoint: &Endpoint,
211    method: &str,
212    path: &str,
213) -> Result<(u16, hyper::HeaderMap, Vec<u8>)> {
214    use http_body_util::{BodyExt, Full, Limited};
215    let (stream, authority): (Box<dyn Socket>, &str) = match endpoint {
216        Endpoint::Tcp {
217            host,
218            port,
219            authority,
220            ..
221        } => (
222            Box::new(
223                tokio::net::TcpStream::connect((host.as_str(), *port))
224                    .await
225                    .context("connect Docker TCP")?,
226            ),
227            authority,
228        ),
229        #[cfg(all(unix, not(target_arch = "wasm32")))]
230        Endpoint::Unix(path) => (
231            Box::new(
232                tokio::net::UnixStream::connect(path)
233                    .await
234                    .context("connect Docker Unix socket")?,
235            ),
236            "localhost",
237        ),
238    };
239    let (mut sender, connection) =
240        hyper::client::conn::http1::handshake(hyper_util::rt::TokioIo::new(stream)).await?;
241    let exchange = async {
242        let response = sender
243            .send_request(
244                hyper::Request::builder()
245                    .method(method)
246                    .uri(path)
247                    .header("Host", authority)
248                    .header("Accept", "application/json")
249                    .header("Accept-Encoding", "identity")
250                    .header("User-Agent", "NetGet-Docker/1.0")
251                    .header("Connection", "close")
252                    .body(Full::new(bytes::Bytes::new()))?,
253            )
254            .await?;
255        let (parts, body) = response.into_parts();
256        if let Some(length) = parts.headers.get("content-length") {
257            ensure!(
258                length.to_str()?.parse::<u64>()? <= schema::MAX_BODY as u64,
259                "Docker body limit"
260            );
261        }
262        let body = Limited::new(body, schema::MAX_BODY)
263            .collect()
264            .await
265            .map_err(|e| anyhow::anyhow!("read bounded Docker body: {e}"))?
266            .to_bytes()
267            .to_vec();
268        Ok((parts.status.as_u16(), parts.headers, body))
269    };
270    tokio::pin!(exchange, connection);
271    // The connection driver belongs to this exchange; cancellation drops both futures/socket.
272    tokio::select! {biased;result=&mut exchange=>result,result=&mut connection=>{
273        // A complete framed body can precede a Unix proxy shutdown error. Consume
274        // the buffered exchange first; partial/malformed bodies still fail closed.
275        match exchange.await {Ok(response)=>Ok(response),Err(error)=>{result.context("Docker connection driver")?;Err(error)}}
276    }}
277}
278fn header<'a>(headers: &'a hyper::HeaderMap, name: &str) -> Result<&'a str> {
279    ensure!(
280        headers.get_all(name).iter().count() == 1,
281        "expected exactly one {name} header"
282    );
283    Ok(headers.get(name).unwrap().to_str()?)
284}
285fn ping(status: u16, headers: &hyper::HeaderMap, body: &[u8], head: bool) -> Result<Value> {
286    ensure!(status == 200, "Docker ping returned HTTP {status}");
287    let api = header(headers, "api-version")?;
288    api_version(api)?;
289    ensure!(head || body == b"OK", "invalid Docker ping body");
290    let os = headers.get("ostype").map(|v| v.to_str()).transpose()?;
291    let experimental = headers
292        .get("docker-experimental")
293        .map(|v| match v.to_str()? {
294            "true" => Ok(true),
295            "false" => Ok(false),
296            _ => anyhow::bail!("invalid Docker-Experimental header"),
297        })
298        .transpose()?;
299    Ok(json!({"api_version":api,"os_type":os,"experimental":experimental}))
300}
301fn response(
302    request: &Request,
303    status: u16,
304    headers: &hyper::HeaderMap,
305    body: &[u8],
306) -> Result<Value> {
307    if request.operation == "ping" {
308        return ping(status, headers, body, false);
309    }
310    ensure!(status == 200, "Docker returned HTTP {status}");
311    let mime = header(headers, "content-type")?
312        .split(';')
313        .next()
314        .unwrap()
315        .trim();
316    ensure!(
317        mime.eq_ignore_ascii_case("application/json"),
318        "expected Docker application/json response"
319    );
320    for encoding in headers.get_all("content-encoding") {
321        ensure!(
322            encoding.to_str()?.eq_ignore_ascii_case("identity"),
323            "unsupported Docker Content-Encoding"
324        );
325    }
326    schema::parse(&request.operation, body)
327}
328type HandlerEvent = (Event, u8);
329type HandlerAction = (Value, u8);
330pub async fn connect(ctx: ConnectContext) -> Result<SocketAddr> {
331    let endpoint = Endpoint::parse(&ctx.remote_addr)?;
332    let mut preferred = DEFAULT_API_VERSION.to_owned();
333    let mut timeout_secs = DEFAULT_TIMEOUT_SECS;
334    if let Some(params) = &ctx.startup_params {
335        preferred = params
336            .get_optional_string("api_version")?
337            .unwrap_or_else(|| DEFAULT_API_VERSION.into());
338        timeout_secs = params
339            .get_optional_u64("request_timeout_secs")?
340            .unwrap_or(DEFAULT_TIMEOUT_SECS);
341    }
342    let preferred_version = api_version(&preferred)?;
343    ensure!(
344        (api_version("1.24")?..=api_version(DEFAULT_API_VERSION)?).contains(&preferred_version),
345        "api_version must be 1.24..1.47"
346    );
347    ensure!(
348        (1..=MAX_TIMEOUT_SECS).contains(&timeout_secs),
349        "request_timeout_secs must be 1..30"
350    );
351    let timeout = Duration::from_secs(timeout_secs);
352    let (status, headers, body) = tokio::time::timeout(timeout, fetch(&endpoint, "HEAD", "/_ping"))
353        .await
354        .context("Docker negotiation deadline")??;
355    let negotiated = ping(status, &headers, &body, true)?;
356    let maximum = api_version(negotiated["api_version"].as_str().unwrap())?;
357    ensure!(
358        maximum >= api_version("1.24")?,
359        "Docker daemon API is below 1.24"
360    );
361    let selected = maximum.min(preferred_version);
362    let version = format!("{}.{}", selected.0, selected.1);
363    let remote = match &endpoint {
364        Endpoint::Tcp { host, port, .. } => {
365            tokio::time::timeout(timeout, tokio::net::lookup_host((host.as_str(), *port)))
366                .await
367                .context("Docker resolution deadline")??
368                .next()
369                .context("Docker resolved no addresses")?
370        }
371        #[cfg(all(unix, not(target_arch = "wasm32")))]
372        Endpoint::Unix(_) => "0.0.0.0:0".parse()?,
373    };
374    let external = command_support::register_command_channel(&ctx.state, ctx.client_id).await;
375    let (actions_tx, actions_rx) = mpsc::channel::<HandlerAction>(QUEUE_CAPACITY);
376    let (events_tx, mut events_rx) = mpsc::channel::<HandlerEvent>(QUEUE_CAPACITY);
377    let handler_ctx = ctx.clone();
378    let handler = tokio::spawn(async move {
379        while let Some((event, depth)) = events_rx.recv().await {
380            let instruction = handler_ctx
381                .state
382                .get_instruction_for_client(handler_ctx.client_id)
383                .await
384                .unwrap_or_default();
385            let memory = handler_ctx
386                .state
387                .get_memory_for_client(handler_ctx.client_id)
388                .await
389                .unwrap_or_default();
390            match call_llm_for_client(
391                &handler_ctx.llm_client,
392                &handler_ctx.state,
393                handler_ctx.client_id.to_string(),
394                &instruction,
395                &memory,
396                Some(&event),
397                &DockerClientProtocol,
398                &handler_ctx.status_tx,
399            )
400            .await
401            {
402                Ok(result) => {
403                    if let Some(memory) = result.memory_updates {
404                        handler_ctx
405                            .state
406                            .set_memory_for_client(handler_ctx.client_id, memory)
407                            .await;
408                    }
409                    for action in result.actions {
410                        if depth >= MAX_FOLLOWUPS && action["type"] != "disconnect" {
411                            Log::new(Some(&handler_ctx.status_tx))
412                                .warn("Docker handler followup limit reached");
413                            continue;
414                        }
415                        if actions_tx.send((action, depth + 1)).await.is_err() {
416                            return;
417                        }
418                    }
419                }
420                Err(e) => {
421                    Log::new(Some(&handler_ctx.status_tx)).warn(format!("Docker handler: {e}"))
422                }
423            }
424        }
425    });
426    let handler_abort = handler.abort_handle();
427    ctx.state.register_client_task(ctx.client_id, handler).await;
428    let session_ctx = ctx.clone();
429    let task = tokio::spawn(async move {
430        let result = session(
431            &session_ctx,
432            &endpoint,
433            &version,
434            negotiated,
435            timeout,
436            external,
437            actions_rx,
438            events_tx,
439        )
440        .await;
441        handler_abort.abort();
442        let status = match result {
443            Ok(()) => ClientStatus::Disconnected,
444            Err(e) => {
445                Log::new(Some(&session_ctx.status_tx)).warn(format!("Docker ended: {e}"));
446                ClientStatus::Error(e.to_string())
447            }
448        };
449        session_ctx
450            .state
451            .update_client_status(session_ctx.client_id, status)
452            .await;
453        session_ctx
454            .state
455            .remove_client_handle(session_ctx.client_id)
456            .await;
457        let _ = session_ctx.status_tx.send("__UPDATE_UI__".into());
458    });
459    ctx.state.register_client_task(ctx.client_id, task).await;
460    Ok(remote)
461}
462#[allow(clippy::too_many_arguments)]
463async fn session(
464    ctx: &ConnectContext,
465    endpoint: &Endpoint,
466    version: &str,
467    negotiated: Value,
468    timeout: Duration,
469    mut external: mpsc::Receiver<ClientCommand>,
470    mut internal: mpsc::Receiver<HandlerAction>,
471    events: mpsc::Sender<HandlerEvent>,
472) -> Result<()> {
473    events
474        .try_send((
475            Event::new(
476                &actions::CONNECTED_EVENT,
477                json!({"endpoint":endpoint.name(),"api_version":version,"daemon":negotiated}),
478            ),
479            0,
480        ))
481        .context("Docker event queue full")?;
482    loop {
483        let (action, depth, command) = tokio::select! {command=external.recv()=>match command {Some(c)=>(c.action.clone(),0,Some(c)),None=>return Ok(())},action=internal.recv()=>match action {Some((a,d))=>(a,d,None),None=>return Ok(())}};
484        if action["type"] == "disconnect" {
485            if let Some(c) = command {
486                command_support::reply(c, Ok(ClientSendOutcome::Disconnected));
487            }
488            return Ok(());
489        }
490        let request = match request(&action, version) {
491            Ok(r) => r,
492            Err(e) => {
493                if let Some(c) = command {
494                    command_support::reply(
495                        c,
496                        Ok(ClientSendOutcome::Rejected {
497                            error: e.to_string(),
498                        }),
499                    );
500                } else {
501                    Log::new(Some(&ctx.status_tx)).warn(format!("Docker action rejected: {e}"));
502                }
503                continue;
504            }
505        };
506        if let Some(c) = command {
507            ctx.state
508                .record_access_log(
509                    AccessLogOwner::Client(ctx.client_id.as_u32()),
510                    "Docker",
511                    None,
512                    "injected_action",
513                    action.clone(),
514                    vec![json!({"accepted":true})],
515                )
516                .await;
517            command_support::reply(
518                c,
519                Ok(ClientSendOutcome::Executed {
520                    detail: "read-only request accepted; result follows as an event".into(),
521                }),
522            );
523        }
524        let result = {
525            let exchange = tokio::time::timeout(timeout, fetch(endpoint, "GET", &request.path));
526            tokio::pin!(exchange);
527            loop {
528                tokio::select! {biased;command=external.recv()=>{let Some(c)=command else{return Ok(())};if c.action["type"]=="disconnect" {command_support::reply(c,Ok(ClientSendOutcome::Disconnected));return Ok(());}command_support::reply(c,Ok(ClientSendOutcome::Rejected{error:"Docker request pending; retry after its event".into()}));},result=&mut exchange=>break result.context("whole Docker request deadline").and_then(|r|r)}
529            }
530        };
531        let event = match result {
532            Ok((status, headers, body)) => {
533                if status != 200 {
534                    let error =
535                        schema::error(&body).unwrap_or_else(|e| format!("HTTP {status}; {e:#}"));
536                    Event::new(
537                        &actions::ERROR_EVENT,
538                        json!({"request":action,"api_version":version,"status":status,"error":error,"category":"http"}),
539                    )
540                } else {
541                    match response(&request, status, &headers, &body) {
542                        Ok(data) => Event::new(
543                            &actions::RESPONSE_EVENT,
544                            json!({"request":action,"api_version":version,"operation":request.operation,"status":status,"data":data}),
545                        ),
546                        Err(e) => Event::new(
547                            &actions::ERROR_EVENT,
548                            json!({"request":action,"api_version":version,"status":status,"error":format!("{e:#}"),"category":"schema"}),
549                        ),
550                    }
551                }
552            }
553            Err(e) => Event::new(
554                &actions::ERROR_EVENT,
555                json!({"request":action,"api_version":version,"status":null,"error":format!("{e:#}"),"category":"transport"}),
556            ),
557        };
558        events
559            .try_send((event, depth))
560            .context("Docker event queue full; disconnecting bounded client")?;
561    }
562}
netget/client/docker/mod.rs

netget/client/docker/
mod.rs
