LTE터널링을 위한 여정(3) Scapy

앞으로 네트워크 패킷을 분석하고 헤더를 떼고 붙이고 변조하고 등의 작업을 위하여 scapy 라는 파이썬 라이브러리 사용이 필수적이라는것을 알게 되었다.

그래서 오늘은 Scapy 에 대해 알아보는 시간을 가져보려 한다.

 

Scapy:

"Scapy는 파이썬(Python) 기반의 강력한 인터랙티브 패킷 조작 라이브러리이자 프로그램입니다. 네트워크 패킷을 직접 위조(Forge)하거나 디코딩하고, 네트워크 선로에 전송하며, 캡처 및 요청-응답 매칭 등 다양한 네트워크 분석 작업을 수행할 수 있습니다. (https://scapy.net/)"

 

라고 한다.

 

그럼 어떤 객체들이 있을까?

모든것을 다 소개할 수는 없지만 최대한 정리해보았다.

 

1.  패킷 생성

Ether()	Ethernet 헤더 생성	Ether(dst="ff:ff:ff:ff:ff:ff")
ARP()	ARP 패킷 생성	ARP(pdst="192.0.2.1")
IP()	IPv4 헤더 생성	IP(dst="192.0.2.1")
IPv6()	IPv6 헤더 생성	IPv6(dst="::1")
TCP()	TCP 헤더 생성	TCP(dport=80, flags="S")
UDP()	UDP 헤더 생성	UDP(dport=53)
ICMP()	ICMPv4 패킷 생성	ICMP(type=8)
ICMPv6EchoRequest()	ICMPv6 Echo Request	ICMPv6EchoRequest()
Raw()	임의의 바이트 데이터	Raw(load=b"hello")
DNS()	DNS 메시지 생성	DNS(rd=1)
DNSQR()	DNS 질문 레코드 생성	DNSQR(qname="example.com")
DNSRR()	DNS 리소스 레코드 생성	DNSRR(rrname="example.com", type="A", rdata="192.0.2.1")
Dot1Q()	VLAN 태그 생성	Dot1Q(vlan=10)
GRE()	GRE 캡슐화 헤더	GRE()
IPOption()	IPv4 옵션 객체	IPOption(b"\x01")

패킷을 네트워크로 실제 전송하지는 않고 메모리에서 구성하는 코드이다.

2. 패킷 구조 확인

pkt.show()	계층별 필드 상세 출력	pkt.show()
pkt.show2()	직렬화한 패킷을 다시 해석해 출력	pkt.show2()
pkt.summary()	패킷을 한 줄로 요약	pkt.summary()
pkt.sprintf()	원하는 형식으로 필드 출력	pkt.sprintf("%IP.src%")
ls()	사용 가능한 필드 목록 확인	ls(TCP)
hexdump()	HEX 및 ASCII 출력	hexdump(pkt)
chexdump()	C 배열 형태의 HEX 출력	chexdump(pkt)
linehexdump()	한 줄 HEX 출력	linehexdump(pkt)
raw()	패킷을 바이트로 변환	raw(pkt)
bytes()	패킷 바이트 표현 생성	bytes(pkt)
len()	패킷 바이트 길이 확인	len(pkt)
pkt.json()	패킷 정보를 JSON 문자열로 표현	pkt.json()
pkt.command()	재생성 가능한 Scapy 표현 생성	pkt.command()

show2()는 패킷을 직렬화한 뒤 다시 해석하므로 길이와 체크섬처럼 자동 계산되는 필드를 확인할 때 특히 유용하다.

3. 패킷 계층 접근 및 분리

pkt[IP]	IPv4 계층 객체 접근	pkt[IP].src
pkt[TCP]	TCP 계층 객체 접근	pkt[TCP].dport
pkt.getlayer()	특정 계층 객체 반환	pkt.getlayer(UDP)
pkt.haslayer()	특정 계층 존재 여부 확인	pkt.haslayer(TCP)
pkt.layers()	패킷에 포함된 계층 클래스 목록	pkt.layers()
pkt.getlayer(Raw)	Raw 계층 가져오기	pkt.getlayer(Raw)
pkt.copy()	패킷 복사	copy_pkt = pkt.copy()
pkt.remove_payload()	해당 계층 뒤의 페이로드 제거	pkt.remove_payload()
pkt.add_payload()	페이로드 연결	pkt.add_payload(Raw(b"test"))
pkt.lastlayer()	마지막 계층 반환	pkt.lastlayer()
pkt.getfieldval()	지정한 계층의 필드 값 읽기	pkt[IP].getfieldval("ttl")
pkt.setfieldval()	지정한 계층의 필드 값 설정	pkt[IP].setfieldval("ttl", 32)
pkt.decode_payload_as()	페이로드를 지정한 계층으로 해석	pkt.decode_payload_as(Raw)
pkt.getlayer()	원하는 계층이 없으면 None 반환	pkt.getlayer(DNS)

 

