本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
# 監控資料庫活動串流
資料庫活動串流會監控和報告活動。活動資料串流會收集並傳送至 Amazon Kinesis。從 Kinesis 中，您可以監視活動串流，或者其他服務和應用程式可以使用活動串流以供進一步分析。您可以使用 AWS CLI 命令 `describe-db-instances` 或 RDS API `DescribeDBInstances` 操作來尋找基礎 Kinesis 串流名稱。
Amazon RDS 會為您管理 Kinesis 串流，如下所示：
+ Amazon RDS 會自動建立保留時間為 24 小時的 Kinesis 串流。
+ 如有必要，Amazon RDS 可擴展 Kinesis 串流。
+ 如果您停止資料庫活動串流或刪除資料庫執行個體，則 Amazon RDS 會刪除 Kinesis 串流。
系統會監控以下類別的活動並將其放在活動串流稽核日誌中：
+ **SQL 命令** – 所有 SQL 命令都經稽核，也是預備陳述式、內建函數和 PL/SQL 函數。對預存程序的呼叫會進行稽核。在儲存的程序或函數中發出的任何 SQL 語句也被稽核。
+ **其他資料庫資訊** – 監控的活動包含完整的 SQL 陳述式、因 DML 命令而受影響的資料列數、經存取的物件和唯一的資料庫名稱。資料庫活動串流還會監控連結變數和已儲存的程序參數。
**重要**
活動資料串流稽核記錄中會顯示每個陳述式的完整 SQL 文字，包括任何敏感資料。但是，如果 Oracle 可以從內容 (例如下列 SQL 陳述式) 判斷資料庫使用者密碼，則該密碼將會被修訂。
```
ALTER ROLE role-name WITH password
```
+ **連線資訊** – 監控的活動包含工作階段和網路資訊、伺服器程序 ID 和結束代碼。
如果活動串流在監控資料庫執行個體時失敗，系統會透過 RDS 事件來通知您。
在下列各節中，您可以存取、稽核和處理資料庫活動串流。
**Topics**
+ [透過 Amazon Kinesis 存取活動串流](DBActivityStreams.KinesisAccess.md)
+ [資料庫活動串流的稽核日誌內容和範例](DBActivityStreams.AuditLog.md)
+ [適用於資料庫活動串流的 databaseActivityEventList JSON 陣列](DBActivityStreams.AuditLog.databaseActivityEventList.md)
+ [使用 AWS SDK 來處理資料庫活動串流](DBActivityStreams.CodeExample.md)
# 透過 Amazon Kinesis 存取活動串流
在您啟用資料庫的活動串流時，系統就會為您建立 Kinesis 串流。透過 Kinesis，您就可以即時監控資料庫活動。若要進一步分析資料庫活動，您可以將 Kinesis 串流連線至消費者應用程式。您也可以將串流連線至合規性管理應用程式，例如 IBM 的 Security Guardium 或 Imperva 的 SecureSphere Database Audit and Protection。
您可以從 RDS 主控台或 Kinesis 主控台存取 Kinesis 串流。
**使用 RDS 主控台從 Kinesis 存取活動串流**
1. 前往 [https://console.aws.amazon.com/rds/](https://console.aws.amazon.com/rds/)，開啟 Amazon RDS 主控台。
1. 在導覽窗格中，選擇 **Databases** (資料庫)。
1. 選擇已在其上啟動活動串流的Amazon RDS 資料課執行個體。
1. 選擇 **Configuration (組態)**。
1. 在 **Database activity stream** (資料庫活動串流) 下，選擇 **Kinesis stream** (Kinesis 串流) 下的連結。
1. 在 Kinesis 主控台中，選擇 **Monitoring** (監控) 來開始觀察資料庫活動。
**使用 Kinesis 主控台從 Kinesis 存取活動串流**
1. 在 [https://console.aws.amazon.com/kinesis](https://console.aws.amazon.com/kinesis) 上開啟 Kinesis 主控台。
1. 從 Kinesis 串流清單中選擇活動串流。
活動串流的名稱包含字首 `aws-rds-das-db-`，後接資料庫的資源 ID。以下是範例。
```
aws-rds-das-db-NHVOV4PCLWHGF52NP
```
若要使用 Amazon RDS 主控台來尋找資料庫的資源 ID，請從資料庫清單中選擇您的資料庫執行個體，然後選擇**Configuration (組態)** 索引標籤。
若要使用 AWS CLI 來尋找活動串流的完整 Kinesis 串流名稱，請使用 [describe-db-instances](https://docs.aws.amazon.com/cli/latest/reference/rds/describe-db-instances.html) CLI 請求，並記下回應中的 `ActivityStreamKinesisStreamName` 值。
1. 選擇 **Monitoring (監控)** 來開始觀察資料庫活動。
如需使用 Amazon Kinesis 的詳細資訊，請參閱[什麼是 Amazon Kinesis Data Streams？](https://docs.aws.amazon.com/streams/latest/dev/introduction.html)。
# 資料庫活動串流的稽核日誌內容和範例
受監控的事件會以 JSON 字串的形式在資料庫活動串流中顯示。此結構包含 JSON 物件，內含的 `DatabaseActivityMonitoringRecord` 會依序包含活動事件的 `databaseActivityEventList` 陣列。
**注意**
對於資料庫活動串流，`paramList` JSON 陣列不包含休眠應用程式的 null 值。
**Topics**
+ [活動串流的稽核記錄範例](#DBActivityStreams.AuditLog.Examples)
+ [DatabaseActivityMonitoringRecords JSON 物件](#DBActivityStreams.AuditLog.DatabaseActivityMonitoringRecords)
+ [databaseActivityEvents JSON 物件](#DBActivityStreams.AuditLog.databaseActivityEvents)
## 活動串流的稽核記錄範例
以下是活動事件記錄的範例解密 JSON 稽核日誌。
**Example a CONNECT SQL 陳述式的活動事件記錄**
以下活動事件記錄顯示 JDBC 精簡用戶端 (`clientApplication`) 使用 `CONNECT` SQL 陳述式 (`command`) 登入您的 Oracle 資料庫。
```
{
"class": "Standard",
"clientApplication": "JDBC Thin Client",
"command": "LOGON",
"commandText": null,
"dbid": "0123456789",
"databaseName": "ORCL",
"dbProtocol": "oracle",
"dbUserName": "TEST",
"endTime": null,
"errorMessage": null,
"exitCode": 0,
"logTime": "2021-01-15 00:15:36.233787",
"netProtocol": "tcp",
"objectName": null,
"objectType": null,
"paramList": [],
"pid": 17904,
"remoteHost": "123.456.789.012",
"remotePort": "25440",
"rowCount": null,
"serverHost": "987.654.321.098",
"serverType": "oracle",
"serverVersion": "19.0.0.0.ru-2020-01.rur-2020-01.r1.EE.3",
"serviceName": "oracle-ee",
"sessionId": 987654321,
"startTime": null,
"statementId": 1,
"substatementId": null,
"transactionId": "0000000000000000",
"engineNativeAuditFields": {
"UNIFIED_AUDIT_POLICIES": "TEST_POL_EVERYTHING",
"FGA_POLICY_NAME": null,
"DV_OBJECT_STATUS": null,
"SYSTEM_PRIVILEGE_USED": "CREATE SESSION",
"OLS_LABEL_COMPONENT_TYPE": null,
"XS_SESSIONID": null,
"ADDITIONAL_INFO": null,
"INSTANCE_ID": 1,
"DBID": 123456789
"DV_COMMENT": null,
"RMAN_SESSION_STAMP": null,
"NEW_NAME": null,
"DV_ACTION_NAME": null,
"OLS_PROGRAM_UNIT_NAME": null,
"OLS_STRING_LABEL": null,
"RMAN_SESSION_RECID": null,
"OBJECT_PRIVILEGES": null,
"OLS_OLD_VALUE": null,
"XS_TARGET_PRINCIPAL_NAME": null,
"XS_NS_ATTRIBUTE": null,
"XS_NS_NAME": null,
"DBLINK_INFO": null,
"AUTHENTICATION_TYPE": "(TYPE\u003d(DATABASE));(CLIENT ADDRESS\u003d((ADDRESS\u003d(PROTOCOL\u003dtcp)(HOST\u003d205.251.233.183)(PORT\u003d25440))));",
"OBJECT_EDITION": null,
"OLS_PRIVILEGES_GRANTED": null,
"EXCLUDED_USER": null,
"DV_ACTION_OBJECT_NAME": null,
"OLS_LABEL_COMPONENT_NAME": null,
"EXCLUDED_SCHEMA": null,
"DP_TEXT_PARAMETERS1": null,
"XS_USER_NAME": null,
"XS_ENABLED_ROLE": null,
"XS_NS_ATTRIBUTE_NEW_VAL": null,
"DIRECT_PATH_NUM_COLUMNS_LOADED": null,
"AUDIT_OPTION": null,
"DV_EXTENDED_ACTION_CODE": null,
"XS_PACKAGE_NAME": null,
"OLS_NEW_VALUE": null,
"DV_RETURN_CODE": null,
"XS_CALLBACK_EVENT_TYPE": null,
"USERHOST": "a1b2c3d4e5f6.amazon.com",
"GLOBAL_USERID": null,
"CLIENT_IDENTIFIER": null,
"RMAN_OPERATION": null,
"TERMINAL": "unknown",
"OS_USERNAME": "sumepate",
"OLS_MAX_READ_LABEL": null,
"XS_PROXY_USER_NAME": null,
"XS_DATASEC_POLICY_NAME": null,
"DV_FACTOR_CONTEXT": null,
"OLS_MAX_WRITE_LABEL": null,
"OLS_PARENT_GROUP_NAME": null,
"EXCLUDED_OBJECT": null,
"DV_RULE_SET_NAME": null,
"EXTERNAL_USERID": null,
"EXECUTION_ID": null,
"ROLE": null,
"PROXY_SESSIONID": 0,
"DP_BOOLEAN_PARAMETERS1": null,
"OLS_POLICY_NAME": null,
"OLS_GRANTEE": null,
"OLS_MIN_WRITE_LABEL": null,
"APPLICATION_CONTEXTS": null,
"XS_SCHEMA_NAME": null,
"DV_GRANTEE": null,
"XS_COOKIE": null,
"DBPROXY_USERNAME": null,
"DV_ACTION_CODE": null,
"OLS_PRIVILEGES_USED": null,
"RMAN_DEVICE_TYPE": null,
"XS_NS_ATTRIBUTE_OLD_VAL": null,
"TARGET_USER": null,
"XS_ENTITY_TYPE": null,
"ENTRY_ID": 1,
"XS_PROCEDURE_NAME": null,
"XS_INACTIVITY_TIMEOUT": null,
"RMAN_OBJECT_TYPE": null,
"SYSTEM_PRIVILEGE": null,
"NEW_SCHEMA": null,
"SCN": 5124715
}
}
```
下列活動事件記錄顯示 SQL Server 資料庫的登入失敗。
```
{
"type": "DatabaseActivityMonitoringRecord",
"clusterId": "",
"instanceId": "db-4JCWQLUZVFYP7DIWP6JVQ77O3Q",
"databaseActivityEventList": [
{
"class": "LOGIN",
"clientApplication": "Microsoft SQL Server Management Studio",
"command": "LOGIN FAILED",
"commandText": "Login failed for user 'test'. Reason: Password did not match that for the login provided. [CLIENT: local-machine]",
"databaseName": "",
"dbProtocol": "SQLSERVER",
"dbUserName": "test",
"endTime": null,
"errorMessage": null,
"exitCode": 0,
"logTime": "2022-10-06 21:34:42.7113072+00",
"netProtocol": null,
"objectName": "",
"objectType": "LOGIN",
"paramList": null,
"pid": null,
"remoteHost": "local machine",
"remotePort": null,
"rowCount": 0,
"serverHost": "172.31.30.159",
"serverType": "SQLSERVER",
"serverVersion": "15.00.4073.23.v1.R1",
"serviceName": "sqlserver-ee",
"sessionId": 0,
"startTime": null,
"statementId": "0x1eb0d1808d34a94b9d3dcf5432750f02",
"substatementId": 1,
"transactionId": "0",
"type": "record",
"engineNativeAuditFields": {
"target_database_principal_id": 0,
"target_server_principal_id": 0,
"target_database_principal_name": "",
"server_principal_id": 0,
"user_defined_information": "",
"response_rows": 0,
"database_principal_name": "",
"target_server_principal_name": "",
"schema_name": "",
"is_column_permission": false,
"object_id": 0,
"server_instance_name": "EC2AMAZ-NFUJJNO",
"target_server_principal_sid": null,
"additional_information": "00x000048188local machineB"-->,
"duration_milliseconds": 0,
"permission_bitmask": "0x00000000000000000000000000000000",
"data_sensitivity_information": "",
"session_server_principal_name": "",
"connection_id": "98B4F537-0F82-49E3-AB08-B9D33B5893EF",
"audit_schema_version": 1,
"database_principal_id": 0,
"server_principal_sid": null,
"user_defined_event_id": 0,
"host_name": "EC2AMAZ-NFUJJNO"
}
}
]
}
```
若未啟用資料庫活動串流，則 JSON 文件中的最後一個欄位為 `"engineNativeAuditFields": { }`。
**Example CREATE TABLE 陳述式的活動事件記錄**
以下是 您 Oracle 資料庫的 `CREATE TABLE` 事件範例。
```
{
"class": "Standard",
"clientApplication": "sqlplus@ip-12-34-5-678 (TNS V1-V3)",
"command": "CREATE TABLE",
"commandText": "CREATE TABLE persons(\n person_id NUMBER GENERATED BY DEFAULT AS IDENTITY,\n first_name VARCHAR2(50) NOT NULL,\n last_name VARCHAR2(50) NOT NULL,\n PRIMARY KEY(person_id)\n)",
"dbid": "0123456789",
"databaseName": "ORCL",
"dbProtocol": "oracle",
"dbUserName": "TEST",
"endTime": null,
"errorMessage": null,
"exitCode": 0,
"logTime": "2021-01-15 00:22:49.535239",
"netProtocol": "beq",
"objectName": "PERSONS",
"objectType": "TEST",
"paramList": [],
"pid": 17687,
"remoteHost": "123.456.789.0",
"remotePort": null,
"rowCount": null,
"serverHost": "987.654.321.01",
"serverType": "oracle",
"serverVersion": "19.0.0.0.ru-2020-01.rur-2020-01.r1.EE.3",
"serviceName": "oracle-ee",
"sessionId": 1234567890,
"startTime": null,
"statementId": 43,
"substatementId": null,
"transactionId": "090011007F0D0000",
"engineNativeAuditFields": {
"UNIFIED_AUDIT_POLICIES": "TEST_POL_EVERYTHING",
"FGA_POLICY_NAME": null,
"DV_OBJECT_STATUS": null,
"SYSTEM_PRIVILEGE_USED": "CREATE SEQUENCE, CREATE TABLE",
"OLS_LABEL_COMPONENT_TYPE": null,
"XS_SESSIONID": null,
"ADDITIONAL_INFO": null,
"INSTANCE_ID": 1,
"DV_COMMENT": null,
"RMAN_SESSION_STAMP": null,
"NEW_NAME": null,
"DV_ACTION_NAME": null,
"OLS_PROGRAM_UNIT_NAME": null,
"OLS_STRING_LABEL": null,
"RMAN_SESSION_RECID": null,
"OBJECT_PRIVILEGES": null,
"OLS_OLD_VALUE": null,
"XS_TARGET_PRINCIPAL_NAME": null,
"XS_NS_ATTRIBUTE": null,
"XS_NS_NAME": null,
"DBLINK_INFO": null,
"AUTHENTICATION_TYPE": "(TYPE\u003d(DATABASE));(CLIENT ADDRESS\u003d((PROTOCOL\u003dbeq)(HOST\u003d123.456.789.0)));",
"OBJECT_EDITION": null,
"OLS_PRIVILEGES_GRANTED": null,
"EXCLUDED_USER": null,
"DV_ACTION_OBJECT_NAME": null,
"OLS_LABEL_COMPONENT_NAME": null,
"EXCLUDED_SCHEMA": null,
"DP_TEXT_PARAMETERS1": null,
"XS_USER_NAME": null,
"XS_ENABLED_ROLE": null,
"XS_NS_ATTRIBUTE_NEW_VAL": null,
"DIRECT_PATH_NUM_COLUMNS_LOADED": null,
"AUDIT_OPTION": null,
"DV_EXTENDED_ACTION_CODE": null,
"XS_PACKAGE_NAME": null,
"OLS_NEW_VALUE": null,
"DV_RETURN_CODE": null,
"XS_CALLBACK_EVENT_TYPE": null,
"USERHOST": "ip-10-13-0-122",
"GLOBAL_USERID": null,
"CLIENT_IDENTIFIER": null,
"RMAN_OPERATION": null,
"TERMINAL": "pts/1",
"OS_USERNAME": "rdsdb",
"OLS_MAX_READ_LABEL": null,
"XS_PROXY_USER_NAME": null,
"XS_DATASEC_POLICY_NAME": null,
"DV_FACTOR_CONTEXT": null,
"OLS_MAX_WRITE_LABEL": null,
"OLS_PARENT_GROUP_NAME": null,
"EXCLUDED_OBJECT": null,
"DV_RULE_SET_NAME": null,
"EXTERNAL_USERID": null,
"EXECUTION_ID": null,
"ROLE": null,
"PROXY_SESSIONID": 0,
"DP_BOOLEAN_PARAMETERS1": null,
"OLS_POLICY_NAME": null,
"OLS_GRANTEE": null,
"OLS_MIN_WRITE_LABEL": null,
"APPLICATION_CONTEXTS": null,
"XS_SCHEMA_NAME": null,
"DV_GRANTEE": null,
"XS_COOKIE": null,
"DBPROXY_USERNAME": null,
"DV_ACTION_CODE": null,
"OLS_PRIVILEGES_USED": null,
"RMAN_DEVICE_TYPE": null,
"XS_NS_ATTRIBUTE_OLD_VAL": null,
"TARGET_USER": null,
"XS_ENTITY_TYPE": null,
"ENTRY_ID": 12,
"XS_PROCEDURE_NAME": null,
"XS_INACTIVITY_TIMEOUT": null,
"RMAN_OBJECT_TYPE": null,
"SYSTEM_PRIVILEGE": null,
"NEW_SCHEMA": null,
"SCN": 5133083
}
}
```
以下範例顯示 SQL Server 資料庫的 `CREATE TABLE` 事件。
```
{
"type": "DatabaseActivityMonitoringRecord",
"clusterId": "",
"instanceId": "db-4JCWQLUZVFYP7DIWP6JVQ77O3Q",
"databaseActivityEventList": [
{
"class": "SCHEMA",
"clientApplication": "Microsoft SQL Server Management Studio - Query",
"command": "ALTER",
"commandText": "Create table [testDB].[dbo].[TestTable2](\r\ntextA varchar(6000),\r\n textB varchar(6000)\r\n)",
"databaseName": "testDB",
"dbProtocol": "SQLSERVER",
"dbUserName": "test",
"endTime": null,
"errorMessage": null,
"exitCode": 1,
"logTime": "2022-10-06 21:44:38.4120677+00",
"netProtocol": null,
"objectName": "dbo",
"objectType": "SCHEMA",
"paramList": null,
"pid": null,
"remoteHost": "local machine",
"remotePort": null,
"rowCount": 0,
"serverHost": "172.31.30.159",
"serverType": "SQLSERVER",
"serverVersion": "15.00.4073.23.v1.R1",
"serviceName": "sqlserver-ee",
"sessionId": 84,
"startTime": null,
"statementId": "0x5178d33d56e95e419558b9607158a5bd",
"substatementId": 1,
"transactionId": "4561864",
"type": "record",
"engineNativeAuditFields": {
"target_database_principal_id": 0,
"target_server_principal_id": 0,
"target_database_principal_name": "",
"server_principal_id": 2,
"user_defined_information": "",
"response_rows": 0,
"database_principal_name": "dbo",
"target_server_principal_name": "",
"schema_name": "",
"is_column_permission": false,
"object_id": 1,
"server_instance_name": "EC2AMAZ-NFUJJNO",
"target_server_principal_sid": null,
"additional_information": "",
"duration_milliseconds": 0,
"permission_bitmask": "0x00000000000000000000000000000000",
"data_sensitivity_information": "",
"session_server_principal_name": "test",
"connection_id": "EE1FE3FD-EF2C-41FD-AF45-9051E0CD983A",
"audit_schema_version": 1,
"database_principal_id": 1,
"server_principal_sid": "0x010500000000000515000000bdc2795e2d0717901ba6998cf4010000",
"user_defined_event_id": 0,
"host_name": "EC2AMAZ-NFUJJNO"
}
}
]
}
```
**Example SELECT 陳述式的活動事件記錄**
下列範例顯示 Oracle 資料庫 的 `SELECT` 事件。
```
{
"class": "Standard",
"clientApplication": "sqlplus@ip-12-34-5-678 (TNS V1-V3)",
"command": "SELECT",
"commandText": "select count(*) from persons",
"databaseName": "1234567890",
"dbProtocol": "oracle",
"dbUserName": "TEST",
"endTime": null,
"errorMessage": null,
"exitCode": 0,
"logTime": "2021-01-15 00:25:18.850375",
"netProtocol": "beq",
"objectName": "PERSONS",
"objectType": "TEST",
"paramList": [],
"pid": 17687,
"remoteHost": "123.456.789.0",
"remotePort": null,
"rowCount": null,
"serverHost": "987.654.321.09",
"serverType": "oracle",
"serverVersion": "19.0.0.0.ru-2020-01.rur-2020-01.r1.EE.3",
"serviceName": "oracle-ee",
"sessionId": 1080639707,
"startTime": null,
"statementId": 44,
"substatementId": null,
"transactionId": null,
"engineNativeAuditFields": {
"UNIFIED_AUDIT_POLICIES": "TEST_POL_EVERYTHING",
"FGA_POLICY_NAME": null,
"DV_OBJECT_STATUS": null,
"SYSTEM_PRIVILEGE_USED": null,
"OLS_LABEL_COMPONENT_TYPE": null,
"XS_SESSIONID": null,
"ADDITIONAL_INFO": null,
"INSTANCE_ID": 1,
"DV_COMMENT": null,
"RMAN_SESSION_STAMP": null,
"NEW_NAME": null,
"DV_ACTION_NAME": null,
"OLS_PROGRAM_UNIT_NAME": null,
"OLS_STRING_LABEL": null,
"RMAN_SESSION_RECID": null,
"OBJECT_PRIVILEGES": null,
"OLS_OLD_VALUE": null,
"XS_TARGET_PRINCIPAL_NAME": null,
"XS_NS_ATTRIBUTE": null,
"XS_NS_NAME": null,
"DBLINK_INFO": null,
"AUTHENTICATION_TYPE": "(TYPE\u003d(DATABASE));(CLIENT ADDRESS\u003d((PROTOCOL\u003dbeq)(HOST\u003d123.456.789.0)));",
"OBJECT_EDITION": null,
"OLS_PRIVILEGES_GRANTED": null,
"EXCLUDED_USER": null,
"DV_ACTION_OBJECT_NAME": null,
"OLS_LABEL_COMPONENT_NAME": null,
"EXCLUDED_SCHEMA": null,
"DP_TEXT_PARAMETERS1": null,
"XS_USER_NAME": null,
"XS_ENABLED_ROLE": null,
"XS_NS_ATTRIBUTE_NEW_VAL": null,
"DIRECT_PATH_NUM_COLUMNS_LOADED": null,
"AUDIT_OPTION": null,
"DV_EXTENDED_ACTION_CODE": null,
"XS_PACKAGE_NAME": null,
"OLS_NEW_VALUE": null,
"DV_RETURN_CODE": null,
"XS_CALLBACK_EVENT_TYPE": null,
"USERHOST": "ip-12-34-5-678",
"GLOBAL_USERID": null,
"CLIENT_IDENTIFIER": null,
"RMAN_OPERATION": null,
"TERMINAL": "pts/1",
"OS_USERNAME": "rdsdb",
"OLS_MAX_READ_LABEL": null,
"XS_PROXY_USER_NAME": null,
"XS_DATASEC_POLICY_NAME": null,
"DV_FACTOR_CONTEXT": null,
"OLS_MAX_WRITE_LABEL": null,
"OLS_PARENT_GROUP_NAME": null,
"EXCLUDED_OBJECT": null,
"DV_RULE_SET_NAME": null,
"EXTERNAL_USERID": null,
"EXECUTION_ID": null,
"ROLE": null,
"PROXY_SESSIONID": 0,
"DP_BOOLEAN_PARAMETERS1": null,
"OLS_POLICY_NAME": null,
"OLS_GRANTEE": null,
"OLS_MIN_WRITE_LABEL": null,
"APPLICATION_CONTEXTS": null,
"XS_SCHEMA_NAME": null,
"DV_GRANTEE": null,
"XS_COOKIE": null,
"DBPROXY_USERNAME": null,
"DV_ACTION_CODE": null,
"OLS_PRIVILEGES_USED": null,
"RMAN_DEVICE_TYPE": null,
"XS_NS_ATTRIBUTE_OLD_VAL": null,
"TARGET_USER": null,
"XS_ENTITY_TYPE": null,
"ENTRY_ID": 13,
"XS_PROCEDURE_NAME": null,
"XS_INACTIVITY_TIMEOUT": null,
"RMAN_OBJECT_TYPE": null,
"SYSTEM_PRIVILEGE": null,
"NEW_SCHEMA": null,
"SCN": 5136972
}
}
```
以下範例顯示 SQL Server 資料庫的 `SELECT` 事件。
```
{
"type": "DatabaseActivityMonitoringRecord",
"clusterId": "",
"instanceId": "db-4JCWQLUZVFYP7DIWP6JVQ77O3Q",
"databaseActivityEventList": [
{
"class": "TABLE",
"clientApplication": "Microsoft SQL Server Management Studio - Query",
"command": "SELECT",
"commandText": "select * from [testDB].[dbo].[TestTable]",
"databaseName": "testDB",
"dbProtocol": "SQLSERVER",
"dbUserName": "test",
"endTime": null,
"errorMessage": null,
"exitCode": 1,
"logTime": "2022-10-06 21:24:59.9422268+00",
"netProtocol": null,
"objectName": "TestTable",
"objectType": "TABLE",
"paramList": null,
"pid": null,
"remoteHost": "local machine",
"remotePort": null,
"rowCount": 0,
"serverHost": "172.31.30.159",
"serverType": "SQLSERVER",
"serverVersion": "15.00.4073.23.v1.R1",
"serviceName": "sqlserver-ee",
"sessionId": 62,
"startTime": null,
"statementId": "0x03baed90412f564fad640ebe51f89b99",
"substatementId": 1,
"transactionId": "4532935",
"type": "record",
"engineNativeAuditFields": {
"target_database_principal_id": 0,
"target_server_principal_id": 0,
"target_database_principal_name": "",
"server_principal_id": 2,
"user_defined_information": "",
"response_rows": 0,
"database_principal_name": "dbo",
"target_server_principal_name": "",
"schema_name": "dbo",
"is_column_permission": true,
"object_id": 581577110,
"server_instance_name": "EC2AMAZ-NFUJJNO",
"target_server_principal_sid": null,
"additional_information": "",
"duration_milliseconds": 0,
"permission_bitmask": "0x00000000000000000000000000000001",
"data_sensitivity_information": "",
"session_server_principal_name": "test",
"connection_id": "AD3A5084-FB83-45C1-8334-E923459A8109",
"audit_schema_version": 1,
"database_principal_id": 1,
"server_principal_sid": "0x010500000000000515000000bdc2795e2d0717901ba6998cf4010000",
"user_defined_event_id": 0,
"host_name": "EC2AMAZ-NFUJJNO"
}
}
]
}
```
## DatabaseActivityMonitoringRecords JSON 物件
資料庫活動事件記錄位於 JSON 物件中，其中包含下列資訊。
****
| JSON 欄位 | 資料類型 | 描述 |
| --- | --- | --- |
| `type` | string | JSON 記錄類型。值為 `DatabaseActivityMonitoringRecords`。 |
| version | string | 資料庫活動監控記錄的版本。Oracle 資料庫使用 1.3 版，SQL Server 則使用 1.4 版。這些引擎版本引入 engineNativeAuditFields JSON 物件。 |
| [databaseActivityEvents](#DBActivityStreams.AuditLog.databaseActivityEvents) | string | 包含活動事件的 JSON 物件。 |
| 金鑰 | string | 您用來解密 [databaseActivityEventList JSON 陣列](DBActivityStreams.AuditLog.databaseActivityEventList.md) 的加密金鑰 |
## databaseActivityEvents JSON 物件
`databaseActivityEvents` JSON 物件包含以下資訊。
### JSON 記錄中的最上層欄位
稽核記錄檔中的每個事件都會包裝在 JSON 格式的記錄中。此記錄包含下列欄位。
**type**
此欄位永遠具有值 `DatabaseActivityMonitoringRecords`。
**version**
此欄位代表資料庫活動串流資料通訊協定或合約的版本。其會定義哪些欄位可用。
**databaseActivityEvents**
代表一或多個活動事件的加密字串。它被表示為一個 base64 位元組陣列。在您解密字串時，結果會是 JSON 格式的記錄，其中包含欄位，如本節範例所示。
**金鑰**
用來加密 `databaseActivityEvents` 字串的加密資料金鑰。這與您啟動資料庫活動串流時 AWS KMS key 提供的相同。
下列範例顯示此記錄的格式。
```
{
"type":"DatabaseActivityMonitoringRecords",
"version":"1.3",
"databaseActivityEvents":"encrypted audit records",
"key":"encrypted key"
}
```
```
"type":"DatabaseActivityMonitoringRecords",
"version":"1.4",
"databaseActivityEvents":"encrypted audit records",
"key":"encrypted key"
```
採取下列步驟來解密 `databaseActivityEvents` 欄位的內容：
1. 使用您在啟動資料庫活動串流時提供的 KMS 金鑰，解密 `key` JSON 欄位中的值。這麼做會以純文字傳回資料加密金鑰。
1. Base64 解碼 `databaseActivityEvents` JSON 欄位中的值，以取得稽核承載的二進位格式的加密文字。
1. 使用您在第一個步驟中解碼的資料加密金鑰來解密二進位密文。
1. 解壓縮已解密的承載。
+ 加密的承載在 `databaseActivityEvents` 欄位。
+ 該 `databaseActivityEventList` 欄位包含稽核記錄的陣列。陣列中的 `type` 欄位可以是 `record` 或 `heartbeat`。
稽核日誌活動事件記錄是包含以下資訊的 JSON 物件。
****
| JSON 欄位 | 資料類型 | 描述 |
| --- | --- | --- |
| `type` | string | JSON 記錄類型。值為 `DatabaseActivityMonitoringRecord`。 |
| instanceId | string | 資料庫執行個體資源識別符。它對應於資料庫執行個體屬性 DbiResourceId。 |
| [databaseActivityEventList JSON 陣列](DBActivityStreams.AuditLog.databaseActivityEventList.md) | string | 活動稽核記錄或活動訊號訊息的陣列。 |
# 適用於資料庫活動串流的 databaseActivityEventList JSON 陣列
稽核日誌承載是加密的 `databaseActivityEventList` JSON 陣列。以下資料表列出稽核記錄中已解密 `DatabaseActivityEventList` 陣列中，每個活動事件的欄位 (按英文字母順序列出)。
在 Oracle 資料庫中啟用統一稽核時，稽核記錄會填入此新稽核權杖中。此 `UNIFIED_AUDIT_TRAIL` 檢視會從稽核權杖擷取稽核記錄，以表格形式顯示稽核記錄。在您開始資料庫活動串流時，`UNIFIED_AUDIT_TRAIL` 中的欄位會映射至 `databaseActivityEventList` 陣列。
**重要**
事件結構可能會改變。Amazon RDS 可能會在未來將新的欄位新增至活動事件。在剖析 JSON 資料的應用程式中，請確定程式碼可以忽略，或針對未知欄位名稱採取適當的動作。
## Amazon RDS for Oracle 的 databaseActivityEventList 欄位
以下是 Amazon RDS for Oracle 的 `databaseActivityEventList` 欄位。
| 欄位 | 資料類型 | 來源 | 描述 |
| --- | --- | --- | --- |
| `class` | string | `UNIFIED_AUDIT_TRAIL` 中的 `AUDIT_TYPE` 欄位 | 活動事件的類別。這對應於 `AUDIT_TYPE` 欄位中的 `UNIFIED_AUDIT_TRAIL` 檢視。Amazon RDS for Oracle 的有效值如下： [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/AmazonRDS/latest/UserGuide/DBActivityStreams.AuditLog.databaseActivityEventList.html) 如需更多資訊，請參閱 Oracle 文件中的 [UNIFIED\$1AUDIT\$1TRAIL](https://docs.oracle.com/en/database/oracle/oracle-database/19/refrn/UNIFIED_AUDIT_TRAIL.html#GUID-B7CE1C02-2FD4-47D6-80AA-CF74A60CDD1D)。 |
| `clientApplication` | string | `CLIENT_PROGRAM_NAME` 中的 `UNIFIED_AUDIT_TRAIL` | 用戶端報告用來連接的應用程式。用戶端不需提供此資訊，因此此值可以是 Null。範例值為 `JDBC Thin Client`。 |
| `command` | string | `UNIFIED_AUDIT_TRAIL` 中的 `ACTION_NAME` 欄位 | 由使用者執行的動作名稱。若要瞭解完整動作，請讀取命令名稱和 `AUDIT_TYPE` 值。範例值為 `ALTER DATABASE`。 |
| `commandText` | string | `UNIFIED_AUDIT_TRAIL` 中的 `SQL_TEXT` 欄位 | 與事件相關聯的 SQL 陳述式。範例值為 `ALTER DATABASE BEGIN BACKUP`。 |
| `databaseName` | string | `V$DATABASE` 中的 `NAME` 欄位 | 資料庫的名稱。 |
| `dbid` | 數字 | `UNIFIED_AUDIT_TRAIL` 中的 `DBID` 欄位 | 資料庫的數字識別符。範例值為 `1559204751`。 |
| `dbProtocol` | string | N/A | 資料庫通訊協定。在此測試版中，值為 `oracle`。 |
| `dbUserName` | string | `UNIFIED_AUDIT_TRAIL` 中的 `DBUSERNAME` 欄位 | 動作已受稽核的資料庫使用者名稱。範例值為 `RDSADMIN`。 |
| `endTime` | string | N/A | 此欄位不會用於 RDS for Oracle 且一律為空。 |
| `engineNativeAuditFields` | 物件 | `UNIFIED_AUDIT_TRAIL` | 根據預設，此物件為空。在您使用 `--engine-native-audit-fields-included` 選項開始活動串流時，此物件會包含下列欄位及其值： ADDITIONAL_INFO
APPLICATION_CONTEXTS
AUDIT_OPTION
AUTHENTICATION_TYPE
CLIENT_IDENTIFIER
CURRENT_USER
DBLINK_INFO
DBPROXY_USERNAME
DIRECT_PATH_NUM_COLUMNS_LOADED
DP_BOOLEAN_PARAMETERS1
DP_TEXT_PARAMETERS1
DV_ACTION_CODE
DV_ACTION_NAME
DV_ACTION_OBJECT_NAME
DV_COMMENT
DV_EXTENDED_ACTION_CODE
DV_FACTOR_CONTEXT
DV_GRANTEE
DV_OBJECT_STATUS
DV_RETURN_CODE
DV_RULE_SET_NAME
ENTRY_ID
EXCLUDED_OBJECT
EXCLUDED_SCHEMA
EXCLUDED_USER
EXECUTION_ID
EXTERNAL_USERID
FGA_POLICY_NAME
GLOBAL_USERID
INSTANCE_ID
KSACL_SERVICE_NAME
KSACL_SOURCE_LOCATION
KSACL_USER_NAME
NEW_NAME
NEW_SCHEMA
OBJECT_EDITION
OBJECT_PRIVILEGES
OLS_GRANTEE
OLS_LABEL_COMPONENT_NAME
OLS_LABEL_COMPONENT_TYPE
OLS_MAX_READ_LABEL
OLS_MAX_WRITE_LABEL
OLS_MIN_WRITE_LABEL
OLS_NEW_VALUE
OLS_OLD_VALUE
OLS_PARENT_GROUP_NAME
OLS_POLICY_NAME
OLS_PRIVILEGES_GRANTED
OLS_PRIVILEGES_USED
OLS_PROGRAM_UNIT_NAME
OLS_STRING_LABEL
OS_USERNAME
PROTOCOL_ACTION_NAME
PROTOCOL_MESSAGE
PROTOCOL_RETURN_CODE
PROTOCOL_SESSION_ID
PROTOCOL_USERHOST
PROXY_SESSIONID
RLS_INFO
RMAN_DEVICE_TYPE
RMAN_OBJECT_TYPE
RMAN_OPERATION
RMAN_SESSION_RECID
RMAN_SESSION_STAMP
ROLE
SCN
SYSTEM_PRIVILEGE
SYSTEM_PRIVILEGE_USED
TARGET_USER
TERMINAL
UNIFIED_AUDIT_POLICIES
USERHOST
XS_CALLBACK_EVENT_TYPE
XS_COOKIE
XS_DATASEC_POLICY_NAME
XS_ENABLED_ROLE
XS_ENTITY_TYPE
XS_INACTIVITY_TIMEOUT
XS_NS_ATTRIBUTE
XS_NS_ATTRIBUTE_NEW_VAL
XS_NS_ATTRIBUTE_OLD_VAL
XS_NS_NAME
XS_PACKAGE_NAME
XS_PROCEDURE_NAME
XS_PROXY_USER_NAME
XS_SCHEMA_NAME
XS_SESSIONID
XS_TARGET_PRINCIPAL_NAME
XS_USER_NAME
如需更多資訊，請參閱 Oracle 資料庫文件中的 [UNIFIED\$1AUDIT\$1TRAIL](https://docs.oracle.com/database/121/REFRN/GUID-B7CE1C02-2FD4-47D6-80AA-CF74A60CDD1D.htm#REFRN29162)。 |
| `errorMessage` | string | N/A | 此欄位不會用於 RDS for Oracle 且一律為空。 |
| `exitCode` | 數字 | `UNIFIED_AUDIT_TRAIL` 中的 `RETURN_CODE` 欄位 | 因動作而產生的 Oracle 資料庫錯誤代碼。若動作成功，則值為 `0`。 |
| `logTime` | string | `UNIFIED_AUDIT_TRAIL` 中的 `EVENT_TIMESTAMP_UTC` 欄位 | 建立稽核權杖項目的時間戳記。範例值為 `2020-11-27 06:56:14.981404`。 |
| `netProtocol` | string | `UNIFIED_AUDIT_TRAIL` 中的 `AUTHENTICATION_TYPE` 欄位 | 網路通訊協定。範例值為 `TCP`。 |
| `objectName` | string | `UNIFIED_AUDIT_TRAIL` 中的 `OBJECT_NAME` 欄位 | 受動作影響的物件名稱。範例值為 `employees`。 |
| `objectType` | string | `UNIFIED_AUDIT_TRAIL` 中的 `OBJECT_SCHEMA` 欄位 | 受動作影響的物件結構描述名稱。範例值為 `hr`。 |
| `paramList` | 清單 | `UNIFIED_AUDIT_TRAIL` 中的 `SQL_BINDS` 欄位 | 與 `SQL_TEXT` 相關聯的綁定變數清單 (如有)。範例值為 `parameter_1,parameter_2`。 |
| `pid` | 數字 | `UNIFIED_AUDIT_TRAIL` 中的 `OS_PROCESS` 欄位 | Oracle 資料庫程序的作業系統程序識別符。範例值為 `22396`。 |
| `remoteHost` | string | `UNIFIED_AUDIT_TRAIL` 中的 `AUTHENTICATION_TYPE` 欄位 | 在產生工作階段後的用戶端 IP 地址或主機名稱。範例值為 `123.456.789.123`。 |
| `remotePort` | string | `UNIFIED_AUDIT_TRAIL` 中的 `AUTHENTICATION_TYPE` 欄位 | 用戶端連接埠號碼。Oracle 資料庫環境中的典型值為 `1521`。 |
| `rowCount` | 數字 | N/A | 此欄位不會用於 RDS for Oracle 且一律為空。 |
| `serverHost` | string | 資料庫主機 | 資料庫伺服器主機的 IP 地址。範例值為 `123.456.789.123`。 |
| `serverType` | string | N/A | 資料庫伺服器類型。值一律為 `ORACLE`。 |
| `serverVersion` | string | 資料庫主機 | Amazon RDS for Oracle 版本，發布更新 (RU)，以及發布更新修訂版 (RUR)。範例值為 `19.0.0.0.ru-2020-01.rur-2020-01.r1.EE.3`。 |
| `serviceName` | string | 資料庫主機 | 服務的名稱。範例值為 `oracle-ee`。 |
| `sessionId` | 數字 | `UNIFIED_AUDIT_TRAIL` 中的 `SESSIONID` 欄位 | 稽核的工作階段識別符。例如，`1894327130`。 |
| `startTime` | string | N/A | 此欄位不會用於 RDS for Oracle 且一律為空。 |
| `statementId` | 數字 | `UNIFIED_AUDIT_TRAIL` 中的 `STATEMENT_ID` 欄位 | 每個陳述式執行的數字 ID。陳述式可能會引起許多動作。範例值為 `142197`。 |
| `substatementId` | N/A | N/A | 此欄位不會用於 RDS for Oracle 且一律為空。 |
| `transactionId` | string | `UNIFIED_AUDIT_TRAIL` 中的 `TRANSACTION_ID` 欄位 | 其中物件已修改的交易識別符。範例值為 `02000800D5030000`。 |
## Amazon RDS for SQL Server 的 databaseActivityEventList 欄位
以下是 Amazon RDS for SQL Server 的 `databaseActivityEventList` 欄位。
| 欄位 | 資料類型 | 來源 | 描述 |
| --- | --- | --- | --- |
| `class` | string | ` sys.fn_get_audit_file.class_type` 對應至 `sys.dm_audit_class_type_map.class_type_desc` | 活動事件的類別。如需詳細資訊，請參閱 Microsoft 文件中的 [SQL Server 稽核 (資料庫引擎)](https://learn.microsoft.com/en-us/sql/relational-databases/security/auditing/sql-server-audit-database-engine?view=sql-server-ver16)。 |
| `clientApplication` | string | `sys.fn_get_audit_file.application_name` | 根據用戶端報告，用戶端連線的應用程式 (SQL Server 第 14 版及更高版本)。此欄位在 SQL Server 第 13 版為空值。 |
| `command` | string | `sys.fn_get_audit_file.action_id` 對應至 `sys.dm_audit_actions.name` | SQL 陳述式的一般類別。此欄位的值取決於該類別的值。 |
| `commandText` | string | `sys.fn_get_audit_file.statement` | 此欄位會顯示該 SQL 陳述式。 |
| `databaseName` | string | `sys.fn_get_audit_file.database_name` | 資料庫的名稱。 |
| `dbProtocol` | string | N/A | 資料庫通訊協定。此值為 `SQLSERVER`。 |
| `dbUserName` | string | `sys.fn_get_audit_file.server_principal_name` | 用戶端驗證所用的資料庫使用者。 |
| `endTime` | string | N/A | Amazon RDS for SQL Server 不使用此欄位，且其值為空值。 |
| `engineNativeAuditFields` | 物件 | 未列於此欄位的 `sys.fn_get_audit_file` 中各欄。 | 根據預設，此物件為空。使用 `--engine-native-audit-fields-included` 選項開啟活動串流時，此物件會包含此 JSON 對應不會回傳的其他原生引擎稽核欄位。 |
| `errorMessage` | string | N/A | Amazon RDS for SQL Server 不使用此欄位，且其值為空值。 |
| `exitCode` | integer | `sys.fn_get_audit_file.succeeded` | 顯示是否成功啟動事件。此欄不能為空值。針對登入事件以外的所有事件，此欄會顯示許可檢查成功與否，但不會顯示操作是否成功。 數值包含： [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/AmazonRDS/latest/UserGuide/DBActivityStreams.AuditLog.databaseActivityEventList.html) |
| `logTime` | string | `sys.fn_get_audit_file.event_time` | SQL Server 記錄的事件時間戳記。 |
| `netProtocol` | string | N/A | Amazon RDS for SQL Server 不使用此欄位，且其值為空值。 |
| `objectName` | string | `sys.fn_get_audit_file.object_name` | SQL 陳述式在其中操作的資料庫物件名稱。 |
| `objectType` | string | `sys.fn_get_audit_file.class_type` 對應至 `sys.dm_audit_class_type_map.class_type_desc` | SQL 陳述式在其中操作的資料庫物件類型。 |
| `paramList` | string | N/A | Amazon RDS for SQL Server 不使用此欄位，且其值為空值。 |
| `pid` | integer | N/A | Amazon RDS for SQL Server 不使用此欄位，且其值為空值。 |
| `remoteHost` | string | `sys.fn_get_audit_file.client_ip` | 發出 SQL 陳述式 (SQL Server 第 14 版及更高版本) 之用戶端的 IP 地址或主機名稱。此欄位在 SQL Server 第 13 版為空值。 |
| `remotePort` | integer | N/A | Amazon RDS for SQL Server 不使用此欄位，且其值為空值。 |
| `rowCount` | integer | `sys.fn_get_audit_file.affected_rows` | SQL 陳述式影響的資料列數 (SQL Server 第 14 版及更高版本)。此欄位在 SQL Server 第 13 版中。 |
| `serverHost` | string | 資料庫主機 | 主機資料庫伺服器的 IP 地址。 |
| `serverType` | string | N/A | 資料庫伺服器類型。值為 `SQLSERVER`。 |
| `serverVersion` | string | 資料庫主機 | 資料庫伺服器版本，例如，SQL Server 2017 年的 15.00.4073.23.v1.R1。 |
| `serviceName` | string | 資料庫主機 | 服務的名稱。範例值為 `sqlserver-ee`。 |
| `sessionId` | integer | `sys.fn_get_audit_file.session_id` | 工作階段的唯一識別碼。 |
| `startTime` | string | N/A | Amazon RDS for SQL Server 不使用此欄位，且其值為空值。 |
| `statementId` | string | `sys.fn_get_audit_file.sequence_group_id` | 唯一用戶端 SQL 陳述式的識別碼。每個產生出的事件的識別碼都不同。範例值為 `0x38eaf4156267184094bb82071aaab644`。 |
| `substatementId` | integer | `sys.fn_get_audit_file.sequence_number` | 用來決定陳述式序號的識別碼。當大量記錄分成多個記錄時，此識別碼將有所助益。 |
| `transactionId` | integer | `sys.fn_get_audit_file.transaction_id` | 交易的識別碼。如果沒有任何作用中交易，則該值為零。 |
| `type` | string | 已產生的資料庫活動串流 | 事件的類型。值為 `record` 或 `heartbeat`。 |
# 使用 AWS SDK 來處理資料庫活動串流
您可以透過使用 AWS 開發套件來以程式設計的方式處理活動串流。下列是功能完整的 Java 和 Python 範例，其示範使用資料庫活動串流記錄進行執行個體型啟用。
------
#### [ Java ]
```
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.security.NoSuchAlgorithmException;
import java.security.NoSuchProviderException;
import java.security.Security;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.zip.GZIPInputStream;
import javax.crypto.Cipher;
import javax.crypto.NoSuchPaddingException;
import javax.crypto.spec.SecretKeySpec;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.encryptionsdk.AwsCrypto;
import com.amazonaws.encryptionsdk.CryptoInputStream;
import com.amazonaws.encryptionsdk.jce.JceMasterKey;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.InvalidStateException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ShutdownException;
import com.amazonaws.services.kinesis.clientlibrary.exceptions.ThrottlingException;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker;
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker.Builder;
import com.amazonaws.services.kinesis.model.Record;
import com.amazonaws.services.kms.AWSKMS;
import com.amazonaws.services.kms.AWSKMSClientBuilder;
import com.amazonaws.services.kms.model.DecryptRequest;
import com.amazonaws.services.kms.model.DecryptResult;
import com.amazonaws.util.Base64;
import com.amazonaws.util.IOUtils;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.annotations.SerializedName;
import org.bouncycastle.jce.provider.BouncyCastleProvider;
public class DemoConsumer {
private static final String STREAM_NAME = "aws-rds-das-[instance-external-resource-id]"; // aws-rds-das-db-ABCD123456
private static final String APPLICATION_NAME = "AnyApplication"; //unique application name for dynamo table generation that holds kinesis shard tracking
private static final String AWS_ACCESS_KEY = "[AWS_ACCESS_KEY_TO_ACCESS_KINESIS]";
private static final String AWS_SECRET_KEY = "[AWS_SECRET_KEY_TO_ACCESS_KINESIS]";
private static final String RESOURCE_ID = "[external-resource-id]"; // db-ABCD123456
private static final String REGION_NAME = "[region-name]"; //us-east-1, us-east-2...
private static final BasicAWSCredentials CREDENTIALS = new BasicAWSCredentials(AWS_ACCESS_KEY, AWS_SECRET_KEY);
private static final AWSStaticCredentialsProvider CREDENTIALS_PROVIDER = new AWSStaticCredentialsProvider(CREDENTIALS);
private static final AwsCrypto CRYPTO = new AwsCrypto();
private static final AWSKMS KMS = AWSKMSClientBuilder.standard()
.withRegion(REGION_NAME)
.withCredentials(CREDENTIALS_PROVIDER).build();
class Activity {
String type;
String version;
String databaseActivityEvents;
String key;
}
class ActivityEvent {
@SerializedName("class") String _class;
String clientApplication;
String command;
String commandText;
String databaseName;
String dbProtocol;
String dbUserName;
String endTime;
String errorMessage;
String exitCode;
String logTime;
String netProtocol;
String objectName;
String objectType;
List paramList;
String pid;
String remoteHost;
String remotePort;
String rowCount;
String serverHost;
String serverType;
String serverVersion;
String serviceName;
String sessionId;
String startTime;
String statementId;
String substatementId;
String transactionId;
String type;
}
class ActivityRecords {
String type;
String clusterId; // note that clusterId will contain an empty string on RDS Oracle and RDS SQL Server
String instanceId;
List databaseActivityEventList;
}
static class RecordProcessorFactory implements IRecordProcessorFactory {
@Override
public IRecordProcessor createProcessor() {
return new RecordProcessor();
}
}
static class RecordProcessor implements IRecordProcessor {
private static final long BACKOFF_TIME_IN_MILLIS = 3000L;
private static final int PROCESSING_RETRIES_MAX = 10;
private static final long CHECKPOINT_INTERVAL_MILLIS = 60000L;
private static final Gson GSON = new GsonBuilder().serializeNulls().create();
private static final Cipher CIPHER;
static {
Security.insertProviderAt(new BouncyCastleProvider(), 1);
try {
CIPHER = Cipher.getInstance("AES/GCM/NoPadding", "BC");
} catch (NoSuchAlgorithmException | NoSuchPaddingException | NoSuchProviderException e) {
throw new ExceptionInInitializerError(e);
}
}
private long nextCheckpointTimeInMillis;
@Override
public void initialize(String shardId) {
}
@Override
public void processRecords(final List records, final IRecordProcessorCheckpointer checkpointer) {
for (final Record record : records) {
processSingleBlob(record.getData());
}
if (System.currentTimeMillis() > nextCheckpointTimeInMillis) {
checkpoint(checkpointer);
nextCheckpointTimeInMillis = System.currentTimeMillis() + CHECKPOINT_INTERVAL_MILLIS;
}
}
@Override
public void shutdown(IRecordProcessorCheckpointer checkpointer, ShutdownReason reason) {
if (reason == ShutdownReason.TERMINATE) {
checkpoint(checkpointer);
}
}
private void processSingleBlob(final ByteBuffer bytes) {
try {
// JSON $Activity
final Activity activity = GSON.fromJson(new String(bytes.array(), StandardCharsets.UTF_8), Activity.class);
// Base64.Decode
final byte[] decoded = Base64.decode(activity.databaseActivityEvents);
final byte[] decodedDataKey = Base64.decode(activity.key);
Map context = new HashMap<>();
context.put("aws:rds:db-id", RESOURCE_ID);
// Decrypt
final DecryptRequest decryptRequest = new DecryptRequest()
.withCiphertextBlob(ByteBuffer.wrap(decodedDataKey)).withEncryptionContext(context);
final DecryptResult decryptResult = KMS.decrypt(decryptRequest);
final byte[] decrypted = decrypt(decoded, getByteArray(decryptResult.getPlaintext()));
// GZip Decompress
final byte[] decompressed = decompress(decrypted);
// JSON $ActivityRecords
final ActivityRecords activityRecords = GSON.fromJson(new String(decompressed, StandardCharsets.UTF_8), ActivityRecords.class);
// Iterate throught $ActivityEvents
for (final ActivityEvent event : activityRecords.databaseActivityEventList) {
System.out.println(GSON.toJson(event));
}
} catch (Exception e) {
// Handle error.
e.printStackTrace();
}
}
private static byte[] decompress(final byte[] src) throws IOException {
ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(src);
GZIPInputStream gzipInputStream = new GZIPInputStream(byteArrayInputStream);
return IOUtils.toByteArray(gzipInputStream);
}
private void checkpoint(IRecordProcessorCheckpointer checkpointer) {
for (int i = 0; i < PROCESSING_RETRIES_MAX; i++) {
try {
checkpointer.checkpoint();
break;
} catch (ShutdownException se) {
// Ignore checkpoint if the processor instance has been shutdown (fail over).
System.out.println("Caught shutdown exception, skipping checkpoint." + se);
break;
} catch (ThrottlingException e) {
// Backoff and re-attempt checkpoint upon transient failures
if (i >= (PROCESSING_RETRIES_MAX - 1)) {
System.out.println("Checkpoint failed after " + (i + 1) + "attempts." + e);
break;
} else {
System.out.println("Transient issue when checkpointing - attempt " + (i + 1) + " of " + PROCESSING_RETRIES_MAX + e);
}
} catch (InvalidStateException e) {
// This indicates an issue with the DynamoDB table (check for table, provisioned IOPS).
System.out.println("Cannot save checkpoint to the DynamoDB table used by the Amazon Kinesis Client Library." + e);
break;
}
try {
Thread.sleep(BACKOFF_TIME_IN_MILLIS);
} catch (InterruptedException e) {
System.out.println("Interrupted sleep" + e);
}
}
}
}
private static byte[] decrypt(final byte[] decoded, final byte[] decodedDataKey) throws IOException {
// Create a JCE master key provider using the random key and an AES-GCM encryption algorithm
final JceMasterKey masterKey = JceMasterKey.getInstance(new SecretKeySpec(decodedDataKey, "AES"),
"BC", "DataKey", "AES/GCM/NoPadding");
try (final CryptoInputStream decryptingStream = CRYPTO.createDecryptingStream(masterKey, new ByteArrayInputStream(decoded));
final ByteArrayOutputStream out = new ByteArrayOutputStream()) {
IOUtils.copy(decryptingStream, out);
return out.toByteArray();
}
}
public static void main(String[] args) throws Exception {
final String workerId = InetAddress.getLocalHost().getCanonicalHostName() + ":" + UUID.randomUUID();
final KinesisClientLibConfiguration kinesisClientLibConfiguration =
new KinesisClientLibConfiguration(APPLICATION_NAME, STREAM_NAME, CREDENTIALS_PROVIDER, workerId);
kinesisClientLibConfiguration.withInitialPositionInStream(InitialPositionInStream.LATEST);
kinesisClientLibConfiguration.withRegionName(REGION_NAME);
final Worker worker = new Builder()
.recordProcessorFactory(new RecordProcessorFactory())
.config(kinesisClientLibConfiguration)
.build();
System.out.printf("Running %s to process stream %s as worker %s...\n", APPLICATION_NAME, STREAM_NAME, workerId);
try {
worker.run();
} catch (Throwable t) {
System.err.println("Caught throwable while processing data.");
t.printStackTrace();
System.exit(1);
}
System.exit(0);
}
private static byte[] getByteArray(final ByteBuffer b) {
byte[] byteArray = new byte[b.remaining()];
b.get(byteArray);
return byteArray;
}
}
```
------
#### [ Python ]
```
import base64
import json
import zlib
import aws_encryption_sdk
from aws_encryption_sdk import CommitmentPolicy
from aws_encryption_sdk.internal.crypto import WrappingKey
from aws_encryption_sdk.key_providers.raw import RawMasterKeyProvider
from aws_encryption_sdk.identifiers import WrappingAlgorithm, EncryptionKeyType
import boto3
REGION_NAME = '' # us-east-1
RESOURCE_ID = '' # db-ABCD123456
STREAM_NAME = 'aws-rds-das-' + RESOURCE_ID # aws-rds-das-db-ABCD123456
enc_client = aws_encryption_sdk.EncryptionSDKClient(commitment_policy=CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT)
class MyRawMasterKeyProvider(RawMasterKeyProvider):
provider_id = "BC"
def __new__(cls, *args, **kwargs):
obj = super(RawMasterKeyProvider, cls).__new__(cls)
return obj
def __init__(self, plain_key):
RawMasterKeyProvider.__init__(self)
self.wrapping_key = WrappingKey(wrapping_algorithm=WrappingAlgorithm.AES_256_GCM_IV12_TAG16_NO_PADDING,
wrapping_key=plain_key, wrapping_key_type=EncryptionKeyType.SYMMETRIC)
def _get_raw_key(self, key_id):
return self.wrapping_key
def decrypt_payload(payload, data_key):
my_key_provider = MyRawMasterKeyProvider(data_key)
my_key_provider.add_master_key("DataKey")
decrypted_plaintext, header = enc_client.decrypt(
source=payload,
materials_manager=aws_encryption_sdk.materials_managers.default.DefaultCryptoMaterialsManager(master_key_provider=my_key_provider))
return decrypted_plaintext
def decrypt_decompress(payload, key):
decrypted = decrypt_payload(payload, key)
return zlib.decompress(decrypted, zlib.MAX_WBITS + 16)
def main():
session = boto3.session.Session()
kms = session.client('kms', region_name=REGION_NAME)
kinesis = session.client('kinesis', region_name=REGION_NAME)
response = kinesis.describe_stream(StreamName=STREAM_NAME)
shard_iters = []
for shard in response['StreamDescription']['Shards']:
shard_iter_response = kinesis.get_shard_iterator(StreamName=STREAM_NAME, ShardId=shard['ShardId'],
ShardIteratorType='LATEST')
shard_iters.append(shard_iter_response['ShardIterator'])
while len(shard_iters) > 0:
next_shard_iters = []
for shard_iter in shard_iters:
response = kinesis.get_records(ShardIterator=shard_iter, Limit=10000)
for record in response['Records']:
record_data = record['Data']
record_data = json.loads(record_data)
payload_decoded = base64.b64decode(record_data['databaseActivityEvents'])
data_key_decoded = base64.b64decode(record_data['key'])
data_key_decrypt_result = kms.decrypt(CiphertextBlob=data_key_decoded,
EncryptionContext={'aws:rds:db-id': RESOURCE_ID})
print (decrypt_decompress(payload_decoded, data_key_decrypt_result['Plaintext']))
if 'NextShardIterator' in response:
next_shard_iters.append(response['NextShardIterator'])
shard_iters = next_shard_iters
if __name__ == '__main__':
main()
```
------