[core] improvement: preventing command injection through the search mechanism

CHANGELOG.md

@@ -7,6 +7,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 ## [0.10.3]
 
 ### Improvements
+- General
+  - preventing command injection through the search mechanism [#266](https://github.com/vinifmor/bauh/issues/266)
+  
 - UI
   - manage window minimum width related to the table columns 
   - table columns width for maximized window

bauh/commons/util.py

@@ -1,9 +1,14 @@
 import logging
+import re
 from abc import ABC
 from datetime import datetime
 from logging import Logger
 from typing import Optional, Union
 
+re_command_forbidden_symbols = re.compile(r'[\'\"%$#*<>|&]')
+re_several_spaces = re.compile(r'\s+')
+re_command_parameter = re.compile(r'(^|\s)-+\w+')
+
 
 class NullLoggerFactory(ABC):
 
@@ -68,3 +73,16 @@ def datetime_as_milis(date: datetime = datetime.utcnow()) -> int:
 def map_timestamp_file(file_path: str) -> str:
     path_split = file_path.split('/')
     return '/'.join(path_split[0:-1]) + '/' + path_split[-1].split('.')[0] + '.ts'
+
+
+def sanitize_command_input(input_: str) -> str:
+    final_input = input_
+
+    for op in ('|', '&'):
+        final_input = final_input.split(op)[0]
+
+    for remove_re in (re_command_forbidden_symbols, re_command_parameter):
+        final_input = remove_re.sub('', final_input)
+
+    final_input = re_several_spaces.sub(' ', final_input)
+    return final_input.strip()

bauh/view/core/controller.py

@@ -4,7 +4,7 @@ import time
 import traceback
 from subprocess import Popen, STDOUT
 from threading import Thread
-from typing import List, Set, Type, Tuple, Dict, Optional, Generator, Callable
+from typing import List, Set, Type, Tuple, Dict, Optional, Generator, Callable, Pattern
 
 from bauh.api.abstract.controller import SoftwareManager, SearchResult, ApplicationContext, UpgradeRequirements, \
     UpgradeRequirement, TransactionResult, SoftwareAction, SettingsView, SettingsController
@@ -16,6 +16,7 @@ from bauh.api.abstract.view import ViewComponent, TabGroupComponent, MessageType
 from bauh.api.exception import NoInternetException
 from bauh.commons.boot import CreateConfigFile
 from bauh.commons.html import bold
+from bauh.commons.util import sanitize_command_input
 from bauh.view.core.config import CoreConfigManager
 from bauh.view.core.settings import GenericSettingsManager
 from bauh.view.core.update import check_for_update
@@ -155,16 +156,17 @@ class GenericSoftwareManager(SoftwareManager, SettingsController):
         res = SearchResult.empty()
 
         if self.context.is_internet_available():
-            norm_word = words.strip().lower()
+            norm_query = sanitize_command_input(words).lower()
+            self.logger.info(f"Search query: {norm_query}")
 
-            is_url = bool(RE_IS_URL.match(norm_word))
+            is_url = bool(RE_IS_URL.match(norm_query))
             disk_loader = self.disk_loader_factory.new()
             disk_loader.start()
 
             threads = []
 
             for man in self.managers:
-                t = Thread(target=self._search, args=(norm_word, is_url, man, disk_loader, res))
+                t = Thread(target=self._search, args=(norm_query, is_url, man, disk_loader, res))
                 t.start()
                 threads.append(t)
 

tests/common/test_util.py

@@ -1,6 +1,7 @@
+import time
 from unittest import TestCase
 
-from bauh.commons.util import size_to_byte
+from bauh.commons.util import size_to_byte, sanitize_command_input
 
 
 class SizeToByteTest(TestCase):
@@ -46,3 +47,46 @@ class SizeToByteTest(TestCase):
 
     def test_must_treat_string_sizes_before_converting(self):
         self.assertEqual(57300, size_to_byte(' 57 , 3  ', ' K '))
+
+
+class SanitizeCommandInputTest(TestCase):
+
+    def test__must_remove_any_forbidden_symbols(self):
+        input_ = ' #$%* abc-def@ #%<<>> '
+        res = sanitize_command_input(input_)
+        self.assertEqual('abc-def@', res)
+
+    def test__must_remove_single_quotes(self):
+        input_ = " 'abc'-'xpto' "
+        res = sanitize_command_input(input_)
+        self.assertEqual('abc-xpto', res)
+
+    def test__must_remove_double_quotes(self):
+        input_ = ' "abc"-"xpto" '
+        res = sanitize_command_input(input_)
+        self.assertEqual('abc-xpto', res)
+
+    def test__must_remove_the_pipe_operator(self):
+        input_ = '  abc | ls /home/xpto | cat /home/xpto/secret.txt'
+        res = sanitize_command_input(input_)
+        self.assertEqual('abc', res)
+
+    def test__must_remove_the_and_operator(self):
+        input_ = '  abc && ls /home/xpto & cat /home/xpto/secret.txt'
+        res = sanitize_command_input(input_)
+        self.assertEqual('abc', res)
+
+    def test__must_remove_several_operator(self):
+        input_ = '  abc | ls /home/xpto && cat /home/xpto/secret.txt'
+        res = sanitize_command_input(input_)
+        self.assertEqual('abc', res)
+
+    def test__must_remove_single_dash_parameters(self):
+        input_ = '-cat abc-def -user -system ghi--jkl -xpto '
+        res = sanitize_command_input(input_)
+        self.assertEqual('abc-def ghi--jkl', res)
+
+    def test__must_remove_double_dash_parameters(self):
+        input_ = '--cat abc-def --user -system ghi--jkl --xpto '
+        res = sanitize_command_input(input_)
+        self.assertEqual('abc-def ghi--jkl', res)