Merge pull request #186 from hecfernandez/hectorfernandez/prevent-csv-injection

Hectorfernandez/prevent csv injection

Author: CarlosLVar

Diff for: taiga/base/utils/text.py

@@ -5,6 +5,7 @@
 #
 # Copyright (c) 2021-present Kaleidos INC
 
+
 def strip_lines(text):
     """
     Given text, try remove unnecessary spaces and
@@ -16,3 +17,12 @@ def strip_lines(text):
 def split_in_lines(text):
     """Split a block of text in lines removing unnecessary spaces from each line."""
     return (line for line in map(str.strip, text.split("\n")) if line)
+
+
+def sanitize_csv_text_value(value):
+    """Sanitize CSV data to prevent formula injection attacks."""
+    return (
+        "'" + value
+        if isinstance(value, str) and value.startswith(("=", "+", "-", "@"))
+        else value
+    )

Diff for: taiga/projects/epics/services.py

@@ -32,6 +32,7 @@
 # Bulk actions
 #####################################################
 
+
 def get_epics_from_bulk(bulk_data, **additional_fields):
     """Convert `bulk_data` into a list of epics.
 
@@ -40,8 +41,10 @@ def get_epics_from_bulk(bulk_data, **additional_fields):
 
     :return: List of `Epic` instances.
     """
-    return [models.Epic(subject=line, **additional_fields)
-            for line in text.split_in_lines(bulk_data)]
+    return [
+        models.Epic(subject=line, **additional_fields)
+        for line in text.split_in_lines(bulk_data)
+    ]
 
 
 def create_epics_in_bulk(bulk_data, callback=None, precall=None, **additional_fields):
@@ -79,9 +82,9 @@ def update_epics_order_in_bulk(bulk_data: list, field: str, project: object):
     apply_order_updates(epic_orders, new_epic_orders)
 
     epic_ids = epic_orders.keys()
-    events.emit_event_for_ids(ids=epic_ids,
-                              content_type="epics.epic",
-                              projectid=project.pk)
+    events.emit_event_for_ids(
+        ids=epic_ids, content_type="epics.epic", projectid=project.pk
+    )
 
     db.update_attr_in_bulk_for_ids(epic_orders, field, models.Epic)
     return epic_orders
@@ -111,10 +114,7 @@ def create_related_userstories_in_bulk(bulk_data, epic, **additional_fields):
         related_userstories = []
         for userstory in userstories:
             related_userstories.append(
-                models.RelatedUserStory(
-                    user_story=userstory,
-                    epic=epic
-                )
+                models.RelatedUserStory(user_story=userstory, epic=epic)
             )
         db.save_in_bulk(related_userstories)
         project.update_role_points(user_stories=userstories)
@@ -137,16 +137,21 @@ def update_epic_related_userstories_order_in_bulk(bulk_data: list, epic: object)
     rus_orders = {rus.id: rus.order for rus in related_user_stories}
 
     rus_conversion = {rus.user_story_id: rus.id for rus in related_user_stories}
-    new_rus_orders = {rus_conversion[e["us_id"]]: e["order"] for e in bulk_data
-                      if e["us_id"] in rus_conversion}
+    new_rus_orders = {
+        rus_conversion[e["us_id"]]: e["order"]
+        for e in bulk_data
+        if e["us_id"] in rus_conversion
+    }
 
     apply_order_updates(rus_orders, new_rus_orders)
 
     if rus_orders:
         related_user_story_ids = rus_orders.keys()
-        events.emit_event_for_ids(ids=related_user_story_ids,
-                                  content_type="epics.relateduserstory",
-                                  projectid=epic.project_id)
+        events.emit_event_for_ids(
+            ids=related_user_story_ids,
+            content_type="epics.relateduserstory",
+            projectid=epic.project_id,
+        )
 
         db.update_attr_in_bulk_for_ids(rus_orders, "order", models.RelatedUserStory)
 
@@ -157,25 +162,39 @@ def update_epic_related_userstories_order_in_bulk(bulk_data: list, epic: object)
 # CSV
 #####################################################
 
+
 def epics_to_csv(project, queryset):
     csv_data = io.StringIO()
-    fieldnames = ["id", "ref", "subject", "description", "owner", "owner_full_name",
-                  "assigned_to", "assigned_to_full_name", "status", "epics_order",
-                  "client_requirement", "team_requirement", "attachments", "tags",
-                  "watchers", "voters", "created_date", "modified_date",
-                  "related_user_stories"]
+    fieldnames = [
+        "id",
+        "ref",
+        "subject",
+        "description",
+        "owner",
+        "owner_full_name",
+        "assigned_to",
+        "assigned_to_full_name",
+        "status",
+        "epics_order",
+        "client_requirement",
+        "team_requirement",
+        "attachments",
+        "tags",
+        "watchers",
+        "voters",
+        "created_date",
+        "modified_date",
+        "related_user_stories",
+    ]
 
     custom_attrs = project.epiccustomattributes.all()
     for custom_attr in custom_attrs:
         fieldnames.append(custom_attr.name)
 
-    queryset = queryset.prefetch_related("attachments",
-                                         "custom_attributes_values",
-                                         "user_stories__project")
-    queryset = queryset.select_related("owner",
-                                       "assigned_to",
-                                       "status",
-                                       "project")
+    queryset = queryset.prefetch_related(
+        "attachments", "custom_attributes_values", "user_stories__project"
+    )
+    queryset = queryset.select_related("owner", "assigned_to", "status", "project")
 
     queryset = attach_total_voters_to_queryset(queryset)
     queryset = attach_watchers_to_queryset(queryset)
@@ -186,12 +205,20 @@ def epics_to_csv(project, queryset):
         epic_data = {
             "id": epic.id,
             "ref": epic.ref,
-            "subject": epic.subject,
-            "description": epic.description,
+            "subject": text.sanitize_csv_text_value(epic.subject),
+            "description": text.sanitize_csv_text_value(epic.description),
             "owner": epic.owner.username if epic.owner else None,
-            "owner_full_name": epic.owner.get_full_name() if epic.owner else None,
+            "owner_full_name": (
+                text.sanitize_csv_text_value(epic.owner.get_full_name())
+                if epic.owner
+                else None
+            ),
             "assigned_to": epic.assigned_to.username if epic.assigned_to else None,
-            "assigned_to_full_name": epic.assigned_to.get_full_name() if epic.assigned_to else None,
+            "assigned_to_full_name": (
+                text.sanitize_csv_text_value(epic.assigned_to.get_full_name())
+                if epic.assigned_to
+                else None
+            ),
             "status": epic.status.name if epic.status else None,
             "epics_order": epic.epics_order,
             "client_requirement": epic.client_requirement,
@@ -202,16 +229,21 @@ def epics_to_csv(project, queryset):
             "voters": epic.total_voters,
             "created_date": epic.created_date,
             "modified_date": epic.modified_date,
-            "related_user_stories": ",".join([
-                "{}#{}".format(us.project.slug, us.ref) for us in epic.user_stories.all()
-            ]),
+            "related_user_stories": ",".join(
+                [
+                    "{}#{}".format(us.project.slug, us.ref)
+                    for us in epic.user_stories.all()
+                ]
+            ),
         }
 
         for custom_attr in custom_attrs:
             if not hasattr(epic, "custom_attributes_values"):
                 continue
-            value = epic.custom_attributes_values.attributes_values.get(str(custom_attr.id), None)
-            epic_data[custom_attr.name] = value
+            value = epic.custom_attributes_values.attributes_values.get(
+                str(custom_attr.id), None
+            )
+            epic_data[custom_attr.name] = text.sanitize_csv_text_value(value)
 
         writer.writerow(epic_data)
 
@@ -222,8 +254,11 @@ def epics_to_csv(project, queryset):
 # Api filter data
 #####################################################
 
+
 def _get_epics_statuses(project, queryset):
-    compiler = connection.ops.compiler(queryset.query.compiler)(queryset.query, connection, None)
+    compiler = connection.ops.compiler(queryset.query.compiler)(
+        queryset.query, connection, None
+    )
     queryset_where_tuple = queryset.query.where.as_sql(compiler, connection)
     where = queryset_where_tuple[0]
     where_params = queryset_where_tuple[1]
@@ -241,26 +276,32 @@ def _get_epics_statuses(project, queryset):
         FROM "projects_epicstatus"
        WHERE "projects_epicstatus"."project_id" = %s
     ORDER BY "projects_epicstatus"."order";
-    """.format(where=where)
+    """.format(
+        where=where
+    )
 
     with closing(connection.cursor()) as cursor:
         cursor.execute(extra_sql, where_params + [project.id])
         rows = cursor.fetchall()
 
     result = []
     for id, name, color, order, count in rows:
-        result.append({
-            "id": id,
-            "name": _(name),
-            "color": color,
-            "order": order,
-            "count": count,
-        })
+        result.append(
+            {
+                "id": id,
+                "name": _(name),
+                "color": color,
+                "order": order,
+                "count": count,
+            }
+        )
     return sorted(result, key=itemgetter("order"))
 
 
 def _get_epics_assigned_to(project, queryset):
-    compiler = connection.ops.compiler(queryset.query.compiler)(queryset.query, connection, None)
+    compiler = connection.ops.compiler(queryset.query.compiler)(
+        queryset.query, connection, None
+    )
     queryset_where_tuple = queryset.query.where.as_sql(compiler, connection)
     where = queryset_where_tuple[0]
     where_params = queryset_where_tuple[1]
@@ -292,7 +333,9 @@ def _get_epics_assigned_to(project, queryset):
              INNER JOIN "projects_project" ON ("epics_epic"."project_id" = "projects_project"."id")
                   WHERE {where} AND "epics_epic"."assigned_to_id" IS NULL
                GROUP BY assigned_to_id
-    """.format(where=where)
+    """.format(
+        where=where
+    )
 
     with closing(connection.cursor()) as cursor:
         cursor.execute(extra_sql, where_params + [project.id] + where_params)
@@ -301,28 +344,34 @@ def _get_epics_assigned_to(project, queryset):
     result = []
     none_valued_added = False
     for id, full_name, username, count in rows:
-        result.append({
-            "id": id,
-            "full_name": full_name or username or "",
-            "count": count,
-        })
+        result.append(
+            {
+                "id": id,
+                "full_name": full_name or username or "",
+                "count": count,
+            }
+        )
 
         if id is None:
             none_valued_added = True
 
     # If there was no epic with null assigned_to we manually add it
     if not none_valued_added:
-        result.append({
-            "id": None,
-            "full_name": "",
-            "count": 0,
-        })
+        result.append(
+            {
+                "id": None,
+                "full_name": "",
+                "count": 0,
+            }
+        )
 
     return sorted(result, key=itemgetter("full_name"))
 
 
 def _get_epics_owners(project, queryset):
-    compiler = connection.ops.compiler(queryset.query.compiler)(queryset.query, connection, None)
+    compiler = connection.ops.compiler(queryset.query.compiler)(
+        queryset.query, connection, None
+    )
     queryset_where_tuple = queryset.query.where.as_sql(compiler, connection)
     where = queryset_where_tuple[0]
     where_params = queryset_where_tuple[1]
@@ -357,7 +406,9 @@ def _get_epics_owners(project, queryset):
                    FROM users_user
         LEFT OUTER JOIN counters ON ("users_user"."id" = "counters"."owner_id")
                   WHERE ("users_user"."is_system" IS TRUE)
-    """.format(where=where)
+    """.format(
+        where=where
+    )
 
     with closing(connection.cursor()) as cursor:
         cursor.execute(extra_sql, where_params + [project.id])
@@ -366,16 +417,20 @@ def _get_epics_owners(project, queryset):
     result = []
     for id, full_name, username, count in rows:
         if count > 0:
-            result.append({
-                "id": id,
-                "full_name": full_name or username or "",
-                "count": count,
-            })
+            result.append(
+                {
+                    "id": id,
+                    "full_name": full_name or username or "",
+                    "count": count,
+                }
+            )
     return sorted(result, key=itemgetter("full_name"))
 
 
 def _get_epics_tags(project, queryset):
-    compiler = connection.ops.compiler(queryset.query.compiler)(queryset.query, connection, None)
+    compiler = connection.ops.compiler(queryset.query.compiler)(
+        queryset.query, connection, None
+    )
     queryset_where_tuple = queryset.query.where.as_sql(compiler, connection)
     where = queryset_where_tuple[0]
     where_params = queryset_where_tuple[1]
@@ -401,19 +456,23 @@ def _get_epics_tags(project, queryset):
         FROM project_tags
    LEFT JOIN epics_tags ON project_tags.tag_color[1] = epics_tags.tag
     ORDER BY tag
-    """.format(where=where)
+    """.format(
+        where=where
+    )
 
     with closing(connection.cursor()) as cursor:
         cursor.execute(extra_sql, where_params + [project.id])
         rows = cursor.fetchall()
 
     result = []
     for name, color, count in rows:
-        result.append({
-            "name": name,
-            "color": color,
-            "count": count,
-        })
+        result.append(
+            {
+                "name": name,
+                "color": color,
+                "count": count,
+            }
+        )
     return sorted(result, key=itemgetter("name"))
 
 
@@ -422,11 +481,13 @@ def get_epics_filters_data(project, querysets):
     Given a project and an epics queryset, return a simple data structure
     of all possible filters for the epics in the queryset.
     """
-    data = OrderedDict([
-        ("statuses", _get_epics_statuses(project, querysets["statuses"])),
-        ("assigned_to", _get_epics_assigned_to(project, querysets["assigned_to"])),
-        ("owners", _get_epics_owners(project, querysets["owners"])),
-        ("tags", _get_epics_tags(project, querysets["tags"])),
-    ])
+    data = OrderedDict(
+        [
+            ("statuses", _get_epics_statuses(project, querysets["statuses"])),
+            ("assigned_to", _get_epics_assigned_to(project, querysets["assigned_to"])),
+            ("owners", _get_epics_owners(project, querysets["owners"])),
+            ("tags", _get_epics_tags(project, querysets["tags"])),
+        ]
+    )
 
     return data