netget/server/jmap/
mod.rs

1//! JMAP (RFC 8620, RFC 8621) server. Rust owns the session resource, authentication, request
2//! validation and limits, capabilities, accounts, Core/echo, result references and creation ids;
3//! the handler is the store and answers every other method call.
4pub mod actions;
5pub mod request;
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::accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS};
12use crate::server::connection::ConnectionId;
13use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
14use base64::Engine;
15use bytes::Bytes;
16use http_body_util::{BodyExt, Full, Limited};
17use hyper::{
18    body::Incoming, header, server::conn::http1, service::service_fn, Method, Request, Response,
19    StatusCode,
20};
21use hyper_util::rt::{TokioIo, TokioTimer};
22use request::{method_error, Problem};
23use serde_json::{json, Map, Value};
24use std::collections::BTreeMap;
25use std::{convert::Infallible, net::SocketAddr, sync::Arc, time::Duration};
26
27pub const MAX_BODY: usize = 1024 * 1024;
28const HEADER_TIMEOUT: Duration = Duration::from_secs(30);
29const TLS_TIMEOUT: Duration = Duration::from_secs(10);
30
31struct Shared {
32    ctx: SpawnContext,
33    /// (id, name), the first primary.
34    accounts: Vec<(String, String)>,
35    users: BTreeMap<String, String>,
36    tokens: BTreeMap<String, String>,
37    scheme: &'static str,
38    state: String,
39}
40
41fn string_map(
42    p: Option<&crate::protocol::StartupParams>,
43    name: &str,
44) -> anyhow::Result<BTreeMap<String, String>> {
45    let mut out = BTreeMap::new();
46    if let Some(m) = p
47        .map(|p| p.get_optional_object(name))
48        .transpose()?
49        .flatten()
50    {
51        anyhow::ensure!(m.len() <= 64, "at most 64 entries in {name}");
52        for (k, v) in m {
53            let v = v
54                .as_str()
55                .ok_or_else(|| anyhow::anyhow!("{name} values are strings"))?;
56            anyhow::ensure!(
57                !k.is_empty() && !v.is_empty(),
58                "{name} entries are not empty"
59            );
60            out.insert(k.clone(), v.to_owned());
61        }
62    }
63    Ok(out)
64}
65
66pub async fn spawn(ctx: SpawnContext) -> anyhow::Result<SocketAddr> {
67    let p = ctx.startup_params.as_ref();
68    let mut accounts = Vec::new();
69    for a in p
70        .map(|p| p.get_optional_array("accounts"))
71        .transpose()?
72        .flatten()
73        .into_iter()
74        .flatten()
75    {
76        let id = a["id"]
77            .as_str()
78            .filter(|i| {
79                !i.is_empty()
80                    && i.len() <= 255
81                    && i.bytes()
82                        .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
83            })
84            .ok_or_else(|| anyhow::anyhow!("each account has an id of letters, digits, - and _"))?;
85        let name = a["name"].as_str().unwrap_or(id);
86        anyhow::ensure!(accounts.len() < 64, "at most 64 accounts");
87        accounts.push((id.to_owned(), name.to_owned()));
88    }
89    if accounts.is_empty() {
90        accounts.push(("a1".into(), "netget".into()));
91    }
92    let users = string_map(p, "users")?;
93    let tokens = string_map(p, "api_tokens")?;
94    let get = |name| {
95        p.map(|p| p.get_optional_string(name))
96            .transpose()
97            .map(Option::flatten)
98    };
99    let tls_on = p
100        .map(|p| p.get_optional_bool("tls"))
101        .transpose()?
102        .flatten()
103        .unwrap_or(true);
104    let tls = match (tls_on, get("tls_cert_file")?, get("tls_key_file")?) {
105        (false, None, None) => None,
106        (false, _, _) => anyhow::bail!("tls is false but a certificate was given"),
107        (true, Some(c), Some(k)) => Some(tokio_rustls::TlsAcceptor::from(
108            crate::server::tls_cert_manager::load_tls_config_from_files(&c, &k)?,
109        )),
110        (true, None, None) => {
111            let spec = crate::server::tls_cert_manager::CertificateSpec {
112                common_name: "localhost".into(),
113                san_dns_names: vec!["localhost".into()],
114                validity_days: 30,
115                organization: Some("NetGet".into()),
116                organizational_unit: Some("JMAP".into()),
117            };
118            let (cert, key) = crate::server::tls_cert_manager::generate_self_signed_cert(&spec)?;
119            let pem = cert.pem();
120            ctx.state
121                .with_server_mut(ctx.server_id, |s| {
122                    s.set_protocol_field("certificate_pem".into(), json!(pem))
123                })
124                .await;
125            Some(tokio_rustls::TlsAcceptor::from(
126                crate::server::tls_cert_manager::create_rustls_server_config(&cert, &key)?,
127            ))
128        }
129        _ => anyhow::bail!("tls_cert_file and tls_key_file go together"),
130    };
131    let scheme = if tls.is_some() { "https" } else { "http" };
132    let state = {
133        use std::hash::{Hash, Hasher};
134        let mut h = std::collections::hash_map::DefaultHasher::new();
135        accounts.hash(&mut h);
136        format!("{:x}", h.finish())
137    };
138    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
139    let local = listener.local_addr()?;
140    Log::new(Some(&ctx.status_tx)).info(format!(
141        "JMAP session at {scheme}://{local}/.well-known/jmap"
142    ));
143    let shared = Arc::new(Shared {
144        ctx: ctx.clone(),
145        accounts,
146        users,
147        tokens,
148        scheme,
149        state,
150    });
151    let server_id = ctx.server_id;
152    let accept = tokio::spawn(async move {
153        let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
154        loop {
155            let (stream, peer, permit) = match accept_bounded(&listener, &limiter, b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", "JMAP", Some(&shared.ctx.status_tx)).await {
156                Ok(v) => v,
157                Err(_) => break,
158            };
159            let id = ConnectionId::new(shared.ctx.state.get_next_unified_id().await);
160            let now = crate::utils::clock::Instant::now();
161            shared
162                .ctx
163                .state
164                .add_connection_to_server(
165                    server_id,
166                    ConnectionState {
167                        id,
168                        remote_addr: peer,
169                        local_addr: local,
170                        bytes_sent: 0,
171                        bytes_received: 0,
172                        packets_sent: 0,
173                        packets_received: 0,
174                        last_activity: now,
175                        status: ConnectionStatus::Active,
176                        status_changed_at: now,
177                        protocol_info: ProtocolConnectionInfo::empty(),
178                    },
179                )
180                .await;
181            let child = shared.clone();
182            let tls = tls.clone();
183            shared
184                .ctx
185                .state
186                .spawn_server_task(server_id, async move {
187                    let _permit = permit;
188                    let svc_shared = child.clone();
189                    let service = service_fn(move |req| {
190                        let shared = svc_shared.clone();
191                        async move { Ok::<_, Infallible>(handle(&shared, id, req).await) }
192                    });
193                    let mut builder = http1::Builder::new();
194                    builder
195                        .timer(TokioTimer::new())
196                        .header_read_timeout(HEADER_TIMEOUT)
197                        .max_headers(64)
198                        .max_buf_size(32 * 1024);
199                    let served = match tls {
200                        Some(acceptor) => {
201                            match tokio::time::timeout(TLS_TIMEOUT, acceptor.accept(stream)).await {
202                                Ok(Ok(s)) => builder
203                                    .serve_connection(TokioIo::new(s), service)
204                                    .await
205                                    .map_err(|e| e.to_string()),
206                                Ok(Err(e)) => Err(format!("TLS handshake: {e}")),
207                                Err(_) => Err("TLS handshake timed out".into()),
208                            }
209                        }
210                        None => builder
211                            .serve_connection(TokioIo::new(stream), service)
212                            .await
213                            .map_err(|e| e.to_string()),
214                    };
215                    if let Err(e) = served {
216                        Log::new(Some(&child.ctx.status_tx))
217                            .debug(format!("JMAP connection {id}: {e}"));
218                    }
219                    child
220                        .ctx
221                        .state
222                        .update_connection_status(server_id, id, ConnectionStatus::Closed)
223                        .await;
224                    let _ = child.ctx.status_tx.send("__UPDATE_UI__".into());
225                })
226                .await;
227        }
228    });
229    ctx.state.register_server_task(server_id, accept).await;
230    Ok(local)
231}
232
233type Reply = Response<Full<Bytes>>;
234
235fn reply(status: u16, content_type: &str, body: Vec<u8>) -> Reply {
236    let mut r = Response::new(Full::new(Bytes::from(body)));
237    *r.status_mut() = StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
238    if let Ok(ct) = header::HeaderValue::from_str(content_type) {
239        r.headers_mut().insert(header::CONTENT_TYPE, ct);
240    }
241    r.headers_mut().insert(
242        header::CACHE_CONTROL,
243        header::HeaderValue::from_static("no-cache, no-store"),
244    );
245    r
246}
247
248fn json_reply(status: u16, body: &Value) -> Reply {
249    reply(
250        status,
251        "application/json",
252        serde_json::to_vec(body).unwrap_or_default(),
253    )
254}
255
256fn problem(status: u16, kind: &str, detail: &str) -> Reply {
257    reply(
258        status,
259        "application/problem+json",
260        serde_json::to_vec(&json!({"type": kind, "status": status, "detail": detail}))
261            .unwrap_or_default(),
262    )
263}
264
265fn log_decision(ctx: &SpawnContext, id: ConnectionId, what: &str, decision: &str) {
266    let line = format!("JMAP connection {id} {what} decision={decision}");
267    let log = Log::new(Some(&ctx.status_tx));
268    if decision.starts_with("fail_closed_") || decision == "model_silent" {
269        log.error(line);
270    } else {
271        log.info(line);
272    }
273}
274
275async fn handle(shared: &Shared, id: ConnectionId, req: Request<Incoming>) -> Reply {
276    let ctx = &shared.ctx;
277    ctx.state
278        .update_connection_stats(
279            ctx.server_id,
280            id,
281            Some(req.uri().to_string().len() as u64),
282            None,
283            Some(1),
284            None,
285        )
286        .await;
287    let response = route(shared, id, req).await;
288    let sent = hyper::body::Body::size_hint(response.body())
289        .exact()
290        .unwrap_or(0);
291    ctx.state
292        .update_connection_stats(ctx.server_id, id, None, Some(sent), None, Some(1))
293        .await;
294    response
295}
296
297/// The authenticated user, None when credentials are required and wrong or missing.
298fn authenticate(shared: &Shared, req: &Request<Incoming>) -> Option<String> {
299    let given = req
300        .headers()
301        .get(header::AUTHORIZATION)
302        .and_then(|a| a.to_str().ok());
303    if shared.users.is_empty() && shared.tokens.is_empty() {
304        return Some(
305            given
306                .and_then(basic)
307                .map(|(u, _)| u)
308                .unwrap_or_else(|| shared.accounts[0].1.clone()),
309        );
310    }
311    let given = given?;
312    if let Some(token) = given.strip_prefix("Bearer ") {
313        return shared.tokens.get(token.trim()).cloned();
314    }
315    let (user, password) = basic(given)?;
316    (shared.users.get(&user) == Some(&password)).then_some(user)
317}
318
319fn basic(header: &str) -> Option<(String, String)> {
320    let raw = header.strip_prefix("Basic ")?;
321    let decoded = base64::engine::general_purpose::STANDARD
322        .decode(raw.trim())
323        .ok()?;
324    let text = String::from_utf8(decoded).ok()?;
325    let (u, p) = text.split_once(':')?;
326    Some((u.to_owned(), p.to_owned()))
327}
328
329async fn route(shared: &Shared, id: ConnectionId, req: Request<Incoming>) -> Reply {
330    let path = req.uri().path().to_owned();
331    let Some(user) = authenticate(shared, &req) else {
332        log_decision(
333            &shared.ctx,
334            id,
335            &path,
336            "protocol_refusal reason=unauthenticated",
337        );
338        let mut r = problem(401, "about:blank", "authentication is required");
339        r.headers_mut().insert(
340            header::WWW_AUTHENTICATE,
341            header::HeaderValue::from_static("Basic realm=\"JMAP\""),
342        );
343        return r;
344    };
345    let host = req
346        .headers()
347        .get(header::HOST)
348        .and_then(|h| h.to_str().ok())
349        .filter(|h| h.len() <= 255 && !h.contains(['/', ' ']))
350        .unwrap_or("localhost")
351        .to_owned();
352    match (req.method(), path.as_str()) {
353        (&Method::GET, "/.well-known/jmap" | "/jmap/session") => {
354            json_reply(200, &session(shared, &host, &user))
355        }
356        (&Method::POST, "/jmap" | "/jmap/") => api(shared, id, &user, req).await,
357        (_, p)
358            if p.starts_with("/jmap/upload/")
359                || p.starts_with("/jmap/download/")
360                || p.starts_with("/jmap/eventsource") =>
361        {
362            problem(
363                501,
364                "about:blank",
365                "blobs and push are not implemented by this server",
366            )
367        }
368        (_, "/.well-known/jmap" | "/jmap" | "/jmap/" | "/jmap/session") => {
369            problem(405, "about:blank", "method not allowed")
370        }
371        _ => problem(404, "about:blank", "no such resource"),
372    }
373}
374
375fn session(shared: &Shared, host: &str, user: &str) -> Value {
376    let base = format!("{}://{host}", shared.scheme);
377    let mut accounts = Map::new();
378    for (id, name) in &shared.accounts {
379        accounts.insert(id.clone(), json!({
380            "name": name,
381            "isPersonal": true,
382            "isReadOnly": false,
383            "accountCapabilities": {
384                request::MAIL: {"maxMailboxesPerEmail": null, "maxMailboxDepth": 10, "maxSizeMailboxName": 255, "maxSizeAttachmentsPerEmail": 50_000_000, "emailQuerySortOptions": ["receivedAt", "sentAt", "size", "from", "to", "subject"], "mayCreateTopLevelMailbox": true},
385                request::SUBMISSION: {"maxDelayedSend": 0, "submissionExtensions": {}},
386                request::VACATION: {},
387            },
388        }));
389    }
390    let primary = &shared.accounts[0].0;
391    json!({
392        "capabilities": {
393            request::CORE: {
394                "maxSizeUpload": 0,
395                "maxConcurrentUpload": 1,
396                "maxSizeRequest": MAX_BODY,
397                "maxConcurrentRequests": 4,
398                "maxCallsInRequest": request::MAX_CALLS,
399                "maxObjectsInGet": request::MAX_OBJECTS_IN_GET,
400                "maxObjectsInSet": request::MAX_OBJECTS_IN_SET,
401                "collationAlgorithms": ["i;ascii-numeric", "i;ascii-casemap", "i;octet"],
402            },
403            request::MAIL: {},
404            request::SUBMISSION: {},
405            request::VACATION: {},
406        },
407        "accounts": accounts,
408        "primaryAccounts": {request::MAIL: primary, request::SUBMISSION: primary, request::VACATION: primary},
409        "username": user,
410        "apiUrl": format!("{base}/jmap/"),
411        "downloadUrl": format!("{base}/jmap/download/{{accountId}}/{{blobId}}/{{name}}?accept={{type}}"),
412        "uploadUrl": format!("{base}/jmap/upload/{{accountId}}/"),
413        "eventSourceUrl": format!("{base}/jmap/eventsource/?types={{types}}&closeafter={{closeafter}}&ping={{ping}}"),
414        "state": shared.state,
415    })
416}
417
418fn problem_reply(p: &Problem) -> Reply {
419    reply(
420        400,
421        "application/problem+json",
422        serde_json::to_vec(&p.body()).unwrap_or_default(),
423    )
424}
425
426async fn api(shared: &Shared, id: ConnectionId, user: &str, req: Request<Incoming>) -> Reply {
427    let bytes = match Limited::new(req.into_body(), MAX_BODY).collect().await {
428        Ok(b) => b.to_bytes(),
429        Err(_) => {
430            return problem_reply(&Problem {
431                kind: "limit",
432                detail: format!("requests are at most {MAX_BODY} bytes"),
433                limit: Some("maxSizeRequest"),
434            })
435        }
436    };
437    let request = match request::parse(&bytes) {
438        Ok(r) => r,
439        Err(p) => {
440            log_decision(
441                &shared.ctx,
442                id,
443                "request",
444                &format!("protocol_refusal reason={}", p.kind),
445            );
446            return problem_reply(&p);
447        }
448    };
449    let mut created = request.created_ids.clone().unwrap_or_default();
450    let mut responses: Vec<(String, Value, String)> = Vec::new();
451    for call in &request.calls {
452        let answers = match prepare(shared, &request.using, call, &responses, &created) {
453            Err(e) => vec![("error".to_owned(), e)],
454            Ok(args) if call.name == "Core/echo" => vec![(call.name.clone(), Value::Object(args))],
455            Ok(args) => answer(shared, id, user, call, args).await,
456        };
457        for (name, args) in answers {
458            request::record_created(&args, &mut created);
459            responses.push((name, args, call.id.clone()));
460        }
461    }
462    let mut body = json!({
463        "methodResponses": responses.into_iter().map(|(n, a, i)| json!([n, a, i])).collect::<Vec<_>>(),
464        "sessionState": shared.state,
465    });
466    if request.created_ids.is_some() {
467        body["createdIds"] = Value::Object(created);
468    }
469    json_reply(200, &body)
470}
471
472/// Everything Rust decides about a call before the handler sees it.
473fn prepare(
474    shared: &Shared,
475    using: &[String],
476    call: &request::Call,
477    earlier: &[(String, Value, String)],
478    created: &Map<String, Value>,
479) -> Result<Map<String, Value>, Value> {
480    let args = request::resolve_references(&call.arguments, earlier)?;
481    let args = match request::substitute_creation_ids(&Value::Object(args), created) {
482        Value::Object(m) => m,
483        _ => unreachable!("an object stays an object"),
484    };
485    let Some(capability) = request::capability(&call.name) else {
486        return Err(method_error(
487            "unknownMethod",
488            &format!("{} is not a method of this server", call.name),
489        ));
490    };
491    if !using.iter().any(|u| u == capability) {
492        return Err(method_error(
493            "unknownMethod",
494            &format!("{} needs {capability} in using", call.name),
495        ));
496    }
497    if capability != request::CORE {
498        let account = args
499            .get("accountId")
500            .and_then(Value::as_str)
501            .ok_or_else(|| method_error("invalidArguments", "accountId is required"))?;
502        if !shared.accounts.iter().any(|(a, _)| a == account) {
503            return Err(method_error("accountNotFound", account));
504        }
505    }
506    if let Some(e) = request::check_limits(&call.name, &args) {
507        return Err(e);
508    }
509    Ok(args)
510}
511
512/// One handler turn for a call: its responses as (name, arguments).
513async fn answer(
514    shared: &Shared,
515    id: ConnectionId,
516    user: &str,
517    call: &request::Call,
518    args: Map<String, Value>,
519) -> Vec<(String, Value)> {
520    let ctx = &shared.ctx;
521    let account = args.get("accountId").cloned();
522    let event = Event::new(
523        &actions::METHOD_CALL_EVENT,
524        json!({"method": call.name, "account_id": account, "arguments": args, "call_id": call.id, "username": user}),
525    );
526    let result = match call_llm(
527        &ctx.llm_client,
528        &ctx.state,
529        ctx.server_id,
530        Some(id),
531        &event,
532        &actions::JmapProtocol,
533    )
534    .await
535    {
536        Ok(r) => r,
537        Err(e) => {
538            log_decision(ctx, id, &call.name, "fail_closed_llm_error");
539            let kind = if crate::utils::WireFailure::classify(&e).is_overloaded() {
540                "serverUnavailable"
541            } else {
542                "serverFail"
543            };
544            return vec![("error".into(), method_error(kind, ""))];
545        }
546    };
547    if !result.failures.is_empty() {
548        log_decision(ctx, id, &call.name, "fail_closed_invalid_reply");
549        return vec![("error".into(), method_error("serverFail", ""))];
550    }
551    let mut out = Vec::new();
552    for r in result.protocol_results {
553        let ActionResult::Custom { data, .. } = r else {
554            continue;
555        };
556        if data["type"] == "jmap_method_error" {
557            log_decision(ctx, id, &call.name, "model_reject");
558            out.push((
559                "error".to_owned(),
560                method_error(
561                    data["error_type"].as_str().unwrap_or("serverFail"),
562                    data["description"].as_str().unwrap_or_default(),
563                ),
564            ));
565        } else {
566            let mut a = data["arguments"].clone();
567            if let (Some(acc), Some(m)) = (&account, a.as_object_mut()) {
568                m.entry("accountId").or_insert_with(|| acc.clone());
569            }
570            let name = data["method"].as_str().unwrap_or(&call.name).to_owned();
571            out.push((name, a));
572        }
573    }
574    if out.is_empty() {
575        log_decision(ctx, id, &call.name, "model_silent");
576        return vec![("error".into(), method_error("serverFail", ""))];
577    }
578    if out.iter().all(|(n, _)| n != "error") {
579        log_decision(ctx, id, &call.name, "model_answer");
580    }
581    out
582}
netget/server/jmap/mod.rs

netget/server/jmap/
mod.rs
