这个项目支持校验某个用户对某个表是否具有增删改权限吗？

[KuanKuanya]

No description provided.

[melin]

只是解析sql，可以结合解析出来的信息，自己实现权限校验。例如我们的实现：

/**
 * @TODO
 * 1、支持存储过程权限校验
 * 2、支持flink cdc 权限校验
 */
@Service
public class AuthorizationService {

    private static final Logger LOG = LoggerFactory.getLogger(AuthorizationService.class);

    @Autowired
    private TableAccessLogService tableAccessLogService;

    @Autowired
    private UserInfoService userInfoService;

    @Autowired
    private SuperiorBeeConfigClient configClient;

    @Autowired
    private FunctionService functionService;

    @Autowired
    private DataSourceService dataSourceService;

    @Autowired
    private TableService tableService;

    @Autowired
    private WorkspaceService workspaceService;

    @Autowired
    private SecTablePrivsService tablePrivsService;

    @Autowired
    private SecDatabasePrivsService databasePrivsService;

    @Transactional
    public void checkAuthority(AuthContext context, String sql, String[] sparkTempTables) {
        String userId = context.getUserId();
        if (StringUtils.isBlank(context.getUserId())) {
            throw new IllegalArgumentException("userId can not empty");
        }

        Statement statement = SparkSqlHelper.parseStatement(sql);
        boolean supportedSql = SparkSqlHelper.checkSupportedSQL(statement.getStatementType());
        if (!supportedSql) {
            throw new SQLParserException("not support sql: " + sql);
        }

        String[] superTableOwners = configClient.getStringArray(SuperiorConf.SUPERIOR_SKIP_TABLE_AUTH_CHECK_USERS);
        if (!ArrayUtils.contains(superTableOwners, userId)) {
            // 校验sql执行权限
            this.checkAuthority(context, statement, sparkTempTables);
        }
    }

