Python3 port

red_ttp/common.py

@@ -2,17 +2,18 @@
 import socket
 import time
 import threading
-import SimpleHTTPServer
+import http.server
 import binascii
 import shutil
 import sys
-import SocketServer
+import socketserver
 import re
 import os
 import getpass
 import functools
 
 
+
 try:
     HOSTNAME = socket.gethostname().lower()
     LOCAL_IP = socket.gethostbyname(HOSTNAME)
@@ -64,7 +65,7 @@ def decorator(f):
         @functools.wraps(f)
         def decorated(*args, **kwargs):
             if len(missing):
-                log("Missing dependencies for %s:%s()" % (f.func_code.co_filename, f.func_code.co_name), "!")
+                log("Missing dependencies for %s:%s()" % (f.__code__.co_filename, f.__code__.co_name), "!")
                 for dep in missing:
                     print("    - %s" % os.path.relpath(dep, BASE_DIR))
                 return MISSING_DEPENDENCIES
@@ -80,7 +81,7 @@ def get_path(*path):
 def execute(command, hide_log=False, mute=False, timeout=30, wait=True, kill=False, drop=False, shell=False):
     """Execute a process and get the output."""
     if isinstance(command, list):
-        command = subprocess.list2cmdline([unicode(arg) for arg in command])
+        command = subprocess.list2cmdline([str(arg) for arg in command])
 
     if not hide_log:
         print("%s > %s" % (HOSTNAME, command))
