
점프 투 파이썬학습을 마친 후 배운 내용을 실제로 활용하기 위해 시작한 개인 프로젝트이다.
Python의socket을 이용해 TCP Port Scanner를 직접 구현하고, 순차 스캔의 속도 문제를ThreadPoolExecutor를 이용한 멀티스레딩으로 개선해보았다.
오늘의 목표는 다음과 같다.
socket 모듈 이해timeout 설정ThreadPoolExecutor를 이용한 멀티스레딩⚠️ 포트 스캔 실습은 본인의 PC 또는 직접 관리하는 테스트 환경을 대상으로 진행한다.
프로젝트 폴더를 생성하고 main.py 파일부터 시작했다.
python-security-scanner/
└── main.py
처음부터 복잡하게 파일을 나누기보다는 하나의 파일에서 기능을 구현하고, 프로젝트가 커지면 추후 모듈화할 예정이다.
가장 먼저 스캔할 대상의 IP와 Port를 사용자에게 입력받았다.
target_ip = input("스캔할 IP를 입력하세요: ")
target_port = int(input("스캔할 Port를 입력하세요: "))
여기서 IP와 Port의 자료형이 다르다는 점이 중요하다.
IP
"127.0.0.1"
→ str
Port
80
→ int
IP 주소는 127.0.0.1처럼 .이 포함되어 있기 때문에 정수로 변환할 수 없다.
따라서 IP는 문자열 그대로 입력받고, Port만 int()를 이용해 정수로 변환한다.
출력에는 f-string을 사용했다.
print(f"[+] {target_ip}:{target_port}")
실행 예시:
스캔할 IP를 입력하세요: 127.0.0.1
스캔할 Port를 입력하세요: 80
[+] 127.0.0.1:80
실제로 네트워크 연결을 시도하기 위해 Python의 socket 모듈을 사용했다.
import socket
그리고 다음과 같이 Socket을 생성한다.
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
각 옵션의 의미는 다음과 같다.
| 코드 | 의미 |
|---|---|
socket.socket() | Socket 생성 |
AF_INET | IPv4 사용 |
SOCK_STREAM | TCP 사용 |
즉,
socket.socket(socket.AF_INET, socket.SOCK_STREAM)
은 쉽게 말하면
IPv4 주소를 사용하는 TCP Socket을 생성한다.
라는 의미이다.
생성한 Socket을 이용해 대상 IP와 Port에 TCP 연결을 시도한다.
result = sock.connect_ex((target_ip, target_port))
connect_ex()에는 IP와 Port를 하나의 튜플로 전달한다.
("127.0.0.1", 80)
구조는 다음과 같다.
sock.connect_ex(
(target_ip, target_port)
)
connect_ex()는 연결 결과를 숫자로 반환한다.
TCP 연결 성공
↓
result == 0
TCP 연결 실패
↓
result != 0
따라서 if문을 이용해 Port의 상태를 판단할 수 있다.
if result == 0:
print(f"[+] {target_ip}:{target_port} OPEN")
else:
print(f"[-] {target_ip}:{target_port} CLOSED")
단, 정확히 말하면 result != 0은 무조건 하나의 의미인 CLOSED만 나타내는 것은 아니다.
Connection Refused, Timeout 등 여러 이유로 TCP 연결이 성공하지 않을 수 있기 때문에 이후에는 OPEN 여부를 중심으로 판단하도록 구현했다.
생성한 Socket은 사용이 끝나면 종료해준다.
sock.close()
기본적인 흐름은 다음과 같다.
Socket 생성
↓
TCP 연결 시도
↓
결과 확인
↓
Socket 종료
처음에는 하나의 Port만 입력받았다.
target_port = int(input("스캔할 Port를 입력하세요: "))
하지만 Port Scanner라면 여러 Port를 자동으로 검사할 수 있어야 한다.
따라서 for문과 range()를 이용했다.
for target_port in range(1, 101):
range()의 마지막 숫자는 포함되지 않기 때문에
range(1, 101)
은 실제로
1
2
3
...
99
100
까지 반복한다.
각 Port마다 새로운 TCP 연결을 시도해야 하기 때문에 Socket 역시 반복문 내부에서 생성하고 종료했다.
for target_port in range(1, 101):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
result = sock.connect_ex((target_ip, target_port))
if result == 0:
print(f"[+] {target_ip}:{target_port} OPEN")
sock.close()
CLOSED Port까지 모두 출력하면 결과가 너무 많아지기 때문에 이후에는 OPEN Port만 출력하도록 했다.
여러 Port를 검사해보니 스캔 속도가 매우 느린 문제가 발생했다.
TCP 연결을 시도했는데 상대방이 바로 응답하지 않으면 프로그램이 응답을 기다릴 수 있기 때문이다.
이를 방지하기 위해 Timeout을 설정했다.
sock.settimeout(0.5)
의미는 다음과 같다.
TCP 연결 시도
↓
최대 0.5초 동안 기다림
↓
응답이 없으면 연결 시도를 계속 기다리지 않음
따라서 Socket 생성 이후 다음과 같이 Timeout을 설정했다.
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(0.5)
result = sock.connect_ex((target_ip, target_port))
Timeout 값을 너무 크게 설정하면 스캔 속도가 느려질 수 있고, 반대로 너무 작게 설정하면 응답이 느린 서비스의 연결 성공 여부를 놓칠 수 있다.
Scanner가 실제로 OPEN Port를 탐지하는지 검증하기 위해 Python의 간단한 HTTP Server를 사용했다.
새로운 터미널에서 다음 명령어를 실행했다.
python -m http.server 8000
실행하면 Python HTTP Server가 TCP 8000번 Port에서 연결을 기다린다.
Python HTTP Server
↓
TCP 8000
↓
OPEN
브라우저에서도 다음 주소로 접근할 수 있다.
http://127.0.0.1:8000
Scanner의 범위도 테스트를 위해 다음과 같이 변경했다.
for target_port in range(7995, 8006):
그러면 다음 Port들을 검사한다.
7995
7996
7997
7998
7999
8000
8001
8002
8003
8004
8005
실행 결과:
[+] 127.0.0.1:8000 OPEN
정상적으로 8000번 Port가 탐지되었다.
HTTP Server를 Ctrl + C로 종료한 뒤 다시 Scanner를 실행하면 8000번 Port가 더 이상 OPEN으로 탐지되지 않는다.
이를 통해 직접 만든 Scanner가 실제 TCP 연결 상태를 기준으로 동작한다는 것을 확인했다.
기존에는 모든 코드를 for문 안에서 처리했다.
이후 멀티스레딩을 적용하기 위해 Port 하나를 검사하는 기능을 함수로 분리했다.
def scan_port(target_ip, target_port):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(0.5)
result = sock.connect_ex((target_ip, target_port))
if result == 0:
print(f"[+] {target_ip}:{target_port} OPEN")
sock.close()
함수에 필요한 매개변수는 두 개이다.
def scan_port(target_ip, target_port):
target_ip : 검사할 대상 IPtarget_port : 검사할 Port예를 들어
scan_port("127.0.0.1", 8000)
을 실행하면
target_ip = "127.0.0.1"
target_port = 8000
이 함수 내부로 전달된다.
여기서는 return을 사용하지 않았다.
현재 함수의 목적은 결과값을 다른 곳에 반환하는 것이 아니라 OPEN Port가 발견되었을 때 직접 print()로 출력하는 것이기 때문이다.
함수를 사용하면서 다시 정리한 중요한 개념이다.
다음 코드는 함수를 정의하는 것이다.
def scan_port(target_ip, target_port):
함수를 정의했다고 바로 실행되는 것은 아니다.
실제로 실행하려면 함수를 호출해야 한다.
scan_port(target_ip, target_port)
따라서 반복문과 함수를 다음과 같이 분리할 수 있다.
def scan_port(target_ip, target_port):
# Port 하나 검사
...
for target_port in range(1, 101):
scan_port(target_ip, target_port)
구조는 다음과 같다.
Port 1
↓
scan_port(IP, 1)
Port 2
↓
scan_port(IP, 2)
Port 3
↓
scan_port(IP, 3)
...
즉,
함수는 한 번 정의하고, for문에서 필요한 만큼 호출한다.
함수화를 했지만 여전히 Port는 순서대로 검사한다.
Port 1 검사
↓
완료
↓
Port 2 검사
↓
완료
↓
Port 3 검사
Timeout이 0.5초라고 가정하면 응답이 늦은 Port가 많을 경우 시간이 상당히 오래 걸릴 수 있다.
예를 들어 1000개의 Port가 각각 최대 0.5초를 기다린다면 이론적으로 상당한 시간이 필요할 수 있다.
따라서 여러 Port를 동시에 검사하도록 개선하기로 했다.
Python의 concurrent.futures 모듈에서 ThreadPoolExecutor를 사용했다.
from concurrent.futures import ThreadPoolExecutor
ThreadPoolExecutor는 여러 Worker Thread가 작업을 나눠 처리할 수 있도록 해준다.
예를 들어
ThreadPoolExecutor(max_workers=10)
은 최대 10개의 Worker Thread를 이용해 작업을 처리한다.
기존 방식:
Worker 1 → Port 1 → 완료
↓
Port 2 → 완료
↓
Port 3 → 완료
멀티스레딩:
Worker 1 → Port 1
Worker 2 → Port 2
Worker 3 → Port 3
Worker 4 → Port 4
...
Worker 10 → Port 10
네트워크 연결처럼 응답을 기다리는 시간이 많은 I/O 작업에서는 여러 연결을 동시에 진행할 수 있기 때문에 성능 개선 효과를 얻을 수 있다.
Thread Pool에 작업을 전달하기 위해 submit()을 사용했다.
executor.submit(scan_port, target_ip, target_port)
구조를 풀어보면 다음과 같다.
executor.submit(
scan_port, ← 실행할 함수
target_ip, ← 첫 번째 인수
target_port ← 두 번째 인수
)
즉 Worker에게
scan_port(target_ip, target_port)
작업을 실행하도록 요청하는 것이다.
for문과 ThreadPoolExecutor를 결합했다.
with ThreadPoolExecutor(max_workers=10) as executor:
for target_port in range(1, 1001):
executor.submit(scan_port, target_ip, target_port)
with 블록을 사용하면 Thread Pool을 사용한 뒤 필요한 정리를 처리할 수 있고, 블록을 빠져나올 때 제출된 작업들이 완료될 때까지 기다린다.
이제 하나의 Worker가 Port를 순서대로 검사하는 대신 여러 Worker가 Port Scan 작업을 나누어 처리한다.
멀티스레딩 적용 전후의 성능을 확인하기 위해 time 모듈을 사용했다.
import time
스캔 직전에 시작 시간을 기록한다.
start_time = time.time()
모든 작업이 완료된 이후 종료 시간을 기록한다.
end_time = time.time()
그리고 두 값의 차이를 계산한다.
print(f"스캔 소요 시간: {end_time - start_time}초")
중요한 점은 start_time과 end_time의 위치이다.
start_time
↓
전체 Port Scan
↓
모든 Thread 작업 완료
↓
end_time
각 Port를 검사할 때마다 시간을 측정하는 것이 아니라 전체 Port Scan 작업의 시작과 끝을 한 번씩 측정해야 한다.
127.0.0.1의 1~1000번 Port를 대상으로 테스트했다.
Timeout:
sock.settimeout(0.5)
ThreadPoolExecutor(max_workers=10)
실행 결과:
스캔 소요 시간: 약 51.03초
ThreadPoolExecutor(max_workers=50)
실행 결과:
스캔 소요 시간: 약 10.22초
결과를 정리하면 다음과 같다.
| Worker | Port 범위 | Timeout | 소요 시간 |
|---|---|---|---|
| 10 | 1~1000 | 0.5초 | 약 51.03초 |
| 50 | 1~1000 | 0.5초 | 약 10.22초 |
50 Workers를 사용했을 때 10 Workers보다 약 5배 빠른 결과를 확인할 수 있었다.
대략적으로 생각하면 10 Workers의 경우:
1000개 Port ÷ 10 Workers
= Worker당 약 100개
100개 × 최대 0.5초
≈ 50초
실제 결과는 약 51초였다.
50 Workers의 경우:
1000개 Port ÷ 50 Workers
= Worker당 약 20개
20개 × 최대 0.5초
≈ 10초
실제 결과 역시 약 10.22초였다.
따라서 단순히 "멀티스레딩이 빠르다"가 아니라 실제 실행 시간을 측정하여 성능 차이를 확인할 수 있었다.
단, Worker 수를 무조건 크게 설정한다고 항상 좋은 것은 아니므로 이후 적절한 Worker 수와 Timeout 값도 테스트해볼 예정이다.
1~1000번 Port를 스캔하면서 내 PC에서 다음과 같은 OPEN Port들이 탐지되었다.
[+] 127.0.0.1:135 OPEN
[+] 127.0.0.1:445 OPEN
[+] 127.0.0.1:902 OPEN
[+] 127.0.0.1:912 OPEN
[+] 127.0.0.1:945 OPEN
현재 단계에서는 TCP 연결 성공 여부만 판단하고 있기 때문에 각 Port에서 실제로 어떤 서비스가 동작하고 있는지는 확인하지 않는다.
서비스 식별은 이후 기능으로 추가할 예정이다.
import time
import socket
from concurrent.futures import ThreadPoolExecutor
# 스캔할 대상 IP 입력
target_ip = input("스캔할 IP를 입력하세요: ")
# 하나의 Port를 검사하는 함수
def scan_port(target_ip, target_port):
# IPv4(AF_INET) + TCP(SOCK_STREAM) 소켓 생성
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# TCP 연결 시 최대 0.5초까지만 응답을 기다림
sock.settimeout(0.5)
# 대상 IP와 Port에 TCP 연결을 시도하고 결과 코드 반환
result = sock.connect_ex((target_ip, target_port))
# connect_ex()의 반환값이 0이면 TCP 연결 성공
if result == 0:
print(f"[+] {target_ip}:{target_port} OPEN")
# 현재 Port 검사가 끝났으므로 Socket 종료
sock.close()
# 전체 Port Scan 시작 시간 기록
start_time = time.time()
# 최대 50개의 작업자 Thread를 사용하는 Thread Pool 생성
with ThreadPoolExecutor(max_workers=50) as executor:
# 1번부터 1000번 Port까지 반복
for target_port in range(1, 1001):
# scan_port() 작업을 Thread Pool에 제출
executor.submit(scan_port, target_ip, target_port)
# 모든 Port Scan 작업 완료 후 종료 시간 기록
end_time = time.time()
# 전체 Port Scan 소요 시간 출력
print(f"스캔 소요 시간: {end_time - start_time}초")
사용자에게 Target IP 입력
↓
Port 1 ~ 1000
↓
ThreadPoolExecutor
(50 Workers)
↓
scan_port()
↓
TCP Socket 생성
↓
Timeout 설정
↓
connect_ex()
↓
TCP 연결 시도
↓
┌─────┴─────┐
↓ ↓
result == 0 result != 0
↓ ↓
OPEN 출력하지 않음
↓
Socket 종료
↓
모든 작업 완료
↓
실행 시간 출력
이번 프로젝트를 진행하면서 기존에 공부했던 Python 문법이 실제 프로그램에서 어떻게 사용되는지 확인할 수 있었다.
| Python 학습 내용 | 프로젝트 적용 |
|---|---|
import | socket, time 모듈 사용 |
| 변수 | IP, Port, 실행 시간 저장 |
input() | 스캔 대상 IP 입력 |
if | TCP 연결 성공 여부 판단 |
for | 여러 Port 반복 |
range() | Port Scan 범위 지정 |
함수 def | Port Scan 기능 분리 |
| 매개변수 | IP와 Port를 함수에 전달 |
| f-string | Scan 결과 출력 |
with | Thread Pool 관리 |
그리고 새롭게 다음 내용을 접했다.
AF_INETSOCK_STREAMconnect_ex()ThreadPoolExecutorexecutor.submit()int()를 사용함처음에는 다음과 같이 작성했다.
target_ip = int(input(...))
하지만 IP는 127.0.0.1과 같은 문자열이므로 int()로 변환할 수 없다.
target_ip = input(...)
로 수정했다.
sock.close()를 for문 밖에 작성Port마다 새로운 Socket을 생성하면서 close()를 반복문 밖에 작성하면 마지막 Socket만 종료하게 된다.
따라서 각 Port의 검사가 끝날 때마다 Socket을 종료하도록 반복문 내부에 배치했다.
처음 함수화할 때
def scan_port(target_ip, target_port):
for target_port in range(...):
형태로 작성했다.
하지만 scan_port()의 역할은 Port 하나를 검사하는 것으로 정하고 반복은 함수 외부에서 담당하도록 분리했다.
for target_port in range(...):
scan_port(target_ip, target_port)
start_time을 함수 내부에 작성함수 내부에서 시작 시간을 기록하면 Port마다 새로운 시작 시간이 만들어진다.
내가 측정하고 싶은 것은 전체 Port Scan 시간이므로 스캔 시작 직전에 한 번만 기록하도록 수정했다.
end_time을 for문 내부에 작성executor.submit()은 작업을 Thread Pool에 제출하는 것이므로 작업 제출 직후가 전체 Scan 종료 시점은 아니다.
따라서 with ThreadPoolExecutor(...) 블록이 종료된 이후 end_time을 기록했다.
현재 Scanner는
IP 입력
↓
1~1000 Port Scan
↓
OPEN Port 탐지
까지 구현했다.
Day 2에서는 여기서 기능을 조금 더 발전시켜볼 예정이다.
OPEN Port
↓
어떤 서비스인가?
↓
22 → SSH
80 → HTTP
443 → HTTPS
...
단순히
[+] 127.0.0.1:445 OPEN
이라고 출력하는 것을 넘어
[+] 127.0.0.1:445 OPEN
Service : ?
처럼 열린 Port의 서비스 정보를 확인하는 기능을 만들어보는 것이 다음 목표다.
이후에는 결과 저장, 예외 처리, 입력값 검증 등을 하나씩 추가하면서 Scanner를 발전시켜볼 예정이다.
Python
socket으로 TCP Port Scanner를 구현하고, 순차 스캔에서 발생한 속도 문제를ThreadPoolExecutor기반 멀티스레딩으로 개선한 뒤 실제 실행 시간을 측정해 성능 차이를 확인했다.