プロパティファイルを使用した監査ログの構成

このトピックでは、server.properties ファイルを使用して監査ログを構成する方法について説明します。

監査ログの構成

監査ログを構成するには、次の各セクションの説明に従って server.properties ファイルを使用することも、以下を使用することもできます。

システムに監査ログをセットアップする前に、次のデフォルト設定をご確認ください。

デフォルトで有効:

  • トピック: 作成と削除
  • ACL: 作成と削除
  • RBAC に関連する認可リクエスト
  • MANAGEMENT カテゴリ のすべてのイベント

デフォルトで無効:

  • ブローカー間のやりとり
  • 生成と消費に関連したイベント
  • ハートビートイベントと DESCRIBE イベント

詳細な監査ログ構成

ほとんどの組織には、厳格で具体的なセキュリティガバナンスの要件とポリシーが存在します。その場合は、監査ログをより詳細なレベルで構成できます。具体的には、以下のような構成が可能です。

  • どのイベントカテゴリをキャプチャするか(デフォルトで無効になっている生成、消費、ブローカー間などのカテゴリを含む)
  • 重要度の異なるログをキャプチャするための複数のトピック
  • セキュリティおよびパフォーマンス用に最適化されたトピックデスティネーションのルート
  • 組織のニーズに対応する保持期間
  • メッセージの量が多すぎてもパフォーマンスを低下させないための 除外プリンシパル
  • 監査ログクラスターとやり取りするための Kafka ポート

生成および消費について監査ログを有効にするときは、ログに記録するイベントを厳選し、特に注意を要するトピックのみを対象としてログを構成します。

注釈

特定のトピックに対する READ または WRITE の認可を受けていないクライアントには、DESCRIBE の認可もないことがきわめて一般的です。詳細については「生成および消費の拒否が監査ログに表示されない」を参照してください。

../../_images/audit-log-recommended-config.ja.png

Kafka server.properties で監査ログを有効にした後、confluent.security.event.router.config を設定して、次の詳細な監査ログ構成例に示すような JSON 構成にします。

{
    "destinations": {
        "bootstrap_servers": [
            "audit.example.com:9092"
        ],
        "topics": {
            "confluent-audit-log-events_denied": {
                "retention_ms": 2592000000
            },
            "confluent-audit-log-events_secure": {
                "retention_ms": 7776000000
            },
            "confluent-audit-log-events": {
                "retention_ms": 2592000000
            }
        }
    },
    "default_topics": {
        "allowed": "confluent-audit-log-events",
        "denied": "confluent-audit-log-events"
    },
    "excluded_principals": [
        "User:Alice",
        "User:service_account_id"
    ],
    "routes": {
        "crn://mds1.example.com/kafka=*/topic=secure-*": {
            "produce": {
                "allowed": "confluent-audit-log-events_secure",
                "denied": ""
            },
            "consume": {
                "allowed": "confluent-audit-log-events_secure",
                "denied": "confluent-audit-log-events_denied"
            }
        }
    }
}

Kafka のダイナミック構成 を使用すると、クラスターを再起動せずに confluent.security.event.router.config 設定を更新できます。ルーター構成は複雑な構造であるため、(server.properties の場合とまったく同じように)目的の値で .properties ファイルを作成し、kafka-configs を使用して構成を追加する必要があります。

bin/kafka-configs --bootstrap-server localhost:9092 --entity-type brokers --entity-default \
    --alter --add-config-file new.properties

ダイナミック構成は server.properties の構成より優先されるため、後で server.properties ファイルの値を変更してクラスターを再起動すると、kafka-configs を使用してダイナミック構成を削除するまで、ダイナミック構成が引き続き使用されます。

監査ログのデスティネーションの構成

destinations オプションでは、ブートストラップサーバーによって提供される監査ログクラスターを識別します。この設定は、監査ログクラスターと Kafka の間の通信チャネルを識別するために使用します。bootstrap_server 設定を使用すると、監査ログメッセージの保持のみを目的として確保されているクラスターに監査ログメッセージを配信できます。これにより、組織の監査ログへのアクセスや改ざんを防止し、機密性の低いデータのログ量を抑えながら、機密性データの詳細な監査を選択的に実行できます。

