(.*?)', re.S)
- link_regex = re.compile('(.*?)
', re.S)
- links = []
- try:
- results_tbl = tbl_regex.findall(resp)[0]
- except IndexError:
- results_tbl = ''
- links_list = link_regex.findall(results_tbl)
- links = list(set(links_list))
- for link in links:
- subdomain = link.strip()
- if not subdomain.endswith(self.domain):
- continue
- if subdomain and subdomain not in self.subdomains and subdomain != self.domain:
- self.subdomains.append(subdomain.strip())
- return links
-
-# Virustotal (updated to v3)
-class Virustotal(EnumeratorBaseThreaded):
- def __init__(self, domain, subdomains=None, q=None, silent=False, verbose=True):
- subdomains = subdomains or []
- self.engine_name = "VirusTotal"
- self.q = q
- super(Virustotal, self).__init__('', self.engine_name, domain, subdomains, q=q, silent=silent, verbose=verbose)
- self.url = f"https://www.virustotal.com/api/v3/domains/{self.domain}/relationships/subdomains"
- api_key = os.getenv('VT_API_KEY')
- if api_key:
- self.headers['x-apikey'] = api_key
- else:
- self.print_(Y + "[!] No VT_API_KEY set, using public API (rate limited)" + W)
-
- def send_req(self, url):
- try:
- resp = self.session.get(url, headers=self.headers, timeout=self.timeout)
- resp.raise_for_status()
- except Exception as e:
- self.print_(f"{R}[!] VT Error: {e}{W}")
- resp = None
- return self.get_response(resp)
-
- def enumerate(self):
- while self.url:
- resp = self.send_req(self.url)
- if not resp:
- break
- try:
- data = json.loads(resp)
- if 'error' in data:
- self.print_(R + f"[!] VT Error: {data['error']['message']}" + W)
- break
- if 'data' in data:
- for item in data['data']:
- subdomain = item['id']
- if subdomain.endswith(self.domain) and subdomain not in self.subdomains and subdomain != self.domain:
- if self.verbose:
- self.print_(f"{R}{self.engine_name}: {W}{subdomain}")
- self.subdomains.append(subdomain)
- if 'links' in data and 'next' in data['links']:
- self.url = data['links']['next']
- else:
- self.url = None
- except json.JSONDecodeError:
- break
- time.sleep(15) # Rate limit
- return self.subdomains
-
- def extract_domains(self, resp):
- pass
-
-# ThreatCrowd (from original, note: ThreatCrowd is deprecated, but keeping for compatibility)
-class ThreatCrowd(EnumeratorBaseThreaded):
- def __init__(self, domain, subdomains=None, q=None, silent=False, verbose=True):
- subdomains = subdomains or []
- base_url = 'https://www.threatcrowd.org/searchApi/v2/domain/report/?domain={domain}'
- self.engine_name = "ThreatCrowd"
- self.q = q
- super(ThreatCrowd, self).__init__(base_url, self.engine_name, domain, subdomains, q=q, silent=silent, verbose=verbose)
-
- def req(self, url):
- try:
- resp = self.session.get(url, headers=self.headers, timeout=self.timeout)
- except Exception:
- resp = None
- return self.get_response(resp)
-
- def enumerate(self):
- url = self.base_url.format(domain=self.domain)
- resp = self.req(url)
- self.extract_domains(resp)
- return self.subdomains
-
- def extract_domains(self, resp):
- try:
- links = json.loads(resp)['subdomains']
- for link in links:
- subdomain = link.strip()
- if not subdomain.endswith(self.domain):
- continue
- if subdomain not in self.subdomains and subdomain != self.domain:
- if self.verbose:
- self.print_(f"{R}{self.engine_name}: {W}{subdomain}")
- self.subdomains.append(subdomain.strip())
- except Exception as e:
- pass
-
-# CrtSearch (from original)
-class CrtSearch(EnumeratorBaseThreaded):
- def __init__(self, domain, subdomains=None, q=None, silent=False, verbose=True):
- subdomains = subdomains or []
- base_url = 'https://crt.sh/?q=%25.{domain}'
- self.engine_name = "SSL Certificates"
- self.q = q
- super(CrtSearch, self).__init__(base_url, self.engine_name, domain, subdomains, q=q, silent=silent, verbose=verbose)
-
- def req(self, url):
- try:
- resp = self.session.get(url, headers=self.headers, timeout=self.timeout)
- except Exception:
- resp = None
- return self.get_response(resp)
-
- def enumerate(self):
- url = self.base_url.format(domain=self.domain)
- resp = self.req(url)
- if resp:
- self.extract_domains(resp)
- return self.subdomains
-
- def extract_domains(self, resp):
- link_regx = re.compile(' | (.*?) | ')
- try:
- links = link_regx.findall(resp)
- for link in links:
- link = link.strip()
- subdomains = []
- if '
' in link:
- subdomains = link.split('
')
- else:
- subdomains.append(link)
- for subdomain in subdomains:
- if not subdomain.endswith(self.domain) or '*' in subdomain:
- continue
- if '@' in subdomain:
- subdomain = subdomain[subdomain.find('@')+1:]
- if subdomain not in self.subdomains and subdomain != self.domain:
- if self.verbose:
- self.print_(f"{R}{self.engine_name}: {W}{subdomain}")
- self.subdomains.append(subdomain.strip())
- except Exception as e:
- pass
-
-# PassiveDNS (from original)
-class PassiveDNS(EnumeratorBaseThreaded):
- def __init__(self, domain, subdomains=None, q=None, silent=False, verbose=True):
- subdomains = subdomains or []
- base_url = 'https://api.sublist3r.com/search.php?domain={domain}'
- self.engine_name = "PassiveDNS"
- self.q = q
- super(PassiveDNS, self).__init__(base_url, self.engine_name, domain, subdomains, q=q, silent=silent, verbose=verbose)
-
- def req(self, url):
- try:
- resp = self.session.get(url, headers=self.headers, timeout=self.timeout)
- except Exception as e:
- resp = None
- return self.get_response(resp)
-
- def enumerate(self):
- url = self.base_url.format(domain=self.domain)
- resp = self.req(url)
- if not resp:
- return self.subdomains
- self.extract_domains(resp)
- return self.subdomains
-
- def extract_domains(self, resp):
- try:
- subdomains = json.loads(resp)
- for subdomain in subdomains:
- if subdomain not in self.subdomains and subdomain != self.domain:
- if self.verbose:
- self.print_(f"{R}{self.engine_name}: {W}{subdomain}")
- self.subdomains.append(subdomain.strip())
- except Exception as e:
- pass
-
-# BufferOverRunEnum (new in v2, kept)
-class BufferOverRunEnum(EnumeratorBaseThreaded):
- def __init__(self, domain, subdomains=None, q=None, silent=False, verbose=True):
- subdomains = subdomains or []
- self.engine_name = "BufferOverRun"
- self.q = q
- super(BufferOverRunEnum, self).__init__('', self.engine_name, domain, subdomains, q=q, silent=silent, verbose=verbose)
- self.url = f"https://dns.bufferover.run/dns?q=.{self.domain}"
-
- def send_req(self, url):
- try:
- resp = self.session.get(url, headers=self.headers, timeout=self.timeout)
- resp.raise_for_status()
- except Exception:
- resp = None
- return self.get_response(resp)
-
- def enumerate(self):
- resp = self.send_req(self.url)
- if not resp:
- return self.subdomains
- try:
- data = json.loads(resp)
- all_dns = data.get('FDNS_A', []) + data.get('FDNS_AAAA', [])
- for dns_entry in all_dns:
- parts = [p.strip() for p in dns_entry.split(',')]
- if len(parts) > 1:
- subdomain = parts[1]
- if subdomain.endswith(self.domain) and subdomain not in self.subdomains and subdomain != self.domain:
- if self.verbose:
- self.print_(f"{R}{self.engine_name}: {W}{subdomain}")
- self.subdomains.append(subdomain)
- except Exception as e:
- self.print_(f"{R}[!] BufferOverRun Error: {e}{W}")
- return self.subdomains
-
- def extract_domains(self, resp):
- pass
-
-# New for v3.0: CertSpotter
-class CertSpotterEnum(EnumeratorBaseThreaded):
- def __init__(self, domain, subdomains=None, q=None, silent=False, verbose=True):
- subdomains = subdomains or []
- self.engine_name = "CertSpotter"
- self.q = q
- super(CertSpotterEnum, self).__init__('', self.engine_name, domain, subdomains, q=q, silent=silent, verbose=verbose)
- self.url = f"https://certspotter.com/api/v0/certs?domain={self.domain}&expand=dns_names"
-
- def send_req(self, url):
- try:
- resp = self.session.get(url, headers=self.headers, timeout=self.timeout)
- resp.raise_for_status()
- except Exception:
- resp = None
- return self.get_response(resp)
-
- def enumerate(self):
- resp = self.send_req(self.url)
- if not resp:
- return self.subdomains
- try:
- data = json.loads(resp)
- for cert in data:
- for dns_name in cert.get('dns_names', []):
- if dns_name.endswith(self.domain) and dns_name not in self.subdomains and dns_name != self.domain:
- if self.verbose:
- self.print_(f"{R}{self.engine_name}: {W}{dns_name}")
- self.subdomains.append(dns_name)
- except Exception as e:
- self.print_(f"{R}[!] CertSpotter Error: {e}{W}")
- return self.subdomains
-
- def extract_domains(self, resp):
- pass
-
-class PortScan:
- def __init__(self, subdomains, ports):
- self.subdomains = subdomains
- self.ports = ports
- self.lock = None
-
- def port_scan(self, host, ports):
- openports = []
- self.lock.acquire()
- for port in ports:
- try:
- s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- s.settimeout(2)
- result = s.connect_ex((host, int(port)))
- if result == 0:
- openports.append(port)
- s.close()
- except Exception:
- pass
- self.lock.release()
- if openports:
- print(f"{G}{host}{W} - {R}Found open ports:{W} {Y}{', '.join(openports)}{W}")
-
- def run(self):
- self.lock = threading.BoundedSemaphore(value=50)
- threads = []
- for subdomain in self.subdomains:
- t = threading.Thread(target=self.port_scan, args=(subdomain, self.ports))
- t.start()
- threads.append(t)
- for t in threads:
- t.join()
-
-def main(domain, threads, savefile, ports, silent, verbose, enable_bruteforce, engines, json_output):
- bruteforce_list = set()
- search_list = set()
-
- subdomains_queue = multiprocessing.Manager().list()
-
- if enable_bruteforce or enable_bruteforce is None:
- enable_bruteforce = True
-
- domain_check = re.compile(r"^(http|https)?[a-zA-Z0-9]+([\-\.]{1}[a-zA-Z0-9]+)*\.[a-zA-Z]{2,}$")
- if not domain_check.match(domain):
- if not silent:
- print(R + "Error: Please enter a valid domain" + W)
- return []
-
- if not domain.startswith(('http://', 'https://')):
- domain = 'http://' + domain
-
- parsed_domain = urlparse(domain).netloc
-
- if not silent:
- print(B + f"[-] Enumerating subdomains now for {parsed_domain}" + W)
-
- if verbose and not silent:
- print(Y + "[-] Verbosity enabled, showing results in realtime" + W)
-
- supported_engines = {
- 'baidu': BaiduEnum,
- 'yahoo': YahooEnum,
- 'google': GoogleEnum,
- 'bing': BingEnum,
- 'ask': AskEnum,
- 'netcraft': NetcraftEnum,
- 'dnsdumpster': DNSdumpster,
- 'virustotal': Virustotal,
- 'threatcrowd': ThreatCrowd,
- 'crt': CrtSearch,
- 'passivedns': PassiveDNS,
- 'bufferover': BufferOverRunEnum,
- 'certspotter': CertSpotterEnum # New in v3.0
- }
-
- chosen_enums = []
-
- if engines is None:
- chosen_enums = [
- GoogleEnum, BingEnum, YahooEnum, AskEnum, BaiduEnum,
- NetcraftEnum, DNSdumpster, Virustotal, ThreatCrowd,
- CrtSearch, BufferOverRunEnum, PassiveDNS, CertSpotterEnum # Added CertSpotter
- ]
- else:
- engines_list = [e.lower().strip() for e in engines.split(',')]
- for engine in engines_list:
- if engine in supported_engines:
- chosen_enums.append(supported_engines[engine])
-
- # Start enumeration
- enums = [enum_class(domain, [], q=subdomains_queue, silent=silent, verbose=verbose) for enum_class in chosen_enums]
- for enum in enums:
- enum.start()
- for enum in enums:
- enum.join()
-
- subdomains = set(subdomains_queue)
- for subdomain in subdomains:
- search_list.add(subdomain)
-
- if enable_bruteforce:
- if not silent:
- print(G + "[-] Starting bruteforce with subbrute.." + W)
- path_to_file = os.path.dirname(os.path.realpath(__file__))
- subs_file = os.path.join(path_to_file, 'subbrute', 'names.txt')
- resolvers_file = os.path.join(path_to_file, 'subbrute', 'resolvers.txt')
- process_count = threads
- output = False
- json_out = False
- bruteforce_list = subbrute.print_target(parsed_domain, False, subs_file, resolvers_file, process_count, output, json_out, search_list, verbose)
-
- all_subdomains = search_list.union(bruteforce_list)
-
- if all_subdomains:
- all_subdomains = sorted(all_subdomains, key=subdomain_sorting_key)
-
- if savefile:
- write_file(savefile, all_subdomains, json_output=False)
-
- if json_output:
- json_filename = f"{parsed_domain}.json"
- write_file(json_filename, all_subdomains, json_output=True)
-
- if not silent:
- print(Y + f"[-] Total Unique Subdomains Found: {len(all_subdomains)}" + W)
-
- if not json_output:
- for subdomain in all_subdomains:
- print(G + subdomain + W)
-
- if ports:
- if not silent:
- print(G + f"[-] Starting port scan for ports: {Y}{ports}" + W)
- ports_list = ports.split(',')
- pscan = PortScan(all_subdomains, ports_list)
- pscan.run()
-
- return list(all_subdomains)
-
-def interactive():
- args = parse_args()
- domain = args.domain
- threads = args.threads
- savefile = args.output
- ports = args.ports
- enable_bruteforce = args.bruteforce
- verbose = args.verbose or args.verbose is None
- engines = args.engines
- json_out = args.json
- if args.no_color:
- no_color()
- banner()
- main(domain, threads, savefile, ports, silent=False, verbose=verbose, enable_bruteforce=enable_bruteforce, engines=engines, json_output=json_out)
-
-if __name__ == "__main__":
- interactive()
+ resp = self.session.get(url, headers=self.headers