Function attemptDoneOnlyStream

dgraph/cmd/dgraphimport/import_security_test.go:190–229

attemptDoneOnlyStream opens StreamExtSnapshot, names the target group, then immediately sends Done with no data packets — the advisory PoC. Stream errors are tolerated and logged: after the fix the server rejects the stream, which is the desired outcome. The assertion that matters is made by the cal

(t *testing.T, ctx context.Context, dc api.DgraphClient, groupId uint32)

// the fix the server rejects the stream, which is the desired outcome. The assertion that
// matters is made by the caller, on whether the store survived.
func attemptDoneOnlyStream(t *testing.T, ctx context.Context, dc api.DgraphClient, groupId uint32) {
	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()

	out, err := dc.StreamExtSnapshot(ctx)
	if err != nil {
		t.Logf("StreamExtSnapshot rejected at open (expected after fix): %v", err)
		return
	}

	// First message names the target group, exactly as the legitimate client does.
	if err := out.Send(&api.StreamExtSnapshotRequest{GroupId: groupId}); err != nil {
		t.Logf("send group id failed (expected after fix): %v", err)
		return
	}
	if _, err := out.Recv(); err != nil {
		t.Logf("recv group ack failed (expected after fix): %v", err)
		return
	}

	// Done-only: no data packets. This alone triggered Prepare()/dropAll() before the fix.
	if err := out.Send(&api.StreamExtSnapshotRequest{Pkt: &api.StreamPacket{Done: true}}); err != nil {
		t.Logf("send done failed (expected after fix): %v", err)
		return
	}
	_ = out.CloseSend()

	// Drain until the server signals completion or closes the stream.
	for {
		resp, err := out.Recv()
		if err != nil {
			t.Logf("stream closed: %v", err)
			return
		}
		if resp.Finish {
			t.Logf("server reported Finish=true")
			return
		}
	}
}