Rinjae
344 lines
13 KiB
Python
344 lines
13 KiB
Python
import argparse
|
|
from datetime import datetime
|
|
import http.cookiejar
|
|
from http.cookies import SimpleCookie
|
|
import json
|
|
import ssl
|
|
import sys
|
|
import time
|
|
import unicodedata
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
DEFAULT_HOST = 'https://api.jejubank.co.kr'
|
|
COOKIE_NAME = 'JSESSIONID_PORTAL'
|
|
USER_AGENT = (
|
|
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
|
|
'AppleWebKit/537.36 (KHTML, like Gecko) '
|
|
'Chrome/140.0.0.0 Safari/537.36'
|
|
)
|
|
COLUMNS = (
|
|
('회', 3),
|
|
('대상', 4),
|
|
('결과', 10),
|
|
('쿠키', 8),
|
|
('CSRF', 8),
|
|
('HTTP', 4),
|
|
('요청 값', 10),
|
|
('요청 서버ID', 11),
|
|
('응답 값', 10),
|
|
('응답 서버ID', 11),
|
|
)
|
|
BORDER = '+' + '+'.join('-' * (width + 2) for _, width in COLUMNS) + '+'
|
|
|
|
|
|
class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, request, fp, code, message, headers, new_url):
|
|
# 두 서버 비교에서는 지정한 주소의 첫 응답만 검사한다.
|
|
# 공유 세션 쿠키를 리다이렉트 대상에 전달하지 않는다.
|
|
return None
|
|
|
|
|
|
def integer_between(minimum, maximum):
|
|
def parse(value):
|
|
try:
|
|
number = int(value)
|
|
except ValueError:
|
|
raise argparse.ArgumentTypeError(
|
|
f'{minimum}~{maximum} 사이의 정수를 입력하세요.'
|
|
)
|
|
if not minimum <= number <= maximum:
|
|
raise argparse.ArgumentTypeError(
|
|
f'{minimum}~{maximum} 사이의 정수를 입력하세요.'
|
|
)
|
|
return number
|
|
return parse
|
|
|
|
|
|
def host_address(value):
|
|
value = value.strip()
|
|
message = 'http://호스트[:포트] 또는 https://호스트[:포트] 형식으로 입력하세요.'
|
|
try:
|
|
parsed = urllib.parse.urlsplit(value)
|
|
parsed.port # 포트 형식과 범위도 검사한다.
|
|
except ValueError:
|
|
raise argparse.ArgumentTypeError(message)
|
|
if (parsed.scheme not in ('http', 'https') or not parsed.hostname
|
|
or parsed.path not in ('', '/') or parsed.query or parsed.fragment
|
|
or parsed.username is not None or parsed.password is not None
|
|
or any(char.isspace() for char in value)):
|
|
raise argparse.ArgumentTypeError(message)
|
|
return urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, '', '', ''))
|
|
|
|
|
|
def parse_args(argv=None):
|
|
parser = argparse.ArgumentParser(
|
|
description='새 익명 세션으로 쿠키와 CSRF 토큰의 유지 여부를 표로 확인합니다.'
|
|
)
|
|
parser.add_argument(
|
|
'--host', type=host_address, default=DEFAULT_HOST,
|
|
metavar='주소', help=f'대상 호스트 주소 (기본값: {DEFAULT_HOST})',
|
|
)
|
|
parser.add_argument(
|
|
'--host2', type=host_address, metavar='주소',
|
|
help='두 번째 서버. A/B를 번갈아 요청 (리다이렉트 미추적)',
|
|
)
|
|
parser.add_argument(
|
|
'--cookie-mode', choices=('shared', 'separate'), default='shared',
|
|
help='shared: 직전 세션 쿠키 공유, separate: 서버 주소/포트별 저장소 분리 (기본값: shared)',
|
|
)
|
|
parser.add_argument(
|
|
'-n', '--count', type=integer_between(1, 100), default=10,
|
|
metavar='횟수', help='전체 요청 횟수: 1~100회 (기본값: 10회, 두 서버 모드도 합산)',
|
|
)
|
|
parser.add_argument(
|
|
'-d', '--delay', '--delay-ms', type=integer_between(0, 10000), default=50,
|
|
metavar='밀리초', help='요청 사이 대기시간: 0~10000ms (기본값: 50ms)',
|
|
)
|
|
tls_options = parser.add_mutually_exclusive_group()
|
|
tls_options.add_argument(
|
|
'--cacert', metavar='CA파일',
|
|
help='신뢰할 사설 CA 인증서 또는 인증서 묶음 파일 (PEM 형식)',
|
|
)
|
|
tls_options.add_argument(
|
|
'-k', '--insecure', action='store_true',
|
|
help='HTTPS 서버 인증서와 호스트명 검증 생략',
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def tls_context(args):
|
|
context = ssl.create_default_context()
|
|
if args.cacert:
|
|
context.load_verify_locations(cafile=args.cacert)
|
|
if args.insecure:
|
|
context.check_hostname = False
|
|
context.verify_mode = ssl.CERT_NONE
|
|
return context
|
|
|
|
|
|
def cookie_from_headers(headers):
|
|
value = None
|
|
for header in headers:
|
|
parsed = SimpleCookie()
|
|
parsed.load(header)
|
|
if COOKIE_NAME in parsed:
|
|
value = parsed[COOKIE_NAME].value
|
|
return value
|
|
|
|
|
|
def cookie_for_url(jar, url):
|
|
request = urllib.request.Request(url)
|
|
jar.add_cookie_header(request)
|
|
return cookie_from_headers([request.get_header('Cookie', '')])
|
|
|
|
|
|
def host_key(url):
|
|
parsed = urllib.parse.urlsplit(url)
|
|
port = parsed.port if parsed.port is not None else (443 if parsed.scheme == 'https' else 80)
|
|
return parsed.scheme, parsed.hostname, port
|
|
|
|
|
|
def carry_session_cookie(request, value):
|
|
# 현재 호스트에 해당하는 다른 쿠키는 유지하고 포털 세션 쿠키만 이어 보낸다.
|
|
cookies = SimpleCookie()
|
|
cookies.load(request.get_header('Cookie', ''))
|
|
cookies[COOKIE_NAME] = value
|
|
request.add_unredirected_header(
|
|
'Cookie', cookies.output(header='', sep='; ').strip()
|
|
)
|
|
|
|
|
|
def server_id(cookie):
|
|
return cookie.split('!')[1] if cookie and '!' in cookie else '-'
|
|
|
|
|
|
def cell(value, width):
|
|
text = str(value)
|
|
# 한글은 보통 터미널에서 두 칸을 차지하므로 표시 폭을 기준으로 정렬한다.
|
|
display_width = sum(
|
|
0 if unicodedata.combining(char) else
|
|
2 if unicodedata.east_asian_width(char) in ('W', 'F') else 1
|
|
for char in text
|
|
)
|
|
return text + ' ' * max(0, width - display_width)
|
|
|
|
|
|
def print_row(values):
|
|
print('| ' + ' | '.join(
|
|
cell(value, width) for value, (_, width) in zip(values, COLUMNS)
|
|
) + ' |', flush=True)
|
|
|
|
|
|
def change_state(previous, current):
|
|
if previous is None:
|
|
return '최초'
|
|
return '유지' if previous == current else '!!변경!!'
|
|
|
|
|
|
def run_probe(args):
|
|
targets = [('A', args.host + '/api/session/csrf')]
|
|
if args.host2:
|
|
targets.append(('B', args.host2 + '/api/session/csrf'))
|
|
try:
|
|
context = tls_context(args)
|
|
except (OSError, ValueError) as error:
|
|
print(f'TLS 설정 실패: {error}', file=sys.stderr)
|
|
return 1
|
|
jar = http.cookiejar.CookieJar()
|
|
cookie_processor = urllib.request.HTTPCookieProcessor(jar)
|
|
handlers = [
|
|
urllib.request.HTTPSHandler(context=context),
|
|
cookie_processor,
|
|
]
|
|
if args.host2:
|
|
handlers.append(NoRedirectHandler())
|
|
client = urllib.request.build_opener(*handlers)
|
|
previous_cookie = None
|
|
previous_token = None
|
|
cookie_stores = {}
|
|
session_states = {}
|
|
totals = {'최초': 0, '유지': 0, '변경': 0}
|
|
completed = 0
|
|
failure = None
|
|
exit_code = 0
|
|
|
|
for target, url in targets:
|
|
print(f'대상 {target}: {url}')
|
|
if len(targets) == 2:
|
|
print('호출 순서: A -> B -> A -> B ... (클라이언트 1개, 전체 요청 횟수 기준)')
|
|
print('두 서버 비교에서는 3xx 리다이렉트를 따라가지 않습니다.')
|
|
if args.cookie_mode == 'separate':
|
|
print('쿠키 모드: separate | 서버 주소/포트별 저장소 및 변경 비교 기준 분리')
|
|
elif len(targets) == 2:
|
|
print(f'쿠키 모드: shared | 직전 {COOKIE_NAME}을 다음 서버로 전달')
|
|
if any(url.startswith('https://') for _, url in targets):
|
|
tls_mode = ('생략 (--insecure)' if args.insecure else
|
|
f'사설 CA 추가 ({args.cacert})' if args.cacert else '기본 CA 사용')
|
|
print(f'TLS 인증서 검증: {tls_mode}')
|
|
print(f'쿠키 키: {COOKIE_NAME} | 값: 앞 10자리 | 횟수: {args.count}회 | 딜레이: {args.delay}ms')
|
|
print(BORDER)
|
|
print_row([title for title, _ in COLUMNS])
|
|
print(BORDER, flush=True)
|
|
|
|
try:
|
|
for number in range(1, args.count + 1):
|
|
if number > 1 and args.delay:
|
|
time.sleep(args.delay / 1000)
|
|
|
|
target, url = targets[(number - 1) % len(targets)]
|
|
if args.cookie_mode == 'separate':
|
|
key = host_key(url)
|
|
if key not in cookie_stores:
|
|
cookie_stores[key] = http.cookiejar.CookieJar()
|
|
jar = cookie_stores[key]
|
|
# 순차 요청마다 같은 클라이언트의 쿠키 저장소만 교체한다.
|
|
cookie_processor.cookiejar = jar
|
|
previous_cookie, previous_token = session_states.get(key, (None, None))
|
|
request = urllib.request.Request(
|
|
url,
|
|
headers={
|
|
'Accept': 'application/json',
|
|
'Cache-Control': 'no-cache',
|
|
'User-Agent': USER_AGENT,
|
|
},
|
|
)
|
|
# 응답이 저장소를 갱신하기 전에 실제 전송할 쿠키를 기록한다.
|
|
jar.add_cookie_header(request)
|
|
if args.host2 and args.cookie_mode == 'shared' and previous_cookie is not None:
|
|
carry_session_cookie(request, previous_cookie)
|
|
request_cookie = cookie_from_headers([request.get_header('Cookie', '')])
|
|
response_cookie = None
|
|
status = '-'
|
|
try:
|
|
with client.open(request, timeout=10) as response:
|
|
status = response.status
|
|
response_url = response.geturl()
|
|
# 이번 응답의 Set-Cookie만 표시한다. 재발급이 없으면 '없음'.
|
|
response_cookie = cookie_from_headers(
|
|
response.headers.get_all('Set-Cookie') or []
|
|
)
|
|
data = json.load(response)
|
|
|
|
if args.host2 and args.cookie_mode == 'shared' and response_cookie is None:
|
|
# 다른 호스트에 직접 이어 보낸 쿠키는 저장소에 없을 수 있다.
|
|
# 재발급이 없으면 이번 요청에 실었던 세션을 계속 사용한다.
|
|
cookie = request_cookie
|
|
else:
|
|
cookie = cookie_for_url(jar, response_url)
|
|
token = data.get('token') if isinstance(data, dict) else None
|
|
if not cookie or not token:
|
|
raise ValueError('세션 쿠키 또는 CSRF 토큰 없음')
|
|
except urllib.error.HTTPError as error:
|
|
status = error.code
|
|
response_cookie = cookie_from_headers(
|
|
error.headers.get_all('Set-Cookie') or []
|
|
)
|
|
failure = f'{number}회: HTTP {status}'
|
|
error.close()
|
|
except Exception as error:
|
|
failure = f'{number}회: {error}'
|
|
|
|
if failure:
|
|
result, cookie_state, token_state = '!!실패!!', '-', '-'
|
|
exit_code = 1
|
|
else:
|
|
# 변경 판정은 앞 10자리가 아니라 전체 쿠키와 전체 토큰으로 비교한다.
|
|
cookie_state = change_state(previous_cookie, cookie)
|
|
token_state = change_state(previous_token, token)
|
|
if '!!변경!!' in (cookie_state, token_state):
|
|
outcome, result = '변경', '>>>변경<<<'
|
|
elif previous_cookie is None:
|
|
outcome, result = '최초', '[최초]'
|
|
else:
|
|
outcome, result = '유지', '[유지]'
|
|
totals[outcome] += 1
|
|
completed += 1
|
|
previous_cookie, previous_token = cookie, token
|
|
if args.cookie_mode == 'separate':
|
|
session_states[host_key(url)] = (cookie, token)
|
|
|
|
print_row((
|
|
number, target, result, cookie_state, token_state, status,
|
|
request_cookie[:10] if request_cookie else '없음',
|
|
server_id(request_cookie),
|
|
response_cookie[:10] if response_cookie else '없음',
|
|
server_id(response_cookie),
|
|
))
|
|
if failure:
|
|
break
|
|
except KeyboardInterrupt:
|
|
failure = '사용자가 중단했습니다.'
|
|
exit_code = 130
|
|
|
|
print(BORDER)
|
|
print(f'정상 조회: {completed}/{args.count}회 | 최초: {totals["최초"]}회 | 유지: {totals["유지"]}회 | 변경: {totals["변경"]}회')
|
|
print('응답 값=없음: Set-Cookie 재발급 없음. 서버ID는 각 쿠키의 ! 뒤 식별값입니다.')
|
|
if failure:
|
|
print(f'실패/중단: {failure}')
|
|
return exit_code
|
|
|
|
|
|
def main(argv=None):
|
|
args = parse_args(argv)
|
|
started_counter = time.perf_counter()
|
|
started_at = datetime.now().astimezone()
|
|
print(f'시작 시각: {started_at.isoformat(sep=" ", timespec="milliseconds")}', flush=True)
|
|
try:
|
|
return run_probe(args)
|
|
except KeyboardInterrupt:
|
|
print('실패/중단: 사용자가 중단했습니다.')
|
|
return 130
|
|
finally:
|
|
finished_at = datetime.now().astimezone()
|
|
# 시스템 시각 보정에 영향받지 않도록 경과 시간은 별도 시계로 측정한다.
|
|
elapsed = time.perf_counter() - started_counter
|
|
print(f'종료 시각: {finished_at.isoformat(sep=" ", timespec="milliseconds")}')
|
|
print(f'총 소요 시간: {elapsed:.3f}초 (요청 간 딜레이 포함)', flush=True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|