Pyhton模拟Proxy实现MITM

参考了Python渗透测试的思路，只是修改消息传递方式为queue，固定有客户端发起链接。

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

# -*- coding: utf-8 -*-

import socket

import ssl

import threading

import queue

import time

fromCliQueue = queue.Queue()

fromSrvQueue = queue.Queue()

sFlag = False

## Client -- proxyServer -- proxyClient -- Server

class proxyClient(threading.Thread):

def __init__(self):

threading.Thread.__init__(self)

self.sock = None

self.ssl_context = ssl.create_default_context(cafile='RootCA.pem')

self.ssl_context.load_cert_chain("Client.pem", "Client.key")

self.ssl_context.protocol = ssl.PROTOCOL_TLSv1_2

self.ssl_context.check_hostname = False

def run(self):

global fromCliQueue

global fromSrvQueue

while True:

#if mesg queue from client is not empty, send it to server and recv response

if not fromCliQueue.empty():

if self.sock is None:

self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)

self.sock.connect(('127.0.0.1', 9999))

else:

sdata = fromCliQueue.get()

print("[proxyClient]send to server: {}".format(sdata))

self.sock.send(sdata)

rdata = self.sock.recv(4096)

#解决recv总是收到空字符

if rdata != b'':

print("[proxyClient]recv from server: {}".format(rdata))

fromSrvQueue.put(rdata)

self.sslDetected(rdata)

def sslDetected(self, data):

if b'\x01\x00\x00\x00' in data:

self.sock = self.ssl_context.wrap_socket(self.sock)

print("[proxyClient]ssl neogotiation")

return True

else:

return False

class proxyServer(threading.Thread):

def __init__(self):

threading.Thread.__init__(self)

self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)

self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

self.sock.bind(('127.0.0.1',8888))

self.sock.listen(1)

self.ssl_context = ssl.SSLContext()

# self.ssl_context.verify_mode = ssl.CERT_REQUIRED

self.ssl_context.load_verify_locations(cafile='RootCA.pem')

# self.ssl_context.check_hostname = False

# test this protocol with client's PROTOCOL_TLSv1_2

self.ssl_context.protocol = ssl.PROTOCOL_TLS_SERVER

self.ssl_context.load_cert_chain("Server.pem", "Server.key")

def run(self):

global fromSrvQueue

global fromCliQueue

self.proxySocket, addr = self.sock.accept()

while self.proxySocket:

rdata = self.proxySocket.recv(4096)

if rdata != b'':

print("[proxyServer]recv from client: {}".format(rdata))

fromCliQueue.put(rdata)

while True:

if fromSrvQueue.empty():

time.sleep(1)

continue

else:

sdata = fromSrvQueue.get()

print("[proxyServer]send to client: {}".format(sdata))

self.proxySocket.send(sdata)

self.sslDetected(sdata)

break;

def sslDetected(self, data):

if b'\x01\x00\x00\x00' in data:

self.proxySocket = self.ssl_context.wrap_socket(self.proxySocket, server_side=True)

print("[proxyClient]ssl neogotiation")

return True

else:

return False

if __name__ == "__main__":

proxyCli = proxyClient()

proxyCli.daemon = True

proxyCli.start()

proxySrv = proxyServer()

proxySrv.daemon = True

proxySrv.start()

proxySrv.join()

sslDetected的作用是为了匹配某种特殊的情况： client-server先交互一些普通socket数据，然后开始ssl协商假设已经获取server端和客户端的证书，此Proxy脚本可以伪造客户端和服务端，实现MITM的场景。如果 […]

re子组和xml.etree.ElementTree写文件

关键信息临时中转╭(╯^╰)╮，

import xml.etree.ElementTree as ET

import re

def handle_version_str(src_str):

# 获取匹配的自组数据

m = re.search(r'(.*)(V\d{3}R\d{3}C\d{2})(.*)', src_str)

if m is not None:

ver = m.group(2)

print(ver)

# 通过获取的匹配内容，将其删除

dst_str = re.sub(ver, '', src_str)

else:

return None

return dst_str

def create_result_xml():

model = '''<data><country name="China">test</country></data>'''

root = ET.fromstring(model)

# 添加子节点

sub_attrib = {"location": "cn", "color":"black"}

root.append("sub1", sub_attrib)

# 如何通过ET写入不是parse方式创建的ET到一个xml文件

tree = ET.ElementTree(root)

tree.write('result.xml')

if __name__ == "__main__":

ver_str = "Hello_V001R002C20_Release"

dst_str = handle_version_str(ver_str)

print(dst_str)

create_result_xml()

socketserver

#-*- coding: utf-8 -*-

import socketserver

import socket

"""

测试环境：Ubuntu

Pyhton3.6.5

web访问: http://127.0.0.1:9999 通过

curl: curl http://127.0.0.1:9999 通过

"""

class myHandler(socketserver.StreamRequestHandler):

def handle(self):

self.rdata = self.rfile.readline()

# 测试for方法独到最后会持续阻塞

