Method checkPermission

worker/dbms.go:584–642

(addr proto.AccountAddress,
	dbID proto.DatabaseID, queryType types.QueryType, queries []types.Query)

}

func (dbms *DBMS) checkPermission(addr proto.AccountAddress,
	dbID proto.DatabaseID, queryType types.QueryType, queries []types.Query) (err error) {
	log.Debugf("in checkPermission, database id: %s, user addr: %s", dbID, addr.String())

	var (
		permStat *types.PermStat
		ok       bool
	)

	// get database perm stat
	permStat, ok = dbms.busService.RequestPermStat(dbID, addr)

	// perm stat not exists
	if !ok {
		err = errors.Wrap(ErrPermissionDeny, "database not exists")
		return
	}

	// check if query is enabled
	if !permStat.Status.EnableQuery() {
		err = errors.Wrapf(ErrPermissionDeny, "cannot query, status: %d", permStat.Status)
		return
	}

	// check query type permission
	switch queryType {
	case types.ReadQuery:
		if !permStat.Permission.HasReadPermission() {
			err = errors.Wrapf(ErrPermissionDeny, "cannot read, permission: %v", permStat.Permission)
			return
		}
	case types.WriteQuery:
		if !permStat.Permission.HasWritePermission() {
			err = errors.Wrapf(ErrPermissionDeny, "cannot write, permission: %v", permStat.Permission)
			return
		}
	default:
		err = errors.Wrapf(ErrInvalidPermission,
			"invalid permission, permission: %v", permStat.Permission)
		return
	}

	// check for query pattern
	var (
		disallowedQuery    string
		hasDisallowedQuery bool
	)

	if disallowedQuery, hasDisallowedQuery = permStat.Permission.HasDisallowedQueryPatterns(queries); hasDisallowedQuery {
		err = errors.Wrapf(ErrPermissionDeny, "disallowed query %s", disallowedQuery)
		log.WithError(err).WithFields(log.Fields{
			"permission": permStat.Permission,
			"query":      disallowedQuery,
		}).Debug("can not query")
		return
	}

	return