sigb.us/dendrite
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases 1 Wiki Activity

Finish implementing redactions (#1189)

Browse source 
* Add a bit more logging to the fedsender

* bugfix: continue sending PDUs if ones are added whilst sending another PDU

Without this, the queue goes back to sleep on `<-oq.notifyPDUs` which won't
fire because `pendingPDUs` is already > 0. This should fix a flakey sytest.

* Break if no txn is sent

* WIP syncapi work

* More debugging

* Bump GMSL version to pull in working Event.Redact

* Remove logging

* Make redactions work on v3+

* Fix more tests
This commit is contained in:
Kegsay 2020-07-08 17:45:39 +01:00 committed by GitHub
parent a5a51b4141
commit d9648b0615
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 131 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
currentstateserver/storage/shared/storage.go
View file

@ -23,7 +23,6 @@ import (
"github.com/matrix-org/dendrite/internal/eventutil"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
)
type Database struct {
@ -45,10 +44,7 @@ func (d *Database) RedactEvent(ctx context.Context, redactedEventID string, reda
return err
}
if len(events) != 1 {
// this should never happen but is non-fatal
util.GetLogger(ctx).WithField("redacted_event_id", redactedEventID).WithField("redaction_event_id", redactedBecause.EventID()).Warnf(
"RedactEvent: missing redacted event",
)
// this will happen for all non-state events
return nil
}
redactionEvent := redactedBecause.Unwrap()

2
go.mod
View file

@ -21,7 +21,7 @@ require (
github.com/matrix-org/go-http-js-libp2p v0.0.0-20200518170932-783164aeeda4
github.com/matrix-org/go-sqlite3-js v0.0.0-20200522092705-bc8506ccbcf3
github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26
github.com/matrix-org/gomatrixserverlib v0.0.0-20200707103800-7470b03f069b
github.com/matrix-org/gomatrixserverlib v0.0.0-20200708152912-d034ccb75e2d
github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f
github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7
github.com/mattn/go-sqlite3 v2.0.2+incompatible

6
go.sum
View file

@ -422,8 +422,10 @@ github.com/matrix-org/go-sqlite3-js v0.0.0-20200522092705-bc8506ccbcf3 h1:Yb+Wlf
github.com/matrix-org/go-sqlite3-js v0.0.0-20200522092705-bc8506ccbcf3/go.mod h1:e+cg2q7C7yE5QnAXgzo512tgFh1RbQLC0+jozuegKgo=
github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26 h1:Hr3zjRsq2bhrnp3Ky1qgx/fzCtCALOoGYylh2tpS9K4=
github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26/go.mod h1:3fxX6gUjWyI/2Bt7J1OLhpCzOfO/bB3AiX0cJtEKud0=
github.com/matrix-org/gomatrixserverlib v0.0.0-20200707103800-7470b03f069b h1:g1ueoPHI5tpafw/QysVfDw43FwRTPqz8sT+MZbK54yk=
github.com/matrix-org/gomatrixserverlib v0.0.0-20200707103800-7470b03f069b/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU=
github.com/matrix-org/gomatrixserverlib v0.0.0-20200708143827-8bfb7222929b h1:mHFK2pcy+fhettE42aoq+JTcj3ysqFqkQLlUpC33Fwg=
github.com/matrix-org/gomatrixserverlib v0.0.0-20200708143827-8bfb7222929b/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU=
github.com/matrix-org/gomatrixserverlib v0.0.0-20200708152912-d034ccb75e2d h1:v3MGEwCLlFjzZcYJu+aI3kUoNxQyCM3DUurjJFlaI04=
github.com/matrix-org/gomatrixserverlib v0.0.0-20200708152912-d034ccb75e2d/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU=
github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f h1:pRz4VTiRCO4zPlEMc3ESdUOcW4PXHH4Kj+YDz1XyE+Y=
github.com/matrix-org/naffka v0.0.0-20200422140631-181f1ee7401f/go.mod h1:y0oDTjZDv5SM9a2rp3bl+CU+bvTRINQsdb7YlDql5Go=
github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7 h1:ntrLa/8xVzeSs8vHFHK25k0C+NV74sYMJnNSg5NoSRo=

5
internal/eventutil/events.go
View file

@ -162,5 +162,10 @@ func RedactEvent(redactionEvent, redactedEvent *gomatrixserverlib.Event) (*gomat
if err != nil {
return nil, err
}
// NOTSPEC: sytest relies on this unspecced field existing :(
err = r.SetUnsignedField("redacted_by", redactionEvent.EventID())
if err != nil {
return nil, err
}
return &r, nil
}

16
roomserver/api/output.go
View file

@ -29,6 +29,22 @@ const (
// OutputTypeRetireInviteEvent indicates that the event is an OutputRetireInviteEvent
OutputTypeRetireInviteEvent OutputType = "retire_invite_event"
// OutputTypeRedactedEvent indicates that the event is an OutputRedactedEvent
//
// This event is emitted when a redaction has been 'validated' (meaning both the redaction and the event to redact are known).
// Redaction validation happens when the roomserver receives either:
// - A redaction for which we have the event to redact.
// - Any event for which we have a redaction.
// When the roomserver receives an event, it will check against the redactions table to see if there is a matching redaction
// for the event. If there is, it will mark the redaction as validated and emit this event. In the common case of a redaction
// happening after receiving the event to redact, the roomserver will emit a OutputTypeNewRoomEvent of m.room.redaction
// immediately followed by a OutputTypeRedactedEvent. In the uncommon case of receiving the redaction BEFORE the event to redact,
// the roomserver will emit a OutputTypeNewRoomEvent of the event to redact immediately followed by a OutputTypeRedactedEvent.
//
// In order to honour redactions correctly, downstream components must ignore m.room.redaction events emitted via OutputTypeNewRoomEvent.
// When downstream components receive an OutputTypeRedactedEvent they must:
// - Pull out the event to redact from the database. They should have this because the redaction is validated.
// - Redact the event and set the corresponding `unsigned` fields to indicate it as redacted.
// - Replace the event in the database.
OutputTypeRedactedEvent OutputType = "redacted_event"
)

11
roomserver/internal/query.go
View file

@ -20,6 +20,7 @@ import (
"context"
"fmt"
"github.com/matrix-org/dendrite/internal/eventutil"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/auth"
"github.com/matrix-org/dendrite/roomserver/state"
@ -867,7 +868,7 @@ func getAuthChain(
func persistEvents(ctx context.Context, db storage.Database, events []gomatrixserverlib.HeaderedEvent) (types.RoomNID, map[string]types.Event) {
var roomNID types.RoomNID
backfilledEventMap := make(map[string]types.Event)
for _, ev := range events {
for j, ev := range events {
nidMap, err := db.EventNIDs(ctx, ev.AuthEventIDs())
if err != nil { // this shouldn't happen as RequestBackfill already found them
logrus.WithError(err).WithField("auth_events", ev.AuthEventIDs()).Error("Failed to find one or more auth events")
@ -891,12 +892,14 @@ func persistEvents(ctx context.Context, db storage.Database, events []gomatrixse
// It's also possible for this event to be a redaction which results in another event being
// redacted, which we don't care about since we aren't returning it in this backfill.
if redactedEventID == ev.EventID() {
ev = ev.Redact().Headered(ev.RoomVersion)
err = ev.SetUnsignedField("redacted_because", redactionEvent)
eventToRedact := ev.Unwrap()
redactedEvent, err := eventutil.RedactEvent(redactionEvent, &eventToRedact)
if err != nil {
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("Failed to set unsigned field")
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("Failed to redact event")
continue
}
ev = redactedEvent.Headered(ev.RoomVersion)
events[j] = ev
}
backfilledEventMap[ev.EventID()] = types.Event{
EventNID: stateAtEvent.StateEntry.EventNID,

4
roomserver/storage/shared/storage.go
View file

@ -563,6 +563,10 @@ func (d *Database) handleRedactions(
// we've seen this redaction before or there is nothing to redact
return nil, "", nil
}
if redactedEvent.RoomID() != redactionEvent.RoomID() {
// redactions across rooms aren't allowed
return nil, "", nil
}
// mark the event as redacted
err = redactedEvent.SetUnsignedField("redacted_because", redactionEvent)

34
syncapi/consumers/roomserver.go
View file

@ -81,11 +81,23 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
switch output.Type {
case api.OutputTypeNewRoomEvent:
// Ignore redaction events. We will add them to the database when they are
// validated (when we receive OutputTypeRedactedEvent)
event := output.NewRoomEvent.Event
if event.Type() == gomatrixserverlib.MRoomRedaction && event.StateKey() == nil {
// in the special case where the event redacts itself, just pass the message through because
// we will never see the other part of the pair
if event.Redacts() != event.EventID() {
return nil
}
}
return s.onNewRoomEvent(context.TODO(), *output.NewRoomEvent)
case api.OutputTypeNewInviteEvent:
return s.onNewInviteEvent(context.TODO(), *output.NewInviteEvent)
case api.OutputTypeRetireInviteEvent:
return s.onRetireInviteEvent(context.TODO(), *output.RetireInviteEvent)
case api.OutputTypeRedactedEvent:
return s.onRedactEvent(context.TODO(), *output.RedactedEvent)
default:
log.WithField("type", output.Type).Debug(
"roomserver output log: ignoring unknown output type",
@ -94,11 +106,25 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
}
}
func (s *OutputRoomEventConsumer) onRedactEvent(
ctx context.Context, msg api.OutputRedactedEvent,
) error {
err := s.db.RedactEvent(ctx, msg.RedactedEventID, &msg.RedactedBecause)
if err != nil {
log.WithError(err).Error("RedactEvent error'd")
return err
}
// fake a room event so we notify clients about the redaction, as if it were
// a normal event.
return s.onNewRoomEvent(ctx, api.OutputNewRoomEvent{
Event: msg.RedactedBecause,
})
}
func (s *OutputRoomEventConsumer) onNewRoomEvent(
ctx context.Context, msg api.OutputNewRoomEvent,
) error {
ev := msg.Event
addsStateEvents := msg.AddsState()
ev, err := s.updateStateEvent(ev)
@ -173,12 +199,10 @@ func (s *OutputRoomEventConsumer) onRetireInviteEvent(
}
func (s *OutputRoomEventConsumer) updateStateEvent(event gomatrixserverlib.HeaderedEvent) (gomatrixserverlib.HeaderedEvent, error) {
var stateKey string
if event.StateKey() == nil {
stateKey = ""
} else {
stateKey = *event.StateKey()
return event, nil
}
stateKey := *event.StateKey()
prevEvent, err := s.db.GetStateEvent(
context.TODO(), event.RoomID(), event.Type(), stateKey,

2
syncapi/storage/interface.go
View file

@ -136,4 +136,6 @@ type Database interface {
// Returns the filterID as a string. Otherwise returns an error if something
// goes wrong.
PutFilter(ctx context.Context, localpart string, filter *gomatrixserverlib.Filter) (string, error)
// RedactEvent wipes an event in the database and sets the unsigned.redacted_because key to the redaction event
RedactEvent(ctx context.Context, redactedEventID string, redactedBecause *gomatrixserverlib.HeaderedEvent) error
}

16
syncapi/storage/postgres/output_room_events_table.go
View file

@ -99,6 +99,9 @@ const selectEarlyEventsSQL = "" +
const selectMaxEventIDSQL = "" +
"SELECT MAX(id) FROM syncapi_output_room_events"
const updateEventJSONSQL = "" +
"UPDATE syncapi_output_room_events SET headered_event_json=$1 WHERE event_id=$2"
// In order for us to apply the state updates correctly, rows need to be ordered in the order they were received (id).
const selectStateInRangeSQL = "" +
"SELECT id, headered_event_json, exclude_from_sync, add_state_ids, remove_state_ids" +
@ -120,6 +123,7 @@ type outputRoomEventsStatements struct {
selectRecentEventsForSyncStmt *sql.Stmt
selectEarlyEventsStmt *sql.Stmt
selectStateInRangeStmt *sql.Stmt
updateEventJSONStmt *sql.Stmt
}
func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
@ -149,9 +153,21 @@ func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
if s.selectStateInRangeStmt, err = db.Prepare(selectStateInRangeSQL); err != nil {
return nil, err
}
if s.updateEventJSONStmt, err = db.Prepare(updateEventJSONSQL); err != nil {
return nil, err
}
return s, nil
}
func (s *outputRoomEventsStatements) UpdateEventJSON(ctx context.Context, event *gomatrixserverlib.HeaderedEvent) error {
headeredJSON, err := json.Marshal(event)
if err != nil {
return err
}
_, err = s.updateEventJSONStmt.ExecContext(ctx, headeredJSON, event.EventID())
return err
}
// selectStateInRange returns the state events between the two given PDU stream positions, exclusive of oldPos, inclusive of newPos.
// Results are bucketed based on the room ID. If the same state is overwritten multiple times between the
// two positions, only the most recent state is returned.

21
syncapi/storage/shared/syncserver.go
View file

@ -24,6 +24,7 @@ import (
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/dendrite/eduserver/cache"
"github.com/matrix-org/dendrite/internal/eventutil"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/syncapi/storage/tables"
@ -597,6 +598,26 @@ func (d *Database) IncrementalSync(
return res, nil
}
func (d *Database) RedactEvent(ctx context.Context, redactedEventID string, redactedBecause *gomatrixserverlib.HeaderedEvent) error {
redactedEvents, err := d.Events(ctx, []string{redactedEventID})
if err != nil {
return err
}
if len(redactedEvents) == 0 {
logrus.WithField("event_id", redactedEventID).WithField("redaction_event", redactedBecause.EventID()).Warnf("missing redacted event for redaction")
return nil
}
eventToRedact := redactedEvents[0].Unwrap()
redactionEvent := redactedBecause.Unwrap()
ev, err := eventutil.RedactEvent(&redactionEvent, &eventToRedact)
if err != nil {
return err
}
newEvent := ev.Headered(redactedBecause.RoomVersion)
return d.OutputEvents.UpdateEventJSON(ctx, &newEvent)
}
// getResponseWithPDUsForCompleteSync creates a response and adds all PDUs needed
// to it. It returns toPos and joinedRoomIDs for use of adding EDUs.
// nolint:nakedret

16
syncapi/storage/sqlite3/output_room_events_table.go
View file

@ -76,6 +76,9 @@ const selectEarlyEventsSQL = "" +
const selectMaxEventIDSQL = "" +
"SELECT MAX(id) FROM syncapi_output_room_events"
const updateEventJSONSQL = "" +
"UPDATE syncapi_output_room_events SET headered_event_json=$1 WHERE event_id=$2"
// In order for us to apply the state updates correctly, rows need to be ordered in the order they were received (id).
/*
$1 = oldPos,
@ -109,6 +112,7 @@ type outputRoomEventsStatements struct {
selectRecentEventsForSyncStmt *sql.Stmt
selectEarlyEventsStmt *sql.Stmt
selectStateInRangeStmt *sql.Stmt
updateEventJSONStmt *sql.Stmt
}
func NewSqliteEventsTable(db *sql.DB, streamID *streamIDStatements) (tables.Events, error) {
@ -140,9 +144,21 @@ func NewSqliteEventsTable(db *sql.DB, streamID *streamIDStatements) (tables.Even
if s.selectStateInRangeStmt, err = db.Prepare(selectStateInRangeSQL); err != nil {
return nil, err
}
if s.updateEventJSONStmt, err = db.Prepare(updateEventJSONSQL); err != nil {
return nil, err
}
return s, nil
}
func (s *outputRoomEventsStatements) UpdateEventJSON(ctx context.Context, event *gomatrixserverlib.HeaderedEvent) error {
headeredJSON, err := json.Marshal(event)
if err != nil {
return err
}
_, err = s.updateEventJSONStmt.ExecContext(ctx, headeredJSON, event.EventID())
return err
}
// selectStateInRange returns the state events between the two given PDU stream positions, exclusive of oldPos, inclusive of newPos.
// Results are bucketed based on the room ID. If the same state is overwritten multiple times between the
// two positions, only the most recent state is returned.

1
syncapi/storage/tables/interface.go
View file

@ -49,6 +49,7 @@ type Events interface {
// SelectEarlyEvents returns the earliest events in the given room.
SelectEarlyEvents(ctx context.Context, txn *sql.Tx, roomID string, r types.Range, limit int) ([]types.StreamEvent, error)
SelectEvents(ctx context.Context, txn *sql.Tx, eventIDs []string) ([]types.StreamEvent, error)
UpdateEventJSON(ctx context.Context, event *gomatrixserverlib.HeaderedEvent) error
}
// Topology keeps track of the depths and stream positions for all events.

16
sytest-whitelist
View file

@ -160,14 +160,6 @@ User can create and send/receive messages in a room with version 1
POST /createRoom ignores attempts to set the room version via creation_content
Inbound federation rejects remote attempts to join local users to rooms
Inbound federation rejects remote attempts to kick local users to rooms
# SyTest currently only implements the v1 endpoints for /send_join and /send_leave,
# whereas Dendrite only supports the v2 endpoints for those, so let's ignore this
# test for now.
#An event which redacts itself should be ignored
# SyTest currently only implements the v1 endpoints for /send_join and /send_leave,
# whereas Dendrite only supports the v2 endpoints for those, so let's ignore this
# test for now.
#A pair of events which redact each other should be ignored
Full state sync includes joined rooms
A message sent after an initial sync appears in the timeline of an incremental sync.
Can add tag
@ -295,6 +287,14 @@ POST /rooms/:room_id/redact/:event_id as random user does not redact message
POST /redact disallows redaction of event in different room
An event which redacts itself should be ignored
A pair of events which redact each other should be ignored
Redaction of a redaction redacts the redaction reason
An event which redacts an event in a different room should be ignored
Can receive redactions from regular users over federation in room version 1
Can receive redactions from regular users over federation in room version 2
Can receive redactions from regular users over federation in room version 3
Can receive redactions from regular users over federation in room version 4
Can receive redactions from regular users over federation in room version 5
Can receive redactions from regular users over federation in room version 6
Outbound federation can backfill events
Inbound federation can backfill events
Backfill checks the events requested belong to the room