서브넷 스캐너 만들기

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

def list_subnet_hosts(cidr: str) -> list[str]:
    """CIDR 표기로 서브넷 내 모든 호스트 IP 나열."""
    network = ipaddress.ip_network(cidr, strict=False)
    return [str(ip) for ip in network.hosts()]

# /24 서브넷 — 254개 호스트
hosts = list_subnet_hosts('192.168.1.0/24')
print(len(hosts))  # 254
print(hosts[:3])   # ['192.168.1.1', '192.168.1.2', '192.168.1.3']

# /16 서브넷은 65,534개 — 신중히
# 너무 큰 서브넷 방어
def safe_list_hosts(cidr: str, max_hosts: int = 1024) -> list[str]:
    network = ipaddress.ip_network(cidr, strict=False)
    if network.num_addresses > max_hosts:
        raise ValueError(f'{cidr}: {network.num_addresses}개 호스트 — 너무 큼 (최대 {max_hosts})')
    return [str(ip) for ip in network.hosts()]

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

def is_alive(host: str, timeout: int = 1) -> tuple[str, bool]:
    """단일 호스트 ping 1회로 도달 여부 확인."""
    is_windows = platform.system().lower() == 'windows'
    if is_windows:
        cmd = ['ping', '-n', '1', '-w', str(timeout * 1000), host]
    else:
        cmd = ['ping', '-c', '1', '-W', str(timeout), host]

    try:
        result = subprocess.run(cmd, capture_output=True, timeout=timeout + 1)
        return host, result.returncode == 0
    except subprocess.TimeoutExpired:
        return host, False

def ping_sweep(cidr: str, workers: int = 50) -> list[str]:
    """서브넷 전체 ping sweep — 활성 호스트 추출."""
    hosts = safe_list_hosts(cidr)
    alive = []
    with ThreadPoolExecutor(max_workers=workers) as ex:
        futures = {ex.submit(is_alive, h, 1): h for h in hosts}
        for fut in as_completed(futures):
            host, ok = fut.result()
            if ok:
                alive.append(host)
    return sorted(alive, key=lambda x: ipaddress.ip_address(x))

# /24 서브넷 sweep → 약 5초
# alive_hosts = ping_sweep('192.168.1.0/24')

import socket

def tcp_alive(host: str, ports: list[int] = None, timeout: float = 1.0) -> bool:
    """일반적 포트 중 하나라도 응답하면 활성."""
    if ports is None:
        ports = [80, 443, 22, 8080]

    for port in ports:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(timeout)
        try:
            if s.connect_ex((host, port)) == 0:
                return True
        finally:
            s.close()
    return False

# pip install scapy
# 주의: ARP 스캔은 root/Administrator 권한 필요
# ⚠️ 이 코드는 허가된 환경에서만 사용하세요.

# from scapy.all import ARP, Ether, srp
#
# def arp_scan(cidr: str, timeout: int = 2) -> list[dict]:
#     """LAN ARP 스캔 — IP+MAC 페어 수집."""
#     arp = ARP(pdst=cidr)
#     ether = Ether(dst='ff:ff:ff:ff:ff:ff')
#     packet = ether / arp
#     answered, _ = srp(packet, timeout=timeout, verbose=False)
#     return [{'ip': r.psrc, 'mac': r.hwsrc} for _, r in answered]
#
# 후속 트랙(scapy 모듈)에서 자세히 다룸

def discover_hosts(cidr: str, methods: list[str] = None, workers: int = 50) -> dict:
    """ICMP + TCP 결합 — 어떤 방어를 뚫고도 호스트 탐지."""
    if methods is None:
        methods = ['icmp', 'tcp']

    hosts = safe_list_hosts(cidr)
    result = {'cidr': cidr, 'total': len(hosts), 'alive': []}

    with ThreadPoolExecutor(max_workers=workers) as ex:
        futures = {}
        for h in hosts:
            futures[ex.submit(_check_host, h, methods)] = h

        for fut in as_completed(futures):
            host, info = fut.result()
            if info['alive']:
                result['alive'].append({'host': host, **info})

    result['alive'].sort(key=lambda x: ipaddress.ip_address(x['host']))
    return result


def _check_host(host: str, methods: list[str]) -> tuple[str, dict]:
    info = {'alive': False, 'methods': []}
    if 'icmp' in methods:
        _, ok = is_alive(host)
        if ok:
            info['alive'] = True
            info['methods'].append('icmp')
    if 'tcp' in methods and not info['alive']:
        if tcp_alive(host):
            info['alive'] = True
            info['methods'].append('tcp')
    return host, info

# 안전한 기본값
SAFE_DEFAULTS = {
    'workers': 50,           # 동시 50개
    'timeout': 1.0,          # 1초
    'rate_per_host': 'once', # 호스트당 1회만
    'delay_between_batches': 0.1,  # 배치 간 100ms
}