某借款软件逆向实战实现自动加解密+重新签名

javascript                  // Hook aes 加密方法                  function aeshook(){                  // 指定类                  var HttpUtils = Java.use('com.a.a.network.HttpUtils');                  // 重写 AESEncipherMapBody 方法                  // 保留原始的 AESEncipherMapBody 方法                  var originalAESEncipherMapBody = HttpUtils.AESEncipherMapBody;                  HttpUtils.AESEncipherMapBody.implementation = function (map) {                  // 打印传入的参数                  console.log("发送验证码加密入参：");                  var mapEntries = Java.cast(map.entrySet(), Java.use('java.util.Set'));                  var iterator = mapEntries.iterator();                  while (iterator.hasNext()) {                  var entry = Java.cast(iterator.next(),Java.use('java.util.Map$Entry'));                  console.log(entry.getKey() + ": " + entry.getValue());                  }                  // 调用原始方法获取返回值                  var result = originalAESEncipherMapBody.call(this, map);                  try {                  var buffer = Java.use('okio.Buffer');                  var bufferInstance = buffer.$new();                  result.writeTo(bufferInstance);                  var content = bufferInstance.readUtf8();                  console.log("原始方法的返回值", content);                  } catch (e) {                  console.log("读取 RequestBody 内容时出错：", e)                  }                  return result;                  }                  }                  Java.perform(function(){                  // 启动入口                  aeshook()                  })                  

