-
Notifications
You must be signed in to change notification settings - Fork 581
/
BBScan.py
1147 lines (1019 loc) · 50.7 KB
/
BBScan.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
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
A fast and light-weight web vulnerability scanner. It helps pen-testers pinpoint possibly vulnerable targets from a large number of web servers.
https://github.com/lijiejie/BBScan
Li JieJie my[at]lijiejie.com https://www.lijiejie.com
"""
import os
# first, change working dir
cur_dir = os.path.dirname(os.path.abspath(__file__))
os.chdir(cur_dir)
import sys
import codecs
import asyncio
import httpx
import re
from bs4 import BeautifulSoup
import warnings
import time
import glob
import ipaddress
import ssl
import traceback
import importlib
import copy
import string
import random
import dns.asyncresolver
from urllib.parse import urlparse
from lib.common import clear_queue, parse_url, cal_depth, get_domain_sub, is_port_open, scan_given_ports, \
is_ip_addr, get_dns_resolver, get_http_title, clear_url
from lib.cmdline import parse_args
from lib.report import save_report
import lib.config as conf
from lib.cms_fingerprints import Fingerprint
import hashlib
from lib.javascript_parser import get_urls_in_js_async
if hasattr(ssl, '_create_unverified_context'):
ssl._create_default_https_context = ssl._create_unverified_context
from bs4 import MarkupResemblesLocatorWarning
warnings.filterwarnings('ignore', category=MarkupResemblesLocatorWarning)
fingerprint = Fingerprint()
class Scanner(object):
def __init__(self, timeout=900):
self.q_results = q_results
self.args = args
self.start_time = time.time()
self.time_out = timeout
self.links_limit = 100 # max number of folders allowed to scan
async def init(self):
await self._init_rules()
self._init_scripts()
self.url_queue = asyncio.Queue() # all urls to scan
self.urls_processed = set() # processed urls
self.urls_enqueued = set() # entered queue urls
self.urls_crawled = set()
self.lock = asyncio.Lock()
self.results = {}
self.log_file = None
self._404_status = -1
self.conn_pool = None
self.index_status, self.index_headers, self.index_html_doc = None, {}, ''
self.scheme, self.host, self.port, self.path = None, None, None, None
self.domain_sub = ''
self.base_url = ''
self.max_depth = 0
self.len_404_doc = 0
self.has_http = None
self.ports_open = None
self.ports_closed = None
self.no_scripts = None
self.status_502_count = 0
self.timeout_count = 0
self.timeout_scan_aborted = False
self.fingerprint_check = True
self.js_urls = []
self.index_has_reported = False
self.urls_regex_found = set()
async def print_msg(self, msg):
await self.q_results.put(msg)
def reset_scanner(self):
self.start_time = time.time()
clear_queue(self.url_queue)
self.urls_processed.clear()
self.urls_enqueued.clear()
self.urls_crawled.clear()
self.results.clear()
self.log_file = None
self._404_status = -1
# self.conn_pool = None # Bug Fixed, shouldn't set to None right here, used pool can not be closed
self.index_status, self.index_headers, self.index_html_doc = None, {}, ''
self.scheme, self.host, self.port, self.path = None, None, None, None
self.domain_sub = ''
self.base_url = ''
self.status_502_count = 0
self.timeout_count = 0
self.timeout_scan_aborted = False
self.fingerprint_check = True
self.js_urls = []
self.index_has_reported = False
self.urls_regex_found = set()
# scan from a given URL
async def init_from_url(self, target):
self.reset_scanner()
self.scheme = target['scheme']
self.host = target['host']
self.port = target['port']
self.path = target['path']
self.has_http = target['has_http']
self.ports_open = target['ports_open']
self.ports_closed = target['ports_closed']
self.no_scripts = target['no_scripts'] if 'no_scripts' in target else 0
self.domain_sub = get_domain_sub(self.host)
await self.init_final()
return True
# Fix me: not yet implemented and tested 2024-05-27
async def init_from_log_file(self, log_file):
self.reset_scanner()
self.log_file = log_file
self.scheme, self.host, self.path = self._parse_url_from_file()
self.domain_sub = get_domain_sub(self.host)
if self.host:
if self.host.find(':') > 0:
_ret = self.host.split(':')
self.host = _ret[0]
self.port = _ret[1]
elif self.scheme == 'https':
self.port = 443
elif self.scheme == 'http':
self.port = 80
else:
self.port = None
if await is_port_open(self.host, self.port):
await self.print_msg('[Port Not Open] %s:%s' % (self.host, self.port))
return False
self.has_http = True
self.no_scripts = 1
await self.init_final()
await self.load_all_urls_from_log_file()
return True
else:
host = os.path.basename(log_file).replace('.log', '')
try:
await dns.asyncresolver.resolve(host, "A")
await self.init_from_url(host) # Fix Me
return True
except Exception as e:
await self.print_msg('[ERROR] Invalid host from log name: %s' % host)
return False
async def init_final(self):
try:
if self.conn_pool:
await self.conn_pool.aclose()
except Exception as e:
await self.print_msg('conn_pool.aclose exception: %s' % str(e))
self.conn_pool = None # after close
if self.scheme == 'http' and self.port == 80 or self.scheme == 'https' and self.port == 443:
self.base_url = '%s://%s' % (self.scheme, self.host)
else:
self.base_url = '%s://%s:%s' % (self.scheme, self.host, self.port)
if self.has_http:
await self.print_msg('Scan %s' % self.base_url)
else:
await self.print_msg('Scan %s:%s' % (self.host, self.port) if self.port else 'Scan %s' % self.host)
if self.has_http:
limits = httpx.Limits(max_connections=100, max_keepalive_connections=40)
self.conn_pool = httpx.AsyncClient(headers=conf.default_headers,
proxies=args.proxy, verify=False, limits=limits, follow_redirects=False)
if self.args.require_index_doc:
await self.crawl('/', do_not_process_links=True)
if self.no_scripts != 1: # 不是重复目标 80 443 跳转的，不需要重复扫描
# 当前目标disable， 或者 全局开启插件扫描
if self.args.scripts_only or not self.no_scripts:
for _ in self.user_scripts:
await self.url_queue.put((_, '/'))
if not self.has_http or self.args.scripts_only: # 未发现HTTP服务 或 只依赖插件扫描
return
self.max_depth = cal_depth(self, self.path)[1] + 5
if self.args.no_check404:
self._404_status = 404
else:
await self.check_404_existence()
if self._404_status == -1:
await self.print_msg('[Warning] HTTP 404 check failed: %s' % self.base_url)
# elif self._404_status != 404:
# await self.print_msg('[Warning] %s has no HTTP 404.' % self.base_url)
_path, _depth = cal_depth(self, self.path)
await self.enqueue('/')
if _path != '/' and not self.log_file:
await self.enqueue(_path)
def _parse_url_from_file(self):
url = ''
with open(self.log_file) as infile:
for _line in infile.readlines():
_line = _line.strip()
if _line and len(_line.split()) >= 3:
url = _line.split()[1]
break
return parse_url(url)
# load urls from rules/*.txt
async def _init_rules(self):
self.text_to_find = []
self.regex_to_find = []
self.text_to_exclude = []
self.regex_to_exclude = []
self.rules_set = set()
self.rules_set_root_only = set()
p_tag = re.compile('{tag="(.*?)"}')
p_status = re.compile(r'{status=(\d{3})}')
p_content_type = re.compile('{type="(.*?)"}')
p_content_type_no = re.compile('{type_no="(.*?)"}')
_files = self.args.rule_files if self.args.rule_files else glob.glob('rules/*.txt')
if self.args.fingerprint_only:
_files = []
for rule_file in _files:
with codecs.open(rule_file, 'r', encoding='utf-8') as infile:
vul_type = os.path.basename(rule_file)[:-4]
for url in infile.readlines():
url = url.strip()
if url.startswith('/'):
_ = p_tag.search(url)
tag = _.group(1) if _ else ''
_ = p_status.search(url)
status = int(_.group(1)) if _ else 0
_ = p_content_type.search(url)
content_type = _.group(1) if _ else ''
_ = p_content_type_no.search(url)
content_type_no = _.group(1) if _ else ''
root_only = True if url.find('{root_only}') >= 0 else False
rule = (url.split()[0], tag, status, content_type, content_type_no, root_only, vul_type)
if root_only:
if rule not in self.rules_set_root_only:
self.rules_set_root_only.add(rule)
else:
await self.print_msg('Duplicated root only rule: %s' % str(rule))
else:
if rule not in self.rules_set:
self.rules_set.add(rule)
else:
await self.print_msg('Duplicated rule: %s' % str(rule))
re_text = re.compile('{text="(.*)"}')
re_regex_text = re.compile('{regex_text="(.*)"}')
file_path = 'rules/white.list'
if not os.path.exists(file_path):
await self.print_msg('[ERROR] File not exist: %s' % file_path)
return
for _line in codecs.open(file_path, encoding='utf-8'):
_line = _line.strip()
if not _line or _line.startswith('#'):
continue
_m = re_text.search(_line)
if _m:
self.text_to_find.append(_m.group(1))
else:
_m = re_regex_text.search(_line)
if _m:
self.regex_to_find.append(re.compile(_m.group(1)))
file_path = 'rules/black.list'
if not os.path.exists(file_path):
await self.print_msg('[ERROR] File not exist: %s' % file_path)
return
for _line in codecs.open(file_path, encoding='utf-8'):
_line = _line.strip()
if not _line or _line.startswith('#'):
continue
_m = re_text.search(_line)
if _m:
self.text_to_exclude.append(_m.group(1))
else:
_m = re_regex_text.search(_line)
if _m:
self.regex_to_exclude.append(re.compile(_m.group(1)))
def _init_scripts(self):
self.user_scripts = []
if self.args.no_scripts: # 全局禁用插件，无需导入
return
files = 'scripts/*.py'
if self.args.fingerprint_only:
files = 'scripts/is_admin_site.py'
for _script in glob.glob(files):
script_name_origin = os.path.basename(_script)
script_name = script_name_origin.replace('.py', '')
if self.args.script: # 只导入指定的脚本
if script_name not in self.args.script and script_name_origin not in self.args.script:
continue
if script_name.startswith('_'):
continue
try:
self.user_scripts.append(importlib.import_module('scripts.%s' % script_name))
except Exception as e:
print('[ERROR] Fail to load script %s' % script_name)
async def http_request(self, url, headers=conf.default_headers, timeout=30, follow_redirects=False):
try:
if not url:
url = '/'
if not self.conn_pool or self.timeout_scan_aborted:
return -1, {}, ''
if self.args.debug:
await self.print_msg('--> %s' % self.base_url + url)
resp = await self.conn_pool.get(self.base_url + url,
headers=headers, follow_redirects=follow_redirects, timeout=timeout)
if resp.headers.get('content-type', '').find('text') >= 0 \
or resp.headers.get('content-type', '').find('html') >= 0 \
or int(resp.headers.get('content-length', '0')) <= 20480: # 1024 * 20
html_doc = resp.text
else:
html_doc = ''
if resp.status_code == 502: # 502出现超过3次，排除该站点不再扫描
self.status_502_count += 1
if self.status_502_count > 3:
self.timeout_scan_aborted = True
clear_queue(self.url_queue)
try:
if self.conn_pool:
await self.conn_pool.aclose()
except Exception as e:
pass #
self.conn_pool = None
if self.args.debug:
await self.print_msg('Website 502 exceeded: %s' % self.base_url)
return resp.status_code, resp.headers, html_doc
except httpx.ReadTimeout as e:
self.timeout_count += 1
if self.timeout_count >= 3:
if not self.timeout_scan_aborted:
self.timeout_scan_aborted = True
await self.print_msg('[Warning] timeout exceeded, scan aborted: %s' % self.base_url)
clear_queue(self.url_queue)
return -1, {}, ''
except (httpx.RequestError, httpx.HTTPStatusError, ssl.SSLError) as e:
if self.args.debug:
await self.print_msg('[Request Error] %s %s %s' % (type(e), str(e), self.base_url))
return -1, {}, ''
except Exception as e:
if self.args.debug:
await self.print_msg('[Request Error] %s %s %s' % (type(e), str(e), self.base_url))
return -1, {}, ''
async def check_404_existence(self):
try:
try:
path = ''.join(random.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits) for _ in
range(random.randint(10, 30)))
self._404_status, _, html_doc = await self.http_request('/' + path)
except Exception as e:
await self.print_msg('[Warning] HTTP 404 check failed: %s, %s' % (self.base_url, type(e)))
self._404_status, _, html_doc = -1, {}, ''
if self._404_status != 404:
self.len_404_doc = len(html_doc)
except Exception as e:
await self.print_msg('[Check_404] Exception %s %s' % (self.base_url, str(e)))
#
async def enqueue(self, url):
try:
url = str(url)
except Exception as e:
return False
try:
url_pattern = re.sub(r'\d+', '{num}', url)
if url_pattern in self.urls_processed or len(self.urls_processed) >= self.links_limit:
return False
self.urls_processed.add(url_pattern)
# await self.print_msg('Entered Queue: %s' % url)
if not self.args.no_crawl: # no crawl
await self.crawl(url)
if self._404_status != -1: # valid web service
rule_set_to_process = [self.rules_set, self.rules_set_root_only] if url == '/' else [self.rules_set]
for rule_set in rule_set_to_process:
for _ in rule_set:
if _[5] and url != '/': # root only
continue
try:
full_url = url.rstrip('/') + _[0]
except Exception as e:
continue
if full_url in self.urls_enqueued:
continue
url_description = {'prefix': url.rstrip('/'), 'full_url': full_url}
item = (url_description, _[1], _[2], _[3], _[4], _[5], _[6])
await self.url_queue.put(item)
self.urls_enqueued.add(full_url)
if self.args.full_scan and url.count('/') >= 2:
await self.enqueue('/'.join(url.split('/')[:-2]) + '/') # sub folder enqueue
if url != '/' and not self.no_scripts:
for script in self.user_scripts:
await self.url_queue.put((script, url))
return True
except Exception as e:
await self.print_msg('[_enqueue.exception] %s' % str(e))
return False
#
async def crawl(self, path, do_not_process_links=False):
try:
# increase body size to 200 KB
request_headers = dict(conf.default_headers, Range='bytes=0-204800')
status, headers, html_doc = await self.http_request(path, headers=request_headers)
if path == '/':
self.index_status, self.index_headers, self.index_html_doc = status, headers, html_doc
if not self.index_has_reported:
self.index_has_reported = True
title = get_http_title(html_doc)
location = headers.get('Location', '')
server = headers.get('Server', '')
str_headers = ''
for key in self.index_headers:
# 减少非关键HTTP头的输出
if key.lower() in ['connection', 'content-encoding', 'content-security-policy',
'date', 'p3p', 'x-ua-compatible', 'x-ua-compatible', 'cache-control',
'x-xss-protection', 'transfer-encoding', 'last-modified', 'etag']:
continue
str_headers += '%s: %s\n' % (key, self.index_headers[key])
_ = {'status': status, 'url': clear_url(self.base_url), 'title': title, 'server': server,
'location': location, 'headers': str_headers}
await self.save_result('$Index', _)
if self.fingerprint_check:
# 检查Web指纹
cms_name = fingerprint.get_cms_name('/^^^get^^^{}^^^', status, headers, html_doc)
if cms_name:
await self.save_result(
'$Fingerprint', cms_name,
msg='[Fingerprint] %s found %s' % (('%s%s' % (self.base_url, path)).rstrip('/'), cms_name))
# 首页30x跳转，在第二次请求时，需要parse HTML， follow后获取新的HTML
if not self.args.no_crawl and not do_not_process_links and status in [301, 302]:
resp = await self.conn_pool.get(self.base_url + '/',
headers=conf.default_headers, timeout=20)
location = resp.headers.get('Location', '')
if location.lower().startswith('http'):
scheme, netloc, _path, params, query, fragment = urlparse(location, 'http')
if netloc.find(self.host) < 0: # different host, do not follow
location = ''
else:
location = _path + '?' + query
elif location.lower().startswith('/'):
pass
else:
location = '/' + location
if location:
url, depth = cal_depth(self, resp.headers.get('Location', ''))
if depth <= self.max_depth:
await self.enqueue(url)
# 避免处理错误，直接传入原始path，让httpx处理跳转URL，会重复，多1次请求
status, headers, html_doc = await self.http_request(path, headers=request_headers,
follow_redirects=True)
# 再次检查Web指纹
cms_name = fingerprint.get_cms_name('/^^^get^^^{}^^^', status, headers, html_doc)
if cms_name:
await self.save_result(
'$Fingerprint', cms_name,
msg='[Fingerprint] %s found %s' % (
('%s%s' % (self.base_url, path)).rstrip('/'), cms_name))
if not self.args.no_crawl and not do_not_process_links and html_doc:
fav_url_found = False
soup = BeautifulSoup(html_doc, "html.parser")
for tag in ['link', 'script', 'a']:
for link in soup.find_all(tag):
origin_url = url = link.get('href', '').strip()
if not url:
origin_url = url = link.get('src', '').strip()
if url.startswith('..'):
continue
if not url.startswith('/') and url.find('//') < 0: # relative path
url = path + url
url, depth = cal_depth(self, url)
# print(url, depth)
if depth <= self.max_depth:
await self.enqueue(url)
if self.fingerprint_check and tag == 'link' and str(link.get('rel', '')).find('icon') >= 0:
fav_url_found = True
fav_url, depth = cal_depth(self, link.get('href', '').strip())
if fav_url: # 非当前域名的icon url，不会请求
await self.url_queue.put(('favicon', fav_url, ''))
# 解析js获取URL
if (path == '/' and tag == 'script' and (self.args.api or not self.args.fingerprint_only) and
origin_url not in self.js_urls):
self.js_urls.append(origin_url)
js_url, depth = cal_depth(self, origin_url)
if js_url:
if origin_url.lower().startswith('http') and origin_url.find('://') > 0:
origin_url = origin_url.split('://')[1]
if origin_url.find('/') > 0:
origin_url = '/'.join(origin_url.split('/')[1:])
await self.url_queue.put(('js_file', origin_url, ''))
if path == '/' and self.fingerprint_check and not fav_url_found: # 尝试请求默认favicon，计算hash
await self.url_queue.put(('favicon', '/favicon.ico', ''))
if path == '/' and self.fingerprint_check:
self.fingerprint_check = False # this should only run once for each target
# 将CMS识别的其他请求，添加到队列
for key_name in fingerprint.rules.keys():
if key_name != '/^^^get^^^{}^^^': # 首页已经默认请求过
await self.url_queue.put(('key_name', key_name, ''))
ret = self.find_text(html_doc)
if ret:
title = get_http_title(html_doc)
_ = {'status': status, 'url': '%s%s' % (self.base_url, path), 'title': title, 'vul_type': ret[1]}
await self.save_result('/', _)
except Exception as e:
await self.print_msg('[crawl Exception] %s %s %s' % (path, type(e), str(e)))
async def load_all_urls_from_log_file(self):
try:
with open(self.log_file) as infile:
for _line in infile.readlines():
_ = _line.strip().split()
if len(_) == 3 and (_[2].find('^^^200') > 0 or _[2].find('^^^403') > 0 or _[2].find('^^^302') > 0):
url, depth = cal_depth(self, _[1])
await self.enqueue(url)
except Exception as e:
await self.print_msg('[load_all_urls_from_log_file] %s' % str(e))
def find_text(self, html_doc):
for _text in self.text_to_find:
if html_doc.find(_text) >= 0:
return True, 'Found [%s]' % _text
for _regex in self.regex_to_find:
if _regex.search(html_doc):
return True, 'Found Regex [%s]' % _regex.pattern
return False
def find_exclude_text(self, html_doc):
for _text in self.text_to_exclude:
if html_doc.find(_text) >= 0:
return True
for _regex in self.regex_to_exclude:
if _regex.search(html_doc):
return True
return False
async def is_url_valid(self, url, item):
url_description, tag, status_to_match, content_type, content_type_no, root_only, vul_type = item
status, headers, html_doc = await self.http_request(url)
cur_content_type = headers.get('content-type', '')
cur_content_length = headers.get('content-length', len(html_doc))
if self.find_exclude_text(html_doc): # excluded text found
return False
if 0 <= int(cur_content_length) <= 10: # text too short
return False
if cur_content_type.find('image/') >= 0: # exclude image
return False
if content_type != 'application/json' and cur_content_type.find('application/json') >= 0 and \
not url.endswith('.json'): # invalid json
return False
if content_type and cur_content_type.find(content_type) < 0 \
or content_type_no and cur_content_type.find(content_type_no) >= 0:
return False # content type mismatch
if tag and html_doc.find(tag) < 0:
return False # tag mismatch
if self.find_text(html_doc):
valid_item = True
else:
# status code check
if status_to_match == 206 and status != 206:
return False
if status_to_match in (200, 206) and status in (200, 206):
valid_item = True
elif status_to_match and status != status_to_match:
return False
elif status in (403, 404) and status != status_to_match:
return False
else:
valid_item = True
if status == self._404_status and url != '/':
len_doc = len(html_doc)
len_sum = self.len_404_doc + len_doc
if len_sum == 0 or (0.4 <= float(len_doc) / len_sum <= 0.6):
return False
return valid_item
async def save_result(self, prefix, item, msg=None):
async with self.lock:
if prefix not in self.results:
self.results[prefix] = []
if item not in self.results[prefix]:
self.results[prefix].append(item)
if msg:
await self.print_msg(msg)
async def scan_worker(self):
while True:
if time.time() - self.start_time > self.time_out and not self.timeout_scan_aborted:
self.timeout_scan_aborted = True
clear_queue(self.url_queue)
await self.print_msg('[ERROR] Timed out task: %s' % self.base_url)
return
try:
item = self.url_queue.get_nowait()
except Exception as e:
return
try:
if len(item) == 3:
if item[0] == 'favicon':
resp = await self.conn_pool.get(self.base_url + item[1],
headers=conf.default_headers,
follow_redirects=False, timeout=20)
fav_hash = hashlib.md5(resp.content).hexdigest()
if fav_hash in fingerprint.fav_icons:
cms_name = fingerprint.fav_icons[fav_hash]
await self.save_result('$Fingerprint', cms_name,
msg='[Fingerprint] %s found %s' % (self.base_url, cms_name))
elif item[0] == 'key_name':
key_name = item[1]
req_item = fingerprint.requests_to_do[key_name]
if req_item[2]:
headers = copy.deepcopy(conf.default_headers)
headers.update(req_item[2]) # update headers
else:
headers = conf.default_headers
resp = None
if req_item[1].lower() == 'get':
resp = await self.conn_pool.get(self.base_url + req_item[0], headers=headers)
elif req_item[1].lower() == 'post':
data = req_item[3]
resp = await self.conn_pool.post(self.base_url + req_item[0], headers=headers, data=data)
if resp:
cms_name = fingerprint.get_cms_name(key_name, resp.status_code, resp.headers, resp.text)
if cms_name:
await self.save_result('$Fingerprint', cms_name,
'[Fingerprint] %s found %s' % (self.base_url, cms_name))
elif item[0] == 'js_file':
_path = item[1] if item[1].startswith('/') else '/' + item[1]
status, headers, js_doc = await self.http_request(_path)
if headers['content-type'].find('javascript') >= 0:
urls_regex, all_path_items, data_leak_found = await get_urls_in_js_async(
asyncio.get_event_loop(), js_doc, self.base_url + item[1], self.args.api, self)
# 目前并没有尝试请求匹配到的两组 疑似API接口，有误报，需要先优化正则，减少误报后，再添加
# 对于接口测试，这里应该是1个非常重要的检测点
if self.args.api:
self.urls_regex_found = self.urls_regex_found.union(urls_regex)
for item in all_path_items:
if type(item[2]) is str:
if self.args.api:
urls_regex.add(item[2])
# await self.url_queue.put(('api_endpoint', item[2], ''))
url, depth = cal_depth(self, item[2])
if depth <= self.max_depth:
await self.enqueue(url)
if data_leak_found:
for item in data_leak_found:
_ = {'status': 200, 'url': self.base_url + _path,
'title': '%s (%s)' % (item[1], item[2]), 'vul_type': 'JS Info Leak'}
await self.save_result('/', _, '[JS Info Leak] %s : %s' % (_['url'], _['title']))
continue
elif len(item) == 2: # Script Scan
check_func = getattr(item[0], 'do_check')
# await self.print_msg('Begin %s %s' % (os.path.basename(item[0].__file__), item[1]))
await check_func(self, item[1])
# await self.print_msg('End %s %s' % (os.path.basename(item[0].__file__), item[1]))
continue
else:
url_description, tag, status_to_match, content_type, content_type_no, root_only, vul_type = item
prefix = url_description['prefix']
url = url_description['full_url']
if url.find('{sub}') >= 0:
if not self.domain_sub:
continue
url = url.replace('{sub}', self.domain_sub)
except Exception as e:
await self.print_msg('[scan_worker.1] %s, %s, %s' % (str(e), self.base_url, item))
# await self.print_msg(traceback.format_exc())
continue
if not item or not url:
break
try:
valid_item = await self.is_url_valid(url, item)
if valid_item:
_ = url.split('/')
_[-1] = 'fptest' + _[-1]
url_fp_test = '/'.join(_) # add false positive test prefix
ret = await self.is_url_valid(url_fp_test, item)
if ret:
valid_item = False
if valid_item:
status, headers, html_doc = await self.http_request(url)
title = get_http_title(html_doc)
_ = {'status': status, 'url': '%s%s' % (self.base_url, url), 'title': title, 'vul_type': vul_type}
await self.save_result(prefix, _)
except Exception as e:
await self.print_msg('[scan_worker.2][%s] %s, %s' % (url, str(e), item))
# await self.print_msg(traceback.format_exc())
async def scan(self, threads=6):
try:
all_threads = []
for i in range(threads):
t = self.scan_worker()
all_threads.append(t)
await asyncio.gather(*all_threads)
for key in self.results.keys():
# too many URLs found under this folder, deduplicate results
if len(self.results[key]) > 10:
vul_type_count = {}
for item in copy.deepcopy(self.results[key]):
if item['vul_type'] not in vul_type_count:
vul_type_count[item['vul_type']] = 1
else:
vul_type_count[item['vul_type']] += 1
if vul_type_count[item['vul_type']] >= 3:
self.results[key].remove(item)
return clear_url(self.base_url), self.results, self.urls_regex_found
except Exception as e:
await self.print_msg('[scan exception] %s' % str(e))
finally:
try:
await self.conn_pool.aclose()
except Exception as e:
pass
async def scan_process():
s = Scanner(args.timeout * 60)
await s.init()
while True:
try:
target = q_targets.get_nowait()
except asyncio.queues.QueueEmpty as e:
if conf.process_targets_done and q_targets.qsize() == 0:
break
else:
await asyncio.sleep(0.1)
continue
if 'target' in target:
ret = await s.init_from_url(target['target'])
elif 'file' in target:
ret = await s.init_from_log_file(target['file'])
else:
continue
if ret:
item = await s.scan(threads=args.t)
if item[1]:
await q_results.put(copy.deepcopy(item))
async def add_target(target, is_neighbor=False):
if is_neighbor:
target['no_scripts'] = 1 # 邻居IP，不启用插件. Bug fixed: 2024/05/03
if args.debug:
await q_results.put('New target: %s' % target)
await q_targets.put({'target': target})
if args.save_ports and target['ports_open']:
conf.ports_saved_to_file = True
if not args.ports_file:
args.ports_file = open(args.save_ports, 'w')
for port in target['ports_open']:
args.ports_file.write('%s:%s\n' % (target['host'], port))
args.ports_file.flush()
conf.tasks_count += 1
def is_intranet(ip):
try:
ret = ip.split('.')
if len(ret) != 4:
return True
if ret[0] == '10':
return True
if ret[0] == '172' and 16 <= int(ret[1]) <= 31:
return True
if ret[0] == '192' and ret[1] == '168':
return True
return False
except Exception as e:
return False
resolver = dns.asyncresolver.Resolver()
async def domain_lookup_check(queue_targets_origin, processed_targets, queue_targets):
while True:
try:
url = queue_targets_origin.get_nowait()
except asyncio.queues.QueueEmpty as e:
break
# scheme netloc path
if url.find('://') < 0:
netloc = url[:url.find('/')] if url.find('/') > 0 else url
else:
scheme, netloc, path, params, query, fragment = urlparse(url, 'http')
# host port
host = netloc.split(':')[0] if netloc.find(':') >= 0 else netloc
if is_ip_addr(host):
processed_targets.append(host)
if args.skip_intranet and is_intranet(host):
await q_results.put('Private IP target skipped: %s [%s]' % (url, host))
else:
await queue_targets.put((url, 0, host))
else:
for i in range(5):
try:
answers = await resolver.resolve(host, "A")
processed_targets.append(answers[0].address)
if args.skip_intranet and is_intranet(answers[0].address):
await q_results.put('Private IP target skipped: %s [%s]' % (url, answers[0].address))
else:
await queue_targets.put((url, 0, answers[0].address))
break
except dns.resolver.NXDOMAIN as e:
await q_results.put('No such domain: %s' % host)
break
except Exception as e:
if i == 4: # Failed after 4 retries
await q_results.put('Domain lookup failed [%s]: %s' % (e.__class__.__name__, host))
async def do_port_scan_check(queue_targets):
"""
检测目标的端口是否开放，输入的目标是URL，也可能是网段下的相邻IP
"""
while True:
try:
url, is_neighbor, ip_addr = queue_targets.get_nowait() # is_neighbor = 1 为相邻网段的IP，优先级降低
except asyncio.queues.QueueEmpty as e:
break
try:
# scheme netloc path
if url.find('://') < 0:
scheme = 'unknown'
netloc = url[:url.find('/')] if url.find('/') > 0 else url
path = ''
else:
scheme, netloc, path, params, query, fragment = urlparse(url, 'http')
# host port
if netloc.find(':') >= 0:
_ = netloc.split(':')
host = _[0]
try:
port = int(_[1])
except:
port = None
else:
host = netloc
port = None
if scheme == 'https' and port is None:
port = 443
elif scheme == 'http' and port is None:
port = 80
if scheme == 'unknown':
if port == 80:
scheme = 'http'
if port == 443:
scheme = 'https'
ports_open = set()
ports_closed = set()
# 插件不依赖HTTP连接池， 且仅启用插件扫描， 则不需要检查80/443端口的HTTP服务， 直接扫描 require_ports
if args.scripts_only and args.require_no_http:
ports_open, ports_closed = await scan_given_ports(ip_addr, args.require_ports, ports_open, ports_closed)
target = {'scheme': scheme, 'host': host, 'port': port, 'path': path,
'has_http': False, 'ports_open': ports_open, 'ports_closed': ports_closed}
await add_target(target) # 在只扫插件的情况下，相邻IP也需要启用
continue
if port:
# 指定了 标准端口 或 非标准端口
has_http = await is_port_open(ip_addr, port)
if has_http:
ports_open.add(port)
else:
ports_closed.add(port)
if not args.no_scripts:
ports_open, ports_closed = \
await scan_given_ports(ip_addr, args.require_ports, ports_open, ports_closed)
target = {'scheme': scheme, 'host': host, 'port': port, 'path': path, 'has_http': has_http,
'ports_open': ports_open, 'ports_closed': ports_closed}
await add_target(target)
else:
# 只有域名和IP情况下， 扫默认端口
port_open_80 = await is_port_open(ip_addr, 80)
port_open_443 = await is_port_open(ip_addr, 443)
if port_open_80:
ports_open.add(80)
else:
ports_closed.add(80)
if port_open_443:
ports_open.add(443)
else:
ports_closed.add(443)
if not args.no_scripts:
ports_open, ports_closed = \
await scan_given_ports(ip_addr, args.require_ports, ports_open, ports_closed)
if port_open_80 and port_open_443:
target = {'scheme': 'https', 'host': host, 'port': 443, 'path': path,
'has_http': True, 'ports_open': ports_open, 'ports_closed': ports_closed}
await add_target(target, is_neighbor)
# 排除 301 HTTP 跳转 HTTPS的目标
async with httpx.AsyncClient() as client:
r = await client.get('http://%s' % host, follow_redirects=False, timeout=20)
if r and not \
(r.status_code == 301 and r.headers.get('Location', '').lower().startswith('https')):
target = {'scheme': 'http', 'host': host, 'port': 80, 'path': path,
'has_http': True, 'no_scripts': 1,
'ports_open': ports_open, 'ports_closed': ports_closed}
await add_target(target)
elif port_open_443:
target = {'scheme': 'https', 'host': host, 'port': 443, 'path': path,
'has_http': True, 'ports_open': ports_open, 'ports_closed': ports_closed}
# 即使指定的目标，允许插件扫描，邻居也将不启用，节省扫描时间
await add_target(target, is_neighbor)
elif port_open_80:
target = {'scheme': 'http', 'host': host, 'port': 80, 'path': path,
'has_http': True, 'ports_open': ports_open, 'ports_closed': ports_closed}
await add_target(target, is_neighbor)
elif args.no_scripts:
# 80 443 端口不开放， 禁用插件扫描
await q_results.put('No ports open: %s' % host)
elif not is_neighbor or args.scripts_only:
# 直接输入目标 或者 对相邻IP应用插件
# 80 443 未开放，此时只能检测其他端口的漏洞
# 如果没有任何开放的端口，直接跳过该目标
if ports_open:
target = {'scheme': 'http', 'host': host, 'port': 80, 'path': path,
'has_http': False, 'ports_open': ports_open, 'ports_closed': ports_closed}