From 64da36aa208cc46d88a1a732a3700fd07d14c705 Mon Sep 17 00:00:00 2001 From: Flomp Date: Thu, 18 Jun 2026 16:36:57 +0200 Subject: [PATCH] fix: harden ActivityPub federation and fix N+1 follower fanout (#1056) * initial commit * improve activitypub recipient query * small fixes --------- Co-authored-by: Christian Beutel <> Co-authored-by: slothful-vassal <89943360+slothful-vassal@users.noreply.github.com> --- db/federation/activity.go | 32 +++++++++++++++++-- db/federation/actor.go | 33 +++++++++++++++----- db/federation/announce.go | 6 ++-- db/federation/create.go | 66 +++++++++++++-------------------------- db/federation/delete.go | 39 +++++------------------ db/hooks/list.go | 2 +- 6 files changed, 86 insertions(+), 92 deletions(-) diff --git a/db/federation/activity.go b/db/federation/activity.go index af359b02..bc2030a7 100644 --- a/db/federation/activity.go +++ b/db/federation/activity.go @@ -18,11 +18,40 @@ import ( "github.com/go-ap/jsonld" "github.com/go-fed/httpsig" + "github.com/pocketbase/dbx" "github.com/pocketbase/pocketbase/core" "github.com/pocketbase/pocketbase/tools/security" "golang.org/x/sync/semaphore" ) +var httpClient = &http.Client{Timeout: 10 * time.Second} + +// followerInboxes returns inbox URLs for all accepted followers of actorId +// in a single JOIN query instead of one query per follower. +func followerInboxes(app core.App, actorId string) ([]string, error) { + rows, err := app.DB(). + Select("aa.inbox"). + From("follows f"). + InnerJoin("activitypub_actors aa", dbx.NewExp("f.follower = aa.id")). + Where(dbx.NewExp("f.followee = {:followee} AND f.status = 'accepted' AND aa.inbox != ''", + dbx.Params{"followee": actorId})). + Rows() + if err != nil { + return nil, err + } + defer rows.Close() + + var inboxes []string + for rows.Next() { + var inbox string + if err := rows.Scan(&inbox); err != nil { + return nil, err + } + inboxes = append(inboxes, inbox) + } + return inboxes, rows.Err() +} + func PostActivity(app core.App, actor *core.Record, activity *pub.Activity, recipients []string) error { go func() { defer func() { @@ -67,7 +96,6 @@ func PostActivity(app core.App, actor *core.Record, activity *pub.Activity, reci } pubID := actor.GetString("iri") + "#main-key" - client := &http.Client{} sem := semaphore.NewWeighted(5) slices.Sort(recipients) @@ -105,7 +133,7 @@ func PostActivity(app core.App, actor *core.Record, activity *pub.Activity, reci return } - resp, err := client.Do(req) + resp, err := httpClient.Do(req) if err != nil { app.Logger().Error(fmt.Sprintf("Error sending to inbox %s: %s", inbox, err)) return diff --git a/db/federation/actor.go b/db/federation/actor.go index 26a05ce3..d75d243e 100644 --- a/db/federation/actor.go +++ b/db/federation/actor.go @@ -194,12 +194,21 @@ func assembleActor(app core.App, ctx context.Context, dbActor *core.Record, incl dbActor.Set("last_fetched", time.Now()) + // an empty privacy field is the default for users who never touched + // their privacy settings and is treated as public. A non-empty but + // corrupt value fails closed (private) so a broken setting can't + // silently expose a profile. privacy := settings.GetString("privacy") - result := make(map[string]interface{}) - json.Unmarshal([]byte(privacy), &result) - - // check that it's not our own profile - private = result["account"] == "private" && dbActor.Id != strings.TrimPrefix(ctx.Value("actor").(string), "actor:") + if privacy != "" { + result := make(map[string]interface{}) + if err := json.Unmarshal([]byte(privacy), &result); err != nil { + private = true + } else { + // check that it's not our own profile + actorVal, _ := ctx.Value("actor").(string) + private = result["account"] == "private" && dbActor.Id != strings.TrimPrefix(actorVal, "actor:") + } + } } else { @@ -279,6 +288,9 @@ func fetchRemoteActor(app core.App, ctx context.Context, iri string, includeFoll client := util.SafeHTTPClient() req, err := http.NewRequestWithContext(ctx, "GET", iri, nil) + if err != nil { + return nil, nil, nil, err + } headers := map[string]string{ "Accept": "application/ld+json", @@ -291,7 +303,8 @@ func fetchRemoteActor(app core.App, ctx context.Context, iri string, includeFoll req.Header.Add(k, v) } - userActorId := strings.TrimPrefix(ctx.Value("actor").(string), "actor:") + actorVal, _ := ctx.Value("actor").(string) + userActorId := strings.TrimPrefix(actorVal, "actor:") userActor, err := app.FindRecordById("activitypub_actors", userActorId) if userActor != nil && userActor.GetString("private_key") != "" { dbPrivateKey := userActor.GetString("private_key") @@ -332,7 +345,7 @@ func fetchRemoteActor(app core.App, ctx context.Context, iri string, includeFoll defer resp.Body.Close() var pubActor pub.Actor - if err := json.NewDecoder(resp.Body).Decode(&pubActor); err != nil { + if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&pubActor); err != nil { return nil, nil, nil, err } @@ -365,6 +378,9 @@ func FetchCollection(app core.App, ctx context.Context, collectionURL string) (* } req, err := http.NewRequestWithContext(ctx, "GET", collectionURL, nil) + if err != nil { + return nil, err + } headers := map[string]string{ "Accept": "application/ld+json", @@ -376,7 +392,8 @@ func FetchCollection(app core.App, ctx context.Context, collectionURL string) (* for k, v := range headers { req.Header.Add(k, v) } - userActorId := strings.TrimPrefix(ctx.Value("actor").(string), "actor:") + actorVal, _ := ctx.Value("actor").(string) + userActorId := strings.TrimPrefix(actorVal, "actor:") userActor, err := app.FindRecordById("activitypub_actors", userActorId) if userActor != nil && userActor.GetString("private_key") != "" { dbPrivateKey := userActor.GetString("private_key") diff --git a/db/federation/announce.go b/db/federation/announce.go index 2ca9a9f3..46c39e18 100644 --- a/db/federation/announce.go +++ b/db/federation/announce.go @@ -113,14 +113,12 @@ func ProcessAnnounceActivity(app core.App, actor *core.Record, activity pub.Acti object := activity.Object.GetID().String() if strings.Contains(object, "/api/v1/trail") { - processTrailAnnounceActivity(app, actor, activity) - + return processTrailAnnounceActivity(app, actor, activity) } else if strings.Contains(object, "/api/v1/list") { - processListAnnounceActivity(app, actor, activity) + return processListAnnounceActivity(app, actor, activity) } else { return fmt.Errorf("unknown announce type") } - return nil } diff --git a/db/federation/create.go b/db/federation/create.go index 17405fe2..0cf686e9 100644 --- a/db/federation/create.go +++ b/db/federation/create.go @@ -12,7 +12,6 @@ import ( "pocketbase/util" pub "github.com/go-ap/activitypub" - "github.com/pocketbase/dbx" "github.com/pocketbase/pocketbase/core" "github.com/pocketbase/pocketbase/tools/filesystem" "github.com/pocketbase/pocketbase/tools/security" @@ -89,19 +88,11 @@ func CreateTrailActivity(app core.App, ctx context.Context, trail *core.Record, return err } - follows, err := app.FindRecordsByFilter("follows", "followee={:followee}&&status='accepted'", "", -1, 0, dbx.Params{"followee": trailAuthor.Id}) + inboxes, err := followerInboxes(app, trailAuthor.Id) if err != nil { return err } - - recipients := mentions - for _, f := range follows { - follower, err := app.FindRecordById("activitypub_actors", f.GetString("follower")) - if err != nil { - return err - } - recipients = append(recipients, follower.GetString("inbox")) - } + recipients := append(mentions, inboxes...) return PostActivity(app, trailAuthor, activity, recipients) } @@ -264,15 +255,14 @@ func CreateSummitLogActivity(app core.App, ctx context.Context, summitLog *core. gpx = fmt.Sprintf("%s/api/v1/files/summit_logs/%s/%s", origin, summitLog.Id, summitLog.GetString("gpx")) } - attachments := make(pub.ItemCollection, max(len(photos), 2)) + attachments := make(pub.ItemCollection, 0, len(photos)+1) for i := range len(photos) { iri := fmt.Sprintf("%s/api/v1/files/summit_logs/%s/%s", origin, summitLog.Id, photos[i]) - - attachments[i] = pub.Document{ + attachments.Append(pub.Document{ Type: pub.ImageType, MediaType: "image/jpeg", URL: pub.IRI(iri), - } + }) } if gpx != "" { attachments.Append(pub.Document{ @@ -328,20 +318,11 @@ func CreateSummitLogActivity(app core.App, ctx context.Context, summitLog *core. activity.CC = cc activity.Published = time.Now() - follows, err := app.FindRecordsByFilter("follows", "followee={:followee}&&status='accepted'", "", -1, 0, dbx.Params{"followee": summitLogAuthor.Id}) + inboxes, err := followerInboxes(app, summitLogAuthor.Id) if err != nil { return err } - - recipients := mentions - - for _, f := range follows { - follower, err := app.FindRecordById("activitypub_actors", f.GetString("follower")) - if err != nil { - return err - } - recipients = append(recipients, follower.GetString("inbox")) - } + recipients := append(mentions, inboxes...) if summitLogAuthor.Id != summitLogTrailAuthor.Id { recipients = append(recipients, summitLogTrailAuthor.GetString("inbox")) @@ -405,20 +386,11 @@ func CreateListActivity(app core.App, list *core.Record, typ pub.ActivityVocabul return err } - follows, err := app.FindRecordsByFilter("follows", "followee={:followee}&&status='accepted'", "", -1, 0, dbx.Params{"followee": listAuthor.Id}) + recipients, err := followerInboxes(app, listAuthor.Id) if err != nil { return err } - recipients := []string{} - for _, f := range follows { - follower, err := app.FindRecordById("activitypub_actors", f.GetString("follower")) - if err != nil { - return err - } - recipients = append(recipients, follower.GetString("inbox")) - } - err = PostActivity(app, listAuthor, activity, recipients) if err != nil { return err @@ -469,7 +441,10 @@ func processCreateOrUpdateTrailActivity(activity pub.Activity, app core.App, act return err } - trailObject, _ := pub.ToObject(activity.Object) + trailObject, err := pub.ToObject(activity.Object) + if err != nil { + return err + } for _, t := range trailObject.Tag { if t.GetType() == pub.MentionType { @@ -487,11 +462,11 @@ func processCreateOrUpdateTrailActivity(activity pub.Activity, app core.App, act Seen: false, Author: actor.Id, } - return util.SendNotification(app, notification, mentionedActor) + util.SendNotification(app, notification, mentionedActor) } } - return err + return nil } func processCreateOrUpdateCommentActivity(activity pub.Activity, app core.App, actor *core.Record) error { @@ -578,7 +553,7 @@ func processCreateOrUpdateCommentActivity(activity pub.Activity, app core.App, a Seen: false, Author: actor.Id, } - return util.SendNotification(app, notification, mentionedActor) + util.SendNotification(app, notification, mentionedActor) } } if activity.Type == pub.CreateType { @@ -629,6 +604,11 @@ func processCreateOrUpdateSummitLogActivity(activity pub.Activity, app core.App, return err } + // no need to do anything else if the actor is local + if actor.GetBool("is_local") { + return nil + } + newSummitLog := false record, err := app.FindFirstRecordByData("summit_logs", "iri", logObject.ID.String()) if err != nil { @@ -644,10 +624,6 @@ func processCreateOrUpdateSummitLogActivity(activity pub.Activity, app core.App, return err } } - // no need to do anything else if the actor is local - if actor.GetBool("is_local") { - return nil - } var distance, duration, elevation_gain, elevation_loss float64 tags, err := pub.ToItemCollection(logObject.Tag) @@ -752,7 +728,7 @@ func processCreateOrUpdateSummitLogActivity(activity pub.Activity, app core.App, Seen: false, Author: actor.Id, } - return util.SendNotification(app, notification, mentionedActor) + util.SendNotification(app, notification, mentionedActor) } } diff --git a/db/federation/delete.go b/db/federation/delete.go index 0deba908..26cbd98a 100644 --- a/db/federation/delete.go +++ b/db/federation/delete.go @@ -9,7 +9,6 @@ import ( pub "github.com/go-ap/activitypub" "github.com/meilisearch/meilisearch-go" - "github.com/pocketbase/dbx" "github.com/pocketbase/pocketbase/core" "github.com/pocketbase/pocketbase/tools/security" ) @@ -29,6 +28,10 @@ func CreateTrailDeleteActivity(app core.App, r *core.Record) error { return err } + if !author.GetBool("is_local") { + return nil + } + collection, err := app.FindCollectionByNameOrId("activitypub_activities") if err != nil { return err @@ -62,20 +65,11 @@ func CreateTrailDeleteActivity(app core.App, r *core.Record) error { activity.CC = pub.ItemCollection{pub.IRI(cc)} activity.Published = time.Now() - follows, err := app.FindRecordsByFilter("follows", "followee={:followee}&&status='accepted'", "", -1, 0, dbx.Params{"followee": author.Id}) + recipients, err := followerInboxes(app, author.Id) if err != nil { return err } - recipients := []string{} - for _, f := range follows { - follower, err := app.FindRecordById("activitypub_actors", f.GetString("follower")) - if err != nil { - return err - } - recipients = append(recipients, follower.GetString("inbox")) - } - return PostActivity(app, author, activity, recipients) } @@ -185,21 +179,11 @@ func CreateSummitLogDeleteActivity(app core.App, r *core.Record) error { activity.CC = cc activity.Published = time.Now() - follows, err := app.FindRecordsByFilter("follows", "followee={:followee}&&status='accepted'", "", -1, 0, dbx.Params{"followee": author.Id}) + recipients, err := followerInboxes(app, author.Id) if err != nil { return err } - recipients := []string{} - - for _, f := range follows { - follower, err := app.FindRecordById("activitypub_actors", f.GetString("follower")) - if err != nil { - return err - } - recipients = append(recipients, follower.GetString("inbox")) - } - if author.Id != summitLogTrailAuthor.Id { recipients = append(recipients, summitLogTrailAuthor.GetString("inbox")) } @@ -256,20 +240,11 @@ func CreateListDeleteActivity(app core.App, r *core.Record) error { activity.CC = pub.ItemCollection{pub.IRI(cc)} activity.Published = time.Now() - follows, err := app.FindRecordsByFilter("follows", "followee={:followee}&&status='accepted'", "", -1, 0, dbx.Params{"followee": author.Id}) + recipients, err := followerInboxes(app, author.Id) if err != nil { return err } - recipients := []string{} - for _, f := range follows { - follower, err := app.FindRecordById("activitypub_actors", f.GetString("follower")) - if err != nil { - return err - } - recipients = append(recipients, follower.GetString("inbox")) - } - err = PostActivity(app, author, activity, recipients) if err != nil { return err diff --git a/db/hooks/list.go b/db/hooks/list.go index fadce824..582ceb66 100644 --- a/db/hooks/list.go +++ b/db/hooks/list.go @@ -87,7 +87,7 @@ func UpdateListHandler(client meilisearch.ServiceManager) func(e *core.RecordEve return err } - err = federation.CreateListActivity(e.App, e.Record, pub.CreateType) + err = federation.CreateListActivity(e.App, e.Record, pub.UpdateType) if err != nil { return err }