Dart实现MikroTik路由器连接类问题：无法收发命令

问题现象 #

• Socket连接可成功建立，但登录流程无法完成，无法发送命令或接收路由器响应
• 登录阶段无法获取路由器返回的确认信息

代码实现 #

import 'dart:async';
import 'dart:io';
import 'dart:convert';
import 'dart:typed_data';
import 'package:crypto/crypto.dart';

/// This class manages a connection to a MikroTik router, allowing commands to be sent and responses to be read.
class RouterConnection {
  final String ip; // Router IP address
  final int port; // Router port (default is 8728 for API)
  Socket? socket; // Socket for the connection
  bool connected = false; // Indicates if the connection is successful
  final int timeout; // Timeout for connection attempts
  final int attempts; // Number of connection attempts
  final int delay; // Delay between attempts

  RouterConnection({
    required this.ip,
    required this.port,
    this.timeout = 5,
    this.attempts = 3,
    this.delay = 2,
  });

  /// Connects to the router and logs in with the provided credentials. It retries based on `attempts`.
  Future<bool> connect(String login, String password) async {
    for (int attempt = 1; attempt <= attempts; attempt++) {
      print('Connection attempt #$attempt to $ip:$port...');
      try {
        socket = await Socket.connect(ip, port, timeout: Duration(seconds: timeout));
        print('Connected to $ip:$port');

        // Send login command
        await write('/login');
        await write('=name=$login', false);

        // Read response
        List<String> response = await read();

        // Check the response
        if (response.isNotEmpty) {
          if (response[0] == '!trap') {
            print('Login failed: ${response[1]}');
            return false;
          } else if (response[0].startsWith('=ret=')) {
            String challenge = response[0].substring(5);
            String responseHash = _generateResponseHash(password, challenge);
            await write('=response=$responseHash');

            // Read the second response
            List<String> secondResponse = await read();
            if (secondResponse.isNotEmpty && secondResponse[0] == '!done') {
              connected = true;
              print('Login successful');
              break;
            } else {
              print('Login failed: ${secondResponse[0]}');
            }
          }
        }

        // Close the connection if login fails
        socket?.destroy();
        await Future.delayed(Duration(seconds: delay));

      } catch (e) {
        print('Connection failed: $e');
      }
    }

    return connected;
  }

  // Sends data to the router
  Future<void> write(String word, [bool terminate = true]) async {
    if (socket == null) {
      print('Socket is not initialized');
      return;
    }

    Uint8List encodedWord = _encodeWord(word);
    socket!.add(encodedWord);
    if (terminate) {
      socket!.add(Uint8List.fromList([0])); // Send zero word to indicate end of command
    }
    await socket!.flush();
  }

  // Reads the response from the router
  Future<List<String>> read() async {
    if (socket == null) {
      print('Socket is not initialized');
      return [];
    }

    List<String> response = [];
    Completer<List<String>> completer = Completer();
    socket!.listen((Uint8List data) {
      String decodedResponse = utf8.decode(data);
      List<String> words = decodedResponse.split('\n');
      response.addAll(words);
      if (words.contains('!done') || words.contains('!trap')) {
        completer.complete(response);
      }
    });
    return completer.future;
  }

  // Generates the MD5 hash for login
  String _generateResponseHash(String password, String challenge) {
    final challengeBytes = Uint8List.fromList(utf8.encode(challenge));
    final passwordBytes = Uint8List.fromList(utf8.encode(password));
    final combined = Uint8List(1 + passwordBytes.length + challengeBytes.length);
    combined[0] = 0;
    combined.setRange(1, passwordBytes.length + 1, passwordBytes);
    combined.setRange(passwordBytes.length + 1, combined.length, challengeBytes);

    final hash = md5.convert(combined).toString();
    return '00$hash';
  }

  // Encodes words according to the MikroTik protocol
  Uint8List _encodeWord(String word) {
    final wordBytes = utf8.encode(word);
    final int len = wordBytes.length;
    late Uint8List encodedLen;

    if (len <= 0x7F) {
      encodedLen = Uint8List(1)..[0] = len;
    } else if (len <= 0x3FFF) {
      encodedLen = Uint8List(2)
        ..[0] = (len >> 8) | 0x80
        ..[1] = len & 0xFF;
    } else if (len <= 0x1FFFFF) {
      encodedLen = Uint8List(3)
        ..[0] = (len >> 16) | 0xC0
        ..[1] = (len >> 8) & 0xFF
        ..[2] = len & 0xFF;
    } else {
      encodedLen = Uint8List(4)
        ..[0] = (len >> 24) | 0xE0
        ..[1] = (len >> 16) & 0xFF
        ..[2] = (len >> 8) & 0xFF
        ..[3] = len & 0xFF;
    }

    return Uint8List.fromList(encodedLen + wordBytes);
  }
}

