阶段三：数据安全持续运营

DataWorks的OpenEvent为您提供消息推送订阅的能力，同时，您可将服务程序注册为DataWorks的扩展程序，通过扩展程序来卡点并响应订阅的事件消息，实现通过扩展程序对特定事件进行消息通知与流程管控。详情请参见扩展程序概述。

1. 开启并配置消息订阅（OpenEvent）。

   • 由于“查询结果下载”为非工作空间内的操作，因此本案例将使用Default总线来承接操作事件消息。

     说明

     如需通过RAM子账号/RAM角色身份读取Default总线中的事件，请提前进行授权。

     

   • 查询结果下载的事件名称为“dataworks:ResourcesDownload:DownloadResources”，可参考下图查看详情。

     

     事件详情及关键参数含义。

     {
       "datacontenttype": "application/json;charset=utf-8",
       "aliyunaccountid": "1107550004253538",
       "aliyunpublishtime": "2023-12-05T07:25:31.708Z",
       "data": {
         "eventCode": "download-resources",
         "extensionBizId": "audit_4d7ebb42b805428483148295a97a8404",
         "extensionBizName": "DataWorks_IDE_Query_20231205152530.csv",
         "requestId": "77cac0c2fc12cecbf1d289128897cc7b@@ac15054317017611303051804e8b43",
         "appId": 3159,
         "tenantId": 524257424564736,
         "blockBusiness": true,
         "eventBody": {
           "sqlText": "SELECT * FROM table_1",   // 查询SQL
           "queryDwProjectId": "3159",    // 查询数据源所在的工作空间ID
           "moduleType": "develop_query",  // 下载来源：develop_query（数据开发查询）/sqlx_query（数据分析查询）/dw_excel（数据分析电子表格）
           "operatorBaseId": "1107550004253538", //操作者的UID
           "datasourceId": "18889", //查询的数据源ID
           "queryDwProjectName": "yongxunQA_emr_chengdu1", // 查询数据源所在的工作空间名称
           "dataRowSize": 4577,  // 下载的数据量
           "datasourceName": "odps_source",  // 查询的数据源名称
           "operatorUid": "1107550004253538"
         },
         "operator": "1107550004253538"
       },
       "aliyunoriginalaccountid": "1107550004253538",
       "specversion": "1.0",
       "aliyuneventbusname": "default",
       "id": "169d171c-d523-4370-a874-bb0fa083194d",
       "source": "acs.dataworks",
       "time": "2023-12-05T15:25:31.588Z",
       "aliyunregionid": "cn-chengdu",
       "type": "dataworks:ResourcesDownload:DownloadResources"
     }

