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