    /**
     * 校验用户是否用执行sql语句权限
     * 创建表：
     * 1. 检测用户是否数据工作空间成员
     * 删除表：
     * 1. 检测用户是否为表owner
     * 修改表 & 添加列 & 修改列：
     * 1. 检测用户是否为表owner
     * 查询表：
     * 1. select table 是否为owner，或者有insert权限
     * 写入表:
     * 1. insertInstanceDependent into table 是否为owner，或者有insert权限
     * 2. select table 是否为owner，或者有insert权限
     */
    @Transactional(rollbackFor = Exception.class)
    public void checkAuthority(AuthContext context, Statement statement, String[] sparkTempTables) {
        final StatementType statementType = statement.getStatementType();
        if (SHOW == statementType) {
            return;
        }
        final PrivilegeType privilegeType = statement.getPrivilegeType();

        if (CREATE_TABLE == statementType || CREATE_TABLE_AS_LIKE == statementType) {
            CreateTable createTable = (CreateTable) statement;
            checkAccessTableAuth(context, createTable.getTableId(), privilegeType);
        } else if (CREATE_VIEW == statementType) {
            CreateView createView = (CreateView) statement;
            checkAccessTableAuth(context, createView.getTableId(), privilegeType);
            QueryStmt queryStmt = createView.getQueryStmt();
            checkSelectTableAuth(context, queryStmt.getInputTables(), sparkTempTables);
            checkFunctionAuth(context, queryStmt.getFunctionNames());
        } else if (CREATE_TABLE_AS_SELECT == statementType) {
            CreateTableAsSelect tableAsSelect = (CreateTableAsSelect) statement;
            checkAccessTableAuth(context, tableAsSelect.getTableId(), privilegeType);
            QueryStmt queryStmt = tableAsSelect.getQueryStmt();
            checkSelectTableAuth(context, queryStmt.getInputTables(), sparkTempTables);
            checkFunctionAuth(context, queryStmt.getFunctionNames());
            addTabAccessLog(context, queryStmt.getInputTables()); // 插入查询记录
        } else if (DROP_TABLE == statementType) {
            DropTable dropTable = (DropTable) statement;
            TableId tableId = dropTable.getTableId();
            checkAccessTableAuth(context, tableId, privilegeType, dropTable.getIfExists());
        } else if (DROP_VIEW == statementType) {
            DropView view = (DropView) statement;
            TableId tableId = view.getTableId();
            checkAccessTableAuth(context, tableId, privilegeType, view.getIfExists());
        } else if (TRUNCATE_TABLE == statementType) {
            TruncateTable truncateTable = (TruncateTable) statement;
            TableId tableId = truncateTable.getTableId();
            checkAccessTableAuth(context, tableId, privilegeType);
        } else if (SELECT == statementType) {
            QueryStmt queryStmt = (QueryStmt) statement;
            checkSelectTableAuth(context, queryStmt.getInputTables(), sparkTempTables);
            checkFunctionAuth(context, queryStmt.getFunctionNames());
            addTabAccessLog(context, queryStmt.getInputTables()); // 插入查询记录
        } else if (INSERT == statementType) { // 多路输出
            InsertTable multiInsertStmt = (InsertTable) statement;
            QueryStmt queryStmt = multiInsertStmt.getQueryStmt();
            checkSelectTableAuth(context, queryStmt.getInputTables(), sparkTempTables);
            checkFunctionAuth(context, queryStmt.getFunctionNames());

            for (TableId tableId : multiInsertStmt.getOutputTables()) {
                checkAccessTableAuth(context, tableId, privilegeType);
            }

            addTabAccessLog(context, queryStmt.getInputTables()); // 插入查询记录
        } else if (DELETE == statementType) {
            DeleteTable deleteTable = (DeleteTable) statement;
            checkAccessTableAuth(context, deleteTable.getTableId(), privilegeType);
        } else if (UPDATE == statementType) {
            UpdateTable updateTable = (UpdateTable) statement;
            checkAccessTableAuth(context, updateTable.getTableId(), privilegeType);
        } else if (MERGE == statementType) {
            MergeTable mergeIntoTable = (MergeTable) statement;
            mergeIntoTable.getInputTables().forEach(tableId -> {
                boolean sparkTempTable = isSparkTempTable(context.getCurrentDatabase(), sparkTempTables, tableId);

                if (!sparkTempTable) {
                    checkAccessTableAuth(context, tableId, privilegeType);
                }
            });

            checkAccessTableAuth(context, mergeIntoTable.getTargetTable(), privilegeType);
        } else if (EXPORT_TABLE == statementType) { // export table
            ExportTable tableData = (ExportTable) statement;
            for (TableId tableId : tableData.getInputTables()) {
                checkAccessTableAuth(context, tableId, privilegeType);
            }
            checkFunctionAuth(context, tableData.getFunctionNames());
        } else if (DATATUNNEL == statementType && statement instanceof DataTunnelExpr) {
            DataTunnelExpr dataTunnelExpr = (DataTunnelExpr) statement;
            for (TableId tableId : dataTunnelExpr.getInputTables()) {
                checkAccessTableAuth(context, tableId, privilegeType);
            }

            long tenantId = context.getTenantId();
            String regionCode = context.getRegionCode();
            String userId = context.getUserId();
            checkDatatunnelAuthority(tenantId, regionCode, userId, dataTunnelExpr);
            checkFunctionAuth(context, dataTunnelExpr.getFunctionNames());
        } else if (CACHE == statementType) { // spark cache
            CacheTable cacheTable = (CacheTable) statement;
            if (cacheTable.getQueryStmt() != null) {
                QueryStmt queryStmt = cacheTable.getQueryStmt();
                this.checkAuthority(context, queryStmt, sparkTempTables);
            } else {
                checkAccessTableAuth(context, cacheTable.getTableId(), privilegeType);
            }

        } else if (ALTER_TABLE == statementType) {
            checkAlterTableAuthority(context, statement, sparkTempTables, privilegeType);
        } else if (CALL == statementType) {
            CallProcedure procedure = (CallProcedure) statement;
            if (procedure.getProperties().containsKey("table")) {
                String tableId = procedure.getProperties().get("table");
                String[] items = StringUtils.split(tableId, ".");
                String databaseName = null;
                String tableName = null;
                if (items.length == 1) {
                    tableName = items[0];
                } else if (items.length == 2) {
                    databaseName = items[0];
                    tableName = items[1];
                } else {
                    throw new SuperiorException("Unsupported identifier " + tableId);
                }
                checkAdminTableAuthority(context, databaseName, tableName);
            }
        }
    }

