Fix checkmarx security issues (#227)

Signed-off-by: Lv, Liang1 <liang1.lv@intel.com>

Author: lvliang-intel

intel_extension_for_transformers/neural_chat/pipeline/tools/cut_video.py

@@ -111,7 +111,11 @@ def cut_video(args, outdir):
  parser.add_argument("--sr", type=str, default=16000)
  parser.add_argument("--out_path", type=str, default="../raw")
  args = parser.parse_args()
- 
+
+ # Validate and normalize input and output paths
+ if not os.path.exists(args.path):
+ raise FileNotFoundError(f"Input path '{args.path}' does not exist.")
+
  outdir = os.path.join(shlex.quote(args.path), shlex.quote(args.out_path))
  if not os.path.exists(outdir):
  os.mkdir(outdir)

intel_extension_for_transformers/neural_chat/ui/basic_frontend/fastchat/serve/cli.py

@@ -35,10 +35,24 @@
 
 from fastchat.serve.inference import chat_loop, ChatIO
 
+def is_safe_input(input_text):
+ # Define a regular expression pattern to match safe input
+ safe_pattern = r'^[a-zA-Z0-9\s,.!?]+$'
+ return re.match(safe_pattern, input_text) is not None
 
 class SimpleChatIO(ChatIO):
  def prompt_for_input(self, role) -> str:
- return input(f"{role}: ").strip()
+ query = input(f"{role}: ").strip()
+ # Validate user input
+ if not query:
+ print('Input cannot be empty. Please try again.')
+ return None
+
+ # Perform input validation
+ if not is_safe_input(query):
+ print('Invalid characters in input. Please use only letters, numbers, and common punctuation.')
+ return None
+ return query
 
  def prompt_for_output(self, role: str):
  print(f"{role}: ", end="", flush=True)

workflows/chatbot/demo/basic_frontend/fastchat/serve/cli.py

@@ -18,10 +18,24 @@
 
 from fastchat.serve.inference import chat_loop, ChatIO
 
+def is_safe_input(input_text):
+ # Define a regular expression pattern to match safe input
+ safe_pattern = r'^[a-zA-Z0-9\s,.!?]+$'
+ return re.match(safe_pattern, input_text) is not None
 
 class SimpleChatIO(ChatIO):
  def prompt_for_input(self, role) -> str:
- return input(f"{role}: ").strip()
+ query = input(f"{role}: ").strip()
+ # Validate user input
+ if not query:
+ print('Input cannot be empty. Please try again.')
+ return None
+
+ # Perform input validation
+ if not is_safe_input(query):
+ print('Invalid characters in input. Please use only letters, numbers, and common punctuation.')
+ return None
+ return query
 
  def prompt_for_output(self, role: str):
  print(f"{role}: ", end="", flush=True)

workflows/chatbot/fine_tuning/instruction_tuning_pipeline/finetune_clm.py

@@ -347,7 +347,7 @@ def main():
  parser = HfArgumentParser(
  (ModelArguments, DataArguments, GaudiTrainingArguments, FinetuneArguments)
  )
- if len(sys.argv) == 2 and sys.argv[1].endswith(".json"):
+ if len(sys.argv) == 2 and sys.argv[1].endswith(".json") and os.path.exists(sys.argv[1]):
  # If we pass only one argument to the script and it's the path to a json file,
  # let's parse it to get our arguments.
  model_args, data_args, training_args, finetune_args = parser.parse_json_file(

workflows/chatbot/inference/backend/chat/model_worker.py

@@ -244,22 +244,50 @@ async def talkingbot(request: Request):
 async def api_get_status(request: Request):
  return worker.get_status()
 
+def validate_port(value):
+ try:
+ port = int(value)
+ if 1 <= port <= 65535:
+ return port
+ else:
+ raise argparse.ArgumentTypeError("Port number must be between 1 and 65535.")
+ except ValueError:
+ raise argparse.ArgumentTypeError("Invalid port number. Must be an integer.")
+
+def validate_device(value):
+ valid_devices = ["cpu", "cuda", "mps"]
+ if value in valid_devices:
+ return value
+ else:
+ raise argparse.ArgumentTypeError(f"Invalid device. Must be one of {', '.join(valid_devices)}.")
+
+def validate_limit_model_concurrency(value):
+ if value >= 0:
+ return value
+ else:
+ raise argparse.ArgumentTypeError("Limit model concurrency must be a non-negative integer.")
+
+def validate_stream_interval(value):
+ if value > 0:
+ return value
+ else:
+ raise argparse.ArgumentTypeError("Stream interval must be a positive integer.")
 
 if __name__ == "__main__":
  parser = argparse.ArgumentParser()
  parser.add_argument("--host", type=str, default="0.0.0.0")
- parser.add_argument("--port", type=int, default=8080)
+ parser.add_argument("--port", type=validate_port, default=8080)
  parser.add_argument("--worker-address", type=str,
  default="http://localhost:8080")
  parser.add_argument("--controller-address", type=str,
  default="http://localhost:80")
  parser.add_argument("--model-path", type=str, default="facebook/opt-350m")
  parser.add_argument("--model-name", type=str)
- parser.add_argument("--device", type=str, choices=["cpu", "cuda", "mps"], default="cuda")
+ parser.add_argument("--device", type=validate_device, choices=["cpu", "cuda", "mps"], default="cpu")
  parser.add_argument("--num-gpus", type=int, default=1)
  parser.add_argument("--load-8bit", action="store_true")
- parser.add_argument("--limit-model-concurrency", type=int, default=5)
- parser.add_argument("--stream-interval", type=int, default=2)
+ parser.add_argument("--limit-model-concurrency", type=validate_limit_model_concurrency, default=5)
+ parser.add_argument("--stream-interval", type=validate_stream_interval, default=2)
  parser.add_argument("--no-register", action="store_true")
  parser.add_argument("--ipex", action="store_true")
  parser.add_argument("--itrex", action="store_true")

workflows/chatbot/inference/memory_controller/chat_with_memory.py

@@ -1,4 +1,4 @@
-import os
+import os, re
 from langchain.llms import HuggingFacePipeline
 from langchain.prompts import PromptTemplate
 from langchain.memory import ConversationBufferWindowMemory
@@ -45,6 +45,10 @@ def inference(args, query, memory):
  print("inference cost {} seconds.".format(end_time - start_time))
  return result, memory
 
+def is_safe_input(input_text):
+ # Define a regular expression pattern to match safe input
+ safe_pattern = r'^[a-zA-Z0-9\s,.!?]+$'
+ return re.match(safe_pattern, input_text) is not None
 
 if __name__ == "__main__":
 
@@ -63,7 +67,7 @@ def inference(args, query, memory):
  "max_length": args.max_length,
  "device_map": "auto",
  "repetition_penalty": args.penalty,
- }
+ })
  if args.memory_type == "buffer_window":
  memory = ConversationBufferWindowMemory(memory_key="chat_history", k=3)
  elif args.memory_type == "buffer":
@@ -74,8 +78,19 @@ def inference(args, query, memory):
 
  while True:
  query = input("Enter input (or 'exit' to quit): ").strip()
- if query == 'exit':
+ if query.lower() == 'exit':
  print('exit')
  break
+
+ # Validate user input
+ if not query:
+ print('Input cannot be empty. Please try again.')
+ continue
+
+ # Perform input validation
+ if not is_safe_input(query):
+ print('Invalid characters in input. Please use only letters, numbers, and common punctuation.')
+ continue
+
  result, memory = inference(args, query, memory)
  print("Input:" + query + '\nResponse:' + result + '\n')