Coverage for graphqler / fuzzer / engine / detectors / sql_injection / sql_injection_detector.py: 57%

1from typing import Type, override 

2 

3import requests 

4import random 

5 

6from graphqler.utils.api import API 

7 

8from ...materializers.getter import Getter 

9from ...materializers.injection_materializer import InjectionMaterializer 

10from ..detector import Detector 

11 

12 

13SQL_INJECTION_STRINGS = [ 

14 '"aaa \' OR 1=1--"', 

15 '"\' OR \'1\'=\'1"', 

16 '"1; DROP TABLE users--"', 

17 '"1\' UNION SELECT null,null,null--"', 

18 '"1\' AND SLEEP(3)--"', 

19 '"\' OR 1=1 LIMIT 1--"', 

20 '"admin\'--"', 

21 '"1\' AND 1=CONVERT(int, (SELECT TOP 1 table_name FROM information_schema.tables))--"', 

22] 

23 

24# Error messages commonly emitted by SQL databases that indicate injection success 

25SQL_ERROR_PATTERNS = [ 

26 "syntax error", 

27 "you have an error in your sql syntax", 

28 "unclosed quotation mark", 

29 "quoted string not properly terminated", 

30 "ora-", 

31 "pg_query", 

32 "mysql_fetch", 

33 "sqlite_", 

34 "sqlstate", 

35 "jdbc", 

36 "odbc", 

37 "invalid query", 

38 "sql syntax", 

39 "unexpected token", 

40 "unterminated string", 

41] 

42 

43 

44class SQLInjectionMaterializer(InjectionMaterializer): 

45 def __init__(self, api: API, fail_on_hard_dependency_not_met: bool = False, max_depth: int = 20): 

46 super().__init__(api, fail_on_hard_dependency_not_met) 

47 self.api = api 

48 self.fail_on_hard_dependency_not_met = fail_on_hard_dependency_not_met 

49 self.getter = SQLInjectionGetter() 

50 

51 

52class SQLInjectionGetter(Getter): 

53 def __init__(self): 

54 super().__init__() 

55 

56 @override 

57 def get_random_string(self, input_name: str) -> str: 

58 if input_name in ['filter', 'search', 'query', 'name', 'username', 'password', 'email', 'id', 'text', 'message', 'input', 'value']: 

59 injection_str = random.choice(SQL_INJECTION_STRINGS) 

60 return injection_str 

61 else: 

62 return super().get_random_string(input_name) 

63 

64 

65# The main class that's being used 

66class SQLInjectionDetector(Detector): 

67 @property 

68 def DETECTION_NAME(self) -> str: 

69 return "SQL Injection (SQLi) Injection" 

70 

71 @property 

72 def detect_only_once_for_api(self) -> bool: 

73 return False 

74 

75 @property 

76 def detect_only_once_for_node(self) -> bool: 

77 return True 

78 

79 @property 

80 def materializer(self) -> Type[SQLInjectionMaterializer]: 

81 return SQLInjectionMaterializer 

82 

83 def _is_vulnerable(self, graphql_response: dict, request_response: requests.Response) -> bool: 

84 response_text_lower = request_response.text.lower() 

85 return any(pattern in response_text_lower for pattern in SQL_ERROR_PATTERNS) 

86 

87 def _is_potentially_vulnerable(self, graphql_response: dict, request_response: requests.Response) -> bool: 

88 if graphql_response is None or 'data' not in graphql_response or graphql_response['data'] is None: 

89 return False 

90 # Flag if the server returned data successfully on an injection payload (possible blind SQLi) 

91 if request_response.status_code == 200 and graphql_response['data'] and any(keyword in self.payload for keyword in SQL_INJECTION_STRINGS): 

92 return True 

93 return False