BaseProxy:非同步http/https中間人
BaseProxy
非同步http/https代理,可攔截並修改報文,可以作為中間人工具.僅支援py3.5+.專案地址:BaseProxy。
意義
BaseProxy專案的本意是為了使HTTP/HTTPS攔截更加純粹,更加易操作,學習成本更低。
在Python領域,中間人工具非常強大和成功的是MitmProxy,但是有些地方不是很喜歡。
- Windows上安裝比較費時費力
- 功能太多了,可惜我用不到這麼多(似乎不是它的錯,哈哈)
- 隨著版本升級,採用外掛化框架,需要定製功能,需要寫個外掛成為它的一部分(我只是想整合它而已).
因此BaseProxy就誕生了,不僅支援HTTPS透明傳輸,還支援HTTP/HTTPS攔截,簡單易用,可以很好地整合到你們的專案中。
安裝
安裝非常簡單,本專案已經發布到PyPI中...
pip3 install baseproxy
使用配置
啟動baseproxy
在test資料夾下,有很多測試用例。以startserver.py為例。
from baseproxy.proxy import AsyncMitmProxy
baseproxy = AsyncMitmProxy(https=True)
baseproxy.serve_forever()
使用上述程式碼,就可以將HTTPServer執行起來了.對程式碼的解釋如下:
https=True
是對https進行解密;https=False
是對於https實行透傳- baseproxy預設執行在8788埠,如果想改變埠的話,修改為
AsyncMitmProxy(server_addr=('',port),https=True)
執行結果如下:
[2018-06-22 18:46:32] INFO HTTPServer is running at address( , 8788 )......
安裝CA證書
1.將chrome瀏覽器代理伺服器設定為127.0.0.1:8788,推薦使用SwitchyOmega外掛.
2.設定好代理,並將baseproxy執行後,訪問www.baidu.com.
3.這時候訪問被拒絕,需要安裝證書.在當前網頁訪問 baseproxy.ca,下載證書.
4.雙擊下載的證書,並安裝到合法機構中.
5.接著訪問百度就可以了.
注意:只有https=True
時,才需要安裝CA證書。
開發
經過上一步的使用配置,baseproxy已經可以正常運行了,但是這樣是遠遠不夠的.baseproxy還提供了介面,方便開發者對http請求和響應進行修改.
介面
baseproxy提供了兩個介面,一個是修改請求,一個是修改響應.
攔截請求
class ReqIntercept(InterceptPlug):
def deal_request(self,request):
pass
對於請求的攔截,需要繼承ReqIntercept類,並重寫其中的deal_request函式.在deal_request函式的最後,需要將修改後的request引數返回出去.
如果想拋棄這個請求,直接返回None.
request引數
deal_request函式中的request引數型別為Request類
成員變數
Name | 型別 | 含義 |
---|---|---|
hostname | str | 域名 |
port | int | 埠 |
command | str | 請求型別 |
path | str | 請求路徑 |
request_version | str | HTTP協議版本 |
成員函式
def set_headers(self,headers)
- headers:型別為dict
- 用於設定頭部
def get_header(self,key):
- key:型別為str
- 用於獲取指定頭部,返回str
def get_headers(self):
- 用於獲取整個頭部,返回為dict
def set_header(self,key,value):
- 頭部 key,型別str
- 頭部 value,型別str
- 用於設定頭資訊
def get_body_data(self):
- 獲取請求體內容,返回型別為bytes
def set_body_data(self,body):
- 設定請求體內容,body型別為bytes
攔截響應
class RspIntercept(InterceptPlug):
def deal_response(self,response):
pass
對於響應的攔截,需要繼承RspIntercept類,並重寫其中的deal_response函式.在deal_response函式的最後,需要將修改後的response引數返回出去.
如果想拋棄這個響應,直接返回None.
response引數
deal_response函式中的response引數型別為Response類
成員變數
Name | 型別 | 含義 |
---|---|---|
hostname | str | 域名 |
port | int | 埠 |
status | int | 狀態碼 |
reason | str | 狀態描述 |
response_version | str | HTTP協議版本 |
request | Request | 響應對應的請求例項 |
成員函式
def set_headers(self,headers)
- headers:型別為dict
- 用於設定頭部
def get_header(self,key):
- key:型別為str
- 用於獲取指定頭部,返回str
def get_headers(self):
- 用於獲取整個頭部,返回為dict
def set_header(self,key,value):
- 頭部 key,型別str
- 頭部 value,型別str
- 用於設定頭資訊
def get_body_data(self):
- 獲取響應體內容,返回型別為bytes
def set_body_data(self,body):
- 設定響應體內容,body型別為bytes
def get_body_str(self,decoding=None):
- decoding:編碼,預設為None,內部採用chardet探測
- 返回響應體,型別為str.如果無法解碼,返回None
def set_body_str(self,body_str,encoding=None):
- encoding:編碼,預設為None,內部採用chardet探測
- 設定響應體,body_str型別為str
註冊攔截外掛
將攔截類完成後,需要註冊到baseproxy中,需要呼叫AsyncMitmProxy的register函式.示例如下:
from baseproxy.proxy import ReqIntercept, RspIntercept, AsyncMitmProxy
__author__ = 'qiye'
__date__ = '2018/6/21 23:35'
class DebugInterceptor(ReqIntercept, RspIntercept):
def deal_request(self, request):
return request
def deal_response(self, response):
return response
if __name__=="__main__":
baseproxy = AsyncMitmProxy(https=False)
baseproxy.register(DebugInterceptor)
baseproxy.serve_forever()
小例子
將淘寶中的所有產品圖片換成我公眾號的二維碼.程式碼在test資料夾的replace_image.py中,內容如下:
from baseproxy.proxy import RspIntercept, AsyncMitmProxy
class ImageInterceptor( RspIntercept):
def deal_response(self, response):
if response.get_header("Content-Type") and 'image' in response.get_header("Content-Type"):
with open("../img/qiye2.jpg",'rb') as f:
response.set_body_data(f.read())
return response
if __name__ == "__main__":
baseproxy = AsyncMitmProxy(https=True)
baseproxy.register(ImageInterceptor)
baseproxy.serve_forever()
效果如下:
參考專案
福利大放送
關注公眾號:七夜安全部落格
- 回覆【1】:領取 Python資料分析 教程大禮包
- 回覆【2】:領取 Python Flask 全套教程
- 回覆【3】:領取 某學院 機器學習 教程
- 回覆【4】:領取 爬蟲 教程
知識星球已經50人了,隨著人數的增多,價格之後會上漲,越早關注越多優惠。星球的福利有很多:
- 比如上面的教程,已經提前在知識星球中分享
- 可以發表一些問題,大家一塊解決
- 我之後寫的電子書,錄製的教學視訊,對於知識星球的朋友都是優惠的(基本上免費)
- 一些節假日會給大家發個紅包或者贈書