hoernschen/dendrite
1
0
Fork 0
mirror of https://github.com/hoernschen/dendrite.git synced 2025-07-29 12:42:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Implement rejected events (#1426)

Browse source 
* WIP Event rejection

* Still send back errors for rejected events

Instead, discard them at the federationapi /send layer rather than
re-implementing checks at the clientapi/PerformJoin layer.

* Implement rejected events

Critically, rejected events CAN cause state resolution to happen
as it can merge forks in the DAG. This is fine, _provided_ we
do not add the rejected event when performing state resolution,
which is what this PR does. It also fixes the error handling
when NotAllowed happens, as we were checking too early and needlessly
handling NotAllowed in more than one place.

* Update test to match reality

* Modify InputRoomEvents to no longer return an error

Errors do not serialise across HTTP boundaries in polylith mode,
so instead set fields on the InputRoomEventsResponse. Add `Err()`
function to make the API shape basically the same.

* Remove redundant returns; linting

* Update blacklist
This commit is contained in:
Kegsay 2020-09-16 13:00:52 +01:00 • committed by GitHub
parent ba6c7c4a5c
commit 18231f25b4
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
27 changed files with 114 additions and 74 deletions
Download patch file Download diff file Expand all files Collapse all files

2
roomserver/api/api.go
View file

