今天在GitHub上看到的一款代理工具，小巧易用，推荐一下： https://github.com/kklis/proxy
合并xml文件中attrib中含有相同标记的element
示例xml内容：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
<peach> <DataModel name="test"> <String value="select" mutable="false" /> <String value=" " mutable="false" /> <String value="id" /> <String value="," mutable="false" /> <String value="name" /> <String value=" " mutable="false" /> <String value="from" mutable="false" /> <String value=" " mutable="false" /> <String value="table_for_test" /> </DataModel> </peach> |
如上这段内容，其实是peach数据模型的一段定义，今天通过脚本模拟自动建模，生成的文档含有大量的mutable="false"的内容。根据peach的描述，这一属性的意义可以让peach忽略对其进行mute操作(实际好像还是会mute)，为了简洁起见，减少建模过程中的费脑，直接将数据模型定义文件中重复出现的muta […]
openssl功能汇总
openssl实在是太过于博大精深了，经常会遇到各种需求不得不查看手册或者Google，现在将已经使用过的部分功能分类收集一下，便于日后在此查看。 使用openssl验证OCSP 通过openssl模拟OCSP服务 openssl ocsp -index index.txt -port 8888 -rsigner RootCA.pem -rkey RootCA.key -CA RootCA.pem […]
折叠屏
京东上已经开卖华为Mate X折叠屏手机了，号称展开后8英寸沉浸式体验，5G支持一步到位。暂且不理会5G的速度在资费不变的条件下能不能提供同等时长的业务，只问折叠屏这个新奇的诞生到底能对手机自身带来多大的价值呢？ 防碎？防弯曲？从卖家的宣传里只能看得到折叠两个字，按照这样的思路折叠屏的唯一优点就是抢夺平板电脑的市场！可是在习惯了这种16：9的模式下单手操作的人，都多少时间展开那明晃晃的8寸屏举着双 […]
peach入门示例
peach入门
win-mysql8开启ssl
首先获取配置文件的路径，windows版本简明的方式就是查看mysql的服务，在启动参数中有配置文件的路径，如下： >sc qc mysql80 [SC] QueryServiceConfig 成功 BINARY_PATH_NAME : "C:\Program Files\MySQL\MySQL Server 8.0\bin\mysqld.exe" --defaults-file="C:\P […]
Python内部类通过外部类属性值传递参数一例
为了解决两个线程类通过global 变量传递数据的方法在并发情况下全局变量被覆盖的问题，不得已将两个代理类用一个class来包装，试验了许久才得出这个简单粗暴的方法，留存以备将来参考
|
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 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 |
# coding: utf-8 import threading import queue import time class MainClass: # class attribute to be used in subClass queueTest = None queueStopThread = None def __init__(self): MainClass.queueTest = queue.Queue() MainClass.queueStopThread = queue.Queue() def __del__(self): MainClass.queueTest = None MainClass.queueStopThread = None class SubClass1(threading.Thread): def __init__(self): threading.Thread.__init__(self) def run(self): while MainClass.queueStopThread.empty(): if not MainClass.queueTest.empty(): print("{} - {}".format(threading.currentThread().ident, MainClass.queueTest.get())) print("SubClass1: queueStopThread: {}".format(MainClass.queueStopThread.get())) class SubClass2(threading.Thread): def __init__(self): threading.Thread.__init__(self) def run(self): for i in range(1, 10): MainClass.queueTest.put(i) time.sleep(0.1) MainClass.queueStopThread.put(True) @staticmethod def run_main(): p_subclass1 = MainClass.SubClass1() p_subclass1.daemon = True p_subclass1.start() p_subclass2 = MainClass.SubClass2() p_subclass2.daemon = True p_subclass2.start() p_subclass1.join() del p_subclass1 del p_subclass2 if __name__ == "__main__": tM1 = MainClass() tM2 = MainClass() t1 = threading.Thread(target=tM1.run_main()) t1.daemon = True t2 = threading.Thread(target=tM2.run_main()) t2.daemon = True t1.start() t2.start() main_class_test = MainClass() main_class_test.run_main() |
运行结果如下：
|
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 31 32 33 34 35 |
F:\Python\classInClass\venv\Scripts\python.exe F:/Python/classInClass/classInClass.py 14596 - 1 14596 - 2 14596 - 3 14596 - 4 14596 - 5 14596 - 6 14596 - 7 14596 - 8 14596 - 9 SubClass1: queueStopThread: True 11812 - 1 11812 - 2 11812 - 3 11812 - 4 11812 - 5 11812 - 6 11812 - 7 11812 - 8 11812 - 9 SubClass1: queueStopThread: True 11120 - 1 11120 - 2 11120 - 3 11120 - 4 11120 - 5 11120 - 6 11120 - 7 11120 - 8 11120 - 9 SubClass1: queueStopThread: True Process finished with exit code 0 |
Pyhton模拟Proxy实现MITM
参考了Python渗透测试的思路，只是修改消息传递方式为queue，固定有客户端发起链接。
|
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 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 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的场景。 如果 […]