-
Notifications
You must be signed in to change notification settings - Fork 9
/
Substr3am.py
executable file
·215 lines (171 loc) · 8.08 KB
/
Substr3am.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
#!/usr/local/bin/python3
####################
# Substr3am v1.0
#
# @nexxai
# github.com/nexxai/Substr3am
####################
import sys
import certstream
import tldextract
import re
import argparse
from sqlalchemy import create_engine, update
from declarative_sql import Subdomain, Base
from sqlalchemy.orm import sessionmaker
def print_callback(message, context):
# Add any subdomains (or partial strings) you want to ignore here
subdomains_to_ignore = [
"www",
"*",
"azuregateway",
"direwolf",
"devshell-vm-",
"device-local",
"-local",
"sni"
]
subdomains_regex_to_ignore = [
# 81d556ba781237c92f0c410f
"[a-f0-9]{24}",
# device1650096-3a628f22
"device[a-f0-9]{7}-[a-f0-9]{8}",
# e4751426-33f2-4239-9765-56b4cbcb505d
"[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}",
#device-3e90cd1b-50dc-48f1-90ac-6389856ccb2e
"device-[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}",
]
if message['message_type'] == "heartbeat":
return
# These are the messages we care about
if message['message_type'] == "certificate_update":
# Some certificates have SAN addresses attached to them
# so we want to know about all of them
domains = message['data']['leaf_cert']['all_domains']
# Get the domain name from the arguments
domain_filter = parse_args().filter
for domain in domains:
# Use the tldextract library to break the domain into its parts
extract = tldextract.extract(domain)
# Determine the root domain by concatenating the chosen name ('microsoft') with the TLD/suffix ('com')
computed_domain = extract.domain + '.' + extract.suffix
# Only continue if one of the following are true:
# - We are not filtering by domain at all
# - We are filtering by domain and we have a match
if domain_filter == None or (domain_filter != None and computed_domain in domain_filter):
# Set the domain based on whether or not we're filtering
if domain_filter == None:
subdomain = extract.subdomain
else:
subdomain = domain
# Make sure there's actually something there
if len(subdomain) > 0:
# Sometimes extract.subdomain gives us something like "www.testing.box"
# which is fine when we're filtering by domain by not when we're not
# so search for anything with a period in it
multi_level = subdomain.find(".")
# And then if it find something that has a period in it and we're not filtering...
if multi_level != -1 and domain_filter == None:
# Split it into two parts, the "www" and the "testing.box"...
subdomain_split = subdomain.split('.', 1)
# ...and we only care about the first entry
subdomain = subdomain_split[0]
# This counter needs to stay at 0 if we're going to act on this entry
i = 0
# See if any of the subdomains_to_ignore are substrings of the one we're checking
# e.g. "devshell-vm" is a substring of "devshell-vm-0000-0000-00000000"
for search in subdomains_to_ignore:
# If it matches, increase the counter
if search in subdomain:
i += 1
# See if any of the subdomains_regex_to_ignore are substrings of the one we're checking
for search in subdomains_regex_to_ignore:
# If it matches, increase the counter
if re.search(search, subdomain):
i += 1
# As long as none of the substrings or regexes match, continue on
if i == 0:
# Set up the connection to the sqlite db
engine = create_engine('sqlite:///subdomains.db')
Base.metadata.bind = engine
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()
# Check to see if it already exists in the database...
subdomain_exists = session.query(Subdomain).filter_by(subdomain=subdomain).first()
# It doesn't exist...
if not subdomain_exists:
# ...so create it
subdomain_new = Subdomain(subdomain=subdomain, count=1)
session.add(subdomain_new)
session.commit()
# Debug line
print("[+] " + subdomain)
# It does exist
if subdomain_exists:
# Add one to the counter to track its popularity
counter = subdomain_exists.count + 1
# Add 1 to the counter
session.query(Subdomain).filter(Subdomain.id == subdomain_exists.id).\
update({'count': counter})
session.commit()
if (counter % 50 == 0):
print("[#] " + subdomain + " (seen " + str(counter) + " times)")
def dump():
# Set up the connection to the sqlite db
engine = create_engine('sqlite:///subdomains.db')
Base.metadata.bind = engine
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()
# Get all the subdomains from the DB and sort them by popularity
subdomains = session.execute("SELECT * FROM subdomains ORDER BY count DESC").fetchall()
# Assuming there's anything in the list...
if len(subdomains) > 0:
# Open the file
f = open("names.txt", "w")
for subdomain in subdomains:
# And write them
f.write(subdomain.subdomain)
f.write("\r\n")
f.close()
sys.exit('names.txt has been written')
def parse_args():
# parse the arguments
parser = argparse.ArgumentParser(epilog='\tExample: \r\npython ' + sys.argv[0] + " -d")
parser.error = parser_error
parser._optionals.title = "OPTIONS"
parser.add_argument('-d', '--dump', help="Dump the list of collected subdomains to names.txt", action='store_true')
parser.add_argument('-f', '--filter', help="A space-separated list of domain names to filter for (e.g. 'google.com' or 'tesco.co.uk tesco.com harrods.com'). BE PATIENT.", nargs='+')
return parser.parse_args()
def parser_error(errmsg):
banner()
print("Usage: python " + sys.argv[0] + " [Options] use -h for help")
print("Error: " + errmsg)
sys.exit()
def main():
# Actually connect to the certstream firehose, and listen for events
certstream.listen_for_events(print_callback, url='wss://certstream.calidog.io/')
def interactive():
args = parse_args()
if args.dump:
dump()
banner()
main()
def banner():
G = '\033[92m' # green
Y = '\033[93m' # yellow
B = '\033[94m' # blue
R = '\033[91m' # red
W = '\033[0m' # white
print("""%s
_________ ___. __ ________
/ _____/__ _\_ |__ _______/ |_______\_____ \_____ _____
\_____ \| | \ __ \ / ___/\ __\_ __ \_(__ <\__ \ / \
/ \ | / \_\ \\\\___ \ | | | | \/ \/ __ \| Y Y \\
/_______ /____/|___ /____ > |__| |__| /______ (____ /__|_| /
\/ \/ \/ \/ \/ \/ %s%s
# Coded By Justin Smith - @nexxai
""" % (R, W, Y))
if __name__ == "__main__":
interactive()