Python Network Security Scanner DAY1

정지범·2일 전

파이썬 프로젝트1

목록 보기
1/2

🐍 Python Security Scanner 프로젝트 - Day 1

점프 투 파이썬 학습을 마친 후 배운 내용을 실제로 활용하기 위해 시작한 개인 프로젝트이다.
Python의 socket을 이용해 TCP Port Scanner를 직접 구현하고, 순차 스캔의 속도 문제를 ThreadPoolExecutor를 이용한 멀티스레딩으로 개선해보았다.


📌 Day 1 목표

오늘의 목표는 다음과 같다.

  • Python socket 모듈 이해
  • 특정 IP / Port에 TCP 연결 시도
  • OPEN Port 판단
  • 여러 Port를 자동으로 스캔
  • 함수로 Port Scan 기능 분리
  • timeout 설정
  • ThreadPoolExecutor를 이용한 멀티스레딩
  • 스캔 실행 시간 측정
  • 직접 테스트 서버를 만들어 OPEN Port 탐지 검증

⚠️ 포트 스캔 실습은 본인의 PC 또는 직접 관리하는 테스트 환경을 대상으로 진행한다.


1. 프로젝트 생성

프로젝트 폴더를 생성하고 main.py 파일부터 시작했다.

python-security-scanner/
└── main.py

처음부터 복잡하게 파일을 나누기보다는 하나의 파일에서 기능을 구현하고, 프로젝트가 커지면 추후 모듈화할 예정이다.


2. IP와 Port 입력받기

가장 먼저 스캔할 대상의 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

3. TCP Socket 생성

실제로 네트워크 연결을 시도하기 위해 Python의 socket 모듈을 사용했다.

import socket

그리고 다음과 같이 Socket을 생성한다.

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

각 옵션의 의미는 다음과 같다.

코드의미
socket.socket()Socket 생성
AF_INETIPv4 사용
SOCK_STREAMTCP 사용

즉,

socket.socket(socket.AF_INET, socket.SOCK_STREAM)

은 쉽게 말하면

IPv4 주소를 사용하는 TCP Socket을 생성한다.

라는 의미이다.


4. connect_ex()를 이용한 TCP 연결

생성한 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 여부를 중심으로 판단하도록 구현했다.


5. Socket 종료

생성한 Socket은 사용이 끝나면 종료해준다.

sock.close()

기본적인 흐름은 다음과 같다.

Socket 생성
    ↓
TCP 연결 시도
    ↓
결과 확인
    ↓
Socket 종료

6. 여러 Port 자동 스캔

처음에는 하나의 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만 출력하도록 했다.


7. Timeout 설정

여러 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 값을 너무 크게 설정하면 스캔 속도가 느려질 수 있고, 반대로 너무 작게 설정하면 응답이 느린 서비스의 연결 성공 여부를 놓칠 수 있다.


8. 직접 8000번 Port 열어보기

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 연결 상태를 기준으로 동작한다는 것을 확인했다.


9. Port Scan 기능 함수화

기존에는 모든 코드를 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 : 검사할 대상 IP
  • target_port : 검사할 Port

예를 들어

scan_port("127.0.0.1", 8000)

을 실행하면

target_ip   = "127.0.0.1"
target_port = 8000

이 함수 내부로 전달된다.

여기서는 return을 사용하지 않았다.

현재 함수의 목적은 결과값을 다른 곳에 반환하는 것이 아니라 OPEN Port가 발견되었을 때 직접 print()로 출력하는 것이기 때문이다.


10. 함수 정의와 함수 호출

함수를 사용하면서 다시 정리한 중요한 개념이다.

다음 코드는 함수를 정의하는 것이다.

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문에서 필요한 만큼 호출한다.


11. 순차 스캔의 문제

함수화를 했지만 여전히 Port는 순서대로 검사한다.

Port 1 검사
    ↓
완료
    ↓
Port 2 검사
    ↓
완료
    ↓
Port 3 검사

Timeout이 0.5초라고 가정하면 응답이 늦은 Port가 많을 경우 시간이 상당히 오래 걸릴 수 있다.

예를 들어 1000개의 Port가 각각 최대 0.5초를 기다린다면 이론적으로 상당한 시간이 필요할 수 있다.

따라서 여러 Port를 동시에 검사하도록 개선하기로 했다.


12. ThreadPoolExecutor

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 작업에서는 여러 연결을 동시에 진행할 수 있기 때문에 성능 개선 효과를 얻을 수 있다.