    private void checkAdminTableAuthority(AuthContext context, String workspaceCode, String tableName) {

        long tenantId = context.getTenantId();
        String regionCode = context.getRegionCode();
        String userId = context.getUserId();
        String currentCatalog = context.getCurrentCatalog();
        String currentDatabase = context.getCurrentDatabase();

        String catalogName = workspaceService.getCatalogName(tenantId, regionCode, workspaceCode, currentCatalog);
        String databaseName = CommonUtils.getDatabaseName(workspaceCode, currentDatabase);
        tableName = StringUtils.lowerCase(tableName);

        TableEntity table = tableService.queryTable(tenantId, regionCode, catalogName, databaseName, tableName);
        if (table == null) {
            String msg = String.format("table not exist: %s.%s.%s", catalogName, databaseName, tableName);
            throw new AccessTableException(msg);
        }
        if (!tablePrivsService.checkOwner(table, userId)) {
            String msg = String.format("%s 不是表owner: %s.%s.%s", userId, catalogName, databaseName, tableName);
            throw new AccessTableException(msg);
        }
    }

    private void checkDatatunnelAuthority(
            Long tenantId, String regionCode, String userId, DataTunnelExpr dataTunnelExpr) {
        Map<String, Object> sourceOptions = dataTunnelExpr.getSourceOptions();
        if (sourceOptions.containsKey("datasource")) {
            String datasource = (String) sourceOptions.get("datasource");
            String catalogName = dataSourceService.queryCatalogName(tenantId, regionCode, datasource);
            String schemaName = (String) sourceOptions.getOrDefault("databaseName", null);
            if (schemaName == null) {
                schemaName = (String) sourceOptions.get("schemaName");
            }
            String tableName = (String) sourceOptions.get("tableName");

            AuthContext context = new AuthContext(tenantId, regionCode, userId, HIVE, catalogName, schemaName);
            if (StringUtils.isNotBlank(tableName) && StringUtils.isNotBlank(schemaName)) {
                checkAccessTableAuth(context, new TableId(tableName), PrivilegeType.READ);
            }
        }

        Map<String, Object> sinkOptions = dataTunnelExpr.getSinkOptions();
        if (sinkOptions.containsKey("datasource")) {
            String datasource = (String) sinkOptions.get("datasource");
            String catalogName = dataSourceService.queryCatalogName(tenantId, regionCode, datasource);
            String schemaName = (String) sinkOptions.getOrDefault("databaseName", null);
            if (schemaName == null) {
                schemaName = (String) sinkOptions.get("schemaName");
            }
            String tableName = (String) sinkOptions.get("tableName");

            AuthContext context = new AuthContext(tenantId, regionCode, userId, HIVE, catalogName, schemaName);
            if (StringUtils.isNotBlank(tableName) && StringUtils.isNotBlank(schemaName)) {
                checkAccessTableAuth(context, new TableId(tableName), PrivilegeType.WRITE);
            }
        }
    }

    private void checkAlterTableAuthority(
            AuthContext context, Statement statement, String[] sparkTempTables, PrivilegeType privilegeType) {
        AlterTable alterTable = (AlterTable) statement;
        TableId tableId = alterTable.getTableId();

        checkAccessTableAuth(context, tableId, privilegeType, alterTable.getIfExists());

        AlterActionType alterType = alterTable.getFirstAlterType();
        if (AlterActionType.ALTER_VIEW_QUERY == alterType) {
            AlterViewAction view = (AlterViewAction) alterTable.firstAction();
            QueryStmt queryStmt = view.getQueryStmt();
            checkSelectTableAuth(context, queryStmt.getInputTables(), sparkTempTables);
            checkFunctionAuth(context, queryStmt.getFunctionNames());
        }
    }

    /**
     * 记录用户查询表信息
     */
    private void addTabAccessLog(AuthContext context, List<TableId> inputTables) {

        long tenantId = context.getTenantId();
        String regionCode = context.getRegionCode();
        String userId = context.getUserId();
        String currentCatalog = context.getCurrentCatalog();
        String currentDatabase = context.getCurrentDatabase();

        UserInfoEntity userInfoEntity = userInfoService.queryUser(tenantId, userId);
        if (userInfoEntity == null) {
            return;
        }
        for (TableId table : inputTables) {
            String userName = userInfoEntity.getCnName();
            String catalogName = CommonUtils.getCatalogName(currentCatalog, table.getCatalogName());
            String databaseName = CommonUtils.getDatabaseName(currentDatabase, table.getSchemaName());
            String tableName = table.getTableName().toLowerCase();
            this.tableAccessLogService.update(
                    tenantId, regionCode, catalogName, databaseName, tableName, userName, userId);
        }
    }

