또는 venv를 직접 관리하고 싶지 않다면 `pipx run`으로 실행할 수 있습니다. PEP-668이 적용되지 않는 구형 시스템에서는 `python3 sparkplug-fuzzer.py --setup`을 바로 실행해도 됩니다.

`--setup`은 다음을 수행합니다:
1. pip 의존성을 설치합니다 (`paho-mqtt`, `protobuf`)
2. [Eclipse Tahu](https://github.com/eclipse/tahu) 저장소의 고정된 태그를 클론합니다 (스크립트의 `TAHU_REF` 참조)
3. `sparkplug_b.py` 및 `array_packer.py` 헬퍼 모듈을 복사합니다
4. `sparkplug_b.proto`를 Python 바인딩으로 컴파일합니다 (`protoc`를 사용할 수 있으면 사용하고, 없으면 `grpcio-tools`로 대체)
5. Tahu 클론을 정리합니다

설정이 완료되면 디렉터리에는 다음이 포함되어야 합니다:```
sparkplug-fuzzer.py     # The fuzzer
sparkplug_b.py          # Sparkplug B helper module (from Tahu)
array_packer.py         # Array packing helper (from Tahu)
sparkplug_b_pb2.py      # Generated protobuf bindings
requirements.txt        # Python dependencies

이 작업은 다음을 수행합니다:
1. `localhost:1883`의 브로커에 연결합니다.
2. 기존 Sparkplug 장치를 발견하기 위해 10초 동안 수신합니다.
3. 퍼저를 Sparkplug 노드/장치로 설정합니다.
4. 12개 퍼즈 카테고리 전부를 실행합니다(~635개 이상의 테스트 케이스).
5. 스푸핑된 메시지로 발견된 모든 장치를 대상으로 합니다.
6. 결과를 `sparkplug_fuzz.jsonl`에 기록합니다.

## 사용법

### 명령줄 옵션```
python3 sparkplug-fuzzer.py [OPTIONS]

**자격 증명을 `ps`에 노출하지 않고 전달하기:**```bash
# Via environment
MQTT_USERNAME=admin MQTT_PASSWORD=secret python3 sparkplug-fuzzer.py -H broker.local

# Or read password from stdin (getpass — no echo)
python3 sparkplug-fuzzer.py -H broker.local -u admin -P -

**수동 인증 평가 + 능동 쓰기 프로브:**```bash
python3 sparkplug-fuzzer.py -H 10.0.1.30 --probe-anon-write -v

**느린 속도로 확장된 탐색 (브로커 부하 최소화):**```bash
python3 sparkplug-fuzzer.py -H 192.168.1.100 --discovery-time 60 --delay 0.5

**별도의 터미널에서 브로커 트래픽을 모니터링하세요:**```bash
mosquitto_sub -h <broker_host> -p 1883 -t 'spBv1.0/#' -F '%I %t %x'

**실행별 출력 레이아웃:**```bash
# Default — directory is auto-named under ./sparkplug-runs/
python3 sparkplug-fuzzer.py -H broker.local
# -> creates ./sparkplug-runs/2026-05-05_1830_broker.local/sparkplug_fuzz.jsonl

# Explicit directory:
python3 sparkplug-fuzzer.py -H broker.local --output-dir ./fuzz-runs/acme-2026Q2

퍼저는 시작 시 `[+] Extra string payloads: loaded N from <path>`를 출력하며, 각 페이로드는 `STRING_FUZZ_VALUES`를 반복하는 모든 위치에서 방출됩니다 — 주로 `string` 범주이지만, 타입 불일치 생성기의 문자열 타입 케이스에서도 방출됩니다.

하드 한도: 파일 크기 10MB, 페이로드 10,000개. 더 필요하다면(그리고 그에 맞는 런타임 예산이 있다면) 스크립트 상단의 `MAX_EXTRA_PAYLOADS_FILE_SIZE` / `MAX_EXTRA_PAYLOADS_COUNT`를 조정하세요.

## v0.2 릴리스 노트

- `--output-dir` 플래그 및 자동 생성되는 `./sparkplug-runs/<UTC-ts>_<host>/` 기본값 — 각 실행이 고유한 디렉터리에 저장되므로 실행 간에 아티팩트가 충돌하지 않습니다.
- `--setup`용 `--tahu-path` 플래그 — 아웃바운드 `git clone`이 차단된 에어갭(air-gapped) 테스트 환경에서 `eclipse/tahu`의 로컬 클론을 가리킵니다. 로컬 소스는 정리 시 절대 삭제되지 않습니다.
- 콘솔 및 JSONL 타임스탬프가 명시적 `Z` 접미사와 함께 UTC로 강제 설정되어 브로커 로그와의 상호 상관관계에 시간대 계산이 필요 없습니다.
- `paho.mqtt` 로거는 기본적으로 WARNING으로 제한되며, `-v`에서는 INFO, `-vv`에서는 DEBUG로 표시됩니다. 패킷별 클라이언트 텔레메트리가 더 이상 퍼즈 신호를 묻지 않습니다.
- `tests/` 아래의 pytest 테스트 하네스 — FuzzLogger, 토픽 헬퍼, 출력 경로 해석, `--tahu-path` 검증을 다루는 23개 테스트. [테스트 실행](#running-the-tests)을 참조하세요.

## 테스트 실행

테스트 하네스는 네트워크 독립적 표면(로거 정확성, 토픽 빌더, 출력 경로 해석, `--tahu-path` 파싱)을 다루며 브로커, paho-mqtt 또는 protobuf가 설치되지 않은 상태에서도 실행됩니다.```bash
pip install -r requirements-dev.txt
pytest tests/

### 네트워크 발견

발견 단계에서 퍼저는 `spBv1.0/#`에 구독하고 모든 Sparkplug 트래픽을 수신합니다. `DeviceTracker` 컴포넌트는 관찰된 메시지를 파싱하여 실시간 네트워크 맵을 구축합니다:

- **NBIRTH** 메시지는 엣지 노드와 해당 메트릭 정의(이름, 별칭, 데이터 유형)를 드러냅니다
- **DBIRTH** 메시지는 디바이스와 해당 메트릭 스키마를 드러냅니다
- **NDEATH/DDEATH** 메시지는 노드/디바이스 수명 주기 상태를 추적합니다
- **STATE** 메시지는 호스트 애플리케이션과 온라인/오프라인 상태를 드러냅니다

이 맵은 대상 퍼징 단계에서 실제 디바이스의 실제 메트릭 스키마에 맞는 맥락적 공격을 보내는 데 사용됩니다.

### 인증 평가

퍼저가 `-u/-P` 없이 연결할 때(그리고 `MQTT_USERNAME`/`MQTT_PASSWORD`가 설정되지 않은 경우), 수동적 발견만으로 브로커 인증 태세를 도출합니다. 이는 로그에 단일 `AUTH_ASSESSMENT` 이벤트와 출력 요약을 생성합니다:

| 신호 | 의미 | 도출 방법 |
|---|---|---|
| `anon_connect_accepted` | 브로커가 자격 증명 없이 CONNECT를 수락함 | 퍼저 자신의 CONNECT가 성공함 |
| `anon_subscribe_accepted` | 브로커가 익명 클라이언트에게 `spBv1.0/#` / `STATE/#`를 전달함 | 수신 대기 창 동안 RX 메시지가 하나 이상 도착함 |
| `anon_publish_accepted` | 브로커가 익명 클라이언트의 PUBLISH를 수락함 | `--probe-anon-write`가 전달된 경우에만 설정됨; QoS=1 프로브 + PUBACK 대기 |
| `unauth_endpoints` | 인증 없이 관찰 가능한 노드 / 디바이스 / 호스트 애플리케이션 | 발견된 네트워크 맵의 모든 엔티티(인증이 생성된 적 없음) |

QoS=1 프로브는 수동적 단계에서 능동적 단계로 넘어가기 때문에 옵트인 방식입니다. QoS=0에서는 브로커가 거부할 메시지를 조용히 버리므로 쓰기 수락을 확인하려면 PUBACK을 읽어야 합니다.

MQTT/Sparkplug에는 엔드포인트별 인증이 없습니다 — 인증은 브로커 수준의 문제입니다. 따라서 "인증 없이 관찰 가능한 엔드포인트"는 엔드포인트 자체의 속성이 아니라 *비용 없이 도달 가능한 대상* 목록으로 보고됩니다.

### 대상 퍼징

체계적인 퍼징 후, 이 도구는 발견된 각 디바이스를 대상으로 다음을 수행합니다:

1. **위조된 사망 알림** — NDEATH/DDEATH를 게시하여 구독자들이 디바이스가 오프라인 상태라고 믿도록 속입니다
2. **위조된 탄생 인증서** — NBIRTH/DBIRTH를 게시하여 발견된 노드/디바이스를 가장합니다
3. **명령 주입** — 알려진 각 메트릭에 대한 경계값을 가진 NCMD/DCMD 메시지를 보내 대상이 수신 명령을 검증하는지 테스트합니다
4. **Rebirth 명령** — `Node Control/Rebirth` NCMD를 보내 디바이스가 자신의 birth를 다시 게시하도록 트리거합니다

## 출력 및 로그 분석

### 로그 형식

로그 파일은 JSON-lines 형식(`.jsonl`)을 사용합니다 — 줄마다 하나의 JSON 객체가 있으며, `jq`, Python 또는 JSON을 지원하는 모든 도구로 분석하기에 적합합니다.

64 KiB보다 큰 페이로드는 hex 인라인 처리되지 않습니다. 대신 `payload_hex`는 `sha256:<digest>+len=<n>`을 담아 매우 큰 퍼징 케이스에서도 로그 크기가 제한되도록 합니다. `payload_len`은 항상 존재합니다.

**TX 레코드**(발신 퍼징 메시지):```json
{
  "ts": "2026-04-10T15:30:00.123456Z",
  "dir": "TX",
  "case_id": "BOUNDARY-0042",
  "category": "boundary",
  "topic": "spBv1.0/Sparkplug B Devices/DDATA/FuzzNode/FuzzDevice",
  "payload_hex": "0800120a0a06...",
  "payload_len": 28,
  "payload_decoded": {"timestamp": 1712345678000, "metrics": [{"name": "fuzz/boundary/Int32", "datatype": 3, "int_value": 2147483647}]},
  "description": "Boundary Int32 = 2147483647 (int_value)"
}

**이벤트 기록** (시스템 이벤트):```json
{
  "ts": "2026-04-10T15:29:50.000000Z",
  "dir": "EVENT",
  "event": "DISCOVERY_COMPLETE",
  "details": {"groups": ["Production"], "node_count": 3, "device_count": 7, "targets": 10}
}

**모든 문자열 삽입 사례 추출:**```bash
jq 'select(.category == "string")' sparkplug_fuzz.jsonl

**브로커 연결 끊김을 유발한 사례를 찾습니다:**```bash
jq 'select(.event == "UNEXPECTED_DISCONNECT" or .event == "RECONNECT_FAIL")' sparkplug_fuzz.jsonl

**인증 없이 도달 가능한 엔드포인트 나열:**```bash
jq -r 'select(.event == "AUTH_ASSESSMENT") | .details.unauth_endpoints[] | [.kind, .group, .node, .device, .host_id, .status] | @tsv' sparkplug_fuzz.jsonl

**게시된 모든 토픽 내보내기:**```bash
jq -r 'select(.dir == "TX") | .topic' sparkplug_fuzz.jsonl | sort -u

## 프로토콜 커버리지

### 메시지 유형

9가지 Sparkplug B 메시지 유형 모두 테스트됩니다:

| 메시지 유형 | 토픽 패턴 | 설명 | 퍼저 사용 |
|---|---|---|---|
| NBIRTH | `spBv1.0/{group}/NBIRTH/{node}` | 노드 탄생 인증서 | 퍼저 존재 확립; 발견된 노드로 위장; 순서 테스트 |
| NDEATH | `spBv1.0/{group}/NDEATH/{node}` | 노드 사망 알림 | MQTT last-will; 발견된 노드로 위장; 순서 테스트 |
| DBIRTH | `spBv1.0/{group}/DBIRTH/{node}/{device}` | 디바이스 탄생 인증서 | 퍼저 디바이스 확립; 발견된 디바이스로 위장; 순서 테스트 |
| DDEATH | `spBv1.0/{group}/DDEATH/{node}/{device}` | 디바이스 사망 알림 | 발견된 디바이스로 위장; 순서 테스트; 고아(orphan) 테스트 |
| NDATA | `spBv1.0/{group}/NDATA/{node}` | 노드 데이터 업데이트 | 경계 값; 시퀀스 번호; 순서 테스트 |
| DDATA | `spBv1.0/{group}/DDATA/{node}/{device}` | 디바이스 데이터 업데이트 | 대부분의 퍼즈 카테고리를 위한 주요 전달 수단 |
| NCMD | `spBv1.0/{group}/NCMD/{node}` | 노드 명령 | 타겟 퍼징(rebirth 명령); 고아(orphan) 테스트 |
| DCMD | `spBv1.0/{group}/DCMD/{node}/{device}` | 디바이스 명령 | 발견된 디바이스 메트릭에 대한 타겟 퍼징; 고아(orphan) 테스트 |
| STATE | `STATE/{host_id}` | 호스트 애플리케이션 상태(JSON) | 잘못된 JSON 인젝션 |

### 데이터 유형

19가지 Sparkplug B 메트릭 데이터 유형 모두 유형별 경계 값으로 테스트됩니다:

| 코드 | 유형 | Protobuf 필드 | 테스트된 경계 값 |
|------|------|---------------|----------------------|
| 1 | Int8 | int_value | 0, -128, 127, 128(오버플로), -129(언더플로) |
| 2 | Int16 | int_value | 0, -32768, 32767, 오버플로/언더플로 |
| 3 | Int32 | int_value | 0, -2^31, 2^31-1, 오버플로/언더플로 |
| 4 | Int64 | long_value | 0, -2^63, 2^63-1, 오버플로 |
| 5 | UInt8 | int_value | 0, 255, 256, -1 |
| 6 | UInt16 | int_value | 0, 65535, 65536, -1 |
| 7 | UInt32 | int_value | 0, 4294967295, -1 |
| 8 | UInt64 | long_value | 0, 2^64-1, -1 |
| 9 | Float | float_value | 0.0, -0.0, 최대, 최소, inf, -inf, NaN |
| 10 | Double | double_value | 0.0, -0.0, 최대, 최소, inf, -inf, NaN |
| 11 | Boolean | boolean_value | True, False; 원시 int 값(0, 1, 2, 255)으로도 테스트 |
| 12 | String | string_value | 빈 값, 긴 값(최대 64KB), 인젝션 페이로드 |
| 13 | DateTime | long_value | Epoch, 최대, 먼 미래/과거 |
| 14 | Text | string_value | String과 동일한 인젝션 페이로드 |
| 15 | UUID | string_value | 빈 값, 유효, 잘못된 형식, 인젝션 |
| 16 | DataSet | dataset_value | dataset 카테고리를 통한 구조적 위반 |
| 17 | Bytes | bytes_value | 빈 값, 널 바이트, 랜덤, 대용량 |
| 18 | File | bytes_value | 빈 값, 매직 바이트, 대용량 |
| 19 | Template | template_value | 정의되지 않은 참조, 고아(orphan) 템플릿 |

### 필드 커버리지

퍼저는 87개 이상의 고유 protobuf 필드 경로를 다룹니다:

- **페이로드 루트 필드**: timestamp, seq, uuid, body, metrics
- **메트릭 필드**: name, alias, timestamp, datatype, is_historical, is_transient, is_null, metadata, properties 및 모든 value oneof 변형
- **MetaData 필드**: is_multi_part, content_type, size, seq, file_name, file_type, md5, description
- **PropertySet/PropertyValue**: keys, values, type, is_null, 재귀적 propertyset_value, propertysets_value
- **DataSet**: num_of_columns, columns, types, rows, elements, 모든 DataSetValue 변형
- **Template**: version, template_ref, is_definition, 중첩 메트릭, parameters

## 아키텍처

퍼저는 다음 구성 요소로 구성된 단일 Python 파일입니다:```
sparkplug-fuzzer.py
    |
    +-- Constants / ALL_METRIC_TYPES / STRING_FUZZ_VALUES
    |       Type definitions and fuzz value tables
    |
    +-- FuzzLogger
    |       JSON-lines file logging + console output
    |       Protobuf payload decoding
    |
    +-- DeviceTracker
    |       Passive network discovery
    |       Tracks groups, nodes, devices, metrics
    |
    +-- PayloadBuilder
    |       Valid payload construction (sparkplug_b helpers)
    |       Raw payload construction (sparkplug_b_pb2 direct)
    |       Binary corruption (truncate, flip, append)
    |
    +-- 12 Fuzz Generators
    |       Each is a Python generator yielding (topic, bytes, desc)
    |       Covers boundary, string, type, seq, timestamp, alias,
    |       orphan, ordering, recursive, dataset, malformed, topic
    |
    +-- SparkplugFuzzer
    |       Orchestration: connect, discover, fuzz, target, report
    |       Centralized publish with logging
    |       Auto-reconnect on disconnect
    |
    +-- CLI (argparse) + main()
            Argument parsing and entry point