13. executor.submit()

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)

작업을 실행하도록 요청하는 것이다.


14. 멀티스레딩 Port Scanner

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 작업을 나누어 처리한다.


15. 스캔 시간 측정

멀티스레딩 적용 전후의 성능을 확인하기 위해 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 작업의 시작과 끝을 한 번씩 측정해야 한다.


16. 멀티스레딩 성능 비교

127.0.0.1의 1~1000번 Port를 대상으로 테스트했다.

Timeout:

sock.settimeout(0.5)

10 Workers

ThreadPoolExecutor(max_workers=10)

실행 결과:

스캔 소요 시간: 약 51.03초

50 Workers

ThreadPoolExecutor(max_workers=50)

실행 결과:

스캔 소요 시간: 약 10.22초

결과를 정리하면 다음과 같다.

WorkerPort 범위Timeout소요 시간
101~10000.5초약 51.03초
501~10000.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 값도 테스트해볼 예정이다.


17. 실제로 탐지된 OPEN Port

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에서 실제로 어떤 서비스가 동작하고 있는지는 확인하지 않는다.

서비스 식별은 이후 기능으로 추가할 예정이다.


18. Day 1 최종 코드

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}초")

📊 Day 1 프로그램 동작 흐름

사용자에게 Target IP 입력
            ↓
      Port 1 ~ 1000
            ↓
   ThreadPoolExecutor
       (50 Workers)
            ↓
      scan_port()
            ↓
     TCP Socket 생성
            ↓
      Timeout 설정
            ↓
      connect_ex()
            ↓
       TCP 연결 시도
            ↓
      ┌─────┴─────┐
      ↓           ↓
 result == 0   result != 0
      ↓           ↓
    OPEN       출력하지 않음
      ↓
  Socket 종료
            ↓
    모든 작업 완료
            ↓
     실행 시간 출력

💡 Day 1에서 배운 내용

이번 프로젝트를 진행하면서 기존에 공부했던 Python 문법이 실제 프로그램에서 어떻게 사용되는지 확인할 수 있었다.

Python 학습 내용프로젝트 적용
importsocket, time 모듈 사용
변수IP, Port, 실행 시간 저장
input()스캔 대상 IP 입력
ifTCP 연결 성공 여부 판단
for여러 Port 반복
range()Port Scan 범위 지정
함수 defPort Scan 기능 분리
매개변수IP와 Port를 함수에 전달
f-stringScan 결과 출력
withThread Pool 관리

그리고 새롭게 다음 내용을 접했다.

  • TCP Socket
  • AF_INET
  • SOCK_STREAM
  • connect_ex()
  • Timeout
  • Thread
  • Thread Pool
  • ThreadPoolExecutor
  • executor.submit()
  • 프로그램 실행 시간 측정

🔍 오늘 발생했던 문제와 해결

1. IP에 int()를 사용함

처음에는 다음과 같이 작성했다.

target_ip = int(input(...))

하지만 IP는 127.0.0.1과 같은 문자열이므로 int()로 변환할 수 없다.

target_ip = input(...)

로 수정했다.

2. sock.close()를 for문 밖에 작성

Port마다 새로운 Socket을 생성하면서 close()를 반복문 밖에 작성하면 마지막 Socket만 종료하게 된다.

따라서 각 Port의 검사가 끝날 때마다 Socket을 종료하도록 반복문 내부에 배치했다.

3. 함수 내부에 for문을 작성

처음 함수화할 때

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)

4. start_time을 함수 내부에 작성

함수 내부에서 시작 시간을 기록하면 Port마다 새로운 시작 시간이 만들어진다.

내가 측정하고 싶은 것은 전체 Port Scan 시간이므로 스캔 시작 직전에 한 번만 기록하도록 수정했다.

5. end_time을 for문 내부에 작성

executor.submit()은 작업을 Thread Pool에 제출하는 것이므로 작업 제출 직후가 전체 Scan 종료 시점은 아니다.

따라서 with ThreadPoolExecutor(...) 블록이 종료된 이후 end_time을 기록했다.


🚀 다음 목표 - Day 2

현재 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를 발전시켜볼 예정이다.


📝 Day 1 한줄 정리

Python socket으로 TCP Port Scanner를 구현하고, 순차 스캔에서 발생한 속도 문제를 ThreadPoolExecutor 기반 멀티스레딩으로 개선한 뒤 실제 실행 시간을 측정해 성능 차이를 확인했다.

profile
성실하자:)

0개의 댓글