From c0fde3686df747460993e84a8367479a564dba79 Mon Sep 17 00:00:00 2001 From: Vinicius Moreira Date: Mon, 2 May 2022 15:20:14 -0300 Subject: [PATCH] [core] improvement: preventing command injection through the search mechanism --- CHANGELOG.md | 3 +++ bauh/commons/util.py | 18 ++++++++++++++ bauh/view/core/controller.py | 10 ++++---- tests/common/test_util.py | 46 +++++++++++++++++++++++++++++++++++- 4 files changed, 72 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 491a05d1..71581e15 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/bauh/commons/util.py b/bauh/commons/util.py index 76beacfe..a8e6e8df 100644 --- a/bauh/commons/util.py +++ b/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() diff --git a/bauh/view/core/controller.py b/bauh/view/core/controller.py index 44a93cc2..c4d8b06a 100755 --- a/bauh/view/core/controller.py +++ b/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) diff --git a/tests/common/test_util.py b/tests/common/test_util.py index 3e9ecd0f..22086bd0 100644 --- a/tests/common/test_util.py +++ b/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)