[OSC-1277] prevent logging PI (#28)

* [OSC-1277] prevent logging PI

* add unit tests for PI leak

* update unit tests to echo properly

* ensure failure 1 works

* ensure failure 2 works

* log sensitive error data to db

Author: seanbudd

appveyor.yml

@@ -66,6 +66,7 @@ test_script:
 
   - test_full_refresh_from_mssql.cmd
   - test_audit.cmd
+  - test_pi_leak.cmd
   - test_mssql_failover_server.cmd
 
 on_finish:

integration_tests/mssql_source/config_fail/CompoundPkTestFail.json

@@ -0,0 +1,33 @@
+{
+  "source_table": {
+    "name": "CompoundPk",
+    "schema": "dbo",
+    "primary_keys": ["Id1"]
+  },
+  "target_schema": "rdl_integration_tests",
+  "stage_table": "stage_compound_pk",
+  "load_table": "load_compound_pk",
+
+  "batch": {
+    "size": 100000
+  },
+  "columns": [
+    {
+      "source_name": "Id1",
+      "destination": {
+        "name": "id_1",
+        "type": "int",
+        "nullable": false,
+        "primary_key": true
+      }
+    },
+    {
+      "source_name": "Id2",
+      "destination": {
+        "name": "id_2",
+        "type": "int",
+        "nullable": true
+      }
+    }
+  ]
+}

rdl/BatchDataLoader.py

@@ -4,6 +4,7 @@
 from io import StringIO
 from rdl.column_transformers.StringTransformers import ToUpper
 from rdl.shared import Constants
+from rdl.shared.Utils import prevent_senstive_data_logging
 
 
 class BatchDataLoader(object):
@@ -49,6 +50,7 @@ def load_batch(self, batch_key_tracker):
 
         self.logger.info(f"Batch keys '{batch_key_tracker.bookmarks}' completed. {batch_tracker.get_statistics()}")
 
+    @prevent_senstive_data_logging
     def write_data_frame_to_table(self, data_frame):
         qualified_target_table = f'{self.target_schema}.{self.target_table}'
         self.logger.debug(f"Starting write to table '{qualified_target_table}'")

rdl/DataLoadManager.py

@@ -10,6 +10,7 @@
 from rdl.data_load_tracking.DataLoadTracker import DataLoadTracker
 from rdl.BatchKeyTracker import BatchKeyTracker
 from rdl.shared import Constants
+from rdl.shared.Utils import SensitiveDataError
 
 
 class DataLoadManager(object):
@@ -137,7 +138,12 @@ def start_single_import(self, model_file, requested_full_refresh, model_number,
 
         batch_key_tracker = BatchKeyTracker(model_config['source_table']['primary_keys'])
         while batch_key_tracker.has_more_data:
-            batch_data_loader.load_batch(batch_key_tracker)
+            try:
+                batch_data_loader.load_batch(batch_key_tracker)
+            except SensitiveDataError as e:
+                data_load_tracker.data_load_failed(e.sensitive_error_args)
+                self.data_load_tracker_repository.save(data_load_tracker)
+                raise e
 
         if full_refresh:
             # Rename the stage table to the load table.
@@ -154,7 +160,7 @@ def start_single_import(self, model_file, requested_full_refresh, model_number,
 
             destination_table_manager.drop_table(model_config['target_schema'],
                                                  model_config['stage_table'])
-        data_load_tracker.completed_successfully()
+        data_load_tracker.data_load_successful()
         self.data_load_tracker_repository.save(data_load_tracker)
         self.logger.info(f"{model_number:0{max_model_number_len}d} of {total_number_of_models}"
                          f" COMPLETED {model_name},"

rdl/data_load_tracking/DataLoadExecution.py

@@ -22,3 +22,4 @@ class DataLoadExecution(Base):
     execution_time_ms = Column(Integer, nullable=False)
     rows_processed = Column(Integer, nullable=False)
     model_checksum = Column(String(100), nullable=False)
+    failure_reason = Column(String(1000), nullable=True)

rdl/data_load_tracking/DataLoadTracker.py

@@ -34,19 +34,26 @@ def __init__(
         self.change_tracking_info = change_tracking_info
         self.correlation_id = correlation_id
         self.full_refresh_reason = full_refresh_reason
+        self.failure_reason = None
 
     def start_batch(self):
         batch = self.Batch()
         self.batches.append(batch)
         return batch
 
-    def completed_successfully(self):
+    def data_load_successful(self):
+        self.data_load_completed(Constants.ExecutionStatus.COMPLETED_SUCCESSFULLY)
+
+    def data_load_failed(self, failure_reason=None):
+        self.data_load_completed(Constants.ExecutionStatus.FAILED, failure_reason)
+
+    def data_load_completed(self, execution_status, failure_reason=None):
         self.completed = datetime.now()
         self.total_execution_time = self.completed - self.started
-        self.status = Constants.ExecutionStatus.COMPLETED_SUCCESSFULLY
+        self.status = execution_status
+        self.failure_reason = failure_reason
         for batch in self.batches:
             self.total_row_count += batch.row_count
-
         self.rows_per_second = self.total_row_count / (self.total_execution_time.total_seconds() + 1e-10)
 
     def get_statistics(self):

rdl/data_load_tracking/DataLoadTrackerRepository.py

@@ -36,7 +36,8 @@ def save(self, data_load_tracker):
             full_refresh_reason=data_load_tracker.full_refresh_reason,
             execution_time_ms=int(data_load_tracker.total_execution_time.total_seconds() * 1000),
             rows_processed=data_load_tracker.total_row_count,
-            model_checksum=data_load_tracker.model_checksum)
+            model_checksum=data_load_tracker.model_checksum,
+            failure_reason=data_load_tracker.failure_reason)
 
         session = self.session_maker()
         session.add(data_load_execution)

rdl/data_sources/CsvDataSource.py

@@ -1,9 +1,11 @@
 import logging
-import pandas
 import os.path
-from rdl.ColumnTypeResolver import ColumnTypeResolver
 from pathlib import Path
+import pandas
+
+from rdl.ColumnTypeResolver import ColumnTypeResolver
 from rdl.data_sources.ChangeTrackingInfo import ChangeTrackingInfo
+from rdl.shared.Utils import prevent_senstive_data_logging
 
 
 class CsvDataSource(object):
@@ -42,6 +44,7 @@ def assert_column_exists(self, column_name, data_frame, csv_file):
 
     # For now, the CSV data sources will get all rows in the CSV regardless of
     # batch size. - Ie, they don't currently support paging.
+    @prevent_senstive_data_logging
     def get_next_data_frame(
             self,
             table_config,

rdl/data_sources/MsSqlDataSource.py

@@ -13,6 +13,7 @@
 from rdl.ColumnTypeResolver import ColumnTypeResolver
 from rdl.data_sources.ChangeTrackingInfo import ChangeTrackingInfo
 from rdl.shared import Constants
+from rdl.shared.Utils import prevent_senstive_data_logging
 
 
 class MsSqlDataSource(object):
@@ -134,6 +135,7 @@ def get_table_columns(self, table_config):
                       autoload_with=self.database_engine)
         return list(map(lambda column: column.name, table.columns))
 
+    @prevent_senstive_data_logging
     def get_next_data_frame(self, table_config, columns, batch_config, batch_tracker, batch_key_tracker,
                             full_refresh, change_tracking_info):
         sql = self.build_select_statement(table_config, columns, batch_config, batch_key_tracker,

rdl/shared/Constants.py

@@ -14,6 +14,7 @@ class FullRefreshReason:
 
 class ExecutionStatus:
     NOT_STARTED = 'Not Started'
+    FAILED = 'Failed'
     EXTRACT_COMPLETED_SUCCESSFULLY = 'Extract Completed Successfully'
     LOAD_COMPLETED_SUCCESSFULLY = 'Load Completed Successfully'
     SKIPPED_AS_ZERO_ROWS = 'Skipped - Zero Rows'