2. 开发部署线扩展程序（Extensions）。

   说明

   实际使用时，请结合上一步消息事件内容进行风险判断。

   处理的扩展点选择“数据下载前置事件”。

   开发扩展程序可参考的示例代码。

   package com.aliyun.dataworks.demo;
   
   import com.alibaba.fastjson.JSON;
   import com.alibaba.fastjson.JSONObject;
   import com.aliyun.dataworks.config.Constants;
   import com.aliyun.dataworks.config.EventCheckEnum;
   import com.aliyun.dataworks.config.ExtensionParamProperties;
   import com.aliyun.dataworks.services.DataWorksOpenApiClient;
   import com.aliyuncs.IAcsClient;
   import com.aliyuncs.dataworks_public.model.v20200518.*;
   import com.aliyuncs.exceptions.ClientException;
   import org.springframework.beans.factory.annotation.Autowired;
   import org.springframework.web.bind.annotation.*;
   
   /**
    * @author dataworks demo
    */
   @RestController
   @RequestMapping("/extensions")
   public class ExtensionsController {
   
       @Autowired(required = false)
       private DataWorksOpenApiClient dataWorksOpenApiClient;
   
       @Autowired
       private ExtensionParamProperties extensionParamProperties;
   
       /**
        * 接收eventBridge推送过来的消息
        * @param jsonParam
        */
       @PostMapping("/consumer")
       public void consumerEventBridge(@RequestBody String jsonParam){
           JSONObject jsonObj = JSON.parseObject(jsonParam);
           String eventCode = jsonObj.getString(Constants.EVENT_CODE_FILED);
           if(Constants.COMMIT_FILE_EVENT_CODE.equals(eventCode) || Constants.DEPLOY_FILE_EVENT_CODE.equals(eventCode)){
               //初始化client
               IAcsClient client = dataWorksOpenApiClient.createClient();
               try {
                   //当前事件参数信息
                   String messageId = jsonObj.getString("id");
                   JSONObject data = jsonObj.getObject("data", JSONObject.class);
                  // Long projectId = data.getLong("appId");
   
                   //初始化事件回调
                   CallbackExtensionRequest callbackExtensionRequest = new CallbackExtensionRequest();
                   callbackExtensionRequest.setMessageId(messageId);
                   callbackExtensionRequest.setExtensionCode(extensionParamProperties.getExtensionCode());
                   JSONObject eventBody = data.getJSONObject("eventBody");
                   Long dataRowSize = eventBody.getLong("dataRowSize");
                   //获取扩展程序选项配置在项目空间下的配置
                   GetOptionValueForProjectRequest getOptionValueForProjectRequest = new GetOptionValueForProjectRequest();
                   //全局扩展点事件的配置信息所属projectId默认为-1
                   getOptionValueForProjectRequest.setProjectId("-1");
                   getOptionValueForProjectRequest.setExtensionCode(extensionParamProperties.getExtensionCode());
                   GetOptionValueForProjectResponse getOptionValueForProjectResponse = client.getAcsResponse(getOptionValueForProjectRequest);
                   JSONObject jsonObject = JSON.parseObject(getOptionValueForProjectResponse.getOptionValue());
                   //这里需根据在DataWorks上实际设置格式来填写
                   Long maxDataRowSize = jsonObject.getLong("dataRowSize");
                   //判断代码是否包含限制函数
                   if(dataRowSize > 1000){
   
                       callbackExtensionRequest.setCheckResult(EventCheckEnum.FAIL.getCode());
                       callbackExtensionRequest.setCheckMessage("下载的行数超过限制数");
                   }else{//成功回调
                       callbackExtensionRequest.setCheckResult(EventCheckEnum.OK.getCode());
                   }
                   //回调DataWorks
                   CallbackExtensionResponse acsResponse = client.getAcsResponse(callbackExtensionRequest);
                   //请求的唯一标识，用于后续错误排查使用
                   System.out.println("acsResponse:" + acsResponse.getRequestId());
               } catch (ClientException e) {
                   //请求的唯一标识，用于后续错误排查使用
                   System.out.println("RequestId:" + e.getRequestId());
                   //错误状态码
                   System.out.println("ErrCode:" + e.getErrCode());
                   //错误描述信息
                   System.out.println("ErrMsg:" + e.getErrMsg());
               }
           }else{
               System.out.println("未能过滤其他事件,请检查配置步骤");
           }
       }
   }

3. 配置风险响应规则。

   进入安全中心 > 风险识别规则，为上述已发布上线的扩展程序配置风险响应，如需审批，则可添加已创建的审批流程。

   说明

   在响应配置时：

   • 如需实现“审批”，则需保证扩展程序在识别到用户风险行为时callbackExtensionRequest.setCheckResult()返回“WARN”。

   • 如需实现“阻断”，则callbackExtensionRequest.setCheckResult()应返回“FAIL”。

   

4. 开启扩展程序。

1. 在数据开发、数据分析模块单击下载数据，将跳转至数据下载页面进行风险检测。

2. 根据检测结果进行后续处理。

   • 若检测通过，则可继续下载。

   

   • 若检测不通过，则下载被阻断，或告知用户需申请权限。

     • 下载被阻断。

       

     • 提示用户申请权限。