@@ -100,7 +101,7 @@ def execute(command, hide_log=False, mute=False, timeout=30, wait=True, kill=Fal
     if kill:
         delta = 0.5
         # Try waiting for the process to die
-        for _ in xrange(int(timeout / delta) + 1):
+        for _ in range(int(timeout / delta) + 1):
             time.sleep(delta)
             if p.poll() is not None:
                 return
@@ -113,15 +114,15 @@ def execute(command, hide_log=False, mute=False, timeout=30, wait=True, kill=Fal
             pass
     elif wait:
         output = ''
-        p.stdin.write(os.linesep)
+        p.stdin.write(os.linesep.encode())
         while p.poll() is None:
-            line = p.stdout.readline()
+            line = p.stdout.readline().decode()
             if line:
                 output += line
                 if not (hide_log or mute):
                     print(line.rstrip())
 
-        output += p.stdout.read()
+        output += p.stdout.read().decode()
         output = output.strip()
 
         # Add artificial sleep to slow down command lines
@@ -179,15 +180,15 @@ def clear_web_cache():
 
 
 def serve_web(ip=LOCAL_IP, port=None, directory=BASE_DIR):
-    handler = SimpleHTTPServer.SimpleHTTPRequestHandler
+    handler = http.server.SimpleHTTPRequestHandler
 
     if port is not None:
-        server = SocketServer.TCPServer((ip, port), handler)
+        server = socketserver.TCPServer((ip, port), handler)
     else:
         # Otherwise, try to find a port
-        for port in xrange(8000, 9000):
+        for port in range(8000, 9000):
             try:
-                server = SocketServer.TCPServer((ip, port), handler)
+                server = socketserver.TCPServer((ip, port), handler)
                 break
             except socket.error:
                 pass
@@ -225,14 +226,14 @@ def patch_regex(source_file, regex, new_bytes, target_file=None):
     log("Patching by regex %s --> %s" % (source_file, target_file))
 
     with open(source_file, "rb") as f:
-        contents = f.read()
+        contents = f.read().decode()
 
     matches = re.findall(regex, contents)
     log("Changing %s -> %s" % (', '.join(matches), new_bytes))
     contents = re.sub(regex, new_bytes, contents)
 
     with open(target_file, "wb") as f:
-        f.write(contents)
+        f.write(contents.encode())
 
 
 def wchar(s):
@@ -263,7 +264,7 @@ def find_remote_host():
 
     if len(pending) > 0:
         # See which ones return first with a success code, and use that host
-        for _ in xrange(20):
+        for _ in range(20):
             for hostname, pending_process in sorted(pending.items()):
                 if pending_process.poll() is None:
                     pending_process.stdin.write(os.linesep)
@@ -359,6 +360,6 @@ def print_file(path):
     else:
         print('-' * 16)
         with open(path, 'rb') as f:
-            print(f.read().rstrip())
+            print(f.read().decode().rstrip())
 
     print('')

red_ttp/office_application_startup.py

@@ -4,7 +4,7 @@
 # Description: Modifies the registry to persist a DLL on Office Startup.
 
 import common
-import _winreg as winreg
+import winreg
 import sys
 import time
 

red_ttp/powershell_args.py

@@ -9,16 +9,16 @@
 
 
 def encode(command):
-    return base64.b64encode(command.encode('utf-16le'))
+    return base64.b64encode(command.encode('utf-16le')).decode()
 
 
 def main():
     common.log("PowerShell Suspicious Commands")
     temp_script = os.path.abspath("tmp.ps1")
 
-    # Create an empty script     
+    # Create an empty script
     with open(temp_script, "wb") as f:
-        f.write("whoami.exe\n")
+        f.write("whoami.exe\n".encode())
 
     powershell_commands = [
         'powershell -encoded %s' % encode('ping google.com'),

red_ttp/registry_persistence_create.py

@@ -3,7 +3,7 @@
 # ATT&CK: T1015, T1103
 # Description: Creates registry persistence for mock malware in Run and RunOnce keys, Services and debuggers.
 
-import _winreg as wreg
+import winreg
 import time
 import common
 
@@ -15,18 +15,18 @@ def pause():
 
 
 def write_reg_string(hive, key, value, data, delete=True):
-    hkey = wreg.CreateKey(hive, key)
+    hkey = winreg.CreateKey(hive, key)
     key = key.rstrip('\\')
     common.log("Writing to registry %s\\%s -> %s" % (key, value, data))
-    wreg.SetValueEx(hkey, value, 0, wreg.REG_SZ, data)
-    stored, code = wreg.QueryValueEx(hkey, value)
+    winreg.SetValueEx(hkey, value, 0, winreg.REG_SZ, data)
+    stored, code = winreg.QueryValueEx(hkey, value)
     if data != stored:
         common.log("Wrote %s but retrieved %s" % (data, stored), log_type="-")
 
     if delete:
         pause()
         common.log("Removing %s\\%s" % (key, value), log_type="-")
-        wreg.DeleteValue(hkey, value)
+        winreg.DeleteValue(hkey, value)
 
     hkey.Close()
     pause()
@@ -37,39 +37,39 @@ def write_reg_string(hive, key, value, data, delete=True):
 def main():
     common.log("Suspicious Registry Persistence")
 
-    for hive in (wreg.HKEY_LOCAL_MACHINE, wreg.HKEY_CURRENT_USER):
+    for hive in (winreg.HKEY_LOCAL_MACHINE, winreg.HKEY_CURRENT_USER):
         write_reg_string(hive, "Software\\Microsoft\\Windows\\CurrentVersion\\RunOnce\\", "RunOnceTest", TARGET_APP)
         write_reg_string(hive, "Software\\Microsoft\\Windows\\CurrentVersion\\Run\\", "RunTest", TARGET_APP)
 
     # create Services subkey for "ServiceTest"
     common.log("Creating ServiceTest registry key")
-    hkey = wreg.CreateKey(wreg.HKEY_LOCAL_MACHINE, "System\\CurrentControlSet\\Services\\ServiceTest\\")
+    hkey = winreg.CreateKey(winreg.HKEY_LOCAL_MACHINE, "System\\CurrentControlSet\\Services\\ServiceTest\\")
 
     # create "ServiceTest" data values
     common.log("Updating ServiceTest metadata")
-    wreg.SetValueEx(hkey, "Description", 0, wreg.REG_SZ, "A fake service")
-    wreg.SetValueEx(hkey, "DisplayName", 0, wreg.REG_SZ, "ServiceTest Service")
-    wreg.SetValueEx(hkey, "ImagePath", 0, wreg.REG_SZ, "c:\\ServiceTest.exe")
-    wreg.SetValueEx(hkey, "ServiceDLL", 0, wreg.REG_SZ, "C:\\ServiceTest.dll")
+    winreg.SetValueEx(hkey, "Description", 0, winreg.REG_SZ, "A fake service")
+    winreg.SetValueEx(hkey, "DisplayName", 0, winreg.REG_SZ, "ServiceTest Service")
+    winreg.SetValueEx(hkey, "ImagePath", 0, winreg.REG_SZ, "c:\\ServiceTest.exe")
+    winreg.SetValueEx(hkey, "ServiceDLL", 0, winreg.REG_SZ, "C:\\ServiceTest.dll")
 
     # modify contents of ServiceDLL and ImagePath
     common.log("Modifying ServiceTest binary")
-    wreg.SetValueEx(hkey, "ImagePath", 0, wreg.REG_SZ, "c:\\ServiceTestMod.exe")
-    wreg.SetValueEx(hkey, "ServiceDLL", 0, wreg.REG_SZ, "c:\\ServiceTestMod.dll")
+    winreg.SetValueEx(hkey, "ImagePath", 0, winreg.REG_SZ, "c:\\ServiceTestMod.exe")
+    winreg.SetValueEx(hkey, "ServiceDLL", 0, winreg.REG_SZ, "c:\\ServiceTestMod.dll")
 
     hkey.Close()
     pause()
 
     # delete Service subkey for "ServiceTest"
     common.log("Removing ServiceTest", log_type="-")
-    hkey = wreg.CreateKey(wreg.HKEY_LOCAL_MACHINE, "System\\CurrentControlSet\\Services\\")
-    wreg.DeleteKeyEx(hkey, "ServiceTest")
+    hkey = winreg.CreateKey(winreg.HKEY_LOCAL_MACHINE, "System\\CurrentControlSet\\Services\\")
+    winreg.DeleteKeyEx(hkey, "ServiceTest")
 
     hkey.Close()
     pause()
 
     # Additional persistence
-    hklm = wreg.HKEY_LOCAL_MACHINE
+    hklm = winreg.HKEY_LOCAL_MACHINE
     common.log("Adding AppInit DLL")
     windows_base = "Software\\Microsoft\\Windows NT\\CurrentVersion\\Windows\\"
     write_reg_string(hklm, windows_base, "AppInit_Dlls", "evil.dll", delete=False)
@@ -84,7 +84,7 @@ def main():
     for victim in debugger_targets:
         common.log("Registering Image File Execution Options debugger for %s -> %s" % (victim, TARGET_APP))
         base_key = "Software\\Microsoft\\Windows NT\\CurrentVersion\\Image File Execution Options\\%s" % victim
-        write_reg_string(wreg.HKEY_LOCAL_MACHINE, base_key, "Debugger", TARGET_APP, delete=True)
+        write_reg_string(winreg.HKEY_LOCAL_MACHINE, base_key, "Debugger", TARGET_APP, delete=True)
 
 
 if __name__ == "__main__":

red_ttp/scrobj_com_hijack.py

@@ -3,7 +3,7 @@
 # ATT&CK: T1122
 # Description: Modifies the Registry to create a new user-defined COM broker, "scrobj.dll".
 
-import _winreg as winreg
+import winreg
 import common
 
 
@@ -19,7 +19,7 @@ def main():
     winreg.DeleteValue(hkey, "")
     winreg.DeleteKey(hkey, "")
     winreg.CloseKey(hkey)
-    
+
     hkey = winreg.CreateKey(winreg.HKEY_CURRENT_USER, "SOFTWARE\\Classes\\CLSID")
     winreg.DeleteKey(hkey, "{00000000-0000-0000-0000-0000DEADBEEF}")
     winreg.CloseKey(hkey)

red_ttp/sip_provider.py

@@ -4,7 +4,7 @@
 # Description: Registers a mock SIP provider to bypass code integrity checks and execute mock malware.
 
 import os
-import _winreg as winreg
+import winreg
 import common
 
 

red_ttp/smb_connection.py

@@ -16,12 +16,11 @@ def main(ip=common.LOCAL_IP):
     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     s.connect((ip, 445))
     common.log("Sending HELLO")
-    s.send("HELLO!")
+    s.send("HELLO!".encode())
     common.log("Shutting down the conection...")
     s.close()
     common.log("Closed connection to {}:{}".format(ip, SMB_PORT))
 
 
 if __name__ == "__main__":
     exit(main(*sys.argv[1:]))
-

red_ttp/trust_provider.py

@@ -4,7 +4,7 @@
 # Description: Substitutes an invalid code authentication policy, enabling trust policy bypass.
 
 import os
-import _winreg as winreg
+import winreg
 import common
 
 FINAL_POLICY_KEY = "Software\\Microsoft\\Cryptography\\providers\\trust\\FinalPolicy\\{00AAC56B-CD44-11D0-8CC2-00C04FC295EE}"

red_ttp/uac_eventviewer.py

@@ -4,7 +4,7 @@
 # Description: Modifies the Registry value to change the handler for MSC files, bypassing UAC.
 
 import sys
-import _winreg as winreg
+import winreg
 import common
 
 # Default machine value:

red_ttp/uac_sdclt.py

@@ -6,7 +6,7 @@
 import subprocess
 import sys
 import os
-import _winreg as winreg
+import winreg
 import common
 
 # HKCU:\Software\Classes\exefile\shell\runas\command value: IsolatedCommand
@@ -26,7 +26,7 @@ def main(target_process=common.get_path("bin", "myapp.exe")):
 
     common.log("Running Sdclt to bypass UAC")
     common.execute([r"c:\windows\system32\sdclt.exe", "/KickOffElev"])
-    
+
     common.log("Clearing registry keys", log_type="-")
     winreg.DeleteValue(hkey, "IsolatedCommand")
     winreg.DeleteKey(hkey, "")