Galaxy实现HTTP自动加解密

‌Galaxy插件‌是一款强大的加解密工具，特别适用于Burp Suite环境，能够帮助渗透测试人员在处理加密报文时提供极大的便利。

Galaxy

项目地址：https://github.com/outlaws-bai/Galaxy

Galaxy可以HOOK如下4个过程，插件会根据自定义的HOOK脚本自动化加解密后续代理的流量，并且可以与sqlmap、xray等扫描工具联动。

request.host修改成需要加解密的网站

AES固定密钥

js看到默认key和iv是1234567890123456，加密模式为AES-CBC

测试过程中发现加密的body在经过AES—BASE64后还需要url编码一下，不然会失败。

由于不存在响应加解密，只需要实现hook_request_to_burp和hook_request_to_server函数即可。

import base64
from urllib.parse import unquote_to_bytes, quote_from_bytes

from fastapi import FastAPI
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from _base_classes import *

KEY = b"1234567890123456"
IV = b"1234567890123456"
JSON_KEY = "data"
app = FastAPI()

@app.post("/hookRequestToBurp", response_model=RequestModel)
async def hook_request_to_burp(request: RequestModel):
    print("------------------------------------------浏览器发送请求------------------------------------------------------")
    """HTTP请求从客户端到达Burp时被调用。在此处完成请求解密的代码就可以在Burp中看到明文的请求报文。"""
    # 获取需要解密的数据
    encrypted_data: bytes = get_data(request.content)
    # 调用函数解密
    if encrypted_data:
        data: bytes = decrypt(encrypted_data)
    else:
        print("此次请求body为空")
        data: bytes = b""
    # 更新body为已解密的数据
    request.content = data
    return request

@app.post("/hookRequestToServer", response_model=RequestModel)
async def hook_request_to_server(request: RequestModel):
    """HTTP请求从Burp将要发送到Server时被调用，在此处完成请求加密的代码就可以将加密后的请求报文发送到Server。"""
    # 获取并加密请求体
    data: bytes = request.content
    print(data)
    encryptedData: bytes = encrypt(data)
    body: bytes = to_data(encryptedData)
    # 对 body 进行 URL 编码
    url_encoded_body = quote_from_bytes(body).encode()
    new_body = b"encryptedData=" + url_encoded_body
    request.content = new_body
    return request

def decrypt(content: bytes) -> bytes:
    cipher = AES.new(KEY, AES.MODE_CBC, IV)
    return unpad(cipher.decrypt(content), AES.block_size)

def encrypt(content: bytes) -> bytes:
    cipher = AES.new(KEY, AES.MODE_CBC, IV)
    return cipher.encrypt(pad(content, AES.block_size))

def get_data(content: bytes) -> bytes:
    parts = content.split(b"encryptedData=")
    if len(parts) > 1:
        decoded_encrypted_part = unquote_to_bytes(parts[1])
        print(decoded_encrypted_part)
        return base64.b64decode(decoded_encrypted_part)
    else:
        print("未找到 encryptedData= 相关内容")
        return content

def to_data(contnet: bytes) -> bytes:

    return base64.b64encode(contnet)

if __name__ == "__main__":
    # 多进程启动
    # uvicorn aes_cbc:app --host 0.0.0.0 --port 5000 --workers 4
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=5000)

Burp重放为明文，发送后python自动加密，服务端也返回了正确的响应。

AES服务端获取Key

首先获取一个key和iv

然后利用获取的key和iv去发送请求

和上次的代码相比只是增加了key和iv的base64解码，提取内容换成json

import base64
import json
import typing

from fastapi import FastAPI
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from _base_classes import *

KEYBase64 = "n+l5EahgQMBysLZ+EBxvNg=="
IVBase64 = "9igMfnRW3OM+9tyU\/dQUIA=="

KEY = base64.b64decode(KEYBase64)
IV = base64.b64decode(IVBase64)

JSON_KEY = "encryptedData"
app = FastAPI()

