Skip to content

Encryption support producer - #560

Merged
cckellogg merged 18 commits into
apache:masterfrom
Fanatics:encryption-support-ext-producer
Sep 3, 2021
Merged

Encryption support producer#560
cckellogg merged 18 commits into
apache:masterfrom
Fanatics:encryption-support-ext-producer

Conversation

@GPrabhudas

Copy link
Copy Markdown
Contributor

Breakdown of PR #552

This PR includes the encryption changes at producer side.

PGarule added 3 commits July 7, 2021 23:02
- use base crypto package for encryption
Comment thread pulsar/internal/commands.go Outdated
}

// copy of the method serializeBatch(....) with an extension to encrypt payload
func serializeBatchWithEncryption(wb Buffer,

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What are the differences between serializeBatch and serializeBatchWithEncryption? It looks like a lot of duplicate code?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I preferred to put this in separate function instead of modifying the exiting one. Let me check if I can remove duplicate code

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

reused the serializeBatch

Comment thread pulsar/producer.go Outdated
PartitionsAutoDiscoveryInterval time.Duration

// EncryptionKeys list of encryption key names to encrypt session key
EncryptionKeys []string

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we wrap these into a struct? Maybe something like this?

type Encryption {
  Keys []string
  KeyReader crypto.KeyReader
  etc
]

@GPrabhudas GPrabhudas Jul 12, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Makes sense

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

Comment thread go.mod
@@ -14,10 +14,7 @@ require (
github.com/google/uuid v1.1.2
github.com/inconshreveable/mousetrap v1.0.0 // indirect
github.com/klauspost/compress v1.10.8

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Curious why is the mod and sum file changing? Can these changes be done in a separate PR?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let me recheck again :)

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

synced with master branch

Comment thread pulsar/internal/commands.go Outdated
wb.PutUint32(checksum, checksumIdx)
}

func encryptPayload(msgMetadata *pb.MessageMetadata,

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should this return an error?

Comment thread pulsar/internal/commands.go Outdated

// there was a error in encrypting the payload and
// crypto failure action is set to crypto.ProducerCryptoFailureActionFail
if encryptedPayload == nil {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure panic is the correct behavior here.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Its better to throw an exception because client is setting ProducerCryptoFailureActionFail to fail.

Comment thread pulsar/internal/commands.go Outdated
crypto.NewMessageMetadataSupplier(msgMetadata),
compressedPayload)

if err != nil {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should this error be log somewhere?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

// [TOTAL_SIZE] [CMD_SIZE][CMD] [MAGIC_NUMBER][CHECKSUM] [METADATA_SIZE][METADATA] [PAYLOAD]

// compress the payload
compressedPayload := compressionProvider.Compress(nil, uncompressedPayload.ReadableSlice())

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We want to compress before encrypting?

@GPrabhudas GPrabhudas Jul 12, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As per java implementation => Yes compress and then encrypt.

Comment thread pulsar/producer_partition.go Outdated
}
}

func (p *partitionProducer) updateDataKey() error {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does this need to be thread safe? Also, can we move the key refreshing to another MR.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can move it to next MR.
Just ask: why this to be moved to another MR?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

provider, _ = GetBatcherBuilderProvider(DefaultBatchBuilder)

} else if p.batchBuilder == nil {
provider, err := GetBatcherBuilderProvider(p.options.BatcherBuilderType)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why change this? The p masks the partitionProducer and make the code harder to follow.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just tried to remove duplicate code :) If it is not ok, I'll revert it.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

reverted

Comment thread pulsar/internal/batch_builder.go Outdated
return &bc, nil
}

// UseEncryptionKeys encryption key names to use

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What about removing these config options and passing an Encryptor interface/struct. I think the config functions work better when things are scoped in their own package - it make it name spacing clearer. Also, it doesn't really match the rest of the code.

type Encrypter interface {
 Encrypt(msgMetadata, payload []byte) ([]byte, error)
}

If encryption is not provided there can be a noop encryter

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done

@wolfstudy

Copy link
Copy Markdown
Member

@cckellogg @GPrabhudas Any update for this? We are ready to initiate a new version of the release, hoping to include this new change in the new version (because some of the code has been merged in), can we consider advancing the progress here.

@GPrabhudas

Copy link
Copy Markdown
Contributor Author

@cckellogg @GPrabhudas Any update for this? We are ready to initiate a new version of the release, hoping to include this new change in the new version (because some of the code has been merged in), can we consider advancing the progress here.