@ -16,7 +16,7 @@ type RoomserverInternalAPI interface {
ctx context.Context,
request *InputRoomEventsRequest,
response *InputRoomEventsResponse,
) error
)
PerformInvite(
ctx context.Context,

7
roomserver/api/api_trace.go
View file

@ -23,10 +23,9 @@ func (t *RoomserverInternalAPITrace) InputRoomEvents(
ctx context.Context,
req *InputRoomEventsRequest,
res *InputRoomEventsResponse,
) error {
err := t.Impl.InputRoomEvents(ctx, req, res)
util.GetLogger(ctx).WithError(err).Infof("InputRoomEvents req=%+v res=%+v", js(req), js(res))
return err
) {
t.Impl.InputRoomEvents(ctx, req, res)
util.GetLogger(ctx).Infof("InputRoomEvents req=%+v res=%+v", js(req), js(res))
}
func (t *RoomserverInternalAPITrace) PerformInvite(

16
roomserver/api/input.go
View file

@ -16,6 +16,8 @@
package api
import (
"fmt"
"github.com/matrix-org/gomatrixserverlib"
)
@ -87,4 +89,18 @@ type InputRoomEventsRequest struct {
// InputRoomEventsResponse is a response to InputRoomEvents
type InputRoomEventsResponse struct {
ErrMsg string // set if there was any error
NotAllowed bool // true if an event in the input was not allowed.
}
func (r *InputRoomEventsResponse) Err() error {
if r.ErrMsg == "" {
return nil
}
if r.NotAllowed {
return &gomatrixserverlib.NotAllowed{
Message: r.ErrMsg,
}
}
return fmt.Errorf("InputRoomEventsResponse: %s", r.ErrMsg)
}

3
roomserver/api/wrapper.go
View file

@ -187,7 +187,8 @@ func SendInputRoomEvents(
) error {
request := InputRoomEventsRequest{InputRoomEvents: ires}
var response InputRoomEventsResponse
return rsAPI.InputRoomEvents(ctx, &request, &response)
rsAPI.InputRoomEvents(ctx, &request, &response)
return response.Err()
}
// SendInvite event to the roomserver.

3
roomserver/internal/alias.go
View file

@ -271,5 +271,6 @@ func (r *RoomserverInternalAPI) sendUpdatedAliasesEvent(
var inputRes api.InputRoomEventsResponse
// Send the request
return r.InputRoomEvents(ctx, &inputReq, &inputRes)
r.InputRoomEvents(ctx, &inputReq, &inputRes)
return inputRes.Err()
}

8
roomserver/internal/input/input.go
View file

@ -110,7 +110,7 @@ func (r *Inputer) InputRoomEvents(
ctx context.Context,
request *api.InputRoomEventsRequest,
response *api.InputRoomEventsResponse,
) error {
) {
// Create a wait group. Each task that we dispatch will call Done on
// this wait group so that we know when all of our events have been
// processed.
@ -156,8 +156,10 @@ func (r *Inputer) InputRoomEvents(
// that back to the caller.
for _, task := range tasks {
if task.err != nil {
return task.err
response.ErrMsg = task.err.Error()
_, rejected := task.err.(*gomatrixserverlib.NotAllowed)
response.NotAllowed = rejected
return
}
}
return nil
}

31
roomserver/internal/input/input_events.go
View file

@ -46,10 +46,11 @@ func (r *Inputer) processRoomEvent(
// Check that the event passes authentication checks and work out
// the numeric IDs for the auth events.
authEventNIDs, err := helpers.CheckAuthEvents(ctx, r.DB, headered, input.AuthEventIDs)
if err != nil {
logrus.WithError(err).WithField("event_id", event.EventID()).WithField("auth_event_ids", input.AuthEventIDs).Error("processRoomEvent.checkAuthEvents failed for event")
return
isRejected := false
authEventNIDs, rejectionErr := helpers.CheckAuthEvents(ctx, r.DB, headered, input.AuthEventIDs)
if rejectionErr != nil {
logrus.WithError(rejectionErr).WithField("event_id", event.EventID()).WithField("auth_event_ids", input.AuthEventIDs).Error("processRoomEvent.checkAuthEvents failed for event, rejecting event")
isRejected = true
}
// If we don't have a transaction ID then get one.
@ -65,12 +66,13 @@ func (r *Inputer) processRoomEvent(
}
// Store the event.
_, stateAtEvent, redactionEvent, redactedEventID, err := r.DB.StoreEvent(ctx, event, input.TransactionID, authEventNIDs)
_, stateAtEvent, redactionEvent, redactedEventID, err := r.DB.StoreEvent(ctx, event, input.TransactionID, authEventNIDs, isRejected)
if err != nil {
return "", fmt.Errorf("r.DB.StoreEvent: %w", err)
}
// if storing this event results in it being redacted then do so.
if redactedEventID == event.EventID() {
if !isRejected && redactedEventID == event.EventID() {
r, rerr := eventutil.RedactEvent(redactionEvent, &event)
if rerr != nil {
return "", fmt.Errorf("eventutil.RedactEvent: %w", rerr)
@ -101,12 +103,22 @@ func (r *Inputer) processRoomEvent(
if stateAtEvent.BeforeStateSnapshotNID == 0 {
// We haven't calculated a state for this event yet.
// Lets calculate one.
err = r.calculateAndSetState(ctx, input, *roomInfo, &stateAtEvent, event)
err = r.calculateAndSetState(ctx, input, *roomInfo, &stateAtEvent, event, isRejected)
if err != nil {
return "", fmt.Errorf("r.calculateAndSetState: %w", err)
}
}
// We stop here if the event is rejected: We've stored it but won't update forward extremities or notify anyone about it.
if isRejected {
logrus.WithFields(logrus.Fields{
"event_id": event.EventID(),
"type": event.Type(),
"room": event.RoomID(),
}).Debug("Stored rejected event")
return event.EventID(), rejectionErr
}
if input.Kind == api.KindRewrite {
logrus.WithFields(logrus.Fields{
"event_id": event.EventID(),
@ -157,11 +169,12 @@ func (r *Inputer) calculateAndSetState(
roomInfo types.RoomInfo,
stateAtEvent *types.StateAtEvent,
event gomatrixserverlib.Event,
isRejected bool,
) error {
var err error
roomState := state.NewStateResolution(r.DB, roomInfo)
if input.HasState {
if input.HasState && !isRejected {
// Check here if we think we're in the room already.
stateAtEvent.Overwrite = true
var joinEventNIDs []types.EventNID
@ -188,7 +201,7 @@ func (r *Inputer) calculateAndSetState(
stateAtEvent.Overwrite = false
// We haven't been told what the state at the event is so we need to calculate it from the prev_events
if stateAtEvent.BeforeStateSnapshotNID, err = roomState.CalculateAndStoreStateBeforeEvent(ctx, event); err != nil {
if stateAtEvent.BeforeStateSnapshotNID, err = roomState.CalculateAndStoreStateBeforeEvent(ctx, event, isRejected); err != nil {
return fmt.Errorf("roomState.CalculateAndStoreStateBeforeEvent: %w", err)
}
}

2
roomserver/internal/perform/perform_backfill.go
View file

@ -535,7 +535,7 @@ func persistEvents(ctx context.Context, db storage.Database, events []gomatrixse
var stateAtEvent types.StateAtEvent
var redactedEventID string
var redactionEvent *gomatrixserverlib.Event
roomNID, stateAtEvent, redactionEvent, redactedEventID, err = db.StoreEvent(ctx, ev.Unwrap(), nil, authNids)
roomNID, stateAtEvent, redactionEvent, redactedEventID, err = db.StoreEvent(ctx, ev.Unwrap(), nil, authNids, false)
if err != nil {
logrus.WithError(err).WithField("event_id", ev.EventID()).Error("Failed to persist event")
continue

3
roomserver/internal/perform/perform_invite.go
View file

@ -183,7 +183,8 @@ func (r *Inviter) PerformInvite(
},
}
inputRes := &api.InputRoomEventsResponse{}
if err = r.Inputer.InputRoomEvents(context.Background(), inputReq, inputRes); err != nil {
r.Inputer.InputRoomEvents(context.Background(), inputReq, inputRes)
if err = inputRes.Err(); err != nil {
return nil, fmt.Errorf("r.InputRoomEvents: %w", err)
}
} else {

3
roomserver/internal/perform/perform_join.go
View file

@ -247,7 +247,8 @@ func (r *Joiner) performJoinRoomByID(
},
}
inputRes := api.InputRoomEventsResponse{}
if err = r.Inputer.InputRoomEvents(ctx, &inputReq, &inputRes); err != nil {
r.Inputer.InputRoomEvents(ctx, &inputReq, &inputRes)
if err = inputRes.Err(); err != nil {
var notAllowed *gomatrixserverlib.NotAllowed
if errors.As(err, &notAllowed) {
return "", &api.PerformError{

3
roomserver/internal/perform/perform_leave.go
View file

@ -139,7 +139,8 @@ func (r *Leaver) performLeaveRoomByID(
},
}
inputRes := api.InputRoomEventsResponse{}
if err = r.Inputer.InputRoomEvents(ctx, &inputReq, &inputRes); err != nil {
r.Inputer.InputRoomEvents(ctx, &inputReq, &inputRes)
if err = inputRes.Err(); err != nil {
return nil, fmt.Errorf("r.InputRoomEvents: %w", err)
}

1
roomserver/internal/query/query.go
View file

@ -70,6 +70,7 @@ func (r *Queryer) QueryStateAfterEvents(
if err != nil {
switch err.(type) {
case types.MissingEventError:
util.GetLogger(ctx).Errorf("QueryStateAfterEvents: MissingEventError: %s", err)
return nil
default:
return err

7
roomserver/inthttp/client.go
View file

@ -149,12 +149,15 @@ func (h *httpRoomserverInternalAPI) InputRoomEvents(
ctx context.Context,
request *api.InputRoomEventsRequest,
response *api.InputRoomEventsResponse,
) error {
) {
span, ctx := opentracing.StartSpanFromContext(ctx, "InputRoomEvents")
defer span.Finish()
apiURL := h.roomserverURL + RoomserverInputRoomEventsPath
return httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
err := httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
if err != nil {
response.ErrMsg = err.Error()
}
}
func (h *httpRoomserverInternalAPI) PerformInvite(

4
roomserver/inthttp/server.go
View file

@ -20,9 +20,7 @@ func AddRoutes(r api.RoomserverInternalAPI, internalAPIMux *mux.Router) {
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
return util.MessageResponse(http.StatusBadRequest, err.Error())
}
if err := r.InputRoomEvents(req.Context(), &request, &response); err != nil {
return util.ErrorResponse(err)
}
r.InputRoomEvents(req.Context(), &request, &response)
return util.JSONResponse{Code: http.StatusOK, JSON: &response}
}),
)

8
roomserver/roomserver_test.go
View file

@ -140,14 +140,6 @@ func mustCreateEvents(t *testing.T, roomVer gomatrixserverlib.RoomVersion, event
return
}
func eventsJSON(events []gomatrixserverlib.Event) []json.RawMessage {
result := make([]json.RawMessage, len(events))
for i := range events {
result[i] = events[i].JSON()
}
return result
}
func mustLoadRawEvents(t *testing.T, ver gomatrixserverlib.RoomVersion, events []json.RawMessage) []gomatrixserverlib.HeaderedEvent {
t.Helper()
hs := make([]gomatrixserverlib.HeaderedEvent, len(events))

5
roomserver/state/state.go
View file

@ -159,7 +159,7 @@ func (v StateResolution) LoadCombinedStateAfterEvents(
}
fullState = append(fullState, entries...)
}
if prevState.IsStateEvent() {
if prevState.IsStateEvent() && !prevState.IsRejected {
// If the prev event was a state event then add an entry for the event itself
// so that we get the state after the event rather than the state before.
fullState = append(fullState, prevState.StateEntry)
@ -523,6 +523,7 @@ func init() {
func (v StateResolution) CalculateAndStoreStateBeforeEvent(
ctx context.Context,
event gomatrixserverlib.Event,
isRejected bool,
) (types.StateSnapshotNID, error) {
// Load the state at the prev events.
prevEventRefs := event.PrevEvents()
@ -561,7 +562,7 @@ func (v StateResolution) CalculateAndStoreStateAfterEvents(
if len(prevStates) == 1 {
prevState := prevStates[0]
if prevState.EventStateKeyNID == 0 {
if prevState.EventStateKeyNID == 0 || prevState.IsRejected {
// 3) None of the previous events were state events and they all
// have the same state, so this event has exactly the same state
// as the previous events.

1
roomserver/storage/interface.go
View file

@ -70,6 +70,7 @@ type Database interface {
// Stores a matrix room event in the database. Returns the room NID, the state snapshot and the redacted event ID if any, or an error.
StoreEvent(
ctx context.Context, event gomatrixserverlib.Event, txnAndSessionID *api.TransactionID, authEventNIDs []types.EventNID,
isRejected bool,
) (types.RoomNID, types.StateAtEvent, *gomatrixserverlib.Event, string, error)
// Look up the state entries for a list of string event IDs
// Returns an error if the there is an error talking to the database

12
roomserver/storage/postgres/events_table.go
View file

@ -65,13 +65,14 @@ CREATE TABLE IF NOT EXISTS roomserver_events (
-- Needed for setting reference hashes when sending new events.
reference_sha256 BYTEA NOT NULL,
-- A list of numeric IDs for events that can authenticate this event.
auth_event_nids BIGINT[] NOT NULL
auth_event_nids BIGINT[] NOT NULL,
is_rejected BOOLEAN NOT NULL DEFAULT FALSE
);
`
const insertEventSQL = "" +
"INSERT INTO roomserver_events (room_nid, event_type_nid, event_state_key_nid, event_id, reference_sha256, auth_event_nids, depth)" +
" VALUES ($1, $2, $3, $4, $5, $6, $7)" +
"INSERT INTO roomserver_events (room_nid, event_type_nid, event_state_key_nid, event_id, reference_sha256, auth_event_nids, depth, is_rejected)" +
" VALUES ($1, $2, $3, $4, $5, $6, $7, $8)" +
" ON CONFLICT ON CONSTRAINT roomserver_event_id_unique" +
" DO NOTHING" +
" RETURNING event_nid, state_snapshot_nid"
@ -88,7 +89,7 @@ const bulkSelectStateEventByIDSQL = "" +
" ORDER BY event_type_nid, event_state_key_nid ASC"
const bulkSelectStateAtEventByIDSQL = "" +
"SELECT event_type_nid, event_state_key_nid, event_nid, state_snapshot_nid FROM roomserver_events" +
"SELECT event_type_nid, event_state_key_nid, event_nid, state_snapshot_nid, is_rejected FROM roomserver_events" +
" WHERE event_id = ANY($1)"
const updateEventStateSQL = "" +
@ -174,12 +175,14 @@ func (s *eventStatements) InsertEvent(
referenceSHA256 []byte,
authEventNIDs []types.EventNID,
depth int64,
isRejected bool,
) (types.EventNID, types.StateSnapshotNID, error) {
var eventNID int64
var stateNID int64
err := s.insertEventStmt.QueryRowContext(
ctx, int64(roomNID), int64(eventTypeNID), int64(eventStateKeyNID),
eventID, referenceSHA256, eventNIDsAsArray(authEventNIDs), depth,
isRejected,
).Scan(&eventNID, &stateNID)
return types.EventNID(eventNID), types.StateSnapshotNID(stateNID), err
}
@ -255,6 +258,7 @@ func (s *eventStatements) BulkSelectStateAtEventByID(
&result.EventStateKeyNID,
&result.EventNID,
&result.BeforeStateSnapshotNID,
&result.IsRejected,
); err != nil {
return nil, err
}

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

@ -382,7 +382,7 @@ func (d *Database) GetLatestEventsForUpdate(
// nolint:gocyclo
func (d *Database) StoreEvent(
ctx context.Context, event gomatrixserverlib.Event,
txnAndSessionID *api.TransactionID, authEventNIDs []types.EventNID,
txnAndSessionID *api.TransactionID, authEventNIDs []types.EventNID, isRejected bool,
) (types.RoomNID, types.StateAtEvent, *gomatrixserverlib.Event, string, error) {
var (
roomNID types.RoomNID
@ -446,6 +446,7 @@ func (d *Database) StoreEvent(
event.EventReference().EventSHA256,
authEventNIDs,
event.Depth(),
isRejected,
); err != nil {
if err == sql.ErrNoRows {
// We've already inserted the event so select the numeric event ID
@ -459,7 +460,9 @@ func (d *Database) StoreEvent(
if err = d.EventJSONTable.InsertEventJSON(ctx, txn, eventNID, event.JSON()); err != nil {
return fmt.Errorf("d.EventJSONTable.InsertEventJSON: %w", err)
}
redactionEvent, redactedEventID, err = d.handleRedactions(ctx, txn, eventNID, event)
if !isRejected { // ignore rejected redaction events
redactionEvent, redactedEventID, err = d.handleRedactions(ctx, txn, eventNID, event)
}
return nil
})
if err != nil {

13
roomserver/storage/sqlite3/events_table.go
View file

@ -41,13 +41,14 @@ const eventsSchema = `
depth INTEGER NOT NULL,
event_id TEXT NOT NULL UNIQUE,
reference_sha256 BLOB NOT NULL,
auth_event_nids TEXT NOT NULL DEFAULT '[]'
auth_event_nids TEXT NOT NULL DEFAULT '[]',
is_rejected BOOLEAN NOT NULL DEFAULT FALSE
);
`
const insertEventSQL = `
INSERT INTO roomserver_events (room_nid, event_type_nid, event_state_key_nid, event_id, reference_sha256, auth_event_nids, depth)
VALUES ($1, $2, $3, $4, $5, $6, $7)
INSERT INTO roomserver_events (room_nid, event_type_nid, event_state_key_nid, event_id, reference_sha256, auth_event_nids, depth, is_rejected)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
ON CONFLICT DO NOTHING;
`
@ -63,7 +64,7 @@ const bulkSelectStateEventByIDSQL = "" +
" ORDER BY event_type_nid, event_state_key_nid ASC"
const bulkSelectStateAtEventByIDSQL = "" +
"SELECT event_type_nid, event_state_key_nid, event_nid, state_snapshot_nid FROM roomserver_events" +
"SELECT event_type_nid, event_state_key_nid, event_nid, state_snapshot_nid, is_rejected FROM roomserver_events" +
" WHERE event_id IN ($1)"
const updateEventStateSQL = "" +
@ -150,13 +151,14 @@ func (s *eventStatements) InsertEvent(
referenceSHA256 []byte,
authEventNIDs []types.EventNID,
depth int64,
isRejected bool,
) (types.EventNID, types.StateSnapshotNID, error) {
// attempt to insert: the last_row_id is the event NID
var eventNID int64
insertStmt := sqlutil.TxStmt(txn, s.insertEventStmt)
result, err := insertStmt.ExecContext(
ctx, int64(roomNID), int64(eventTypeNID), int64(eventStateKeyNID),
eventID, referenceSHA256, eventNIDsAsArray(authEventNIDs), depth,
eventID, referenceSHA256, eventNIDsAsArray(authEventNIDs), depth, isRejected,
)
if err != nil {
return 0, 0, err
@ -261,6 +263,7 @@ func (s *eventStatements) BulkSelectStateAtEventByID(
&result.EventStateKeyNID,
&result.EventNID,
&result.BeforeStateSnapshotNID,
&result.IsRejected,
); err != nil {
return nil, err
}

5
roomserver/storage/tables/interface.go
View file

@ -34,7 +34,10 @@ type EventStateKeys interface {
}
type Events interface {
InsertEvent(c context.Context, txn *sql.Tx, i types.RoomNID, j types.EventTypeNID, k types.EventStateKeyNID, eventID string, referenceSHA256 []byte, authEventNIDs []types.EventNID, depth int64) (types.EventNID, types.StateSnapshotNID, error)
InsertEvent(
ctx context.Context, txn *sql.Tx, i types.RoomNID, j types.EventTypeNID, k types.EventStateKeyNID, eventID string,
referenceSHA256 []byte, authEventNIDs []types.EventNID, depth int64, isRejected bool,
) (types.EventNID, types.StateSnapshotNID, error)
SelectEvent(ctx context.Context, txn *sql.Tx, eventID string) (types.EventNID, types.StateSnapshotNID, error)
// bulkSelectStateEventByID lookups a list of state events by event ID.
// If any of the requested events are missing from the database it returns a types.MissingEventError

3
roomserver/types/types.go
View file

@ -101,6 +101,9 @@ type StateAtEvent struct {
Overwrite bool
// The state before the event.
BeforeStateSnapshotNID StateSnapshotNID
// True if this StateEntry is rejected. State resolution should then treat this
// StateEntry as being a message event (not a state event).
IsRejected bool
// The state entry for the event itself, allows us to calculate the state after the event.
StateEntry
}