    private void checkSelectTableAuth(AuthContext context, List<TableId> inputTables, String[] sparkTempTables) {
        DataSourceType dataSourceType = context.getDataSourceType();
        for (TableId tableId : inputTables) {
            if (ORACLE == dataSourceType || DAMENG == dataSourceType || OCEANBASE == dataSourceType) {
                if (StringUtils.equalsIgnoreCase("dual", tableId.getTableName())) {
                    continue;
                }
            }

            boolean sparkTempTable = isSparkTempTable(context.getCurrentDatabase(), sparkTempTables, tableId);

            if (!sparkTempTable) {
                checkAccessTableAuth(context, tableId, PrivilegeType.READ);
            }
        }
    }

    private void checkFunctionAuth(AuthContext context, HashSet<FunctionId> functionNames) {
        if (functionNames == null) {
            return;
        }

        LOG.info("context: {} function names: {}", context, StringUtils.join(functionNames, ","));

        long tenantId = context.getTenantId();
        String regionCode = context.getRegionCode();
        String userId = context.getUserId();
        String currentCatalog = context.getCurrentCatalog();
        String currentDatabase = context.getCurrentDatabase();

        for (FunctionId functionId : functionNames) {
            String database = functionId.getSchemaName();
            String funcName = functionId.getFunctionName();
            database = database == null ? context.getCurrentDatabase() : database;

            database = StringUtils.lowerCase(database);
            funcName = StringUtils.lowerCase(funcName);
            FunctionEntity function =
                    functionService.queryFunction(tenantId, regionCode, currentCatalog, database, funcName);
            if (function != null) {
                if (AUTH_ONESELF.equals(function.getAuthType())) {
                    if (!userId.equals(function.getCreater())) {
                        throw new AccessFunctionException("无权访问函数: {}, 函数访问范围：仅个人可用", funcName);
                    }
                } else if (AUTH_WORKSPACE_USERS.equals(function.getAuthType())) {
                    if (StringUtils.isNotBlank(database) && !currentDatabase.equals(database)) {
                        throw new AccessFunctionException("无权访问函数: , 函数访问范围：仅项目空间成员可用", funcName);
                    }
                } else if (AUTH_ASSIGN_USERS.equals(function.getAuthType())) {
                    if (!function.getAuthUsers().contains(userId)) {
                        throw new AccessFunctionException("无权访问函数: {}, 函数访问范围：指定用户可用", funcName);
                    }
                }
            }
        }
    }

    private boolean isSparkTempTable(String currentDatabaseName, String[] sparkTempTables, TableId tableId) {
        boolean sparkTempTable = false;

        if (sparkTempTables != null && sparkTempTables.length > 0) {
            String schemaName = CommonUtils.getDatabaseName(currentDatabaseName, tableId.getSchemaName());
            String tableName = StringUtils.lowerCase(tableId.getTableName());

            if (StringUtils.equalsIgnoreCase(currentDatabaseName, schemaName)) {
                for (String name : sparkTempTables) {
                    if (StringUtils.equalsIgnoreCase(name, tableName)) {
                        sparkTempTable = true;
                        break;
                    }
                }
            }
        }
        return sparkTempTable;
    }

    @Transactional
    public void checkAccessTableAuth(AuthContext authContext, TableId tableId, PrivilegeType privilegeType) {
        this.checkAccessTableAuth(authContext, tableId, privilegeType, false);
    }

