security: replace unsafe eval() with ast.literal_eval()

backend/database/redis_db.py

@@ -1,6 +1,7 @@
 import base64
 import json
 import os
+import ast
 from typing import List, Union, Optional
 from datetime import datetime, timedelta, timezone
 
@@ -114,7 +115,7 @@ def get_app_usage_count_cache(app_id: str) -> int | None:
     count = r.get(f'apps:{app_id}:usage_count')
     if not count:
         return None
-    return eval(count)
+    return ast.literal_eval(count.decode())
 
 
 def set_app_money_made_amount_cache(app_id: str, amount: float):
@@ -125,7 +126,7 @@ def get_app_money_made_amount_cache(app_id: str) -> float | None:
     amount = r.get(f'apps:{app_id}:money_made')
     if not amount:
         return None
-    return eval(amount)
+    return ast.literal_eval(amount.decode())
 
 
 def set_app_usage_history_cache(app_id: str, usage: List[dict]):
@@ -161,7 +162,7 @@ def set_app_review_cache(app_id: str, uid: str, data: dict):
     if not reviews:
         reviews = {}
     else:
-        reviews = eval(reviews)
+        reviews = ast.literal_eval(reviews.decode())
     reviews[uid] = data
     r.set(f'plugins:{app_id}:reviews', str(reviews))
 
@@ -170,7 +171,7 @@ def get_specific_user_review(app_id: str, uid: str) -> dict:
     reviews = r.get(f'plugins:{app_id}:reviews')
     if not reviews:
         return {}
-    reviews = eval(reviews)
+    reviews = ast.literal_eval(reviews.decode())
     return reviews.get(uid, {})
 
 
@@ -221,7 +222,7 @@ def get_app_reviews(app_id: str) -> dict:
     reviews = r.get(f'plugins:{app_id}:reviews')
     if not reviews:
         return {}
-    return eval(reviews)
+    return ast.literal_eval(reviews.decode())
 
 
 def get_apps_reviews(app_ids: list) -> dict:
@@ -232,7 +233,7 @@ def get_apps_reviews(app_ids: list) -> dict:
     reviews = r.mget(keys)
     if reviews is None:
         return {}
-    return {app_id: eval(review) if review else {} for app_id, review in zip(app_ids, reviews)}
+    return {app_id: ast.literal_eval(review.decode()) if review else {} for app_id, review in zip(app_ids, reviews)}
 
 
 def set_app_installs_count(app_id: str, count: int):
@@ -284,7 +285,7 @@ def get_cached_user_geolocation(uid: str):
     geolocation = r.get(f'users:{uid}:geolocation')
     if not geolocation:
         return None
-    return eval(geolocation)
+    return ast.literal_eval(geolocation.decode())
 
 
 # VISIIBILTIY OF CONVERSATIONS

backend/migrations/002_populate_historical_usage.py

@@ -29,7 +29,7 @@
         # This path is for Modal environment
         service_account_info = os.environ["SERVICE_ACCOUNT_JSON"]
         cred = credentials.Certificate(
-            eval(service_account_info) if service_account_info.startswith('{') else service_account_info
+            json.loads(service_account_info) if service_account_info.startswith('{') else service_account_info
         )
     else:
         # This path is for local development, GOOGLE_APPLICATION_CREDENTIALS should be set

backend/migrations/003_report_top_transcription_users.py

@@ -1,5 +1,6 @@
 import argparse
 import csv
+import json
 import os
 import sys
 from concurrent.futures import ThreadPoolExecutor, as_completed
@@ -20,7 +21,7 @@
         # This path is for Modal environment
         service_account_info = os.environ["SERVICE_ACCOUNT_JSON"]
         cred = credentials.Certificate(
-            eval(service_account_info) if service_account_info.startswith('{') else service_account_info
+            json.loads(service_account_info) if service_account_info.startswith('{') else service_account_info
         )
     else:
         # This path is for local development, GOOGLE_APPLICATION_CREDENTIALS should be set

plugins/db.py

@@ -1,4 +1,5 @@
 import os
+import ast
 from typing import List
 
 import redis
@@ -58,7 +59,7 @@ def append_segment_to_transcript(uid: str, session_id: str, new_segments: list[T
     if not segments:
         segments = []
     else:
-        segments = eval(segments)
+        segments = ast.literal_eval(segments.decode())
 
     segments.extend([segment.dict() for segment in new_segments])
 
@@ -169,7 +170,7 @@ def get_upsert_segment_to_transcript_plugin(
     if not segments:
         segments = []
     else:
-        segments = eval(segments)
+        segments = ast.literal_eval(segments.decode())
 
     segments.extend([segment.dict() for segment in new_segments])