Sanitize exported CSVs to avoid CSV injection.

kolibri/core/auth/csv_utils.py

@@ -17,6 +17,7 @@
 from kolibri.core.auth.models import FacilityUser
 from kolibri.core.query import SQCount
 from kolibri.core.utils.csv import open_csv_for_writing
+from kolibri.core.utils.csv import sanitize
 
 
 logger = logging.getLogger(__name__)
@@ -103,6 +104,7 @@ def map_output(obj):
             mapped_obj[label] = output_mappings[header](obj)
         elif header in obj:
             mapped_obj[label] = obj[header]
+        mapped_obj[label] = sanitize(mapped_obj[label])
     return mapped_obj
 
 

kolibri/core/auth/management/commands/bulkexportusers.py

@@ -27,6 +27,7 @@
 from kolibri.core.tasks.management.commands.base import AsyncCommand
 from kolibri.core.tasks.utils import get_current_job
 from kolibri.core.utils.csv import open_csv_for_writing
+from kolibri.core.utils.csv import sanitize
 from kolibri.utils import conf
 
 try:
@@ -114,6 +115,7 @@ def map_output(obj):
             mapped_obj[label] = output_mappings[header](obj)
         elif header in obj:
             mapped_obj[label] = obj[header]
+        mapped_obj[label] = sanitize(mapped_obj[label])
     return mapped_obj
 
 

kolibri/core/logger/csv_export.py

@@ -24,6 +24,7 @@
 from kolibri.core.content.models import ChannelMetadata
 from kolibri.core.content.models import ContentNode
 from kolibri.core.utils.csv import open_csv_for_writing
+from kolibri.core.utils.csv import sanitize
 from kolibri.utils import conf
 
 
@@ -94,6 +95,7 @@ def map_object(obj):
             mapped_obj[label] = mappings[header](obj)
         elif header in obj:
             mapped_obj[label] = obj[header]
+        mapped_obj[label] = sanitize(mapped_obj[label])
     return mapped_obj
 
 

kolibri/core/utils/csv.py

@@ -1,5 +1,7 @@
 import io
+import re
 import sys
+from numbers import Number
 
 
 def open_csv_for_writing(filepath):
@@ -12,3 +14,22 @@ def open_csv_for_reading(filepath):
     if sys.version_info[0] < 3:
         return io.open(filepath, "rb")
     return io.open(filepath, "r", newline="", encoding="utf-8-sig")
+
+
+negative_number_regex = re.compile("^-?[0-9,\\.]+$")
+csv_injection_chars = {"@", "+", "-", "=", "|", "%"}
+
+
+def sanitize(value):
+    if value is None or isinstance(value, Number):
+        return value
+
+    value = str(value)
+    if (
+        value
+        and value[0] in csv_injection_chars
+        and not negative_number_regex.match(value)
+    ):
+        value = value.replace("|", "\\|")
+        value = "'" + value
+    return value