1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
|
import argparse
import importlib
import sys
import json
from jinja2 import TemplateError
from jinja2.sandbox import SandboxedEnvironment
from types import ModuleType, FunctionType, MethodType
from contextlib import contextmanager
def safe_import(module_name):
try:
module = importlib.import_module(module_name)
return module, None
except ImportError as e:
return None, f"Failed to import module '{module_name}': {str(e)}"
def enumerate_object(obj, depth=0, max_depth=3, path=None, visited=None):
if path is None:
path = []
if visited is None:
visited = set()
obj_id = id(obj)
if obj_id in visited:
return None, "Cycle detected"
visited.add(obj_id)
try:
result = {
'path': '.'.join(path),
'attributes': dir(obj)[:50],
'globals': list(getattr(obj, '__globals__', {}).keys())[:50] if hasattr(obj, '__globals__') else None,
'subclasses': [cls.__name__ for cls in obj.__subclasses__()][:50] if hasattr(obj, '__subclasses__') else None
}
if depth < max_depth:
result['nested'] = {}
skip_attrs = {'__class__', '__base__', '__bases__', '__mro__'}
for attr in dir(obj)[:10]:
if attr in skip_attrs:
continue
try:
value = getattr(obj, attr)
if isinstance(value, (ModuleType, type)) or hasattr(value, '__dict__') or hasattr(value, '__globals__'):
nested_result, error = enumerate_object(value, depth + 1, max_depth, path + [attr], visited)
result['nested'][attr] = nested_result if nested_result else 'Inaccessible'
except:
result['nested'][attr] = 'Inaccessible'
return result, None
except Exception as e:
return None, f"Error enumerating object: {str(e)}"
def discover_rce_vectors(context, framework="jinja2"):
dangerous_modules = [
'os', 'subprocess', 'sys', 'shutil', 'platform', 'ctypes', 'pickle', 'code', 'builtins',
'threading', 'multiprocessing', 'socket', 'webbrowser', 'tempfile', 'asyncio', 'signal', 'popen2'
]
dangerous_functions = [
'popen', 'system', 'exec', 'eval', 'execfile', 'compile', 'open', 'execvp', 'execl', 'execle',
'execv', 'execlp', 'call', 'run', 'communicate', 'load', 'getattr', 'setattr', '__import__',
'check_call', 'check_output', 'getoutput', 'getstatusoutput'
]
dangerous_classes = [
'Popen', '_wrap_close', 'InteractiveConsole', 'CodeType', 'Process', 'Thread',
'TCPServer', 'UDPServer', 'AsyncIOEventLoop'
]
execution_keywords = ['exec', 'run', 'call', 'eval', 'system', 'load', 'spawn', 'fork']
rce_vectors = []
def check_path(obj, obj_name, path=None):
if path is None:
path = [obj_name]
try:
enumeration, _ = enumerate_object(obj, path=path)
if not enumeration:
return
if enumeration.get('globals'):
for mod in enumeration['globals']:
if mod in dangerous_modules:
rce_vectors.append({
'path': '.'.join(path + ['__globals__', mod]),
'type': 'potentially dangerous module, investigate manually',
'details': f"access to '{mod}' module"
})
for attr in enumeration['attributes']:
if attr in dangerous_functions or any(keyword in attr.lower() for keyword in execution_keywords):
try:
value = getattr(obj, attr)
if isinstance(value, (FunctionType, MethodType)):
rce_vectors.append({
'path': '.'.join(path + [attr]),
'type': 'potentially dangerous function, investigate manually',
'details': f"access to '{attr}' function"
})
except:
pass
if enumeration.get('subclasses'):
for cls in enumeration['subclasses']:
if cls in dangerous_classes or any(keyword in cls.lower() for keyword in execution_keywords):
rce_vectors.append({
'path': '.'.join(path + ['__subclasses__', cls]),
'type': 'potentially dangerous class, investigate manually',
'details': f"access to {cls}"
})
for attr, nested in enumeration.get('nested', {}).items():
if nested != 'Inaccessible':
try:
value = getattr(obj, attr)
check_path(value, attr, path + [attr])
except:
pass
except:
pass
for key, value in context.items():
if value is not None:
check_path(value, key)
return rce_vectors
@contextmanager
def flask_context():
from flask import Flask, request
app = Flask(__name__)
with app.test_request_context():
yield {
'request': request,
'config': app.config,
'self': None
}
@contextmanager
def django_context():
from django.http import HttpRequest
from django.test import RequestFactory
from django.conf import settings
if not settings.configured:
settings.configure(DEFAULT_CHARSET='utf-8')
factory = RequestFactory()
request = factory.get('/')
yield {
'request': request,
'self': None
}
def main():
parser = argparse.ArgumentParser(description="SSTI RCE Vector Discovery Tool")
parser.add_argument('--module', required=True, help="Module to import (e.g., os, numpy, myutils)")
parser.add_argument('--framework', choices=['jinja2', 'django'], default='jinja2',
help="Template framework to simulate (jinja2 or django)")
parser.add_argument('--output', help="Output file for results (default: console)")
args = parser.parse_args()
module, import_error = safe_import(args.module)
if import_error:
print(f"Error: {import_error}")
sys.exit(1)
context = {'utils': module}
rce_vectors = []
if args.framework == 'jinja2':
try:
with flask_context() as flask_ctx:
context.update(flask_ctx)
env = SandboxedEnvironment()
env.globals.update(context)
rce_vectors = discover_rce_vectors(env.globals, framework="jinja2")
except ImportError:
print("Error: Flask not installed. Run 'pip install flask'")
sys.exit(1)
else:
try:
from django.template import Template, Context
with django_context() as django_ctx:
context.update(django_ctx)
rce_vectors = discover_rce_vectors(context, framework="django")
except ImportError:
print("Error: Django not installed. Run 'pip install django'")
sys.exit(1)
results = {
'module': args.module,
'framework': args.framework,
'rce_vectors': rce_vectors
}
output = json.dumps(results, indent=2)
if args.output:
with open(args.output, 'w') as f:
f.write(output)
print(f"Results written to {args.output}")
else:
print(output)
if __name__ == '__main__':
main()
|