RDS OracleでDatabase Activity Streamsを試してみる

Audit

Kinesis

Oracle

RDS

Scala

ツイート

こんにちは、プロダクト開発本部の川原です！前回ではRDS Oracleで標準監査ログの取得について説明しました。しかしOracle 21cでは標準監査が非推奨となり統合監査が推奨されることになります。RDS Oracleは2021年現在21cを対応していませんが、Oracleで今後標準監査が取れない可能性があります。では統合監査を取得するにはこの二つが上げられます。

1. OracleにログインをしUNIFIED_AUDIT_TRAILを参照

2. Database Activity Streams

タイトルにある通りDatabase Activity Streams をメインに話しますが、その前に 1.について軽く説明したいと思います。

OracleにログインをしUNIFIED_AUDIT_TRAILを参照

この方法は一般のOracleと同じ方法となります。取得するのが一度であるならいいのですが、定期的に取るとしたら

• UNIFIED_AUDIT_TRAILをメンテナンスするタイミングはどうするのか

• どのくらいのタイミングで取得すればいいのか

• AWSの他のサービスと連携がしにくい

と考える事がたくさんあります。

Database Activity Streams

先ほど説明したように、標準監査は非推奨になるので対策しないといけない。でも統合監査使うのは難しいと悩んでいた折、AWSがRDS OracleでDatabase Activity Streamsの提供を始めました。このDatabase Activity Streamsは Amazon Aurora Postgres、 Amazon Aurora MySQL で提供されていた機能ですが、RDS Oracleの場合は監査データに 統合監査 を使用しています。一番のネックとなっていた 監査データの取得がAWSによってサポートされたのです！統合監査データはGzipに圧縮をしAWS KMSで暗号化をしたものをBase64形式にエンコードされてKinesis Data Streams に送信されます。Kinesis Data Streams から直接監査データを取得する事も可能ですし、Kinesis Data Firehoseを使えば、S3に監査データを容易に保存できます。

Database Activity Streams を有効にする

有効にする前にRDS Oracleで次の準備が必要です。

• 監査データLのバックアップ

• 監査ポリシーの設定

Database Activity Streamsが有効になると監査データは消去されるのでバックアップをする必要があります。
また有効になる事で次の操作が出来なくなります。

• 統合監査証跡レコードの消去

• 統合監査ポリシーを追加、削除、または変更する

• 最後にアーカイブされたタイムスタンプを更新する

このため事前に監査ポリシーを定義しておかないと監査データの取得が上手く出来ない場合があります。

この制限については、AWSがデータベースユーザが自由にPOLICYを変更できるのはおかしいという目的があるようです。

Database Activity Streams の開始

Auroraの時と大体一緒ですが一点異なります。 Database Activity Streamsで取得するフィールドはAuroraとRDS Oracleで共通です。ですが、統合監査の項目はAuroraの時の項目よりも多いため共通のフィールドだけでは賄いきれません。全ての統合監査のフィールドを取得したい場合はデータベースアクティビティストリームの開始ウインドウで Oracle ネイティブ監査フィールドを含める を選択します。

RDS Oracleの変化

Database Activity Streams が有効になったことで、実際にRDS Oracleがどのように変わったのかをみてみます。

有効になる前の AUDIT_UNIFIED_ENABLED_POLICIES

POLICY_NAME の TOPLEVEL は私が作ったポリシーで TOPEVEL を全て取得する設定を与えています。
ORA_SECURECONFIG と ORA_LOGON_FAILURES はデフォルトで有効になっていたポリシーです。

有効にした AUDIT_UNIFIED_ENABLED_POLICIES

RDS_DAS_PROTECTION_POLICY が追加されました。ではこのPOLICYが何者かを観てみます。

select * from AUDIT_UNIFIED_POLICIES where POLICY_NAME = 'RDS_DAS_PROTECTION_POLICY';

どうやら、AUDIT情報を変更しようとしたユーザを監視しているようです。この事からもポリシーを変更しようとしたユーザは監査データとして取得されるようにチェックされるようになっているのが分かります。

ユーザ一覧について

Database Activity Streams を有効にする事でユーザ一覧にも変化が生じます。

有効にすることで RDSSEC が追加されます。 RDSSECユーザで統合監査データを管理しているようです。

UNIFIED AUDITの削除

RDSSEC は定期的に以下のプロシージャを実行しておりUNIFIED AUDITを削除しています。もしバックアップデータが欲しくなった場合は容易にS3へ保存できます。

