-
Notifications
You must be signed in to change notification settings - Fork 7
/
autoredirect.py
192 lines (149 loc) · 6.52 KB
/
autoredirect.py
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
import regex
import argparse
import requests
import os
import threading
currentPath = os.path.dirname(__file__)
CANARY_TEXT = 'CANARY049'
CANARY_DOMAIN = 'canaryredirect.fr'
FUZZ_PLACE_HOLDER = '??????'
TIMEOUT_DELAY = 1.75
LOCK = threading.Lock()
parser = argparse.ArgumentParser()
parser.add_argument("--file", "-f", type=str, required=False, help= 'file of all URLs to be tested against Open Redirect')
parser.add_argument("--url", "-u", type=str, required=False, help= 'url to be tested against Open Redirect')
parser.add_argument("--threads", "-n", type=int, required=False, help= 'number of threads for the tool')
parser.add_argument("--output", "-o", type=str, required=False, help='output file path')
parser.add_argument("--verbose", "-v", action='store_true', help='activate verbose mode for the tool')
parser.add_argument("--smart", "-s", action='store_true', help='activate context-based payload generation for each tested URL')
parser.add_argument("--oneshot", "-t", action='store_true', help='fuzz with only one basic payload - to be activated in case of time constraints')
args = parser.parse_args()
if not (args.file or args.url):
parser.error('No input selected: Please add --file or --url as arguments.')
if args.smart and args.oneshot:
parser.error('Incompatible modes chosen : oneshot mode implies that only one payload is used.')
defaultPayloadFile = open(f"{currentPath}/default-payloads.txt", "r")
if args.oneshot:
payloads = [f"http://{CANARY_DOMAIN}"]
else:
payloads = [payload.replace('\n', '') for payload in defaultPayloadFile]
if args.file :
allURLs = [line.replace('\n','') for line in open(args.file, "r")]
regexParams = regex.compile('(?<=(Url|URL|Open|callback|continue|data|dest|destination|dir|domain|file|file_name|forward|go|goto|host|html|load_file|login\?to|logout|navigation|next|next_page|out|path|redir|redirect|redirect_to|uri|URI|Uri|return|returnTo|return_path|return_to|target|url)=)(.*)(?=(&|$))', flags=regex.IGNORECASE)
if args.output:
output = open(args.output, "w")
else:
output = open("open-redirect-output.txt", "w")
def splitURLS(threadsSize): #Multithreading
splitted = []
URLSsize = len(allURLs)
width = int(URLSsize/threadsSize)
if width == 0:
width = 1
endVal = 0
i = 0
while endVal != URLSsize:
if URLSsize <= i + 2 * width:
if len(splitted) == threadsSize - 2:
endVal = int(i + (URLSsize - i)/2)
else:
endVal = URLSsize
else:
endVal = i + width
splitted.append(allURLs[i: endVal])
i += width
return splitted
def exception_verbose_message(exceptionType):
if args.verbose:
if exceptionType == "timeout":
print("\nTimeout detected... URL skipped")
elif exceptionType == "redirects":
print("\nToo many redirects... URL skipped")
elif exceptionType == "others":
print("\nRequest error... URL skipped")
def smart_extract_host(url, matchedElement):
urlDecodedElem = requests.utils.unquote(matchedElement)
hostExtractorRegex = '(?<=(https|http):\/\/)(.*?)(?=\/)'
extractedHost = regex.search(hostExtractorRegex, urlDecodedElem)
if not extractedHost:
extractedHost = regex.search(hostExtractorRegex, url)
return extractedHost.group()
def generate_payloads(whitelistedHost):
generated =[
f"http://{whitelistedHost}.{CANARY_DOMAIN}", # whitelisted.attacker.com
f"http://{CANARY_DOMAIN}?{whitelistedHost}",
f"http://{CANARY_DOMAIN}/{whitelistedHost}",
f"http://{CANARY_DOMAIN}%ff@{whitelistedHost}",
f"http://{CANARY_DOMAIN}%ff.{whitelistedHost}",
f"http://{whitelistedHost}%25253F@{CANARY_DOMAIN}",
f"http://{whitelistedHost}%253F@{CANARY_DOMAIN}",
f"http://{whitelistedHost}%3F@{CANARY_DOMAIN}",
f"http://{whitelistedHost}@{CANARY_DOMAIN}",
f"http://foo@{CANARY_DOMAIN}:80@{whitelistedHost}",
f"http://foo@{CANARY_DOMAIN}%20@{whitelistedHost}",
f"http://foo@{CANARY_DOMAIN}%09@{whitelistedHost}"
]
return generated
def prepare_url_with_regex(url):
replacedURL = regexParams.sub(FUZZ_PLACE_HOLDER, url)
matchedElem = regexParams.search(url)
if matchedElem:
matchedElem = matchedElem.group()
return replacedURL, matchedElem
def fuzz_open_redirect(url, payloadsList = payloads):
replacedURL, matchedElem = prepare_url_with_regex(url)
if not matchedElem: #No relevant parameter matching
return
if args.smart:
host = smart_extract_host(url, matchedElem)
payloadsList += generate_payloads(host)
if args.verbose:
if not args.threads:
print(f"+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +")
print(f"Starting fuzzing {replacedURL}")
for payload in payloadsList:
if detected_vuln_with_payload(replacedURL, payload):
print(f"Open Redirect detected in {replacedURL} with payload {payload}.")
with LOCK:
output.write(f"Open Redirect detected in {replacedURL} with payload {payload}\n")
return
if args.verbose:
print(f"\nNothing detected for {replacedURL}\n")
def detected_vuln_with_payload(url, payload):
fuzzedUrl = url.replace(FUZZ_PLACE_HOLDER, payload)
if args.verbose:
if not args.threads:
print(f"Testing payload: {payload} ", end="\r")
response = requests.get(fuzzedUrl, timeout=TIMEOUT_DELAY)
return (CANARY_TEXT in response.text)
def sequential_url_scan(urlList):
for url in urlList:
try:
fuzz_open_redirect(url)
except requests.exceptions.Timeout:
exception_verbose_message("timeout")
except requests.exceptions.TooManyRedirects:
exception_verbose_message("redirects")
except requests.exceptions.RequestException:
exception_verbose_message("others")
def main():
if args.url:
try:
fuzz_open_redirect(args.url)
except:
print("\nInvalid URL")
elif args.file:
if not args.threads or args.threads == 1:
sequential_url_scan(allURLs)
else:
workingThreads = []
split = splitURLS(args.threads)
for subList in split:
t = threading.Thread(target=sequential_url_scan, args=[subList])
t.start()
workingThreads.append(t)
for thread in workingThreads:
thread.join()
output.close()
if __name__ == '__main__':
main()