- 전체
- 보안뉴스
- 제로데이취약점
- 해킹프로그래밍
- 웹해킹
- 해킹기법
- 정보보호
- 정보보안기사 - 국가기술자격
- 악성코드분석_리버싱
- 시큐어코딩_개발보안진단원
- CISSP
- CISA
- 모의해킹_penetration-test
- deepweb / tor network
- Kali Linux
모의해킹_penetration-test 모의해킹(Penetration Testing) PortScanner 6. 업그레이드1: 스레드를 활용한 고속 스캔
2023.03.20 08:59
모의해킹(Penetration Testing) PortScanner 6. 업그레이드1: 스레드를 활용한 고속 스캔
6. 업그레이드1: 스레드를 활용한 고속 스캔
성공적으로 포트 스캐너를 만들었지만, 이 프로그램의 실행의 흐름은 하나이기 때문에 매우 느리다. 한 포트의 스캔이 끝나야만 다음 포트를 스캔할 수 있기에 대기 시간이 상당히 길다는 단점이 존재한다. 만약 이 속도로 65536(0~65535)개의 포트를 모두 검사해야 된다면 여러분은 시간 내의 업무를 마치지 못할 수도 있다. 그렇다면 부득이하게 야근을 해야 하는 경우도 발생할 것이다.
이번에는 스레드를 활용하여 프로그램의 실행을 여러 개로 나누어 보자. 대상 서버로 요청을 한꺼번에 여러 개 보낸 뒤 여러 응답을 한꺼번에 기다리면 프로그램의 효율이 증가할 것이다.
스레드란 한 프로세스 내에서의 실행 흐름 단위를 이야기한다. 우리는 멀티 스레드 기법을 사용하여 한 프로세스 내에 여러 스레드가 존재하도록 만들 것이다.
그림 일반 포트 스캐너와 스레드 포트 스캐너의 차이
스레드를 활용하기 위해서는 스레드를 제어할 줄 알아야 한다. CPU가 스레드를 돌아가며 실행하기 때문에 결과를 연속해서 출력하길 원하거나, 스레드의 수를 제어하기 원한다면(임계구역), 또 자원 원자성을 보호하기 원한다면 세마포어를 사용해야 한다. 세마포어는 특정 작업을 하는 동안에 락(Lock)을 걸어 임계구역을 설정하여 다른 스레드가 작업을 하지 않도록 제어한다. 언락(Unlock)을 실행하면 다시 원래대로 동작한다.
많은 사람들이 세마포어를 보통 화장실에 비교하여 많이 설명한다. 이 설명이 매우 단순하면서도 실제로 경험해보는 원리이기 때문에 잘 이해할 수 있다.
변기 3대가 있는 화장실이 있다고 가정하자. 사람 10명이 와서 이 화장실을 사용하려고 할 때 10명이 모두 동시에 이 화장실을 사용할 수 없다. 한번에 3명만이 가능하다. 3명은 화장실에 들어가 문을 걸어 잠그고(Lock), 다 사용한 뒤 문을 열고 나올 것이다(Unlock). 그 후 다음 사람들이 차례대로 이 화장실을 사용할 수 있을 것이다.
그림 10명의 사람들과 3개의 화장실
이러한 방식으로 스레드 생성량을 조절하지 않으면 프로그램은 스레드를 더 이상 만들 수 없다는 메시지를 출력하고 죽어버리니 꼭 주의하자. 먼저 실습을 통해 스레드의 동작 원리를 파악해 하자. time 라이브러리의 sleep함수를 사용하여 위에서 설명한 화장실 시뮬레이션을 직접 구현해보도록 한다. 아래 코드를 에디터 모드로 작성하여 실행([F5])하자.
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
#! python
# -*- coding: utf-8 -*-
# toilet simulation
import threading # 스레드 라이브러리
import time # 시간 라이브러리
import random # 난수 라이브러리
numOfToilet = threading.Semaphore(value=3) # 화장실이 3개 존재한다.
# 스레드 함수
def useToilet(peopleNum): # 스레드를 돌릴 함수를 정의해야 한다.
time.sleep(random.randint(5,10)) # 화장실 사용 시간(5~10초 랜덤생성)
# 해당 시간 동안 슬립
print("he's happy now, people #" + peopleNum)
numOfToilet.release() # 도어락 해제
# 메인 함수
def main():
print("10 people start using 3 toilets...")
for peopleNum in range(10): # 10명의 사람들이 대기 중
numOfToilet.acquire() # 도어락 설정
t=threading.Thread(target=useToilet, args=(str(peopleNum))) # 스레드 생성
t.start() # 스레드 시작
time.sleep(11)
print("All People Happy~~~!!!") # 모든 사람들 사용 종료
if __name__ =='__main__':
main()
|
cs |
결과 화면
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
10 people start using 3 toilets...
he's happy now, people #2
he's happy now, people #0
he's happy now, people #1
he's happy now, people #3
he's happy now, people #4
he's happy now, people #5
he's happy now, people #7
he's happy now, people #6
he's happy now, people #8
he's happy now, people #9
All People Happy~~~!!!
|
|
사람들의 사용 시간이 다르기 때문에 순서가 뒤죽박죽 진짜 화장실을 사용한 것처럼 결과가 실시간으로 출력된다. 3명만이 동시에 사용할 수 있기 때문에 중간에 사람이 나오면 다음 사람이 들어가 화장실을 사용할 수 있게 된다. 아래 그림을 통해서 위 결과를 조금 더 자세하게 이해해보기 바란다.
실행 시간을 정확히 측정하지 못해 정확하지는 않지만 대략 이러한 흐름으로 흘러갔을 것이라 예상할 수 있다. #2가 끝나면 #3이 시작하고, #0이 끝나면 #4가 시작하고, #1이 끝나면 #5가 시작한다. #9가 실행될 때까지 이 반복 작업을 수행한다.
새롭게 등장한 코드들에 대해서 살펴보자.
numOfToilet = threading.Semaphore(value=3)
threading.Semaphore함수의 인자 값을 3으로 세팅하여 한 번에 실행될 수 있는 스레드의 개수를 3개로 제한 했다.
time.sleep(random.randint(5,10))
time.sleep(s)는 s초 만큼 스레드가 슬립(대기) 상태로 머물게 한다. s에 int 값 대신에 random.randint(5,10)을 넣었다. 이 함수는 5~10초를 랜덤으로 생성하여 int 값으로 반환한다. 즉 5~10의 int 값을 생성하여 time.sleep(s)의 인자로 넘겨 5~10초 동안 스레드가 쉬도록 하는 코드이다.
print "\nhe's happy now, people #" + str(peopleNum),
맨 마지막 문자인 ‘,’(쉼표)는 오타가 아니다. 쉼표를 입력하면 문자열 마지막에 newline 대신에 공백을 삽입한다. 위의 결과는 거의 동일하다. 굳이 이렇게 다른 방식으로 출력하는 이유는 스레드를 여러 개 실행하면 print 함수가 한 번에 실행되지 않고 문자열 출력과 새로운 줄 출력을 나눠서 실행하는 경우가 있기 때문이다. 그러면 결과 출력 시 결과가 지저분하게 섞여서 출력된다. 위 코드처럼 쉼표와 ‘\n’를 삽입하면 모든 문자열을 한번에 출력하여 문자열과 newLine이 뒤섞이지 않고 잘 출력된다.
numOfToilet.acquire()
numOfToilet.release()
이 코드들은 미리 정의한 numOfToilet의 스레드 개수를 추가하고 줄인다. acquire() 함수를 실행하여 실행 중인 스레드가 3개가 되면 더 이상 진행하지 않고 대기하다가, 실행 중이던 스레드가 release() 함수를 실행하여 스레드 개수가 줄어들면 다시 코드를 진행시킨다.
t=threading.Thread(target=useToilet, args=(str(peopleNum)))
t.start()
Thread() 함수는 변수 t를 만들어 스레드 객체를 생성한다. target에는 위에서 미리 정의한 함수인 useToilet 값을 넣었고, args에는 함수에 넘겨줄 파라미터를 넣었다. t.start()는 스레드를 실행한다.
지금까지 세마포어에 대하여 공부해보았다. 이제 이 세마포어를 활용하여 포트 스캐너에 접목시켜보자. 여기서는 두 가지의 세마포어를 생성하였다. 세마포어 화장실 예제에서 사용한 세마포어처럼 스레드 개수를 제한하는 세마포어와 출력이 뒤섞이지 않도록 제어하는 세마포어를 사용하겠다. 마지막에는 결과를 csv 파일로 저장하고, 콘솔에도 출력하도록 하겠다.
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
|
#! python
# -*- coding: utf-8 -*-
# PortScanProgram with threads
import threading # 스레드 관련 라이브러리
import socket # 소켓 관련 라이브러리
import time # 시간 라이브러리
resultLock = threading.Semaphore(value=1) # 결과 출력을 제어할 세마포어
maxConnection = 20 # 스레드 개수를 제어할 세마포어
connection_lock = threading.BoundedSemaphore(value=maxConnection)
portList = [] # 결과를 저장할 리스트1
dataList = [] # 결과를 저장할 리스트2
# 스레드 함수
def scanPort(tgtHost, portNum):
try: # 접속 시도
s=socket.socket()
s.settimeout(2)
s.connect((tgtHost, portNum))
s.send("Python Connect\n")
data = s.recv(1024)
resultLock.acquire() # 결과 저장을 위해 세마포어 설정
if data: # 데이터가 존재할 때 출력 및 저장
print("[+] Port " + str(portNum) + " open: " + data[:20])
portList.append(portNum)
dataList.append('exist data')
else: # 데이터가 존재하지 않을 때 출력 및 저장
print("[+] Port " + str(portNum) + " open: no data")
portList.append(portNum)
dataList.append('no data')
resultLock.release() # 세마포어 해제
except:
pass
finally:
s.close()
connection_lock.release() # 세마포어 해제
# 메인 함수
def main()
tgtHost = "192.168.157.133" # 스캔 대상 tgtHost
for portNum in range(65536): # 반복문 수행 0~65535 포트
connection_lock.acquire()
t = threading.Thread(target=scanPort, args =(tgtHost, portNum)) # 쓰레드 초기화
t.start() # 스레드 실행
if portNum == 65535: # 마지막 스레드 실행 후 결과를 기다림
time.sleep(10)
# csv 파일 저장
f=open("portScanResult.csv",'w') # 결과를 저장할 csv 파일 열기
f.write("portNum, banner\n") # 컬럼 쓰기
for i in range(len(portList)): # 결과 쓰기
f.write(str(portList[i]) + ',' + dataList[i] + '\n')
f.close() # 파일 닫기
# 결과 출력
print "\n\n\n+++++++++++ the result +++++++++++"
print('portNum' + '\t' + 'banner')
for i in range(len(portList)):
print(str(portList[i]) + '\t' + dataList[i])
print "\n\n\n>>>>>>>>> the result in portScanResult.csv"
if __name__ == "__main__":
main()
|
cs |
결과 화면 : 콘솔
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
[+] Port 21 open: 220 (vsFTPd 2.3.4)
[+] Port 23 open: ÿýÿý ÿý#ÿý'
[+] Port 25 open: 220 metasploitable.l
[+] Port 80 open: <!DOCTYPE HTML PUBLI
...중략...
+++++++++++ the result +++++++++++
portNum banner
22 exist data
21 exist data
23 exist data
25 exist data
80 exist data
... 중략...
>>>>>>>>> the result in portScanResult.csv
|
cs |
결과 화면 : csv파일
실행([F5]) 후 결과를 기다리면, 첫 번째 만들었던 스캐너와는 비교할 수 없는 속도로 스캔을 시작한다. 하지만 65535포트를 모두 스캔하는 데에는 시간이 꾀 필요하니 1000번 포트 정도만 스캔하도록 하자. 만약 눈에 띄게 결과를 확인하고 싶다면 두 예제 모두 같은 리스트를 대상으로 스캔하면 결과를 확인하기 쉽다.
※ 시간 기록하기
시간의 결과를 확인하기 쉬운 방법은 시간을 기록하는 것이다. 아래 코드와 같이 time 라이브러리의 time 함수를 사용하면 시작 시간과 마침 시간을 저장할 수 있다. 마침 시간에서 시작을 감산하여 출력하면 총 프로그램의 수행시간을 알 수 있다. 다음 예제부터는 프로그램 수행 시간을 출력하도록 구성하겠다.
|
1
2
3
4
5
6
|
startTime = time.time()
... 프로그램 수행 ...
endTime = time.time()
print “executed Time : ” + (endTime-startTime)
|
cs |
-
csv 파일
csv 파일은 엑셀로 바로 결과를 출력할 수 있다. 업무 중 엑셀을 사용하는 경우가 많으니 이를 사용하면 업무를 더 효과적으로 처리할 수 있다. 형식은 매우 단순하다. MS 엑셀은 열은 쉼표(,)로 구분하고 행은 newLine으로 구분한다.
[출처] https://m.blog.naver.com/isc0304/220373652138
광고 클릭에서 발생하는 수익금은 모두 웹사이트 서버의 유지 및 관리, 그리고 기술 콘텐츠 향상을 위해 쓰여집니다.