監査ログを別のクラスターに配信する場合は、そのクラスターへの接続を構成する必要があります。この接続の構成は、適切な認証情報を含め、プロデューサー構成キーのプレフィックスとして confluent.security.event.logger.exporter.kafka を使用し、クラスターへの書き込みを行うプロデューサーの場合と同じように行います。

たとえば Kafka クラスターで、ホスト audit.example.com のポート 9092 をリッスンしていて、そのクラスターで受け入れる認証方式が SCRAM-SHA-256 認証であり、confluent-audit という名前のプリンシパルが監査ログトピックへの接続と生成を許可されている場合、構成は次のようになります。

confluent.security.event.router.config=\
{ \
    "destinations": { \
        "bootstrap_servers": ["audit.example.com:9092"], \
        "topics": { \
            "confluent-audit-log-events": { \
                "retention_ms": 7776000000
            } \
        } \
    }, \
    "default_topics": { \
        "allowed": "confluent-audit-log-events", \
        "denied": "confluent-audit-log-events" \
    } \
}
confluent.security.event.logger.exporter.kafka.sasl.mechanism=SCRAM-SHA-256
confluent.security.event.logger.exporter.kafka.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
    username="confluent-audit" \
    password="secretP@ssword123";

ブートストラップサーバーは、ルーター構成の JSON またはプロデューサー構成のプロパティのいずれかで指定できます。両方の場所に出現した場合は、ルーター構成が優先されます。

監査ログトピックの作成

ほとんどの組織では、特定のログ記録の目的でトピックが作成されます。単一の Kafka クラスター上で、複数のトピックにさまざまな種類の監査ログメッセージを送信できます。各トピックの保持期間を指定することもできます。たとえば、ログ生成メッセージとログ管理メッセージを同じトピックに送信する場合、管理メッセージの保持期間を 30 日に設定しながら、生成メッセージの保持期間を 1 日に設定することはできません。これらの保持期間を指定するには、それぞれのメッセージを別々のトピックに送信する必要があります。

重要

監査ログ構成で指定したトピック保持期間は、まだ存在していないトピックが監査ログ機能によって自動的に作成される場合にのみ適用されます。既に存在するトピックの構成でこの値の編集を試みないでください。保持期間の更新は適用されません。CLI を使用して既存の監査ログトピックの保持期間を変更する方法については、「レプリケーション係数と保持期間の構成」を参照してください。

すべてのトピックに、プレフィックスとして confluent-audit-log-events を使用する必要があります。この例では、以下のトピックが指定されています。

  • confluent-audit-log-events_denied: 名前がプレフィックス secure- で始まるトピックへの消費リクエストが拒否された際のログメッセージはすべて、このトピックに送信されます。
  • confluent-audit-log-events_secure: プレフィックス secure- が使用されているトピックへの生成および消費リクエストが許可された際のログメッセージはすべて、このトピックに送信されます。
  • confluent-audit-log-events: 許可/拒否を問わず、管理カテゴリおよび認可カテゴリのログメッセージはすべて、このトピックに送信されます。

まだ作成されていないトピックをトピック構成に含めることもできます。その場合、監査ログプリンシパルに必要なアクセス許可があれば、トピックが自動的に作成され、構成ファイルに指定されている保持期間が使用されます。ただし、この構成ファイルを編集するだけで既存のトピックの保持期間を変更することはできません。既存のトピックの保持期間の変更については、「レプリケーション係数と保持期間の構成」を参照してください。

トピックのルーティング

監査ログメッセージは、サブジェクト、カテゴリ、認可の可否(許可されたか、拒否されたか)に基づいて、さまざまなトピックに柔軟にルーティングできます。特定サブジェクトのルート指定や、プレフィックス(機密性の高い認可イベントの場合は _secure-* を使用するなど)とワイルドカードの使用により、より大きなリソースグループを扱うこともできます。

