Merge "[FAB-1809] Enable tls config for Kafka connection"

images/kafka/Dockerfile.in

@@ -15,6 +15,7 @@ ADD payload/kafka-run-class.sh /opt/kafka/bin/kafka-run-class.sh
 ADD payload/docker-entrypoint.sh /docker-entrypoint.sh
 
 EXPOSE 9092
+EXPOSE 9093
 
 ENTRYPOINT ["/docker-entrypoint.sh"]
 CMD ["/opt/kafka/bin/kafka-server-start.sh"]

orderer/kafka/consumer.go

@@ -16,7 +16,10 @@ limitations under the License.
 
 package kafka
 
-import "github.com/Shopify/sarama"
+import (
+ "github.com/Shopify/sarama"
+ "github.com/hyperledger/fabric/orderer/localconfig"
+)
 
 // Consumer allows the caller to receive a stream of blobs from the Kafka cluster for a specific partition.
 type Consumer interface {
@@ -29,8 +32,8 @@ type consumerImpl struct {
  partition sarama.PartitionConsumer
 }
 
-func newConsumer(brokers []string, kafkaVersion sarama.KafkaVersion, cp ChainPartition, offset int64) (Consumer, error) {
- parent, err := sarama.NewConsumer(brokers, newBrokerConfig(kafkaVersion, rawPartition))
+func newConsumer(brokers []string, kafkaVersion sarama.KafkaVersion, tls config.TLS, cp ChainPartition, offset int64) (Consumer, error) {
+ parent, err := sarama.NewConsumer(brokers, newBrokerConfig(kafkaVersion, rawPartition, tls))
  if err != nil {
  return nil, err
  }

orderer/kafka/orderer.go

@@ -29,50 +29,51 @@ import (
 )
 
 // New creates a Kafka-backed consenter. Called by orderer's main.go.
-func New(kv sarama.KafkaVersion, ro config.Retry) multichain.Consenter {
- return newConsenter(kv, ro, bfValue, pfValue, cfValue)
+func New(kv sarama.KafkaVersion, ro config.Retry, tls config.TLS) multichain.Consenter {
+ return newConsenter(kv, ro, tls, bfValue, pfValue, cfValue)
 }
 
 // New calls here because we need to pass additional arguments to
 // the constructor and New() should only read from the config file.
-func newConsenter(kv sarama.KafkaVersion, ro config.Retry, bf bfType, pf pfType, cf cfType) multichain.Consenter {
- return &consenterImpl{kv, ro, bf, pf, cf}
+func newConsenter(kv sarama.KafkaVersion, ro config.Retry, tls config.TLS, bf bfType, pf pfType, cf cfType) multichain.Consenter {
+ return &consenterImpl{kv, ro, tls, bf, pf, cf}
 }
 
 // bfType defines the signature of the broker constructor.
 type bfType func([]string, ChainPartition) (Broker, error)
 
 // pfType defines the signature of the producer constructor.
-type pfType func([]string, sarama.KafkaVersion, config.Retry) Producer
+type pfType func([]string, sarama.KafkaVersion, config.Retry, config.TLS) Producer
 
 // cfType defines the signature of the consumer constructor.
-type cfType func([]string, sarama.KafkaVersion, ChainPartition, int64) (Consumer, error)
+type cfType func([]string, sarama.KafkaVersion, config.TLS, ChainPartition, int64) (Consumer, error)
 
 // bfValue holds the value for the broker constructor that's used in the non-test case.
 var bfValue = func(brokers []string, cp ChainPartition) (Broker, error) {
  return newBroker(brokers, cp)
 }
 
 // pfValue holds the value for the producer constructor that's used in the non-test case.
-var pfValue = func(brokers []string, kafkaVersion sarama.KafkaVersion, retryOptions config.Retry) Producer {
- return newProducer(brokers, kafkaVersion, retryOptions)
+var pfValue = func(brokers []string, kafkaVersion sarama.KafkaVersion, retryOptions config.Retry, tls config.TLS) Producer {
+ return newProducer(brokers, kafkaVersion, retryOptions, tls)
 }
 
 // cfValue holds the value for the consumer constructor that's used in the non-test case.
-var cfValue = func(brokers []string, kafkaVersion sarama.KafkaVersion, cp ChainPartition, offset int64) (Consumer, error) {
- return newConsumer(brokers, kafkaVersion, cp, offset)
+var cfValue = func(brokers []string, kafkaVersion sarama.KafkaVersion, tls config.TLS, cp ChainPartition, offset int64) (Consumer, error) {
+ return newConsumer(brokers, kafkaVersion, tls, cp, offset)
 }
 
 // consenterImpl holds the implementation of type that satisfies the
 // multichain.Consenter and testableConsenter interfaces. The former
 // is needed because that is what the HandleChain contract requires.
 // The latter is needed for testing.
 type consenterImpl struct {
- kv sarama.KafkaVersion
- ro config.Retry
- bf bfType
- pf pfType
- cf cfType
+ kv sarama.KafkaVersion
+ ro config.Retry
+ tls config.TLS
+ bf bfType
+ pf pfType
+ cf cfType
 }
 
 // HandleChain creates/returns a reference to a Chain for the given set of support resources.
@@ -110,7 +111,7 @@ func newChain(consenter testableConsenter, support multichain.ConsenterSupport,
  partition: newChainPartition(support.ChainID(), rawPartition),
  batchTimeout: support.SharedConfig().BatchTimeout(),
  lastOffsetPersisted: lastOffsetPersisted,
- producer: consenter.prodFunc()(support.SharedConfig().KafkaBrokers(), consenter.kafkaVersion(), consenter.retryOptions()),
+ producer: consenter.prodFunc()(support.SharedConfig().KafkaBrokers(), consenter.kafkaVersion(), consenter.retryOptions(), consenter.tlsConfig()),
  halted: false, // Redundant as the default value for booleans is false but added for readability
  exitChan: make(chan struct{}),
  haltedChan: make(chan struct{}),
@@ -123,13 +124,15 @@ func newChain(consenter testableConsenter, support multichain.ConsenterSupport,
 type testableConsenter interface {
  kafkaVersion() sarama.KafkaVersion
  retryOptions() config.Retry
+ tlsConfig() config.TLS
  brokFunc() bfType
  prodFunc() pfType
  consFunc() cfType
 }
 
 func (co *consenterImpl) kafkaVersion() sarama.KafkaVersion { return co.kv }
 func (co *consenterImpl) retryOptions() config.Retry { return co.ro }
+func (co *consenterImpl) tlsConfig() config.TLS { return co.tls }
 func (co *consenterImpl) brokFunc() bfType { return co.bf }
 func (co *consenterImpl) prodFunc() pfType { return co.pf }
 func (co *consenterImpl) consFunc() cfType { return co.cf }
@@ -169,7 +172,7 @@ func (ch *chainImpl) Start() {
  }
 
  // 2. Set up the listener/consumer for this partition.
- consumer, err := ch.consenter.consFunc()(ch.support.SharedConfig().KafkaBrokers(), ch.consenter.kafkaVersion(), ch.partition, ch.lastOffsetPersisted+1)
+ consumer, err := ch.consenter.consFunc()(ch.support.SharedConfig().KafkaBrokers(), ch.consenter.kafkaVersion(), ch.consenter.tlsConfig(), ch.partition, ch.lastOffsetPersisted+1)
  if err != nil {
  logger.Criticalf("Cannot retrieve required offset from Kafka cluster for chain %s: %s", ch.partition, err)
  close(ch.exitChan)

orderer/kafka/orderer_test.go

@@ -51,14 +51,16 @@ func mockNewConsenter(t *testing.T, kafkaVersion sarama.KafkaVersion, retryOptio
  prodDisk := make(chan *ab.KafkaMessage)
  consDisk := make(chan *ab.KafkaMessage)
 
+ mockTLS := config.TLS{Enabled: false}
+
  mockBfValue := func(brokers []string, cp ChainPartition) (Broker, error) {
  return mockNewBroker(t, cp)
  }
- mockPfValue := func(brokers []string, kafkaVersion sarama.KafkaVersion, retryOptions config.Retry) Producer {
+ mockPfValue := func(brokers []string, kafkaVersion sarama.KafkaVersion, retryOptions config.Retry, tls config.TLS) Producer {
  // The first Send on this producer will return a blob with offset #nextProducedOffset
  return mockNewProducer(t, cp, nextProducedOffset, prodDisk)
  }
- mockCfValue := func(brokers []string, kafkaVersion sarama.KafkaVersion, cp ChainPartition, lastPersistedOffset int64) (Consumer, error) {
+ mockCfValue := func(brokers []string, kafkaVersion sarama.KafkaVersion, tls config.TLS, cp ChainPartition, lastPersistedOffset int64) (Consumer, error) {
  if lastPersistedOffset != nextProducedOffset {
  panic(fmt.Errorf("Mock objects about to be set up incorrectly (consumer to seek to %d, producer to post %d)", lastPersistedOffset, nextProducedOffset))
  }
@@ -67,11 +69,12 @@ func mockNewConsenter(t *testing.T, kafkaVersion sarama.KafkaVersion, retryOptio
 
  return &mockConsenterImpl{
  consenterImpl: consenterImpl{
- kv: kafkaVersion,
- ro: retryOptions,
- bf: mockBfValue,
- pf: mockPfValue,
- cf: mockCfValue,
+ kv: kafkaVersion,
+ ro: retryOptions,
+ tls: mockTLS,
+ bf: mockBfValue,
+ pf: mockPfValue,
+ cf: mockCfValue,
  },
  prodDisk: prodDisk,
  consDisk: consDisk,

orderer/kafka/producer.go

@@ -34,10 +34,10 @@ type producerImpl struct {
  producer sarama.SyncProducer
 }
 
-func newProducer(brokers []string, kafkaVersion sarama.KafkaVersion, retryOptions config.Retry) Producer {
+func newProducer(brokers []string, kafkaVersion sarama.KafkaVersion, retryOptions config.Retry, tls config.TLS) Producer {
  var p sarama.SyncProducer
  var err error
- brokerConfig := newBrokerConfig(kafkaVersion, rawPartition)
+ brokerConfig := newBrokerConfig(kafkaVersion, rawPartition, tls)
 
  repeatTick := time.NewTicker(retryOptions.Period)
  panicTick := time.NewTicker(retryOptions.Stop)

orderer/kafka/util.go

@@ -17,15 +17,19 @@ limitations under the License.
 package kafka
 
 import (
+ "crypto/tls"
+ "crypto/x509"
+ "fmt"
  "strconv"
 
  "github.com/Shopify/sarama"
+ "github.com/hyperledger/fabric/orderer/localconfig"
  ab "github.com/hyperledger/fabric/protos/orderer"
 )
 
 // TODO Set the returned config file to more appropriate
 // defaults as we're getting closer to a stable release
-func newBrokerConfig(kafkaVersion sarama.KafkaVersion, chosenStaticPartition int32) *sarama.Config {
+func newBrokerConfig(kafkaVersion sarama.KafkaVersion, chosenStaticPartition int32, tlsConfig config.TLS) *sarama.Config {
  brokerConfig := sarama.NewConfig()
 
  brokerConfig.Version = kafkaVersion
@@ -40,6 +44,31 @@ func newBrokerConfig(kafkaVersion sarama.KafkaVersion, chosenStaticPartition int
  // value of a Kafka broker's socket.request.max.bytes property (100 MiB).
  brokerConfig.Producer.MaxMessageBytes = int(sarama.MaxRequestSize)
 
+ brokerConfig.Net.TLS.Enable = tlsConfig.Enabled
+
+ if brokerConfig.Net.TLS.Enable {
+ // create public/private key pair structure
+ keyPair, err := tls.X509KeyPair([]byte(tlsConfig.Certificate), []byte(tlsConfig.PrivateKey))
+ if err != nil {
+ panic(fmt.Errorf("Unable to decode public/private key pair. Error: %v", err))
+ }
+
+ // create root CA pool
+ rootCAs := x509.NewCertPool()
+ for _, certificate := range tlsConfig.RootCAs {
+ if !rootCAs.AppendCertsFromPEM([]byte(certificate)) {
+ panic(fmt.Errorf("Unable to decode certificate. Error: %v", err))
+ }
+ }
+
+ brokerConfig.Net.TLS.Config = &tls.Config{
+ Certificates: []tls.Certificate{keyPair},
+ RootCAs: rootCAs,
+ MinVersion: 0, // TLS 1.0 (no SSL support)
+ MaxVersion: 0, // Latest supported TLS version
+ }
+ }
+
  return brokerConfig
 }
 

orderer/kafka/util_test.go

@@ -21,6 +21,9 @@ import (
 
  "github.com/Shopify/sarama"
  "github.com/hyperledger/fabric/orderer/common/bootstrap/provisional"
+ "github.com/hyperledger/fabric/orderer/localconfig"
+ "github.com/hyperledger/fabric/orderer/mocks/util"
+ "github.com/stretchr/testify/assert"
 )
 
 func TestProducerConfigMessageMaxBytes(t *testing.T) {
@@ -35,7 +38,8 @@ func TestProducerConfigMessageMaxBytes(t *testing.T) {
  "ProduceRequest": sarama.NewMockProduceResponse(t),
  })
 
- config := newBrokerConfig(testConf.Kafka.Version, rawPartition)
+ mockTLS := config.TLS{Enabled: false}
+ config := newBrokerConfig(testConf.Kafka.Version, rawPartition, mockTLS)
  producer, err := sarama.NewSyncProducer([]string{broker.Addr()}, config)
  if err != nil {
  t.Fatal(err)
@@ -86,7 +90,7 @@ func TestNewBrokerConfig(t *testing.T) {
  "ProduceRequest": sarama.NewMockProduceResponse(t),
  })
 
- config := newBrokerConfig(testConf.Kafka.Version, differentPartition)
+ config := newBrokerConfig(testConf.Kafka.Version, differentPartition, config.TLS{Enabled: false})
  producer, err := sarama.NewSyncProducer([]string{broker.Addr()}, config)
  if err != nil {
  t.Fatal("Failed to create producer:", err)
@@ -105,3 +109,92 @@ func TestNewBrokerConfig(t *testing.T) {
  }
  }
 }
+
+func TestTLSConfigEnabled(t *testing.T) {
+ publicKey, privateKey, err := util.GenerateMockPublicPrivateKeyPairPEM(false)
+ if err != nil {
+ t.Fatalf("Enable to generate a public/private key pair: %v", err)
+ }
+ caPublicKey, _, err := util.GenerateMockPublicPrivateKeyPairPEM(true)
+ if err != nil {
+ t.Fatalf("Enable to generate a signer certificate: %v", err)
+ }
+
+ config := newBrokerConfig(testConf.Kafka.Version, 0, config.TLS{
+ Enabled: true,
+ PrivateKey: privateKey,
+ Certificate: publicKey,
+ RootCAs: []string{caPublicKey},
+ })
+
+ assert.True(t, config.Net.TLS.Enable)
+ assert.NotNil(t, config.Net.TLS.Config)
+ assert.Len(t, config.Net.TLS.Config.Certificates, 1)
+ assert.Len(t, config.Net.TLS.Config.RootCAs.Subjects(), 1)
+ assert.Equal(t, uint16(0), config.Net.TLS.Config.MaxVersion)
+ assert.Equal(t, uint16(0), config.Net.TLS.Config.MinVersion)
+}
+
+func TestTLSConfigDisabled(t *testing.T) {
+ publicKey, privateKey, err := util.GenerateMockPublicPrivateKeyPairPEM(false)
+ if err != nil {
+ t.Fatalf("Enable to generate a public/private key pair: %v", err)
+ }
+ caPublicKey, _, err := util.GenerateMockPublicPrivateKeyPairPEM(true)
+ if err != nil {
+ t.Fatalf("Enable to generate a signer certificate: %v", err)
+ }
+
+ config := newBrokerConfig(testConf.Kafka.Version, 0, config.TLS{
+ Enabled: false,
+ PrivateKey: privateKey,
+ Certificate: publicKey,
+ RootCAs: []string{caPublicKey},
+ })
+
+ assert.False(t, config.Net.TLS.Enable)
+ assert.Zero(t, config.Net.TLS.Config)
+
+}
+
+func TestTLSConfigBadCert(t *testing.T) {
+ publicKey, privateKey, err := util.GenerateMockPublicPrivateKeyPairPEM(false)
+ if err != nil {
+ t.Fatalf("Enable to generate a public/private key pair: %v", err)
+ }
+ caPublicKey, _, err := util.GenerateMockPublicPrivateKeyPairPEM(true)
+ if err != nil {
+ t.Fatalf("Enable to generate a signer certificate: %v", err)
+ }
+
+ t.Run("BadPrivateKey", func(t *testing.T) {
+ assert.Panics(t, func() {
+ newBrokerConfig(testConf.Kafka.Version, 0, config.TLS{
+ Enabled: true,
+ PrivateKey: privateKey,
+ Certificate: "TRASH",
+ RootCAs: []string{caPublicKey},
+ })
+ })
+ })
+ t.Run("BadPublicKey", func(t *testing.T) {
+ assert.Panics(t, func() {
+ newBrokerConfig(testConf.Kafka.Version, 0, config.TLS{
+ Enabled: true,
+ PrivateKey: "TRASH",
+ Certificate: publicKey,
+ RootCAs: []string{caPublicKey},
+ })
+ })
+ })
+ t.Run("BadRootCAs", func(t *testing.T) {
+ assert.Panics(t, func() {
+ newBrokerConfig(testConf.Kafka.Version, 0, config.TLS{
+ Enabled: true,
+ PrivateKey: privateKey,
+ Certificate: publicKey,
+ RootCAs: []string{"TRASH"},
+ })
+ })
+ })
+}