포트 스캐닝: 열린 문 찾기

# ⚠️ 이 코드는 허가된 환경에서만 사용하세요.
import socket

def scan_port(host: str, port: int, timeout: float = 1.0) -> tuple[int, bool, str]:
    """단일 포트 상태 확인."""
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(timeout)
    try:
        # connect_ex는 예외 대신 errno 반환 — 더 빠름
        result = s.connect_ex((host, port))
        if result == 0:
            return port, True, 'open'
        return port, False, f'closed({result})'
    except socket.timeout:
        return port, False, 'timeout'
    finally:
        s.close()

import time

def scan_serial(host: str, ports: list[int]) -> list[int]:
    open_ports = []
    start = time.time()
    for p in ports:
        port, is_open, _ = scan_port(host, p, 0.5)
        if is_open:
            open_ports.append(port)
    print(f'{len(ports)}개 포트, {time.time() - start:.1f}초')
    return open_ports

# 1024개 포트 → 약 8분 (timeout 0.5초 × 1024)

# ⚠️ 이 코드는 허가된 환경에서만 사용하세요.
from concurrent.futures import ThreadPoolExecutor, as_completed

def scan_parallel(host: str, ports: list[int], workers: int = 100) -> list[int]:
    """병렬 포트 스캔."""
    open_ports = []
    with ThreadPoolExecutor(max_workers=workers) as ex:
        futures = {ex.submit(scan_port, host, p, 1.0): p for p in ports}
        for fut in as_completed(futures):
            port, is_open, _ = fut.result()
            if is_open:
                open_ports.append(port)
    return sorted(open_ports)

# 1024개 포트 → 약 10초 (100 worker)

COMMON_PORTS = {
    'top10': [21, 22, 23, 25, 53, 80, 110, 143, 443, 3306],
    'top100_web': list(range(80, 90)) + [443, 8080, 8443, 8888],
    'all_well_known': list(range(1, 1024)),
}

WELL_KNOWN_SERVICES = {
    21:    'FTP',
    22:    'SSH',
    23:    'Telnet',
    25:    'SMTP',
    53:    'DNS',
    80:    'HTTP',
    110:   'POP3',
    143:   'IMAP',
    443:   'HTTPS',
    3306:  'MySQL',
    5432:  'PostgreSQL',
    6379:  'Redis',
    8080:  'HTTP-Alt',
    27017: 'MongoDB',
}

def label_port(port: int) -> str:
    return WELL_KNOWN_SERVICES.get(port, 'Unknown')

# ⚠️ 이 코드는 허가된 환경에서만 사용하세요.
import argparse
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed

logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s')
log = logging.getLogger('portscan')

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--host', default='127.0.0.1')
    parser.add_argument('--start', type=int, default=1)
    parser.add_argument('--end', type=int, default=1024)
    parser.add_argument('--workers', type=int, default=100)
    args = parser.parse_args()

    # 안전 검증 — 자기 호스트만 허용
    import ipaddress
    ip = ipaddress.ip_address(socket.gethostbyname(args.host))
    if not (ip.is_private or ip.is_loopback):
        raise PermissionError(f'공용 IP 스캔 금지: {ip}')

    ports = list(range(args.start, args.end + 1))
    log.info(f'{args.host} 포트 {args.start}~{args.end} 스캔 시작')

    open_ports = []
    with ThreadPoolExecutor(max_workers=args.workers) as ex:
        futures = {ex.submit(scan_port, args.host, p, 1.0): p for p in ports}
        for fut in as_completed(futures):
            port, is_open, _ = fut.result()
            if is_open:
                open_ports.append(port)
                log.warning(f'OPEN: {port}/tcp ({label_port(port)})')

    print(f'\n결과: {len(open_ports)}개 열린 포트')
    for p in sorted(open_ports):
        print(f'  {p:5d}/tcp  {label_port(p)}')

if __name__ == '__main__':
    main()