4. 패킷 전송 및 송수신

send()	L3 패킷 전송	send(IP(dst="127.0.0.1")/ICMP())
sendp()	L2 프레임 전송	sendp(Ether()/IP(dst="127.0.0.1"))
sr()	L3 전송 후 응답 수집	sr(IP(dst="127.0.0.1")/ICMP(), timeout=2)
sr1()	L3 전송 후 첫 응답 반환	sr1(IP(dst="127.0.0.1")/ICMP(), timeout=2)
srp()	L2 전송 및 응답 수집	srp(Ether()/ARP(pdst="192.0.2.1"), timeout=2)
srp1()	L2 전송 후 첫 응답 반환	srp1(Ether()/ARP(pdst="192.0.2.1"), timeout=2)
sendpfast()	외부 재생 도구를 활용한 빠른 L2 전송	sendpfast([Ether()/IP()])
srloop()	L3 패킷을 반복 송수신	srloop(IP(dst="127.0.0.1")/ICMP(), count=3)
srploop()	L2 패킷을 반복 송수신	srploop(Ether()/ARP(pdst="192.0.2.1"), count=3)

 

 

5. 패킷 캡처

sniff()	패킷 캡처	sniff(count=5)
AsyncSniffer()	백그라운드 캡처 객체	AsyncSniffer(iface="lo")
sniff(filter=...)	BPF 조건으로 패킷 캡처	sniff(filter="icmp", count=5)
sniff(prn=...)	캡처한 패킷에 함수 적용	sniff(prn=lambda p: p.summary(), count=5)
sniff(lfilter=...)	Python 조건으로 필터링	sniff(lfilter=lambda p: IP in p, count=5)
sniff(offline=...)	PCAP 파일에서 패킷 읽기	sniff(offline="test.pcap")
sniff(store=False)	패킷을 메모리에 누적하지 않음	sniff(count=10, store=False)
sniff(timeout=...)	지정 시간 동안 캡처	sniff(timeout=5)

 

6. 패킷 필드 수정

pkt[IP].src	IPv4 출발지 읽기·수정	pkt[IP].src = "192.0.2.10"
pkt[IP].dst	IPv4 목적지 읽기·수정	pkt[IP].dst = "192.0.2.20"
pkt[IP].ttl	TTL 수정	pkt[IP].ttl = 32
pkt[TCP].sport	TCP 출발지 포트 수정	pkt[TCP].sport = 12345
pkt[TCP].dport	TCP 목적지 포트 수정	pkt[TCP].dport = 8080
pkt[TCP].flags	TCP 플래그 수정	pkt[TCP].flags = "S"
pkt[TCP].seq	TCP Sequence Number 수정	pkt[TCP].seq = 1000
pkt[TCP].ack	TCP Acknowledgment Number 수정	pkt[TCP].ack = 1001
pkt[Raw].load	Raw 데이터 수정	pkt[Raw].load = b"NEW"
del pkt[IP].chksum	기존 IPv4 체크섬 필드 제거	del pkt[IP].chksum
del pkt[TCP].chksum	기존 TCP 체크섬 필드 제거	del pkt[TCP].chksum
checksum()	바이트 데이터의 인터넷 체크섬 계산	checksum(b"\x00\x01")
in4_chksum()	IPv4 의사 헤더를 고려한 L4 체크섬 계산	in4_chksum(17, pkt[IP], udp_bytes)
fragment()	IPv4 패킷 단편화	fragment(pkt, fragsize=1480)

 

7. PCAP 파일 입출력

rdpcap()	PCAP/PCAPNG 읽기	pkts = rdpcap("capture.pcap")
wrpcap()	PCAP 저장	wrpcap("output.pcap", pkts)
wrpcapng()	PCAPNG 저장	wrpcapng("output.pcapng", pkts)
PcapReader()	PCAP을 순차적으로 읽기	PcapReader("capture.pcap")
PcapWriter()	PCAP 스트리밍 기록	PcapWriter("output.pcap")
RawPcapReader()	원시 패킷 바이트와 메타데이터 읽기	RawPcapReader("capture.pcap")
RawPcapWriter()	원시 패킷을 PCAP 형식으로 기록	RawPcapWriter("output.pcap", linktype=1)

 

8. 패킷 목록 분석

packets.summary()	모든 패킷 간단 요약	packets.summary()
packets.nsummary()	번호와 함께 요약	packets.nsummary()
packets.show()	패킷 목록 출력	packets.show()
packets.filter()	조건에 따라 패킷 목록 필터링	packets.filter(lambda p: TCP in p)
packets.sessions()	패킷을 세션별로 분류	packets.sessions()
packets.make_table()	지정한 항목으로 표 생성	packets.make_table(lambda s,r: (s[IP].src, r[IP].dst, r[TCP].dport))
packets.plot()	패킷 값 시각화	packets.plot(lambda p: len(p))
packets.timeskew_graph()	패킷 시간 편차 관련 그래프	packets.timeskew_graph()

 