javascript                  function hookAesKey() {                  // 指定类                  var AESCBCUtil = Java.use('com.a.a.network.aes.AESCBCUtil')                  // 重写 encrypt 方法                  // 保留原有 encrypt 方法                  var originalencrypt = AESCBCUtil.encrypt;                  AESCBCUtil.encrypt.implementation = function (str1,str2,str3) {                  //打印入参                  console.log("待加密值：", str1)                  console.log("加密key: ", str2)                  console.log("加密iv: ", str3)                  var result = originalencrypt.call(this, str1, str2, str3)                  return result                  }                  }                  Java.perform(function(){                  // 启动入口                  hookAesKey()                  })

python                  # -*- conding: utf-8 -*-                  from Crypto.Cipher import AES                  from Crypto.Util.Padding import pad, unpad                  import base64                  def encrypt(plaintext, key, iv):                  try:                  # 创建 AES 加密器，使用 CBC 模式                  cipher = AES.new(key.encode('utf-8'), AES.MODE_CBC, iv.encode('utf-8'))                  # 对明文进行填充，使其长度为 AES 块大小的整数倍                  padded_plaintext = pad(plaintext.encode('utf-8'), AES.block_size)                  # 进行加密操作                  ciphertext = cipher.encrypt(padded_plaintext)                  # 对密文进行 Base64 编码                  return base64.b64encode(ciphertext).decode('utf-8')                  except Exception as e:                  print(f"加密时出现异常: {e}")                  return None                  def decrypt(ciphertext, key, iv):                  try:                  # 对 Base64 编码的密文进行解码                  ciphertext = base64.b64decode(ciphertext)                  # 创建 AES 解密器，使用 CBC 模式                  cipher = AES.new(key.encode('utf-8'), AES.MODE_CBC, iv.encode('utf-8'))                  # 进行解密操作                  decrypted_data = cipher.decrypt(ciphertext)                  # 去除填充                  unpadded_data = unpad(decrypted_data, AES.block_size)                  return unpadded_data.decode('utf-8')                  except Exception as e:                  print(f"解密时出现异常: {e}")                  return None                  if __name__ == '__main__':                  text = "13333333333"                  key = "8111111188201096"                  iv = "01111111111111"                  print(encrypt(text, key, iv))

python                  import json                  import base64                  import hashlib                  from fastapi import FastAPI                  from Crypto.Cipher import AES                  from Crypto.Util.Padding import pad, unpad                  from _base_classes import *                  IV = "0111111111111111"                  JSON_KEY = "data"                  app = FastAPI()                  @app.post("/hookRequestToBurp", response_model=RequestModel)                  async def hook_request_to_burp(request: RequestModel):                  """HTTP请求从数据客户端到达Burp时被调用。在此处完成请求解密的代码就可以在Burp中看到明文的请求报文。"""print("--------------------------------")                  request_dict = json.loads(request.model_dump_json())                  print("原client向burp请求体")                  print(request_dict)                  content_base64 = request_dict.get("contentBase64")                  timestamp = request_dict["headers"]["X-Timestamp"][0]                  aes_key = f"8222222f{timestamp[-8:]}"                  print("AES Key: " + aes_key)                  if content_base64:                  # 解码 base64                  content_plaintext = base64.b64decode(content_base64)                  # 解码二进制为json字符串                  content_json = content_plaintext.decode("utf-8")                  # json 转 python对象                  content_dict = json.loads(content_json)                  # 获取待解密的值                  need_decrypt = content_dict.get("content")                  # 判断是否有需要解密的数据                  if need_decrypt is not None:                  # 调用解密函数，解密出要传给 burp 的明文                  plaintext = decrypt(need_decrypt, aes_key, IV)                  # 将解密后的值进行base64，准备替换原本 request_dict 中的 contentBase64 的值                  plaintext_base64 = base64.b64encode(plaintext.encode("utf-8"))                  request_dict["contentBase64"] = plaintext_base64.decode("utf-8")                  # 创建新的 RequestModel 实例，将其传回给原来的请求                  new_request = RequestModel(**request_dict)                  return new_request                  else:                  return request                  else:                  return request                  @app.post("/hookRequestToServer", response_model=RequestModel)                  async def hook_request_to_server(request: RequestModel):                  """HTTP请求从Burp将要发送到Server时被调用。在此处完成请求加密的代码就可以将加密后的请求报文发送到Server。"""print("--------------------------------")                  request_dict = json.loads(request.model_dump_json())                  print("原burp向server请求体")                  print(request_dict)                  content_base64 = request_dict.get("contentBase64")                  timestamp = request_dict["headers"]["X-Timestamp"][0]                  aes_key = f"8222222f{timestamp[-8:]}"                  print("AES Key: " + aes_key)                  if content_base64:                  # 解码 base64                  content_plaintext = base64.b64decode(content_base64)                  # 解码二进制为json字符串                  need_encrypt = content_plaintext.decode("utf-8")                  print("原需要加密传输的请求体为：" + need_encrypt)                  # 对请求体进行二次加签，防止修改请求后签名不过                  need_encrypt_dict = json.loads(need_encrypt)                  print("旧签名为: " + need_encrypt_dict["sign"])                  need_encrypt_dict["sign"] = get_sign(need_encrypt_dict)                  print("新签名为: " + need_encrypt_dict["sign"])                  need_encrypt_dict["timestamp"] = timestamp                  need_encrypt = json.dumps(need_encrypt_dict)                  print("新需要加密传输的请求体为：" + need_encrypt)                  # 判断是否有需要加密的值                  if need_encrypt is not None:                  # 调用加密函数，加密出要传给 server 的密文                  ciphertext = encrypt(need_encrypt, aes_key, IV)                  print("Ciphertext: " + ciphertext)                  # 构造原client请求                  origin_ciphertext = json.dumps({"content": ciphertext})                  # 将加密后的值进行base64，准备替换原本 request_dict 中的 contentBase64 的值                  plaintext_base64 = base64.b64encode(origin_ciphertext.encode("utf-8"))                  request_dict["contentBase64"] = plaintext_base64.decode("utf-8")                  # 创建新的 RequestModel 实例，将其传回给原来的请求                  new_request = RequestModel(**request_dict)                  return new_request                  else:                  return request                  else:                  return request                  @app.post("/hookResponseToBurp", response_model=ResponseModel)                  async def hook_response_to_burp(response: ResponseModel):                  """HTTP响应从Server到达Burp时被调用。在此处完成响应解密的代码就可以在Burp中看到明文的响应报文。"""print("--------------------------------")                  # 获取server向burp的响应体                  response_dict = json.loads(response.model_dump_json())                  print("原server向burp的响应体")                  print(response_dict)                  if response_dict.get("contentBase64"):                  content_base64 = response_dict.get("contentBase64")                  # base64解码响应体，以便获取时间戳                  content_b64decode_json = base64.b64decode(content_base64).decode("utf-8")                  content_b64decode_dict = json.loads(content_b64decode_json)                  timestamp = content_b64decode_dict["timestamp"]                  aes_key = f"8222222f{timestamp[-8:]}"                  print("AES Key: " + aes_key)                  print(content_b64decode_dict)                  need_decrypt = content_b64decode_dict["data"]                  # 判断是否有需要解密的值                  if need_decrypt is not None:                  # 对其中的 data 进行解密                  plaintext = decrypt(need_decrypt, aes_key, IV)                  # 拼接解密后的数据到原来的响应体中                  content_b64decode_dict["data"] = plaintext                  # 将新的响应体base64后，准备发送给burp                  plaintext_base64 = base64.b64encode(json.dumps(content_b64decode_dict, ensure_ascii=False).encode("utf-8"))                  response_dict["contentBase64"] = plaintext_base64                  new_response = ResponseModel(**response_dict)                  print("新响应体")                  print(new_response)                  return new_response                  else:                  return response                  else:                  return response                  @app.post("/hookResponseToClient", response_model=ResponseModel)                  async def hook_response_to_client(response: ResponseModel):                  """HTTP响应从Burp将要发送到Client时被调用。在此处完成响应加密的代码就可以将加密后的响应报文返回给Client。"""print("--------------------------------")                  # 获取 burp 向 client 的响应体                  response_dict = json.loads(response.model_dump_json())                  print("原burp向client响应体")                  print(response_dict)                  if response_dict.get("contentBase64"):                  content_base64 = response_dict.get("contentBase64")                  # base64 解码响应体                  content_b64decode_json = base64.b64decode(content_base64).decode("utf-8")                  content_b64decode_dict = json.loads(content_b64decode_json)                  timestamp = content_b64decode_dict["timestamp"]                  aes_key = f"8222222f{timestamp[-8:]}"                  print('AES Key: ' + aes_key)                  # 获取待加密值，并调用加密函数                  need_decrypt = content_b64decode_dict["data"]                  # 判断是否有需要加密的值                  if need_decrypt is not None:                  ciphertext = encrypt(need_decrypt, aes_key, IV)                  print("Ciphertext: " + ciphertext)                  # 将加密值放回到原有的字典中，并生成json字符串                  content_b64decode_dict["data"] = ciphertext                  ciphertext_base64 = base64.b64encode(json.dumps(content_b64decode_dict, ensure_ascii=False).encode("utf-8"))                  response_dict["contentBase64"] = ciphertext_base64                  new_response = ResponseModel(**response_dict)                  print("新响应体")                  print(new_response)                  return new_response                  else:                  return response                  else:                  return response                  def decrypt(ciphertext, key, iv):                  try:                  # 对 Base64 编码的密文进行解码                  ciphertext = base64.b64decode(ciphertext)                  # 创建 AES 解密器，使用 CBC 模式                  cipher = AES.new(key.encode('utf-8'), AES.MODE_CBC, iv.encode('utf-8'))                  # 进行解密操作                  decrypted_data = cipher.decrypt(ciphertext)                  print(decrypted_data)                  # 去除填充                  unpadded_data = unpad(decrypted_data, AES.block_size)                  return unpadded_data.decode('utf-8')                  except Exception as e:                  print(f"解密时出现异常: {e}")                  return None                  def encrypt(plaintext, key, iv):                  try:                  # 创建 AES 加密器，使用 CBC 模式                  cipher = AES.new(key.encode('utf-8'), AES.MODE_CBC, iv.encode('utf-8'))                  # 对明文进行填充，使其长度为 AES 块大小的整数倍                  padded_plaintext = pad(plaintext.encode('utf-8'), AES.block_size)                  # 进行加密操作                  ciphertext = cipher.encrypt(padded_plaintext)                  print(ciphertext)                  # 对密文进行 Base64 编码                  return base64.b64encode(ciphertext).decode("utf-8")                  except Exception as e:                  print(f"加密时出现异常: {e}")                  return None                  def string_no_null(obj):                  return obj is not None and obj != "" and obj != "null" and obj != " "                  def encrypt_md5(text):                  md5 = hashlib.md5()                  md5.update(text.encode('utf-8'))                  return md5.hexdigest()                  def get_sign(request_data):                  request_data.pop('sign', None)                  if request_data["timestamp"]:                  request_data["soltSignKey"] = encrypt_md5(request_data["timestamp"])                  tree_map = {}                  for key, value in request_data.items():                  if not isinstance(value, (dict, list)) and string_no_null(value):                  tree_map[key] = str(value)                  sorted_tree_map = dict(sorted(tree_map.items()))                  input_str = ""                  for v in sorted_tree_map.values():                  input_str += v + "&"                  if input_str:                  input_str = input_str[:-1]                  return encrypt_md5(input_str)                  if __name__ == "__main__":                  # 多进程启动                  # uvicorn aes_cbc_query:app --host 0.0.0.0 --port 5000 --workers 4                  import uvicorn                  uvicorn.run(app, host="0.0.0.0", port=5000)