@app.post("/hookRequestToBurp", response_model=RequestModel)
async def hook_request_to_burp(request: RequestModel):
    print("------------------------------------------浏览器发送请求------------------------------------------------------")
    """HTTP请求从客户端到达Burp时被调用。在此处完成请求解密的代码就可以在Burp中看到明文的请求报文。"""
    # 获取需要解密的数据
    encrypted_data: bytes = get_data(request.content)
    # 调用函数解密
    if encrypted_data:
        data: bytes = decrypt(encrypted_data)
    else:
        print("此次请求body为空")
        data: bytes = b""
    # 更新body为已解密的数据
    request.content = data
    return request

@app.post("/hookRequestToServer", response_model=RequestModel)
async def hook_request_to_server(request: RequestModel):
    """HTTP请求从Burp将要发送到Server时被调用，在此处完成请求加密的代码就可以将加密后的请求报文发送到Server。"""
    # 获取并加密请求体
    data: bytes = request.content
    print(data)
    encryptedData: bytes = encrypt(data)
    body: bytes = to_data(encryptedData)
    request.content = body
    return request

def decrypt(content: bytes) -> bytes:
    cipher = AES.new(KEY, AES.MODE_CBC, IV)
    return unpad(cipher.decrypt(content), AES.block_size)

def encrypt(content: bytes) -> bytes:
    cipher = AES.new(KEY, AES.MODE_CBC, IV)
    return cipher.encrypt(pad(content, AES.block_size))

def get_data(content: bytes) -> bytes:
    body_json: typing.Dict = json.loads(content)
    return base64.b64decode(body_json[JSON_KEY])

def to_data(contnet: bytes) -> bytes:
    body_json = {}
    body_json[JSON_KEY] = base64.b64encode(contnet).decode()
    return json.dumps(body_json).encode()

if __name__ == "__main__":
    # 多进程启动
    # uvicorn aes_cbc:app --host 0.0.0.0 --port 5000 --workers 4
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=5000)

测试案例

测试地址：http://39.98.108.20:8085/

请求体有三个参数Requestid、Sign、timestamp

timestamp是一个时间戳，现在需要找Requestid和Sign

Requestid来源

生成一个符合 UUIDv4（UUID版本4）规范的随机字符串

Sign来源

Sign = MD5(请求体+Requestid+时间戳)

构造请求体的Requestid、Sign、timestamp可以正常获取数据

使用Galaxy提供的模板修改代码，实现更换headers三个参数以及请求体和响应体的加解密。

import hashlib
import json
import base64
import time
import random
import uuid

from fastapi import FastAPI
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from _base_classes import *


KEY = b"1234567891234567"
IV = b"1234567891234567"

app = FastAPI()


@app.post("/hookRequestToBurp", response_model=RequestModel)
async def hook_request_to_burp(request: RequestModel):
    print("------------------------------------------发送请求------------------------------------------------------")
    """HTTP请求从客户端到达Burp时被调用。在此处完成请求解密的代码就可以在Burp中看到明文的请求报文。"""
    # 获取需要解密的数据
    encrypted_data: bytes = get_data(request.content)
    # 调用函数解密
    if encrypted_data:
        data: bytes = decrypt(encrypted_data)
    else:
        print("此次请求body为空")
        data: bytes = b""
    # 更新body为已解密的数据
    request.content = data
    return request

@app.post("/hookRequestToServer", response_model=RequestModel)
async def hook_request_to_server(request: RequestModel):
    """HTTP请求从Burp将要发送到Server时被调用，在此处完成请求加密的代码就可以将加密后的请求报文发送到Server。"""
    # 获取并修改 headers
    my_headers = dict(request.headers)  # 转换为可修改的字典

    #生成requesdid和timestamp
    request_id = uuid.uuid4().hex
    timestamp = str(int(time.time() * 1000))

    #获取body，获取生成的headers
    data: bytes = request.content
    headers = generate_sign(data,request_id,timestamp)

    # 添加 request-id,UUIDv4格式
    my_headers['requestId'] = [headers["requestId"]]

    # 新增时间戳
    my_headers['timestamp'] = [headers["timestamp"]]

    # 生成sign
    my_headers['sign'] = [headers["sign"]]

    print("sign:", my_headers['sign'])
    print("requestId:", my_headers['requestId'])
    print("timestamp:", my_headers['timestamp'])
    print("sign_source:", headers['sign_source'])
    # 获取并加密请求体
    encryptedData: bytes = encrypt(data)
    body: bytes = to_data(encryptedData)
    request.content = body

    # 更新 headers
    request.headers = my_headers
    
    return request

