종합: 네트워크 정찰 도구

$ python netrecon.py --target localhost --ports 1-1024 --output report.json

🎯 대상: localhost (127.0.0.1)
📡 DNS: A=[127.0.0.1] · MX=[]
📊 포트 스캔: 1-1024 (1024개)
✅ 열린 포트: 5
   22/tcp   SSH      OpenSSH_8.4p1
   80/tcp   HTTP     nginx/1.18.0
   443/tcp  HTTPS    nginx/1.18.0
   3000/tcp Unknown  Node.js
   5432/tcp PostgreSQL 13.x

📁 저장: report.json

# netrecon.py — 네트워크 정찰 통합 도구
# ⚠️ 이 코드는 허가된 환경에서만 사용하세요.

import argparse
import json
import logging
import socket
import ipaddress
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
import dns.resolver

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
)
log = logging.getLogger('netrecon')


WELL_KNOWN = {
    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', 8443: 'HTTPS-Alt', 27017: 'MongoDB',
}


def resolve_target(target: str) -> tuple[str, str]:
    """도메인 또는 IP를 받아 (ip, hostname) 반환."""
    try:
        ip = socket.gethostbyname(target)
    except socket.gaierror:
        raise SystemExit(f'❌ DNS 해석 실패: {target}')

    addr = ipaddress.ip_address(ip)
    if not (addr.is_private or addr.is_loopback):
        raise SystemExit(f'❌ 공용 IP 스캔 금지 (허가된 환경만): {ip}')

    try:
        hostname = socket.gethostbyaddr(ip)[0]
    except socket.herror:
        hostname = target
    return ip, hostname


def dns_records(domain: str) -> dict:
    """주요 DNS 레코드 수집."""
    records = {}
    for rtype in ['A', 'AAAA', 'MX', 'NS', 'TXT', 'CNAME']:
        try:
            answers = dns.resolver.resolve(domain, rtype, lifetime=5)
            records[rtype] = [str(r) for r in answers]
        except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN, dns.exception.Timeout):
            records[rtype] = []
    return records


def scan_port(host: str, port: int, timeout: float = 1.0) -> tuple[int, bool]:
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(timeout)
    try:
        return port, s.connect_ex((host, port)) == 0
    except (socket.timeout, OSError):
        return port, False
    finally:
        s.close()


def grab_banner(host: str, port: int, timeout: float = 3.0) -> str:
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(timeout)
    try:
        s.connect((host, port))
        try:
            data = s.recv(1024)
            if data:
                return data.decode('utf-8', errors='replace').strip()[:200]
        except socket.timeout:
            pass
        if port in (80, 8080, 8443, 443):
            s.send(b'HEAD / HTTP/1.0\r\nHost: ' + host.encode() + b'\r\n\r\n')
            return s.recv(2048).decode('utf-8', errors='replace').strip()[:300]
        return ''
    except (socket.timeout, ConnectionRefusedError, OSError):
        return ''
    finally:
        s.close()


def parse_port_range(spec: str) -> list[int]:
    """1-1024 또는 22,80,443 형식 파싱."""
    result = set()
    for part in spec.split(','):
        if '-' in part:
            start, end = part.split('-')
            result.update(range(int(start), int(end) + 1))
        else:
            result.add(int(part))
    return sorted(result)


def reconnaissance(target: str, ports_spec: str, workers: int = 100) -> dict:
    """통합 정찰 흐름."""
    ip, hostname = resolve_target(target)

    # 1. DNS
    log.info(f'DNS 조회: {target}')
    dns_data = dns_records(target) if target != ip else {}

    # 2. 포트 스캔
    ports = parse_port_range(ports_spec)
    log.info(f'포트 스캔: {len(ports)}개')
    open_ports = []
    with ThreadPoolExecutor(max_workers=workers) as ex:
        futures = {ex.submit(scan_port, ip, 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)
    open_ports.sort()
    log.info(f'열린 포트: {len(open_ports)}')

    # 3. 서비스 감지 (병렬)
    services = []
    with ThreadPoolExecutor(max_workers=20) as ex:
        futures = {ex.submit(grab_banner, ip, p, 3.0): p for p in open_ports}
        for fut in as_completed(futures):
            port = futures[fut]
            banner = fut.result()
            services.append({
                'port': port,
                'service': WELL_KNOWN.get(port, 'Unknown'),
                'banner': banner.split('\n')[0] if banner else '',
            })
    services.sort(key=lambda x: x['port'])

    return {
        'target': target,
        'resolved': {'ip': ip, 'hostname': hostname},
        'dns': dns_data,
        'open_ports': open_ports,
        'services': services,
    }


def print_report(report: dict) -> None:
    print(f'\n🎯 대상: {report["target"]} ({report["resolved"]["ip"]})')
    print(f'   호스트명: {report["resolved"]["hostname"]}')

    if report['dns']:
        print('\n📡 DNS:')
        for rtype, records in report['dns'].items():
            if records:
                print(f'   {rtype}: {records[:3]}')

    print(f'\n📊 열린 포트: {len(report["open_ports"])}')
    for svc in report['services']:
        banner_preview = svc['banner'][:60] if svc['banner'] else '-'
        print(f'   {svc["port"]:5d}/tcp  {svc["service"]:10s}  {banner_preview}')


def main():
    parser = argparse.ArgumentParser(description='네트워크 정찰 통합 도구')
    parser.add_argument('--target', '-t', required=True)
    parser.add_argument('--ports', '-p', default='1-1024')
    parser.add_argument('--workers', '-w', type=int, default=100)
    parser.add_argument('--output', '-o', help='JSON 저장 경로')
    parser.add_argument('--json', '-j', action='store_true', help='JSON 출력')
    parser.add_argument('--verbose', '-v', action='store_true')
    args = parser.parse_args()

    if args.verbose:
        log.setLevel(logging.DEBUG)

    report = reconnaissance(args.target, args.ports, args.workers)

    if args.json:
        out = json.dumps(report, indent=2, ensure_ascii=False)
        print(out)
        if args.output:
            with open(args.output, 'w', encoding='utf-8') as f:
                f.write(out)
    else:
        print_report(report)
        if args.output:
            with open(args.output, 'w', encoding='utf-8') as f:
                json.dump(report, f, indent=2, ensure_ascii=False)
            print(f'\n📁 저장: {args.output}')

    sys.exit(0 if report['open_ports'] else 1)


if __name__ == '__main__':
    main()

# 로컬 서버 정찰
python netrecon.py --target localhost --ports 1-1024

# 사내망 서버
python netrecon.py --target 192.168.1.100 --ports 22,80,443,3306,5432 --output server_audit.json

# JSON 출력 (CI/CD 통합)
python netrecon.py --target localhost --json | jq '.services'