Federation (#327)
* initial commit federation * more federation * more federation * more federation * completes follow, accept, undo * add trail create activity * process trail create activity * more trail create activity * adds missing endpoints * adds update and delete activity * adds comment activities * adds activities back * adds summit logs activities * deletes follow counts table * fixes migrations * fixes migrations * fixes migrations * fixes migrations * adds remote profiles * ctd * ctd * we are getting closer... * adds public summit logs * fixes federated trails in lists * adds remote lists * adds list activites * fixes lists * adds notifications * adds iri redirect * fixes comments and summitlogs * fixes follows and profiles * adds asynchronous send * fixes list search * adds encryption key * bug fixes * fixes activity signing * removes custon activity object types * adds html editor * fixes html editor * fixes display issues on mastodon * adds federated sharing * finishes announcements * fixes small summit log issues * adds trail likes * finalizes likes * fixes images for komoot * add disable federation option * adds private profiles * updates docs * adds federated comments * adds trail and comments actvitiypub routes * adds mentions to editor * adds mentions to trails, comments, summit logs * updates theme * updates theme * update docs * update docs * updates docs * fixes various frontend problems * fixes activitypub follows api * updates docs * updates docs --------- Co-authored-by: Christian Beutel <>
This commit is contained in:
Flomp
207
db/federation/activity.go
Normal file
207
db/federation/activity.go
Normal file
@@ -0,0 +1,207 @@
|
||||
package federation
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/x509"
|
||||
"database/sql"
|
||||
"encoding/pem"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
pub "github.com/go-ap/activitypub"
|
||||
|
||||
"sync"
|
||||
|
||||
"github.com/go-ap/jsonld"
|
||||
"github.com/go-fed/httpsig"
|
||||
"github.com/pocketbase/pocketbase/core"
|
||||
"github.com/pocketbase/pocketbase/tools/security"
|
||||
"golang.org/x/sync/semaphore"
|
||||
)
|
||||
|
||||
func PostActivity(app core.App, actor *core.Record, activity *pub.Activity, recipients []string) error {
|
||||
encryptionKey := os.Getenv("POCKETBASE_ENCRYPTION_KEY")
|
||||
if len(encryptionKey) == 0 {
|
||||
return fmt.Errorf("POCKETBASE_ENCRYPTION_KEY not set")
|
||||
}
|
||||
origin := os.Getenv("ORIGIN")
|
||||
if origin == "" {
|
||||
return fmt.Errorf("ORIGIN not set")
|
||||
}
|
||||
|
||||
algs := []httpsig.Algorithm{httpsig.RSA_SHA256}
|
||||
postHeaders := []string{"(request-target)", "Date", "Digest", "Content-Type", "Host"}
|
||||
expiresIn := 60
|
||||
|
||||
body, err := jsonld.WithContext(
|
||||
jsonld.IRI(pub.ActivityBaseURI),
|
||||
jsonld.IRI(pub.SecurityContextURI),
|
||||
).Marshal(activity)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
decryptedPrivateKey, err := security.Decrypt(actor.GetString("private_key"), encryptionKey)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
privateKey, err := x509.ParsePKCS1PrivateKey(decryptedPrivateKey)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
pubID := actor.GetString("iri") + "#main-key"
|
||||
|
||||
client := &http.Client{}
|
||||
var wg sync.WaitGroup
|
||||
|
||||
sem := semaphore.NewWeighted(5) // Limit to 5 concurrent sends
|
||||
|
||||
slices.Sort(recipients)
|
||||
uniqueRecipients := slices.Compact(recipients)
|
||||
|
||||
for _, v := range uniqueRecipients {
|
||||
|
||||
wg.Add(1)
|
||||
go func(inbox string) {
|
||||
defer wg.Done()
|
||||
|
||||
signer, _, err := httpsig.NewSigner(algs, httpsig.DigestSha256, postHeaders, httpsig.Signature, int64(expiresIn))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if err := sem.Acquire(context.Background(), 1); err != nil {
|
||||
app.Logger().Error(fmt.Sprintf("Semaphore acquire failed: %s", err))
|
||||
return
|
||||
}
|
||||
defer sem.Release(1)
|
||||
|
||||
buf := bytes.NewBuffer(body)
|
||||
req, err := http.NewRequest(http.MethodPost, inbox, buf)
|
||||
if err != nil {
|
||||
app.Logger().Error(fmt.Sprintf("Request creation failed: %s", err))
|
||||
return
|
||||
}
|
||||
req.Header.Add("Content-Type", "application/activity+json")
|
||||
req.Header.Add("Date", strings.ReplaceAll(time.Now().UTC().Format(time.RFC1123), "UTC", "GMT"))
|
||||
req.Header.Add("Host", req.Host)
|
||||
|
||||
if err := signer.SignRequest(privateKey, pubID, req, body); err != nil {
|
||||
app.Logger().Error(fmt.Sprintf("Signing request failed: %s", err))
|
||||
return
|
||||
}
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
app.Logger().Error(fmt.Sprintf("Error sending request to inbox %s: %s", inbox, err))
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted {
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
app.Logger().Error(fmt.Sprintf("Inbox %s responded with %d: %s", inbox, resp.StatusCode, body))
|
||||
}
|
||||
|
||||
app.Logger().Info(fmt.Sprintf("Sent %s to %s", activity.Type, inbox))
|
||||
}(v)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
return nil
|
||||
}
|
||||
|
||||
func ProcessActivity(e *core.RequestEvent) error {
|
||||
|
||||
body, err := io.ReadAll(e.Request.Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var activity pub.Activity
|
||||
activity.UnmarshalJSON(body)
|
||||
|
||||
actor, err := e.App.FindFirstRecordByData("activitypub_actors", "iri", activity.Actor.GetID().String())
|
||||
if err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
actor, err = GetActorByIRI(e.App, activity.Actor.GetID().String(), false)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
return err
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
verified, err := verifySignature(e.Request, actor.GetString("public_key"))
|
||||
if err != nil || !verified {
|
||||
return e.UnauthorizedError("Invalid http signature", err)
|
||||
}
|
||||
|
||||
switch activity.Type {
|
||||
case pub.FollowType:
|
||||
ProcessFollowActivity(e.App, actor, activity)
|
||||
case pub.AcceptType:
|
||||
ProcessAcceptActivity(e.App, actor, activity)
|
||||
case pub.UndoType:
|
||||
ProcessUndoActivity(e.App, actor, activity)
|
||||
case pub.UpdateType:
|
||||
fallthrough
|
||||
case pub.CreateType:
|
||||
ProcessCreateOrUpdateActivity(e.App, actor, activity)
|
||||
case pub.DeleteType:
|
||||
ProcessDeleteActivity(e.App, actor, activity)
|
||||
case pub.AnnounceType:
|
||||
ProcessAnnounceActivity(e.App, actor, activity)
|
||||
case pub.LikeType:
|
||||
ProcessLikeActivity(e.App, actor, activity)
|
||||
}
|
||||
return e.JSON(http.StatusOK, nil)
|
||||
}
|
||||
|
||||
func verifySignature(req *http.Request, publicKeyPem string) (bool, error) {
|
||||
origin := os.Getenv("ORIGIN")
|
||||
if origin == "" {
|
||||
return false, fmt.Errorf("ORIGIN not set")
|
||||
}
|
||||
block, _ := pem.Decode([]byte(publicKeyPem))
|
||||
if block == nil || block.Type != "PUBLIC KEY" {
|
||||
return false, fmt.Errorf("could not decode publicKeyPem to PUBLIC KEY pem block type")
|
||||
}
|
||||
|
||||
req.URL = &url.URL{
|
||||
Path: req.Header.Get("X-Forwarded-Path"),
|
||||
}
|
||||
|
||||
url, err := url.Parse(origin)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
req.Header.Set("Host", url.Host)
|
||||
req.Host = url.Host
|
||||
|
||||
publicKey, err := x509.ParsePKIXPublicKey(block.Bytes)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
v, err := httpsig.NewVerifier(req)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
err = v.Verify(publicKey, httpsig.RSA_SHA256)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
return true, nil
|
||||
}
|
||||
Reference in New Issue
Block a user