CVE-2022-1806: Improper Neutralization of Input During Web Page Generation ('Cross-site Scripting') (CWE-79)
value: Cross-site Scripting (XSS) - Reflected in GitHub repository rtxteam/rtx prior to checkpoint_2022-05-18.
CWE description: The product does not neutralize or incorrectly neutralizes user-controllable input before it is placed in output that is used as a web page that is served to other users.
CVSS v3 base score: 6.1
Published: 2022-05-20T09:15Z
Repository: https://github.com/rtxteam/rtx
Language: Python
Files changed: code/autocomplete/server.py
Commit date: 2022-05-18 03:32:13 +0000
Commit message: remove some cruft and implement a sanitizer for the client-supplied callback function name

Diff: diff --git a/code/autocomplete/server.py b/code/autocomplete/server.py
index c27afa667..f6d792164 100644
--- a/code/autocomplete/server.py
+++ b/code/autocomplete/server.py
@@ -4,36 +4,39 @@ import os
 #import sqlite3
 import json
 import sys
 import rtxcomplete
 import traceback
-
-#class MainHandler(tornado.web.RequestHandler):
-#    def get(self):
-#        self.write("Hello, world")
-#print __file__
+import re
 
 root = os.path.dirname(os.path.abspath(__file__))
 rtxcomplete.load()
-#conn = sqlite3.connect('dict.db')
-#conn.enable_load_extension(True)
-#conn.load_extension("./spellfix")
-#cursor = conn.cursor()
+
+
+#### Sanitize the client-provided callback function name
+def sanitize_callback(callback):
+    if callback is None or not isinstance(callback,str):
+        return 'autocomplete_callback'
+    match = re.match(r'([a-zA-Z0-9_]+).*$', callback)
+    if match:
+         callback = match.group(1)
+    else:
+         callback = 'autocomplete_callback'
+    return callback
+
 
 class autoSearch(tornado.web.RequestHandler):
 
     def get(self, arg,word=None):
-        #print "match auto"
         try:
             limit = self.get_argument("limit")
             word = self.get_argument("word")
-            callback = self.get_argument("callback") #jsonp
+            callback = sanitize_callback(self.get_argument("callback"))
 
             result = rtxcomplete.prefix(word,limit)
-
-            result = callback+"("+json.dumps(result)+");" #jsonp
-            #result = json.dumps(result) #typeahead
+            
+            result = callback+"("+json.dumps(result)+");"
             
             self.write(result)
             
         except:
             print(sys.exc_info()[:])
@@ -45,11 +48,11 @@ class fuzzySearch(tornado.web.RequestHandler):
     def get(self, arg,word=None):
         #print "matched fuzzy"
         try:
             limit = self.get_argument("limit")
             word = self.get_argument("word")
-            callback = self.get_argument("callback")
+            callback = sanitize_callback(self.get_argument("callback"))
             #print word
             #cursor.execute("SELECT word FROM spell WHERE word MATCH \"" + word + "\" LIMIT " + limit)
             #cursor.execute("SELECT word FROM spell WHERE word MATCH \"" + word + "*\" LIMIT " + limit)
             result = rtxcomplete.fuzzy(word,limit);
             #rows = cursor.fetchall()
@@ -71,11 +74,11 @@ class autofuzzySearch(tornado.web.RequestHandler):
     def get(self, arg,word=None):
         #print "matched autofuzzy"
         try:
             limit = self.get_argument("limit")
             word = self.get_argument("word")
-            callback = self.get_argument("callback")
+            callback = sanitize_callback(self.get_argument("callback"))
             #print word
             #cursor.execute("SELECT word FROM spell WHERE word MATCH \"" + word + "\" LIMIT " + limit)
             #cursor.execute("SELECT word FROM spell WHERE word MATCH \"" + word + "*\" LIMIT " + limit)
             result = rtxcomplete.autofuzzy(word,limit);
             #rows = cursor.fetchall()
@@ -94,22 +97,21 @@ class autofuzzySearch(tornado.web.RequestHandler):
             self.write("error")
 
 
 class nodesLikeSearch(tornado.web.RequestHandler):
     def get(self, arg,word=None):
-        #try:
-        if 1 == 1:
+        try:
             limit = self.get_argument("limit")
             word = self.get_argument("word")
-            callback = self.get_argument("callback")
+            callback = sanitize_callback(self.get_argument("callback"))
             result = rtxcomplete.get_nodes_like(word,limit);
             result = callback+"("+json.dumps(result)+");"
             self.write(result)
-        #except:
-        #    print(sys.exc_info()[:])
-        #    traceback.print_tb(sys.exc_info()[-1])
-        #    self.write("error")
+        except:
+            print(sys.exc_info()[:])
+            traceback.print_tb(sys.exc_info()[-1])
+            self.write("error")
 
 
 class defineSearch(tornado.web.RequestHandler):
     def get(self, arg,word=None):
         print("matched define search: not implemented")

Vulnerable code: 
#class MainHandler(tornado.web.RequestHandler):
#    def get(self):
#        self.write("Hello, world")
#print __file__
#conn = sqlite3.connect('dict.db')
#conn.enable_load_extension(True)
#conn.load_extension("./spellfix")
#cursor = conn.cursor()
        #print "match auto"
            callback = self.get_argument("callback") #jsonp

            result = callback+"("+json.dumps(result)+");" #jsonp
            #result = json.dumps(result) #typeahead
            callback = self.get_argument("callback")
            callback = self.get_argument("callback")
        #try:
        if 1 == 1:
            callback = self.get_argument("callback")
        #except:
        #    print(sys.exc_info()[:])
        #    traceback.print_tb(sys.exc_info()[-1])
        #    self.write("error")
Fixed code: import re


#### Sanitize the client-provided callback function name
def sanitize_callback(callback):
    if callback is None or not isinstance(callback,str):
        return 'autocomplete_callback'
    match = re.match(r'([a-zA-Z0-9_]+).*$', callback)
    if match:
         callback = match.group(1)
    else:
         callback = 'autocomplete_callback'
    return callback

            callback = sanitize_callback(self.get_argument("callback"))
            
            result = callback+"("+json.dumps(result)+");"
            callback = sanitize_callback(self.get_argument("callback"))
            callback = sanitize_callback(self.get_argument("callback"))
        try:
            callback = sanitize_callback(self.get_argument("callback"))
        except:
            print(sys.exc_info()[:])
            traceback.print_tb(sys.exc_info()[-1])
            self.write("error")