データベース監査のセットアップ

システム・ログ・ファイルのデータベース・アクティビティーを捕捉する監査ロガーを構成します。

audit_logging_options:
    enabled: true
    logger: SLF4JAuditWriter

データベース・テーブルのデータベース・アクティビティーを捕捉する監査ロガーを構成します。

1. audit_logging_optionsをdse.yaml ファイルで設定します。

   audit_logging_options:
       enabled: true
       logger: CassandraAuditWriter
   #    included_categories:
   #    excluded_categories:
   #    included_keyspaces:
   #    excluded_keyspaces:
   #    included_roles:
   #    excluded_roles:
       retention_time: 12
   
       cassandra_audit_writer_options:
           mode: sync
           batch_size: 50
           flush_time: 250
           queue_size: 30000
           write_consistency: QUORUM
   #         dropped_event_log: /var/log/cassandra/dropped_audit_events.log
   #         day_partition_millis: 3600000

   • enabled: true - 次回の起動後にロギングをオンにします。
   • logger: CassandraAuditWriter - ロガー名。
   • retention_time: 12 - dse_audit.audit_logテーブルでTTL（Time-To-Live）として設定する時間数。データを自動的に期限切れにするには、この設定を使用します。デフォルトは0（無効）です。
   • cassandra_audit_writer_optionsパラメーターを必要に応じてカスタマイズします（デフォルトは上記のとおり）。

   注: audit_logテーブルには以下の設定があります。

   DESC TABLE dse_audit.audit_log
   
   CREATE TABLE dse_audit.audit_log (
       date timestamp,
       node inet,
       day_partition int,
       event_time timeuuid,
       authenticated text,
       batch_id uuid,
       category text,
       consistency text,
       keyspace_name text,
       operation text,
       source text,
       table_name text,
       type text,
       username text,
       PRIMARY KEY ((date, node, day_partition), event_time)
   ) WITH CLUSTERING ORDER BY (event_time ASC)
       AND bloom_filter_fp_chance = 0.01
       AND caching = {'keys': 'ALL', 'rows_per_partition': 'NONE'}
       AND comment = ''
       AND compaction = {'class': 'org.apache.cassandra.db.compaction.TimeWindowCompactionStrategy', 'max_threshold': '32', 'min_threshold': '4'}
       AND compression = {'chunk_length_in_kb': '64', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'}
       AND crc_check_chance = 1.0
       AND default_time_to_live = 0
       AND gc_grace_seconds = 7776000
       AND max_index_interval = 2048
       AND memtable_flush_period_in_ms = 3600000
       AND min_index_interval = 128
       AND nodesync = {'enabled': 'true'}
       AND speculative_retry = '99PERCENTILE';

2. （オプション）デフォルトでは、NodeSyncはaudit_logテーブルに対して無効になっています。NodeSyncを使用してデータの一貫性を維持するには、これを有効にします。

   ALTER TABLE dse_audit.audit_log WITH 
   nodesync = { 'enabled' : true };

3. キースペース・レプリケーション・ストラテジをNetworkTopologyStrategyに設定し、監査が有効になっているクラスター内の各データ・センターのレプリケーション係数を設定します。

   ALTER KEYSPACE dse_audit WITH 
   replication = {
                  'class': 'NetworkTopologyStrategy', 
                  'datacenter1' : 2, 
                  'datacenter2' : 1   };

4. 次のいずれかの方法を使用して、新しく追加されたデータ・センターにデータ・レプリケーションを強制します。
   • NodeSyncが有効になっている場合：

     nodesync -cu user_name -cp password \
     validation submit dse_audit.audit_log

     user_nameは、CQLテーブルにアクセスする権限を持つ内部ログイン・ロール名またはLDAP/Kerberosです。
   • NodeSyncが無効になっている場合：

     nodetool repair dse_audit audit_log

ログに書き込まれる前に、機密データをマスクするようにDSE監査ライターを構成します。

1. テキスト・エディターで logback.xml ファイルを開きます。
2. パターン要素を特定します。

   <!--audit log-->  
   <appender name="SLF4JAuditWriterAppender" class="ch.qos.logback.core.rolling.RollingFileAppender">
       <file>${cassandra.logdir}/audit/audit.log</file> <!-- logfile location -->
       <encoder>
         <pattern>%-5level [%thread] %date{ISO8601} %F:%L - %msg%n</pattern> <!-- the layout pattern used to format log entries -->
         <immediateFlush>true</immediateFlush> 
       </encoder>
       <rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
         <fileNamePattern>${cassandra.logdir}/audit/audit.log.%i.zip</fileNamePattern>
         <minIndex>1</minIndex>
         <maxIndex>20</maxIndex> <!-- max number of archived logs that are kept -->
       </rollingPolicy>
       <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
         <maxFileSize>200MB</maxFileSize> <!-- The size of the logfile that triggers a switch to a new logfile, and the current one archived -->
       </triggeringPolicy>
     </appender>
   <logger name="SLF4JAuditWriter" level="INFO" additivity="false">
       <appender-ref ref="SLF4JAuditWriterAppender"/>
     </logger>

3. メッセージ・テキストに置換機能を使用して、最初の用語を2番目の用語に置き換えます。たとえば、任意の文字に設定されたパスワードを検索し、そのパスワード文字をxxxxxに置き換える場合は、次のようにします。

   %replace(%msg){"password\s*=\s*'.*'", "password='xxxxx'"}

   注: DataStaxは、CREATE ROLEコマンドで指定されたパスワードを自動的にマスクします。
4. ノードを再起動するか、構成が自動的に再度読み込まれるのを待ちます。

SLF4JAuditWriterログ・ファイルの場所、名前、ロール頻度、および名前を変更します。

<!--audit log-->  
<appender name="SLF4JAuditWriterAppender" class="ch.qos.logback.core.rolling.RollingFileAppender">
    <file>${cassandra.logdir}/audit/audit.log</file> <!-- logfile location -->
    <encoder>
      <pattern>%-5level [%thread] %date{ISO8601} %F:%L - %msg%n</pattern> <!-- the layout pattern used to format log entries -->
      <immediateFlush>true</immediateFlush> 
    </encoder>
    <rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
      <fileNamePattern>${cassandra.logdir}/audit/audit.log.%i.zip</fileNamePattern>
      <minIndex>1</minIndex>
      <maxIndex>20</maxIndex> <!-- max number of archived logs that are kept -->
    </rollingPolicy>
    <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
      <maxFileSize>200MB</maxFileSize> <!-- The size of the logfile that triggers a switch to a new logfile, and the current one archived -->
    </triggeringPolicy>
  </appender>
<logger name="SLF4JAuditWriter" level="INFO" additivity="false">
    <appender-ref ref="SLF4JAuditWriterAppender"/>
  </logger>

1. テキスト・エディターで logback.xml ファイルを開きます。
2. データ監査を構成するには、プロパティを変更します。
   • file - ロケーション・システム上のログファイルのパス。
   • pattern - ログ・メッセージを書き込む変数。機密データを非表示にするには、「監査ログの機密データの置換」を参照してください。
   • immediateFlush - イベント・メッセージをすぐにログに書き込むかどうか。
   • fileNamePattern - ログ・ファイルがロールされるときの名前の規則。
   • maxFileSize - ログをロールするしきい値（最大ログ・サイズ）。
3. ノードを再起動するか、構成が自動的に再度読み込まれるのを待ちます。

認証デバッグ・メッセージをdebug.logに捕捉します。

1. logback.xml ファイルを見つけて、その末尾に以下の行を含めます。

      <logger name="com.datastax.bdp.transport.server" level="TRACE"/>
      <logger name="com.datastax.bdp.cassandra.auth" level="TRACE"/>

   ローカル・ノードからのアクティビティーのみが記録されます。認証のトラブル・シューティングを行うノードで認証トレースを有効にします。
2. ノードを再起動するか、構成が自動的に再度読み込まれるのを待ちます。
3. debug.logを追跡します。

   tail -f /etc/dse/cassandra/debug.log

4. 外部サーバーのアカウントを使用してログインします。

TRACE [IOThread-0] 2018-04-24 21:07:26,565  LdapUtils.java:484 - [ldap-fetch-user] username: dantest1 connection: org.apache.directory.ldap.client.api.LdapNetworkConnection@430680df
TRACE [IOThread-0] 2018-04-24 21:07:26,566  LdapUtils.java:499 - [ldap-fetch-user] bind to connection
TRACE [IOThread-0] 2018-04-24 21:07:26,568  LdapUtils.java:508 - [ldap-fetch-user] user_search_base: ou=users,dc=qaldap,dc=datastax,dc=lan, user_search_filter: (uid=dantest1)
TRACE [IOThread-0] 2018-04-24 21:07:26,608  LdapUtils.java:517 - [ldap-fetch-user] found entry for username: dantest1
TRACE [IOThread-0] 2018-04-24 21:07:26,610  LdapUtils.java:433 - [ldap-bind] userDN: cn=dantest1,ou=users,dc=qaldap,dc=datastax,dc=lan connection: org.apache.directory.ldap.client.api.LdapNetworkConnection@430680df
TRACE [IOThread-0] 2018-04-24 21:07:26,611  LdapUtils.java:438 - [ldap-bind] SUCCESS - bind succcessful for userDN: cn=dantest1,ou=users,dc=qaldap,dc=datastax,dc=lan
TRACE [IOThread-0] 2018-04-24 21:07:26,613  LdapUtils.java:325 - [ldap-authenticate] SUCCESS - username: dantest1, userDN: cn=dantest1,ou=users,dc=qaldap,dc=datastax,dc=lan

TRACE [IOThread-0] 2018-04-24 21:18:23,434  LdapUtils.java:303 - [ldap-authenticate] username: dantest1 not found in cache
TRACE [IOThread-0] 2018-04-24 21:18:23,434  LdapUtils.java:484 - [ldap-fetch-user] username: dantest1 connection: org.apache.directory.ldap.client.api.LdapNetworkConnection@430680df
TRACE [IOThread-0] 2018-04-24 21:18:23,434  LdapUtils.java:499 - [ldap-fetch-user] bind to connection
TRACE [IOThread-0] 2018-04-24 21:18:23,437  LdapUtils.java:508 - [ldap-fetch-user] user_search_base: ou=users,dc=qaldap,dc=datastax,dc=lan, user_search_filter: (uid=dantest1)
TRACE [IOThread-0] 2018-04-24 21:18:23,440  LdapUtils.java:517 - [ldap-fetch-user] found entry for username: dantest1
TRACE [IOThread-0] 2018-04-24 21:18:23,441  LdapUtils.java:433 - [ldap-bind] userDN: cn=dantest1,ou=users,dc=qaldap,dc=datastax,dc=lan connection: org.apache.directory.ldap.client.api.LdapNetworkConnection@430680df
TRACE [IOThread-0] 2018-04-24 21:18:23,451  LdapUtils.java:447 - [ldap-bind] ERROR - bind failed for userDN: cn=dantest1,ou=users,dc=qaldap,dc=datastax,dc=lan
...