上のルーティング例では、機密性の高いトピック用に作成されたプレフィックス(topic=secure-*)が使用され Kafka のワイルドカードパターン(kafka=*)が含まれている特定の CRN(crn://mds1.example.com/kafka=*/topic=secure-*)を示しています。この CRN パターンに一致する、許可されたすべての生成認可イベントログは、トピック confluent-audit-log-events_secure にルーティングされます。この CRN パターンに一致する、拒否されたすべての生成認可イベントログは、空の文字列("")による指定に従い、どこにもルーティングされません。この CRN パターンに一致する、許可されたすべての消費認可イベントログはトピック confluent-audit-log-events_secure にルーティングされ、この CRN パターンに一致する、拒否されたすべての消費認可イベントログはトピック confluent-audit-log-events_denied にルーティングされます。

監査ログの保護

監査ログを保護するには、構成に関する以下の推奨事項を検討します。

  • セキュリティを強化するには、監査ログを別のクラスターに送信します。監査ログクラスターのアクセス許可を制限して、監査ログの信頼性を確保します。
  • アクセス許可が制限された監査ログプリンシパルを作成します(confluent-audit で識別されるユーザーなど)。デフォルトでは、包括的なアクセス許可を持つブローカープリンシパルが使用されます。

RBAC のロールバインディングを使用して監査ログプリンシパルのアクセス許可をセットアップするには:

bin/confluent iam rolebinding create \
  --principal User:confluent-audit \
  --role DeveloperWrite \
  --resource Topic:confluent-audit-log-events \
  --prefix \
  --kafka-cluster-id <cluster id>

ACL を使用してアクセス許可を設定することもできます。

bin/kafka-acls --bootstrap-server localhost:9092 \
  --add \
  --allow-principal User:confluent-audit \
  --allow-host '*' \
  --producer  \
  --topic confluent-audit-log-events \
  --resource-pattern-type prefixed

監査ログ構成例

このセクションでは、特定のユースケースに対応する追加の監査ログ構成例をご紹介します。

詳細なトピックルーティングのユースケース

次の構成例は、クラスター内の請求データと給与データのルーティング方法を示しています。このユースケースでは、すべての請求トピックでプレフィックス billing- が使用され、すべての給与トピックでプレフィックス payroll- が使用されています。これら 2 つの部門には別々の監査人が存在し、詳細な監査ログ情報を必要としています。特定のトピックに関連する監査ログメッセージを正しい監査人にのみルーティングするには、次のような構成を使用します。

{
    "destinations": {
        "bootstrap_servers": [
            "audit.example.com:9092"
        ],
        "topics": {
            "confluent-audit-log-events": {
                "retention_ms": 2592000000
            },
            "confluent-audit-log-events_payroll": {
                "retention_ms": 2592000000
            },
            "confluent-audit-log-events_billing": {
                "retention_ms": 2592000000
            }
        }
    },
    "default_topics": {
        "allowed": "confluent-audit-log-events",
        "denied": "confluent-audit-log-events"
    },
    "routes": {
        "crn://mds1.example.com/kafka=*/topic=billing-*": {
            "produce": {
                "allowed": "confluent-audit-log-events_billing",
                "denied": "confluent-audit-log-events_billing"
            },
            "consume": {
                "allowed": "confluent-audit-log-events_billing",
                "denied": "confluent-audit-log-events_billing"
            },
            "management": {
                "allowed": "confluent-audit-log-events_billing",
                "denied": "confluent-audit-log-events_billing"
            }
        },
        "crn://mds1.example.com/kafka=*/topic=payroll-*": {
            "produce": {
                "allowed": "confluent-audit-log-events_payroll",
                "denied": "confluent-audit-log-events_payroll"
            },
            "consume": {
                "allowed": "confluent-audit-log-events_payroll",
                "denied": "confluent-audit-log-events_payroll"
            },
            "management": {
                "allowed": "confluent-audit-log-events_payroll",
                "denied": "confluent-audit-log-events_payroll"
            }
        }
    }
}

特定のイベントタイプのメッセージだけをキャプチャするルーター構成を作成することはできません。たとえば、すべての kafka.JoinGroup メッセージをキャプチャするようにルーティングを構成することはできません。ただし、目的のイベントタイプのカテゴリがわかっている場合は、カテゴリレベルでルーター構成を作成し、ksqlDB、Spark、ElasticSerach、Splunk などのツールを使用して結果をフィルター処理することができます。

kafka.JoinGroup イベントは consume カテゴリに属しています。これについてすべての監査ログメッセージのルーティングを構成するには、次のように指定します。

"routes": {
  "crn:///kafka=*/group=*": {
    "consume": {
      "allowed": "confluent-audit-log-events_group_consume",
      "denied": "confluent-audit-log-events_group_consume"
    }
  }
}

注釈

ここで crn:///kafka=/group= が使用されているのは、kafka.JoinGroup がトピックではなくグループを対象としているためです。

この例は、ユーザーがテストクラスター test-cluster-1 でランダムな更新を頻繁に行う構成を示しています。他のクラスターからのログを受信する一方で、このテストクラスターからのログが表示されたり受信することのないようにトピックのルーティングを構成するには:

{
    "destinations": {
        "bootstrap_servers": [
            "audit.example.com:9092"
        ],
        "topics": {
            "confluent-audit-log-events": {
                "retention_ms": 2592000000
            }
        }
    },
    "default_topics": {
        "allowed": "confluent-audit-log-events",
        "denied": "confluent-audit-log-events"
    },
    "routes": {
        "crn://mds1.example.com/kafka=test-cluster-1": {
            "management": {
                "allowed": "",
                "denied": ""
            },
            "authorize": {
                "allowed": "",
                "denied": ""
            }
        }
    }
}

注釈

別の方法として、このテストクラスターに confluent.security.event.logger.enable=false を指定することもできます。

この構成例は、test-1234 のような名前を使用してトピックを作成および削除するテスト自動化クラスターを対象としています。これらのトピックの監査ログメッセージは作成せず、他のトピックからの監査ログメッセージを生成するには次のようにします。

{
    "destinations": {
        "bootstrap_servers": [
            "audit.example.com:9092"
        ],
        "topics": {
            "confluent-audit-log-events": {
                "retention_ms": 2592000000
            }
        }
    },
    "default_topics": {
        "allowed": "confluent-audit-log-events",
        "denied": "confluent-audit-log-events"
    },
    "routes": {
        "crn://mds1.example.com/kafka=*/topic=test-*": {
            "management": {
                "allowed": "",
                "denied": ""
            },
            "authorize": {
                "allowed": "",
                "denied": ""
            }
        }
    }
}

悪意のある攻撃が頻繁に発生する場合は、攻撃者用に特定のトピックを作成できます。この例では、そのトピック名は honeypot です。この構成では、使用パターンの検出を試みるために、honeypot 専用の監査ログをセットアップします。ほとんどのログの保持期間は 1 日ですが、この例では、ハッカーであると疑われるユーザーに関するログを 1 年間保持します。

{
    "destinations": {
        "bootstrap_servers": [
            "audit.example.com:9092"
        ],
         "topics": {
             "confluent-audit-log-events_users": {
                 "retention_ms": 86400000
             },
             "confluent-audit-log-events_hackers": {
                 "retention_ms": 31622400000
             }
         }
    },
    "default_topics": {
        "allowed": "confluent-audit-log-events_users",
        "denied": "confluent-audit-log-events_users"
    },
    "routes": {
        "crn://mds1.example.com/kafka=*/topic=honeypot": {
            "produce": {
                "allowed": "confluent-audit-log-events_hackers",
                "denied": "confluent-audit-log-events_hackers"
            },
            "consume": {
               "allowed": "confluent-audit-log-events_hackers",
               "denied": "confluent-audit-log-events_hackers"
            },
            "management": {
               "allowed": "confluent-audit-log-events_hackers",
               "denied": "confluent-audit-log-events_hackers"
            }
        }
    }
}

監査ログのパーティション分割

監査ログトピックで、メッセージに対して予測可能な順序付けが必要である場合は、単一のパーティションのトピックに移動するようにメッセージを構成できます。

bin/kafka-topics --bootstrap-server localhost:9093 --command-config config/client.properties --create --topic confluent-audit-log-events-ordered --partitions 1

注釈

メッセージの量が非常に多い場合(特に、生成または消費が有効になっている場合)、この構成では持続可能性に欠ける可能性があります。そのような場合は、複数のパーティションで構成される 1 つ以上のトピックに監査ログメッセージを送信する必要があります。その際に、これらのメッセージを読み取るコンシューマーは、送信された順序でメッセージを受信できない可能性がある点に注意してください。

内部トピックの監査ログの抑制

生成および消費の監査ログを検討している場合は、生成される可能性のある監査ログメッセージの数の観点から、影響を慎重に検討してください。このレベルの詳細情報が必要となるトピックまたはトピックプレフィックスを特定できる場合は、それを指定し、それ以外を監査対象から除外することで、シグナルとノイズの比率を妥当なレベルに維持できます。生成と消費の監査ログを有効にする必要があり、次のようなルートを使用している場合:

"crn:///kafka=*/topic=*": {
   "produce": {
     "allowed": "confluent-audit-log-events",
     "denied": "confluent-audit-log-events"
   },
   "consume": {
     "allowed": "confluent-audit-log-events",
     "denied": "confluent-audit-log-events"
   }
 }

Kafka と Confluent Platform では、キャプチャする必要のないさまざまな目的で内部トピックが使用されることに注意してください。Kafka の内部トピックにはプレフィックス _ が付き、Confluent Platform の内部トピックにはプレフィックス _confluent- が付きます。これらのプレフィックスを除外するには、次の構成を使用します。

"crn:///kafka=*/topic=_*": {
    "consume": {
        "allowed": "",
        "denied": ""
    },
    "produce": {
        "allowed": "",
        "denied": ""
    },
    "management": {
        "allowed": "",
        "denied": ""
    }
  }

オプションで、次の構成を使用することもできます。

"crn:///kafka=*/topic=_confluent-*": {
     "consume": {
         "allowed": "",
         "denied": ""
     },
     "produce": {
         "allowed": "",
         "denied": ""
     },
     "management": {
         "allowed": "",
         "denied": ""
     }
   }

生成または消費の監査ログで topic=* を使用し、監査ログを同じクラスターに送信している場合(推奨されていません)、監査ログトピック自体の監査ログ記録は必ず無効にしてください。そうしないと、フィードバックループに関する大量の警告がブローカーログから返されることになります。

"crn:///kafka=*/topic=confluent-audit-log-events*": {
     "consume": {
         "allowed": "",
         "denied": ""
     },
     "produce": {
         "allowed": "",
         "denied": ""
     },
     "management": {
         "allowed": "",
         "denied": ""
     }
   }

レプリケーション係数と保持期間の構成

次の例は、特定のレプリケーション係数と保持期間を使用してログ記録されるトピックを CLI で作成する方法を示しています。

bin/kafka-topics --bootstrap-server localhost:9092 \
  --create --topic confluent-audit-log-events\
  --replication-factor 3 \
  --config retention.ms=7776000000

Kafka ブローカーによって監査ログトピックが自動的に作成される場合、デフォルトでは、使用されるレプリケーション係数はクラスター設定の default.replication.factor によって制御されます。

監査ログトピックのレプリケーション係数をデフォルト以外の値にする必要がある場合は、confluent.security.event.logger.exporter.kafka.topic.replicas オプションを使用してレプリケーション係数を指定できます。

confluent.security.event.logger.exporter.kafka.topic.replicas=2

Kafka ブローカーでのレプリケーション係数の自動構成の詳細については、「レプリケーション係数の構成」を参照してください。

手動でトピックを作成する際に指定した retention.ms が、router.config で指定されている保持期間と競合する場合は、手動で作成した設定が使用されます。デフォルトの保持期間は 90 日(7776000000 ミリ秒)です。

注釈

監査ログ構成で指定した保持期間は、まだ存在していないトピックが監査ログ機能によって自動的に作成される場合にのみ適用されます。既に存在するトピックの監査ログ構成でこの値を編集しても、そのトピックの保持設定は変更されません。

次の例は、CLI を使用して既存のトピックの保持期間を変更する方法を示しています。

bin/kafka-topics --bootstrap-server localhost:9092 \
  --alter --topic confluent-audit-log-events\
  --config retention.ms=12096000000000

監査ログの構成リファレンス

Confluent Platform の監査ログでは、プロデューサーの作成に使用される Kafka プロデューサー構成 がサポートされます。プロデューサー構成オプションのいずれかを監査ログに使用する場合は、プレフィックス confluent.security.event.logger.exporter. を含める必要があります。

confluent.security.event.logger.authentication.enable

認証イベントの監査ログを有効にします。

  • 型: ブール値
  • デフォルト: false
  • 重要度: 高

confluent.security.event.logger.enable

イベントロガーを有効にするかどうかを指定します。

  • 型: ブール値
  • デフォルト: true
  • 重要度: 高

confluent.security.event.logger.exporter.kafka.topic.create

イベントログトピックがまだ存在していない場合は作成します。

  • 型: ブール値
  • デフォルト: true
  • 重要度: 低

confluent.security.event.logger.exporter.kafka.topic.partitions

イベントログトピック内のパーティション数を指定します。この構成オプションは、まだ存在しないトピックの作成時(confluent.security.event.logger.exporter.kafka.topic.create=true)にのみ使用してください。トピックが既に存在する場合にこの構成を実装しようとしても、何も変更されません。

  • 型: int
  • デフォルト: 12
  • 重要度: 低

confluent.security.event.logger.exporter.kafka.topic.replicas

監査ログトピックのレプリケーション係数を指定します。この構成オプションは、まだ存在しないトピックの作成時(confluent.security.event.logger.exporter.kafka.topic.create=true)にのみ使用してください。トピックが既に存在する場合にこの構成を実装しようとしても、何も変更されません。

  • 型: int
  • デフォルト: 3
  • 重要度: 低

confluent.security.event.logger.exporter.kafka.topic.retention.bytes

イベントログトピックの保持バイト数を指定します。この構成オプションは、まだ存在しないトピックの作成時(confluent.security.event.logger.exporter.kafka.topic.create=true)にのみ使用してください。トピックが既に存在する場合にこの構成を実装しようとしても、何も変更されません。

  • 型: long
  • デフォルト: -1(無制限)
  • 重要度: 低

confluent.security.event.logger.exporter.kafka.topic.retention.ms

イベントログトピックの保持時間を指定します。この構成オプションは、まだ存在しないトピックの作成時(confluent.security.event.logger.exporter.kafka.topic.create=true)にのみ使用してください。トピックが既に存在する場合にこの構成を実装しようとしても、何も変更されません。

  • 型: long
  • デフォルト: 2592000000 ミリ秒(30 日間)
  • 重要度: 低

confluent.security.event.logger.exporter.kafka.topic.roll.ms

イベントログトピックのログローリング時間を指定します。このオプションは、イベントログトピックの作成時に使用される segment.ms トピック構成プロパティに相当します。この構成オプションは、まだ存在しないトピックの作成時(confluent.security.event.logger.exporter.kafka.topic.create=true)にのみ使用してください。トピックが既に存在する場合にこの構成を実装しようとしても、何も変更されません。

  • 型: long
  • デフォルト: 14400000 ミリ秒(4 時間)
  • 重要度: 低

confluent.security.event.router.cache.entries

ルーターキャッシュでサポートするリソースエントリの数を指定します。一部のトピックで生成または消費に対する監査ログが有効になっている場合は、そのようなトピックの数より大きい値を指定します。それ以外の場合は、キャッシュによってパフォーマンスに大きな影響が生じないように、ログメッセージのボリュームを十分に小さくする必要があります。

  • 型: int
  • デフォルト: 10000
  • 重要度: 低

confluent.security.event.router.config

イベントをトピックにルーティングするために使用される JSON 構成を識別します。

  • 型: string
  • デフォルト: ""
  • 重要度: 低

監査ログの無効化

Kafka クラスターで監査ログを無効にするには、server.properties で以下を設定します。

confluent.security.event.logger.enable=false

これにより、MANAGEMENT イベントに関するものを含めて、すべての構造化監査ログが無効になります。認可ログは引き続き log4j に Kafka 形式で書き込まれます。詳細については、「Kafka のログ記録」を参照してください。

Docker での監査ログの構成

Confluent Docker コンテナを構成するには、Docker コンテナの実行用の環境変数を使用してプロパティを渡します(通常は docker-compose.yml または Kubernetes 構成を使用)。Docker での Confluent Kafka 構成の使用については、「Confluent Kafka の構成」を参照してください。

Docker コンテナで監査ログを構成するには、監査ログ構成でどの server.properties オプションを設定するかを決定して、そのオプション名の小文字とドットを大文字とアンダースコアに変換し、プレフィックスとして KAFKA_ を付けます。Confluent Platform と Docker における監査ログ構成の違いは、使用される命名規則のみです。たとえば、以下に示す docker-compose.yml の抜粋には、Docker コンテナで confluent.security.event.router.config を構成する方法が示されています。

version: '2'
services:
  broker:
    image: confluentinc/cp-server:5.4.x-latest
    ...
    environment:
      KAFKA_AUTHORIZER_CLASS_NAME: io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
      KAFKA_CONFLUENT_SECURITY_EVENT_ROUTER_CONFIG: "{\"routes\":{\"crn:///kafka=*/group=*\":{\"consume\":{\"allowed\":\"confluent-audit-log-events\",\"denied\":\"confluent-audit-log-events\"}},\"crn:///kafka=*/topic=*\":{\"produce\":{\"allowed\":\"confluent-audit-log-events\",\"denied\":\"confluent-audit-log-events\"},\"consume\":{\"allowed\":\"confluent-audit-log-events\",\"denied\":\"confluent-audit-log-events\"}}},\"destinations\":{\"topics\":{\"confluent-audit-log-events\":{\"retention_ms\":7776000000}}},\"default_topics\":{\"allowed\":\"confluent-audit-log-events\",\"denied\":\"confluent-audit-log-events\"},\"excluded_principals\":[\"User:kafka\",\"User:ANONYMOUS\"]}"

トラブルシューティング

監査ログのメカニズムによって Kafka トピックへの書き込みが試行され、なんらかの理由で正常に処理されなかった場合は、代わりに JSON の監査ログメッセージが log4j に書き込まれます。このメッセージの送信先は、Kafka のブローカーの log4j.propertiesce-kafka/config/)で io.confluent.security.audit.log.fallback を構成することによって指定できます。

