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())