added test

Author: stiwarisemanticbits

apps/logging/sensitive_logging_filters.py

@@ -0,0 +1,69 @@
+import re
+import logging
+import logging.config
+
+MBI_WITH_HYPHEN_PATTERN = r"""\b
+    [1-9](?![SLOIBZsloibz])[A-Za-z](?![SLOIBZsloibz)])[A-Za-z\d]\d
+    -(?![SLOIBZsloibz])[A-Za-z](?![SLOIBZsloibz])[A-Za-z\d]\d
+    -((?![SLOIBZsloibz])[A-Za-z]){2}\d{2}
+    \b
+    """
+
+MBI_WITHOUT_HYPHEN_PATTERN = r"""\b
+    [1-9](?![SLOIBZsloibz])[A-Za-z](?![SLOIBZsloibz)])[A-Za-z\d]\d
+    (?![SLOIBZsloibz])[A-Za-z](?![SLOIBZsloibz])[A-Za-z\d]\d
+    ((?![SLOIBZsloibzd])[A-Za-z]){2}\d{2}
+    \b"""
+
+MBI_PATTERN = f'({MBI_WITH_HYPHEN_PATTERN}|{MBI_WITHOUT_HYPHEN_PATTERN})'
+SENSITIVE_DATA_FILTER = "sensitive_data_filter"
+
+
+def has_mbi_match(text):
+    return bool(re.search(MBI_PATTERN, text, flags=re.VERBOSE))
+
+
+def mask_if_has_mbi(text):
+    return re.sub(MBI_PATTERN, '***MBI***', str(text), flags=re.VERBOSE)
+
+
+def is_not_primitive(value):
+    primitive_types = (int, float, bool, str, bytes)
+    return not isinstance(value, primitive_types)
+
+
+def mask_mbi(value_to_mask):
+    if isinstance(value_to_mask, str):
+        return mask_if_has_mbi(value_to_mask)
+
+    if isinstance(value_to_mask, tuple):
+        return tuple([mask_if_has_mbi(arg) for arg in value_to_mask])
+
+    if isinstance(value_to_mask, list):
+        return [mask_if_has_mbi(arg) for arg in value_to_mask]
+
+    if isinstance(value_to_mask, dict):
+        for key, value in value_to_mask.items():
+            if is_not_primitive(value):
+                mask_mbi(value)
+            else:
+                value_to_mask[key] = mask_if_has_mbi(value)
+
+    return value_to_mask
+
+
+class SensitiveDataFilter(logging.Filter):
+
+    def filter(self, record):
+        try:
+            record.args = mask_mbi(record.args)
+            record.msg = mask_mbi(record.msg)
+            return True
+        except Exception:
+            pass
+
+    def mask_sensitive_args(self, args):
+        if isinstance(args, dict):
+            return mask_mbi(dict)
+
+        return tuple([mask_if_has_mbi(arg) for arg in args])

hhs_oauth_server/request_logging.py

@@ -1,7 +1,6 @@
 import datetime
 import hashlib
 import json
-import re
 import uuid
 
 import apps.logging.request_logger as logging
@@ -23,9 +22,7 @@
 )
 
 audit = logging.getLogger("audit.%s" % __name__)
-MBI_WITH_HYPHEN_PATTERN = r'\b\d[A-Z]{2}\d-[A-Z]{2}\d-[A-Z]{2}\d{2}\b'
-MBI_WITHOUT_HYPHEN_PATTERN = r'\b[1-9](?![SLOIBZ])[A-Z](?![SLOIBZ)])[A-Z\d]\d(?![SLOIBZ])[A-Z](?![SLOIBZ])[A-Z\d]\d(?![SLOIBZ])[A-Z]{2}\d{2}\b'
-MBI_PATTERN = f'({MBI_WITH_HYPHEN_PATTERN}|{MBI_WITHOUT_HYPHEN_PATTERN})'
+
 
 class RequestResponseLog(object):
     """Audit log message to JSON string
@@ -140,20 +137,13 @@ def __init__(self, req, resp):
         self.log_msg["location"] = ""
         self.log_msg["size"] = 0
 
-    
-    def has_mbi_match(text):
-        return bool(re.search(MBI_PATTERN, text))
-    
-    def mask_if_has_mbi(text):
-        return re.sub(MBI_PATTERN, '***MBI***', text)
-        
     def _log_msg_update_from_dict(self, from_dict, key, dict_key):
         # Log message update from a passed in dictionary
         try:
             value = from_dict.get(dict_key, None)
             if value is not None:
                 if len(str(value)) > 0:
-                    self.log_msg[key] = self.mask_if_has_mbi(value)
+                    self.log_msg[key] = value
         except ObjectDoesNotExist:
             self.log_msg[key] = (
                 "ObjectDoesNotExist exception for key " + key + ":" + dict_key
@@ -169,7 +159,7 @@ def _log_msg_update_from_object(self, obj, key, obj_key):
             value = getattr(obj, obj_key, None)
             if value is not None:
                 if len(str(value)) > 0:
-                    self.log_msg[key] = self.mask_if_has_mbi(value)
+                    self.log_msg[key] = value
         except ObjectDoesNotExist:
             self.log_msg[key] = (
                 "ObjectDoesNotExist exception for key " + key + ":" + obj_key
@@ -184,7 +174,6 @@ def _log_msg_update_from_querydict(self, key, qp_key):
         try:
             value_list = self.request.GET.getlist(qp_key, None)
             if value_list is not None:
-                value_list = [self.mask_if_has_mbi(value) for value in value_list]
                 if len(value_list) == 1:
                     self.log_msg[key] = value_list[0]
                 elif len(value_list) > 1:
@@ -532,8 +521,8 @@ def to_dict(self):
                     except ObjectDoesNotExist:
                         pass
         self._sync_app_name()
-        masked_logged_dict = {key: self.mask_if_has_mbi(value) for key, value in self.log_msg.items()}
-        return masked_logged_dict
+
+        return self.log_msg
 
 ##############################################################################
 #

hhs_oauth_server/settings/base.py

@@ -1,4 +1,5 @@
 import os
+from apps.logging.sensitive_logging_filters import SENSITIVE_DATA_FILTER, SensitiveDataFilter
 import dj_database_url
 import socket
 import datetime
@@ -377,6 +378,12 @@
             "console": {
                 "class": "logging.StreamHandler",
                 "formatter": "verbose",
+                "filters": [SENSITIVE_DATA_FILTER],
+            }
+        },
+        "filters": {
+            "sensitive_data_filter": {
+                "()": SensitiveDataFilter,
             }
         },
         "loggers": {
@@ -421,6 +428,11 @@
                 "handlers": ["console"],
                 "level": "INFO",
             },
+            'django': {
+                'handlers': ['console'],
+                'level': 'INFO',
+                'propagate': True,
+            },
         },
     },
 )

hhs_oauth_server/settings/logging_it.py

@@ -21,7 +21,8 @@
     raise ValueError("Bad settings, expecting handlers defined in settings.LOGGING")
 
 logging_handlers['file'] = {'class': 'logging.FileHandler',
-                            'filename': logfile_path, }
+                            'filename': logfile_path,
+                            "filters": [SENSITIVE_DATA_FILTER]}
 
 loggers = LOGGING.get('loggers')