package internal import ( "context" "crypto/ed25519" "encoding/base64" "sync" "time" "github.com/matrix-org/dendrite/federationapi/api" "github.com/matrix-org/dendrite/federationapi/queue" "github.com/matrix-org/dendrite/federationapi/statistics" "github.com/matrix-org/dendrite/federationapi/storage" "github.com/matrix-org/dendrite/federationapi/storage/cache" "github.com/matrix-org/dendrite/internal/caching" roomserverAPI "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/setup/config" "github.com/matrix-org/gomatrix" "github.com/matrix-org/gomatrixserverlib" "github.com/sirupsen/logrus" ) // FederationInternalAPI is an implementation of api.FederationInternalAPI type FederationInternalAPI struct { db storage.Database cfg *config.FederationAPI statistics *statistics.Statistics rsAPI roomserverAPI.RoomserverInternalAPI federation *gomatrixserverlib.FederationClient keyRing *gomatrixserverlib.KeyRing queues *queue.OutgoingQueues joins sync.Map // joins currently in progress } func NewFederationInternalAPI( db storage.Database, cfg *config.FederationAPI, rsAPI roomserverAPI.RoomserverInternalAPI, federation *gomatrixserverlib.FederationClient, statistics *statistics.Statistics, caches *caching.Caches, queues *queue.OutgoingQueues, keyRing *gomatrixserverlib.KeyRing, ) *FederationInternalAPI { serverKeyDB, err := cache.NewKeyDatabase(db, caches) if err != nil { logrus.WithError(err).Panicf("failed to set up caching wrapper for server key database") } if keyRing == nil { keyRing = &gomatrixserverlib.KeyRing{ KeyFetchers: []gomatrixserverlib.KeyFetcher{}, KeyDatabase: serverKeyDB, } addDirectFetcher := func() { keyRing.KeyFetchers = append( keyRing.KeyFetchers, &gomatrixserverlib.DirectKeyFetcher{ Client: federation, }, ) } if cfg.PreferDirectFetch { addDirectFetcher() } else { defer addDirectFetcher() } var b64e = base64.StdEncoding.WithPadding(base64.NoPadding) for _, ps := range cfg.KeyPerspectives { perspective := &gomatrixserverlib.PerspectiveKeyFetcher{ PerspectiveServerName: ps.ServerName, PerspectiveServerKeys: map[gomatrixserverlib.KeyID]ed25519.PublicKey{}, Client: federation, } for _, key := range ps.Keys { rawkey, err := b64e.DecodeString(key.PublicKey) if err != nil { logrus.WithError(err).WithFields(logrus.Fields{ "server_name": ps.ServerName, "public_key": key.PublicKey, }).Warn("Couldn't parse perspective key") continue } perspective.PerspectiveServerKeys[key.KeyID] = rawkey } keyRing.KeyFetchers = append(keyRing.KeyFetchers, perspective) logrus.WithFields(logrus.Fields{ "server_name": ps.ServerName, "num_public_keys": len(ps.Keys), }).Info("Enabled perspective key fetcher") } } return &FederationInternalAPI{ db: db, cfg: cfg, rsAPI: rsAPI, keyRing: keyRing, federation: federation, statistics: statistics, queues: queues, } } func (a *FederationInternalAPI) isBlacklistedOrBackingOff(s gomatrixserverlib.ServerName) (*statistics.ServerStatistics, error) { stats := a.statistics.ForServer(s) until, blacklisted := stats.BackoffInfo() if blacklisted { return stats, &api.FederationClientError{ Blacklisted: true, } } now := time.Now() if until != nil && now.Before(*until) { return stats, &api.FederationClientError{ RetryAfter: time.Until(*until), } } return stats, nil } func failBlacklistableError(err error, stats *statistics.ServerStatistics) (until time.Time, blacklisted bool) { if err == nil { return } mxerr, ok := err.(gomatrix.HTTPError) if !ok { return stats.Failure() } if mxerr.Code == 401 { // invalid signature in X-Matrix header return stats.Failure() } if mxerr.Code >= 500 && mxerr.Code < 600 { // internal server errors return stats.Failure() } return } func (a *FederationInternalAPI) doRequest( s gomatrixserverlib.ServerName, request func() (interface{}, error), ) (interface{}, error) { stats, err := a.isBlacklistedOrBackingOff(s) if err != nil { return nil, err } res, err := request() if err != nil { until, blacklisted := failBlacklistableError(err, stats) now := time.Now() var retryAfter time.Duration if until.After(now) { retryAfter = time.Until(until) } return res, &api.FederationClientError{ Err: err.Error(), Blacklisted: blacklisted, RetryAfter: retryAfter, } } stats.Success() return res, nil } func (a *FederationInternalAPI) GetUserDevices( ctx context.Context, s gomatrixserverlib.ServerName, userID string, ) (gomatrixserverlib.RespUserDevices, error) { ctx, cancel := context.WithTimeout(ctx, time.Second*30) defer cancel() ires, err := a.doRequest(s, func() (interface{}, error) { return a.federation.GetUserDevices(ctx, s, userID) }) if err != nil { return gomatrixserverlib.RespUserDevices{}, err } return ires.(gomatrixserverlib.RespUserDevices), nil } func (a *FederationInternalAPI) ClaimKeys( ctx context.Context, s gomatrixserverlib.ServerName, oneTimeKeys map[string]map[string]string, ) (gomatrixserverlib.RespClaimKeys, error) { ctx, cancel := context.WithTimeout(ctx, time.Second*30) defer cancel() ires, err := a.doRequest(s, func() (interface{}, error) { return a.federation.ClaimKeys(ctx, s, oneTimeKeys) }) if err != nil { return gomatrixserverlib.RespClaimKeys{}, err } return ires.(gomatrixserverlib.RespClaimKeys), nil } func (a *FederationInternalAPI) QueryKeys( ctx context.Context, s gomatrixserverlib.ServerName, keys map[string][]string, ) (gomatrixserverlib.RespQueryKeys, error) { ires, err := a.doRequest(s, func() (interface{}, error) { return a.federation.QueryKeys(ctx, s, keys) }) if err != nil { return gomatrixserverlib.RespQueryKeys{}, err } return ires.(gomatrixserverlib.RespQueryKeys), nil } func (a *FederationInternalAPI) Backfill( ctx context.Context, s gomatrixserverlib.ServerName, roomID string, limit int, eventIDs []string, ) (res gomatrixserverlib.Transaction, err error) { ctx, cancel := context.WithTimeout(ctx, time.Second*30) defer cancel() ires, err := a.doRequest(s, func() (interface{}, error) { return a.federation.Backfill(ctx, s, roomID, limit, eventIDs) }) if err != nil { return gomatrixserverlib.Transaction{}, err } return ires.(gomatrixserverlib.Transaction), nil } func (a *FederationInternalAPI) LookupState( ctx context.Context, s gomatrixserverlib.ServerName, roomID, eventID string, roomVersion gomatrixserverlib.RoomVersion, ) (res gomatrixserverlib.RespState, err error) { ctx, cancel := context.WithTimeout(ctx, time.Second*30) defer cancel() ires, err := a.doRequest(s, func() (interface{}, error) { return a.federation.LookupState(ctx, s, roomID, eventID, roomVersion) }) if err != nil { return gomatrixserverlib.RespState{}, err } return ires.(gomatrixserverlib.RespState), nil } func (a *FederationInternalAPI) LookupStateIDs( ctx context.Context, s gomatrixserverlib.ServerName, roomID, eventID string, ) (res gomatrixserverlib.RespStateIDs, err error) { ctx, cancel := context.WithTimeout(ctx, time.Second*30) defer cancel() ires, err := a.doRequest(s, func() (interface{}, error) { return a.federation.LookupStateIDs(ctx, s, roomID, eventID) }) if err != nil { return gomatrixserverlib.RespStateIDs{}, err } return ires.(gomatrixserverlib.RespStateIDs), nil } func (a *FederationInternalAPI) GetEvent( ctx context.Context, s gomatrixserverlib.ServerName, eventID string, ) (res gomatrixserverlib.Transaction, err error) { ctx, cancel := context.WithTimeout(ctx, time.Second*30) defer cancel() ires, err := a.doRequest(s, func() (interface{}, error) { return a.federation.GetEvent(ctx, s, eventID) }) if err != nil { return gomatrixserverlib.Transaction{}, err } return ires.(gomatrixserverlib.Transaction), nil } func (a *FederationInternalAPI) LookupServerKeys( ctx context.Context, s gomatrixserverlib.ServerName, keyRequests map[gomatrixserverlib.PublicKeyLookupRequest]gomatrixserverlib.Timestamp, ) ([]gomatrixserverlib.ServerKeys, error) { ctx, cancel := context.WithTimeout(ctx, time.Minute) defer cancel() ires, err := a.doRequest(s, func() (interface{}, error) { return a.federation.LookupServerKeys(ctx, s, keyRequests) }) if err != nil { return []gomatrixserverlib.ServerKeys{}, err } return ires.([]gomatrixserverlib.ServerKeys), nil } func (a *FederationInternalAPI) MSC2836EventRelationships( ctx context.Context, s gomatrixserverlib.ServerName, r gomatrixserverlib.MSC2836EventRelationshipsRequest, roomVersion gomatrixserverlib.RoomVersion, ) (res gomatrixserverlib.MSC2836EventRelationshipsResponse, err error) { ctx, cancel := context.WithTimeout(ctx, time.Minute) defer cancel() ires, err := a.doRequest(s, func() (interface{}, error) { return a.federation.MSC2836EventRelationships(ctx, s, r, roomVersion) }) if err != nil { return res, err } return ires.(gomatrixserverlib.MSC2836EventRelationshipsResponse), nil } func (a *FederationInternalAPI) MSC2946Spaces( ctx context.Context, s gomatrixserverlib.ServerName, roomID string, r gomatrixserverlib.MSC2946SpacesRequest, ) (res gomatrixserverlib.MSC2946SpacesResponse, err error) { ctx, cancel := context.WithTimeout(ctx, time.Minute) defer cancel() ires, err := a.doRequest(s, func() (interface{}, error) { return a.federation.MSC2946Spaces(ctx, s, roomID, r) }) if err != nil { return res, err } return ires.(gomatrixserverlib.MSC2946SpacesResponse), nil }