diff --git a/README.md b/README.md index f9ccf473..ca9eda57 100644 --- a/README.md +++ b/README.md @@ -1678,17 +1678,33 @@ You can chain multiple transformers by providing transformer classes as a comma- 3. Custom transformers run before binary properties are calculated. So inserting or removing fields changes the copybook layout. +### GPG Transparent Decryption (Experimental) + +Cobrix supports transparent decryption of GPG/PGP-encrypted files. You just need to provide the content of ASCII-armored +private key, and the passphrase (if non-empty). + +```scala +val df = spark.read + .format("cobol") + .option("copybook", someCopybook) + .option("gpg_private_key", gpgPrivateKey) + .option("gpg_private_key_passphrase", gpgPrivateKeyPassphrase) + .load("/some/path") +``` + ## Summary of all available options ##### File reading options -| Option (usage example) | Description | -|----------------------------------------|:---------------------------------------------------------------------------------------------------------------| -| .option("data_paths", "/path1,/path2") | Allows loading data from multiple unrelated paths on the same filesystem. | -| .option("file_start_offset", "0") | Specifies the number of bytes to skip at the beginning of each file. | -| .option("file_end_offset", "0") | Specifies the number of bytes to skip at the end of each file. | -| .option("record_start_offset", "0") | Specifies the number of bytes to skip at the beginning of each record before applying copybook fields to data. | -| .option("record_end_offset", "0") | Specifies the number of bytes to skip at the end of each record after applying copybook fields to data. | +| Option (usage example) | Description | +|-----------------------------------------------------|:------------------------------------------------------------------------------------------------------------------------------------------------| +| .option("data_paths", "/path1,/path2") | Allows loading data from multiple unrelated paths on the same filesystem. | +| .option("file_start_offset", "0") | Specifies the number of bytes to skip at the beginning of each file. | +| .option("file_end_offset", "0") | Specifies the number of bytes to skip at the end of each file. | +| .option("record_start_offset", "0") | Specifies the number of bytes to skip at the beginning of each record before applying copybook fields to data. | +| .option("record_end_offset", "0") | Specifies the number of bytes to skip at the end of each record after applying copybook fields to data. | +| .option("gpg_private_key", ascGpgKeyContent) | Specifies the ASCII-armored GPG private key for decrypting data files. | +| .option("gpg_private_key_passphrase", "passphrase") | Specifies the passphrase for the ASCII-armored GPG private key used for decrypting data files. If not specified, empty passphrase will be used. | ##### Copybook parsing options diff --git a/build.sbt b/build.sbt index 5d4f34f1..9a167bbf 100644 --- a/build.sbt +++ b/build.sbt @@ -150,6 +150,8 @@ lazy val assemblySettings = Seq( assembly / assemblyShadeRules:= Seq( // Spark may rely on a different version of ANTLR runtime. Renaming the package helps avoid the binary incompatibility ShadeRule.rename("org.antlr.**" -> "za.co.absa.cobrix.cobol.parser.shaded.org.antlr.@1").inAll, + // Environments might rely on different versions of GPG support + ShadeRule.rename("org.bouncycastle.**" -> "za.co.absa.cobrix.cobol.parser.shaded.org.bouncycastle.@1").inAll, // The SLF4j API and implementation are provided by Spark ShadeRule.zap("org.slf4j.**").inAll ), diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/parser/Copybook.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/parser/Copybook.scala index f45d685a..dc2e1394 100644 --- a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/parser/Copybook.scala +++ b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/parser/Copybook.scala @@ -141,7 +141,7 @@ class Copybook(val ast: CopybookAST) extends Logging with Serializable { def getFieldByPathInGroup(group: Group, path: Array[String]): scala.collection.Seq[Statement] = { if (path.length == 0) { - throw new IllegalStateException(s"'$fieldName' is a GROUP and not a primitive field. Cannot extract it's value.") + throw new IllegalStateException(s"'$fieldName' is a GROUP and not a primitive field. Cannot extract its value.") } else { group.children.flatMap { case g: Group => diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParameters.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParameters.scala index ce73e107..d442ca9e 100644 --- a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParameters.scala +++ b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParameters.scala @@ -38,6 +38,8 @@ import za.co.absa.cobrix.cobol.reader.policies.SchemaRetentionPolicy.SchemaReten * @param asciiCharset A charset for ASCII data * @param fieldCodePage Specifies a mapping between a field name and the code page * @param isUtf16BigEndian If true UTF-16 is considered big-endian. + * @param gpgPrivateKey GPG private key content + * @param gpgPrivateKeyPassphrase GPG private key passphrase * @param floatingPointFormat A format of floating-point numbers * @param recordStartOffset A number of bytes to skip at the beginning of the record before parsing a record according to a copybook * @param recordEndOffset A number of bytes to skip at the end of each record @@ -84,6 +86,8 @@ case class CobolParameters( asciiCharset: Option[String], fieldCodePage: Map[String, String], isUtf16BigEndian: Boolean, + gpgPrivateKey: Option[String], + gpgPrivateKeyPassphrase: Option[String], floatingPointFormat: FloatingPointFormat, recordStartOffset: Int, recordEndOffset: Int, diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParametersParser.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParametersParser.scala index 70535602..1f38ca4c 100644 --- a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParametersParser.scala +++ b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParametersParser.scala @@ -68,6 +68,8 @@ object CobolParametersParser extends Logging { val PARAM_RECORD_TRAILER_NAME2 = "file_trailer_field" val PARAM_IS_XCOM = "is_xcom" val PARAM_IS_TEXT = "is_text" + val PARAM_GPG_PRIVATE_KEY = "gpg_private_key" + val PARAM_GPG_PRIVATE_KEY_PASSPHRASE = "gpg_private_key_passphrase" // Schema transformation parameters val PARAM_GENERATE_RECORD_ID = "generate_record_id" @@ -302,6 +304,9 @@ object CobolParametersParser extends Logging { ) } + val gpgPrivateKey = params.get(PARAM_GPG_PRIVATE_KEY) + val gpgPrivateKeyPassphraseOpt = params.get(PARAM_GPG_PRIVATE_KEY_PASSPHRASE) + val variableSizeOccursPolicy = VariableSizeOccursPolicy(params.getOrElse(PARAM_VARIABLE_SIZE_OCCURS, "false")) val writerParameters = if (isWriter) { @@ -326,6 +331,8 @@ object CobolParametersParser extends Logging { asciiCharset, getFieldCodepageMap(params), params.getOrElse(PARAM_IS_UTF16_BIG_ENDIAN, "true").toBoolean, + gpgPrivateKey, + gpgPrivateKeyPassphraseOpt, getFloatingPointFormat(params), params.getOrElse(PARAM_RECORD_START_OFFSET, "0").toInt, params.getOrElse(PARAM_RECORD_END_OFFSET, "0").toInt, @@ -493,6 +500,8 @@ object CobolParametersParser extends Logging { asciiCharset = parameters.asciiCharset, fieldCodePage = parameters.fieldCodePage, isUtf16BigEndian = parameters.isUtf16BigEndian, + gpgPrivateKey = parameters.gpgPrivateKey, + gpgPrivateKeyPassphrase = parameters.gpgPrivateKeyPassphrase, floatingPointFormat = parameters.floatingPointFormat, redefineRuleExpressions = ruleExpressionMap, variableSizeOccurs = parameters.variableSizeOccurs, diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/ReaderParameters.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/ReaderParameters.scala index 402aa7a6..4b31f01f 100644 --- a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/ReaderParameters.scala +++ b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/ReaderParameters.scala @@ -37,6 +37,8 @@ import za.co.absa.cobrix.cobol.reader.policies.SchemaRetentionPolicy.SchemaReten * @param asciiCharset A charset for ASCII data * @param fieldCodePage Specifies a mapping between a field name and the code page * @param isUtf16BigEndian If true UTF-16 strings are considered big-endian. + * @param gpgPrivateKey GPG private key content + * @param gpgPrivateKeyPassphrase GPG private key passphrase * @param floatingPointFormat A format of floating-point numbers * @param redefineRuleExpressions A map of REDEFINE field names to expressions that determine which redefine alternative to use when parsing records. * @param variableSizeOccurs Specifies how to handle OCCURS DEPENDING ON when the actual number of elements in arrays is less than the maximum array size @@ -95,6 +97,8 @@ case class ReaderParameters( asciiCharset: Option[String] = None, fieldCodePage: Map[String, String] = Map.empty[String, String], isUtf16BigEndian: Boolean = true, + gpgPrivateKey: Option[String] = None, + gpgPrivateKeyPassphrase: Option[String] = None, floatingPointFormat: FloatingPointFormat = FloatingPointFormat.IBM, redefineRuleExpressions: Map[String, ExpressionEvaluator] = Map.empty, variableSizeOccurs: VariableSizeOccursPolicy = VariableSizeOccursPolicy.MaxSize, diff --git a/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/parser/extract/BinaryExtractorSpec.scala b/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/parser/extract/BinaryExtractorSpec.scala index 8438c452..cb456c17 100644 --- a/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/parser/extract/BinaryExtractorSpec.scala +++ b/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/parser/extract/BinaryExtractorSpec.scala @@ -204,7 +204,7 @@ class BinaryExtractorSpec extends AnyFunSuite { val thrown4 = intercept[IllegalStateException] { val resultImpossible4: Any = copybook.getFieldValueByName(notPrimitiveName2, bytes, startOffset) } - assert(thrown4.getMessage === s"'$notPrimitiveName2' is a GROUP and not a primitive field. Cannot extract it's value.") + assert(thrown4.getMessage === s"'$notPrimitiveName2' is a GROUP and not a primitive field. Cannot extract its value.") } test("Test set field value by name") { diff --git a/data/test41_copybook.cob b/data/test41_copybook.cob new file mode 100644 index 00000000..c6dc2273 --- /dev/null +++ b/data/test41_copybook.cob @@ -0,0 +1,37 @@ + **************************************************************************** + * * + * Copyright 2018 ABSA Group Limited * + * * + * Licensed 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. * + * * + **************************************************************************** + + 01 RECORD. + 05 ID PIC S9(4) COMP. + 05 COMPANY. + 10 SHORT-NAME PIC X(10). + 10 COMPANY-ID-NUM PIC 9(5) COMP-3. + 10 COMPANY-ID-STR + REDEFINES COMPANY-ID-NUM PIC X(3). + 05 METADATA. + 10 CLIENTID PIC X(15). + 10 REGISTRATION-NUM PIC X(10). + 10 NUMBER-OF-ACCTS PIC 9(03) COMP-3. + 10 ACCOUNT. + 12 ACCOUNT-DETAIL OCCURS 80 + DEPENDING ON NUMBER-OF-ACCTS. + 15 ACCOUNT-NUMBER PIC X(24). + 15 ACCOUNT-TYPE-N PIC 9(5) COMP-3. + 15 ACCOUNT-TYPE-X REDEFINES + ACCOUNT-TYPE-N PIC X(3). + \ No newline at end of file diff --git a/data/test41_data/example.bin.gpg b/data/test41_data/example.bin.gpg new file mode 100644 index 00000000..ce623c8c Binary files /dev/null and b/data/test41_data/example.bin.gpg differ diff --git a/data/test41_expected/test41a.txt b/data/test41_expected/test41a.txt new file mode 100644 index 00000000..72d7ef0f --- /dev/null +++ b/data/test41_expected/test41a.txt @@ -0,0 +1,10 @@ +{"ID":1,"COMPANY":{"SHORT_NAME":"FOO INCORP","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000000000001100220033","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"ID":2,"COMPANY":{"SHORT_NAME":"BARCOMPANY","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"002000000022004000010001","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"ID":3,"COMPANY":{"SHORT_NAME":"EXAMPLE.CO","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000000000001234567890","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"ID":4,"COMPANY":{"SHORT_NAME":"EXAMPLE330","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":2,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000000000009876543210","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000000000001234555561","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""}]}}} +{"ID":5,"COMPANY":{"SHORT_NAME":"EXAMPLE3","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000012131415161718192","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"ID":6,"COMPANY":{"SHORT_NAME":"EXAMPLE4","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":3,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000000000002000400012","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000000000003000400102","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000005006001200301000","ACCOUNT_TYPE_N":2,"ACCOUNT_TYPE_X":""}]}}} +{"ID":7,"COMPANY":{"SHORT_NAME":"EXAMPLE7","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":2,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000100423412301203120","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000000030928973981723","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""}]}}} +{"ID":8,"COMPANY":{"SHORT_NAME":"FOOBAR8","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":3,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000389871238792010200","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000036719283719283713","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000001992837819827389172","ACCOUNT_TYPE_N":2,"ACCOUNT_TYPE_X":""}]}}} +{"ID":9,"COMPANY":{"SHORT_NAME":"DUMMY_CO9","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000731928300100002312","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"ID":10,"COMPANY":{"SHORT_NAME":"NEWEXCOM10","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":2,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000004909239000000233","ACCOUNT_TYPE_N":2,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000000984120003123900","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""}]}}} diff --git a/data/test41_expected/test41b.txt b/data/test41_expected/test41b.txt new file mode 100644 index 00000000..9b0244b7 --- /dev/null +++ b/data/test41_expected/test41b.txt @@ -0,0 +1,10 @@ +{"File_Id":0,"Record_Id":0,"Record_Byte_Length":2202,"ID":1,"COMPANY":{"SHORT_NAME":"FOO INCORP","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000000000001100220033","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":1,"Record_Byte_Length":2202,"ID":2,"COMPANY":{"SHORT_NAME":"BARCOMPANY","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"002000000022004000010001","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":2,"Record_Byte_Length":2202,"ID":3,"COMPANY":{"SHORT_NAME":"EXAMPLE.CO","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000000000001234567890","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":3,"Record_Byte_Length":2202,"ID":4,"COMPANY":{"SHORT_NAME":"EXAMPLE330","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":2,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000000000009876543210","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000000000001234555561","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":4,"Record_Byte_Length":2202,"ID":5,"COMPANY":{"SHORT_NAME":"EXAMPLE3","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000012131415161718192","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":5,"Record_Byte_Length":2202,"ID":6,"COMPANY":{"SHORT_NAME":"EXAMPLE4","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":3,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000000000002000400012","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000000000003000400102","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000005006001200301000","ACCOUNT_TYPE_N":2,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":6,"Record_Byte_Length":2202,"ID":7,"COMPANY":{"SHORT_NAME":"EXAMPLE7","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":2,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000100423412301203120","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000000030928973981723","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":7,"Record_Byte_Length":2202,"ID":8,"COMPANY":{"SHORT_NAME":"FOOBAR8","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":3,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000389871238792010200","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000036719283719283713","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000001992837819827389172","ACCOUNT_TYPE_N":2,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":8,"Record_Byte_Length":2202,"ID":9,"COMPANY":{"SHORT_NAME":"DUMMY_CO9","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000731928300100002312","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":9,"Record_Byte_Length":2202,"ID":10,"COMPANY":{"SHORT_NAME":"NEWEXCOM10","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":2,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000004909239000000233","ACCOUNT_TYPE_N":2,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000000984120003123900","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""}]}}} diff --git a/data/test41_expected/test41c.txt b/data/test41_expected/test41c.txt new file mode 100644 index 00000000..9b0244b7 --- /dev/null +++ b/data/test41_expected/test41c.txt @@ -0,0 +1,10 @@ +{"File_Id":0,"Record_Id":0,"Record_Byte_Length":2202,"ID":1,"COMPANY":{"SHORT_NAME":"FOO INCORP","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000000000001100220033","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":1,"Record_Byte_Length":2202,"ID":2,"COMPANY":{"SHORT_NAME":"BARCOMPANY","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"002000000022004000010001","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":2,"Record_Byte_Length":2202,"ID":3,"COMPANY":{"SHORT_NAME":"EXAMPLE.CO","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000000000001234567890","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":3,"Record_Byte_Length":2202,"ID":4,"COMPANY":{"SHORT_NAME":"EXAMPLE330","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":2,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000000000009876543210","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000000000001234555561","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":4,"Record_Byte_Length":2202,"ID":5,"COMPANY":{"SHORT_NAME":"EXAMPLE3","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000012131415161718192","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":5,"Record_Byte_Length":2202,"ID":6,"COMPANY":{"SHORT_NAME":"EXAMPLE4","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":3,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000000000002000400012","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000000000003000400102","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000005006001200301000","ACCOUNT_TYPE_N":2,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":6,"Record_Byte_Length":2202,"ID":7,"COMPANY":{"SHORT_NAME":"EXAMPLE7","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":2,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000100423412301203120","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000000030928973981723","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":7,"Record_Byte_Length":2202,"ID":8,"COMPANY":{"SHORT_NAME":"FOOBAR8","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":3,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000389871238792010200","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000036719283719283713","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000001992837819827389172","ACCOUNT_TYPE_N":2,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":8,"Record_Byte_Length":2202,"ID":9,"COMPANY":{"SHORT_NAME":"DUMMY_CO9","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":1,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000731928300100002312","ACCOUNT_TYPE_N":0,"ACCOUNT_TYPE_X":""}]}}} +{"File_Id":0,"Record_Id":9,"Record_Byte_Length":2202,"ID":10,"COMPANY":{"SHORT_NAME":"NEWEXCOM10","COMPANY_ID_NUM":0,"COMPANY_ID_STR":""},"METADATA":{"CLIENTID":"","REGISTRATION_NUM":"","NUMBER_OF_ACCTS":2,"ACCOUNT":{"ACCOUNT_DETAIL":[{"ACCOUNT_NUMBER":"000000004909239000000233","ACCOUNT_TYPE_N":2,"ACCOUNT_TYPE_X":""},{"ACCOUNT_NUMBER":"000000000984120003123900","ACCOUNT_TYPE_N":1,"ACCOUNT_TYPE_X":""}]}}} diff --git a/pom.xml b/pom.xml index 70a9805e..6659f04b 100644 --- a/pom.xml +++ b/pom.xml @@ -111,6 +111,7 @@ 4.11.0 1.11.10 1.7.25 + 1.84 @@ -225,6 +226,13 @@ test + + + org.bouncycastle + bcpg-jdk18on + ${bouncycastle.version} + + org.scalatest diff --git a/project/Dependencies.scala b/project/Dependencies.scala index adb614ef..073ba5b2 100644 --- a/project/Dependencies.scala +++ b/project/Dependencies.scala @@ -21,6 +21,7 @@ object Dependencies { private val antlrValue = "4.9.3" private val slf4jVersion = "1.7.25" private val jacksonVersion = "2.15.4" + private val bouncycastleVersion = "1.84" private val scalatestVersion = "3.2.19" private val mockitoVersion = "4.11.0" @@ -56,6 +57,9 @@ object Dependencies { "org.apache.spark" %% "spark-sql" % sparkVersion(scalaVersion) % Provided, "org.apache.spark" %% "spark-streaming" % sparkVersion(scalaVersion) % Provided, + // libraries + "org.bouncycastle" % "bcpg-jdk18on" % bouncycastleVersion, + // test "org.scalatest" %% "scalatest" % scalatestVersion % Test, "org.mockito" % "mockito-core" % mockitoVersion % Test diff --git a/spark-cobol/pom.xml b/spark-cobol/pom.xml index 2906bf1a..bf2052e4 100644 --- a/spark-cobol/pom.xml +++ b/spark-cobol/pom.xml @@ -49,6 +49,12 @@ ${project.version} + + + org.bouncycastle + bcpg-jdk18on + + org.slf4j slf4j-log4j12 diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/SparkCobolProcessor.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/SparkCobolProcessor.scala index 2a78574e..0f6694fd 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/SparkCobolProcessor.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/SparkCobolProcessor.scala @@ -234,8 +234,8 @@ object SparkCobolProcessor { val numOfBytesMsg = if (numOfBytes > 0) s"${numOfBytes / Constants.megabyte} MB" else "until the end" log.info(s"Going to process offsets ${indexEntry.offsetFrom}...${indexEntry.offsetTo} ($numOfBytesMsg) of $fileName") - val dataStream = new FileStreamer(filePathName, sconf.value, indexEntry.offsetFrom, numOfBytes) - val headerStream = new FileStreamer(filePathName, sconf.value) + val dataStream = new FileStreamer(filePathName, sconf.value, indexEntry.offsetFrom, numOfBytes, readerParameters.gpgPrivateKey, readerParameters.gpgPrivateKeyPassphrase) + val headerStream = new FileStreamer(filePathName, sconf.value, 0L, 0L, readerParameters.gpgPrivateKey, readerParameters.gpgPrivateKeyPassphrase) CobolProcessorBase.getRecordExtractor(readerParameters, copybookContents, dataStream, Some(headerStream)) }) @@ -243,7 +243,7 @@ object SparkCobolProcessor { case _ => spark.sparkContext.parallelize(listOfFiles).flatMap { inputFile => log.info(s"Going to process data from $inputFile") - val ifs = new FileStreamer(inputFile, sconf.value) + val ifs = new FileStreamer(inputFile, sconf.value, 0L, 0L, readerParameters.gpgPrivateKey, readerParameters.gpgPrivateKeyPassphrase) CobolProcessorBase.getRecordExtractor(readerParameters, copybookContents, ifs, None) } @@ -282,16 +282,16 @@ object SparkCobolProcessor { } else 0L - val recordCount = UsingUtils.using(new FileStreamer(inputFile, sconf.value, fileStartOffset, maximumBytes)) { ifs => + val recordCount = UsingUtils.using(new FileStreamer(inputFile, sconf.value, fileStartOffset, maximumBytes, readerParameters.gpgPrivateKey, readerParameters.gpgPrivateKeyPassphrase)) { ifs => UsingUtils.using(new BufferedOutputStream(outputFs.create(outputFile, true))) { ofs => if (fileStartOffset > 0 && retainStartAndEndOffsets) { - val tempStream = new FileStreamer(inputFile, sconf.value) + val tempStream = new FileStreamer(inputFile, sconf.value, 0L, 0L, readerParameters.gpgPrivateKey, readerParameters.gpgPrivateKeyPassphrase) ofs.write(tempStream.next(fileStartOffset)) tempStream.close() } val recordsProcessed = cobolProcessor.process(ifs, ofs)(rawRecordProcessor) if (fileEndOffset > 0 && retainStartAndEndOffsets) { - val tempStream = new FileStreamer(inputFile, sconf.value, maximumBytes + fileStartOffset) + val tempStream = new FileStreamer(inputFile, sconf.value, maximumBytes + fileStartOffset, 0L, readerParameters.gpgPrivateKey, readerParameters.gpgPrivateKeyPassphrase) ofs.write(tempStream.next(fileEndOffset)) tempStream.close() } diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/DefaultSource.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/DefaultSource.scala index 80ff26d6..7a741b05 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/DefaultSource.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/DefaultSource.scala @@ -64,7 +64,8 @@ class DefaultSource val filesList = CobolRelation.getListFilesWithOrder(cobolParameters.sourcePaths, sqlContext, isRecursiveRetrieval(sqlContext)) - val hasCompressedFiles = filesList.exists(_.isCompressed) + val hasGpg = cobolParameters.gpgPrivateKey.isDefined + val hasCompressedFiles = hasGpg || filesList.exists(_.isCompressed) if (hasCompressedFiles) { logger.info(s"Compressed files found. Binary parallelism and indexes will be adjusted accordingly.") diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/index/IndexBuilder.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/index/IndexBuilder.scala index e92fcb74..8e724fbf 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/index/IndexBuilder.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/index/IndexBuilder.scala @@ -205,7 +205,7 @@ private[cobol] object IndexBuilder extends Logging { logger.info(s"Going to generate index for the file: $filePath") - val (inputStream, headerStream, maximumBytes) = getStreams(filePath, startOffset, endOffset, config) + val (inputStream, headerStream, maximumBytes) = getStreams(filePath, startOffset, endOffset, reader.getReaderProperties.gpgPrivateKey, reader.getReaderProperties.gpgPrivateKeyPassphrase, config) val index = try { reader.generateIndex(inputStream, headerStream, fileOrder, reader.isRdwBigEndian) } finally { @@ -225,6 +225,8 @@ private[cobol] object IndexBuilder extends Logging { private[cobol] def getStreams(filePath: String, fileStartOffset: Long, fileEndOffset: Long, + gpgPrivateKey: Option[String], + gpgPrivateKeyPassphrase: Option[String], config: Configuration): (SimpleStream, SimpleStream, Long) = { val path = new Path(filePath) val fileSystem = path.getFileSystem(config) @@ -236,8 +238,8 @@ private[cobol] object IndexBuilder extends Logging { 0L } - val inputStream = new FileStreamer(filePath, config, startOffset, maximumBytes) - val headerStream = new FileStreamer(filePath, config) + val inputStream = new FileStreamer(filePath, config, startOffset, maximumBytes, gpgPrivateKey, gpgPrivateKeyPassphrase) + val headerStream = new FileStreamer(filePath, config, 0L, 0L, gpgPrivateKey, gpgPrivateKeyPassphrase) (inputStream, headerStream, maximumBytes) } @@ -252,7 +254,7 @@ private[cobol] object IndexBuilder extends Logging { val endOffset = readerProperties.fileEndOffset readerProperties.recordExtractor.foreach { recordExtractorClass => - val (dataStream, headerStream, _) = getStreams(filePath, startOffset, endOffset, config) + val (dataStream, headerStream, _) = getStreams(filePath, startOffset, endOffset, readerProperties.gpgPrivateKey, readerProperties.gpgPrivateKeyPassphrase, config) try { val extractorOpt = reader.asInstanceOf[ReaderVarLenNestedReader[_]].recordExtractor(0, dataStream, headerStream) @@ -274,7 +276,7 @@ private[cobol] object IndexBuilder extends Logging { headerStream.close() // Getting new streams and record extractor that points directly to the second record - val (dataStream2, headerStream2, _) = getStreams(filePath, offset, endOffset, config) + val (dataStream2, headerStream2, _) = getStreams(filePath, offset, endOffset, readerProperties.gpgPrivateKey, readerProperties.gpgPrivateKeyPassphrase, config) try { val extractorOpt2 = reader.asInstanceOf[ReaderVarLenNestedReader[_]].recordExtractor(1, dataStream2, headerStream2) diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/scanners/CobolScanners.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/scanners/CobolScanners.scala index 8a589334..33522e56 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/scanners/CobolScanners.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/scanners/CobolScanners.scala @@ -48,8 +48,8 @@ private[source] object CobolScanners extends Logging { val numOfBytesMsg = if (numOfBytes > 0) s"${numOfBytes / Constants.megabyte} MB" else "until the end" logger.info(s"Going to process offsets ${indexEntry.offsetFrom}...${indexEntry.offsetTo} ($numOfBytesMsg) of $fileName") - val dataStream = new FileStreamer(filePathName, sconf.value, indexEntry.offsetFrom, numOfBytes) - val headerStream = new FileStreamer(filePathName, sconf.value) + val dataStream = new FileStreamer(filePathName, sconf.value, indexEntry.offsetFrom, numOfBytes, reader.getReaderProperties.gpgPrivateKey, reader.getReaderProperties.gpgPrivateKeyPassphrase) + val headerStream = new FileStreamer(filePathName, sconf.value, 0L, 0L, reader.getReaderProperties.gpgPrivateKey, reader.getReaderProperties.gpgPrivateKeyPassphrase) reader.getRowIterator(dataStream, headerStream, indexEntry.offsetFrom, indexEntry.fileId, indexEntry.recordIndex) }) } @@ -80,8 +80,8 @@ private[source] object CobolScanners extends Logging { fileSize - reader.getReaderProperties.fileEndOffset - startFileOffset } - val dataStream = new FileStreamer(filePath, sconf.value, startFileOffset, maximumFileBytes) - val headerStream = new FileStreamer(filePath, sconf.value, startFileOffset) + val dataStream = new FileStreamer(filePath, sconf.value, startFileOffset, maximumFileBytes, reader.getReaderProperties.gpgPrivateKey, reader.getReaderProperties.gpgPrivateKeyPassphrase) + val headerStream = new FileStreamer(filePath, sconf.value, startFileOffset, 0L, reader.getReaderProperties.gpgPrivateKey, reader.getReaderProperties.gpgPrivateKeyPassphrase) reader.getRowIterator(dataStream, headerStream, startFileOffset, fileOrder, 0L) }) }) diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/streaming/BufferedFSDataInputStream.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/streaming/BufferedFSDataInputStream.scala index 7aa39cbb..8aca4401 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/streaming/BufferedFSDataInputStream.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/streaming/BufferedFSDataInputStream.scala @@ -18,13 +18,22 @@ package za.co.absa.cobrix.spark.cobol.source.streaming import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.{FSDataInputStream, Path} -import za.co.absa.cobrix.spark.cobol.utils.FileUtils - -import java.io.{IOException, InputStream} - -class BufferedFSDataInputStream(filePath: Path, hadoopConfig: Configuration, startOffset: Long, bufferSizeInMegabytes: Int, maximumBytes: Long ) { +import za.co.absa.cobrix.spark.cobol.utils.{FileUtils, GpgUtils} + +import java.io.InputStream +import scala.util.Try +import scala.util.control.NonFatal + +class BufferedFSDataInputStream(filePath: Path, + hadoopConfig: Configuration, + startOffset: Long, + bufferSizeInMegabytes: Int, + maximumBytes: Long, + gpgKeyringAsc: Option[String], + gpgPassphrase: Option[String]) extends AutoCloseable { val bytesInMegabyte: Int = 1048576 private var isCompressedStream = false + private var rawStream: FSDataInputStream = _ // This is the base stream for GPG-encrypted files. Used for closing only, never for actual read. if (bufferSizeInMegabytes <=0 || bufferSizeInMegabytes > 1000) { throw new IllegalArgumentException(s"Invalid buffer size $bufferSizeInMegabytes MB.") @@ -40,11 +49,20 @@ class BufferedFSDataInputStream(filePath: Path, hadoopConfig: Configuration, sta private var bufferContainBytes = 0 private var bytesRead = 0L - @throws[IOException] - def close(): Unit = { + override def close(): Unit = { if (!isStreamClosed) { - in.close() + Try { + if (in != null) { + in.close() + } + } + Try { + if (rawStream != null) { + rawStream.close() + } + } in = null + rawStream = null isStreamClosed = true } } @@ -121,27 +139,55 @@ class BufferedFSDataInputStream(filePath: Path, hadoopConfig: Configuration, sta private def openStream(): InputStream = { val fileSystem = filePath.getFileSystem(hadoopConfig) - val codec = FileUtils.getCompressionCodec(filePath, hadoopConfig) - val fsIn: FSDataInputStream = fileSystem.open(filePath) - val baseStream = if (codec != null) { - isCompressedStream = true - codec.createInputStream(fsIn) - } else { - // No compression detected - fsIn + val baseStream = gpgKeyringAsc match { + case Some(keyring) => + isCompressedStream = true + rawStream = fileSystem.open(filePath) + try { + GpgUtils.decryptStream(rawStream, keyring, gpgPassphrase.map(_.toCharArray).getOrElse(Array.emptyCharArray)) + } catch { + case ex: Throwable => + // Close rawStream only if decryptStream() fails to return a decrypted stream. Ignore errors that might happen on close. + Try { + if (rawStream != null) { + rawStream.close() + } + } + rawStream = null + throw ex + } + case None => + val codec = FileUtils.getCompressionCodec(filePath, hadoopConfig) + val fsIn: FSDataInputStream = fileSystem.open(filePath) + + if (codec != null) { + isCompressedStream = true + codec.createInputStream(fsIn) + } else { + // No compression detected + fsIn + } } if (startOffset > 0) { - if (codec == null) { - fsIn.seek(startOffset) - } else { - var toSkip = startOffset - while (toSkip > 0) { - val skipped = baseStream.skip(toSkip) - if (skipped <= 0) return baseStream - toSkip -= skipped + try { + if (!isCompressedStream) { + baseStream.asInstanceOf[FSDataInputStream].seek(startOffset) + } else { + var toSkip = startOffset + while (toSkip > 0) { + val skipped = baseStream.skip(toSkip) + if (skipped <= 0) return baseStream + toSkip -= skipped + } } + } catch { + case NonFatal(ex) => + Try { + baseStream.close() + } + throw ex } } baseStream diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/streaming/FileStreamer.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/streaming/FileStreamer.scala index 7caa2593..ce942269 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/streaming/FileStreamer.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/streaming/FileStreamer.scala @@ -17,8 +17,8 @@ package za.co.absa.cobrix.spark.cobol.source.streaming import org.apache.hadoop.conf.Configuration -import org.slf4j.{Logger, LoggerFactory} import org.apache.hadoop.fs.{ContentSummary, Path} +import org.slf4j.{Logger, LoggerFactory} import za.co.absa.cobrix.cobol.reader.common.Constants import za.co.absa.cobrix.cobol.reader.stream.SimpleStream import za.co.absa.cobrix.spark.cobol.utils.FileUtils @@ -37,7 +37,12 @@ import java.io.IOException * @param hadoopConfig Hadoop configuration. * @note This class is not thread-safe and should only be accessed from a single thread */ -class FileStreamer(filePath: String, hadoopConfig: Configuration, startOffset: Long = 0L, maximumBytes: Long = 0L) extends SimpleStream { +class FileStreamer(filePath: String, + hadoopConfig: Configuration, + startOffset: Long = 0L, + maximumBytes: Long = 0L, + gpgKeyringAsc: Option[String] = None, + gpgPassphrase: Option[String] = None) extends SimpleStream { private val logger: Logger = LoggerFactory.getLogger(this.getClass) private val hadoopPath = new Path(filePath) @@ -50,7 +55,7 @@ class FileStreamer(filePath: String, hadoopConfig: Configuration, startOffset: L private var wasOpened = false private var bufferedStream: BufferedFSDataInputStream = _ - private lazy val isCompressedStream = FileUtils.isCompressed(hadoopPath, hadoopConfig) + private lazy val isCompressedStream = gpgKeyringAsc.isDefined || FileUtils.isCompressed(hadoopPath, hadoopConfig) private lazy val fileSize = getHadoopFileSize(hadoopPath) @@ -134,13 +139,13 @@ class FileStreamer(filePath: String, hadoopConfig: Configuration, startOffset: L } override def copyStream(): SimpleStream = { - new FileStreamer(filePath, hadoopConfig, startOffset, maximumBytes) + new FileStreamer(filePath, hadoopConfig, startOffset, maximumBytes, gpgKeyringAsc, gpgPassphrase) } @throws[IOException] private def ensureOpened(): Unit = { if (!wasOpened) { - bufferedStream = new BufferedFSDataInputStream(new Path(filePath), hadoopConfig, startOffset, Constants.defaultStreamBufferInMB, maximumBytes) + bufferedStream = new BufferedFSDataInputStream(new Path(filePath), hadoopConfig, startOffset, Constants.defaultStreamBufferInMB, maximumBytes, gpgKeyringAsc, gpgPassphrase) wasOpened = true } } diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/GpgUtils.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/GpgUtils.scala new file mode 100644 index 00000000..1128474f --- /dev/null +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/GpgUtils.scala @@ -0,0 +1,192 @@ +/* + * Copyright 2018 ABSA Group Limited + * + * Licensed 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 za.co.absa.cobrix.spark.cobol.utils + +import org.bouncycastle.jce.provider.BouncyCastleProvider +import org.bouncycastle.openpgp._ +import org.bouncycastle.openpgp.jcajce.JcaPGPObjectFactory +import org.bouncycastle.openpgp.operator.jcajce.{JcaKeyFingerprintCalculator, JcePBESecretKeyDecryptorBuilder, JcePublicKeyDataDecryptorFactoryBuilder} + +import java.io.{ByteArrayInputStream, InputStream} +import java.nio.charset.StandardCharsets +import java.security.{Provider, Security} +import scala.collection.JavaConverters._ + +object GpgUtils { + /** + * Decrypts a PGP encrypted stream using a secret key from the provided ASCII armored keychain. + * + * The method looks up the public key encrypted session keys contained in the input stream and matches them + * against the secret keys available in the keychain. The first matching secret key is unlocked with the given + * passphrase and used to decrypt the data. Compressed messages are transparently decompressed so that the + * returned stream exposes the original, unencrypted content of the PGP literal data packet. + * + * The returned stream is lazily read from the input stream, so the input stream should stay open until the + * decrypted data is fully consumed, and the returned stream should be closed by the caller. + * + * @param in An input stream containing PGP encrypted data, either binary or ASCII armored. + * @param keychainAscii An ASCII armored keychain containing the secret key that can decrypt the data. + * @param passphrase The passphrase protecting the secret key. + * @return An input stream with the decrypted contents of the message. + * @throws IllegalArgumentException if the input stream does not contain PGP encrypted data, if no secret key in + * the keychain matches the encrypted data, or if the decrypted message does not + * contain literal data. + */ + def decryptStream(in: InputStream, + keychainAscii: String, + passphrase: Array[Char]): InputStream = { + val secretKeyRings = readSecretKeyRings(keychainAscii) + + val encryptedDataList = findEncryptedDataList(PGPUtil.getDecoderStream(in)) + + val publicKeyEncryptedData = encryptedDataList + .getEncryptedDataObjects + .asScala + .collect { case data: PGPPublicKeyEncryptedData => data } + .toSeq + + val (encryptedData, privateKey) = publicKeyEncryptedData + .flatMap(data => findPrivateKey(secretKeyRings, data.getKeyIdentifier.getKeyId, passphrase).map(key => (data, key))) + .headOption + .getOrElse(throw new IllegalArgumentException("No secret key in the provided keychain matches the encrypted data.")) + + val decryptorFactory = new JcePublicKeyDataDecryptorFactoryBuilder() + .setProvider(bcProvider) + .build(privateKey) + + getLiteralDataStream(encryptedData.getDataStream(decryptorFactory)) + } + + /** + * The BouncyCastle security provider used for all PGP cryptographic operations. + * + * The provider instance is passed explicitly to the JCE builders so that the code does not depend on the "BC" + * provider alias being present in the JVM security provider list. This is required when the BouncyCastle classes + * are shaded/relocated. + * + * A provider already registered in the JVM under the BouncyCastle provider name is reused only when it is an + * instance of the same, possibly relocated, `BouncyCastleProvider` class used by this code. Otherwise, for + * instance, when a different (non-relocated) BouncyCastle copy occupies that name, a private instance is created + * and used directly, without being registered in the JVM security provider list. + */ + private lazy val bcProvider: Provider = { + Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) match { + case provider: BouncyCastleProvider => provider + case _ => new BouncyCastleProvider + } + } + + /** + * Parses an ASCII armored PGP keychain into a collection of secret key rings. + * + * The keychain text is decoded from its UTF-8 representation and passed through a PGP decoder stream, so that both + * ASCII armored and plain binary keychain contents are accepted. All secret key rings found in the keychain are + * loaded eagerly, therefore the stream used for reading is closed before the collection is returned. + * + * @param keychainAscii An ASCII armored keychain containing one or more secret key rings. + * @return A collection of all secret key rings contained in the keychain. + * @throws java.io.IOException if the keychain cannot be read or is malformed. + * @throws org.bouncycastle.openpgp.PGPException if the keychain does not contain valid secret key ring data. + */ + private def readSecretKeyRings(keychainAscii: String): PGPSecretKeyRingCollection = { + val keyIn = PGPUtil.getDecoderStream(new ByteArrayInputStream(keychainAscii.getBytes(StandardCharsets.UTF_8))) + try { + new PGPSecretKeyRingCollection(keyIn, new JcaKeyFingerprintCalculator) + } finally { + keyIn.close() + } + } + + /** + * Locates the list of public key encrypted session keys in a PGP message stream. + * + * The objects of the PGP message are traversed in order until an encrypted data list is encountered, so that leading + * packets, such as marker packets, are skipped. The stream may contain either binary or already de-armored PGP data. + * + * The returned data list is read lazily from the given stream, therefore the stream must remain open for as long as + * the encrypted data is being processed. + * + * @param inputStream A stream positioned at the beginning of a PGP message. + * @return The encrypted data list holding the encrypted session keys and the encrypted payload of the message. + * @throws IllegalArgumentException if the stream does not contain PGP encrypted data. + */ + private def findEncryptedDataList(inputStream: InputStream): PGPEncryptedDataList = { + val factory = new JcaPGPObjectFactory(inputStream) + + Iterator.continually(factory.nextObject()) + .takeWhile(_ != null) + .collectFirst { case dataList: PGPEncryptedDataList => dataList } + .getOrElse(throw new IllegalArgumentException("The input stream does not contain PGP encrypted data.")) + } + + /** + * Extracts the literal data stream from a decrypted PGP message stream. + * + * The objects of the PGP message are traversed until a literal data packet is found. Compressed data packets are + * transparently unwrapped, so that nested compressed content is also inspected, while any other packet types, such + * as signature or marker packets, are skipped. + * + * The returned stream is read lazily from the given stream, therefore the given stream must remain open until the + * literal data is fully consumed. + * + * @param clearStream A stream containing the decrypted, but still PGP structured, message data. + * @return An input stream exposing the contents of the literal data packet of the message. + * @throws IllegalArgumentException if the message does not contain a literal data packet. + */ + private def getLiteralDataStream(clearStream: InputStream): InputStream = { + var factory = new JcaPGPObjectFactory(clearStream) + var message = factory.nextObject() + var literalStream: Option[InputStream] = None + + while (message != null && literalStream.isEmpty) { + message match { + case compressedData: PGPCompressedData => + factory = new JcaPGPObjectFactory(compressedData.getDataStream) + message = factory.nextObject() + case literalData: PGPLiteralData => + literalStream = Option(literalData.getInputStream) + case _ => + message = factory.nextObject() + } + } + + literalStream.getOrElse(throw new IllegalArgumentException("The decrypted PGP message does not contain literal data.")) + } + + + /** + * Retrieves and unlocks the private key with the given key identifier from a collection of secret key rings. + * + * The secret key rings are searched for a secret key matching the requested key identifier. If such a key exists, + * it is decrypted with the given passphrase using the configured security provider, yielding the usable private key. + * If no secret key with the given identifier is present in the collection, no attempt to decrypt anything is made. + * + * @param secretKeyRings A collection of secret key rings to search for the requested key. + * @param keyId The identifier of the secret key to look for. + * @param passphrase The passphrase protecting the secret key. + * @return The unlocked private key, or `None` if the collection does not contain a key with the given identifier. + * @throws org.bouncycastle.openpgp.PGPException if the secret key cannot be unlocked with the given passphrase. + */ + private def findPrivateKey(secretKeyRings: PGPSecretKeyRingCollection, + keyId: Long, + passphrase: Array[Char]): Option[PGPPrivateKey] = { + Option(secretKeyRings.getSecretKey(keyId)).map { secretKey => + val decryptor = new JcePBESecretKeyDecryptorBuilder().setProvider(bcProvider).build(passphrase) + secretKey.extractPrivateKey(decryptor) + } + } +} diff --git a/spark-cobol/src/test/resources/test/test_gpg_file.gpg b/spark-cobol/src/test/resources/test/test_gpg_file.gpg new file mode 100644 index 00000000..656801a6 Binary files /dev/null and b/spark-cobol/src/test/resources/test/test_gpg_file.gpg differ diff --git a/spark-cobol/src/test/resources/test/test_gpg_key.asc b/spark-cobol/src/test/resources/test/test_gpg_key.asc new file mode 100644 index 00000000..7829468d --- /dev/null +++ b/spark-cobol/src/test/resources/test/test_gpg_key.asc @@ -0,0 +1,74 @@ +# +# Copyright 2018 ABSA Group Limited +# +# Licensed 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. +# +# THIS IS A NON-SECRET GPG KEY USED IN TESTS ONLY. +# +-----BEGIN PGP PRIVATE KEY BLOCK----- + +lQOYBGpwniQBCADWilRsezNYmfLhZrH++APYQO0CYI3AWMqls3Wu4X1/Mm1oF1RT +RHUNJN/3x8LlQndSH5F0MQSNpeS8O0DVrL1QpmsOtCsVS/zvpG2gtx5dfzvUSkQs +9q9NvzaLuYsFwS9B498cZarTc3B0nma9fVY92dSGJzWYNLmnjWuNBVyVUqnjF9LX +om9bfMEr3if+2lNsThN0ZJwhYtlBqsonhRNrYWLynk9uWfxk+aVdbkwi9Bsf0tg1 +HHxCS6XBGcAi8IEdmMhZRWM6qV4lKH4O4xlVJ+TpbN3AmhMbZnlEROZW9oxQAuqH +SyQXqnqqSm5oW09EIu9Z4+L0yxWnehDZazMBABEBAAEAB/oDLfsxTVYy/hs71c+Z +z1SY4cmUdMx5hixc0kwOWTLYv91IFQNOwY6evPwHd0jIQKbpqQB1Em9YY0Xkj/gQ +iaUXtAN1dRhHzkm8pk8qUff0kbDIAzCzchWSEl1GSuPygCbzcf9s54UlmZgIN683 +CAOfTpRHL6dcRNzm+LwsL6UEHNHLYq7nThaklBAJH0CLI3ZFWt4uj74IO2HfwfVQ +/wA6OU2IsHXdcNaLpzE87TmGt0JBsVzK60/6UMK9Hr2yR9CVSsnQNuyjVMabTARo +J84a5leAInmLcSLlsLVCv71bBcIVhEk6vWlugrsXYxPz4E6vWwF3vvkN+3RJwafm +7SRBBADaf/uCen+HqvKFu1k2JdTnq8lW02D4Gq472UztRXa6TjacSzZcSlSwhlvF +IrltXGjAPY846WAvtJjuC2g3qbJnCg7P5VeDpJga1b8eBtJVy8Mm6oHJZVErM7Bm +iar9gtIOtbBh+qJbPPiyjFWKetT4fEO1ZVX6g5P0rp1z4+qQQQQA+1xhDIDE6Isc +80TKXfFAUZJIrOqAw6HIWaapYSJ/GFZMmMVdBa+yp+SqWY2g0IrQGnWDG7PdkX7h +NdDe6RoDrgtJd9URjA0SxjRkcxcgpzcgM0jLBG7OieptG3uyDTr8P49DHrV3aEpZ +hsE3HcSLxXB2VCiyvo0kho7ZZnt38sEEAKmx7av+ydKKxeVw3Q1c11c+y5s8LiUI +wGPURNIEJbP/EAh0+614Bkikswh1HWQGq7fvD+lMn3XYUuDTeYoph5pbzzvYf9X0 +z3sp/oB0ExflJveQXgdClYS5XMaaJG2k5H90Wo9craDdZeDGeEnPgoyrSvVDdTrT +gTduiiCnTobuQda0IENvYnJpeCBUZXN0IDxjb2JyaXhAbm9yZXBseS5jb20+iQFO +BBMBCAA4FiEEflxqvitL4wc9QmK+EcrHhTNTWkIFAmpwniQCGwMFCwkIBwIGFQoJ +CAsCBBYCAwECHgECF4AACgkQEcrHhTNTWkL7TAgAxoFujIMVvzi9ksUoS4tXvtp/ +wMA3MH7QkVftYk47NshU4qL3ZkERJhzXSd5pqdDi2bCiSaVsAwcC1DmBx6G6Z0FL +6+dcPRckGEYyZogKzg0m12k7Rur+VwULDmFy38XDR5zaijNDV8B2v/7BX55tsHqJ +NdYgaOVcEwWdiNf5LF0xwjn5J5tfglHQi6OntAVwh1rLijVDw/pG4yuYgoJexyH+ +AKk+i6jALfQRSYfgXTOT15MwMIzkrwxOAi/o5RHUrQkjPMilXeDdN01IRulD2u6o +m2H8+kkC4qpDQPrsNGqCQ1Ac1Wr2p2xOLYOmkAWdPmbmX/r8l/PyXRr+ohA4RZ0D +mARqcJ4kAQgAxaURPJNeWlOugi9E/yIuiRKKQHIpM813IWEpWC8gNTHiN8zYVRMD +8sdUnsUfzaFw0BzHomSpzOz8c0WaUa1kA7YHIhxoxzSMZY0ChscYbCWzFqXFiBV4 +zgSzDoHNhmzF9KYj5geCjQoHw9G2JK5pNrmmmhUWBir83P8aMW+fRDsSvqhCGTxX +1ohUkIPrlZvem4LSPfRc9yuzZku33DCYmN1gs2Z1HnfVrgRNMJZfes2IaOrQ2VDd +VtSWe5KwpA+sSG1qgOgFgWco/8sc+hkBKVPNw72Z3VpSsX3IGP+XXJT0/wIXWcR9 +PPaClObc5m4ZbQIUugrojCjbHioqgslMWQARAQABAAf9EG6ko/lppQ9jiSHWdfLY +1R8kPO7w44Rv5OfP2OvUHPAYi4SdsNcLoz302F1xwzYqq7bU1zjQ0D+czpWpGqTU +laVm6uxbPsKs8I18egmoC2fH/7hJF8MXP+OuomRi2swE2k4M2oVh0omUszBmmR6i +E0F0q0dZBSrCSrpv6k/KOVEdllr1JGMyGG9QMfnKDavTFzAQSeMZKoABEso+BkYD +nhdxQCfYgfE1ZrL3wQyIUmp6jxX24WALCZPMzG0mhlUhj6f9qvOY4ZjD/HRnYj4z +nQ3n+TORoPbLZqK0RKxkqk2b7gDdJsyiPz49+VODw9BUAjCiryowyBLlhbiv7ZiG +YQQA2iQTiMsNmfixjfv8YM8fujfqDkcDxjDvz0MWmi5KFVur41CKPTCaW2r4a9oF +y2KK4zwJQKermRKlYZS1ELB8eXX2VDUKWmg3neqTYIyIhA4t5rt27XE0k6uEepGz +RhvRLQ+HEMi6HRU6G73UcFHG8/XTxAz2ouPRHpn6mFIRwJEEAOfyWvVDaFUKcS4Y +HlhgO0r/OLWXDWMflo27tkgXuniRoZNkqiBsi5rZYB+XFTHd8KMa4qtAh/cILsQh +oeZPR2tmY7b7Vf0TNkIMfulZTIK7ZTAHCm/uEk0sqNw78O5JCF2YaCSRx2MV8QlI +ETXFYcWtdprTNkRaZqAlCO6a1LNJA/906qRgpy8AG6QYhOZZRxeDtEOX21rL2TXZ +yqRU2frwKUPPnnmCnLauRUS4uWiEM3TjhqNULyybYG/WA40GSE9QAN6e1nfNC57E +CrUaN0ymllfQgjVO4Z1OqBZWb7WEgUw9BIJjSPtQTW5GCeOe+77zcjbTEnwCQQ/L +OEbn9z6z+zwTiQE2BBgBCAAgFiEEflxqvitL4wc9QmK+EcrHhTNTWkIFAmpwniQC +GwwACgkQEcrHhTNTWkKlbQgA1oQvptaovdm8guKG99nC5bFIMtwb2tOo2CupEdYf +DPX+d7tOLaTWUv2Gj7bUOua3Ql+bznUoVEzZkRE2RXmOteV3Y32USXkKGM/xgAHz +ezeHvi00FsqDXdN1lb0QoOZZBH9dfTz3VIz2XR8N14rdNbKq5i0o43G6R0zLVYAp +ioeNS7TZ7tzIGwlp68zqMlpXM8efu9MEOBTAXrp7uBec1897ogrsN/Nhy6Uqeb4g +sIWlEz48Y1Bp1O42S2gZlqWD3veHpOjA2+CEIrNldLiH4hyklnr8YGjVdqUfQAoB +U5rYTB2dK1GMPRRieDC7EGXMXcEcc7J3myuDyJBYdj60Mw== +=aLlG +-----END PGP PRIVATE KEY BLOCK----- diff --git a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/integration/Test45GpgEncryptedFilesSpec.scala b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/integration/Test45GpgEncryptedFilesSpec.scala new file mode 100644 index 00000000..16820bbd --- /dev/null +++ b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/integration/Test45GpgEncryptedFilesSpec.scala @@ -0,0 +1,100 @@ +/* + * Copyright 2018 ABSA Group Limited + * + * Licensed 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 za.co.absa.cobrix.spark.cobol.source.integration + +import org.scalatest.wordspec.AnyWordSpec +import org.slf4j.{Logger, LoggerFactory} +import za.co.absa.cobrix.spark.cobol.source.base.{SimpleComparisonBase, SparkTestBase} +import za.co.absa.cobrix.spark.cobol.source.fixtures.BinaryFileFixture +import za.co.absa.cobrix.spark.cobol.utils.{FileUtils, ResourceUtils} + +import java.nio.charset.StandardCharsets +import java.nio.file.{Files, Paths} + +class Test45GpgEncryptedFilesSpec extends AnyWordSpec with SparkTestBase with BinaryFileFixture with SimpleComparisonBase { + private implicit val logger: Logger = LoggerFactory.getLogger(this.getClass) + + private val exampleName = "Test41" + private val inputCopybookPath = "file://../data/test41_copybook.cob" + private val inputDataPath = "../data/test41_data" + private val expectedResultsAPath = "../data/test41_expected/test41a.txt" + private val actualResultsAPath = "../data/test41_expected/test41a_actual.txt" + private val expectedResultsBPath = "../data/test41_expected/test41b.txt" + private val actualResultsBPath = "../data/test41_expected/test41b_actual.txt" + private val expectedResultsCPath = "../data/test41_expected/test41c.txt" + private val actualResultsCPath = "../data/test41_expected/test41c_actual.txt" + + "gpg encrypted files" should { + "load normally a fixed-record-length file" in { + val gpgPrivateKey = ResourceUtils.readResourceAsString("/test/test_gpg_key.asc") + val df = spark.read + .format("cobol") + .option("copybook", inputCopybookPath) + .option("gpg_private_key", gpgPrivateKey) + .option("pedantic", "true") + .load(inputDataPath) + + val actual = df.toJSON.take(60) + val expected = Files.readAllLines(Paths.get(expectedResultsAPath), StandardCharsets.ISO_8859_1).toArray + + if (!actual.sameElements(expected)) { + FileUtils.writeStringsToFile(actual, actualResultsAPath) + assert(false, s"The actual data doesn't match what is expected for $exampleName example. Please compare contents of $expectedResultsAPath to $actualResultsAPath for details.") + } + } + + "load normally a fixed-record-length file without indexes and with record ids" in { + val gpgPrivateKey = ResourceUtils.readResourceAsString("/test/test_gpg_key.asc") + val df = spark.read + .format("cobol") + .option("copybook", inputCopybookPath) + .option("gpg_private_key", gpgPrivateKey) + .option("pedantic", "true") + .option("generate_record_id", "true") + .option("enable_indexes", "false") + .load(inputDataPath) + + val actual = df.toJSON.take(60) + val expected = Files.readAllLines(Paths.get(expectedResultsBPath), StandardCharsets.ISO_8859_1).toArray + + if (!actual.sameElements(expected)) { + FileUtils.writeStringsToFile(actual, actualResultsBPath) + assert(false, s"The actual data doesn't match what is expected for $exampleName example. Please compare contents of $expectedResultsBPath to $actualResultsBPath for details.") + } + } + + "load normally a variable-record-length file with indexes" in { + val gpgPrivateKey = ResourceUtils.readResourceAsString("/test/test_gpg_key.asc") + val df = spark.read + .format("cobol") + .option("copybook", inputCopybookPath) + .option("gpg_private_key", gpgPrivateKey) + .option("generate_record_id", "true") + .option("input_split_records", 1) + .option("pedantic", "true") + .load(inputDataPath) + + val actual = df.toJSON.take(60) + val expected = Files.readAllLines(Paths.get(expectedResultsCPath), StandardCharsets.ISO_8859_1).toArray + + if (!actual.sameElements(expected)) { + FileUtils.writeStringsToFile(actual, actualResultsCPath) + assert(false, s"The actual data doesn't match what is expected for $exampleName example. Please compare contents of $expectedResultsCPath to $actualResultsCPath for details.") + } + } + } +} diff --git a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/utils/GpgUtilsSuite.scala b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/utils/GpgUtilsSuite.scala new file mode 100644 index 00000000..57b8d918 --- /dev/null +++ b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/utils/GpgUtilsSuite.scala @@ -0,0 +1,58 @@ +/* + * Copyright 2018 ABSA Group Limited + * + * Licensed 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 za.co.absa.cobrix.spark.cobol.utils + +import org.scalatest.wordspec.AnyWordSpec + +import java.io.ByteArrayInputStream +import scala.io.Source + +class GpgUtilsSuite extends AnyWordSpec { + + import za.co.absa.cobrix.cobol.utils.UsingUtils.Implicits._ + + "decryptStream" should { + "decrypt an input stream" in { + val gpgPrivateKey = ResourceUtils.readResourceAsString("/test/test_gpg_key.asc") + + val decryptedText = for { + iss <- getClass.getResourceAsStream("/test/test_gpg_file.gpg") + oss <- GpgUtils.decryptStream(iss, gpgPrivateKey, Array.empty[Char]) + } yield { + Source.fromInputStream(oss).mkString + } + + assert(decryptedText.trim == "This is a test") + } + + "fail if the stream is not GPG-encrypted" in { + val gpgPrivateKey = ResourceUtils.readResourceAsString("/test/test_gpg_key.asc") + + val ex = intercept[RuntimeException] { + for { + iss <- new ByteArrayInputStream(Array[Byte](0, 0, 0, 0)) + oss <- GpgUtils.decryptStream(iss, gpgPrivateKey, Array.empty[Char]) + } yield { + Source.fromInputStream(oss).mkString + } + } + + assert(ex.getMessage.contains("The input stream does not contain PGP encrypted data.")) + } + } + +}