@app.post("/hookResponseToBurp", response_model=ResponseModel)
async def hook_response_to_burp(response: ResponseModel):
    print("------------------------------------------接受请求------------------------------------------------------")
    """HTTP响应从Server到达Burp时被调用。在此处完成响应解密的代码就可以在Burp中看到明文的响应报文。"""
    if not response.content:
        # 如果内容为空，直接返回
        return response
    try:
        # 获取需要解密的数据
        encryptedData: bytes = response.content

        # 响应体按 HTTP 报文可能分隔符查找
        raw_text = encryptedData.decode(errors="ignore")  # 允许忽略部分非 UTF-8 字节
        split_index = raw_text.find("HTTP/1.1")

        # 截取 HTTP/1.1 前的部分作为 base64 加密数据
        if split_index != -1:
            encrypted_b64 = raw_text[:split_index].strip()
        else:
            # 没有第二个 HTTP 报文，整段就是加密的
            encrypted_b64 = raw_text.strip()

        # 解密
        encrypted_b64 = get_data(encrypted_b64)
        data: bytes = decrypt(encrypted_b64)

        # 更新内容
        response.content = data

        print("未能成功解密，保持原响应体")
        return response
    except Exception as e:
        print(f"未能成功解密，保持原响应体: {e}")
        # 如果解密失败，也返回原始响应
        return response

    return response

@app.post("/hookResponseToClient", response_model=ResponseModel)
async def hook_response_to_client(response: ResponseModel):
    """HTTP响应从Burp将要发送到Client时被调用。在此处完成响应加密的代码就可以将加密后的响应报文返回给Client。"""
    # 获取需要加密的数据
    encrypted_data: bytes = encrypt(response.content)
    # 调用函数加密
    data: bytes = to_data(encrypted_data)
    # 更新body为已加密的数据
    response.content = data
    return response

def decrypt(content: bytes) -> bytes:
    cipher = AES.new(KEY, AES.MODE_CBC, IV)
    return unpad(cipher.decrypt(content), AES.block_size)

def encrypt(content: bytes) -> bytes:
    cipher = AES.new(KEY, AES.MODE_CBC, IV)
    return cipher.encrypt(pad(content, AES.block_size))

def get_data(content: bytes) -> bytes:
    try:
        return base64.b64decode(content)
    except Exception as e:
        print("Base64 解码失败:", e)
        return b""


def to_data(contnet: bytes) -> bytes:
    return base64.b64encode(contnet)


#获取requestid
def generate_uuid_v4():
    t = "0123456789abcdef"
    e = [random.choice(t) for _ in range(32)]
    e[14] = "4"
    e[19] = random.choice("89ab")
    e[8] = e[13] = e[18] = e[23]
    return "".join(e)

#获取sign
def generate_sign(raw_body: bytes, request_id: str, timestamp: str) -> str:
    try:
        body_dict = json.loads(raw_body.decode('utf-8'))
    except Exception as e:
        print("请求体不是有效的 JSON：", e)
        body_dict = {}

    body_str = json.dumps(body_dict, separators=(',', ':'), ensure_ascii=False)

    sign_source = body_str + request_id + timestamp
    sign = hashlib.md5(sign_source.encode('utf-8')).hexdigest()

    return {
        "sign": sign,
        "requestId": request_id,
        "timestamp": timestamp,
        "sign_source": sign_source  
    }

if __name__ == "__main__":
    # 多进程启动
    # uvicorn aes_cbc:app --host 0.0.0.0 --port 5000 --workers 4
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=5000)

Tips：Burp大小写敏感，put到burp的参数要以http数据包为准，不然会出现多个参数。