    private void checkAccessTableAuth(
            AuthContext context, TableId tableId, PrivilegeType privilegeType, boolean ifExists) {

        long tenantId = context.getTenantId();
        String regionCode = context.getRegionCode();
        String userId = context.getUserId();
        String catalogName = CommonUtils.getCatalogName(context.getCurrentCatalog(), tableId.getCatalogName());
        String databaseName = CommonUtils.getDatabaseName(context.getCurrentDatabase(), tableId.getSchemaName());
        String tableName = tableId.getTableName();

        // 访问paimon 系统表。格式：SELECT * FROM hive_metastore.bigdata.paimon_users_ods$schemas
        String paimonSysTableName = StringUtils.substringAfterLast(tableName, "$");
        if (StringUtils.isNotBlank(paimonSysTableName)
                && ArrayUtils.contains(PAIMON_SYS_TABLES, paimonSysTableName.toLowerCase())) {
            tableName = StringUtils.substringBeforeLast(tableName, "$");
        }

        if (privilegeType != PrivilegeType.CREATE) {
            TableEntity table = tableService.queryTable(tenantId, regionCode, catalogName, databaseName, tableName);
            if (table == null) {
                if (ifExists) {
                    return; // alter table if exists 语句，找不到表，不报错
                }
                throw new AccessTableException("table not exist: {}.{}.{}", catalogName, databaseName, tableName);
            }
            if (tablePrivsService.checkOwner(table, userId)) {
                return;
            }
        }

        if (StringUtils.isBlank(databaseName)) {
            throw new SuperiorException("databaseName can not blank");
        }

        List<SecTablePrivsEntity> tablePrivsList =
                tablePrivsService.queryTablePrivs(tenantId, userId, catalogName, databaseName, tableName);
        List<SecDatabasePrivsEntity> databasePrivsList = null;
        if (tablePrivsList.isEmpty()) {
            databasePrivsList = databasePrivsService.queryDatabasePrivs(tenantId, userId, catalogName, databaseName);
            if (databasePrivsList.isEmpty()) {
                throw new AccessTableException(
                        "{} 没有申请表: {}.{}.{} 使用权限: {}", userId, catalogName, databaseName, tableName, privilegeType);
            }
        }

        for (SecTablePrivsEntity tablePrivs : tablePrivsList) { // 权限审核通过
            LocalDate currentDate = LocalDate.now();
            // 权限过期
            if (currentDate.isAfter(tablePrivs.getExpireDate())) {
                if (tablePrivs.getStatus() == 1 || tablePrivs.getStatus() == 15) {
                    tablePrivs.setStatus(9);
                    tablePrivsService.updateEntity(tablePrivs);
                }
            }

            switch (privilegeType) {
                case READ:
                    if (tablePrivs.isReadPriv()) {
                        return;
                    }
                    break;
                case WRITE:
                    if (tablePrivs.isWritePriv()) {
                        return;
                    }
                    break;
                case ALTER:
                    if (tablePrivs.isAlterPriv()) {
                        return;
                    }
                    break;
                case DROP:
                    if (tablePrivs.isDropPriv()) {
                        return;
                    }
                    break;
                default:
                    throw new AccessTableException("not support " + privilegeType);
            }
        }

        if (databasePrivsList != null && !databasePrivsList.isEmpty()) {
            for (SecDatabasePrivsEntity databasePrivs : databasePrivsList) { // 权限审核通过
                LocalDate currentDate = LocalDate.now();
                // 权限过期
                if (currentDate.isAfter(databasePrivs.getExpireDate())) {
                    if (databasePrivs.getStatus() == 15) {
                        databasePrivs.setStatus(9);
                        databasePrivsService.updateEntity(databasePrivs);
                    }
                }

                switch (privilegeType) {
                    case CREATE:
                        if (databasePrivs.isCreatePriv()) {
                            return;
                        }
                        break;
                    case READ:
                        if (databasePrivs.isReadPriv()) {
                            return;
                        }
                        break;
                    case WRITE:
                        if (databasePrivs.isWritePriv()) {
                            return;
                        }
                        break;
                    case ALTER:
                        if (databasePrivs.isAlterPriv()) {
                            return;
                        }
                        break;
                    case DROP:
                        if (databasePrivs.isDropPriv()) {
                            return;
                        }
                        break;
                    default:
                        throw new AccessTableException("not support " + privilegeType);
                }
            }
        }

        String msg = String.format(
                "%s 没有申请表: %s.%s.%s 使用权限: %s", userId, catalogName, databaseName, tableName, privilegeType);
        throw new AccessTableException(msg);
    }
}