Federated via graft
12 kB
363 lines
1package activitypub
2
3import (
4 "crypto/rsa"
5 "encoding/json"
6 "io"
7 "log/slog"
8 "net/http"
9 "strconv"
10 "strings"
11 "time"
12
13 "graft/internal/state"
14)
15
16const (
17 activityContentType = "application/activity+json"
18 outboxLimit = 50
19 maxInboxBody = 1 << 20 // 1 MiB, generous for a Follow/Undo/reply
20)
21
22// Handler serves the ActivityPub surface for every series: WebFinger,
23// actor, outbox, individual notes, followers, and an inbox that accepts
24// Follow/Undo{Follow} (replies land here too but are ignored until the
25// comment bridge, M3).
26type Handler struct {
27 store *state.Store
28 host string
29 log *slog.Logger
30 // repoURL and known are supplied by the caller (cmd/sync/main.go),
31 // which already knows every series' topology — avoids this package
32 // needing to know about config.Config or sync.RepoSyncer at all.
33 repoURL func(series string) string
34 known func(series string) bool
35 client *http.Client
36}
37
38func NewHandler(store *state.Store, host string, log *slog.Logger, repoURL func(series string) string, known func(series string) bool) *Handler {
39 return &Handler{
40 store: store,
41 host: host,
42 log: log,
43 repoURL: repoURL,
44 known: known,
45 client: &http.Client{Timeout: 15 * time.Second},
46 }
47}
48
49// Register mounts every ActivityPub route on mux. Call before mounting any
50// catch-all "/" handler — Go's ServeMux resolves the more specific
51// patterns registered here first regardless of registration order, but
52// keeping this explicit avoids surprises.
53func (h *Handler) Register(mux *http.ServeMux) {
54 mux.HandleFunc("/.well-known/webfinger", h.webfinger)
55 mux.HandleFunc("/actors/", h.actorRouter)
56}
57
58func (h *Handler) webfinger(w http.ResponseWriter, r *http.Request) {
59 series, ok := parseAcct(r.URL.Query().Get("resource"), h.host)
60 if !ok || !h.known(series) {
61 http.NotFound(w, r)
62 return
63 }
64 resp := map[string]any{
65 "subject": "acct:" + series + "@" + h.host,
66 "links": []map[string]string{
67 {"rel": "self", "type": activityContentType, "href": ActorURI(h.host, series)},
68 },
69 }
70 writeJSON(w, "application/jrd+json", resp)
71}
72
73// parseAcct extracts the series name from a WebFinger "acct:series@host"
74// resource query, rejecting anything not addressed to our own host.
75func parseAcct(resource, host string) (series string, ok bool) {
76 resource = strings.TrimPrefix(resource, "acct:")
77 at := strings.LastIndex(resource, "@")
78 if at < 0 {
79 return "", false
80 }
81 series, resourceHost := resource[:at], resource[at+1:]
82 if series == "" || resourceHost != host {
83 return "", false
84 }
85 return series, true
86}
87
88func (h *Handler) actorRouter(w http.ResponseWriter, r *http.Request) {
89 parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/actors/"), "/")
90 series := parts[0]
91 if series == "" || !h.known(series) {
92 http.NotFound(w, r)
93 return
94 }
95 switch {
96 case len(parts) == 1:
97 h.actor(w, series)
98 case len(parts) == 2 && parts[1] == "outbox":
99 h.outbox(w, series)
100 case len(parts) == 2 && parts[1] == "followers":
101 h.followers(w, series)
102 case len(parts) == 2 && parts[1] == "inbox":
103 h.inbox(w, r, series)
104 case len(parts) == 3 && parts[1] == "notes":
105 h.note(w, r, series, parts[2])
106 default:
107 http.NotFound(w, r)
108 }
109}
110
111// actorKeys returns a series' stored keypair, generating and persisting
112// one the first time it's asked for.
113func (h *Handler) actorKeys(series string) (privPEM, pubPEM string, err error) {
114 k, err := h.store.ActorKeys(series)
115 if err != nil {
116 return "", "", err
117 }
118 if k != nil {
119 return k.PrivateKeyPEM, k.PublicKeyPEM, nil
120 }
121 priv, pub, err := GenerateKeyPair()
122 if err != nil {
123 return "", "", err
124 }
125 if err := h.store.SaveActorKeys(series, state.ActorKeys{PrivateKeyPEM: priv, PublicKeyPEM: pub}); err != nil {
126 return "", "", err
127 }
128 // Re-read: SaveActorKeys is INSERT OR IGNORE, so a concurrent request
129 // generating the same series' first keypair may have won the race —
130 // this returns whichever copy actually landed in the database.
131 k, err = h.store.ActorKeys(series)
132 if err != nil {
133 return "", "", err
134 }
135 return k.PrivateKeyPEM, k.PublicKeyPEM, nil
136}
137
138func (h *Handler) actor(w http.ResponseWriter, series string) {
139 _, pub, err := h.actorKeys(series)
140 if err != nil {
141 http.Error(w, err.Error(), http.StatusInternalServerError)
142 return
143 }
144 writeJSON(w, activityContentType, BuildActor(h.host, series, h.repoURL(series), pub))
145}
146
147func (h *Handler) outbox(w http.ResponseWriter, series string) {
148 entries, err := h.store.ActivityForSeries(series, outboxLimit)
149 if err != nil {
150 http.Error(w, err.Error(), http.StatusInternalServerError)
151 return
152 }
153 writeJSON(w, activityContentType, BuildOutbox(h.host, series, entries))
154}
155
156func (h *Handler) note(w http.ResponseWriter, r *http.Request, series, idStr string) {
157 id, err := strconv.ParseInt(idStr, 10, 64)
158 if err != nil {
159 http.NotFound(w, r)
160 return
161 }
162 e, err := h.store.ActivityByID(id)
163 if err != nil {
164 http.Error(w, err.Error(), http.StatusInternalServerError)
165 return
166 }
167 if e == nil || e.Series != series {
168 http.NotFound(w, r)
169 return
170 }
171 writeJSON(w, activityContentType, BuildNote(h.host, series, *e))
172}
173
174func (h *Handler) followers(w http.ResponseWriter, series string) {
175 n, err := h.store.FollowerCount(series)
176 if err != nil {
177 http.Error(w, err.Error(), http.StatusInternalServerError)
178 return
179 }
180 writeJSON(w, activityContentType, OrderedCollection{
181 Context: "https://www.w3.org/ns/activitystreams",
182 ID: ActorURI(h.host, series) + "/followers",
183 Type: "OrderedCollection",
184 TotalItems: n,
185 OrderedItems: []Create{},
186 })
187}
188
189// actorPrivateKey returns a series' actor's private key, generating a
190// keypair first if one doesn't exist yet.
191func (h *Handler) actorPrivateKey(series string) (*rsa.PrivateKey, error) {
192 privPEM, _, err := h.actorKeys(series)
193 if err != nil {
194 return nil, err
195 }
196 return ParsePrivateKey(privPEM)
197}
198
199// inboxActivity is the subset of an inbound activity's fields this handler
200// reads. Object is kept raw so, for a Follow, it can be embedded verbatim
201// as the Accept's object — and so an Undo's nested Follow can be decoded
202// separately.
203type inboxActivity struct {
204 Type string `json:"type"`
205 Actor string `json:"actor"`
206 ID string `json:"id"`
207 Object json.RawMessage `json:"object"`
208}
209
210func (h *Handler) inbox(w http.ResponseWriter, r *http.Request, series string) {
211 body, err := io.ReadAll(io.LimitReader(r.Body, maxInboxBody))
212 if err != nil {
213 http.Error(w, "read body", http.StatusBadRequest)
214 return
215 }
216
217 var act inboxActivity
218 if err := json.Unmarshal(body, &act); err != nil || act.Actor == "" {
219 http.Error(w, "invalid activity", http.StatusBadRequest)
220 return
221 }
222
223 remoteActor, err := FetchActor(h.client, act.Actor)
224 if err != nil {
225 h.log.Error("ap inbox: resolve actor", "series", series, "actor", act.Actor, "err", err)
226 http.Error(w, "could not resolve actor", http.StatusBadGateway)
227 return
228 }
229 pub, err := ParsePublicKey(remoteActor.PublicKey.PublicKeyPem)
230 if err != nil {
231 http.Error(w, "invalid actor public key", http.StatusBadGateway)
232 return
233 }
234 if err := VerifyRequest(r, body, pub); err != nil {
235 h.log.Error("ap inbox: signature verification failed", "series", series, "actor", act.Actor, "err", err)
236 http.Error(w, "signature verification failed", http.StatusUnauthorized)
237 return
238 }
239
240 switch act.Type {
241 case "Follow":
242 if err := h.store.SaveFollower(series, act.Actor, remoteActor.Inbox); err != nil {
243 http.Error(w, err.Error(), http.StatusInternalServerError)
244 return
245 }
246 if err := h.sendAccept(series, remoteActor.Inbox, body); err != nil {
247 // The follow is already recorded; a failed Accept delivery
248 // shouldn't undo that — the remote side will likely retry.
249 h.log.Error("ap inbox: send accept", "series", series, "actor", act.Actor, "err", err)
250 }
251 case "Undo":
252 var inner inboxActivity
253 if err := json.Unmarshal(act.Object, &inner); err == nil && inner.Type == "Follow" {
254 if err := h.store.RemoveFollower(series, act.Actor); err != nil {
255 h.log.Error("ap inbox: remove follower", "series", series, "actor", act.Actor, "err", err)
256 }
257 }
258 default:
259 // Replies (Create{Note}) land here too but aren't handled until
260 // the comment bridge (v2 M3) — accepted and ignored for now.
261 }
262
263 w.WriteHeader(http.StatusAccepted)
264}
265
266// sendAccept replies to a Follow with a signed Accept, embedding the
267// original Follow activity verbatim as required by the spec.
268func (h *Handler) sendAccept(series, targetInbox string, followBody []byte) error {
269 priv, err := h.actorPrivateKey(series)
270 if err != nil {
271 return err
272 }
273 actor := ActorURI(h.host, series)
274 accept := map[string]any{
275 "@context": "https://www.w3.org/ns/activitystreams",
276 "id": actor + "/accepts/" + strconv.FormatInt(time.Now().UnixNano(), 10),
277 "type": "Accept",
278 "actor": actor,
279 "object": json.RawMessage(followBody),
280 }
281 body, err := json.Marshal(accept)
282 if err != nil {
283 return err
284 }
285 return PostSigned(h.client, targetInbox, actor+"#main-key", priv, body)
286}
287
288// DeliverNewActivity signs and POSTs a Create{Note} to every follower's
289// inbox for each series' activity_log rows logged since the last
290// delivery. Call it once per sync pass, after runAll has logged whatever
291// happened this pass. Best-effort: a delivery failure to one follower is
292// logged and skipped, never blocks the sync loop or other followers.
293func (h *Handler) DeliverNewActivity() {
294 seriesList, err := h.store.SeriesWithFollowers()
295 if err != nil {
296 h.log.Error("ap delivery: list series with followers", "err", err)
297 return
298 }
299
300 for _, series := range seriesList {
301 cursor, err := h.store.DeliveryCursor(series)
302 if err != nil {
303 h.log.Error("ap delivery: read cursor", "series", series, "err", err)
304 continue
305 }
306 entries, err := h.store.NewActivityForSeries(series, cursor)
307 if err != nil {
308 h.log.Error("ap delivery: list new activity", "series", series, "err", err)
309 continue
310 }
311 if len(entries) == 0 {
312 continue
313 }
314
315 followers, err := h.store.Followers(series)
316 if err != nil {
317 h.log.Error("ap delivery: list followers", "series", series, "err", err)
318 continue
319 }
320 priv, err := h.actorPrivateKey(series)
321 if err != nil {
322 h.log.Error("ap delivery: load actor key", "series", series, "err", err)
323 continue
324 }
325 keyID := ActorURI(h.host, series) + "#main-key"
326
327 maxID := cursor
328 for _, e := range entries {
329 note := BuildNote(h.host, series, e)
330 create := Create{
331 Context: "https://www.w3.org/ns/activitystreams",
332 ID: note.ID + "/activity",
333 Type: "Create",
334 Actor: ActorURI(h.host, series),
335 Published: note.Published,
336 To: []string{publicAudience},
337 Object: note,
338 }
339 body, err := json.Marshal(create)
340 if err != nil {
341 h.log.Error("ap delivery: marshal activity", "series", series, "id", e.ID, "err", err)
342 continue
343 }
344 for _, f := range followers {
345 if err := PostSigned(h.client, f.InboxURI, keyID, priv, body); err != nil {
346 h.log.Error("ap delivery: post to follower", "series", series, "follower", f.ActorURI, "err", err)
347 }
348 }
349 if e.ID > maxID {
350 maxID = e.ID
351 }
352 }
353 if err := h.store.SetDeliveryCursor(series, maxID); err != nil {
354 h.log.Error("ap delivery: set cursor", "series", series, "err", err)
355 }
356 }
357}
358
359func writeJSON(w http.ResponseWriter, contentType string, v any) {
360 w.Header().Set("Content-Type", contentType)
361 w.Header().Set("Cache-Control", "no-store")
362 json.NewEncoder(w).Encode(v)
363}