log4j.properties コンテンツの例を次に示します。

log4j.appender.auditLogAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.auditLogAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.auditLogAppender.File=${kafka.logs.dir}/metadata-service.log
log4j.appender.auditLogAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.auditLogAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

# Fallback logger for audit logging. Used when the Kafka topics are initializing.
log4j.logger.io.confluent.security.audit.log.fallback=INFO, auditLogAppender
log4j.additivity.io.confluent.security.audit.log.fallback=false

監査ログメッセージが作成されない場合や、構成どおりのトピックにルーティングされない場合は、次のことを確認してください。

  • 監査ログプリンシパルに必要なアクセス許可が付与されていること
  • プリンシパルがターゲットクラスターに接続できること
  • プリンシパルがすべてのトピックを作成および生成できること

必要であれば、次のように指定して、ログレベルを一時的に上げます。log4j.logger.io.confluent.security.audit=DEBUG

すべての認可イベントタイプの表示

クラスターで認可の決定を表示およびデバッグする必要がある場合は、すべての認可イベントタイプがキャプチャされるように監査ログを構成できます。大量のメッセージがきわめて高速で生成される構成であるため、このトラブルシューティング構成の使用は 1 つのシナリオに限定してください。保持期間は 30 分以下に設定することをお勧めします。この構成では、内部トピック(プレフィックス _ が付いたトピック)のハートビートやトラフィックはキャプチャされません。

