Skip to content

Commit

Permalink
Use SecurityUtility class (#16)
Browse files Browse the repository at this point in the history
  • Loading branch information
@rdhabalia
rdhabalia authored and Jai Asher committed Feb 13, 2018
1 parent 136c241 commit d8f6068
Show file tree
Hide file tree
Showing 7 changed files with 82 additions and 133 deletions.
36 changes: 2 additions & 34 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarChannelInitializer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,21 +18,14 @@
*/
package org.apache.pulsar.broker.service;

import java.io.File;
import java.security.cert.X509Certificate;

import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.client.impl.auth.AuthenticationDataTls;
import org.apache.pulsar.common.api.PulsarDecoder;
import org.apache.pulsar.common.api.PulsarLengthFieldFrameDecoder;
import org.apache.pulsar.common.util.SecurityUtility;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.ssl.ClientAuth;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;

public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel> {

Expand All @@ -56,32 +49,7 @@ public PulsarChannelInitializer(BrokerService brokerService, ServiceConfiguratio
@Override
protected void initChannel(SocketChannel ch) throws Exception {
if (enableTLS) {
File tlsCert = new File(serviceConfig.getTlsCertificateFilePath());
File tlsKey = new File(serviceConfig.getTlsKeyFilePath());
SslContextBuilder builder = SslContextBuilder.forServer(tlsCert, tlsKey);
if (serviceConfig.isTlsAllowInsecureConnection()) {
builder.trustManager(InsecureTrustManagerFactory.INSTANCE);
} else {
if (serviceConfig.getTlsTrustCertsFilePath().isEmpty()) {
// Use system default
builder.trustManager((File) null);
} else {
File trustCertCollection = new File(serviceConfig.getTlsTrustCertsFilePath());
builder.trustManager(trustCertCollection);
}
}

ServiceConfiguration config = brokerService.pulsar().getConfiguration();
String certFilePath = config.getTlsCertificateFilePath();
String keyFilePath = config.getTlsKeyFilePath();
if (StringUtils.isNotBlank(certFilePath) && StringUtils.isNotBlank(keyFilePath)) {
AuthenticationDataTls authTlsData = new AuthenticationDataTls(certFilePath, keyFilePath);
builder.keyManager(authTlsData.getTlsPrivateKey(),
(X509Certificate[]) authTlsData.getTlsCertificates());
}


SslContext sslCtx = builder.clientAuth(ClientAuth.OPTIONAL).build();
SslContext sslCtx = SecurityUtility.createNettySslContextForServer(serviceConfig.isTlsAllowInsecureConnection(), serviceConfig.getTlsTrustCertsFilePath(), serviceConfig.getTlsCertificateFilePath(), serviceConfig.getTlsKeyFilePath());
ch.pipeline().addLast(TLS_HANDLER, sslCtx.newHandler(ch.alloc()));
}
ch.pipeline().addLast("frameDecoder",
Expand Down
31 changes: 12 additions & 19 deletions pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
package org.apache.pulsar.client.impl;

import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.security.cert.X509Certificate;
Expand All @@ -32,6 +31,7 @@
import org.apache.pulsar.client.api.ClientConfiguration;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.api.PulsarLengthFieldFrameDecoder;
import org.apache.pulsar.common.util.SecurityUtility;
import org.apache.pulsar.common.util.netty.EventLoopUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand All @@ -47,6 +47,9 @@
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
import io.netty.resolver.dns.DnsNameResolver;
import io.netty.resolver.dns.DnsNameResolverBuilder;
import io.netty.util.concurrent.Future;

public class ConnectionPool implements Closeable {
private final ConcurrentHashMap<InetSocketAddress, ConcurrentMap<Integer, CompletableFuture<ClientCnx>>> pool;
Expand Down Expand Up @@ -74,27 +77,17 @@ public ConnectionPool(final PulsarClientImpl client, EventLoopGroup eventLoopGro
public void initChannel(SocketChannel ch) throws Exception {
ClientConfiguration clientConfig = client.getConfiguration();
if (clientConfig.isUseTls()) {
SslContextBuilder builder = SslContextBuilder.forClient();
if (clientConfig.isTlsAllowInsecureConnection()) {
builder.trustManager(InsecureTrustManagerFactory.INSTANCE);
} else {
if (clientConfig.getTlsTrustCertsFilePath().isEmpty()) {
// Use system default
builder.trustManager((File) null);
} else {
File trustCertCollection = new File(clientConfig.getTlsTrustCertsFilePath());
builder.trustManager(trustCertCollection);
}
}

SslContext sslCtx;
// Set client certificate if available
AuthenticationDataProvider authData = clientConfig.getAuthentication().getAuthData();
if (authData.hasDataForTls()) {
builder.keyManager(authData.getTlsPrivateKey(),
(X509Certificate[]) authData.getTlsCertificates());
sslCtx = SecurityUtility.createNettySslContextForClient(clientConfig.isTlsAllowInsecureConnection(),
clientConfig.getTlsTrustCertsFilePath(), (X509Certificate[]) authData.getTlsCertificates(),
authData.getTlsPrivateKey());
} else {
sslCtx = SecurityUtility.createNettySslContextForClient(clientConfig.isTlsAllowInsecureConnection(),
clientConfig.getTlsTrustCertsFilePath());
}

SslContext sslCtx = builder.build();
ch.pipeline().addLast(TLS_HANDLER, sslCtx.newHandler(ch.alloc()));
}
ch.pipeline().addLast("frameDecoder", new PulsarLengthFieldFrameDecoder(MaxMessageSize, 0, 4, 0, 4));
Expand Down Expand Up @@ -184,7 +177,7 @@ private CompletableFuture<ClientCnx> createConnection(InetSocketAddress logicalA
}

cnx.setRemoteHostName(physicalAddress.getHostName());

cnx.connectionFuture().thenRun(() -> {
if (log.isDebugEnabled()) {
log.debug("[{}] Connection handshake completed", cnx.channel());
Expand Down
4 changes: 2 additions & 2 deletions pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -92,10 +92,10 @@ public boolean keepAlive(Request ahcRequest, HttpRequest request, HttpResponse r
// Set client key and certificate if available
AuthenticationDataProvider authData = authentication.getAuthData();
if (authData.hasDataForTls()) {
sslCtx = SecurityUtility.createNettySslContext(tlsAllowInsecureConnection, tlsTrustCertsFilePath,
sslCtx = SecurityUtility.createNettySslContextForClient(tlsAllowInsecureConnection, tlsTrustCertsFilePath,
authData.getTlsCertificates(), authData.getTlsPrivateKey());
} else {
sslCtx = SecurityUtility.createNettySslContext(tlsAllowInsecureConnection, tlsTrustCertsFilePath);
sslCtx = SecurityUtility.createNettySslContextForClient(tlsAllowInsecureConnection, tlsTrustCertsFilePath);
}

confBuilder.setSslContext(sslCtx);
Expand Down
56 changes: 45 additions & 11 deletions pulsar-common/src/main/java/org/apache/pulsar/common/util/SecurityUtility.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,26 +18,36 @@
*/
package org.apache.pulsar.common.util;

import java.io.*;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.security.GeneralSecurityException;
import java.security.KeyFactory;
import java.security.KeyManagementException;
import java.security.KeyStore;
import java.security.PrivateKey;
import java.security.SecureRandom;
import java.security.cert.Certificate;
import java.security.cert.X509Certificate;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.security.spec.KeySpec;
import java.security.spec.PKCS8EncodedKeySpec;
import java.util.Base64;
import java.util.Collection;

import javax.net.ssl.*;
import javax.net.ssl.KeyManager;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLException;
import javax.net.ssl.TrustManager;
import javax.net.ssl.TrustManagerFactory;

import io.netty.handler.ssl.ClientAuth;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.SslProvider;
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;

public class SecurityUtility {
Expand All @@ -47,9 +57,10 @@ public static SSLContext createSslContext(boolean allowInsecureConnection, Certi
return createSslContext(allowInsecureConnection, trustCertificates, (Certificate[]) null, (PrivateKey) null);
}

public static SslContext createNettySslContext(boolean allowInsecureConnection, String trustCertsFilePath)
public static SslContext createNettySslContextForClient(boolean allowInsecureConnection, String trustCertsFilePath)
throws GeneralSecurityException, SSLException, FileNotFoundException {
return createNettySslContext(allowInsecureConnection, trustCertsFilePath, (Certificate[]) null, (PrivateKey) null);
return createNettySslContextForClient(allowInsecureConnection, trustCertsFilePath, (Certificate[]) null,
(PrivateKey) null);
}

public static SSLContext createSslContext(boolean allowInsecureConnection, String trustCertsFilePath,
Expand All @@ -60,15 +71,17 @@ public static SSLContext createSslContext(boolean allowInsecureConnection, Strin
return createSslContext(allowInsecureConnection, trustCertificates, certificates, privateKey);
}

public static SslContext createNettySslContext(boolean allowInsecureConnection, String trustCertsFilePath,
String certFilePath, String keyFilePath) throws GeneralSecurityException, SSLException, FileNotFoundException {
public static SslContext createNettySslContextForClient(boolean allowInsecureConnection, String trustCertsFilePath,
String certFilePath, String keyFilePath)
throws GeneralSecurityException, SSLException, FileNotFoundException {
X509Certificate[] certificates = loadCertificatesFromPemFile(certFilePath);
PrivateKey privateKey = loadPrivateKeyFromPemFile(keyFilePath);
return createNettySslContext(allowInsecureConnection, trustCertsFilePath, certificates, privateKey);
return createNettySslContextForClient(allowInsecureConnection, trustCertsFilePath, certificates, privateKey);
}

public static SslContext createNettySslContext(boolean allowInsecureConnection, String trustCertsFilePath,
Certificate[] certificates, PrivateKey privateKey) throws GeneralSecurityException, SSLException, FileNotFoundException {
public static SslContext createNettySslContextForClient(boolean allowInsecureConnection, String trustCertsFilePath,
Certificate[] certificates, PrivateKey privateKey)
throws GeneralSecurityException, SSLException, FileNotFoundException {
SslContextBuilder builder = SslContextBuilder.forClient();
if (allowInsecureConnection) {
builder.trustManager(InsecureTrustManagerFactory.INSTANCE);
Expand All @@ -81,6 +94,27 @@ public static SslContext createNettySslContext(boolean allowInsecureConnection,
return builder.build();
}

public static SslContext createNettySslContextForServer(boolean allowInsecureConnection, String trustCertsFilePath,
String certFilePath, String keyFilePath)
throws GeneralSecurityException, SSLException, FileNotFoundException {
X509Certificate[] certificates = loadCertificatesFromPemFile(certFilePath);
PrivateKey privateKey = loadPrivateKeyFromPemFile(keyFilePath);

SslContextBuilder builder = SslContextBuilder.forServer(privateKey, (X509Certificate[]) certificates);
if (allowInsecureConnection) {
builder.trustManager(InsecureTrustManagerFactory.INSTANCE);
} else {
if (trustCertsFilePath != null && trustCertsFilePath.length() != 0) {
builder.trustManager(new FileInputStream(trustCertsFilePath));
} else {
builder.trustManager((File) null);
}
}
builder.keyManager(privateKey, (X509Certificate[]) certificates);
builder.clientAuth(ClientAuth.OPTIONAL);
return builder.build();
}

public static SSLContext createSslContext(boolean allowInsecureConnection, Certificate[] trustCertficates,
Certificate[] certificates, PrivateKey privateKey) throws GeneralSecurityException {
KeyStoreHolder ksh = new KeyStoreHolder();
Expand Down
27 changes: 6 additions & 21 deletions ...-service/src/main/java/org/apache/pulsar/discovery/service/ServiceChannelInitializer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,18 +18,14 @@
*/
package org.apache.pulsar.discovery.service;

import java.io.File;

import org.apache.pulsar.common.api.PulsarDecoder;
import org.apache.pulsar.common.api.PulsarLengthFieldFrameDecoder;
import org.apache.pulsar.common.util.SecurityUtility;
import org.apache.pulsar.discovery.service.server.ServiceConfig;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.ssl.ClientAuth;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;

/**
* Initialize service channel handlers.
Expand All @@ -42,7 +38,8 @@ public class ServiceChannelInitializer extends ChannelInitializer<SocketChannel>
private DiscoveryService discoveryService;
private boolean enableTLS;

public ServiceChannelInitializer(DiscoveryService discoveryService, ServiceConfig serviceConfig, boolean enableTLS) {
public ServiceChannelInitializer(DiscoveryService discoveryService, ServiceConfig serviceConfig,
boolean enableTLS) {
super();
this.serviceConfig = serviceConfig;
this.discoveryService = discoveryService;
Expand All @@ -52,21 +49,9 @@ public ServiceChannelInitializer(DiscoveryService discoveryService, ServiceConfi
@Override
protected void initChannel(SocketChannel ch) throws Exception {
if (enableTLS) {
File tlsCert = new File(serviceConfig.getTlsCertificateFilePath());
File tlsKey = new File(serviceConfig.getTlsKeyFilePath());
SslContextBuilder builder = SslContextBuilder.forServer(tlsCert, tlsKey);
if (serviceConfig.isTlsAllowInsecureConnection()) {
builder.trustManager(InsecureTrustManagerFactory.INSTANCE);
} else {
if (serviceConfig.getTlsTrustCertsFilePath().isEmpty()) {
// Use system default
builder.trustManager((File) null);
} else {
File trustCertCollection = new File(serviceConfig.getTlsTrustCertsFilePath());
builder.trustManager(trustCertCollection);
}
}
SslContext sslCtx = builder.clientAuth(ClientAuth.OPTIONAL).build();
SslContext sslCtx = SecurityUtility.createNettySslContextForServer(
serviceConfig.isTlsAllowInsecureConnection(), serviceConfig.getTlsTrustCertsFilePath(),
serviceConfig.getTlsCertificateFilePath(), serviceConfig.getTlsKeyFilePath());
ch.pipeline().addLast(TLS_HANDLER, sslCtx.newHandler(ch.alloc()));
}
ch.pipeline().addLast("frameDecoder",
Expand Down
32 changes: 10 additions & 22 deletions pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,19 +19,20 @@

package org.apache.pulsar.proxy.server;

import java.io.File;
import java.net.URI;
import java.net.URISyntaxException;
import java.security.cert.X509Certificate;

import javax.net.ssl.SSLSession;

import org.apache.http.conn.ssl.DefaultHostnameVerifier;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
import org.apache.pulsar.common.api.Commands;
import org.apache.pulsar.common.api.PulsarDecoder;
import org.apache.pulsar.common.api.PulsarLengthFieldFrameDecoder;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandConnected;
import org.apache.pulsar.common.util.SecurityUtility;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand All @@ -46,12 +47,9 @@
import io.netty.channel.ChannelOption;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.FutureListener;
import org.apache.http.conn.ssl.DefaultHostnameVerifier;

public class DirectProxyHandler {

Expand All @@ -73,7 +71,7 @@ public DirectProxyHandler(ProxyService service, ProxyConnection proxyConnection,
this.clientAuthMethod = proxyConnection.clientAuthMethod;
ProxyConfiguration config = service.getConfiguration();
this.forwardAuthData = service.getConfiguration().forwardAuthorizationCredentials();

// Start the connection attempt.
Bootstrap b = new Bootstrap();
// Tie the backend connection on the same thread to avoid context switches when passing data between the 2
Expand All @@ -84,27 +82,17 @@ public DirectProxyHandler(ProxyService service, ProxyConnection proxyConnection,
@Override
protected void initChannel(SocketChannel ch) throws Exception {
if (config.isTlsEnabledWithBroker()) {
SslContextBuilder builder = SslContextBuilder.forClient();
if (config.isTlsAllowInsecureConnection()) {
builder.trustManager(InsecureTrustManagerFactory.INSTANCE);
} else {
if (config.getTlsTrustCertsFilePath().isEmpty()) {
// Use system default
builder.trustManager((File) null);
} else {
File trustCertCollection = new File(config.getTlsTrustCertsFilePath());
builder.trustManager(trustCertCollection);
}
}

SslContext sslCtx;
// Set client certificate if available
AuthenticationDataProvider authData = authentication.getAuthData();
if (authData.hasDataForTls()) {
builder.keyManager(authData.getTlsPrivateKey(),
(X509Certificate[]) authData.getTlsCertificates());
sslCtx = SecurityUtility.createNettySslContextForClient(config.isTlsAllowInsecureConnection(),
config.getTlsTrustCertsFilePath(), (X509Certificate[]) authData.getTlsCertificates(),
authData.getTlsPrivateKey());
} else {
sslCtx = SecurityUtility.createNettySslContextForClient(config.isTlsAllowInsecureConnection(),
config.getTlsTrustCertsFilePath());
}

SslContext sslCtx = builder.build();
ch.pipeline().addLast(TLS_HANDLER, sslCtx.newHandler(ch.alloc()));
}
ch.pipeline().addLast("frameDecoder",
Expand Down
Loading

0 comments on commit d8f6068

Please sign in to comment.