[鐵人賽] Day 27：⚙️ 用 httpx + socket 進行 IoT 批量可用性與 Port 掃描

✍️ 寫在前面
獎金獵人倒數，今天我們要做一件實用的事：
快速檢測上百個 IoT 設備的可用性與服務開放狀況。

核心概念是把 FreeBuf 的網站檢測邏輯「拆成兩層」：
1️⃣ HTTP 狀態碼檢測（確認 Web 是否可達）
2️⃣ Port 掃描（確認哪些服務端口開放）

這是紅隊賽前的最基本「偵察腳本」。

🧠 為什麼用 httpx 而非 requests

結論：

httpx 是 requests 的升級版，更適合紅隊做批量檢測與資產偵測。

📂 目標清單範例：target_ip.txt

192.168.1.10
192.168.1.20
192.168.1.30

🔍 第一部分：HTTP 可用性檢測（httpx）

此程式會針對清單內每個 IP 嘗試存取 80、443、8080、8443 端口，
回報 HTTP 狀態碼（如 200、403、404、timeout）。

import httpx
import asyncio
from tabulate import tabulate

TARGET_PORTS = [80, 443, 8080, 8443]

async def check_http(ip, port):
    protocol = "https" if port in [443, 8443] else "http"
    url = f"{protocol}://{ip}:{port}"
    try:
        async with httpx.AsyncClient(timeout=5, follow_redirects=True) as client:
            r = await client.get(url)
            return ip, port, r.status_code
    except httpx.RequestError:
        return ip, port, "無法連線"
    except Exception as e:
        return ip, port, f"錯誤: {e}"

async def run_http_scan(ip_list):
    tasks = []
    for ip in ip_list:
        for port in TARGET_PORTS:
            tasks.append(check_http(ip, port))
    results = await asyncio.gather(*tasks)

    table = [[ip, port, status] for ip, port, status in results]
    print(tabulate(table, headers=["IP 位址", "Port", "HTTP 狀態"], tablefmt="grid"))

if __name__ == "__main__":
    with open("target_ip.txt") as f:
        ip_list = [line.strip() for line in f if line.strip()]
    asyncio.run(run_http_scan(ip_list))

🧾 範例輸出

+-------------+--------+-------------+
| IP 位址      | Port   | HTTP 狀態   |
+-------------+--------+-------------+
| 192.168.1.10 | 80     | 200         |
| 192.168.1.20 | 443    | 無法連線     |
| 192.168.1.30 | 8080   | 403         |
+-------------+--------+-------------+

⚙️ 第二部分：Port 掃描（socket 驗證）

有些 IoT 設備不一定有 Web 介面（例如 MQTT、Telnet、FTP），
因此還需做 TCP Port 掃描，確認服務是否開放。

import socket
from concurrent.futures import ThreadPoolExecutor
from tabulate import tabulate

TARGET_PORTS = [21, 22, 23, 80, 443, 8080, 1883, 8883]

def check_port(ip, port, timeout=1):
    try:
        with socket.create_connection((ip, port), timeout=timeout):
            return True
    except:
        return False

def run_port_scan(ip_list):
    results = []
    with ThreadPoolExecutor(max_workers=50) as executor:
        futures = []
        for ip in ip_list:
            for port in TARGET_PORTS:
                futures.append(executor.submit(check_port, ip, port))
        idx = 0
        for ip in ip_list:
            open_ports = []
            for port in TARGET_PORTS:
                if futures[idx].result():
                    open_ports.append(str(port))
                idx += 1
            results.append([ip, ", ".join(open_ports) if open_ports else "無開放"])
    print(tabulate(results, headers=["IP 位址", "開放 Ports"], tablefmt="grid"))
    return results

if __name__ == "__main__":
    with open("target_ip.txt") as f:
        ip_list = [line.strip() for line in f if line.strip()]
    run_port_scan(ip_list)

🧾 範例輸出

+-------------+---------------------+
| IP 位址      | 開放 Ports          |
+-------------+---------------------+
| 192.168.1.10 | 22, 80, 443         |
| 192.168.1.20 | 23, 8080            |
| 192.168.1.30 | 無開放              |
+-------------+---------------------+

🧩 實戰整合建議

⚠️ 注意事項

• 僅針對官方提供清單的 IP 進行掃描。