{
        "routes": {
                  "crn://mds1.example.com/kafka=*": {
                            "interbroker": {
                                      "allowed": "confluent-audit-log-events",
                                      "denied": "confluent-audit-log-events"
                            },
                            "describe": {
                                      "allowed": "confluent-audit-log-events",
                                      "denied": "confluent-audit-log-events"
                            },
                            "management": {
                                      "allowed": "confluent-audit-log-events",
                                      "denied": "confluent-audit-log-events"
                            }
                  },
                  "crn://mds1.example.com/kafka=*/group=*": {
                            "consume": {
                                      "allowed": "confluent-audit-log-events",
                                      "denied": "confluent-audit-log-events"
                            },
                            "describe": {
                                      "allowed": "confluent-audit-log-events",
                                      "denied": "confluent-audit-log-events"
                            },
                            "management": {
                                      "allowed": "confluent-audit-log-events",
                                      "denied": "confluent-audit-log-events"
                            }
                  },
                  "crn://mds1.example.com/kafka=*/topic=*": {
                            "produce": {
                                      "allowed": "confluent-audit-log-events",
                                      "denied": "confluent-audit-log-events"
                            },
                            "consume": {
                                      "allowed": "confluent-audit-log-events",
                                      "denied": "confluent-audit-log-events"
                            },
                            "describe": {
                                      "allowed": "confluent-audit-log-events",
                                      "denied": "confluent-audit-log-events"
                            },
                            "management": {
                                      "allowed": "confluent-audit-log-events",
                                      "denied": "confluent-audit-log-events"
                            }
                  },
                  "crn://mds1.example.com/kafka=*/topic=_*": {
                            "produce": {
                                      "allowed": "",
                                      "denied": ""
                            },
                            "consume": {
                                      "allowed": "",
                                      "denied": ""
                            },
                            "describe": {
                                      "allowed": "",
                                      "denied": ""
                            }
                  }
        },
        "destinations": {
                  "bootstrap_servers": [
                            "host1:port"
                  ],
                  "topics": {
                            "confluent-audit-log-events": {
                                      "retention_ms": 7776000000
                            }
                  }
        },
        "default_topics": {
                  "allowed": "confluent-audit-log-events",
                  "denied": "confluent-audit-log-events"
        }
}

