使用socat模拟噪声串行线路的实现问题求助

一、虚拟串口间添加数据干扰 #

1. 前置准备 #

• 确保已安装socat（1.7.4版本）和Python3
• 创建两个命名管道（FIFO）实现双向数据转发：

  mkfifo /tmp/fifo_tx /tmp/fifo_rx
  

2. 双向干扰脚本修改 #

原脚本仅支持单向处理，串口是全双工通信，需改为双向转发版本：

import sys
import random
import threading

def disturb(data):
    """ 对数据进行干扰：10%概率翻转1位，5%概率丢弃字节 """
    result = bytearray()
    for byte in data:
        # 随机翻转一个比特位（10%概率）
        if random.random() < 0.1:
            byte ^= 1 << random.randint(0, 7)
        # 保留字节（95%概率，5%概率丢弃）
        if random.random() > 0.05:
            result.append(byte)
    return bytes(result)

def forward(src, dst, direction):
    """ 从源读取数据，干扰后写入目标，打印调试信息 """
    while True:
        data = src.read(128)
        if not data:
            break
        print(f"[{direction}] 原始数据: {data}")
        disturbed_data = disturb(data)
        print(f"[{direction}] 干扰后数据: {disturbed_data}")
        dst.write(disturbed_data)
        dst.flush()

def main():
    # 启动双线程处理双向数据流
    thread_v1_to_v2 = threading.Thread(target=forward, args=(sys.stdin.buffer, sys.stdout.buffer, "V1→V2"))
    thread_v2_to_v1 = threading.Thread(target=forward, args=(sys.stdout.buffer, sys.stdin.buffer, "V2→V1"))
    thread_v1_to_v2.start()
    thread_v2_to_v1.start()
    thread_v1_to_v2.join()
    thread_v2_to_v1.join()

if __name__ == '__main__':
    main()

3. 按顺序启动组件 #

1. 绑定虚拟串口ttyV1与FIFO：

   socat -d -d pty,raw,echo=0,link=/tmp/ttyV1 <(cat /tmp/fifo_rx) >/tmp/fifo_tx
   

   作用：将ttyV1的输出写入fifo_tx，从fifo_rx读取数据写入ttyV1。

2. 绑定虚拟串口ttyV2与FIFO（反向）：

   socat -d -d pty,raw,echo=0,link=/tmp/ttyV2 <(cat /tmp/fifo_tx) >/tmp/fifo_rx
   

   作用：将ttyV2的输出写入fifo_rx，从fifo_tx读取数据写入ttyV2。

3. 启动干扰脚本：

   python3 disturb.py < /tmp/fifo_tx > /tmp/fifo_rx
   

   此时ttyV1与ttyV2间的所有数据都会经过干扰处理后传输。

二、真实串口连接虚拟串口并添加加解密 #

1. 核心架构 #

真实串口（如/dev/ttyUSB0）→ socat → 加解密脚本 → socat → 虚拟串口，双向数据均经过加解密。

2. 加解密脚本示例（异或对称加密） #

可替换为实际业务需要的加解密算法：

import sys
import threading

# 自定义对称密钥
ENCRYPT_KEY = 0xAB

def xor_crypt(data):
    """ 异或加解密（加密解密使用同一密钥） """
    return bytes(byte ^ ENCRYPT_KEY for byte in data)

def forward(src, dst):
    """ 读取数据、加解密后转发 """
    while True:
        data = src.read(128)
        if not data:
            break
            processed_data = xor_crypt(data)
        dst.write(processed_data)
        dst.flush()

def main():
    # 双线程处理双向加解密转发
    thread_real_to_virtual = threading.Thread(target=forward, args=(sys.stdin.buffer, sys.stdout.buffer))
    thread_virtual_to_real = threading.Thread(target=forward, args=(sys.stdout.buffer, sys.stdin.buffer))
    thread_real_to_virtual.start()
    thread_virtual_to_real.start()
    thread_real_to_virtual.join()
    thread_virtual_to_real.join()

if __name__ == '__main__':
    main()

3. 启动命令 #

1. 创建虚拟串口对：

   socat -d -d pty,raw,echo=0,link=/tmp/ttyV1 pty,raw,echo=0,link=/tmp/ttyV2
   

2. 连接真实串口与虚拟串口（带加解密）：
   直接用socat串联真实串口、加解密脚本与虚拟串口：

   socat -d -d /dev/ttyUSB0,raw,echo=0 EXEC:"python3 crypt.py" pty,raw,echo=0,link=/tmp/ttyV1
   

   或用FIFO实现更清晰的双向转发（参考干扰脚本的FIFO配置逻辑，替换干扰脚本为加解密脚本即可）。

内容的提问来源于stack exchange，提问作者mastupristi