wandering1/STM32F4-Base
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

freeRTOS支持

Browse Source
This commit is contained in:
wandering1
2026-07-19 13:09:06 +08:00
parent 4eb728583b
commit 9ed4c22022
63 changed files with 45898 additions and 507 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
test/__pycache__/ch395f_socket_test.cpython-311.pyc Normal file
View File

Binary file not shown.

204
test/ch395f_socket_test.py
View File

@@ -608,6 +608,199 @@ def stress_test(args):
log_error("压力测试有失败!")
# ==================== DHCP 测试 ====================
DHCP_SERVER_PORT = 67
DHCP_CLIENT_PORT = 68
DHCP_DISCOVER = 1
DHCP_OFFER = 2
DHCP_REQUEST = 3
DHCP_ACK = 5
OPT_MESSAGE_TYPE = 53
OPT_SERVER_ID = 54
OPT_SUBNET_MASK = 1
OPT_ROUTER = 3
OPT_DNS = 6
OPT_LEASE_TIME = 51
OPT_END = 255
def _ip_to_bytes(ip):
return bytes(map(int, ip.split('.')))
def _build_dhcp_pkt(op, xid, client_mac, yiaddr, siaddr, options):
op_b = bytes([op])
htype = b'\x01'
hlen = b'\x06'
hops = b'\x00'
secs = b'\x00\x00'
flags = b'\x00\x00'
ciaddr = b'\x00\x00\x00\x00'
giaddr = b'\x00\x00\x00\x00'
chaddr = client_mac + b'\x00' * 10
sname = b'\x00' * 64
file = b'\x00' * 128
magic = b'\x63\x82\x53\x63'
return op_b + htype + hlen + hops + xid + secs + flags + ciaddr + yiaddr + siaddr + giaddr + chaddr + sname + file + magic + options
def dhcp_server_test(args):
"""DHCP Server 模式（DISCOVER → OFFER → REQUEST → ACK）"""
log_info("=" * 60)
log_info("DHCP Server 测试")
log_info("=" * 60)
server_ip = args.dhcp_iface
offer_ip = args.ip
subnet_mask = args.dhcp_mask
router = args.dhcp_gw
lease_time = 3600
log_info(f"DHCP 服务器 IP: {server_ip}")
log_info(f"待分配 IP: {offer_ip}")
log_info(f"子网掩码: {subnet_mask}")
log_info(f"网关: {router}")
print()
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
sock.bind(('0.0.0.0', DHCP_SERVER_PORT))
sock.settimeout(60)
log_info("DHCP 服务器已启动，等待 CH395F DISCOVER...")
log_info("（请确保 MCU 端已启用 ENABLE_PHASE6_TESTS 并复位）")
print()
try:
# Phase 1: DISCOVER → OFFER
data, addr = sock.recvfrom(4096)
if len(data) <= 250:
log_error("收到无效 DHCP 包")
return
msg_type = data[242]
if msg_type != DHCP_DISCOVER:
log_error(f"预期 DISCOVER(1)，收到 {msg_type}")
return
xid = data[4:8]
client_mac = data[28:34]
mac_str = ':'.join(f'{b:02X}' for b in client_mac)
log_success(f"收到 DHCP DISCOVER from {mac_str} (xid={xid.hex()})")
# Build OFFER options
options = b''
options += struct.pack('BBB', OPT_MESSAGE_TYPE, 1, DHCP_OFFER)
options += struct.pack('BB', OPT_SUBNET_MASK, 4) + _ip_to_bytes(subnet_mask)
options += struct.pack('BB', OPT_ROUTER, 4) + _ip_to_bytes(router)
options += struct.pack('BB', OPT_DNS, 4) + _ip_to_bytes(router)
options += struct.pack('BB', OPT_SERVER_ID, 4) + _ip_to_bytes(server_ip)
options += bytes([OPT_LEASE_TIME, 4]) + struct.pack('!I', lease_time)
options += bytes([OPT_END])
offer = _build_dhcp_pkt(2, xid, client_mac,
_ip_to_bytes(offer_ip),
_ip_to_bytes(server_ip),
options)
# Try both global broadcast and subnet broadcast
for bcast in ['255.255.255.255', '192.168.1.255']:
sock.sendto(offer, (bcast, DHCP_CLIENT_PORT))
log_success("发送 DHCP OFFER")
# Phase 2: REQUEST → ACK（最多重试 8 次）
sock.settimeout(5)
phase2_ok = False
for retry in range(8):
try:
data, addr = sock.recvfrom(4096)
if len(data) <= 250:
continue
msg_type = data[242]
if msg_type == DHCP_REQUEST:
rx_xid = data[4:8]
rx_mac = data[28:34]
log_success(f"收到 DHCP REQUEST from {':'.join(f'{b:02X}' for b in rx_mac)}")
ack_options = b''
ack_options += struct.pack('BBB', OPT_MESSAGE_TYPE, 1, DHCP_ACK)
ack_options += struct.pack('BB', OPT_SUBNET_MASK, 4) + _ip_to_bytes(subnet_mask)
ack_options += struct.pack('BB', OPT_ROUTER, 4) + _ip_to_bytes(router)
ack_options += struct.pack('BB', OPT_DNS, 4) + _ip_to_bytes(router)
ack_options += struct.pack('BB', OPT_SERVER_ID, 4) + _ip_to_bytes(server_ip)
ack_options += bytes([OPT_LEASE_TIME, 4]) + struct.pack('!I', lease_time)
ack_options += bytes([OPT_END])
ack = _build_dhcp_pkt(2, rx_xid, rx_mac,
_ip_to_bytes(offer_ip),
_ip_to_bytes(server_ip),
ack_options)
for bcast in ['255.255.255.255', '192.168.1.255']:
sock.sendto(ack, (bcast, DHCP_CLIENT_PORT))
log_success(f"发送 DHCP ACK → {offer_ip}")
log_success("DHCP 分配完成！")
phase2_ok = True
break
elif msg_type == DHCP_DISCOVER:
# 重传 DISCOVER，重发 OFFER
xid = data[4:8]
client_mac = data[28:34]
options = b''
options += struct.pack('BBB', OPT_MESSAGE_TYPE, 1, DHCP_OFFER)
options += struct.pack('BB', OPT_SUBNET_MASK, 4) + _ip_to_bytes(subnet_mask)
options += struct.pack('BB', OPT_ROUTER, 4) + _ip_to_bytes(router)
options += struct.pack('BB', OPT_DNS, 4) + _ip_to_bytes(router)
options += struct.pack('BB', OPT_SERVER_ID, 4) + _ip_to_bytes(server_ip)
options += bytes([OPT_LEASE_TIME, 4]) + struct.pack('!I', lease_time)
options += bytes([OPT_END])
offer = _build_dhcp_pkt(2, xid, client_mac,
_ip_to_bytes(offer_ip),
_ip_to_bytes(server_ip),
options)
for bcast in ['255.255.255.255', '192.168.1.255']:
sock.sendto(offer, (bcast, DHCP_CLIENT_PORT))
log_info(f"重发 DHCP OFFER (retry={retry+1})")
except socket.timeout:
log_warn(f"Phase 2 等待超时 (retry={retry+1})")
if not phase2_ok:
log_warn("DHCP REQUEST 未收到")
print()
# Phase 3: 连通性验证（MCU Phase 6 内置 HELLO 端口 = 60000）
hello_port = args.port if args.port != 8080 else 60000
log_info(f"验证 {offer_ip} 连通性（UDP HELLO → port {hello_port}）")
sock.close()
time.sleep(3)
verify = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
verify.settimeout(2)
phase3_ok = False
for retry in range(5):
try:
verify.sendto(b"HELLO", (offer_ip, hello_port))
data, addr = verify.recvfrom(64)
if data == b"HELLO":
log_success(f"连通性验证通过：{addr[0]}:{addr[1]} 响应正常")
phase3_ok = True
break
time.sleep(1)
except (socket.timeout, ConnectionResetError, OSError):
log_info(f"等待 UDP 响应... (retry={retry+1})")
time.sleep(1)
if not phase3_ok:
log_warn("连通性验证超时（MCU 的 UDP 服务可能未运行或端口不正确）")
log_info("提示：CH395F UDP echo 默认端口为 60000，请使用 --port 60000")
except socket.timeout:
log_error("等待 DHCP DISCOVER 超时（60s）")
finally:
sock.close()
# ==================== 数据包大小测试 ====================
def packet_size_test(args):
@@ -668,17 +861,22 @@ def main():
%(prog)s tcp_client --ip 192.168.1.100 --port 8080
%(prog)s tcp_server --port 8080
%(prog)s stress --ip 192.168.1.100 --port 8080 --count 5
%(prog)s dhcp_server --dhcp-iface 192.168.1.2 --ip 192.168.1.100
"""
)
parser.add_argument('mode', choices=['tcp_server', 'tcp_client', 'udp_server', 'udp_client',
'tcp_echo', 'udp_echo', 'stress', 'packet_size'],
'tcp_echo', 'udp_echo', 'stress', 'packet_size',
'dhcp_server'],
help='测试模式')
parser.add_argument('--ip', default='192.168.1.100', help='目标 IP 地址')
parser.add_argument('--ip', default='192.168.1.100', help='目标 IP 地址 / DHCP 待分配 IP')
parser.add_argument('--port', type=int, default=8080, help='端口号')
parser.add_argument('--count', type=int, default=10, help='测试次数')
parser.add_argument('--size', type=int, default=1024, help='数据包大小')
parser.add_argument('--timeout', type=int, default=180, help='接受连接超时（秒，默认180）')
parser.add_argument('--dhcp-iface', default='192.168.1.2', help='DHCP 服务器网卡 IP')
parser.add_argument('--dhcp-mask', default='255.255.255.0', help='DHCP 分配的子网掩码')
parser.add_argument('--dhcp-gw', default='192.168.1.1', help='DHCP 分配的网关')
args = parser.parse_args()
@@ -705,6 +903,8 @@ def main():
stress_test(args)
elif args.mode == 'packet_size':
packet_size_test(args)
elif args.mode == 'dhcp_server':
dhcp_server_test(args)
except KeyboardInterrupt:
log_warn("用户中断")
except Exception as e:

156
test/dhcp_server.py
View File

@@ -1,156 +0,0 @@
#!/usr/bin/env python3
"""
轻量 DHCP Server — 用于 CH395F Phase 6 测试
用法：以管理员身份运行
python dhcp_server.py --iface 192.168.1.2
"""
import socket
import struct
import argparse
import time
DHCP_SERVER_PORT = 67
DHCP_CLIENT_PORT = 68
# DHCP 消息类型
DHCP_DISCOVER = 1
DHCP_OFFER = 2
DHCP_REQUEST = 3
DHCP_ACK = 5
# DHCP 选项
OPT_MESSAGE_TYPE = 53
OPT_SERVER_ID = 54
OPT_REQUESTED_IP = 50
OPT_SUBNET_MASK = 1
OPT_ROUTER = 3
OPT_DNS = 6
OPT_LEASE_TIME = 51
OPT_END = 255
SERVER_IP = "192.168.1.2"
OFFER_IP = "192.168.1.100"
SUBNET_MASK = "255.255.255.0"
ROUTER = "192.168.1.1"
LEASE_TIME = 3600 # 1 小时
def ip_to_bytes(ip):
return bytes(map(int, ip.split('.')))
def build_dhcp_offer(discover_pkt, server_ip, offer_ip):
"""构建 DHCP OFFER 包"""
transaction_id = discover_pkt[232:236] # Bootp header 前 4 字节
client_mac = discover_pkt[28:34]
# BOOTP 头
op = b'\x02' # Boot reply
htype = b'\x01' # Ethernet
hlen = b'\x06' # MAC len
hops = b'\x00'
xid = transaction_id
secs = b'\x00\x00'
flags = b'\x00\x00'
ciaddr = b'\x00\x00\x00\x00'
yiaddr = ip_to_bytes(offer_ip) # Your IP
siaddr = ip_to_bytes(server_ip) # Server IP
giaddr = b'\x00\x00\x00\x00'
chaddr = client_mac + b'\x00' * 10 # Client MAC (padded to 16)
sname = b'\x00' * 64
file = b'\x00' * 128
magic = b'\x63\x82\x53\x63' # DHCP magic cookie
# DHCP 选项
options = b''
options += struct.pack('BBB', OPT_MESSAGE_TYPE, 1, DHCP_OFFER)
options += struct.pack('BB', OPT_SERVER_ID, 4) + ip_to_bytes(server_ip)
options += struct.pack('BB', OPT_REQUESTED_IP, 4) + ip_to_bytes(offer_ip)
options += struct.pack('BB', OPT_SUBNET_MASK, 4) + ip_to_bytes(SUBNET_MASK)
options += struct.pack('BB', OPT_ROUTER, 4) + ip_to_bytes(ROUTER)
options += struct.pack('BB', OPT_DNS, 4) + ip_to_bytes(ROUTER)
options += bytes([OPT_LEASE_TIME, 4]) + struct.pack('!I', LEASE_TIME)
options += bytes([OPT_END])
pkt = op + htype + hlen + hops + xid + secs + flags + ciaddr + yiaddr + siaddr + giaddr + chaddr + sname + file + magic + options
return pkt
def build_dhcp_ack(request_pkt, server_ip, offer_ip):
"""构建 DHCP ACK 包"""
transaction_id = request_pkt[232:236]
client_mac = request_pkt[28:34]
op = b'\x02'
htype = b'\x01'
hlen = b'\x06'
hops = b'\x00'
xid = transaction_id
secs = b'\x00\x00'
flags = b'\x00\x00'
ciaddr = b'\x00\x00\x00\x00'
yiaddr = ip_to_bytes(offer_ip)
siaddr = ip_to_bytes(server_ip)
giaddr = b'\x00\x00\x00\x00'
chaddr = client_mac + b'\x00' * 10
sname = b'\x00' * 64
file = b'\x00' * 128
magic = b'\x63\x82\x53\x63'
options = b''
options += struct.pack('BBB', OPT_MESSAGE_TYPE, 1, DHCP_ACK)
options += struct.pack('BB', OPT_SERVER_ID, 4) + ip_to_bytes(server_ip)
options += struct.pack('BB', OPT_SUBNET_MASK, 4) + ip_to_bytes(SUBNET_MASK)
options += struct.pack('BB', OPT_ROUTER, 4) + ip_to_bytes(ROUTER)
options += struct.pack('BB', OPT_DNS, 4) + ip_to_bytes(ROUTER)
options += bytes([OPT_LEASE_TIME, 4]) + struct.pack('!I', LEASE_TIME)
options += bytes([OPT_END])
return op + htype + hlen + hops + xid + secs + flags + ciaddr + yiaddr + siaddr + giaddr + chaddr + sname + file + magic + options
def main():
parser = argparse.ArgumentParser(description="轻量 DHCP Server for CH395F 测试")
parser.add_argument("--iface", default=SERVER_IP, help="网卡 IP 地址")
args = parser.parse_args()
offer_ip = OFFER_IP
server_ip = args.iface
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
sock.bind(('0.0.0.0', DHCP_SERVER_PORT))
sock.settimeout(60)
print(f"DHCP Server 启动, 网卡 IP={server_ip}, 分配 IP={offer_ip}")
print("等待 CH395F DHCP DISCOVER...")
try:
# 第一阶段：DISCOVER → OFFER
data, addr = sock.recvfrom(4096)
msg_type = data[232 + 4 + 1] if len(data) > 250 else 0
client_mac = ':'.join(f'{b:02X}' for b in data[28:34])
print(f"收到 DHCP DISCOVER from {client_mac}")
offer = build_dhcp_offer(data, server_ip, offer_ip)
sock.sendto(offer, ('255.255.255.255', DHCP_CLIENT_PORT))
print("发送 DHCP OFFER")
# 第二阶段：REQUEST → ACK
sock.settimeout(10)
try:
data, addr = sock.recvfrom(4096)
client_mac = ':'.join(f'{b:02X}' for b in data[28:34])
print(f"收到 DHCP REQUEST from {client_mac}")
ack = build_dhcp_ack(data, server_ip, offer_ip)
sock.sendto(ack, ('255.255.255.255', DHCP_CLIENT_PORT))
print(f"发送 DHCP ACK → {offer_ip}")
print("DHCP 分配完成！")
except socket.timeout:
print("等待 DHCP REQUEST 超时")
except socket.timeout:
print("等待 DHCP DISCOVER 超时（60s）")
finally:
sock.close()
if __name__ == '__main__':
main()