監査ログのオンザフライ表示

初期セットアップ中またはトラブルシューティング中に簡単なコマンドラインツールを使用して、最近の監査ログエントリをすばやく何度でも調べることができます。この機能は、監査ログのトラブルシューティングや、RBAC のロールバインディングのトラブルシューティングに役立ちます。

まず、監査ログトピックをローカルファイルに送ります。ローカルファイルシステムを使用する方が、grep の動作が高速になるためです。

kafka-console-consumer がローカルにインストールされていて、監査ログのデスティネーション Kafka クラスターから直接消費できる場合、コマンドは次のようになります。

./kafka-console-consumer --bootstrap-server auditlog.example.com:9092 --consumer-config ~/auditlog-consumer.properties --whitelist '^confluent-audit-log-events.*' > /tmp/streaming.audit.log

直接アクセスできず、代わりに "ジャンプボックス"(独立したセキュリティゾーンでデバイスにアクセスして管理するために使用するネットワーク上のマシンまたはサーバー)を使用して接続する必要がある場合は、次のようなコマンドを使用します。

ssh -tt -i ~/.ssh/theuser.key theuser@jumpbox './kafka-console-consumer --bootstrap-server auditlog.example.com:9092 --consumer-config ~/auditlog-consumer.properties --whitelist '"'"'^confluent-audit-log-events.*'"'"' ' > /tmp/streaming.audit.log