9. 네트워크 인터페이스와 라우팅

 

get_if_list()	인터페이스 이름 목록	get_if_list()
show_interfaces()	인터페이스 정보 출력	show_interfaces()
get_if_addr()	인터페이스 IPv4 주소 조회	get_if_addr("eth0")
get_if_hwaddr()	인터페이스 MAC 주소 조회	get_if_hwaddr("eth0")
getmacbyip()	IPv4 목적지에 도달할 다음 홉의 MAC 조회	getmacbyip("192.0.2.1")
get_if_addr6()	인터페이스 IPv6 주소 조회	get_if_addr6("eth0")
conf.iface	기본 사용 인터페이스 설정/조회	conf.iface
conf.route	Scapy의 IPv4 라우팅 테이블	conf.route
conf.route.route()	목적지 라우팅 결정 조회	conf.route.route("192.0.2.1")
conf.route6	Scapy IPv6 라우팅 테이블	conf.route6
conf.verb	Scapy 출력 상세도 설정	conf.verb = 0

 

10. 네트워크 진단 및 프로토콜 탐색

 

traceroute()	경로의 홉 탐색	traceroute(["127.0.0.1"], maxttl=3)
traceroute6()	IPv6 경로 탐색	traceroute6(["::1"], maxttl=3)
arping()	ARP 요청/응답 관찰	arping("192.0.2.0/30")
promiscping()	인터페이스의 프로미스큐어스 모드 관련 탐지	promiscping("192.0.2.1")
report_ports()	TCP 포트 응답 관련 보고	report_ports("127.0.0.1", (80, 443))
tshark()	패킷을 텍스트 방식으로 출력	tshark(count=5)

 

11. 패킷 테스트 및 Fuzzing

RandByte()	임의 1바이트 값 생성	RandByte()
RandShort()	임의 16비트 값 생성	RandShort()
RandInt()	임의 32비트 값 생성	RandInt()
RandIP()	임의 IPv4 주소 값	RandIP()
RandMAC()	임의 MAC 주소 값	RandMAC()
RandString()	임의 바이트 문자열	RandString(8)
fuzz()	기본 필드 값을 무작위화한 패킷 생성	fuzz(IP()/TCP())
corrupt_bytes()	지정 비율/개수의 바이트 변경	corrupt_bytes(b"hello", n=1)
corrupt_bits()	지정 비율/개수의 비트 변경	corrupt_bits(b"hello", n=1)

Fuzzing은 비정상적이거나 다양한 입력에 대해 소프트웨어가 어떻게 동작하는지 검증하는 방법이다.

Scapy가 생성한 비정상 패킷이 네트워크에서 실제로 그대로 송신·수신된다는 보장은 없기 때문에 유용할듯 하다.

12. 프로토콜 계층 확장 및 디코딩

bind_layers()	하위 프로토콜과 상위 프로토콜 연결	bind_layers(UDP, Raw, dport=9999)
split_layers()	등록된 계층 바인딩 해제	split_layers(UDP, Raw, dport=9999)
bind_bottom_up()	디코딩 방향의 바인딩 추가	bind_bottom_up(UDP, Raw, dport=9999)
bind_top_down()	패킷 생성 방향의 바인딩 추가	bind_top_down(UDP, Raw, dport=9999)
Packet	사용자 정의 프로토콜 클래스의 기본형	class Custom(Packet): ...
ByteField()	1바이트 필드 정의	ByteField("type", 1)
ShortField()	2바이트 정수 필드 정의	ShortField("length", 0)
IntField()	4바이트 정수 필드 정의	IntField("id", 0)
StrLenField()	길이가 정해지는 바이트 문자열 필드	StrLenField("data", b"", length_from=lambda p: p.length)
FieldLenField()	다른 필드의 길이나 개수 표현	FieldLenField("length", None, length_of="data", fmt="!H")

이 기능을 활용하면 지원되지 않는 자체 프로토콜의 헤더 형식을 정의하고 Scapy가 이를 분석하도록 확장할 수 있다.

더 궁금하다면? https://scapy.readthedocs.io/en/latest/api/scapy.utils.html

scapy.utils — Scapy 2.8.0 documentation

 

이런게 있다 싶으면서도 보기만해도 참 어렵다..

그래도 패킷변조를 위해서는 필수적이니 공부해보기로 한다.

다음에는 scapy 를 활용해 간단한 프로그램을 만들어보며 사용방법을 익힐 예정이다.