python渗透测试入门之http流量图片还原

近期收到了电子工业出版社赠送的一本网络安全书籍《python黑帽子》，书中一共24个实验，今天复现第10个实验（http流量图片还原），我的测试环境是mbp电脑+conda开发环境。访问一个http网站，就能抓取该网站的所有图片，实际测试效果一般，不仅抓不全，而且抓的图片存在大量重复，备注：有可能是我选的网站不好～

ailx10

网络安全优秀回答者

网络安全硕士

去咨询

1、访问一个有图片的http网站，注意是80端口的http，不是https，然后开启wireshark抓包，保存为pcap.pcap

2、在mbp上启动脚本，请忽略告警

3、查看通过流量还原的图片

参考代码：

# -*- coding: utf-8 -*-
# @Time    : 2022/6/13 6:56 PM
# @Author  : ailx10
# @File    : recapper.py

from scapy.all import rdpcap
from scapy.layers.inet import TCP
import collections
import os
import re
import sys
import zlib

OUTDIR = "/Users/ailx10/py3hack/chapter4/picture"
PCAPS = "/Users/ailx10/py3hack/chapter4/download"

Response = collections.namedtuple("Response",["header","payload"])

def get_header(payload):
    try:
        header_raw = payload[:payload.index(b"\r\n\r\n")+2]
    except ValueError:
        sys.stdout.write("-")
        sys.stdout.flush()
        return None
    header = dict(re.findall(r"(?P<name>.*?):(?P<value>.*?)\r\n",header_raw.decode()))
    if "Content-Type" not in header:
        return None
    return header

def extract_content(Response,content_name="image"):
    content,content_type = None,None
    if content_name in Response.header["Content-Type"]:
        content_type = Response.header["Content-Type"].split("/")[1]
        content = Response.payload[Response.payload.index(b"\r\n\r\n")+4:]

        if "Content-Encoding" in Response.header:
            if Response.header["Content-Encoding"] == "gzip":
                content = zlib.decompress(Response.header,zlib.MAX_WBITS | 32)
            elif Response.header["Content-Encoding"] == "deflate":
                content = zlib.decompress(Response.payload)
    return content,content_type

class Recapper:
    def __init__(self,fname):
        pcap = rdpcap(fname)
        self.sessions = pcap.sessions()
        self.response = list()

    def get_responses(self):
        for session in self.sessions:
            payload = b""
            for packet in self.sessions[session]:
                try:
                    if packet[TCP].dport == 80 or packet[TCP].sport == 80:
                        payload += bytes(packet[TCP].payload)
                except IndexError:
                    sys.stdout.write("x")
                    sys.stdout.flush()
                if payload:
                    header = get_header(payload)
                    if header is None:
                        continue
                    self.response.append(Response(header=header,payload=payload))

    def write(self,content_name):
        for i,response in enumerate(self.response):
            content,content_type = extract_content(response,content_name)
            if content and content_type:
                fname = os.path.join(OUTDIR,f"ex_{i}.{content_type}")
                print(f"Writing {fname}")
                with open(fname,"wb") as f:
                    f.write(content)

if __name__ == "__main__":
    pfile = os.path.join(PCAPS,"pcap.pcap")
    recapper = Recapper(pfile)
    recapper.get_responses()
    recapper.write("image")           

发布于 2022-06-13 19:27