Realtime Compute for Apache Flink:Intégration de la gestion des identifiants KMS (bêta)

Présentation

Grâce à l'intégration de la gestion des identifiants KMS, vous pouvez référencer des identifiants hébergés dans KMS (tels que les noms d'utilisateur et mots de passe de base de données) dans vos jobs Flink sans exposer ces informations en texte clair dans le code du job. Au moment de l'exécution, Flink endosse un rôle RAM pour obtenir un jeton STS temporaire, appelle KMS afin de récupérer la valeur réelle de l'identifiant, puis se connecte à la source de données cible.

Principaux avantages :

• Protection des informations sensibles : Aucun mot de passe ou nom d'utilisateur en texte clair n'est exposé dans les jobs. Les identifiants sont hébergés de manière chiffrée dans KMS.

• Identifiants sans maintenance : Avec la rotation des identifiants KMS, Flink récupère automatiquement la dernière valeur de l'identifiant après chaque mise à jour. Le code du job ne nécessite aucune modification.

Remarques

La suppression incorrecte d'identifiants dans KMS, ou la modification ou suppression erronée des autorisations et de la politique d'approbation d'un rôle RAM dans RAM, peut entraîner l'échec des jobs Flink, car ceux-ci ne parviennent pas à obtenir les identifiants KMS. Conformément à l'Accord de niveau de service pour Realtime Compute for Apache Flink, les indisponibilités causées par une utilisation incorrecte des ressources par les clients ne sont pas couvertes par l'SLA.

Limites

• Seuls le connecteur Postgres CDC (VVR 11.8 et versions ultérieures) et le connecteur Kafka (VVR 11.9 et versions ultérieures) prennent en charge l'accès aux ressources cloud avec des identifiants KMS.

• Le nombre de rôles RAM référencés pour accéder à KMS est limité. Pour plus de détails, consultez la rubrique Utiliser un rôle RAM pour accéder aux ressources cloud.

Préparation des autorisations

• Étape 1 (création d'un identifiant KMS) : L'opérateur doit disposer des autorisations de gestion des identifiants KMS.

• Étapes 2 et 3 : L'opérateur doit se voir attribuer la stratégie AliyunRAMFullAccess, ou recevoir séparément les autorisations de gestion des rôles RAM et de gestion des autorisations. Pour plus d'informations, consultez la page Créer une stratégie personnalisée.

• Étapes 4 et 5 (référencement d'un rôle dans Flink et développement de jobs) : L'opérateur doit disposer des autorisations d'édition ou supérieures sur le namespace Flink, ou se voir attribuer des autorisations fines pour référencer des rôles RAM, gérer des fichiers et développer des jobs. Pour plus de détails, consultez la rubrique Accorder des autorisations pour accéder à la console de développement.

Étape 1 : Créer un identifiant KMS

KMS propose plusieurs types d'identifiants. Les étapes suivantes utilisent un identifiant RDS à titre d'exemple :

1. Connectez-vous à la console Key Management Service. Sélectionnez la région dans la barre de menus supérieure, puis dans le volet de navigation de gauche, choisissez Resources > Secrets.

2. Dans l'onglet Self-Managed Secrets, zone Secret Type, cliquez sur Database Secret.

3. Au-dessus de la liste à droite, sous Create Database Secret, cliquez sur Create Single Secret, complétez la configuration, puis cliquez sur OK.

Pour plus d'informations, consultez les rubriques Prise en main de Secrets Manager et Gérer et utiliser les secrets ApsaraDB RDS.

Étape 2 : Créer et configurer un rôle RAM

Ce rôle constitue l'identité que le service Flink utilise pour accéder à KMS. Flink endosse ce rôle pour obtenir des identifiants temporaires et appelle KMS en tant que ce rôle afin de récupérer les valeurs des identifiants. Pour plus de détails sur l'accès sans identifiant à KMS, consultez l'annexe de la rubrique Utiliser un rôle RAM pour accéder aux ressources cloud.

{
  "Statement": [
    {
      "Action": "sts:AssumeRole",
      "Effect": "Allow",
      "Principal": {
        "Service": [
          "stream.aliyuncs.com"
        ]
      }
    }
  ],
  "Version": "1"
}

{
  "Version": "1",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "kms:GetSecretValue",
      "Resource": "acs:kms:<region>:<account-id>:secret/<secret-name>"
    },
    {
      "Effect": "Allow",
      "Action": "kms:Decrypt",
      "Resource": "acs:kms:<region>:<account-id>:key/<key-id>"
    }
  ]
}

Étape 3 : Accorder des autorisations à un utilisateur RAM

Créez une stratégie d'autorisation et attribuez-la à l'utilisateur RAM. Cela permet à l'utilisateur de transmettre le rôle RAM créé à l'étape 2 au service Flink pour qu'il l'endosse.