どちらの方法を使用した場合も、この時点で別のターミナルを開き、ローカルで tail -f /tmp/streaming.audit.log を実行すると、監査ログメッセージをオンザフライで表示できます。取得した監査ログは、grep および jq (または別のユーティリティ)を使用して調べることができます。以下に例を示します。

tail -f /tmp/streaming.audit.log | grep 'connect' | jq -c '[.time, .data.authenticationInfo.principal, .data.authorizationInfo.operation, .data.resourceName]'

生成および消費の拒否が監査ログに表示されない

クライアントがトピックに対する生成または消費を行う場合は、まずトピックのメタデータを取得して、使用可能なパーティションを決定する必要があります。これは、Kafka API の metadata 操作を使用して実行できます。特定のトピックに対する READ または WRITE の認可を受けていないクライアントには、DESCRIBE の認可もないことがきわめて一般的です。そのような場合、metadata 操作が失敗し、Kafka API の produce または consume 操作は試行されないため、監査ログに表示されません。

ノイズが非常に多いため、許可された説明を抑制する一方で、拒否をキャプチャする(拒否は構成の誤りまたは不正アクセスの試行を示している可能性がある)には、次のようなルーティング指定を構成に追加します。

"crn://mds1.example.com/kafka=*/topic=secure-*": {
    "produce": {
        "allowed": "confluent-audit-log-events_secure",
        "denied": "confluent-audit-log-events_denied"
    },
    "consume": {
        "allowed": "confluent-audit-log-events_secure",
        "denied": "confluent-audit-log-events_denied"
    },
    "describe": {
        "allowed": "",
        "denied": "confluent-audit-log-events_denied"
    }
}