1. 程式人生 > >BaseProxy:非同步http/https中間人

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外掛.

0883f5c67029006e0763da718f7536ba.png

2.設定好代理,並將baseproxy執行後,訪問www.baidu.com.

b3c200d0d74defb8b80cff3cbd5c48e3.png

3.這時候訪問被拒絕,需要安裝證書.在當前網頁訪問 baseproxy.ca,下載證書.

5929234995d2868ee4928052a70f2f45.png

4.雙擊下載的證書,並安裝到合法機構中.

77a823fca5b9a4765bf1f7478851ce0d.png

d993a5e66887de15297afafa5a408349.png
d994706e303d68f8a0152d2d510b5997.png
5.接著訪問百度就可以了.

c2fb0549a86bd98e26c52c617b0395fb.png

注意:只有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()

效果如下:

cb202f3182fe9a5d39820fef4516204b.png

參考專案

福利大放送

關注公眾號:七夜安全部落格

d929ae9a466c0f5e20148d538bbc8e17.png

  • 回覆【1】:領取 Python資料分析 教程大禮包
  • 回覆【2】:領取 Python Flask 全套教程
  • 回覆【3】:領取 某學院 機器學習 教程
  • 回覆【4】:領取 爬蟲 教程

知識星球已經50人了,隨著人數的增多,價格之後會上漲,越早關注越多優惠。星球的福利有很多:

  • 比如上面的教程,已經提前在知識星球中分享
  • 可以發表一些問題,大家一塊解決
  • 我之後寫的電子書,錄製的教學視訊,對於知識星球的朋友都是優惠的(基本上免費)
  • 一些節假日會給大家發個紅包或者贈書

3899727832f58c3e9359943ed9a2e047.png