# for item in self.rfile:

# print("{} write: {}".format(self.client_address[0], item))

self.function1()

self.wfile.write(b'HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n\r\n<html>hello</html>')

def function1(self):

if 'GET' in self.rdata.decode('utf-8'):

self.function2()

def function2(self):

"""测试rfile中的数据可以缓存后续继续read"""

while True:

item = self.rfile.readline()

# 此处无法设置非阻塞，通过判断http请求中最后一组数据为\r\n来退出循环

if item == b"\r\n":

break

print("{} write: {}".format(self.client_address[0], item))

if __name__ == "__main__":

# 全局设置端口重用

socketserver.ThreadingTCPServer.allow_reuse_address = True

server = socketserver.ThreadingTCPServer(('127.0.0.1', 9999), myHandler)

# 设置端口重用

# server.allow_reuse_address = True

# 网上关于设置socket参数方法

# server.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)

# 设置非阻塞不生效

server.socket.setblocking(False)

server.serve_forever()

python2: SockServer python3: socketserver 如果pyhton2一直提示找不到模块，尝试 pip install Werkzeug --upgrade

sniff通过signal信号正常终止的一个实现

由于stop_filter是作用在每一个packet的，当终止函数返回True的时候，只有在收到包的时候sniff才会返回，所以如果捕获的是明文数据，可以直接通过过滤Raw load方式处理，但是如果都是密文，通过kill方式来结束sniff也是一个可取的办法

from scapy.all import sniff, Raw, wrpcap

import threading

import time

import signal

e = threading.Event()

def handler(signum, frame):

global e

e.set()

def sniff_with_signal():

global e

pcap = sniff(iface=['wlxe84e063348c8', 'lo'], stop_filter=lambda p: e.is_set())

#pcap = sniff(iface=['wlxe84e063348c8', 'lo'],

# stop_filter=lambda p: p.haslayer(Raw) and 'stopflag' in p.getlayer(Raw).load)

wrpcap('/tmp/test.pcap', pcap)

print('pcap has wrote into file /tmp/test.pcap')

def main():

signal.signal(signal.SIGTERM, handler)

thread = threading.Thread(target=sniff_with_signal)

thread.start()

while True:

thread.join(2)

if thread.is_alive():

print('thread is alive')

else:

break

if __name__ == '__main__':

main()

python数字转ASCII字符

数字转字符 - chr() 字符转数字 - ord() 于是就出现一个有意思的事情，chr()可以将数字转化为字符，而且根据python3的IDLE提示，chr()的范围为0<=i<=0x10ffff，如果我们遍历这之间的数值会有什么现象呢，如下：

for i in range(0,0x10ffff):

print(format("%6X: %s") % (i, chr(i)) )

-----------------------------------------------------------------------

FFFF: <br>Traceback (most recent call last):

File "C:\Users\jma\Desktop\chr.py", line 9, in <module>

print(format("%10X: %s") % (i, chr(i)) )

UnicodeEncodeError: 'UCS-2' codec can't encode characters in position 12-12: Non-BMP character not supported in Tk

在超过65535的时候，系统就提示错误了，可是同样的程序在cmd里面却可 […]

selenium-初试登录路由器

selenium的安装很简单，或者说无需安装。如果需要录制脚本，可以下载Firefox的插件selenium IDE；写python的自动化，则只需要安装Language Bindings(pip install selenium)，然后下载对应浏览器的drivers即可（windows平台上的driver都是.exe程序，统一放到某一目录下然后将目录路径加入path即可），操作起来还是蛮方便的， […]

用python体验2038年问题

这是在构造消息码流的填充数据的时遇到的一个问题，本和python不相干的，但是偶发联想，于是试试python中时间处理的方式是不是也有同样的问题。首先通过计算器看一下4字节能表示的最大数值，（windows默认是signed类型，没有找到可以设置unsigned的地方）不过这不影响我们的测试，因为还有qword类型：最高位为符号为，0表示正数，1表示负数，所以4字节能表示的最大正数是0x7FF […]

Python在linux终端模式下的类似聊天消息对话框

首先看一下最终效果，在终端模式下实现类似于聊天的对话框实现方式如下，通过两个list: iwords和swords分别缓存来自终端输入和socket收到的数据，然后再将数据通过屏幕输出并通过socket接口发送出去，通过四个独立的线程分别完成各自的工作： myinput：获取输入区域内的输入数据，并将数据append到iwords和swords myoutput：检测swords长度并通过swo […]

一	二	三	四	五	六	日
1	2	3	4	5	6	7
8	9	10	11	12	13	14
15	16	17	18	19	20	21
22	23	24	25	26	27	28
29	30

mowblog

一群散乱的记忆

标签： python

Pyhton模拟Proxy实现MITM

re子组和xml.etree.ElementTree写文件

socketserver

sniff通过signal信号正常终止的一个实现

python数字转ASCII字符

selenium-初试登录路由器

用python体验2038年问题

Python在linux终端模式下的类似聊天消息对话框