BEGIN DBMS_AUDIT_MGMT.SET_LAST_ARCHIVE_TIMESTAMP(audit_trail_type => DBMS_AUDIT_MGMT.AUDIT_TRAIL_UNIFIED,last_archive_time => :1
BEGIN DBMS_AUDIT_MGMT.CLEAN_AUDIT_TRAIL(DBMS_AUDIT_MGMT.AUDIT_TRAIL_UNIFIED, true); END;

Database Activity Streamsの取得

Database Activity Streamsが開始されるとaws-rds-db-<RDSOracleのリソースID>のKinesis Data Streams が自動で作成されます。このためDBインスタンスからKinesis Data Streamsを特定出来ます。Kinesis Data StreamsはKCLを使用する事で容易に取得できますが暗号化されているため複合する必要があります。公式ドキュメントでは取得するまでのサンプルコードが掲載されていないです。手順としては、AuroraのDatabase Activity Streamsのサンプルコードとほぼ一緒ですが、使用しているライブラリがAWS SDK v1のため、AWS SDK v2で行っていきます。JDKは8以上を使用していきます。言語はScalaを使用してfs2を使って取得していきます。ソースコードはこちらに載せています。

Kinesis Data Streamsからレコードの取得

Kinesis Data Streamsから取得されるデータは次のようなフィールドを持つJSONデータで送られます。

{
  "type":"DatabaseActivityMonitoringRecords",
  "version":"1.3",
  "databaseActivityEvents":"encrypted audit records",
  "key":"encrypted key"
}

databaseActivityEvents がGzip形式で圧縮して暗号化されたものをBase64でエンコードされたものとなります。keyが複合化に必要なバイト列をBase64でエンコードされています。

キーの復号リクエスト作成

  private def decryptRequest(
      dbResourceId: String, bytes: Array[Byte]): DecryptRequest =
    DecryptRequest.builder().ciphertextBlob(SdkBytes.fromByteArray(bytes))
      .encryptionContext(Map("aws:rds:db-id" -> dbResourceId).asJava)
      .build()

dbResourceIdはRDSOracleのリソースID、bytesは先ほどのkeyをBase64でデコードしたByte列を指します。encryptionContextはaws:rds:db-id を指定します。 Auroraの場合は aws:rds:dbc-idと違っています。このコンテキストキーを特定するのに苦労しました。

復号

  private def decrypt[F[_] : Sync](
      awsCrypto: AwsCrypto,
      decoded: Array[Byte],
      decodedDataKey: Array[Byte]
  ) = { // Create a JCE master key provider using the random key and an AES-GCM encryption algorithm

    import com.amazonaws.encryptionsdk.jce.JceMasterKey

    import javax.crypto.spec.SecretKeySpec
    val masterKey = JceMasterKey.getInstance(new SecretKeySpec(decodedDataKey, "AES"), BouncyCastleProvider.PROVIDER_NAME, "DataKey", "AES/GCM/NoPadding")

    def cryptInputStream(
        decoded: Array[Byte],
    ) = Resource.fromAutoCloseable {
      Sync[F].blocking(awsCrypto.createDecryptingStream(masterKey, new ByteArrayInputStream(decoded)))
    }

    def byteArrayOutputStream() = Resource.fromAutoCloseable {Sync[F].blocking{new ByteArrayOutputStream()}}

    def copy(in: Array[Byte]) = (for {
      inStream <- cryptInputStream(in)
      outStream <- byteArrayOutputStream()
    } yield (inStream, outStream)).use { case (inStream, outStream) =>
      Sync[F].blocking(IoUtils.copy(inStream, outStream)) *> Sync[F].delay(outStream.toByteArray)
    }

    copy(decoded)
  }

AWSCryptoは AWS Encryption SDKを使っています。インスタンスの作成は次のようにやっています。

  val crypto = AwsCrypto.builder().withCommitmentPolicy(CommitmentPolicy.ForbidEncryptAllowDecrypt).build()

decodedはdatabaseActivityEventsのBase64でデコードしたものでdecodedDataKeyは KMSでkeyを復号したものになります。

解凍

復号したものはGzipで圧縮されているので解凍します

  private def decompress[F[_] : Sync](src: Array[Byte]) = {
    Stream.emits(src).through(Compression[F].gunzip()).flatMap { _.content }
      .compile.toList.map(_.toArray)
  }

fs2のGzipメソッドを使って解凍しました

JSONのマッピング

何気にこれが苦労します。取得したデータは100くらいのフィールドがあります。統合監査のフィールドであるEngineNativeAuditFieldsはSNAKE_CASEで書かれているためマッピングするのに注意が必要です。

AuditLogの例

またJSONフィールドは増えたり減る可能性があるのでその対策もする必要があります。

まとめ

いかがだったでしょうか？ 統合監査でネックだった監査データの取得もDatabase Activity Streamsでより簡単でセキュアで取得出来るのが分かったと思います。今まで監査データはDB毎に監査ログのフォーマットが異なっていたので解析するのもひと苦労で、DBを移行するにも監査が出来ないなどが理由で選択肢から外れる事があったかと思いますが、それも外れていきそうです。ただDatabase Activity Streamsも完璧ではなくてCDBインスタンスに対してはまだ対応していない。 RDSSECユーザを除外指定出来ないなどのデメリットがあります。Database Activity Streamsは監査製品で対応しているものも存在しています。もし興味がわけば是非試してみてください。

お待ちしております

インサイトテクノロジーはデータ活用に興味を持つエンジニアを募集しています。もしこの記事で興味を持たれた方はこちらで応募をお待ちしております。