From 2ae8f52b4b43ab8755e7058f36437192af566c7c Mon Sep 17 00:00:00 2001 From: Alan Johnson Date: Mon, 3 Aug 2026 16:40:18 -0400 Subject: [PATCH] Wrap unsigned public activities for relay subscribers --- api/fep_ae0c_characterization_test.go | 56 +- api/relay_activity_wrapper_test.go | 726 ++++++++++++++++++++++++++ api/resolver.go | 201 ++++++- testdata/fep-ae0c/cases.json | 14 +- 4 files changed, 961 insertions(+), 36 deletions(-) create mode 100644 api/relay_activity_wrapper_test.go diff --git a/api/fep_ae0c_characterization_test.go b/api/fep_ae0c_characterization_test.go index fce335e..f3487a5 100644 --- a/api/fep_ae0c_characterization_test.go +++ b/api/fep_ae0c_characterization_test.go @@ -361,7 +361,6 @@ func TestFEPAE0CTraditionalPublicAudienceFixtures(t *testing.T) { ) }) addFEPTraditionalReceiver(t, receiverDomain) - activity, body := fixtureActivity(t, identifier) actor := actorForFixtureActivity(t, activity, "Person") recorder := postFEPAE0CFixture(t, activity, actor, body) @@ -373,9 +372,36 @@ func TestFEPAE0CTraditionalPublicAudienceFixtures(t *testing.T) { http.StatusAccepted, ) } - waitForFEPStoredBody(t, body) + + objectID, err := activity.UnwrapInnerObjectId() + if err != nil { + t.Fatalf("fixture object ID: %v", err) + } + _, targetCount, found := waitForRelayWrapper( + t, + objectID, + ) + if !found { + t.Fatal( + "unsigned fixture did not produce relay-authored Announce", + ) + } + if targetCount < 1 { + t.Fatalf( + "relay Announce target count = %d; want at least 1", + targetCount, + ) + } + if storedRelayBodyEquals(t, body) { + t.Fatal( + "unsigned fixture body must not be forwarded unchanged", + ) + } if !RelayState.IsPublisher(sourceDomain) { - t.Fatalf("source %s was not recorded as publisher", sourceDomain) + t.Fatalf( + "source %s was not recorded as publisher", + sourceDomain, + ) } }) } @@ -507,7 +533,6 @@ func TestFEPAE0COpenPublisherCreate(t *testing.T) { activity, body := fixtureActivity(t, "open-publisher-public-create") actor := actorForFixtureActivity(t, activity, "Person") recorder := postFEPAE0CFixture(t, activity, actor, body) - if recorder.Code != http.StatusAccepted { t.Fatalf( "status = %d; want %d", @@ -522,7 +547,28 @@ func TestFEPAE0COpenPublisherCreate(t *testing.T) { sourceDomain, ) } - waitForFEPStoredBody(t, body) + + objectID, err := activity.UnwrapInnerObjectId() + if err != nil { + t.Fatalf("fixture object ID: %v", err) + } + _, targetCount, found := waitForRelayWrapper(t, objectID) + if !found { + t.Fatal( + "unsigned open-publisher fixture did not produce relay Announce", + ) + } + if targetCount < 1 { + t.Fatalf( + "relay Announce target count = %d; want at least 1", + targetCount, + ) + } + if storedRelayBodyEquals(t, body) { + t.Fatal( + "unsigned open-publisher body must not be forwarded unchanged", + ) + } if !RelayState.IsPublisher(sourceDomain) { t.Fatalf("open publisher %s was not recorded", sourceDomain) } diff --git a/api/relay_activity_wrapper_test.go b/api/relay_activity_wrapper_test.go new file mode 100644 index 0000000..160a350 --- /dev/null +++ b/api/relay_activity_wrapper_test.go @@ -0,0 +1,726 @@ +// File: api/relay_activity_wrapper_test.go + +package api + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "testing" + "time" + + "github.com/thystra/Activity-Relay/models" +) + +func relayWrapperTestActor(sourceDomain string) *models.Actor { + return &models.Actor{ + ID: "https://" + sourceDomain + "/?author=0", + Type: "Group", + Inbox: "https://" + sourceDomain + "/inbox", + Endpoints: &models.Endpoints{ + SharedInbox: "https://" + sourceDomain + "/shared-inbox", + }, + } +} + +func relayWrapperTestActivity( + activityType string, + sourceDomain string, + objectID string, +) *models.Activity { + objectType := "Article" + if activityType == "Delete" { + objectType = "Tombstone" + } + + return &models.Activity{ + Context: []string{ + "https://www.w3.org/ns/activitystreams", + }, + ID: fmt.Sprintf( + "https://%s/activities/%s", + sourceDomain, + activityType, + ), + Actor: "https://" + sourceDomain + "/?author=0", + Type: activityType, + Object: map[string]interface{}{ + "id": objectID, + "type": objectType, + }, + To: []string{ + "https://www.w3.org/ns/activitystreams#Public", + }, + } +} + +func relayWrapperSignedBody( + t *testing.T, + activity *models.Activity, +) []byte { + t.Helper() + + plain, err := json.Marshal(activity) + if err != nil { + t.Fatalf("marshal activity: %v", err) + } + + var envelope map[string]interface{} + if err := json.Unmarshal(plain, &envelope); err != nil { + t.Fatalf("decode activity envelope: %v", err) + } + envelope["signature"] = map[string]interface{}{ + "type": "RsaSignature2017", + "creator": activity.Actor + "#main-key", + "created": "2026-08-03T20:00:00Z", + "signatureValue": "fixture-not-a-real-signature", + } + + signed, err := json.Marshal(envelope) + if err != nil { + t.Fatalf("marshal signed activity: %v", err) + } + return signed +} + +func relayStoredActivities( + t *testing.T, +) map[string][]byte { + t.Helper() + + ctx := context.Background() + keys, err := models.ScanKeys( + ctx, + RelayState.RedisClient, + "relay:activity:*", + ) + if err != nil { + t.Fatalf("scan relay activities: %v", err) + } + + result := make(map[string][]byte, len(keys)) + for _, key := range keys { + body, err := RelayState.RedisClient. + HGet(ctx, key, "body"). + Bytes() + if err != nil { + continue + } + result[key] = body + } + return result +} + +func waitForRelayWrapper( + t *testing.T, + objectID string, +) (models.Activity, int, bool) { + t.Helper() + + ctx := context.Background() + deadline := time.Now().Add(3 * time.Second) + + for time.Now().Before(deadline) { + for key, storedBody := range relayStoredActivities(t) { + var relayed models.Activity + if err := json.Unmarshal(storedBody, &relayed); err != nil { + continue + } + if relayed.Type != "Announce" || + relayed.Actor != RelayActor.ID || + relayed.Object != objectID { + continue + } + + targetCount, err := RelayState.RedisClient. + HGet(ctx, key, "remain_count"). + Int() + if err != nil { + t.Fatalf( + "read relay target count for %s: %v", + key, + err, + ) + } + return relayed, targetCount, true + } + + time.Sleep(25 * time.Millisecond) + } + + return models.Activity{}, 0, false +} + +func waitForStoredRelayBody( + t *testing.T, + expected []byte, +) (int, bool) { + t.Helper() + + ctx := context.Background() + deadline := time.Now().Add(3 * time.Second) + + for time.Now().Before(deadline) { + for key, storedBody := range relayStoredActivities(t) { + if !bytes.Equal(storedBody, expected) { + continue + } + + targetCount, err := RelayState.RedisClient. + HGet(ctx, key, "remain_count"). + Int() + if err != nil { + t.Fatalf( + "read relay target count for %s: %v", + key, + err, + ) + } + return targetCount, true + } + + time.Sleep(25 * time.Millisecond) + } + + return 0, false +} + +func storedRelayBodyEquals( + t *testing.T, + expected []byte, +) bool { + t.Helper() + + for _, storedBody := range relayStoredActivities(t) { + if bytes.Equal(storedBody, expected) { + return true + } + } + return false +} + +func storedRelayWrapperExists( + t *testing.T, + objectID string, +) bool { + t.Helper() + + for _, storedBody := range relayStoredActivities(t) { + var relayed models.Activity + if err := json.Unmarshal(storedBody, &relayed); err != nil { + continue + } + if relayed.Type == "Announce" && + relayed.Actor == RelayActor.ID && + relayed.Object == objectID { + return true + } + } + return false +} + +func assertRelayPublisherType( + t *testing.T, + sourceDomain string, + activityType string, +) { + t.Helper() + + publisher := RelayState.SelectPublisher(sourceDomain) + if publisher == nil { + t.Fatal("source was not recorded as a publisher") + } + if publisher.LastActivityType != activityType { + t.Fatalf( + "publisher activity type = %q; want %q", + publisher.LastActivityType, + activityType, + ) + } +} + +func TestExecuteRelayActivityWrapsUnsignedPublicActivitiesForAllReceivers( + t *testing.T, +) { + const ( + sourceDomain = "wordpress-wrapper.example" + traditionalDomain = "mastodon-wrapper.example" + followerDomain = "friendica-wrapper.example" + ) + + ctx := context.Background() + removeRelayActivityTestKeys(t) + RelayState.RedisClient.Del( + ctx, + "relay:publisher:"+sourceDomain, + ) + + RelayState.AddSubscriber(models.Subscriber{ + Domain: traditionalDomain, + InboxURL: "https://" + traditionalDomain + "/inbox", + }) + RelayState.AddFollower(models.Follower{ + Domain: followerDomain, + InboxURL: "https://" + followerDomain + "/inbox", + ActorID: "https://" + followerDomain + "/actor", + MutuallyFollow: true, + }) + + t.Cleanup(func() { + RelayState.DelSubscriber(traditionalDomain) + RelayState.DelFollower(followerDomain) + RelayState.RedisClient.Del( + ctx, + "relay:publisher:"+sourceDomain, + ) + removeRelayActivityTestKeys(t) + }) + + for _, activityType := range []string{ + "Create", + "Update", + "Delete", + "Move", + } { + t.Run(activityType, func(t *testing.T) { + removeRelayActivityTestKeys(t) + + objectID := fmt.Sprintf( + "https://%s/objects/%s", + sourceDomain, + activityType, + ) + activity := relayWrapperTestActivity( + activityType, + sourceDomain, + objectID, + ) + plainBody, err := json.Marshal(activity) + if err != nil { + t.Fatalf( + "marshal %s activity: %v", + activityType, + err, + ) + } + + if err := executeRelayActivity( + activity, + relayWrapperTestActor(sourceDomain), + plainBody, + ); err != nil { + t.Fatalf( + "executeRelayActivity(%s): %v", + activityType, + err, + ) + } + + relayed, targetCount, found := waitForRelayWrapper( + t, + objectID, + ) + if !found { + t.Fatalf( + "relay-authored Announce for %s was not queued", + activityType, + ) + } + if relayed.ID == "" { + t.Fatal("relay-authored Announce has no ID") + } + if !contains( + relayed.To, + RelayActor.Followers(), + ) { + t.Fatalf( + "relay Announce targets %v; want %q", + relayed.To, + RelayActor.Followers(), + ) + } + if targetCount != 2 { + t.Fatalf( + "relay Announce target count = %d; want 2", + targetCount, + ) + } + if storedRelayBodyEquals(t, plainBody) { + t.Fatalf( + "unsigned %s body must not be queued", + activityType, + ) + } + assertRelayPublisherType( + t, + sourceDomain, + activityType, + ) + }) + } +} + +func TestExecuteRelayActivityPreservesLDSignatureForSubscriberAndWrapsFollower( + t *testing.T, +) { + const ( + sourceDomain = "signed-publisher.example" + traditionalDomain = "signed-mastodon.example" + followerDomain = "signed-follower.example" + ) + + ctx := context.Background() + removeRelayActivityTestKeys(t) + RelayState.RedisClient.Del( + ctx, + "relay:publisher:"+sourceDomain, + ) + RelayState.AddSubscriber(models.Subscriber{ + Domain: traditionalDomain, + InboxURL: "https://" + traditionalDomain + "/inbox", + }) + RelayState.AddFollower(models.Follower{ + Domain: followerDomain, + InboxURL: "https://" + followerDomain + "/inbox", + ActorID: "https://" + followerDomain + "/actor", + MutuallyFollow: true, + }) + + t.Cleanup(func() { + RelayState.DelSubscriber(traditionalDomain) + RelayState.DelFollower(followerDomain) + RelayState.RedisClient.Del( + ctx, + "relay:publisher:"+sourceDomain, + ) + removeRelayActivityTestKeys(t) + }) + + objectID := "https://" + sourceDomain + "/objects/signed" + activity := relayWrapperTestActivity( + "Create", + sourceDomain, + objectID, + ) + signedBody := relayWrapperSignedBody(t, activity) + + if err := executeRelayActivity( + activity, + relayWrapperTestActor(sourceDomain), + signedBody, + ); err != nil { + t.Fatalf("executeRelayActivity: %v", err) + } + + directTargets, found := waitForStoredRelayBody(t, signedBody) + if !found { + t.Fatal("exact LD-signed body was not queued") + } + if directTargets != 1 { + t.Fatalf( + "direct LD-signed target count = %d; want 1", + directTargets, + ) + } + + _, wrapperTargets, found := waitForRelayWrapper(t, objectID) + if !found { + t.Fatal("follower relay Announce was not queued") + } + if wrapperTargets != 1 { + t.Fatalf( + "follower wrapper target count = %d; want 1", + wrapperTargets, + ) + } + assertRelayPublisherType(t, sourceDomain, "Create") +} + +func TestExecuteRelayActivityDeduplicatesUnsignedReceiverStylesAndExcludesSource( + t *testing.T, +) { + const ( + sourceDomain = "source-overlap.example" + targetDomain = "receiver-overlap.example" + ) + + ctx := context.Background() + removeRelayActivityTestKeys(t) + RelayState.RedisClient.Del( + ctx, + "relay:publisher:"+sourceDomain, + ) + + RelayState.AddSubscriber(models.Subscriber{ + Domain: sourceDomain, + InboxURL: "https://" + sourceDomain + "/inbox", + }) + RelayState.AddSubscriber(models.Subscriber{ + Domain: targetDomain, + InboxURL: "https://" + targetDomain + "/shared-inbox", + }) + RelayState.AddFollower(models.Follower{ + Domain: targetDomain, + InboxURL: "https://" + targetDomain + "/actor/inbox", + ActorID: "https://" + targetDomain + "/actor", + MutuallyFollow: true, + }) + + t.Cleanup(func() { + RelayState.DelSubscriber(sourceDomain) + RelayState.DelSubscriber(targetDomain) + RelayState.DelFollower(targetDomain) + RelayState.RedisClient.Del( + ctx, + "relay:publisher:"+sourceDomain, + ) + removeRelayActivityTestKeys(t) + }) + + objectID := "https://" + sourceDomain + "/objects/overlap" + activity := relayWrapperTestActivity( + "Create", + sourceDomain, + objectID, + ) + plainBody, err := json.Marshal(activity) + if err != nil { + t.Fatalf("marshal overlap activity: %v", err) + } + + if err := executeRelayActivity( + activity, + relayWrapperTestActor(sourceDomain), + plainBody, + ); err != nil { + t.Fatalf("executeRelayActivity: %v", err) + } + + _, targetCount, found := waitForRelayWrapper(t, objectID) + if !found { + t.Fatal("relay-authored Announce was not queued") + } + if targetCount != 1 { + t.Fatalf( + "deduplicated target count = %d; want 1", + targetCount, + ) + } + if storedRelayBodyEquals(t, plainBody) { + t.Fatal("unsigned source activity must not be queued") + } +} + +func TestExecuteRelayActivitySignedOverlapPrefersTraditionalRoute( + t *testing.T, +) { + const ( + sourceDomain = "signed-source-overlap.example" + targetDomain = "signed-receiver-overlap.example" + ) + + ctx := context.Background() + removeRelayActivityTestKeys(t) + RelayState.RedisClient.Del( + ctx, + "relay:publisher:"+sourceDomain, + ) + RelayState.AddSubscriber(models.Subscriber{ + Domain: targetDomain, + InboxURL: "https://" + targetDomain + "/shared-inbox", + }) + RelayState.AddFollower(models.Follower{ + Domain: targetDomain, + InboxURL: "https://" + targetDomain + "/actor/inbox", + ActorID: "https://" + targetDomain + "/actor", + MutuallyFollow: true, + }) + + t.Cleanup(func() { + RelayState.DelSubscriber(targetDomain) + RelayState.DelFollower(targetDomain) + RelayState.RedisClient.Del( + ctx, + "relay:publisher:"+sourceDomain, + ) + removeRelayActivityTestKeys(t) + }) + + objectID := "https://" + sourceDomain + "/objects/signed-overlap" + activity := relayWrapperTestActivity( + "Create", + sourceDomain, + objectID, + ) + signedBody := relayWrapperSignedBody(t, activity) + + if err := executeRelayActivity( + activity, + relayWrapperTestActor(sourceDomain), + signedBody, + ); err != nil { + t.Fatalf("executeRelayActivity: %v", err) + } + + targetCount, found := waitForStoredRelayBody(t, signedBody) + if !found { + t.Fatal("exact signed body was not queued") + } + if targetCount != 1 { + t.Fatalf( + "signed overlap target count = %d; want 1", + targetCount, + ) + } + + time.Sleep(250 * time.Millisecond) + if keys := len(relayStoredActivities(t)); keys != 1 { + t.Fatalf( + "signed overlap created %d stored payloads; want 1", + keys, + ) + } + if storedRelayWrapperExists(t, objectID) { + t.Fatal( + "overlapping follower route produced duplicate Announce", + ) + } +} + +func TestExecuteRelayActivityUnsignedMissingObjectIDFailsClosed( + t *testing.T, +) { + const ( + sourceDomain = "missing-object-id.example" + targetDomain = "subscriber-missing-id.example" + ) + + ctx := context.Background() + removeRelayActivityTestKeys(t) + RelayState.RedisClient.Del( + ctx, + "relay:publisher:"+sourceDomain, + ) + RelayState.AddSubscriber(models.Subscriber{ + Domain: targetDomain, + InboxURL: "https://" + targetDomain + "/inbox", + }) + + t.Cleanup(func() { + RelayState.DelSubscriber(targetDomain) + RelayState.RedisClient.Del( + ctx, + "relay:publisher:"+sourceDomain, + ) + removeRelayActivityTestKeys(t) + }) + + activity := relayWrapperTestActivity( + "Update", + sourceDomain, + "https://"+sourceDomain+"/objects/missing", + ) + activity.Object = map[string]interface{}{ + "type": "Article", + } + + plainBody, err := json.Marshal(activity) + if err != nil { + t.Fatalf("marshal missing-ID activity: %v", err) + } + + if err := executeRelayActivity( + activity, + relayWrapperTestActor(sourceDomain), + plainBody, + ); err != nil { + t.Fatalf("executeRelayActivity: %v", err) + } + + time.Sleep(300 * time.Millisecond) + if keys := len(relayStoredActivities(t)); keys != 0 { + t.Fatalf( + "unsigned missing-ID activity created %d payloads; want 0", + keys, + ) + } + assertRelayPublisherType(t, sourceDomain, "Update") +} + +func TestExecuteRelayActivitySignedMissingObjectIDForwardsSubscriberOnly( + t *testing.T, +) { + const ( + sourceDomain = "signed-missing-id.example" + traditionalDomain = "signed-missing-id-subscriber.example" + followerDomain = "signed-missing-id-follower.example" + ) + + ctx := context.Background() + removeRelayActivityTestKeys(t) + RelayState.RedisClient.Del( + ctx, + "relay:publisher:"+sourceDomain, + ) + RelayState.AddSubscriber(models.Subscriber{ + Domain: traditionalDomain, + InboxURL: "https://" + traditionalDomain + "/inbox", + }) + RelayState.AddFollower(models.Follower{ + Domain: followerDomain, + InboxURL: "https://" + followerDomain + "/inbox", + ActorID: "https://" + followerDomain + "/actor", + MutuallyFollow: true, + }) + + t.Cleanup(func() { + RelayState.DelSubscriber(traditionalDomain) + RelayState.DelFollower(followerDomain) + RelayState.RedisClient.Del( + ctx, + "relay:publisher:"+sourceDomain, + ) + removeRelayActivityTestKeys(t) + }) + + activity := relayWrapperTestActivity( + "Update", + sourceDomain, + "https://"+sourceDomain+"/objects/missing", + ) + activity.Object = map[string]interface{}{ + "type": "Article", + } + signedBody := relayWrapperSignedBody(t, activity) + + if err := executeRelayActivity( + activity, + relayWrapperTestActor(sourceDomain), + signedBody, + ); err != nil { + t.Fatalf("executeRelayActivity: %v", err) + } + + targetCount, found := waitForStoredRelayBody(t, signedBody) + if !found { + t.Fatal("signed missing-ID body was not queued") + } + if targetCount != 1 { + t.Fatalf( + "signed missing-ID target count = %d; want 1", + targetCount, + ) + } + + time.Sleep(250 * time.Millisecond) + if keys := len(relayStoredActivities(t)); keys != 1 { + t.Fatalf( + "signed missing-ID activity created %d payloads; want 1", + keys, + ) + } + assertRelayPublisherType(t, sourceDomain, "Update") +} + +// EOF: api/relay_activity_wrapper_test.go diff --git a/api/resolver.go b/api/resolver.go index 2afa8eb..0d670fd 100644 --- a/api/resolver.go +++ b/api/resolver.go @@ -254,13 +254,6 @@ func enqueueActivityExcept( body []byte, excludedDomains ...string, ) bool { - if len(subscriptions) > GlobalConfig.MaxFanoutTargets() { - logrus.Warn("Skipped relay activity: fan-out exceeds MAX_FANOUT_TARGETS") - recordQueueAdmission("relay", "rejected", "fanout_limit") - recordFanoutTargets("rejected", len(subscriptions)) - return false - } - excluded := make(map[string]struct{}, len(excludedDomains)) for _, domain := range excludedDomains { normalized := normalizedStoredDomain(domain) @@ -269,16 +262,46 @@ func enqueueActivityExcept( } } + // A server can appear in both the traditional subscriber set and the + // follower-style receiver set. Deliver once per normalized domain so one + // relay-authored activity cannot be duplicated merely because the server + // supports both registration styles. Subscriber entries are loaded first + // and therefore remain the preferred route for an overlapping domain. + seen := make(map[string]struct{}, len(subscriptions)) targets := make([]models.Subscriber, 0, len(subscriptions)) for _, subscription := range subscriptions { - if _, skip := excluded[normalizedStoredDomain(subscription.Domain)]; !skip { - targets = append(targets, subscription) + domain := normalizedStoredDomain(subscription.Domain) + if _, skip := excluded[domain]; skip { + continue } + + dedupeKey := domain + if dedupeKey == "" { + dedupeKey = strings.ToLower( + strings.TrimSpace(subscription.InboxURL), + ) + } + if dedupeKey != "" { + if _, duplicate := seen[dedupeKey]; duplicate { + continue + } + seen[dedupeKey] = struct{}{} + } + + targets = append(targets, subscription) + } + + if len(targets) > GlobalConfig.MaxFanoutTargets() { + logrus.Warn("Skipped relay activity: fan-out exceeds MAX_FANOUT_TARGETS") + recordQueueAdmission("relay", "rejected", "fanout_limit") + recordFanoutTargets("rejected", len(targets)) + return false } if len(targets) < 1 { recordQueueAdmission("relay", "skipped", "no_targets") return true } + accepted, reason := reserveQueueCapacityWithReason(len(targets)) if !accepted { recordQueueAdmission("relay", "rejected", reason) @@ -286,6 +309,7 @@ func enqueueActivityExcept( return false } defer releaseQueueCapacity(len(targets)) + activityID := uuid.NewString() signatures := make([]*tasks.Signature, 0, len(targets)) for _, target := range targets { @@ -641,32 +665,161 @@ func executeEmbeddedAnnounceActivity(activity *models.Activity, actor *models.Ac return nil } -func executeRelayActivity(activity *models.Activity, actor *models.Actor, body []byte) error { +func hasMastodonLinkedDataSignature(body []byte) bool { + var envelope struct { + Signature json.RawMessage `json:"signature"` + } + if err := json.Unmarshal(body, &envelope); err != nil { + return false + } + + rawSignature := strings.TrimSpace(string(envelope.Signature)) + if rawSignature == "" || rawSignature == "null" { + return false + } + + // This is deliberately only a shape check. Activity-Relay preserves the + // exact source bytes so the receiving Mastodon-style server can perform + // the actual LD-signature verification with the originating actor's key. + var signature struct { + Type string `json:"type"` + Creator string `json:"creator"` + Created string `json:"created"` + SignatureValue string `json:"signatureValue"` + } + if err := json.Unmarshal(envelope.Signature, &signature); err != nil { + return false + } + + return signature.Type == "RsaSignature2017" && + strings.TrimSpace(signature.Creator) != "" && + strings.TrimSpace(signature.Created) != "" && + strings.TrimSpace(signature.SignatureValue) != "" +} + +func executeRelayActivity( + activity *models.Activity, + actor *models.Actor, + body []byte, +) error { actorID, err := url.Parse(actor.ID) if err != nil || normalizedActorDomain(actorID) == "" { return errors.New("activity actor has an invalid ID") } + sourceDomain := normalizedActorDomain(actorID) if isActorBlocked(actorID) { - return errors.New(normalizedActorDomain(actorID) + " is blocked") + return errors.New(sourceDomain + " is blocked") + } + if !isActorAbleToRelay(actor) { + logrus.Debug("Skipped Relay Activity : ", activity.Actor) + return nil } - if isActorAbleToRelay(actor) { - if err := recordPublisherActivity(activity, actor); err != nil { - return err + if err := recordPublisherActivity(activity, actor); err != nil { + return err + } + + snapshot := RelayState.Snapshot() + preserveOriginal := hasMastodonLinkedDataSignature(body) + + // Mastodon-style direct forwarding is safe only when the source body + // carries the historical LD-signature shape. Preserve those bytes exactly; + // the receiver, not the relay, verifies the document proof. + if preserveOriginal { + go enqueueActivity( + snapshot.Subscribers, + sourceDomain, + body, + ) + } + + objectID, err := activity.UnwrapInnerObjectId() + if err != nil { + if preserveOriginal { + logrus.WithFields(logrus.Fields{ + "actor": activity.Actor, + "type": activity.Type, + }).Debug( + "Accepted LD-signed relay activity without follower wrapper: object has no ID", + ) + return nil } - go enqueueActivityForSubscriber(normalizedActorDomain(actorID), body) - var innnerObjectId, err = activity.UnwrapInnerObjectId() - if err != nil { - logrus.Debug("Accepted Relay Activity (Announce Failed) : ", activity.Actor) - } else { - announce := models.NewActivityPubActivity(RelayActor, []string{RelayActor.Followers()}, innnerObjectId, "Announce") - jsonData, _ := json.Marshal(&announce) - go enqueueActivityForFollower(normalizedActorDomain(actorID), jsonData) - logrus.Debug("Accepted Relay Activity : ", activity.Actor) + // An unsigned source activity cannot be forwarded under the relay's + // HTTP signature because signer and JSON actor would differ. + logrus.WithFields(logrus.Fields{ + "actor": activity.Actor, + "type": activity.Type, + }).Debug("Skipped unsigned relay fan-out: activity object has no ID") + return nil + } + + announce := models.NewActivityPubActivity( + RelayActor, + []string{RelayActor.Followers()}, + objectID, + "Announce", + ) + jsonData, err := json.Marshal(&announce) + if err != nil { + return err + } + + if preserveOriginal { + // Traditional subscribers get the exact LD-signed source body. + // Follower-style receivers get the relay-authored Announce. If a + // domain is registered through both styles, prefer the traditional + // route and suppress the duplicate follower delivery. + followerTargets := make( + []models.Subscriber, + 0, + len(snapshot.Followers), + ) + for _, follower := range snapshot.Followers { + followerTargets = append( + followerTargets, + models.Subscriber{ + Domain: follower.Domain, + InboxURL: follower.InboxURL, + ActivityID: follower.ActivityID, + ActorID: follower.ActorID, + }, + ) } + + excludedDomains := make( + []string, + 0, + len(snapshot.Subscribers)+1, + ) + excludedDomains = append(excludedDomains, sourceDomain) + for _, subscriber := range snapshot.Subscribers { + excludedDomains = append( + excludedDomains, + subscriber.Domain, + ) + } + + go enqueueActivityExcept( + followerTargets, + jsonData, + excludedDomains..., + ) } else { - logrus.Debug("Skipped Relay Activity : ", activity.Actor) + // Unsigned activities use one relay-authored Announce for every + // receiver style, aligning the JSON actor with the HTTP signer. + go enqueueActivity( + snapshot.SubscribersAndFollowers, + sourceDomain, + jsonData, + ) } + + logrus.WithFields(logrus.Fields{ + "actor": activity.Actor, + "object": objectID, + "type": activity.Type, + "preserved_ld_body": preserveOriginal, + }).Debug("Accepted Relay Activity") return nil } diff --git a/testdata/fep-ae0c/cases.json b/testdata/fep-ae0c/cases.json index 1ce29cb..8401721 100644 --- a/testdata/fep-ae0c/cases.json +++ b/testdata/fep-ae0c/cases.json @@ -98,8 +98,8 @@ "description": "Public Create addresses Public in to.", "profile": "mastodon", "status": "covered", - "decision": "preserve", - "distribution": "forward-original", + "decision": "document-extension", + "distribution": "relay-announce", "activity": { "@context": "https://www.w3.org/ns/activitystreams", "id": "https://publisher.example/activities/create-public", @@ -116,7 +116,7 @@ } }, "current_v2_5": "Queues the original received body for relay fan-out.", - "v3_assertion": "Preserve public forwarding and exact-body characterization." + "v3_assertion": "Unsigned public activities use a relay-authored Announce; exact-body forwarding is preserved only for the historical Mastodon LD-signature shape." }, { "id": "traditional-create-public-cc", @@ -124,7 +124,7 @@ "profile": "mastodon", "status": "specified", "decision": "investigate", - "distribution": "forward-original", + "distribution": "relay-announce", "activity": { "@context": "https://www.w3.org/ns/activitystreams", "id": "https://publisher.example/activities/create-unlisted", @@ -144,7 +144,7 @@ } }, "current_v2_5": "Routes identically to Public in to and fans out the original body.", - "v3_assertion": "Measure receiver behavior and select an explicit configurable policy." + "v3_assertion": "Under the current broad public policy, unsigned cc-only Public activities use a relay-authored Announce; the configurable v3 distribution policy remains separate work." }, { "id": "traditional-forward-preserves-ld-proof", @@ -293,7 +293,7 @@ "profile": "activity-relay-extension", "status": "covered", "decision": "preserve", - "distribution": "forward-original", + "distribution": "relay-announce", "activity": { "@context": "https://www.w3.org/ns/activitystreams", "id": "https://blog.example/activities/create-1", @@ -310,7 +310,7 @@ } }, "current_v2_5": "May fan out after HTTP-signature, actor-host, and domain-policy validation without adding receiver membership.", - "v3_assertion": "Preserve and repeat the security matrix under every signing profile." + "v3_assertion": "Preserve open-publisher acceptance while wrapping unsigned source activities; repeat the security matrix under every HTTP signing profile." }, { "id": "server-actor-arbitrary-path",