{
  "Version": "1",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "ram:ListRoles"
      ],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": "ram:PassRole",
      "Resource": "acs:ram::<account-id>:role/<role-name>",
      "Condition": {
        "StringEquals": {
          "acs:Service": "stream.aliyuncs.com"
        }
      }
    }
  ]
}

Étape 4 : Référencer le rôle RAM dans Flink

Référencez le rôle RAM créé à l'étape 2 dans le namespace Flink afin que les jobs puissent utiliser ce rôle pour accéder à KMS.

1. Connectez-vous à la console Realtime Compute et accédez à l'espace de travail cible.

2. Dans le volet de navigation de gauche, choisissez Security > Security.

3. Cliquez sur l'onglet RAM Roles.

4. Cliquez sur Reference RAM Role.

5. Dans la boîte de dialogue, sélectionnez le rôle RAM créé à l'étape 2 (par exemple, FlinkRoleForKMSRead). La recherche floue par nom de rôle est prise en charge.

   Remarque

   La liste des rôles contient les rôles de service réguliers sous le compte Alibaba Cloud propriétaire du namespace actuel. L'opérateur doit appartenir au même compte Alibaba Cloud que le namespace actuel et disposer de l'autorisation ram:ListRoles pour afficher les rôles.

6. Cliquez sur Test. Le système vérifie automatiquement les deux éléments suivants :

   • Vérification de l'autorisation PassRole : L'utilisateur actuel dispose de l'autorisation ram:PassRole sur le rôle RAM sélectionné.

   • Vérification de la politique d'approbation : Le rôle RAM sélectionné approuve le namespace Flink actuel pour l'endosser.

7. Une fois les vérifications réussies, cliquez sur OK.

Pour plus d'informations, consultez la rubrique Utiliser un rôle RAM pour accéder aux ressources cloud.

Étape 5 : Utiliser les identifiants KMS dans les jobs

Une fois les étapes précédentes terminées, spécifiez les identifiants en utilisant le format de référence secret:// dans le code du job et configurez les paramètres de connexion KMS. Flink récupère automatiquement les valeurs des identifiants au moment de l'exécution pour accéder à la source de données.

CREATE TEMPORARY TABLE postgrescdc_source (
    id INT NOT NULL,
    name STRING,
    description STRING,
    weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = 'secret://kms.my-db-secret.AccountName',
  'password' = 'secret://kms.my-db-secret.AccountPassword',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>',
  'slot.name' = '<slot name>',
  'decoding.plugin.name' = 'pgoutput',
  'kms.endpoint' = '<your-kms-instance-vpc-endpoint>',
  'kms.ca' = 'oss://<bucket>/kms-ca.pem',
  'kms.akless.assume-role.role-name' = 'FlinkRoleForKMSRead'
);

CREATE TEMPORARY TABLE kafka_source (
  order_id BIGINT,
  user_id BIGINT,
  amount INT,
  payload STRING
) WITH (
  'connector' = 'kafka',
  'topic' = '<topic name>',
  'properties.bootstrap.servers' = '<broker 1>,<broker 2>,<broker 3>',
  'properties.group.id' = '<group id>',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv',
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN',
  /* Reference KMS credentials in username and password */
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="secret://kms.my-kafka-secret.AccountName" password="secret://kms.my-kafka-secret.AccountPassword";',
  'kms.endpoint' = 'kms.cn-beijing.aliyuncs.com',
  'kms.akless.assume-role.role-name' = 'FlinkRoleForKMSRead'
);

source:
  type: kafka
  name: Kafka Source
  topic: <topic name>
  value.format: json
  scan.startup.mode: earliest-offset
  properties.bootstrap.servers: <broker 1>,<broker 2>,<broker 3>
  properties.group.id: <group id>
  properties.security.protocol: SASL_PLAINTEXT
  properties.sasl.mechanism: PLAIN
  properties.sasl.jaas.config: org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="secret://kms.my-kafka-secret.AccountName" password="secret://kms.my-kafka-secret.AccountPassword";
  # Shared gateway: the endpoint is fixed per region, and kms.ca is not required
  kms.endpoint: kms.cn-beijing.aliyuncs.com
  kms.akless.assume-role.role-name: FlinkRoleForKMSRead

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: <broker 1>,<broker 2>,<broker 3>
  # ApsaraMQ for Kafka does not support idempotent writes. Disable idempotence on VVR 8.0.0 and later
  properties.enable.idempotence: false
  properties.security.protocol: SASL_PLAINTEXT
  properties.sasl.mechanism: PLAIN
  properties.sasl.jaas.config: org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="secret://kms.my-kafka-secret.AccountName" password="secret://kms.my-kafka-secret.AccountPassword";
  # Dedicated gateway: kms.ca is also required
  kms.endpoint: <your-kms-instance-vpc-endpoint>
  kms.ca: oss://flink-fullymanaged-<workspace-id>/artifacts/namespaces/<namespace>/PrivateKmsCA_kst-******.pem
  kms.akless.assume-role.role-name: FlinkRoleForKMSRead