fix(sanitizing): Return ampersand after sanitizing in validator and p…

…atch for de-sanitize & to &

src/apps/shared/commands/patch_commands.py

@@ -84,6 +84,11 @@
     task_id="M2-5116",
     description="[Subject] Populate alerts with subject ids",
 )
+PatchRegister.register(
+    file_path="m2_6757_replace_amp_sanitizer.py",
+    task_id="M2-6757",
+    description="Change ampersand sanitizer to symbol '&'",
+)
 
 app = typer.Typer()
 

src/apps/shared/commands/patches/m2_6757_replace_amp_sanitizer.py

@@ -0,0 +1,116 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+UPDATE_APPLET_SQL = """
+    UPDATE applets
+    SET
+        "display_name" = regexp_replace("display_name", '&', '&', 'g'),
+        "about" = regexp_replace(about::text, '&', '&', 'g')::jsonb,
+        "description" = regexp_replace(description::text, '&', '&', 'g')::jsonb
+    WHERE
+        "display_name" LIKE '%&%'
+        OR (about->>'en' LIKE '%&%' OR about->>'fr' LIKE '%&%')
+        OR (description->>'en' LIKE '%&%' OR description->>'fr' LIKE '%&%');
+"""
+
+UPDATE_APPLET_HISTORY_SQL = """
+    UPDATE applet_histories
+    SET
+        "display_name" = regexp_replace("display_name", '&', '&', 'g'),
+        "about" = regexp_replace(about::text, '&', '&', 'g')::jsonb,
+        "description" = regexp_replace(description::text, '&', '&', 'g')::jsonb
+    WHERE
+        "display_name" LIKE '%&%'
+        OR (about->>'en' LIKE '%&%' OR about->>'fr' LIKE '%&%')
+        OR (description->>'en' LIKE '%&%' OR description->>'fr' LIKE '%&%');
+"""
+
+UPDATE_ACTIVITY_SQL = """
+    UPDATE activities
+    SET 
+        "name" = regexp_replace("name", '&', '&', 'g'),
+        "description" = regexp_replace(description::text, '&', '&', 'g')::jsonb
+    WHERE "name" LIKE '%&%' OR (description->>'en' LIKE '%&%' OR description->>'fr' LIKE '%&%');
+"""
+
+UPDATE_ACTIVITY_HISTORY_SQL = """
+    UPDATE activity_histories
+    SET 
+        "name" = regexp_replace("name", '&', '&', 'g'),
+        "description" = regexp_replace(description::text, '&', '&', 'g')::jsonb
+    WHERE "name" LIKE '%&%' OR (description->>'en' LIKE '%&%' OR description->>'fr' LIKE '%&%');
+"""
+
+
+UPDATE_ACTIVITY_SCORES_AND_REPORTS_SQL = """
+    UPDATE activities
+    SET scores_and_reports = regexp_replace(scores_and_reports::text, '&', '&', 'g')::jsonb
+    WHERE EXISTS (
+        SELECT 1
+        FROM jsonb_array_elements(scores_and_reports->'reports') AS report
+        WHERE report->>'message' LIKE '%&%'
+    );
+"""
+
+UPDATE_ACTIVITY_HISTORY_SCORES_AND_REPORTS_SQL = """
+    UPDATE activity_histories
+    SET scores_and_reports = regexp_replace(scores_and_reports::text, '&', '&', 'g')::jsonb
+    WHERE EXISTS (
+        SELECT 1
+        FROM jsonb_array_elements(scores_and_reports->'reports') AS report
+        WHERE report->>'message' LIKE '%&%'
+    );
+"""
+
+UPDATE_ACTIVITY_ITEM_SQL = """
+    UPDATE activity_items
+    SET 
+        "name" = regexp_replace("name", '&', '&', 'g'),
+        "question" = regexp_replace(question::text, '&', '&', 'g')::jsonb
+    WHERE "name" LIKE '%&%' OR (question->>'en' LIKE '%&%' OR question->>'fr' LIKE '%&%');
+"""
+UPDATE_ACTIVITY_ITEM_HISTORY_SQL = """
+    UPDATE activity_item_histories
+    SET 
+        "name" = regexp_replace("name", '&', '&', 'g'),
+        "question" = regexp_replace(question::text, '&', '&', 'g')::jsonb
+    WHERE "name" LIKE '%&%' OR (question->>'en' LIKE '%&%' OR question->>'fr' LIKE '%&%');
+"""
+
+UPDATE_FLOW_SQL = """
+    UPDATE flows
+    SET 
+        "name" = regexp_replace("name", '&', '&', 'g'),
+        "description" = regexp_replace(description::text, '&', '&', 'g')::jsonb
+    WHERE "name" LIKE '%&%' OR (description->>'en' LIKE '%&%' OR description->>'fr' LIKE '%&%');
+"""
+
+UPDATE_FLOW_HISTORY__SQL = """
+    UPDATE flow_histories
+    SET 
+        "name" = regexp_replace("name", '&', '&', 'g'),
+        "description" = regexp_replace(description::text, '&', '&', 'g')::jsonb
+    WHERE "name" LIKE '%&%' OR (description->>'en' LIKE '%&%' OR description->>'fr' LIKE '%&%');
+"""
+
+QUERIES = [
+    UPDATE_APPLET_SQL,
+    UPDATE_APPLET_HISTORY_SQL,
+    UPDATE_ACTIVITY_SQL,
+    UPDATE_ACTIVITY_HISTORY_SQL,
+    UPDATE_ACTIVITY_SCORES_AND_REPORTS_SQL,
+    UPDATE_ACTIVITY_HISTORY_SCORES_AND_REPORTS_SQL,
+    UPDATE_ACTIVITY_ITEM_SQL,
+    UPDATE_ACTIVITY_ITEM_HISTORY_SQL,
+    UPDATE_FLOW_SQL,
+    UPDATE_FLOW_HISTORY__SQL,
+]
+
+
+async def main(session: AsyncSession, *args, **kwargs):
+    try:
+        for sql in QUERIES:
+            await session.execute(sql)
+        await session.commit()
+    except Exception as ex:
+        await session.rollback()
+        raise ex

src/apps/shared/domain/custom_validations.py

@@ -153,6 +153,7 @@ def _array_from_string(val):
 
 
 nh3_attributes = deepcopy(nh3.ALLOWED_ATTRIBUTES)
+nh3_rollback = {"&": "&"}
 default_attributes = {
     "id",
     "data-line",
@@ -186,4 +187,6 @@ def _array_from_string(val):
 
 def sanitize_string(value: str) -> str:
     value = nh3.clean(value, attributes=nh3_attributes, link_rel=None)
+    for key in nh3_rollback:
+        value = value.replace(key, nh3_rollback[key])
     return value