コンテナ上のマイクロサービスの認証強化　～StrimziとKeycloak～

リスト1：Strimziをインストール

$ wget https://github.com/strimzi/strimzi-kafka-operator/releases/download/0.20.0/strimzi-0.20.0.zip
$ apt install unzip
$ unzip strimzi-0.20.0.zip
$ cd strimzi-0.20.0/
$ kubectl create namespace kafka
$ sed -i 's/namespace: .*/namespace: kafka/' install/cluster-operator/*RoleBinding*.yaml
$ kubectl create namespace my-kafka-project

リスト2：設定を書き換え

    env:
      - name: STRIMZI_NAMESPACE
        value: my-kafka-project           // この行を追加
        valueFrom:                        // 以下3行を削除
          fieldRef:
            fieldPath: metadata.namespace // ここまで削除

リスト3：Kubernetesにデプロイ

$ kubectl apply -f install/cluster-operator/ -n kafka
$ kubectl apply -f install/cluster-operator/020-RoleBinding-strimzi-cluster-operator.yaml -n my-kafka-project
$ kubectl apply -f install/cluster-operator/032-RoleBinding-strimzi-cluster-operator-topic-operator-delegation.yaml -n my-kafka-project
$ kubectl apply -f install/cluster-operator/031-RoleBinding-strimzi-cluster-operator-entity-operator-delegation.yaml -n my-kafka-project

リスト4：Kafkaクラスタの作成

apiVersion: kafka.strimzi.io/v1beta1
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    replicas: 1
    listeners:
      plain:
        port: 9092
        type: internal
        tls: false
      tls:
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
      external:
        port: 9094
        type: nodeport
        tls: false
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 100Mi
        deleteClaim: false
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
  zookeeper:
    replicas: 1
    storage:
      type: persistent-claim
      size: 100Mi
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}

リスト5：CustomResourceDefinitionの適用

$ kubectl create -n my-kafka-project -f kafka.yaml

リスト6：KafakaクラスタのIPアドレスとポートを確認する

$ echo $(minikube ip)
172.30.10.93
$ echo $(kubectl get services/my-cluster-kafka-external-bootstrap -o go-template='{{(index .spec.ports 0).nodePort}}' -n my-kafka-project)
31105

リスト7：smallrye-reactive-messaging-kafakのエクステンションを追加

$ cd security-keycloak-authorization-quickstart/
$ ./mvnw quarkus:add-extension -Dextensions="smallrye-reactive-messaging-kafka"

リスト8：Producer部分をBooksResource.javaに追加で実装

…
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;
…
import io.smallrye.reactive.messaging.annotations.Broadcast;

@Path("/")
public class BooksResource {
…
    @Inject
    @Broadcast
    @Channel("my-channel-out")
    Emitter<String> myData;
…
    @GET
    @Path("/details/0")
    @NoCache
    public Response details() {
        myData.send(identity.getPrincipal().getName());
        …
    }

}

リスト9：Consumer部分をSampleConsumer.javaに実装

package org.acme.security.keycloak.authorization;

import javax.enterprise.context.ApplicationScoped;

import org.eclipse.microprofile.reactive.messaging.Incoming;

@ApplicationScoped
public class SampleConsumer {
    @Incoming("my-channel-in")
    public void consume(String myData) {
        System.out.println("Hello " + myData + " !");
    }
}

リスト10：application.propertiesに設定追加

…
quarkus.kubernetes.labels.version=v3
…
# Configure the SmallRye Kafka connector
kafka.bootstrap.servers=172.30.10.93:31105

# Configure the Kafka sink (we write to it)
mp.messaging.outgoing.my-channel-out.connector=smallrye-kafka
mp.messaging.outgoing.my-channel-out.topic=book-info
mp.messaging.outgoing.my-channel-out.value.serializer=org.apache.kafka.common.serialization.StringSerializer

# Configure the Kafka source (we read from it)
mp.messaging.incoming.my-channel-in.connector=smallrye-kafka
mp.messaging.incoming.my-channel-in.topic=book-info
mp.messaging.incoming.my-channel-in.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

リスト11：Quarkusアプリケーションのビルド、デプロイ

$ ./mvnw clean package -Dquarkus.kubernetes.deploy=true

リスト12：DestinationRuleの書き換え

…
---
apiVersion: networking.istio.io/v1alpha3
kind: DestinationRule
metadata:
  name: details
spec:
  host: details
  trafficPolicy:
    tls:
      mode: ISTIO_MUTUAL
  subsets:
  - name: v3
    labels:
      version: v3
---

リスト13：変更したDestinationRuleの適用

$ kubectl apply -f destination-rule-all-mtls-v2.yaml

リスト14：メッセージ送受信されていることを確認

$ kubectl logs details-xxx -c details
…
Hello admin_user !

リスト15：Kafkaクラスタの設定を変更

…
spec:
  kafka:
    …
    listeners:
      …
      external:
        port: 9094
        type: nodeport
        tls: false
        authentication:
          type: oauth
          clientId: kafka-broker
          clientSecret:
            secretName: my-cluster-oauth
            key: clientSecret
          validIssuerUri: http://172.30.10.93:31385/auth/realms/bookinfo
          introspectionEndpointUri: http://172.30.10.93:31385/auth/realms/bookinfo/protocol/openid-connect/token/introspect
          userNameClaim: preferred_username
…

リスト16：Secretを作成

apiVersion: v1
kind: Secret
metadata:
  name: my-cluster-oauth
type: Opaque
data:
  clientSecret: ＜Kafkaクラスタ用クライアントのクライアントシークレットをBase64エンコードしたもの＞

リスト17：Secret、Kafkaクラスタの設定を適用

$ kubectl apply -n my-kafka-project -f kafka-broker-client-secret.yaml
$ kubectl apply -n my-kafka-project -f kafka.yaml

リスト18：動作の確認

$ kubectl logs details-xxx -c details
…
2020-12-10 08:36:45,003 INFO  [org.apa.kaf.cli.FetchSessionHandler] (kafka-coordinator-heartbeat-thread | dfd2724b-17ff-4dbe-948b-30768b4fe02a) [Consumer clientId=consumer-dfd2724b-17ff-4dbe-948b-30768b4fe02a-1, groupId=dfd2724b-17ff-4dbe-948b-30768b4fe02a] Error sending fetch request (sessionId=284737104, epoch=INITIAL) to node 0: {}.: org.apache.kafka.common.errors.DisconnectException

リスト19：pom.xmlの依存関係にkafka-oauth-clientを追加

…
        <!-- https://mvnrepository.com/artifact/io.strimzi/kafka-oauth-client -->
        <dependency>
            <groupId>io.strimzi</groupId>
            <artifactId>kafka-oauth-client</artifactId>
            <version>0.6.1</version>
        </dependency>
…

リスト20：applicationpropertiesに設定を追加

# Configure the Kafka sink (we write to it)
mp.messaging.outgoing.my-channel-out.connector=smallrye-kafka
mp.messaging.outgoing.my-channel-out.topic=book-info
mp.messaging.outgoing.my-channel-out.value.serializer=org.apache.kafka.common.serialization.StringSerializer
mp.messaging.outgoing.my-channel-out.security.protocol=SASL_PLAINTEXT
mp.messaging.outgoing.my-channel-out.sasl.mechanism=OAUTHBEARER
mp.messaging.outgoing.my-channel-out.sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
mp.messaging.outgoing.my-channel-out.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.client.id="kafka-producer" \
  oauth.client.secret="＜Producer用クライアントのクライアントシークレット＞" \
  oauth.token.endpoint.uri="http://172.30.10.93:31385/auth/realms/bookinfo/protocol/openid-connect/token";

# Configure the Kafka source (we read from it)
mp.messaging.incoming.my-channel-in.connector=smallrye-kafka
mp.messaging.incoming.my-channel-in.topic=book-info
mp.messaging.incoming.my-channel-in.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
mp.messaging.incoming.my-channel-in.security.protocol=SASL_PLAINTEXT
mp.messaging.incoming.my-channel-in.sasl.mechanism=OAUTHBEARER
mp.messaging.incoming.my-channel-in.sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
mp.messaging.incoming.my-channel-in.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.client.id="kafka-consumer" \
  oauth.client.secret="＜Consumer用クライアントのクライアントシークレット＞" \
  oauth.token.endpoint.uri="http://172.30.10.93:31385/auth/realms/bookinfo/protocol/openid-connect/token";

リスト21：Quarkusアプリケーションのビルド、デプロイ

$ cd security-keycloak-authorization-quickstart/
$ ./mvnw clean package -Dquarkus.kubernetes.deploy=true

リスト22：動作を確認

$ kubectl logs details-xxx -c details
…
Hello admin_user !