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