496 lines
18 KiB
Python
496 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
"""Teach unknown-frame analysis with Netzob on Tianwen-1 raw frame data.
|
||
|
||
这个脚本的目标不是“直接使用已知的 Tianwen-1 / CCSDS 解析器”，而是假设我们
|
||
只拿到一段连续的二进制帧数据，不知道具体空间帧协议，然后用 Netzob 的核心
|
||
概念做一次可运行的协议探索教学。
|
||
|
||
重点演示的 Netzob 概念：
|
||
1. RawMessage：把每一帧原始字节包装成 Netzob 消息。
|
||
2. Symbol：把一组相似消息放进同一个协议符号。
|
||
3. Format.splitStatic：根据样本中固定/变化的字节位置自动切字段。
|
||
4. Format.clusterByKeyField：选择某个字段作为 key，把消息按字段值聚类。
|
||
5. Field / Raw：在已有观察基础上，手工建立一个“候选帧格式”模型。
|
||
|
||
注意：
|
||
- 本脚本不会 import Tianwen.ccsds，也不会调用 AOSFrame.parse。
|
||
- 为了教学和运行速度，默认只抽样前 96 帧做 Netzob 推断。
|
||
- 原始数据较大，完整协议逆向通常需要多轮实验；这里侧重方法和工具用法。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import math
|
||
import sys
|
||
from collections import Counter
|
||
from pathlib import Path
|
||
|
||
# 让 print 尽量按行输出，便于长流程运行时看到进度。
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(line_buffering=True)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 路径准备
|
||
# ---------------------------------------------------------------------------
|
||
# 当前脚本位于 /home/zjz/CCSDS_study/test/。
|
||
# parent.parent 回到项目根目录 /home/zjz/CCSDS_study。
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
|
||
# 本仓库里有一份本地 Netzob 源码/测试目录。优先使用它，保证和仓库现有示例一致。
|
||
LOCAL_NETZOB_SRC = PROJECT_ROOT / "netzob-030" / "test" / "src"
|
||
if LOCAL_NETZOB_SRC.exists():
|
||
sys.path.insert(0, str(LOCAL_NETZOB_SRC))
|
||
|
||
# 导入 Netzob 公共 API。
|
||
# noqa 注释是告诉代码检查器：星号导入是教学脚本为了贴近 Netzob 教程而保留。
|
||
from netzob.all import * # noqa: F401,F403,E402
|
||
|
||
|
||
# 原始天问一号帧字节文件。这里把它当成未知二进制样本，不使用已知解析器。
|
||
DEFAULT_INPUT = PROJECT_ROOT / "Tianwen" / "tianwen1_frames_20200730.u8"
|
||
|
||
|
||
def section(title: str) -> None:
|
||
"""打印一个清晰的教学分节标题。"""
|
||
|
||
print("\n" + "=" * 78)
|
||
print(title)
|
||
print("=" * 78)
|
||
|
||
|
||
def short_hex(data: bytes, max_bytes: int = 48) -> str:
|
||
"""把 bytes 转成短十六进制字符串，避免一帧 220 字节全部刷屏。"""
|
||
|
||
head = data[:max_bytes].hex(" ")
|
||
return head + (" ..." if len(data) > max_bytes else "")
|
||
|
||
|
||
def entropy(values) -> float:
|
||
"""计算一组离散值的 Shannon entropy。
|
||
|
||
entropy 越低，说明这个字节位置越稳定，越像版本号、固定标识或填充。
|
||
entropy 越高，说明这个字节位置变化越丰富，越像计数器、时间戳或载荷。
|
||
"""
|
||
|
||
counts = Counter(values)
|
||
total = len(values)
|
||
return -sum((count / total) * math.log2(count / total) for count in counts.values())
|
||
|
||
|
||
def load_raw_bytes(path: Path) -> bytes:
|
||
"""读取原始二进制文件。"""
|
||
|
||
if not path.exists():
|
||
raise FileNotFoundError(f"input file not found: {path}")
|
||
return path.read_bytes()
|
||
|
||
|
||
def estimate_frame_size(
|
||
raw: bytes,
|
||
candidate_min: int,
|
||
candidate_max: int,
|
||
sample_frames: int,
|
||
header_columns: int,
|
||
):
|
||
"""在不知道帧长时，用“候选帧长打分”的方式找可能帧长。
|
||
|
||
思路很简单：
|
||
- 如果帧长猜对了，那么每一行的开头会对齐到真实帧头。
|
||
- 真实帧头通常包含版本号、ID、计数器等结构化字段。
|
||
- 这些字段的熵一般比随机载荷低。
|
||
- 所以对每个候选帧长，把数据切成多行，计算前若干列的平均熵。
|
||
- 平均熵越低，越可能是正确帧长。
|
||
|
||
这不是严格证明，只是未知协议分析中常用的启发式方法。
|
||
"""
|
||
|
||
results = []
|
||
for frame_size in range(candidate_min, candidate_max + 1):
|
||
frame_count = min(len(raw) // frame_size, sample_frames)
|
||
if frame_count < 8:
|
||
continue
|
||
|
||
frames = [
|
||
raw[i * frame_size : (i + 1) * frame_size] for i in range(frame_count)
|
||
]
|
||
columns = min(header_columns, frame_size)
|
||
entropies = []
|
||
unique_counts = []
|
||
for offset in range(columns):
|
||
values = [frame[offset] for frame in frames]
|
||
entropies.append(entropy(values))
|
||
unique_counts.append(len(set(values)))
|
||
|
||
results.append(
|
||
{
|
||
"frame_size": frame_size,
|
||
"avg_entropy": sum(entropies) / len(entropies),
|
||
"avg_unique": sum(unique_counts) / len(unique_counts),
|
||
"frame_count": frame_count,
|
||
}
|
||
)
|
||
|
||
return sorted(results, key=lambda item: (item["avg_entropy"], item["avg_unique"]))
|
||
|
||
|
||
def slice_frames(raw: bytes, frame_size: int, limit: int | None = None) -> list[bytes]:
|
||
"""把连续字节流切成固定长度帧。"""
|
||
|
||
total_frames = len(raw) // frame_size
|
||
if limit is not None:
|
||
total_frames = min(total_frames, limit)
|
||
return [raw[i * frame_size : (i + 1) * frame_size] for i in range(total_frames)]
|
||
|
||
|
||
def build_symbol(frames: list[bytes], name: str) -> Symbol:
|
||
"""把 bytes 帧列表包装成 Netzob RawMessage，再放入 Symbol。"""
|
||
|
||
messages = [RawMessage(data=frame) for frame in frames]
|
||
symbol = Symbol(messages=messages, name=name)
|
||
|
||
# HexaString 让 Netzob 打印 Symbol 时用十六进制展示，更适合二进制协议。
|
||
symbol.encodingFunctions.add(TypeEncodingFunction(HexaString))
|
||
return symbol
|
||
|
||
|
||
def byte_statistics(frames: list[bytes]) -> list[dict]:
|
||
"""按字节偏移统计唯一值数量、熵和最常见取值。"""
|
||
|
||
frame_size = len(frames[0])
|
||
stats = []
|
||
for offset in range(frame_size):
|
||
values = [frame[offset] for frame in frames]
|
||
counts = Counter(values)
|
||
stats.append(
|
||
{
|
||
"offset": offset,
|
||
"unique": len(counts),
|
||
"entropy": entropy(values),
|
||
"top": counts.most_common(4),
|
||
}
|
||
)
|
||
return stats
|
||
|
||
|
||
def print_byte_stats(stats: list[dict], first_columns: int = 32) -> None:
|
||
"""打印前若干字节位置的统计表。"""
|
||
|
||
print("offset unique entropy most common byte values")
|
||
print("------ ------ ------- -----------------------")
|
||
for item in stats[:first_columns]:
|
||
top = ", ".join(f"0x{value:02x}:{count}" for value, count in item["top"])
|
||
print(
|
||
f"{item['offset']:>6} {item['unique']:>6} "
|
||
f"{item['entropy']:>7.3f} {top}"
|
||
)
|
||
|
||
|
||
def print_static_dynamic_regions(stats: list[dict]) -> None:
|
||
"""根据 unique 数量粗略标出固定区、低变化区和高变化区。"""
|
||
|
||
labels = []
|
||
for item in stats:
|
||
if item["unique"] == 1:
|
||
labels.append("static")
|
||
elif item["unique"] <= 8:
|
||
labels.append("low-var")
|
||
else:
|
||
labels.append("dynamic")
|
||
|
||
regions = []
|
||
start = 0
|
||
current = labels[0]
|
||
for index, label in enumerate(labels[1:], start=1):
|
||
if label != current:
|
||
regions.append((start, index - 1, current))
|
||
start = index
|
||
current = label
|
||
regions.append((start, len(labels) - 1, current))
|
||
|
||
print("candidate byte regions from simple statistics:")
|
||
for start, end, label in regions[:40]:
|
||
width = end - start + 1
|
||
print(f" bytes {start:03d}-{end:03d} width={width:03d} {label}")
|
||
if len(regions) > 40:
|
||
print(f" ... {len(regions) - 40} more regions omitted")
|
||
|
||
|
||
def demonstrate_split_static(frames: list[bytes]) -> Symbol:
|
||
"""用 Netzob splitStatic 展示自动字段切分。"""
|
||
|
||
symbol = build_symbol(frames, "unknown_tianwen_frames")
|
||
print("Before splitStatic, Netzob sees one raw field:")
|
||
print(f" number of fields: {len(symbol.fields)}")
|
||
|
||
# splitStatic 会比较同一 Symbol 下所有消息的每个位置：
|
||
# - 所有样本都相同的位置会变成 static field。
|
||
# - 样本之间变化的位置会变成 dynamic field。
|
||
Format.splitStatic(
|
||
symbol,
|
||
unitSize=UnitSize.SIZE_8,
|
||
mergeAdjacentStaticFields=True,
|
||
mergeAdjacentDynamicFields=True,
|
||
)
|
||
|
||
print("\nAfter splitStatic(unitSize=8, merge adjacent static/dynamic fields):")
|
||
print(f" number of inferred fields: {len(symbol.fields)}")
|
||
print(
|
||
" teaching note: if most byte positions vary at least once, adjacent "
|
||
"dynamic bytes can merge into one large dynamic field."
|
||
)
|
||
|
||
bytewise_symbol = build_symbol(frames, "unknown_tianwen_frames_bytewise")
|
||
Format.splitStatic(
|
||
bytewise_symbol,
|
||
unitSize=UnitSize.SIZE_8,
|
||
mergeAdjacentStaticFields=False,
|
||
mergeAdjacentDynamicFields=False,
|
||
)
|
||
|
||
print("\nAfter splitStatic(unitSize=8, do not merge adjacent fields):")
|
||
print(f" number of inferred byte-level fields: {len(bytewise_symbol.fields)}")
|
||
print(" first inferred field labels:")
|
||
for index, field in enumerate(bytewise_symbol.fields[:24]):
|
||
print(f" field[{index:02d}] {field}")
|
||
if len(bytewise_symbol.fields) > 24:
|
||
print(f" ... {len(bytewise_symbol.fields) - 24} more fields")
|
||
return bytewise_symbol
|
||
|
||
|
||
def build_bytewise_symbol(frames: list[bytes]) -> Symbol:
|
||
"""把每个字节都切成独立 Field，方便选择某个 offset 做聚类 key。"""
|
||
|
||
symbol = build_symbol(frames, "unknown_tianwen_frames_bytewise")
|
||
Format.splitStatic(
|
||
symbol,
|
||
unitSize=UnitSize.SIZE_8,
|
||
mergeAdjacentStaticFields=False,
|
||
mergeAdjacentDynamicFields=False,
|
||
)
|
||
return symbol
|
||
|
||
|
||
def demonstrate_cluster_by_key_field(
|
||
frames: list[bytes], stats: list[dict], cluster_sample_size: int
|
||
) -> None:
|
||
"""演示如何用某个候选字段作为 key 进行 Netzob 聚类。"""
|
||
|
||
# 在未知协议中，低变化字段常常适合作为聚类 key，比如版本、航天器 ID、
|
||
# 虚拟信道 ID、消息类型等。这里先用统计找出一些候选 offset。
|
||
candidates = [
|
||
item
|
||
for item in stats[:32]
|
||
if 1 < item["unique"] <= 12
|
||
]
|
||
candidates = sorted(candidates, key=lambda item: (item["unique"], item["entropy"]))
|
||
|
||
if not candidates:
|
||
print("No low-variation key candidates found in the first 32 bytes.")
|
||
return
|
||
|
||
print("candidate key byte offsets from the first 32 bytes:")
|
||
for item in candidates[:8]:
|
||
top = ", ".join(f"0x{value:02x}:{count}" for value, count in item["top"])
|
||
print(
|
||
f" offset {item['offset']:02d}: unique={item['unique']}, "
|
||
f"entropy={item['entropy']:.3f}, top=[{top}]"
|
||
)
|
||
|
||
# 选择最靠前且变化种类较少的字段作为演示 key。
|
||
key_offset = candidates[0]["offset"]
|
||
cluster_frames = frames[:cluster_sample_size]
|
||
print(
|
||
f" using {len(cluster_frames)} frames for this cluster demo "
|
||
f"(kept small because Netzob clustering can be expensive)"
|
||
)
|
||
|
||
bytewise_symbol = build_bytewise_symbol(cluster_frames)
|
||
|
||
print(f"\nNetzob clusterByKeyField demo on byte offset {key_offset}:")
|
||
print(f" bytewise fields available: {len(bytewise_symbol.fields)}")
|
||
|
||
# clusterByKeyField 会把拥有相同 key 字段值的消息放进同一个 Symbol。
|
||
clusters = Format.clusterByKeyField(bytewise_symbol, bytewise_symbol.fields[key_offset])
|
||
|
||
print(f" clusters created: {len(clusters)}")
|
||
for key, cluster_symbol in list(clusters.items())[:12]:
|
||
key_hex = key.hex() if isinstance(key, (bytes, bytearray)) else str(key)
|
||
print(
|
||
f" key=0x{key_hex:<4} messages={len(cluster_symbol.messages):>4} "
|
||
f"fields={len(cluster_symbol.fields)}"
|
||
)
|
||
|
||
|
||
def demonstrate_manual_candidate_model(frame_size: int) -> None:
|
||
"""用 Netzob Field/Raw 手工搭建一个候选格式模型。
|
||
|
||
这一步不是声称字段含义已经确定，而是演示逆向分析常见工作流：
|
||
先用统计和 splitStatic 找到疑似字段边界，再用 Netzob 明确描述一个候选模型。
|
||
"""
|
||
|
||
section("Step 7 - Manual candidate model with Netzob Field/Raw")
|
||
|
||
# 这里故意使用“candidate_”前缀，表示这些字段只是初步假设。
|
||
# 对未知空间帧，通常可以先把开头若干字节当成候选头部，
|
||
# 中间大段当成候选数据区，末尾若干字节当成候选尾部/校验/填充。
|
||
candidate_header = Field(Raw(nbBytes=6), name="candidate_header_0_5")
|
||
candidate_insert_or_secondary = Field(
|
||
Raw(nbBytes=8), name="candidate_insert_or_secondary_6_13"
|
||
)
|
||
candidate_payload = Field(
|
||
Raw(nbBytes=max(frame_size - 18, 0)), name="candidate_payload"
|
||
)
|
||
candidate_tail = Field(Raw(nbBytes=4), name="candidate_tail_4_bytes")
|
||
|
||
symbol = Symbol(
|
||
name="manual_candidate_tianwen_like_frame",
|
||
fields=[
|
||
candidate_header,
|
||
candidate_insert_or_secondary,
|
||
candidate_payload,
|
||
candidate_tail,
|
||
],
|
||
)
|
||
|
||
print("This is a teaching model, not a confirmed Tianwen-1 specification:")
|
||
print(symbol.str_structure())
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
"""命令行参数。"""
|
||
|
||
parser = argparse.ArgumentParser(
|
||
description=(
|
||
"Use Netzob to teach unknown binary frame analysis on Tianwen-1 raw data."
|
||
)
|
||
)
|
||
parser.add_argument(
|
||
"--input",
|
||
type=Path,
|
||
default=DEFAULT_INPUT,
|
||
help=f"raw binary input file, default: {DEFAULT_INPUT}",
|
||
)
|
||
parser.add_argument(
|
||
"--frame-size",
|
||
type=int,
|
||
default=None,
|
||
help=(
|
||
"known or chosen frame size. If omitted, the script estimates it "
|
||
"from candidate sizes."
|
||
),
|
||
)
|
||
parser.add_argument(
|
||
"--candidate-min",
|
||
type=int,
|
||
default=180,
|
||
help="minimum frame-size candidate used when estimating frame length",
|
||
)
|
||
parser.add_argument(
|
||
"--candidate-max",
|
||
type=int,
|
||
default=260,
|
||
help="maximum frame-size candidate used when estimating frame length",
|
||
)
|
||
parser.add_argument(
|
||
"--sample-size",
|
||
type=int,
|
||
default=96,
|
||
help="number of frames used for Netzob analysis",
|
||
)
|
||
parser.add_argument(
|
||
"--show-samples",
|
||
type=int,
|
||
default=4,
|
||
help="number of raw sample frames to print as short hex",
|
||
)
|
||
parser.add_argument(
|
||
"--cluster-sample-size",
|
||
type=int,
|
||
default=8,
|
||
help="number of frames used only for the Netzob clusterByKeyField demo",
|
||
)
|
||
return parser.parse_args()
|
||
|
||
|
||
def main() -> None:
|
||
"""Run the teaching analysis."""
|
||
|
||
args = parse_args()
|
||
|
||
section("Step 1 - Read unknown binary data")
|
||
raw = load_raw_bytes(args.input)
|
||
print(f"input file: {args.input}")
|
||
print(f"total bytes: {len(raw):,}")
|
||
|
||
section("Step 2 - Estimate or choose a fixed frame size")
|
||
if args.frame_size is None:
|
||
ranked = estimate_frame_size(
|
||
raw,
|
||
candidate_min=args.candidate_min,
|
||
candidate_max=args.candidate_max,
|
||
sample_frames=args.sample_size,
|
||
header_columns=16,
|
||
)
|
||
print("Top frame-size candidates by low average header entropy:")
|
||
for item in ranked[:10]:
|
||
print(
|
||
f" size={item['frame_size']:>3} "
|
||
f"avg_entropy={item['avg_entropy']:.3f} "
|
||
f"avg_unique={item['avg_unique']:.2f} "
|
||
f"frames_tested={item['frame_count']}"
|
||
)
|
||
frame_size = ranked[0]["frame_size"]
|
||
print(f"\nChosen frame size for the rest of this teaching run: {frame_size}")
|
||
else:
|
||
frame_size = args.frame_size
|
||
print(f"Using user-provided frame size: {frame_size}")
|
||
|
||
all_frame_count = len(raw) // frame_size
|
||
frames = slice_frames(raw, frame_size, limit=args.sample_size)
|
||
print(f"complete frames in file with this size: {all_frame_count:,}")
|
||
print(f"frames sampled for Netzob: {len(frames):,}")
|
||
|
||
print("\nFirst sample frames as short hex:")
|
||
for index, frame in enumerate(frames[: args.show_samples]):
|
||
print(f" frame[{index:03d}] {short_hex(frame)}")
|
||
|
||
section("Step 3 - Wrap samples as Netzob RawMessage and Symbol")
|
||
teaching_symbol = build_symbol(frames, "unknown_tianwen_frames")
|
||
print("Netzob objects created:")
|
||
print(f" RawMessage count: {len(teaching_symbol.messages)}")
|
||
print(f" Symbol name: {teaching_symbol.name}")
|
||
print(f" Initial field count: {len(teaching_symbol.fields)}")
|
||
print(
|
||
"Teaching point: at the beginning Netzob only knows each frame is raw bytes; "
|
||
"it does not know the protocol fields."
|
||
)
|
||
|
||
section("Step 4 - Byte-position statistics before protocol knowledge")
|
||
stats = byte_statistics(frames)
|
||
print_byte_stats(stats, first_columns=40)
|
||
print()
|
||
print_static_dynamic_regions(stats)
|
||
|
||
section("Step 5 - Netzob Format.splitStatic field inference")
|
||
demonstrate_split_static(frames)
|
||
|
||
section("Step 6 - Netzob clusterByKeyField on candidate key bytes")
|
||
demonstrate_cluster_by_key_field(frames, stats, args.cluster_sample_size)
|
||
|
||
demonstrate_manual_candidate_model(frame_size)
|
||
|
||
section("Done - What to try next")
|
||
print(
|
||
"1. Increase --sample-size to see whether inferred fields remain stable.\n"
|
||
"2. Try --frame-size with another candidate and compare splitStatic results.\n"
|
||
"3. Choose another low-variation offset as cluster key and inspect clusters.\n"
|
||
"4. After a candidate field map is stable, then compare it with known CCSDS/Tianwen parsing.\n"
|
||
"5. Treat this as protocol-discovery scaffolding, not as a final specification."
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|