encryption_properties.go

Documentation: github.com/apache/arrow/go/v15/parquet

     1  // Licensed to the Apache Software Foundation (ASF) under one
     2  // or more contributor license agreements.  See the NOTICE file
     3  // distributed with this work for additional information
     4  // regarding copyright ownership.  The ASF licenses this file
     5  // to you under the Apache License, Version 2.0 (the
     6  // "License"); you may not use this file except in compliance
     7  // with the License.  You may obtain a copy of the License at
     8  //
     9  // http://www.apache.org/licenses/LICENSE-2.0
    10  //
    11  // Unless required by applicable law or agreed to in writing, software
    12  // distributed under the License is distributed on an "AS IS" BASIS,
    13  // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14  // See the License for the specific language governing permissions and
    15  // limitations under the License.
    16  
    17  package parquet
    18  
    19  import (
    20  	"crypto/rand"
    21  	"unicode/utf8"
    22  
    23  	format "github.com/apache/arrow/go/v15/parquet/internal/gen-go/parquet"
    24  )
    25  
    26  // Constants that will be used as the default values with encryption/decryption
    27  const (
    28  	// By default we'll use AesGCM as our encryption algorithm
    29  	DefaultEncryptionAlgorithm       = AesGcm
    30  	MaximalAadMetadataLength   int32 = 256
    31  	// if encryption is turned on, we will default to also encrypting the footer
    32  	DefaultEncryptedFooter = true
    33  	DefaultCheckSignature  = true
    34  	// by default if you set the file decryption properties, we will error
    35  	// on any plaintext files unless otherwise specified.
    36  	DefaultAllowPlaintextFiles       = false
    37  	AadFileUniqueLength        int32 = 8
    38  )
    39  
    40  // ColumnPathToDecryptionPropsMap maps column paths to decryption properties
    41  type ColumnPathToDecryptionPropsMap map[string]*ColumnDecryptionProperties
    42  
    43  // ColumnPathToEncryptionPropsMap maps column paths to encryption properties
    44  type ColumnPathToEncryptionPropsMap map[string]*ColumnEncryptionProperties
    45  
    46  // ColumnEncryptionProperties specifies how to encrypt a given column
    47  type ColumnEncryptionProperties struct {
    48  	columnPath             string
    49  	encrypted              bool
    50  	encryptedWithFooterKey bool
    51  	key                    string
    52  	keyMetadata            string
    53  	utilized               bool
    54  }
    55  
    56  // ColumnPath returns which column these properties are for
    57  func (ce *ColumnEncryptionProperties) ColumnPath() string {
    58  	return ce.columnPath
    59  }
    60  
    61  // IsEncrypted returns true if this column is encrypted.
    62  func (ce *ColumnEncryptionProperties) IsEncrypted() bool { return ce.encrypted }
    63  
    64  // IsEncryptedWithFooterKey returns if this column was encrypted with the footer key itself, or false if a separate
    65  // key was used for encrypting this column.
    66  func (ce *ColumnEncryptionProperties) IsEncryptedWithFooterKey() bool {
    67  	return ce.encryptedWithFooterKey
    68  }
    69  
    70  // Key returns the key used for encrypting this column if it isn't encrypted by the footer key
    71  func (ce *ColumnEncryptionProperties) Key() string { return ce.key }
    72  
    73  // KeyMetadata returns the key identifier which is used with a KeyRetriever to get the key for this column if it is not
    74  // encrypted using the footer key
    75  func (ce *ColumnEncryptionProperties) KeyMetadata() string { return ce.keyMetadata }
    76  
    77  // WipeOutEncryptionKey Clears the encryption key, used after completion of file writing
    78  func (ce *ColumnEncryptionProperties) WipeOutEncryptionKey() { ce.key = "" }
    79  
    80  // IsUtilized returns whether or not these properties have already been used, if the key is empty
    81  // then this is always false
    82  func (ce *ColumnEncryptionProperties) IsUtilized() bool {
    83  	if ce.key == "" {
    84  		return false
    85  	}
    86  	return ce.utilized
    87  }
    88  
    89  // SetUtilized is used for marking it as utilized once it is used in FileEncryptionProperties
    90  // as the encryption key will be wiped out on completion of writing
    91  func (ce *ColumnEncryptionProperties) SetUtilized() {
    92  	ce.utilized = true
    93  }
    94  
    95  // Clone returns a instance of ColumnEncryptionProperties with the same key and metadata
    96  func (ce *ColumnEncryptionProperties) Clone() *ColumnEncryptionProperties {
    97  	copy := ce.key
    98  	return NewColumnEncryptionProperties(ce.columnPath, WithKey(copy), WithKeyMetadata(ce.keyMetadata))
    99  }
   100  
   101  type colEncryptConfig struct {
   102  	key         string
   103  	keyMetadata string
   104  	encrypted   bool
   105  }
   106  
   107  // ColumnEncryptOption how to specify options to the NewColumnEncryptionProperties function.
   108  type ColumnEncryptOption func(*colEncryptConfig)
   109  
   110  // WithKey sets a column specific key.
   111  // If key is not set on an encrypted column, the column will be encrypted with the footer key.
   112  // key length must be either 16, 24, or 32 bytes
   113  // the key is cloned and will be wiped out (array values set to 0) upon completion of file writing.
   114  // Caller is responsible for wiping out input key array
   115  func WithKey(key string) ColumnEncryptOption {
   116  	return func(c *colEncryptConfig) {
   117  		if key != "" {
   118  			c.key = key
   119  		}
   120  	}
   121  }
   122  
   123  // WithKeyMetadata sets the key retrieval metadata, use either KeyMetadata or KeyID but not both
   124  func WithKeyMetadata(keyMeta string) ColumnEncryptOption {
   125  	return func(c *colEncryptConfig) {
   126  		c.keyMetadata = keyMeta
   127  	}
   128  }
   129  
   130  // WithKeyID is a convenience function to set the key metadata using a string id.
   131  // Set a key retrieval metadata (converted from String). and use either KeyMetadata or KeyID, not both.
   132  // KeyID will be converted to metadata (UTF-8 Array)
   133  func WithKeyID(keyID string) ColumnEncryptOption {
   134  	if !utf8.ValidString(keyID) {
   135  		panic("parquet: key id should be UTF8 encoded")
   136  	}
   137  	return WithKeyMetadata(keyID)
   138  }
   139  
   140  // NewColumnEncryptionProperties constructs properties for the provided column path, modified by the options provided
   141  func NewColumnEncryptionProperties(name string, opts ...ColumnEncryptOption) *ColumnEncryptionProperties {
   142  	var cfg colEncryptConfig
   143  	cfg.encrypted = true
   144  	for _, o := range opts {
   145  		o(&cfg)
   146  	}
   147  	return &ColumnEncryptionProperties{
   148  		utilized:               false,
   149  		encrypted:              cfg.encrypted,
   150  		encryptedWithFooterKey: cfg.encrypted && cfg.key == "",
   151  		keyMetadata:            cfg.keyMetadata,
   152  		key:                    cfg.key,
   153  		columnPath:             name,
   154  	}
   155  }
   156  
   157  // ColumnDecryptionProperties are the specifications for how to decrypt a given column.
   158  type ColumnDecryptionProperties struct {
   159  	columnPath string
   160  	key        string
   161  	utilized   bool
   162  }
   163  
   164  // NewColumnDecryptionProperties constructs a new ColumnDecryptionProperties for the given column path, modified by
   165  // the provided options
   166  func NewColumnDecryptionProperties(column string, opts ...ColumnDecryptOption) *ColumnDecryptionProperties {
   167  	var cfg columnDecryptConfig
   168  	for _, o := range opts {
   169  		o(&cfg)
   170  	}
   171  
   172  	return &ColumnDecryptionProperties{
   173  		columnPath: column,
   174  		utilized:   false,
   175  		key:        cfg.key,
   176  	}
   177  }
   178  
   179  // ColumnPath returns which column these properties describe how to decrypt
   180  func (cd *ColumnDecryptionProperties) ColumnPath() string { return cd.columnPath }
   181  
   182  // Key returns the key specified to decrypt this column, or is empty if the Footer Key should be used.
   183  func (cd *ColumnDecryptionProperties) Key() string { return cd.key }
   184  
   185  // IsUtilized returns whether or not these properties have been used for decryption already
   186  func (cd *ColumnDecryptionProperties) IsUtilized() bool { return cd.utilized }
   187  
   188  // SetUtilized is used by the reader to specify when we've decrypted the column and have used the key so we know
   189  // to wipe out the keys.
   190  func (cd *ColumnDecryptionProperties) SetUtilized() { cd.utilized = true }
   191  
   192  // WipeOutDecryptionKey is called after decryption to ensure the key doesn't stick around and get re-used.
   193  func (cd *ColumnDecryptionProperties) WipeOutDecryptionKey() { cd.key = "" }
   194  
   195  // Clone returns a new instance of ColumnDecryptionProperties with the same key and column
   196  func (cd *ColumnDecryptionProperties) Clone() *ColumnDecryptionProperties {
   197  	return NewColumnDecryptionProperties(cd.columnPath, WithDecryptKey(cd.key))
   198  }
   199  
   200  type columnDecryptConfig struct {
   201  	key string
   202  }
   203  
   204  // ColumnDecryptOption is the type of the options passed for constructing Decryption Properties
   205  type ColumnDecryptOption func(*columnDecryptConfig)
   206  
   207  // WithDecryptKey specifies the key to utilize for decryption
   208  func WithDecryptKey(key string) ColumnDecryptOption {
   209  	return func(cfg *columnDecryptConfig) {
   210  		if key != "" {
   211  			cfg.key = key
   212  		}
   213  	}
   214  }
   215  
   216  // AADPrefixVerifier is an interface for any object that can be used to verify the identity of the file being decrypted.
   217  // It should panic if the provided AAD identity is bad.
   218  //
   219  // In a data set, AAD Prefixes should be collected, and then checked for missing files.
   220  type AADPrefixVerifier interface {
   221  	// Verify identity of file. panic if bad
   222  	Verify(string)
   223  }
   224  
   225  // DecryptionKeyRetriever is an interface for getting the desired key for decryption from metadata. It should take in
   226  // some metadata identifier and return the actual Key to use for decryption.
   227  type DecryptionKeyRetriever interface {
   228  	GetKey(keyMetadata []byte) string
   229  }
   230  
   231  // FileDecryptionProperties define the File Level configuration for decrypting a parquet file. Once constructed they are
   232  // read only.
   233  type FileDecryptionProperties struct {
   234  	footerKey                     string
   235  	aadPrefix                     string
   236  	checkPlaintextFooterIntegrity bool
   237  	plaintextAllowed              bool
   238  	utilized                      bool
   239  	columnDecryptProps            ColumnPathToDecryptionPropsMap
   240  	Verifier                      AADPrefixVerifier
   241  	KeyRetriever                  DecryptionKeyRetriever
   242  }
   243  
   244  // NewFileDecryptionProperties takes in the options for constructing a new FileDecryptionProperties object, otherwise
   245  // it will use the default configuration which will check footer integrity of a plaintext footer for an encrypted file
   246  // for unencrypted parquet files, the decryption properties should not be set.
   247  func NewFileDecryptionProperties(opts ...FileDecryptionOption) *FileDecryptionProperties {
   248  	var cfg fileDecryptConfig
   249  	cfg.checkFooterIntegrity = DefaultCheckSignature
   250  	cfg.plaintextAllowed = DefaultAllowPlaintextFiles
   251  	for _, o := range opts {
   252  		o(&cfg)
   253  	}
   254  	return &FileDecryptionProperties{
   255  		Verifier:                      cfg.verifier,
   256  		footerKey:                     cfg.footerKey,
   257  		checkPlaintextFooterIntegrity: cfg.checkFooterIntegrity,
   258  		KeyRetriever:                  cfg.retriever,
   259  		aadPrefix:                     cfg.aadPrefix,
   260  		columnDecryptProps:            cfg.colDecrypt,
   261  		plaintextAllowed:              cfg.plaintextAllowed,
   262  		utilized:                      false,
   263  	}
   264  }
   265  
   266  // ColumnKey returns the key to be used for decrypting the provided column.
   267  func (fd *FileDecryptionProperties) ColumnKey(path string) string {
   268  	if d, ok := fd.columnDecryptProps[path]; ok {
   269  		if d != nil {
   270  			return d.Key()
   271  		}
   272  	}
   273  	return ""
   274  }
   275  
   276  // FooterKey returns the key utilized for decrypting the Footer if encrypted and any columns that are encrypted with
   277  // the footer key.
   278  func (fd *FileDecryptionProperties) FooterKey() string { return fd.footerKey }
   279  
   280  // AadPrefix returns the prefix to be supplied for constructing the identification strings when decrypting
   281  func (fd *FileDecryptionProperties) AadPrefix() string { return fd.aadPrefix }
   282  
   283  // PlaintextFooterIntegrity returns whether or not an integrity check will be performed on a plaintext footer for an
   284  // encrypted file.
   285  func (fd *FileDecryptionProperties) PlaintextFooterIntegrity() bool {
   286  	return fd.checkPlaintextFooterIntegrity
   287  }
   288  
   289  // PlaintextFilesAllowed returns whether or not this instance of decryption properties are allowed on a plaintext file.
   290  func (fd *FileDecryptionProperties) PlaintextFilesAllowed() bool { return fd.plaintextAllowed }
   291  
   292  // SetUtilized is called to mark this instance as utilized once it is used to read a file. A single instance
   293  // can be used for reading one file only. Setting this ensures the keys will be wiped out upon completion of file reading.
   294  func (fd *FileDecryptionProperties) SetUtilized() { fd.utilized = true }
   295  
   296  // IsUtilized returns whether or not this instance has been used to decrypt a file. If the footer key and prefix are
   297  // empty and there are no column decryption properties, then this is always false.
   298  func (fd *FileDecryptionProperties) IsUtilized() bool {
   299  	if fd.footerKey == "" && len(fd.columnDecryptProps) == 0 && fd.aadPrefix == "" {
   300  		return false
   301  	}
   302  	return fd.utilized
   303  }
   304  
   305  // WipeOutDecryptionKeys will clear all the keys for this instance including the column level ones, this will be called
   306  // after this instance has been utilized.
   307  func (fd *FileDecryptionProperties) WipeOutDecryptionKeys() {
   308  	fd.footerKey = ""
   309  	for _, cd := range fd.columnDecryptProps {
   310  		cd.WipeOutDecryptionKey()
   311  	}
   312  }
   313  
   314  // Clone returns a new instance of these properties, changing the prefix if set (keeping the same prefix if left empty)
   315  func (fd *FileDecryptionProperties) Clone(newAadPrefix string) *FileDecryptionProperties {
   316  	keyCopy := fd.footerKey
   317  	colDecryptMapCopy := make(ColumnPathToDecryptionPropsMap)
   318  	for k, v := range fd.columnDecryptProps {
   319  		colDecryptMapCopy[k] = v.Clone()
   320  	}
   321  	if newAadPrefix == "" {
   322  		newAadPrefix = fd.aadPrefix
   323  	}
   324  	return &FileDecryptionProperties{
   325  		footerKey:                     keyCopy,
   326  		KeyRetriever:                  fd.KeyRetriever,
   327  		checkPlaintextFooterIntegrity: fd.checkPlaintextFooterIntegrity,
   328  		Verifier:                      fd.Verifier,
   329  		columnDecryptProps:            colDecryptMapCopy,
   330  		aadPrefix:                     newAadPrefix,
   331  		plaintextAllowed:              fd.plaintextAllowed,
   332  		utilized:                      false,
   333  	}
   334  }
   335  
   336  type fileDecryptConfig struct {
   337  	footerKey            string
   338  	aadPrefix            string
   339  	verifier             AADPrefixVerifier
   340  	colDecrypt           ColumnPathToDecryptionPropsMap
   341  	retriever            DecryptionKeyRetriever
   342  	checkFooterIntegrity bool
   343  	plaintextAllowed     bool
   344  }
   345  
   346  // FileDecryptionOption is how to supply options to constructing a new FileDecryptionProperties instance.
   347  type FileDecryptionOption func(*fileDecryptConfig)
   348  
   349  // WithFooterKey sets an explicit footer key. If Applied on a file that contains footer key
   350  // metadata the metadata will be ignored, the footer will be decrypted/verified with this key.
   351  //
   352  // If the explicit key is not set, footer key will be fetched from the key retriever.
   353  // With explicit keys or AAD prefix, new encryption properties object must be created for each
   354  // encrypted file.
   355  //
   356  // Explicit encryption keys (footer and column) are cloned.
   357  // Upon completion of file reading, the cloned encryption keys in the properties will be wiped out
   358  // Caller is responsible for wiping out the input key array
   359  // footer key length must be either 16, 24, or 32 bytes
   360  func WithFooterKey(key string) FileDecryptionOption {
   361  	return func(cfg *fileDecryptConfig) {
   362  		if key != "" {
   363  			cfg.footerKey = key
   364  		}
   365  	}
   366  }
   367  
   368  // WithPrefixVerifier supplies a verifier object to use for verifying the AAD Prefixes stored in the file.
   369  func WithPrefixVerifier(verifier AADPrefixVerifier) FileDecryptionOption {
   370  	return func(cfg *fileDecryptConfig) {
   371  		if verifier != nil {
   372  			cfg.verifier = verifier
   373  		}
   374  	}
   375  }
   376  
   377  // WithColumnKeys sets explicit column keys.
   378  //
   379  // It's also possible to set a key retriever on this property object.
   380  //
   381  // Upon file decryption, availability of explicit keys is checked before invocation
   382  // of the retriever callback.
   383  //
   384  // If an explicit key is available for a footer or a column, its key metadata will be ignored.
   385  func WithColumnKeys(decrypt ColumnPathToDecryptionPropsMap) FileDecryptionOption {
   386  	return func(cfg *fileDecryptConfig) {
   387  		if len(decrypt) == 0 {
   388  			return
   389  		}
   390  		if len(cfg.colDecrypt) != 0 {
   391  			panic("column properties already set")
   392  		}
   393  		for _, v := range decrypt {
   394  			if v.IsUtilized() {
   395  				panic("parquet: column properties utilized in another file")
   396  			}
   397  			v.SetUtilized()
   398  		}
   399  		cfg.colDecrypt = decrypt
   400  	}
   401  }
   402  
   403  // WithKeyRetriever sets a key retriever callback. It's also possible to set explicit footer or column keys.
   404  func WithKeyRetriever(retriever DecryptionKeyRetriever) FileDecryptionOption {
   405  	return func(cfg *fileDecryptConfig) {
   406  		if retriever != nil {
   407  			cfg.retriever = retriever
   408  		}
   409  	}
   410  }
   411  
   412  // DisableFooterSignatureVerification skips integrity verification of plaintext footers.
   413  //
   414  // If not called, integrity of plaintext footers will be checked in runtime, and will panic
   415  // if the footer signing key is not available
   416  // or if the footer content and signature don't match
   417  func DisableFooterSignatureVerification() FileDecryptionOption {
   418  	return func(cfg *fileDecryptConfig) {
   419  		cfg.checkFooterIntegrity = false
   420  	}
   421  }
   422  
   423  // WithPlaintextAllowed sets allowing plaintext files.
   424  //
   425  // By default, reading plaintext (unencrypted) files is not allowed when using
   426  // a decryptor.
   427  //
   428  // In order to detect files that were not encrypted by mistake.
   429  // However the default behavior can be overridden by using this method.
   430  func WithPlaintextAllowed() FileDecryptionOption {
   431  	return func(cfg *fileDecryptConfig) {
   432  		cfg.plaintextAllowed = true
   433  	}
   434  }
   435  
   436  // WithDecryptAadPrefix explicitly supplies the file aad prefix.
   437  //
   438  // A must when a prefix is used for file encryption, but not stored in the file.
   439  func WithDecryptAadPrefix(prefix string) FileDecryptionOption {
   440  	return func(cfg *fileDecryptConfig) {
   441  		if prefix != "" {
   442  			cfg.aadPrefix = prefix
   443  		}
   444  	}
   445  }
   446  
   447  // Algorithm describes how something was encrypted, representing the EncryptionAlgorithm object from the
   448  // parquet.thrift file.
   449  type Algorithm struct {
   450  	Algo Cipher
   451  	Aad  struct {
   452  		AadPrefix       []byte
   453  		AadFileUnique   []byte
   454  		SupplyAadPrefix bool
   455  	}
   456  }
   457  
   458  // ToThrift returns an instance to be used for serializing when writing a file.
   459  func (e Algorithm) ToThrift() *format.EncryptionAlgorithm {
   460  	if e.Algo == AesGcm {
   461  		return &format.EncryptionAlgorithm{
   462  			AES_GCM_V1: &format.AesGcmV1{
   463  				AadPrefix:       e.Aad.AadPrefix,
   464  				AadFileUnique:   e.Aad.AadFileUnique,
   465  				SupplyAadPrefix: &e.Aad.SupplyAadPrefix,
   466  			},
   467  		}
   468  	}
   469  	return &format.EncryptionAlgorithm{
   470  		AES_GCM_CTR_V1: &format.AesGcmCtrV1{
   471  			AadPrefix:       e.Aad.AadPrefix,
   472  			AadFileUnique:   e.Aad.AadFileUnique,
   473  			SupplyAadPrefix: &e.Aad.SupplyAadPrefix,
   474  		},
   475  	}
   476  }
   477  
   478  // AlgorithmFromThrift converts the thrift object to the Algorithm struct for easier usage.
   479  func AlgorithmFromThrift(enc *format.EncryptionAlgorithm) (ret Algorithm) {
   480  	if enc.IsSetAES_GCM_V1() {
   481  		ret.Algo = AesGcm
   482  		ret.Aad.AadFileUnique = enc.AES_GCM_V1.AadFileUnique
   483  		ret.Aad.AadPrefix = enc.AES_GCM_V1.AadPrefix
   484  		ret.Aad.SupplyAadPrefix = *enc.AES_GCM_V1.SupplyAadPrefix
   485  		return
   486  	}
   487  	ret.Algo = AesCtr
   488  	ret.Aad.AadFileUnique = enc.AES_GCM_CTR_V1.AadFileUnique
   489  	ret.Aad.AadPrefix = enc.AES_GCM_CTR_V1.AadPrefix
   490  	ret.Aad.SupplyAadPrefix = *enc.AES_GCM_CTR_V1.SupplyAadPrefix
   491  	return
   492  }
   493  
   494  // FileEncryptionProperties describe how to encrypt a parquet file when writing data.
   495  type FileEncryptionProperties struct {
   496  	alg                  Algorithm
   497  	footerKey            string
   498  	footerKeyMetadata    string
   499  	encryptedFooter      bool
   500  	fileAad              string
   501  	utilized             bool
   502  	storeAadPrefixInFile bool
   503  	aadPrefix            string
   504  	encryptedCols        ColumnPathToEncryptionPropsMap
   505  }
   506  
   507  // EncryptedFooter returns if the footer for this file should be encrypted or left in plaintext.
   508  func (fe *FileEncryptionProperties) EncryptedFooter() bool { return fe.encryptedFooter }
   509  
   510  // Algorithm returns the description of how we will perform the encryption, the algorithm, prefixes, and so on.
   511  func (fe *FileEncryptionProperties) Algorithm() Algorithm { return fe.alg }
   512  
   513  // FooterKey returns the actual key used to encrypt the footer if it is encrypted, or to encrypt any columns which
   514  // will be encrypted with it rather than their own keys.
   515  func (fe *FileEncryptionProperties) FooterKey() string { return fe.footerKey }
   516  
   517  // FooterKeyMetadata is used for retrieving a key from the key retriever in order to set the footer key
   518  func (fe *FileEncryptionProperties) FooterKeyMetadata() string { return fe.footerKeyMetadata }
   519  
   520  // FileAad returns the aad identification to be used at the file level which gets concatenated with the row and column
   521  // information for encrypting data.
   522  func (fe *FileEncryptionProperties) FileAad() string { return fe.fileAad }
   523  
   524  // IsUtilized returns whether or not this instance has been used to encrypt a file
   525  func (fe *FileEncryptionProperties) IsUtilized() bool { return fe.utilized }
   526  
   527  // SetUtilized is called after writing a file. A FileEncryptionProperties object can be used for writing one file only,
   528  // the encryption keys will be wiped out upon completion of writing the file.
   529  func (fe *FileEncryptionProperties) SetUtilized() { fe.utilized = true }
   530  
   531  // EncryptedColumns returns the mapping of column paths to column encryption properties
   532  func (fe *FileEncryptionProperties) EncryptedColumns() ColumnPathToEncryptionPropsMap {
   533  	return fe.encryptedCols
   534  }
   535  
   536  // ColumnEncryptionProperties returns the properties for encrypting a given column.
   537  //
   538  // This may be nil for columns that aren't encrypted or may be default properties.
   539  func (fe *FileEncryptionProperties) ColumnEncryptionProperties(path string) *ColumnEncryptionProperties {
   540  	if len(fe.encryptedCols) == 0 {
   541  		return NewColumnEncryptionProperties(path)
   542  	}
   543  	if c, ok := fe.encryptedCols[path]; ok {
   544  		return c
   545  	}
   546  	return nil
   547  }
   548  
   549  // Clone allows returning an identical property setup for another file with the option to update the aadPrefix,
   550  // (if given the empty string, the current aad prefix will be used) since a single instance can only be used
   551  // to encrypt one file before wiping out the keys.
   552  func (fe *FileEncryptionProperties) Clone(newAadPrefix string) *FileEncryptionProperties {
   553  	footerKeyCopy := fe.footerKey
   554  	encryptedColsCopy := make(ColumnPathToEncryptionPropsMap)
   555  	for k, v := range fe.encryptedCols {
   556  		encryptedColsCopy[k] = v.Clone()
   557  	}
   558  	if newAadPrefix == "" {
   559  		newAadPrefix = fe.aadPrefix
   560  	}
   561  
   562  	opts := []EncryptOption{
   563  		WithAlg(fe.alg.Algo), WithFooterKeyMetadata(fe.footerKeyMetadata),
   564  		WithAadPrefix(newAadPrefix), WithEncryptedColumns(encryptedColsCopy),
   565  	}
   566  	if !fe.encryptedFooter {
   567  		opts = append(opts, WithPlaintextFooter())
   568  	}
   569  	if !fe.storeAadPrefixInFile {
   570  		opts = append(opts, DisableAadPrefixStorage())
   571  	}
   572  	return NewFileEncryptionProperties(footerKeyCopy, opts...)
   573  }
   574  
   575  // WipeOutEncryptionKeys clears all of the encryption keys for this and the columns
   576  func (fe *FileEncryptionProperties) WipeOutEncryptionKeys() {
   577  	fe.footerKey = ""
   578  	for _, elem := range fe.encryptedCols {
   579  		elem.WipeOutEncryptionKey()
   580  	}
   581  }
   582  
   583  type configEncrypt struct {
   584  	cipher               Cipher
   585  	encryptFooter        bool
   586  	keyMetadata          string
   587  	aadprefix            string
   588  	storeAadPrefixInFile bool
   589  	encryptedCols        ColumnPathToEncryptionPropsMap
   590  }
   591  
   592  // EncryptOption is used for specifying values when building FileEncryptionProperties
   593  type EncryptOption func(*configEncrypt)
   594  
   595  // WithPlaintextFooter sets the writer to write the footer in plain text, otherwise the footer will be encrypted
   596  // too (which is the default behavior).
   597  func WithPlaintextFooter() EncryptOption {
   598  	return func(cfg *configEncrypt) {
   599  		cfg.encryptFooter = false
   600  	}
   601  }
   602  
   603  // WithAlg sets the encryption algorithm to utilize. (default is AesGcm)
   604  func WithAlg(cipher Cipher) EncryptOption {
   605  	return func(cfg *configEncrypt) {
   606  		cfg.cipher = cipher
   607  	}
   608  }
   609  
   610  // WithFooterKeyID sets a key retrieval metadata to use (converted from string), this must be a utf8 string.
   611  //
   612  // use either WithFooterKeyID or WithFooterKeyMetadata, not both.
   613  func WithFooterKeyID(key string) EncryptOption {
   614  	if !utf8.ValidString(key) {
   615  		panic("parquet: footer key id should be UTF8 encoded")
   616  	}
   617  	return WithFooterKeyMetadata(key)
   618  }
   619  
   620  // WithFooterKeyMetadata sets a key retrieval metadata to use for getting the key.
   621  //
   622  // Use either WithFooterKeyID or WithFooterKeyMetadata, not both.
   623  func WithFooterKeyMetadata(keyMeta string) EncryptOption {
   624  	return func(cfg *configEncrypt) {
   625  		if keyMeta != "" {
   626  			cfg.keyMetadata = keyMeta
   627  		}
   628  	}
   629  }
   630  
   631  // WithAadPrefix sets the AAD prefix to use for encryption and by default will store it in the file
   632  func WithAadPrefix(aadPrefix string) EncryptOption {
   633  	return func(cfg *configEncrypt) {
   634  		if aadPrefix != "" {
   635  			cfg.aadprefix = aadPrefix
   636  			cfg.storeAadPrefixInFile = true
   637  		}
   638  	}
   639  }
   640  
   641  // DisableAadPrefixStorage will set the properties to not store the AadPrefix in the file. If this isn't called
   642  // and the AadPrefix is set, then it will be stored. This needs to in the options *after* WithAadPrefix to have an effect.
   643  func DisableAadPrefixStorage() EncryptOption {
   644  	return func(cfg *configEncrypt) {
   645  		cfg.storeAadPrefixInFile = false
   646  	}
   647  }
   648  
   649  // WithEncryptedColumns sets the map of columns and their properties (keys etc.) If not called, then all columns will
   650  // be encrypted with the footer key. If called, then columns not in the map will be left unencrypted.
   651  func WithEncryptedColumns(encrypted ColumnPathToEncryptionPropsMap) EncryptOption {
   652  	none := func(*configEncrypt) {}
   653  	if len(encrypted) == 0 {
   654  		return none
   655  	}
   656  	return func(cfg *configEncrypt) {
   657  		if len(cfg.encryptedCols) != 0 {
   658  			panic("column properties already set")
   659  		}
   660  		for _, v := range encrypted {
   661  			if v.IsUtilized() {
   662  				panic("column properties utilized in another file")
   663  			}
   664  			v.SetUtilized()
   665  		}
   666  		cfg.encryptedCols = encrypted
   667  	}
   668  }
   669  
   670  // NewFileEncryptionProperties returns a new File Encryption description object using the options provided.
   671  func NewFileEncryptionProperties(footerKey string, opts ...EncryptOption) *FileEncryptionProperties {
   672  	var cfg configEncrypt
   673  	cfg.cipher = DefaultEncryptionAlgorithm
   674  	cfg.encryptFooter = DefaultEncryptedFooter
   675  	for _, o := range opts {
   676  		o(&cfg)
   677  	}
   678  
   679  	props := &FileEncryptionProperties{
   680  		footerKey:            footerKey,
   681  		footerKeyMetadata:    cfg.keyMetadata,
   682  		encryptedFooter:      cfg.encryptFooter,
   683  		aadPrefix:            cfg.aadprefix,
   684  		storeAadPrefixInFile: cfg.storeAadPrefixInFile,
   685  		encryptedCols:        cfg.encryptedCols,
   686  		utilized:             false,
   687  	}
   688  
   689  	aadFileUnique := [AadFileUniqueLength]uint8{}
   690  	_, err := rand.Read(aadFileUnique[:])
   691  	if err != nil {
   692  		panic(err)
   693  	}
   694  
   695  	supplyAadPrefix := false
   696  	if props.aadPrefix == "" {
   697  		props.fileAad = string(aadFileUnique[:])
   698  	} else {
   699  		props.fileAad = props.aadPrefix + string(aadFileUnique[:])
   700  		if !props.storeAadPrefixInFile {
   701  			supplyAadPrefix = true
   702  		}
   703  	}
   704  	props.alg.Algo = cfg.cipher
   705  	props.alg.Aad.AadFileUnique = aadFileUnique[:]
   706  	props.alg.Aad.SupplyAadPrefix = supplyAadPrefix
   707  	if cfg.aadprefix != "" && cfg.storeAadPrefixInFile {
   708  		props.alg.Aad.AadPrefix = []byte(props.aadPrefix)
   709  	}
   710  	return props
   711  }
   712
View as plain text