How to Secure a Flink Job with Confluent Manager for Apache Flink

The following sections provide an overview of security risks and available security controls for Flink jobs deployed with Confluent Manager for Apache Flink® (CMF).

Risks

Apache Flink® is a framework for executing user code. As such, users that have permission to deploy Flink jobs have the ability to execute arbitrary code. It is therefore critical that you set up authentication/authorization and limit networked access. Unconfigured Flink clusters should never be deployed in an Internet-facing environment.

TLS/SSL for authentication and encryption

Flink supports TLS/SSL for authentication and encryption of network traffic between Flink processes, both for internal connectivity and external connectivity.

You should follow the SSL guidelines in the Flink documentation.

Internal connectivity in Flink includes control messages between Flink processes and the data connections between TaskManagers. All internal connections can be configured to use TLS/SSL for authentication and encryption, which is in the SSL guidelines. When configured, the connections use mutual authentication, meaning both server and client side of each connection need to present the certificate to each other. The certificate acts effectively as a shared secret when a dedicated CA is used to exclusively sign an internal certificate. The certificate for internal communication is not needed by any other party to interact with Flink, and can be simply added to the container images.

External connectivity in Flink happens with HTTP/REST endpoints. For example, these endpoints are used by the web UI and the Flink CLI. These endpoints can be configured to require TLS/SSL connections. The server will, however, accept connections from any client by default, meaning the REST endpoint does not authenticate the client. Simple mutual authentication may be enabled by configuration (as outlined in the Flink documentation mentioned previously) if authentication of connections to the REST endpoints is required. However, Confluent recommends you deploy a “side car proxy” meaning you bind the REST endpoint to the loopback interface (or the pod-local interface in Kubernetes) and start a REST proxy that authenticates and forwards the requests to Flink. Examples for proxies that Flink users have deployed are Envoy Proxy or NGINX with MOD_AUTH.

Connect CMF to clusters with TLS on the REST endpoint

CMF uses the REST endpoint of an Flink cluster to proxy the Flink Web UI and to fetch the results of SELECT statements. When a cluster sets security.ssl.rest.enabled: true, CMF connects to it over HTTPS. Give CMF the following stores:

• A trust store with the certificate of the REST endpoint, or with the CA that signed it. Without a trust store, CMF trusts the default CAs of the JVM.

• A key store with a client certificate, if the cluster also sets security.ssl.rest.authentication-enabled: true.

CMF uses the same stores for all clusters. The Confluent Platform for Apache Flink Kubernetes Operator needs the same stores when you install it with tls.create: true, so you can mount the flink-operator-cert secret of the Kubernetes Operator into CMF. A pod can only use secrets from its own namespace. If CMF and the Kubernetes Operator run in different namespaces, copy the flink-operator-cert and operator-certificate-password secrets into the namespace of CMF.

To configure the stores:

1. Add the stores to your Helm values file. The following example mounts the flink-operator-cert secret and reads the password from the operator-certificate-password secret:

   cmf:
     flink-rest:
       ssl:
         trust-store: /opt/flink-rest-tls/truststore.jks
         trust-store-password: ${FLINK_REST_TLS_PASSWORD}
         key-store: /opt/flink-rest-tls/keystore.jks
         key-store-password: ${FLINK_REST_TLS_PASSWORD}
   mountedVolumes:
     volumes:
       - name: flink-rest-tls
         secret:
           secretName: flink-operator-cert
     volumeMounts:
       - name: flink-rest-tls
         mountPath: /opt/flink-rest-tls
         readOnly: true
   extraEnv:
     - name: FLINK_REST_TLS_PASSWORD
       valueFrom:
         secretKeyRef:
           name: operator-certificate-password
           key: password
   

2. Upgrade CMF with these values:

   helm upgrade --install cmf confluentinc/confluent-manager-for-apache-flink \
     --version "~2.5.0" \
     --namespace <namespace> \
     --reuse-values \
     --values values.yaml
   

The following settings are available:

Helm value	Default	Description
cmf.flink-rest.ssl.trust-store	None	Path to the trust store file.
cmf.flink-rest.ssl.trust-store-password	None	Password of the trust store.
cmf.flink-rest.ssl.trust-store-type	PKCS12	Type of the trust store. With the default type, CMF also reads JKS files.
cmf.flink-rest.ssl.key-store	None	Path to the key store file. Needed for mutual TLS.
cmf.flink-rest.ssl.key-store-password	None	Password of the key store.
cmf.flink-rest.ssl.key-store-type	PKCS12	Type of the key store. With the default type, CMF also reads JKS files.
cmf.flink-rest.ssl.key-password	The key store password	Password of the private key in the key store.
cmf.flink-rest.ssl.verify-hostname	true	Whether CMF checks that the certificate of the REST endpoint names the host that CMF connects to.

Note the following:

• With verify-hostname, the certificate must name <name>-rest.<namespace>.svc, where <name> is the name of the application, statement, or compute pool. A wildcard name, *.<namespace>.svc, covers all clusters in a namespace. CMF connects to clusters on remote Kubernetes clusters through a port-forward and doesn’t check their host name.

• CMF loads the stores when it first connects to a cluster with TLS, and loads them again every five minutes. It picks up renewed certificates without a restart. CMF reads the passwords only at startup. If you change a password, restart CMF.

• CMF reads security.ssl.rest.enabled from the flinkConfiguration of the FlinkDeployment. If you set it only in the default configuration of the Kubernetes Operator, or as a -D argument in kubernetes.jobmanager.entrypoint.args, CMF connects over HTTP and the request fails.

The cluster itself has the following requirements:

• The Kubernetes Operator reads the files that security.ssl.rest.keystore and security.ssl.rest.truststore name when it deploys the cluster. Mount the secret into the Flink pods at the path where the Kubernetes Operator has it, /opt/flink/tls-cert.

• Flink versions before 1.20 accept only the cipher suite TLS_RSA_WITH_AES_128_CBC_SHA by default, which the Kubernetes Operator doesn’t offer. Set security.ssl.algorithms, for example to TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384.

Kubernetes-level security controls

CMF supports Flink deployments via Flink’s Kubernetes operator. You have tight control over who can deploy Flink applications with the security controls available in Kubernetes.

You should not provide direct access to the Kubernetes resources managed by the Flink Kubernetes operator using Kubernetes RBAC. All access should go through CMF to comply with the authentication and authorization requirements.

The Flink Kubernetes operator installs two custom roles: flink-operator and flink. flink-operator is used to manage FlinkDeployment resources, meaning it creates and manages the JobManager deployment for each Flink job (and related resources). The flink role is used by the jobManager process of each job to create and manage the taskManager and configMap resources.

The standalone Kubernetes deployment mode is recommended when running untrusted code inside the Flink cluster, as there is no requirement for the Flink pod service account to have permissions to launch additional pods.

When you install CMF, you create the roles to create and manage FlinkDeployment resources on Kubernetes.

Related content

• Get Help with Confluent Platform for Apache Flink

• Get Started with Applications in Confluent Manager for Apache Flink

• Kubernetes best practices

• Flink RBAC documentation