netget/server/activitypub/
http.rs

1//! The HTTP surface an ActivityPub instance serves, for either role: WebFinger, NodeInfo,
2//! actor documents, followers/following/outbox collections, objects, and the inboxes —
3//! where every POST is verified (HTTP Signature, Digest, Date, actor = signer) before it
4//! is queued for whoever answers it.
5use super::instance::{Inbound, Instance, MAX_DOCUMENT};
6use bytes::Bytes;
7use http_body_util::{BodyExt, Full, Limited};
8use hyper::{body::Incoming, header::CONTENT_TYPE, Method, Request, Response, StatusCode};
9use serde_json::{json, Value};
10use std::collections::HashMap;
11use std::sync::Arc;
12use tokio::sync::mpsc;
13
14pub type Reply = Response<Full<Bytes>>;
15pub const ACTIVITY_JSON: &str = "application/activity+json";
16
17fn reply(status: u16, content_type: &str, body: Vec<u8>) -> Reply {
18    let mut r = Response::new(Full::new(Bytes::from(body)));
19    *r.status_mut() = StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
20    if let Ok(v) = hyper::header::HeaderValue::from_str(content_type) {
21        r.headers_mut().insert(CONTENT_TYPE, v);
22    }
23    r
24}
25
26fn json_reply(status: u16, content_type: &str, v: &Value) -> Reply {
27    reply(
28        status,
29        content_type,
30        serde_json::to_vec(v).unwrap_or_default(),
31    )
32}
33
34fn error(status: u16, message: &str) -> Reply {
35    json_reply(status, "application/json", &json!({"error": message}))
36}
37
38/// What happened to a request, for the caller's log.
39pub enum Outcome {
40    Served,
41    Queued,
42    Refused(String),
43}
44
45pub async fn handle(
46    instance: &Arc<Instance>,
47    request: Request<Incoming>,
48    inbound: &mpsc::Sender<Inbound>,
49) -> (Reply, Outcome) {
50    let (parts, body) = request.into_parts();
51    let path = parts.uri.path().to_string();
52    let query = parts.uri.query().unwrap_or_default().to_string();
53    let segs: Vec<&str> = path.trim_matches('/').split('/').collect();
54    let served = |r: Reply| (r, Outcome::Served);
55    match (&parts.method, segs.as_slice()) {
56        (&Method::GET, [".well-known", "webfinger"]) => {
57            let resource = query
58                .split('&')
59                .find_map(|p| p.strip_prefix("resource="))
60                .map(|r| {
61                    urlencoding::decode(r)
62                        .map(|c| c.into_owned())
63                        .unwrap_or_default()
64                })
65                .unwrap_or_default();
66            served(match instance.webfinger(&resource) {
67                Some(j) => json_reply(200, "application/jrd+json", &j),
68                None => error(404, "no such account"),
69            })
70        }
71        (&Method::GET, [".well-known", "nodeinfo"]) => served(json_reply(
72            200,
73            "application/json",
74            &json!({"links": [{"rel": "http://nodeinfo.diaspora.software/ns/schema/2.1",
75                              "href": format!("{}/nodeinfo/2.1", instance.base)}]}),
76        )),
77        (&Method::GET, ["nodeinfo", "2.1"]) => {
78            let users = instance
79                .actors
80                .lock()
81                .unwrap_or_else(|e| e.into_inner())
82                .len();
83            served(json_reply(
84                200,
85                "application/json; profile=\"http://nodeinfo.diaspora.software/ns/schema/2.1#\"",
86                &json!({"version": "2.1",
87                        "software": {"name": "netget", "version": env!("CARGO_PKG_VERSION")},
88                        "protocols": ["activitypub"], "services": {"inbound": [], "outbound": []},
89                        "openRegistrations": false,
90                        "usage": {"users": {"total": users}}, "metadata": {}}),
91            ))
92        }
93        (&Method::GET, ["users", name]) => served(match instance.actor_document(name) {
94            Some(d) => json_reply(200, ACTIVITY_JSON, &d),
95            None => error(404, "no such actor"),
96        }),
97        (&Method::GET, ["users", name, which @ ("followers" | "following" | "outbox")]) => {
98            served(match instance.collection(name, which) {
99                Some(c) => json_reply(200, ACTIVITY_JSON, &c),
100                None => error(404, "no such actor"),
101            })
102        }
103        (&Method::GET, ["notes", _]) => {
104            let id = format!("{}{}", instance.base, path);
105            served(match instance.object(&id) {
106                Some(o) => json_reply(200, ACTIVITY_JSON, &o),
107                None => error(404, "no such object"),
108            })
109        }
110        (&Method::POST, ["users", name, "inbox"]) | (&Method::POST, [name @ "inbox"]) => {
111            let to = (*name != "inbox").then(|| name.to_string());
112            if let Some(n) = &to {
113                if !instance.has_actor(n) {
114                    return served(error(404, "no such actor"));
115                }
116            }
117            let bytes = match Limited::new(body, MAX_DOCUMENT).collect().await {
118                Ok(b) => b.to_bytes(),
119                Err(_) => {
120                    return (
121                        error(413, "activity too large"),
122                        Outcome::Refused("too large".into()),
123                    )
124                }
125            };
126            let headers: HashMap<String, String> = parts
127                .headers
128                .iter()
129                .filter_map(|(k, v)| {
130                    v.to_str()
131                        .ok()
132                        .map(|v| (k.as_str().to_string(), v.to_string()))
133                })
134                .collect();
135            match instance.verify_inbound(to, &path, &headers, &bytes).await {
136                Err(e) => {
137                    let why = format!("{e:#}");
138                    (error(401, &why), Outcome::Refused(why))
139                }
140                Ok(inb) => match inbound.try_send(inb) {
141                    Ok(()) => (reply(202, "text/plain", Vec::new()), Outcome::Queued),
142                    Err(_) => (
143                        error(503, "inbox busy"),
144                        Outcome::Refused("queue full".into()),
145                    ),
146                },
147            }
148        }
149        _ => served(error(404, "not found")),
150    }
151}
netget/server/activitypub/http.rs

netget/server/activitypub/
http.rs
