| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.flink.configuration; |
| |
| import org.apache.flink.annotation.Experimental; |
| import org.apache.flink.annotation.docs.Documentation; |
| import org.apache.flink.configuration.description.Description; |
| import org.apache.flink.core.security.token.DelegationTokenProvider; |
| |
| import java.time.Duration; |
| import java.util.List; |
| |
| import static org.apache.flink.configuration.ConfigOptions.key; |
| import static org.apache.flink.configuration.description.LineBreakElement.linebreak; |
| import static org.apache.flink.configuration.description.LinkElement.link; |
| import static org.apache.flink.configuration.description.TextElement.code; |
| import static org.apache.flink.configuration.description.TextElement.text; |
| import static org.apache.flink.util.Preconditions.checkNotNull; |
| |
| /** The set of configuration options relating to security. */ |
| public class SecurityOptions { |
| |
| public static final String DELEGATION_TOKEN_PROVIDER_PREFIX = |
| DelegationTokenProvider.CONFIG_PREFIX + ".<serviceName>"; |
| |
| // ------------------------------------------------------------------------ |
| // Custom Security Service Loader |
| // ------------------------------------------------------------------------ |
| |
| public static final ConfigOption<List<String>> SECURITY_CONTEXT_FACTORY_CLASSES = |
| key("security.context.factory.classes") |
| .stringType() |
| .asList() |
| .defaultValues( |
| "org.apache.flink.runtime.security.contexts.HadoopSecurityContextFactory", |
| "org.apache.flink.runtime.security.contexts.NoOpSecurityContextFactory") |
| .withDescription( |
| "List of factories that should be used to instantiate a security context. " |
| + "If multiple are configured, Flink will use the first compatible " |
| + "factory. You should have a NoOpSecurityContextFactory in this list " |
| + "as a fallback."); |
| |
| public static final ConfigOption<List<String>> SECURITY_MODULE_FACTORY_CLASSES = |
| key("security.module.factory.classes") |
| .stringType() |
| .asList() |
| .defaultValues( |
| "org.apache.flink.runtime.security.modules.HadoopModuleFactory", |
| "org.apache.flink.runtime.security.modules.JaasModuleFactory", |
| "org.apache.flink.runtime.security.modules.ZookeeperModuleFactory") |
| .withDescription( |
| "List of factories that should be used to instantiate security " |
| + "modules. All listed modules will be installed. Keep in mind that the " |
| + "configured security context might rely on some modules being present."); |
| |
| // ------------------------------------------------------------------------ |
| // Kerberos Options |
| // ------------------------------------------------------------------------ |
| |
| @Documentation.Section(Documentation.Sections.SECURITY_AUTH_KERBEROS) |
| public static final ConfigOption<String> KERBEROS_LOGIN_PRINCIPAL = |
| key("security.kerberos.login.principal") |
| .stringType() |
| .noDefaultValue() |
| .withDeprecatedKeys("security.principal") |
| .withDescription("Kerberos principal name associated with the keytab."); |
| |
| @Documentation.Section(Documentation.Sections.SECURITY_AUTH_KERBEROS) |
| public static final ConfigOption<String> KERBEROS_LOGIN_KEYTAB = |
| key("security.kerberos.login.keytab") |
| .stringType() |
| .noDefaultValue() |
| .withDeprecatedKeys("security.keytab") |
| .withDescription( |
| "Absolute path to a Kerberos keytab file that contains the user credentials."); |
| |
| public static final ConfigOption<String> KERBEROS_KRB5_PATH = |
| key("security.kerberos.krb5-conf.path") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "Specify the local location of the krb5.conf file. If defined, this conf would be mounted on the JobManager and " |
| + "TaskManager containers/pods for Kubernetes and Yarn. Note: The KDC defined needs to be visible from inside the containers."); |
| |
| @Documentation.Section(Documentation.Sections.SECURITY_AUTH_KERBEROS) |
| public static final ConfigOption<Boolean> KERBEROS_LOGIN_USETICKETCACHE = |
| key("security.kerberos.login.use-ticket-cache") |
| .booleanType() |
| .defaultValue(true) |
| .withDescription("Indicates whether to read from your Kerberos ticket cache."); |
| |
| @Documentation.Section(Documentation.Sections.SECURITY_AUTH_KERBEROS) |
| public static final ConfigOption<String> KERBEROS_LOGIN_CONTEXTS = |
| key("security.kerberos.login.contexts") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "A comma-separated list of login contexts to provide the Kerberos credentials to" |
| + " (for example, `Client,KafkaClient` to use the credentials for ZooKeeper authentication and for" |
| + " Kafka authentication)"); |
| |
| /** @deprecated Use {@link #DELEGATION_TOKENS_ENABLED}. */ |
| @Deprecated |
| @Documentation.Section(Documentation.Sections.SECURITY_AUTH_KERBEROS) |
| public static final ConfigOption<Boolean> KERBEROS_FETCH_DELEGATION_TOKEN = |
| key("security.kerberos.fetch.delegation-token") |
| .booleanType() |
| .defaultValue(true) |
| .withDescription( |
| "Indicates whether to fetch the delegation tokens for external services the Flink job needs to contact. " |
| + "Only HDFS and HBase are supported. It is used in Yarn deployments. " |
| + "If true, Flink will fetch HDFS and HBase delegation tokens and inject them into Yarn AM containers. " |
| + "If false, Flink will assume that the delegation tokens are managed outside of Flink. " |
| + "As a consequence, it will not fetch delegation tokens for HDFS and HBase. " |
| + "You may need to disable this option, if you rely on submission mechanisms, e.g. Apache Oozie, " |
| + "to handle delegation tokens."); |
| |
| @Documentation.Section(Documentation.Sections.SECURITY_AUTH_KERBEROS) |
| public static final ConfigOption<Duration> KERBEROS_RELOGIN_PERIOD = |
| key("security.kerberos.relogin.period") |
| .durationType() |
| .defaultValue(Duration.ofMinutes(1)) |
| .withDescription( |
| "The time period when keytab login happens automatically in order to always have a valid TGT."); |
| |
| /** @deprecated Use {@link #DELEGATION_TOKENS_RENEWAL_RETRY_BACKOFF}. */ |
| @Deprecated |
| @Documentation.Section(Documentation.Sections.SECURITY_AUTH_KERBEROS) |
| public static final ConfigOption<Duration> KERBEROS_TOKENS_RENEWAL_RETRY_BACKOFF = |
| key("security.kerberos.tokens.renewal.retry.backoff") |
| .durationType() |
| .defaultValue(Duration.ofHours(1)) |
| .withDescription( |
| "The time period how long to wait before retrying to obtain new delegation tokens after a failure."); |
| |
| /** @deprecated Use {@link #DELEGATION_TOKENS_RENEWAL_TIME_RATIO}. */ |
| @Deprecated |
| @Documentation.Section(Documentation.Sections.SECURITY_AUTH_KERBEROS) |
| public static final ConfigOption<Double> KERBEROS_TOKENS_RENEWAL_TIME_RATIO = |
| key("security.kerberos.tokens.renewal.time-ratio") |
| .doubleType() |
| .defaultValue(0.75) |
| .withDescription( |
| "Ratio of the tokens's expiration time when new credentials should be re-obtained."); |
| |
| @Documentation.Section(Documentation.Sections.SECURITY_AUTH_KERBEROS) |
| public static final ConfigOption<List<String>> KERBEROS_HADOOP_FILESYSTEMS_TO_ACCESS = |
| key("security.kerberos.access.hadoopFileSystems") |
| .stringType() |
| .asList() |
| .noDefaultValue() |
| .withDeprecatedKeys("yarn.security.kerberos.additionalFileSystems") |
| .withDescription( |
| "A semicolon-separated list of Kerberos-secured Hadoop filesystems Flink is going to access. For example, " |
| + "security.kerberos.access.hadoopFileSystems=hdfs://namenode2:9002;hdfs://namenode3:9003. " |
| + "The JobManager needs to have access to these filesystems to retrieve the security tokens."); |
| |
| // ------------------------------------------------------------------------ |
| // Delegation Token Options |
| // ------------------------------------------------------------------------ |
| |
| @Documentation.Section(value = Documentation.Sections.SECURITY_DELEGATION_TOKEN, position = 1) |
| public static final ConfigOption<Boolean> DELEGATION_TOKENS_ENABLED = |
| key("security.delegation.tokens.enabled") |
| .booleanType() |
| .defaultValue(true) |
| .withDeprecatedKeys(KERBEROS_FETCH_DELEGATION_TOKEN.key()) |
| .withDescription( |
| "Indicates whether to start delegation tokens system for external services."); |
| |
| @Documentation.Section(value = Documentation.Sections.SECURITY_DELEGATION_TOKEN, position = 2) |
| public static final ConfigOption<Duration> DELEGATION_TOKENS_RENEWAL_RETRY_BACKOFF = |
| key("security.delegation.tokens.renewal.retry.backoff") |
| .durationType() |
| .defaultValue(Duration.ofHours(1)) |
| .withDeprecatedKeys(KERBEROS_TOKENS_RENEWAL_RETRY_BACKOFF.key()) |
| .withDescription( |
| "The time period how long to wait before retrying to obtain new delegation tokens after a failure."); |
| |
| @Documentation.Section(value = Documentation.Sections.SECURITY_DELEGATION_TOKEN, position = 3) |
| public static final ConfigOption<Double> DELEGATION_TOKENS_RENEWAL_TIME_RATIO = |
| key("security.delegation.tokens.renewal.time-ratio") |
| .doubleType() |
| .defaultValue(0.75) |
| .withDeprecatedKeys(KERBEROS_TOKENS_RENEWAL_TIME_RATIO.key()) |
| .withDescription( |
| "Ratio of the tokens's expiration time when new credentials should be re-obtained."); |
| |
| @Documentation.SuffixOption(DELEGATION_TOKEN_PROVIDER_PREFIX) |
| @Documentation.Section(value = Documentation.Sections.SECURITY_DELEGATION_TOKEN, position = 4) |
| public static final ConfigOption<Boolean> DELEGATION_TOKEN_PROVIDER_ENABLED = |
| key("enabled") |
| .booleanType() |
| .defaultValue(true) |
| .withDescription( |
| "Controls whether to obtain credentials for services when security is " |
| + "enabled. By default, credentials for all supported services " |
| + "are retrieved when those services are configured, but it's " |
| + "possible to disable that behavior if it somehow conflicts " |
| + "with the application being run."); |
| |
| /** |
| * Returns a view over the given configuration via which options can be set/retrieved for the |
| * given provider. |
| * |
| * <pre> |
| * Configuration config = ... |
| * SecurityOptions.forProvider(config, "my_provider") |
| * .set(SecurityOptions.DELEGATION_TOKEN_PROVIDER_ENABLED, false) |
| * ... |
| * </pre> |
| * |
| * @param configuration backing configuration |
| * @param providerName provider name |
| * @return view over configuration |
| */ |
| @Experimental |
| public static Configuration forProvider(Configuration configuration, String providerName) { |
| return new DelegatingConfiguration( |
| configuration, DelegationTokenProvider.CONFIG_PREFIX + "." + providerName + "."); |
| } |
| |
| // ------------------------------------------------------------------------ |
| // ZooKeeper Security Options |
| // ------------------------------------------------------------------------ |
| |
| @Documentation.Section(Documentation.Sections.SECURITY_AUTH_ZOOKEEPER) |
| public static final ConfigOption<Boolean> ZOOKEEPER_SASL_DISABLE = |
| key("zookeeper.sasl.disable").booleanType().defaultValue(false); |
| |
| @Documentation.Section(Documentation.Sections.SECURITY_AUTH_ZOOKEEPER) |
| public static final ConfigOption<String> ZOOKEEPER_SASL_SERVICE_NAME = |
| key("zookeeper.sasl.service-name").stringType().defaultValue("zookeeper"); |
| |
| @Documentation.Section(Documentation.Sections.SECURITY_AUTH_ZOOKEEPER) |
| public static final ConfigOption<String> ZOOKEEPER_SASL_LOGIN_CONTEXT_NAME = |
| key("zookeeper.sasl.login-context-name").stringType().defaultValue("Client"); |
| |
| // ------------------------------------------------------------------------ |
| // SSL Security Options |
| // ------------------------------------------------------------------------ |
| |
| /** |
| * Enable SSL for internal (rpc, data transport, blob server) and external (HTTP/REST) |
| * communication. |
| * |
| * @deprecated Use {@link #SSL_INTERNAL_ENABLED} and {@link #SSL_REST_ENABLED} instead. |
| */ |
| @Deprecated |
| public static final ConfigOption<Boolean> SSL_ENABLED = |
| key("security.ssl.enabled") |
| .booleanType() |
| .defaultValue(false) |
| .withDescription( |
| "Turns on SSL for internal and external network communication." |
| + "This can be overridden by 'security.ssl.internal.enabled', 'security.ssl.external.enabled'. " |
| + "Specific internal components (rpc, data transport, blob server) may optionally override " |
| + "this through their own settings."); |
| |
| /** Enable SSL for internal communication (pekko rpc, netty data transport, blob server). */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<Boolean> SSL_INTERNAL_ENABLED = |
| key("security.ssl.internal.enabled") |
| .booleanType() |
| .defaultValue(false) |
| .withDescription( |
| "Turns on SSL for internal network communication. " |
| + "Optionally, specific components may override this through their own settings " |
| + "(rpc, data transport, REST, etc)."); |
| |
| /** Enable SSL for external REST endpoints. */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<Boolean> SSL_REST_ENABLED = |
| key("security.ssl.rest.enabled") |
| .booleanType() |
| .defaultValue(false) |
| .withDescription( |
| "Turns on SSL for external communication via the REST endpoints."); |
| |
| /** Enable mututal SSL authentication for external REST endpoints. */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<Boolean> SSL_REST_AUTHENTICATION_ENABLED = |
| key("security.ssl.rest.authentication-enabled") |
| .booleanType() |
| .defaultValue(false) |
| .withDescription( |
| "Turns on mutual SSL authentication for external communication via the REST endpoints."); |
| |
| // ----------------- certificates (internal + external) ------------------- |
| |
| /** The Java keystore file containing the flink endpoint key and certificate. */ |
| @Documentation.ExcludeFromDocumentation( |
| "The SSL Setup encourages separate configs for internal and REST security.") |
| public static final ConfigOption<String> SSL_KEYSTORE = |
| key("security.ssl.keystore") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The Java keystore file to be used by the flink endpoint for its SSL Key and Certificate."); |
| |
| /** Secret to decrypt the keystore file. */ |
| @Documentation.ExcludeFromDocumentation( |
| "The SSL Setup encourages separate configs for internal and REST security.") |
| public static final ConfigOption<String> SSL_KEYSTORE_PASSWORD = |
| key("security.ssl.keystore-password") |
| .stringType() |
| .noDefaultValue() |
| .withDescription("The secret to decrypt the keystore file."); |
| |
| /** Secret to decrypt the server key. */ |
| @Documentation.ExcludeFromDocumentation( |
| "The SSL Setup encourages separate configs for internal and REST security.") |
| public static final ConfigOption<String> SSL_KEY_PASSWORD = |
| key("security.ssl.key-password") |
| .stringType() |
| .noDefaultValue() |
| .withDescription("The secret to decrypt the server key in the keystore."); |
| |
| /** The truststore file containing the public CA certificates to verify the ssl peers. */ |
| @Documentation.ExcludeFromDocumentation( |
| "The SSL Setup encourages separate configs for internal and REST security.") |
| public static final ConfigOption<String> SSL_TRUSTSTORE = |
| key("security.ssl.truststore") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The truststore file containing the public CA certificates to be used by flink endpoints" |
| + " to verify the peer’s certificate."); |
| |
| /** Secret to decrypt the truststore. */ |
| @Documentation.ExcludeFromDocumentation( |
| "The SSL Setup encourages separate configs for internal and REST security.") |
| public static final ConfigOption<String> SSL_TRUSTSTORE_PASSWORD = |
| key("security.ssl.truststore-password") |
| .stringType() |
| .noDefaultValue() |
| .withDescription("The secret to decrypt the truststore."); |
| |
| // ----------------------- certificates (internal) ------------------------ |
| |
| /** For internal SSL, the Java keystore file containing the private key and certificate. */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<String> SSL_INTERNAL_KEYSTORE = |
| key("security.ssl.internal.keystore") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The Java keystore file with SSL Key and Certificate, " |
| + "to be used Flink's internal endpoints (rpc, data transport, blob server)."); |
| |
| /** For internal SSL, the password to decrypt the keystore file containing the certificate. */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<String> SSL_INTERNAL_KEYSTORE_PASSWORD = |
| key("security.ssl.internal.keystore-password") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The secret to decrypt the keystore file for Flink's " |
| + "for Flink's internal endpoints (rpc, data transport, blob server)."); |
| |
| /** For internal SSL, the password to decrypt the private key. */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<String> SSL_INTERNAL_KEY_PASSWORD = |
| key("security.ssl.internal.key-password") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The secret to decrypt the key in the keystore " |
| + "for Flink's internal endpoints (rpc, data transport, blob server)."); |
| |
| /** |
| * For internal SSL, the truststore file containing the public CA certificates to verify the ssl |
| * peers. |
| */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<String> SSL_INTERNAL_TRUSTSTORE = |
| key("security.ssl.internal.truststore") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The truststore file containing the public CA certificates to verify the peer " |
| + "for Flink's internal endpoints (rpc, data transport, blob server)."); |
| |
| /** For internal SSL, the secret to decrypt the truststore. */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<String> SSL_INTERNAL_TRUSTSTORE_PASSWORD = |
| key("security.ssl.internal.truststore-password") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The password to decrypt the truststore " |
| + "for Flink's internal endpoints (rpc, data transport, blob server)."); |
| |
| /** For internal SSL, the sha1 fingerprint of the internal certificate to verify the client. */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<String> SSL_INTERNAL_CERT_FINGERPRINT = |
| key("security.ssl.internal.cert.fingerprint") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The sha1 fingerprint of the internal certificate. " |
| + "This further protects the internal communication to present the exact certificate used by Flink." |
| + "This is necessary where one cannot use private CA(self signed) or there is internal firm wide CA is required"); |
| |
| // ----------------------- certificates (external) ------------------------ |
| |
| /** |
| * For external (REST) SSL, the Java keystore file containing the private key and certificate. |
| */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<String> SSL_REST_KEYSTORE = |
| key("security.ssl.rest.keystore") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The Java keystore file with SSL Key and Certificate, " |
| + "to be used Flink's external REST endpoints."); |
| |
| /** |
| * For external (REST) SSL, the password to decrypt the keystore file containing the |
| * certificate. |
| */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<String> SSL_REST_KEYSTORE_PASSWORD = |
| key("security.ssl.rest.keystore-password") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The secret to decrypt the keystore file for Flink's " |
| + "for Flink's external REST endpoints."); |
| |
| /** For external (REST) SSL, the password to decrypt the private key. */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<String> SSL_REST_KEY_PASSWORD = |
| key("security.ssl.rest.key-password") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The secret to decrypt the key in the keystore " |
| + "for Flink's external REST endpoints."); |
| |
| /** |
| * For external (REST) SSL, the truststore file containing the public CA certificates to verify |
| * the ssl peers. |
| */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<String> SSL_REST_TRUSTSTORE = |
| key("security.ssl.rest.truststore") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The truststore file containing the public CA certificates to verify the peer " |
| + "for Flink's external REST endpoints."); |
| |
| /** For external (REST) SSL, the secret to decrypt the truststore. */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<String> SSL_REST_TRUSTSTORE_PASSWORD = |
| key("security.ssl.rest.truststore-password") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The password to decrypt the truststore " |
| + "for Flink's external REST endpoints."); |
| |
| /** For external (REST) SSL, the sha1 fingerprint of the rest client certificate to verify. */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<String> SSL_REST_CERT_FINGERPRINT = |
| key("security.ssl.rest.cert.fingerprint") |
| .stringType() |
| .noDefaultValue() |
| .withDescription( |
| "The sha1 fingerprint of the rest certificate. " |
| + "This further protects the rest REST endpoints to present certificate which is only used by proxy server" |
| + "This is necessary where once uses public CA or internal firm wide CA"); |
| |
| // ------------------------ ssl parameters -------------------------------- |
| |
| /** SSL protocol version to be supported. */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<String> SSL_PROTOCOL = |
| key("security.ssl.protocol") |
| .stringType() |
| .defaultValue("TLSv1.2") |
| .withDescription( |
| "The SSL protocol version to be supported for the ssl transport. Note that it doesn’t" |
| + " support comma separated list."); |
| |
| /** |
| * The standard SSL algorithms to be supported. |
| * |
| * <p>More options here - |
| * http://docs.oracle.com/javase/8/docs/technotes/guides/security/StandardNames.html#ciphersuites |
| */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<String> SSL_ALGORITHMS = |
| key("security.ssl.algorithms") |
| .stringType() |
| .defaultValue("TLS_RSA_WITH_AES_128_CBC_SHA") |
| .withDescription( |
| Description.builder() |
| .text( |
| "The comma separated list of standard SSL algorithms to be supported. Read more %s", |
| link( |
| "http://docs.oracle.com/javase/8/docs/technotes/guides/security/StandardNames.html#ciphersuites", |
| "here")) |
| .build()); |
| |
| /** Flag to enable/disable hostname verification for the ssl connections. */ |
| @Documentation.Section(Documentation.Sections.SECURITY_SSL) |
| public static final ConfigOption<Boolean> SSL_VERIFY_HOSTNAME = |
| key("security.ssl.verify-hostname") |
| .booleanType() |
| .defaultValue(true) |
| .withDescription( |
| "Flag to enable peer’s hostname verification during ssl handshake."); |
| |
| /** SSL engine provider. */ |
| @Documentation.Section(Documentation.Sections.EXPERT_SECURITY_SSL) |
| public static final ConfigOption<String> SSL_PROVIDER = |
| key("security.ssl.provider") |
| .stringType() |
| .defaultValue("JDK") |
| .withDescription( |
| Description.builder() |
| .text("The SSL engine provider to use for the ssl transport:") |
| .list( |
| text("%s: default Java-based SSL engine", code("JDK")), |
| text( |
| "%s: openSSL-based SSL engine using system libraries", |
| code("OPENSSL"))) |
| .text( |
| "%s is based on %s and comes in two flavours:", |
| code("OPENSSL"), |
| link( |
| "http://netty.io/wiki/forked-tomcat-native.html#wiki-h2-4", |
| "netty-tcnative")) |
| .list( |
| text( |
| "dynamically linked: This will use your system's openSSL libraries " |
| + "(if compatible) and requires %s to be copied to %s", |
| code( |
| "opt/flink-shaded-netty-tcnative-dynamic-*.jar"), |
| code("lib/")), |
| text( |
| "statically linked: Due to potential licensing issues with " |
| + "openSSL (see %s), we cannot ship pre-built libraries. However, " |
| + "you can build the required library yourself and put it into %s:%s%s", |
| link( |
| "https://issues.apache.org/jira/browse/LEGAL-393", |
| "LEGAL-393"), |
| code("lib/"), |
| linebreak(), |
| code( |
| "git clone https://github.com/apache/flink-shaded.git && " |
| + "cd flink-shaded && " |
| + "mvn clean package -Pinclude-netty-tcnative-static -pl flink-shaded-netty-tcnative-static"))) |
| .build()); |
| |
| // ------------------------ ssl parameters -------------------------------- |
| |
| /** SSL session cache size. */ |
| @Documentation.Section(Documentation.Sections.EXPERT_SECURITY_SSL) |
| public static final ConfigOption<Integer> SSL_INTERNAL_SESSION_CACHE_SIZE = |
| key("security.ssl.internal.session-cache-size") |
| .intType() |
| .defaultValue(-1) |
| .withDescription( |
| Description.builder() |
| .text( |
| "The size of the cache used for storing SSL session objects. " |
| + "According to %s, you should always set " |
| + "this to an appropriate number to not run into a bug with stalling IO threads " |
| + "during garbage collection. (-1 = use system default).", |
| link( |
| "https://github.com/netty/netty/issues/832", |
| "here")) |
| .build()) |
| .withDeprecatedKeys("security.ssl.session-cache-size"); |
| |
| /** SSL session timeout. */ |
| @Documentation.Section(Documentation.Sections.EXPERT_SECURITY_SSL) |
| public static final ConfigOption<Integer> SSL_INTERNAL_SESSION_TIMEOUT = |
| key("security.ssl.internal.session-timeout") |
| .intType() |
| .defaultValue(-1) |
| .withDescription( |
| "The timeout (in ms) for the cached SSL session objects. (-1 = use system default)") |
| .withDeprecatedKeys("security.ssl.session-timeout"); |
| |
| /** SSL session timeout during handshakes. */ |
| @Documentation.Section(Documentation.Sections.EXPERT_SECURITY_SSL) |
| public static final ConfigOption<Integer> SSL_INTERNAL_HANDSHAKE_TIMEOUT = |
| key("security.ssl.internal.handshake-timeout") |
| .intType() |
| .defaultValue(-1) |
| .withDescription( |
| "The timeout (in ms) during SSL handshake. (-1 = use system default)") |
| .withDeprecatedKeys("security.ssl.handshake-timeout"); |
| |
| /** SSL session timeout after flushing the <tt>close_notify</tt> message. */ |
| @Documentation.Section(Documentation.Sections.EXPERT_SECURITY_SSL) |
| public static final ConfigOption<Integer> SSL_INTERNAL_CLOSE_NOTIFY_FLUSH_TIMEOUT = |
| key("security.ssl.internal.close-notify-flush-timeout") |
| .intType() |
| .defaultValue(-1) |
| .withDescription( |
| "The timeout (in ms) for flushing the `close_notify` that was triggered by closing a " |
| + "channel. If the `close_notify` was not flushed in the given timeout the channel will be closed " |
| + "forcibly. (-1 = use system default)") |
| .withDeprecatedKeys("security.ssl.close-notify-flush-timeout"); |
| |
| /** |
| * Checks whether SSL for internal communication (rpc, data transport, blob server) is enabled. |
| */ |
| public static boolean isInternalSSLEnabled(Configuration sslConfig) { |
| @SuppressWarnings("deprecation") |
| final boolean fallbackFlag = sslConfig.getBoolean(SSL_ENABLED); |
| return sslConfig.getBoolean(SSL_INTERNAL_ENABLED, fallbackFlag); |
| } |
| |
| /** Checks whether SSL for the external REST endpoint is enabled. */ |
| public static boolean isRestSSLEnabled(Configuration sslConfig) { |
| @SuppressWarnings("deprecation") |
| final boolean fallbackFlag = sslConfig.getBoolean(SSL_ENABLED); |
| return sslConfig.getBoolean(SSL_REST_ENABLED, fallbackFlag); |
| } |
| |
| /** Checks whether mutual SSL authentication for the external REST endpoint is enabled. */ |
| public static boolean isRestSSLAuthenticationEnabled(Configuration sslConfig) { |
| checkNotNull(sslConfig, "sslConfig"); |
| return isRestSSLEnabled(sslConfig) && sslConfig.getBoolean(SSL_REST_AUTHENTICATION_ENABLED); |
| } |
| } |
