netget/server/consul/
mod.rs

1//! Consul agent HTTP API. Rust owns HTTP, Consul's JSON shapes, base64 values, the index
2//! header and the agent's own description; the handler answers every KV, catalog and
3//! registration request. NetGet stores nothing.
4pub mod actions;
5pub mod wire;
6
7use crate::llm::action_helper::call_llm;
8use crate::llm::actions::protocol_trait::ActionResult;
9use crate::logging::emit::Log;
10use crate::protocol::{Event, SpawnContext};
11use crate::server::{
12    accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS},
13    connection::ConnectionId,
14};
15use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
16use anyhow::Result;
17use bytes::Bytes;
18use http_body_util::{BodyExt, Full, Limited};
19use hyper::{
20    body::Incoming,
21    header::{HeaderValue, CONTENT_TYPE},
22    server::conn::http1,
23    service::service_fn,
24    Method, Request, Response, StatusCode,
25};
26use hyper_util::rt::{TokioIo, TokioTimer};
27use serde_json::{json, Map, Value};
28use std::{
29    collections::HashMap,
30    convert::Infallible,
31    net::SocketAddr,
32    sync::{
33        atomic::{AtomicU64, Ordering},
34        Arc,
35    },
36    time::Duration,
37};
38
39pub const HEADER_TIMEOUT: Duration = Duration::from_secs(30);
40pub const BODY_TIMEOUT: Duration = Duration::from_secs(30);
41pub const MAX_HEADER_BYTES: usize = 32 * 1024;
42pub const MAX_HEADERS: usize = 64;
43const CAP_REFUSAL: &[u8] = b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\nRetry-After: 5\r\n\r\n";
44
45type Reply = Response<Full<Bytes>>;
46
47pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
48    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
49    let local = listener.local_addr()?;
50    Log::new(Some(&ctx.status_tx)).info(format!("Consul agent API listening on {local}"));
51    let index = Arc::new(AtomicU64::new(1));
52    let server_id = ctx.server_id;
53    let state = ctx.state.clone();
54    let accept = tokio::spawn(async move {
55        let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
56        loop {
57            let (stream, peer, permit) = match accept_bounded(
58                &listener,
59                &limiter,
60                CAP_REFUSAL,
61                "Consul",
62                Some(&ctx.status_tx),
63            )
64            .await
65            {
66                Ok(v) => v,
67                Err(_) => break,
68            };
69            let id = ConnectionId::new(ctx.state.get_next_unified_id().await);
70            let now = crate::utils::clock::Instant::now();
71            ctx.state
72                .add_connection_to_server(
73                    server_id,
74                    ConnectionState {
75                        id,
76                        remote_addr: peer,
77                        local_addr: local,
78                        bytes_sent: 0,
79                        bytes_received: 0,
80                        packets_sent: 0,
81                        packets_received: 0,
82                        last_activity: now,
83                        status: ConnectionStatus::Active,
84                        status_changed_at: now,
85                        protocol_info: ProtocolConnectionInfo::empty(),
86                    },
87                )
88                .await;
89            let child = ctx.clone();
90            let index = index.clone();
91            ctx.state
92                .spawn_server_task(server_id, async move {
93                    let _permit = permit;
94                    let request_ctx = child.clone();
95                    let service = service_fn(move |request| {
96                        let ctx = request_ctx.clone();
97                        let index = index.clone();
98                        async move { Ok::<_, Infallible>(handle(request, id, &ctx, &index).await) }
99                    });
100                    let mut builder = http1::Builder::new();
101                    builder
102                        .keep_alive(false)
103                        .timer(TokioTimer::new())
104                        .header_read_timeout(HEADER_TIMEOUT)
105                        .max_headers(MAX_HEADERS)
106                        .max_buf_size(MAX_HEADER_BYTES);
107                    if let Err(e) = builder
108                        .serve_connection(TokioIo::new(stream), service)
109                        .await
110                    {
111                        Log::new(Some(&child.status_tx))
112                            .warn(format!("Consul connection {id} HTTP error: {e}"));
113                    }
114                    child
115                        .state
116                        .update_connection_status(server_id, id, ConnectionStatus::Closed)
117                        .await;
118                    let _ = child.status_tx.send("__UPDATE_UI__".into());
119                })
120                .await;
121        }
122    });
123    state.register_server_task(server_id, accept).await;
124    Ok(local)
125}
126
127fn reply(status: u16, content_type: &str, body: Vec<u8>, index: Option<u64>) -> Reply {
128    let mut r = Response::new(Full::new(Bytes::from(body)));
129    *r.status_mut() = StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
130    if let Ok(v) = HeaderValue::from_str(content_type) {
131        r.headers_mut().insert(CONTENT_TYPE, v);
132    }
133    if let Some(i) = index {
134        let h = r.headers_mut();
135        h.insert("X-Consul-Index", HeaderValue::from(i));
136        h.insert("X-Consul-KnownLeader", HeaderValue::from_static("true"));
137        h.insert("X-Consul-LastContact", HeaderValue::from_static("0"));
138    }
139    r
140}
141
142fn json_reply(v: &Value, index: u64) -> Reply {
143    reply(
144        200,
145        "application/json",
146        serde_json::to_vec(v).unwrap_or_default(),
147        Some(index),
148    )
149}
150
151fn text(status: u16, body: &str) -> Reply {
152    reply(
153        status,
154        "text/plain; charset=utf-8",
155        body.as_bytes().to_vec(),
156        None,
157    )
158}
159
160fn outcome(ctx: &SpawnContext, id: ConnectionId, op: &str, decision: &str) {
161    let line = format!("Consul connection {id} operation={op} decision={decision}");
162    let log = Log::new(Some(&ctx.status_tx));
163    if decision.starts_with("fail_closed") || decision == "model_silent" {
164        log.error(line)
165    } else {
166        log.info(line)
167    }
168}
169
170fn failure(e: Option<&anyhow::Error>) -> Reply {
171    let message = match e {
172        Some(e) => crate::utils::wire_failure::prefixed_wire_failure_text(e),
173        None => crate::utils::WireFailure::Unavailable.prefixed_text(),
174    };
175    text(500, message)
176}
177
178/// Ask the handler; one `consul_*` answer, or the reply a failure earns.
179async fn ask(
180    ctx: &SpawnContext,
181    id: ConnectionId,
182    event: Event,
183    op: &str,
184) -> std::result::Result<(String, Value), Reply> {
185    let result = match call_llm(
186        &ctx.llm_client,
187        &ctx.state,
188        ctx.server_id,
189        Some(id),
190        &event,
191        &actions::ConsulProtocol,
192    )
193    .await
194    {
195        Ok(r) => r,
196        Err(e) => {
197            outcome(ctx, id, op, "fail_closed_llm_error");
198            return Err(failure(Some(&e)));
199        }
200    };
201    if !result.failures.is_empty() {
202        outcome(ctx, id, op, "fail_closed_invalid_reply");
203        return Err(failure(None));
204    }
205    let mut answers = Vec::new();
206    let mut stack = result.protocol_results;
207    while let Some(r) = stack.pop() {
208        match r {
209            ActionResult::Custom { name, data } if name.starts_with("consul_") => {
210                answers.push((name, data))
211            }
212            ActionResult::Multiple(items) => stack.extend(items),
213            _ => {}
214        }
215    }
216    match answers.len() {
217        1 => {
218            let (name, data) = answers.pop().unwrap();
219            if name == "consul_error" {
220                outcome(ctx, id, op, "model_reject");
221                return Err(text(
222                    data["status"].as_u64().unwrap_or(500) as u16,
223                    data["message"].as_str().unwrap_or_default(),
224                ));
225            }
226            Ok((name, data))
227        }
228        0 => {
229            outcome(ctx, id, op, "model_silent");
230            Err(failure(None))
231        }
232        _ => {
233            outcome(ctx, id, op, "fail_closed_invalid_reply");
234            Err(failure(None))
235        }
236    }
237}
238
239fn query(q: Option<&str>) -> HashMap<String, String> {
240    q.unwrap_or_default()
241        .split('&')
242        .filter(|p| !p.is_empty())
243        .map(|p| {
244            let (k, v) = p.split_once('=').unwrap_or((p, ""));
245            (
246                urlencoding::decode(k)
247                    .map(|c| c.into_owned())
248                    .unwrap_or_default(),
249                urlencoding::decode(v)
250                    .map(|c| c.into_owned())
251                    .unwrap_or_default(),
252            )
253        })
254        .collect()
255}
256
257async fn handle(
258    request: Request<Incoming>,
259    id: ConnectionId,
260    ctx: &SpawnContext,
261    index: &AtomicU64,
262) -> Reply {
263    let (parts, body) = request.into_parts();
264    let bytes = match tokio::time::timeout(
265        BODY_TIMEOUT,
266        Limited::new(body, wire::MAX_VALUE_BYTES).collect(),
267    )
268    .await
269    {
270        Err(_) => return text(408, "request body deadline exceeded"),
271        Ok(Err(e)) => {
272            return if e
273                .downcast_ref::<http_body_util::LengthLimitError>()
274                .is_some()
275            {
276                text(413, "Request body too large, max size: 524288 bytes")
277            } else {
278                text(400, "incomplete HTTP body")
279            }
280        }
281        Ok(Ok(b)) => b.to_bytes(),
282    };
283    ctx.state
284        .update_connection_stats(
285            ctx.server_id,
286            id,
287            Some(bytes.len() as u64),
288            None,
289            Some(1),
290            None,
291        )
292        .await;
293    let path = parts.uri.path().to_string();
294    let q = query(parts.uri.query());
295    let method = parts.method.clone();
296    let now = index.load(Ordering::SeqCst);
297    let Some(rest) = path.strip_prefix("/v1/") else {
298        return text(404, "Not found");
299    };
300    if let Some(key) = rest.strip_prefix("kv/") {
301        let key = urlencoding::decode(key)
302            .map(|c| c.into_owned())
303            .unwrap_or_default();
304        if wire::check_key(&key).is_err() {
305            return text(400, "Invalid key");
306        }
307        return kv(ctx, id, index, &method, &key, &q, &bytes).await;
308    }
309    match (method.clone(), rest) {
310        (Method::GET, "status/leader") => json_reply(&json!("127.0.0.1:8300"), now),
311        (Method::GET, "status/peers") => json_reply(&json!(["127.0.0.1:8300"]), now),
312        (Method::GET, "agent/self") => json_reply(&wire::agent_self(), now),
313        (Method::GET, "catalog/datacenters") => json_reply(&json!([wire::DATACENTER]), now),
314        (Method::GET, "catalog/nodes") => json_reply(
315            &json!([{"ID": "00000000-0000-0000-0000-00000000c0de", "Node": wire::NODE, "Address": "127.0.0.1",
316                     "Datacenter": wire::DATACENTER, "TaggedAddresses": {"lan": "127.0.0.1", "wan": "127.0.0.1"},
317                     "Meta": {}, "CreateIndex": 1, "ModifyIndex": 1}]),
318            now,
319        ),
320        (Method::GET, "catalog/services") => catalog(ctx, id, now, "services", None).await,
321        (Method::GET, "agent/services") => catalog(ctx, id, now, "agent_services", None).await,
322        (Method::GET, r) if r.starts_with("catalog/service/") => {
323            catalog(
324                ctx,
325                id,
326                now,
327                "service",
328                Some(&r["catalog/service/".len()..]),
329            )
330            .await
331        }
332        (Method::GET, r) if r.starts_with("health/service/") => {
333            catalog(ctx, id, now, "health", Some(&r["health/service/".len()..])).await
334        }
335        (Method::PUT, "agent/service/register") => {
336            let def = match serde_json::from_slice::<Value>(&bytes)
337                .map_err(anyhow::Error::from)
338                .and_then(|b| wire::registration(&b))
339            {
340                Ok(d) => d,
341                Err(e) => return text(400, &format!("Request decode failed: {e:#}")),
342            };
343            register(
344                ctx,
345                id,
346                index,
347                json!({"operation": "register", "service": def}),
348            )
349            .await
350        }
351        (Method::PUT, r) if r.starts_with("agent/service/deregister/") => {
352            let sid = &r["agent/service/deregister/".len()..];
353            register(
354                ctx,
355                id,
356                index,
357                json!({"operation": "deregister", "id": sid}),
358            )
359            .await
360        }
361        _ => text(
362            404,
363            "Not found: this agent serves kv, catalog, health, agent services and status",
364        ),
365    }
366}
367
368async fn kv(
369    ctx: &SpawnContext,
370    id: ConnectionId,
371    index: &AtomicU64,
372    method: &Method,
373    key: &str,
374    q: &HashMap<String, String>,
375    body: &[u8],
376) -> Reply {
377    let now = index.load(Ordering::SeqCst);
378    let flag = |k: &str| q.contains_key(k);
379    match *method {
380        Method::GET => {
381            let recurse = flag("recurse");
382            let keys_only = flag("keys");
383            let event = Event::new(
384                &actions::KV_READ_EVENT,
385                json!({"key": key, "recurse": recurse, "keys_only": keys_only}),
386            );
387            let (name, data) = match ask(ctx, id, event, "kv_read").await {
388                Ok(a) => a,
389                Err(r) => return r,
390            };
391            if name == "consul_not_found" {
392                outcome(ctx, id, "kv_read", "model_answer");
393                let mut r = text(404, "");
394                r.headers_mut()
395                    .insert("X-Consul-Index", HeaderValue::from(now));
396                return r;
397            }
398            if name != "consul_kv_entries" {
399                outcome(ctx, id, "kv_read", "fail_closed_invalid_reply");
400                return failure(None);
401            }
402            let entries: Vec<Value> = match data["entries"].as_array().map(|a| {
403                a.iter()
404                    .map(|e| wire::kv_entry(e, now))
405                    .collect::<Result<Vec<_>>>()
406            }) {
407                Some(Ok(e)) => e,
408                _ => {
409                    outcome(ctx, id, "kv_read", "fail_closed_invalid_reply");
410                    return failure(None);
411                }
412            };
413            outcome(ctx, id, "kv_read", "model_answer");
414            if entries.is_empty() {
415                return text(404, "");
416            }
417            if keys_only {
418                let keys: Vec<&Value> = entries.iter().map(|e| &e["Key"]).collect();
419                return json_reply(&json!(keys), now);
420            }
421            if flag("raw") {
422                let v = entries[0]["Value"]
423                    .as_str()
424                    .map(wire::unb64)
425                    .transpose()
426                    .ok()
427                    .flatten()
428                    .unwrap_or_default();
429                return reply(200, "text/plain; charset=utf-8", v, Some(now));
430            }
431            let shown = if recurse {
432                entries
433            } else {
434                entries.into_iter().take(1).collect()
435            };
436            json_reply(&Value::Array(shown), now)
437        }
438        Method::PUT | Method::DELETE => {
439            let cas = q.get("cas").and_then(|c| c.parse::<u64>().ok());
440            let (event, op) = if *method == Method::PUT {
441                let (value, encoding) = wire::shown(body);
442                let flags = q
443                    .get("flags")
444                    .and_then(|f| f.parse::<u64>().ok())
445                    .unwrap_or(0);
446                (
447                    Event::new(
448                        &actions::KV_WRITE_EVENT,
449                        json!({"key": key, "value": value, "value_encoding": encoding, "flags": flags, "cas": cas}),
450                    ),
451                    "kv_write",
452                )
453            } else {
454                (
455                    Event::new(
456                        &actions::KV_DELETE_EVENT,
457                        json!({"key": key, "recurse": flag("recurse"), "cas": cas}),
458                    ),
459                    "kv_delete",
460                )
461            };
462            let (name, _) = match ask(ctx, id, event, op).await {
463                Ok(a) => a,
464                Err(r) => return r,
465            };
466            let accepted = match name.as_str() {
467                "consul_ok" => true,
468                "consul_refuse" => false,
469                _ => {
470                    outcome(ctx, id, op, "fail_closed_invalid_reply");
471                    return failure(None);
472                }
473            };
474            outcome(
475                ctx,
476                id,
477                op,
478                if accepted {
479                    "model_answer"
480                } else {
481                    "model_reject"
482                },
483            );
484            let i = if accepted {
485                index.fetch_add(1, Ordering::SeqCst) + 1
486            } else {
487                now
488            };
489            json_reply(&json!(accepted), i)
490        }
491        _ => text(405, "Method not allowed"),
492    }
493}
494
495async fn catalog(
496    ctx: &SpawnContext,
497    id: ConnectionId,
498    now: u64,
499    endpoint: &str,
500    name: Option<&str>,
501) -> Reply {
502    let name = name.map(|n| {
503        urlencoding::decode(n)
504            .map(|c| c.into_owned())
505            .unwrap_or_default()
506    });
507    let event = Event::new(
508        &actions::CATALOG_EVENT,
509        json!({"endpoint": endpoint, "name": name}),
510    );
511    let (action, data) = match ask(ctx, id, event, endpoint).await {
512        Ok(a) => a,
513        Err(r) => return r,
514    };
515    let body = match (endpoint, action.as_str()) {
516        ("services", "consul_services") => {
517            let mut out = Map::new();
518            for (k, v) in data["services"].as_object().cloned().unwrap_or_default() {
519                let tags: Vec<Value> = v
520                    .as_array()
521                    .cloned()
522                    .unwrap_or_default()
523                    .into_iter()
524                    .filter(Value::is_string)
525                    .collect();
526                out.insert(k, Value::Array(tags));
527            }
528            Value::Object(out)
529        }
530        (_, "consul_instances") if endpoint != "services" => {
531            let list: Result<Vec<wire::Instance>> = data["instances"]
532                .as_array()
533                .cloned()
534                .unwrap_or_default()
535                .iter()
536                .map(wire::instance)
537                .collect();
538            let Ok(list) = list else {
539                outcome(ctx, id, endpoint, "fail_closed_invalid_reply");
540                return failure(None);
541            };
542            match endpoint {
543                "service" => {
544                    Value::Array(list.iter().map(|i| wire::catalog_entry(i, now)).collect())
545                }
546                "health" => Value::Array(list.iter().map(|i| wire::health_entry(i, now)).collect()),
547                _ => Value::Object(
548                    list.iter()
549                        .map(|i| (i.id.clone(), wire::agent_service(i)))
550                        .collect(),
551                ),
552            }
553        }
554        _ => {
555            outcome(ctx, id, endpoint, "fail_closed_invalid_reply");
556            return failure(None);
557        }
558    };
559    outcome(ctx, id, endpoint, "model_answer");
560    json_reply(&body, now)
561}
562
563async fn register(ctx: &SpawnContext, id: ConnectionId, index: &AtomicU64, data: Value) -> Reply {
564    let op = data["operation"].as_str().unwrap_or("register").to_string();
565    let (name, _) = match ask(ctx, id, Event::new(&actions::REGISTER_EVENT, data), &op).await {
566        Ok(a) => a,
567        Err(r) => return r,
568    };
569    if name != "consul_ok" {
570        outcome(ctx, id, &op, "fail_closed_invalid_reply");
571        return failure(None);
572    }
573    outcome(ctx, id, &op, "model_answer");
574    let i = index.fetch_add(1, Ordering::SeqCst) + 1;
575    reply(200, "text/plain; charset=utf-8", Vec::new(), Some(i))
576}
netget/server/consul/mod.rs

netget/server/consul/
mod.rs
