Python Network Security Scanner DAY1

Об этом предложении

🐍 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_INET IPv4 사용 SOCK_STREAM TCP 사용 즉, 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초 결과를 정리하면 다음과 같다. 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 값도 테스트해볼 예정이다. 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 학습 내용 프로젝트 적용 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 관리 그리고 새롭게 다음 내용을 접했다. 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 을 함수 내부에 작성