Fixed static domain

Author: jonluca

anubis/commands/target.py

@@ -24,32 +24,6 @@
 from .base import Base
 
 
-def clean_domains(domains):
-  cleaned = []
-  for subdomain in domains:
-    subdomain = subdomain.lower()
-    if subdomain.find("//") != -1:
-      subdomain = subdomain[subdomain.find("//") + 2:]
-    # Some pkey return instances like example.com. - remove the final .
-    if subdomain.endswith('.'):
-      subdomain = subdomain[:-1]
-    # sometimes we'll get something like /www.example.com
-    if subdomain[0] in ["\\", ".", "/", "#", "$", "%"]:
-      subdomain = subdomain[1:]
-    # If it's an email address, only take the domain part
-    if "@" in subdomain:
-      subdomain = subdomain.split("@")
-      # If it's an actual email like mail@example.com, take example.com
-      if len(subdomain) > 1:
-        subdomain = subdomain[1]
-      else:
-        # If for some reason it's example.com@, take example.com
-        subdomain = subdomain[0]
-
-    cleaned.append(subdomain.strip())
-  return cleaned
-
-
 class Target(Base):
   """Main enumeration module"""
   domains = list()
@@ -95,8 +69,7 @@ def run(self):
     for i in range(len(self.options["TARGET"])):
       # Default scans that run every time
       target = self.options["TARGET"][i]
-      processes = [
-        threading.Thread(target=dns_zonetransfer, args=(self, target)),
+      threads = [threading.Thread(target=dns_zonetransfer, args=(self, target)),
         threading.Thread(target=search_subject_alt_name, args=(self, target)),
         threading.Thread(target=subdomain_hackertarget, args=(self, target)),
         threading.Thread(target=search_virustotal, args=(self, target)),
@@ -108,37 +81,37 @@ def run(self):
       print('test')
       # Additional options - ssl cert scan
       if self.options["--ssl"]:
-        processes.append(threading.Thread(target=ssl_scan, args=(self, target)))
+        threads.append(threading.Thread(target=ssl_scan, args=(self, target)))
 
       # Additional options - shodan.io scan
       if self.options["--additional-info"]:
-        processes.append(threading.Thread(target=search_shodan, args=(self,)))
+        threads.append(threading.Thread(target=search_shodan, args=(self,)))
 
       # Additional options - nmap scan of dnssec script and a host/port scan
       if self.options["--with-nmap"]:
-        processes.append(
+        threads.append(
           threading.Thread(target=dnssecc_subdomain_enum, args=(self, target)))
-        processes.append(threading.Thread(target=scan_host, args=(self)))
+        threads.append(threading.Thread(target=scan_host, args=(self,)))
 
       # Additional options - brute force common subdomains
       if self.options["--brute-force"]:
-        processes.append(
+        threads.append(
           threading.Thread(target=brute_force, args=(self, target)))
 
-      # Start all processes
-    for x in processes:
+      # Start all threads
+    for x in threads:
       x.start()
 
     # Wait for all of them to finish
-    for x in processes:
+    for x in threads:
       x.join()
 
     # remove duplicates and clean up
 
     if self.options["--recursive"]:
       recursive_search(self)
 
-    self.domains = clean_domains(self.domains)
+    self.domains = self.clean_domains(self.domains)
     self.dedupe = set(self.domains)
 
     print("Found", len(self.dedupe), "subdomains")
@@ -172,3 +145,29 @@ def resolve_ips(self):
       # String truthiness ignores empty strings
       if ip:
         ColorPrint.green(ip)
+
+  @staticmethod
+  def clean_domains(domains):
+    cleaned = []
+    for subdomain in domains:
+      subdomain = subdomain.lower()
+      if subdomain.find("//") != -1:
+        subdomain = subdomain[subdomain.find("//") + 2:]
+      # Some pkey return instances like example.com. - remove the final .
+      if subdomain.endswith('.'):
+        subdomain = subdomain[:-1]
+      # sometimes we'll get something like /www.example.com
+      if subdomain[0] in ["\\", ".", "/", "#", "$", "%"]:
+        subdomain = subdomain[1:]
+      # If it's an email address, only take the domain part
+      if "@" in subdomain:
+        subdomain = subdomain.split("@")
+        # If it's an actual email like mail@example.com, take example.com
+        if len(subdomain) > 1:
+          subdomain = subdomain[1]
+        else:
+          # If for some reason it's example.com@, take example.com
+          subdomain = subdomain[0]
+
+      cleaned.append(subdomain.strip())
+    return cleaned

anubis/scanners/recursive.py

@@ -11,7 +11,7 @@
 
 def recursive_search(self):
   print("Starting recursive search - warning, might take a long time")
-  domains = clean_domains(self.domains)
+  domains = self.clean_domains(self.domains)
   domains_unique = set(domains)
   num_workers = 10
 

tests/commands/test_target.py

@@ -105,10 +105,9 @@ def test_netcraft(self):
   # As of 1/18/18, Pkey currently constantly times out
   def test_pkey(self):
     search_pkey(self, "google.com")
-    #self.assertIn("google.com", self.domains)
+    # self.assertIn("google.com", self.domains)
     self.assertTrue(True)
 
-
   def test_shodan(self):
     self.ip = "138.197.125.24"
     self.options = {}
@@ -141,13 +140,14 @@ def test_recursive(self):
 
   # Pass through function for recursive search
   def clean_domains(self, domains):
-    return clean_domains(domains)
+    return Target.clean_domains(domains)
 
   def test_sigints(self):
     # Declare function to send sigint, after timer
 
+    proc1 = popen(['anubis', '-tr', 'neverssl.com'], stdout=PIPE)
 
-    proc1 = popen(['anubis', '-tr','neverssl.com'], stdout=PIPE)
+    # Function to send sigint to our processes, make sure that it outputss "Quitting" then ends
     def send_siginit():
       popen.send_signal(proc1, signal.SIGINT)
       self.assertTrue("Quitting" in sys.stdout.getvalue())
@@ -165,7 +165,6 @@ def test_exception(self):
       self.assertTrue("Test" in sys.stdout.getvalue())
 
 
-
 class TestColorPrint(TestCase):
 
   def setUp(self):