void main() async {
  // Setup connection details
  String routerIp = '91e209389285.sn.mynetname.net'; // Router IP address
  int routerPort = 8728; // Router port (typically 8728 for API)
  String username = 'a'; // Username
  String password = '1234'; // Password

  // Create a connection to the router
  RouterConnection connection = RouterConnection(ip: routerIp, port: routerPort);

  // Attempt to connect and log in
  bool isConnected = await connection.connect(username, password);

  // Display the result
  if (isConnected) {
    print('Successfully logged in to the router at $routerIp:$routerPort');
  } else {
    print('Failed to log in to the router at $routerIp:$routerPort');
  }
}

具体疑问 #

1. 连接建立后收发数据失败的原因是什么？
2. Dart处理Socket连接有哪些常见错误需要注意？
3. 如何确保发送的命令符合MikroTik API的格式要求？
4. 登录响应处理是否正确实现，若失败该如何排查？

问题分析与解决方案 #

1. 收发数据失败的核心原因 #

你的read()方法存在两个致命问题：

• 重复注册监听器：每次调用read()都会新增Socket监听器，导致同一响应被多次处理，Completer触发逻辑混乱
• 错误的响应拆分方式：MikroTik API是长度前缀的二进制协议，不是文本换行分隔，用\n拆分完全不符合协议规范，无法正确解析响应

2. Dart Socket常见坑点 #

• 重复监听：多次调用socket.listen()会叠加监听器，数据被重复处理，必须确保一个Socket只注册一次监听器
• 分块数据处理：Socket数据可能分多次到达，必须维护全局缓冲区，积累足够数据后再解析
• 编码与二进制边界：MikroTik API用UTF-8编码，但必须严格按二进制长度前缀解析，不能直接转字符串后拆分
• 异步顺序：socket.add()和flush()是异步操作，必须保证命令发送顺序，避免乱序
• 资源清理：登录失败销毁Socket后，要避免后续操作调用已销毁的Socket

3. 符合MikroTik API格式的关键要求 #

MikroTik API命令由多个**字（word）**组成，每个字格式：

1. 长度前缀：1-4字节，最高位标识长度字节数（0x7F=1字节，0x80=2字节，0xC0=3字节，0xE0=4字节）
2. 字内容：UTF-8编码字符串
3. 命令结束：发送长度为0的字（单字节0x00）

你的_encodeWord()方法是正确的，但登录命令发送逻辑错误：

• 原代码中await write('/login')会自动发送结束符，导致路由器只收到/login，未收到用户名，无法返回正确的challenge
• 修复后的登录命令发送逻辑：

// 正确发送登录初始命令：/login + =name=login，最后发送结束符
await write('/login', false); // 不发送结束符
await write('=name=$login', false); // 不发送结束符
await write(''); // 发送空字符串触发结束符0x00

4. 登录响应处理的修复与排查 #

修复后的核心逻辑

首先添加全局缓冲区和二进制解析方法：

final Uint8List _buffer = Uint8List(0);

// 从缓冲区解析一个完整的字，数据不足返回null
String? _parseWord() {
  if (_buffer.length < 1) return null;

  int lenBytes = 1;
  int lenFlag = _buffer[0];
  int actualLen = lenFlag & 0x7F;
  if (lenFlag >= 0x80) {
    lenBytes++;
    actualLen = (actualLen << 8) | _buffer[1];
    if (lenFlag >= 0xC0) {
      lenBytes++;
      actualLen = (actualLen << 8) | _buffer[2];
      if (lenFlag >= 0xE0) {
        lenBytes++;
        actualLen = (actualLen << 8) | _buffer[3];
      }
    }
  }

  if (_buffer.length < lenBytes + actualLen) return null;

  String word = utf8.decode(_buffer.sublist(lenBytes, lenBytes + actualLen));
  // 移除已解析的部分
  Uint8List newBuffer = Uint8List(_buffer.length - (lenBytes + actualLen));
  newBuffer.setRange(0, newBuffer.length, _buffer.sublist(lenBytes + actualLen));
  _buffer.setRange(0, newBuffer.length, newBuffer);
  return word;
}

重写read()方法：

Future<List<String>> read() async {
  if (socket == null) return [];

  List<String> response = [];
  Completer<List<String>> completer = Completer();

  void processData() {
    String? word;
    while ((word = _parseWord()) != null) {
      if (word == '') continue;
      response.add(word);
      if (word == '!done' || word == '!trap') {
        completer.complete(response);
        socket!.pause();
        return;
      }
    }
  }

  // 处理已有缓冲区数据
  processData();
  if (completer.isCompleted) return completer.future;

  // 监听新数据
  StreamSubscription? subscription;
  subscription = socket!.listen((Uint8List data) {
    // 追加新数据到缓冲区
    Uint8List newBuffer = Uint8List(_buffer.length + data.length);
    newBuffer.setRange(0, _buffer.length, _buffer);
    newBuffer.setRange(_buffer.length, newBuffer.length, data);
    _buffer.setRange(0, newBuffer.length, newBuffer);

    processData();
    if (completer.isCompleted) subscription?.cancel();
  });

  return completer.future;
}

登录排查步骤

• 在路由器启用API日志：/system logging add topics=api,debug，查看路由器收到的命令格式是否正确
• 在代码中打印发送的二进制字节，对比MikroTik API文档的格式要求
• 检查用户权限：确保登录用户所在组拥有api权限（/user group set full policy=api,...）

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