I'll work on it and address the review suggestions.

@cckellogg

Copy link
Copy Markdown
Contributor

@wolfstudy what is your timeframe for a new release?

@wolfstudy

wolfstudy commented Jul 20, 2021

Copy link
Copy Markdown
Member

@wolfstudy what is your timeframe for a new release?

What I want is to start the release process after merging the change related to this function. If the development of this feature takes some time, we can consider including it in the next version.

@cckellogg

Copy link
Copy Markdown
Contributor

@wolfstudy what is your timeframe for a new release?

What I want is to start the release process after merging the change related to this function. If the development of this feature takes some time, we can consider including it in the next version.

There are additional MRs (besides this one) to get this feature working. Currently, just the crypto package has been merged. It's probably better to cut a release now and then add the crypto implementation (producer and consumer) into the next release. To me that is better than rushing to get MRs in.

@wolfstudy wolfstudy modified the milestones: 0.6.0, 0.7.0 Jul 21, 2021
@GPrabhudas
GPrabhudas requested a review from cckellogg July 24, 2021 16:30
// Encrypt producer encryptor
func (e *producerEncryptor) Encrypt(payload []byte, msgMetadata crypto.MessageMetadataSupplier) ([]byte, error) {
// encryption is enabled but KeyReader interface is not implemented
if e.keyReader == nil {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should this be detected and an error raised while setting up the producer?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

Comment thread pulsar/producer_partition.go Outdated
logCtx := fmt.Sprintf("[%v] [%v] [%v]", p.topic, p.producerName, p.producerID)
messageCrypto, err := crypto.NewDefaultMessageCrypto(logCtx, true, logger)
if err != nil {
logger.WithError(err).Error("Unable to get MessageCrypto instance. Producer creation is abandoned")

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Will there be more context in the err of why this failed?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes. Error is returned if generation of data key is failed. Here we are logging the same error with other information.

Comment thread pulsar/encryption.go Outdated
// ProducerEncryptionInfo encryption related fields required by the producer
type ProducerEncryptionInfo struct {
// KeyReader read RSA public/private key pairs
Keyreader crypto.KeyReader

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

KeyReader

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

Comment thread pulsar/internal/crypto/encryptor.go Outdated

// Encryptor support encryption
type Encryptor interface {
Encrypt([]byte, crypto.MessageMetadataSupplier) ([]byte, error)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this is internal we can pass the *pb.MessageMetadata and avoid having to create a supplier for each message when encryption is not being used.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

Comment thread pulsar/internal/commands.go Outdated
Comment thread pulsar/internal/commands.go Outdated
// send unencrypted message
if e.producerCryptoFailureAction == crypto.ProducerCryptoFailureActionSend {
e.logger.Errorf("Encryption of payload failed : %v", err)
e.logger.Warn("ProducerCryptoFailureAction is set to send, sending unecrypted message")

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would log only one warning message here.

logger.WithError(err).Warnf("Encryption failed for payload sending unencrypted message ProducerCryptoFailureAction is set to send")

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done.

Comment thread pulsar/producer_partition.go Outdated
func (p *partitionProducer) generateDataKey() error {
if p.options.Encryption != nil {
if p.options.Encryption.KeyReader != nil {
return p.options.Encryption.MessageCrypto.AddPublicKeyCipher(p.options.Encryption.Keys,

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could the MessageCrypto be nil here?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No. It won't be nil.
If not provided, it is getting initialized with default one.
https://github.com/Fanatics/pulsar-client-go/blob/encryption-support-ext-producer/pulsar/producer_partition.go#L141

}

if encryption.MessageCrypto == nil {
logCtx := fmt.Sprintf("[%v] [%v] [%v]", p.topic, p.producerName, p.producerID)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe this can be update in another MR. It might be better to create a context logger instead of sending a string

cl := logger.WithFields(log.Fields{
    "topic": topic,
    "producer": producerName,
    "producerID", producerID
  })

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure

batchesData[idx] = b
sequenceIDs[idx] = s
callbacks[idx] = c
} else if err != nil {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should the error be set with the other arrays so they are always the same length? the only way b can be nil is if there are no messages right.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@cckellogg can this PR me merged if all looks good :)

@cckellogg
cckellogg merged commit 1d3a9cc into apache:master Sep 3, 2021
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

4 participants