sevenhill/Qlyra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge: kolibri-migration into FullStack

Browse Source 
# Conflicts:
#	lib/frontend/screens/profile/customization_section.dart
This commit is contained in:
klockky
2026-07-14 12:40:22 +03:00
parent 99bede10d4 5f3ebb2ade
commit f7103da880
30 changed files with 683 additions and 1875 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/core/calls/conversation_params.dart
+12 -69
View File
@@ -1,7 +1,4 @@
import 'dart:convert';
import 'dart:typed_data';
import '../protocol/lz4_block.dart';
import 'package:kolibri/kolibri.dart' as kb;
/// Параметры подключения к звонку (`vcp`), которые сервер присылает в пуше
/// входящего звонка (opcode 137) и в ответе на инициацию исходящего.
@@ -79,73 +76,19 @@ class ConversationParams {
return nowSec >= expiresAt! - 5;
}
static List<String> _splitTurn(Object? value) {
if (value is! String || value.isEmpty) return const [];
return value
.split(',')
.map((e) => e.trim())
.where((e) => e.isNotEmpty)
.toList();
}
static List<String> _stringList(Object? value) {
if (value is! List) return const [];
return value.whereType<String>().toList();
}
/// Распаковывает и парсит строку `vcp`. Возвращает `null`, если формат
/// не распознан.
/// Распаковывает и парсит строку `vcp` через Rust-ядро (kolibri). Возвращает
/// `null`, если формат не распознан. Требует инициализации `initKolibri()`.
static ConversationParams? decode(String vcp) {
final sep = vcp.indexOf(':');
if (sep <= 0) return null;
final rawLen = int.tryParse(vcp.substring(0, sep));
if (rawLen == null || rawLen <= 0) return null;
final Uint8List compressed;
try {
compressed = base64.decode(vcp.substring(sep + 1));
} catch (_) {
return null;
}
final Uint8List bytes;
try {
final decompressed = lz4BlockDecompress(compressed, rawLen);
bytes = decompressed.length > rawLen
? Uint8List.sublistView(decompressed, 0, rawLen)
: decompressed;
} catch (_) {
return null;
}
final Object? json;
try {
json = jsonDecode(utf8.decode(bytes));
} catch (_) {
return null;
}
if (json is! Map) return null;
final token = json['tkn'];
final wse = json['wse'];
if (token is! String || wse is! String) return null;
final kb.CallParams? p = kb.decodeVcp(vcp: vcp, conversationId: '');
if (p == null) return null;
return ConversationParams(
token: token,
wsEndpoint: wse,
wsIps: _stringList(json['wsip']),
wtEndpoint: json['wte'] as String?,
wtIps: _stringList(json['wtip']),
callsApiEndpoint: json['vcae'] as String?,
callsApiIps: _stringList(json['vcaip']),
clientType: json['srcp'] as String?,
expiresAt: json['et'] is int ? json['et'] as int : null,
stun: json['stne'] as String?,
turn: _splitTurn(json['trne']),
turnUser: json['trnu'] as String?,
turnPassword: json['trnp'] as String?,
isVideo: json['iv'] == true,
token: p.token,
wsEndpoint: p.wsEndpoint,
stun: p.stun,
turn: p.turn,
turnUser: p.turnUser,
turnPassword: p.turnPassword,
isVideo: p.isVideo,
);
}
}
lib/core/calls/ws2_signaling.dart
+45 -87
View File
@@ -1,8 +1,8 @@
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import '../utils/logger.dart';
import 'package:kolibri/kolibri.dart' as kb;
import 'conversation_params.dart';
/// Параметры подключения к сигналинг-сокету ws2.
@@ -84,17 +84,19 @@ class Ws2CommandException implements Exception {
/// Клиент сигналинга звонка поверх WebSocket `ws2`.
///
/// Конверт сообщений (подтверждено захватом `docs/ws2_capture.log`):
/// Тонкий адаптер над Rust-ядром (kolibri [kb.CallSignaling]): ядро держит
/// WebSocket, корреляцию `sequence`/`response`, keepalive `ping`→`pong` и
/// разбор кадров; здесь — прежний Dart-интерфейс для [call_session].
///
/// Конверт сообщений:
/// - запрос: `{"command": ..., ..., "sequence": N}`
/// - ответ: `{"sequence": N, "response": "<command>", "type": "response"}`
/// - пуш: `{..., "notification": "<name>", "type": "notification"}`
/// - keepalive: текстовый кадр `ping` → ответ `pong`.
class Ws2Signaling {
final Ws2Config config;
WebSocket? _socket;
int _sequence = 0;
final Map<int, Completer<Map<String, dynamic>>> _pending = {};
kb.CallSignaling? _call;
StreamSubscription<String>? _notifSub;
final _notifications = StreamController<Map<String, dynamic>>.broadcast();
final _closed = Completer<Object?>();
@@ -104,106 +106,60 @@ class Ws2Signaling {
/// Пуши сервера (`type == "notification"`). Фильтруй по полю `notification`.
Stream<Map<String, dynamic>> get notifications => _notifications.stream;
/// Завершается, когда сокет закрыт (значение — причина закрытия, если была).
/// Завершается, когда сокет закрыт.
Future<Object?> get done => _closed.future;
bool get isConnected => _socket != null;
bool get isConnected => _call?.isConnected() ?? false;
Future<void> connect() async {
final socket = await WebSocket.connect(
config.uri.toString(),
headers: {'User-Agent': 'okhttp/4.12.0'},
final call = await kb.connectCallSignaling(
url: config.uri.toString(),
userAgent: 'okhttp/4.12.0',
);
_socket = socket;
socket.listen(
_onFrame,
onError: _onDone,
_call = call;
_notifSub = call.notifications().listen(
(json) {
Object? decoded;
try {
decoded = jsonDecode(json);
} catch (_) {
return;
}
if (decoded is Map<String, dynamic>) _notifications.add(decoded);
},
onError: (_) => _onDone(null),
onDone: () => _onDone(null),
cancelOnError: false,
);
}
void _onFrame(dynamic frame) {
if (frame is String && frame == 'ping') {
_socket?.add('pong');
return;
}
final String text;
if (frame is String) {
text = frame;
} else if (frame is List<int>) {
text = utf8.decode(frame);
} else {
return;
}
Object? decoded;
try {
decoded = jsonDecode(text);
} catch (_) {
return;
}
if (decoded is! Map<String, dynamic>) return;
final label =
decoded['notification'] ?? decoded['response'] ?? decoded['type'];
final dump = jsonEncode(decoded);
logger.t('[ws2] ← $label');
logger.t(dump.length > 1500
? '${dump.substring(0, 1500)}… (${dump.length}b)'
: dump);
final type = decoded['type'];
if (type == 'response' || type == 'error') {
final seq = decoded['sequence'];
if (seq is int) {
final completer = _pending.remove(seq);
if (completer != null && !completer.isCompleted) {
completer.complete(decoded);
}
}
if (type == 'error') _notifications.add(decoded);
return;
}
if (type == 'notification' || decoded.containsKey('notification')) {
_notifications.add(decoded);
}
}
void _onDone(Object? error) {
for (final c in _pending.values) {
if (!c.isCompleted) c.completeError(error ?? const SocketException('ws2 closed'));
}
_pending.clear();
if (!_closed.isCompleted) _closed.complete(error);
if (!_notifications.isClosed) _notifications.close();
}
/// Отправляет команду и ждёт ответ сервера. Бросает [Ws2CommandException],
/// если в ответе есть поле `error`.
/// если сервер вернул ошибку.
Future<Map<String, dynamic>> sendCommand(
String command, {
Map<String, dynamic> extra = const {},
Duration timeout = const Duration(seconds: 15),
}) {
final socket = _socket;
if (socket == null) {
}) async {
final call = _call;
if (call == null) {
return Future.error(StateError('ws2 не подключён'));
}
final seq = ++_sequence;
final completer = Completer<Map<String, dynamic>>();
_pending[seq] = completer;
socket.add(jsonEncode({'command': command, ...extra, 'sequence': seq}));
return completer.future.timeout(timeout).then((response) {
final error = response['error'];
if (error != null) throw Ws2CommandException(command, error);
return response;
});
try {
final response = await call
.sendCommand(command: command, extraJson: jsonEncode(extra))
.timeout(timeout);
final decoded = jsonDecode(response);
return decoded is Map<String, dynamic>
? decoded
: <String, dynamic>{};
} catch (e) {
throw Ws2CommandException(command, e);
}
}
/// Передаёт SDP (offer/answer) другому участнику.
@@ -329,7 +285,9 @@ class Ws2Signaling {
extra: {'mediaSource': mediaSource, 'layers': layers});
Future<void> close() async {
await _socket?.close();
_socket = null;
await _notifSub?.cancel();
_notifSub = null;
_call?.close();
_call = null;
}
}
lib/core/protocol/lz4_block.dart
-71
View File
@@ -1,71 +0,0 @@
import 'dart:typed_data';
/// LZ4 block декомпрессия (без frame-заголовка).
///
/// Сервер шлёт block-формат как в транспорте (payload пакетов), так и в
/// `vcp`-параметрах звонка. dart_lz4 поддерживает только frame-формат, поэтому
/// block распаковывается вручную.
Uint8List lz4BlockDecompress(Uint8List src, int maxSize) {
var out = Uint8List(1024);
int outLen = 0;
int pos = 0;
void ensure(int extra) {
if (outLen + extra > maxSize) throw StateError('LZ4: превышен лимит');
if (outLen + extra <= out.length) return;
var newCap = out.length * 2;
while (newCap < outLen + extra) {
newCap *= 2;
}
if (newCap > maxSize) newCap = maxSize;
final grown = Uint8List(newCap);
grown.setRange(0, outLen, out);
out = grown;
}
while (pos < src.length) {
final token = src[pos++];
var litLen = token >> 4;
if (litLen == 15) {
while (pos < src.length) {
final b = src[pos++];
litLen += b;
if (b != 255) break;
}
}
if (litLen > 0) {
ensure(litLen);
out.setRange(outLen, outLen + litLen, src, pos);
outLen += litLen;
pos += litLen;
}
if (pos >= src.length) break;
if (pos + 1 >= src.length) throw StateError('LZ4: unexpected end of input');
final offset = src[pos] | (src[pos + 1] << 8);
pos += 2;
if (offset == 0) throw StateError('LZ4: offset = 0');
var matchLen = (token & 0x0F) + 4;
if ((token & 0x0F) == 0x0F) {
while (pos < src.length) {
final b = src[pos++];
matchLen += b;
if (b != 255) break;
}
}
ensure(matchLen);
final start = outLen - offset;
if (start < 0) throw StateError('LZ4: offset за пределами вывода');
for (var i = 0; i < matchLen; i++) {
out[outLen + i] = out[start + i];
}
outLen += matchLen;
}
return Uint8List.sublistView(out, 0, outLen);
}
lib/core/protocol/packet.dart
+4 -152
View File
@@ -1,16 +1,3 @@
import 'dart:typed_data';
import 'dart:isolate';
import 'package:dart_lz4/dart_lz4.dart';
import 'package:libcompress/libcompress.dart';
import 'package:msgpack_dart/msgpack_dart.dart' as msgpack;
import 'lz4_block.dart';
/// ver(1) + cmd(1) + seq(2) + opcode(2) + packedLen(4) = 10
const int headerSize = 10;
/// Потолок распаковки payload (анти-бомба); буфер растёт динамически до него.
const int _maxDecompressedSize = 32 * 1024 * 1024; // 32 MB
/// Типы команд в протоколе
abstract class CmdType {
static const int request =
@@ -22,17 +9,11 @@ abstract class CmdType {
static const int error = 3; // ответ: ошибка
}
/// Распакованный бинарный пакет
/// Распакованный пакет.
///
/// Формат заголовка (10 байт):
/// ```
/// [0] ver — версия протокола (uint8) (по умолчанию 10)
/// [1] cmd — тип команды (uint8) (при отправке от клиента равно 0)
/// [2..3] seq — порядковый номер (uint16 BE)
/// [4..5] opcode — код операции (uint16 BE)
/// [6..9] packedLen — флаг сжатия [6] + длина payload [7..9] (uint32 BE)
/// [10..] payload — данные в MsgPack, опционально сжатые LZ4
/// ```
/// Провод (фрейминг, MsgPack, сжатие) живёт в Rust-ядре kolibri; здесь пакет —
/// это уже декодированный [payload] (Map/List/скаляр, бинарь — Uint8List) плюс
/// метаданные заголовка.
class Packet {
int api;
int cmd;
@@ -109,132 +90,3 @@ bool isSessionStateError(Object error) {
text.contains('авторизационная сессия') ||
text.contains('сессия не онлайн');
}
/// Payload меньше этого размера отправляется без сжатия (как в оригинале).
const int _compressionThreshold = 32;
/// Упаковка пакета для отправки на сервер.
///
/// Payload сериализуется в MsgPack и при размере >= [_compressionThreshold]
/// сжимается LZ4-block. Старший байт поля packedLen — флаг сжатия:
/// `0` — без сжатия, иначе `(rawLen ~/ compLen) + 1` (множитель размера, по
/// которому получатель выделяет буфер под распаковку).
Uint8List packPacket(int opcode, Map<dynamic, dynamic> payload, {int seq = 0}) {
final Uint8List raw = msgpack.serialize(payload);
final List<int> body;
final int flag;
if (raw.length < _compressionThreshold) {
body = raw;
flag = 0;
} else {
body = lz4Compress(raw);
flag = (raw.length ~/ body.length) + 1;
}
final out = Uint8List(headerSize + body.length);
final header = ByteData.view(out.buffer, out.offsetInBytes, headerSize);
header.setUint8(0, 10);
header.setUint8(1, CmdType.request);
header.setUint16(2, seq, Endian.big);
header.setUint16(4, opcode, Endian.big);
header.setUint32(
6,
((flag & 0xFF) << 24) | (body.length & 0xFFFFFF),
Endian.big,
);
out.setRange(headerSize, out.length, body);
return out;
}
const int _isolateDecodeThreshold = 4096;
Future<Packet> unpackPacket(Uint8List packet) async {
final header = ByteData.sublistView(packet);
final apiVer = header.getUint8(0) & 0xFF;
final cmd = header.getUint8(1) & 0xFF;
final seq = header.getUint16(2) & 0xFFFF;
final opcode = header.getUint16(4) & 0xFFFF;
final packedLen = header.getUint32(6);
final compFlag = packedLen >> 24;
final payloadLength = packedLen & 0xFFFFFF;
if (payloadLength == 0) {
return Packet(api: apiVer, cmd: cmd, seq: seq, opcode: opcode);
}
final end = headerSize + payloadLength;
if (end > packet.length) {
throw Exception('Packet payload length $payloadLength exceeds buffer');
}
final slice = Uint8List.sublistView(packet, headerSize, end);
dynamic payload;
if (compFlag == 0 && slice.length < _isolateDecodeThreshold) {
payload = _deserializePayload(slice, compFlag);
} else {
final owned = Uint8List.fromList(slice);
payload = await Isolate.run(() => _deserializePayload(owned, compFlag));
}
return Packet(
api: apiVer,
cmd: cmd,
seq: seq,
opcode: opcode,
payload: payload,
);
}
dynamic _deserializePayload(Uint8List payloadBytes, int compFlag) {
var bytes = payloadBytes;
if (compFlag != 0) {
bytes = _decompressPayload(bytes);
}
if (bytes.isEmpty) return null;
try {
return msgpack.deserialize(bytes);
} catch (e) {
throw Exception('MsgPack deserialization error: $e');
}
}
/// Определяет формат сжатия по magic-number и распаковывает payload.
/// Сервер может присылать LZ4 block ИЛИ Zstandard в зависимости от ответа.
Uint8List _decompressPayload(Uint8List src) {
// Zstandard: magic 28 B5 2F FD (little-endian)
if (src.length >= 4 &&
src[0] == 0x28 &&
src[1] == 0xB5 &&
src[2] == 0x2F &&
src[3] == 0xFD) {
try {
return ZstdCodec(
maxDecompressedSize: _maxDecompressedSize,
).decompress(src);
} catch (e) {
throw Exception('Zstd decompression error: $e');
}
}
// LZ4 frame: magic 04 22 4D 18
if (src.length >= 4 &&
src[0] == 0x04 &&
src[1] == 0x22 &&
src[2] == 0x4D &&
src[3] == 0x18) {
try {
return lz4Decompress(src, decompressedSize: _maxDecompressedSize);
} catch (e) {
throw Exception('LZ4 frame decompression error: $e');
}
}
// По умолчанию — LZ4 block (без magic)
try {
return lz4BlockDecompress(src, _maxDecompressedSize);
} catch (e) {
throw Exception('LZ4 block decompression error: $e');
}
}
lib/core/push/push_service.dart
+5
View File
@@ -5,6 +5,7 @@ import 'package:firebase_core/firebase_core.dart';
import 'package:firebase_messaging/firebase_messaging.dart';
import 'package:flutter/widgets.dart';
import 'package:flutter_local_notifications/flutter_local_notifications.dart';
import 'package:kolibri/kolibri.dart' show initKolibri;
import 'package:shared_preferences/shared_preferences.dart';
import '../../backend/api.dart';
@@ -53,6 +54,9 @@ Future<void> _handleCallDecline(String payloadJson) async {
}
if (vcp.isEmpty || conversationId.isEmpty) return;
// Фоновый изолят: инициализируем ядро перед vcp-декодом/сигналингом.
await initKolibri();
final params = ConversationParams.decode(vcp);
if (params == null) return;
@@ -83,6 +87,7 @@ Future<void> _handleReply(String payloadJson, String text) async {
if (account == 0 || chatId == 0) return;
WidgetsFlutterBinding.ensureInitialized();
await initKolibri();
if (AppInstance.isNamed) {
try {
SharedPreferences.setPrefix('flutter.${AppInstance.id}.');
lib/core/transport/connection.dart
-189
View File
@@ -1,189 +0,0 @@
import 'dart:async';
import 'dart:io';
import 'dart:typed_data';
import '../config/proxy_config.dart';
import '../utils/logger.dart';
import 'proxy_connector.dart';
import 'tls_config.dart';
import 'traffic_monitor.dart';
import 'vpn_bypass.dart';
enum SocketState { disconnected, connecting, connected }
/// Обёртка над TCP + TLS сокетом.
/// Отдаёт сырые байты через [dataStream], сборкой пакетов занимается [PacketReceiver].
class Connection {
static const Duration _defaultConnectTimeout = Duration(seconds: 15);
static const Duration _proxyLoadTimeout = Duration(seconds: 8);
static const Duration _vpnCallTimeout = Duration(seconds: 5);
SecureSocket? _socket;
StreamSubscription<Uint8List>? _subscription;
SocketState _state = SocketState.disconnected;
final _dataController = StreamController<Uint8List>.broadcast();
final _stateController = StreamController<SocketState>.broadcast();
Stream<Uint8List> get dataStream => _dataController.stream;
Stream<SocketState> get stateStream => _stateController.stream;
SocketState get state => _state;
bool get isConnected => _state == SocketState.connected;
void _setState(SocketState newState) {
if (_state == newState) return;
_state = newState;
if (!_stateController.isClosed) _stateController.add(newState);
}
Future<void> connect(
String host,
int port, {
bool bypassVpn = false,
Duration? timeout,
}) async {
if (_state != SocketState.disconnected) {
logger.w('Connection.connect пропущен: state=$_state (уже $_state)');
return;
}
_setState(SocketState.connecting);
try {
logger.i('Connection: загрузка прокси-конфига');
ProxySettings proxySettings;
try {
proxySettings = await ProxyConfig.load().timeout(_proxyLoadTimeout);
} catch (e) {
logger.w('Connection: ProxyConfig.load завис/упал ($e) — без прокси');
proxySettings = const ProxySettings();
}
logger.i(
'Connection: VPN ${bypassVpn ? 'bind (обход)' : 'restoreDefault'}',
);
try {
if (bypassVpn) {
await VpnBypassService.instance.bind().timeout(_vpnCallTimeout);
} else {
await VpnBypassService.instance
.restoreDefault()
.timeout(_vpnCallTimeout);
}
} catch (e) {
logger.w('Connection: VPN-вызов завис/упал ($e) — продолжаю');
}
logger.i(
'Connection: открываю сокет $host:$port '
'(прокси: ${proxySettings.isEnabled ? proxySettings.type.name : 'нет'})',
);
final socket = await _openSecureSocket(
host,
port,
proxySettings,
timeout: timeout,
);
_socket = socket;
_setState(SocketState.connected);
logger.i('Подключено к $host:$port');
final route = proxySettings.isEnabled
? 'через прокси ${proxySettings.type.name}'
: bypassVpn
? 'напрямую (обход VPN)'
: 'прямое соединение';
TrafficMonitor.instance.recordEvent(
'Подключено',
detail: '$host:$port · TLS · $route',
endpoint: '$host:$port',
);
_subscription = _socket!.listen(
(data) {
if (!_dataController.isClosed) _dataController.add(data);
},
onError: (Object error) {
logger.e('Ошибка сокета: $error');
disconnect();
},
onDone: () {
logger.w('Сокет закрыт сервером');
disconnect();
},
);
} catch (e) {
logger.e('Не удалось подключиться: $e');
_setState(SocketState.disconnected);
rethrow;
}
}
Future<SecureSocket> _openSecureSocket(
String host,
int port,
ProxySettings proxySettings, {
Duration? timeout,
}) async {
final connectTimeout = timeout ?? _defaultConnectTimeout;
Socket socket;
if (proxySettings.isEnabled) {
final connector = ProxyConnector(proxySettings);
socket = await connector.connect(host, port).timeout(connectTimeout);
logger.i('Подключено через прокси ${proxySettings.type.name}');
} else {
logger.i('Connection: TCP connect $host:$port (лимит ${connectTimeout.inSeconds}с)');
socket = await Socket.connect(host, port, timeout: connectTimeout);
logger.i('Connection: TCP установлен, начинаю TLS');
}
final allowInsecure = await TlsConfig.isInsecureAllowed();
if (allowInsecure) {
logger.w(
'TLS: проверка сертификата отключена (дебаг) — соединение уязвимо к MitM',
);
}
final secured = allowInsecure
? SecureSocket.secure(socket, host: host, onBadCertificate: (_) => true)
: SecureSocket.secure(socket, host: host);
try {
final result = await secured.timeout(connectTimeout);
logger.i('Connection: TLS-handshake завершён');
return result;
} on TimeoutException {
logger.w('Connection: TLS-handshake таймаут ${connectTimeout.inSeconds}с');
socket.destroy();
rethrow;
}
}
void write(Uint8List data) {
if (_socket == null || !isConnected) {
throw StateError('Нельзя писать: сокет не подключён');
}
_socket!.add(data);
}
Future<void> disconnect() async {
_subscription?.cancel();
_subscription = null;
final socket = _socket;
_socket = null;
if (socket != null) {
TrafficMonitor.instance.recordEvent('Соединение закрыто');
try {
await socket.close();
} catch (e) {
logger.w('Ошибка при закрытии сокета: $e');
}
}
_setState(SocketState.disconnected);
}
Future<void> dispose() async {
await disconnect();
await _dataController.close();
await _stateController.close();
}
}
lib/core/transport/proxy_connector.dart
-360
View File
@@ -1,360 +0,0 @@
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'dart:typed_data';
import '../config/proxy_config.dart';
import '../utils/logger.dart';
class ProxyConnector {
final ProxySettings settings;
ProxyConnector(this.settings);
Future<Socket> connect(String targetHost, int targetPort) async {
switch (settings.type) {
case ProxyType.socks5:
return _connectSocks5(targetHost, targetPort);
case ProxyType.httpConnect:
return _connectHttpConnect(targetHost, targetPort);
case ProxyType.none:
return Socket.connect(targetHost, targetPort);
}
}
// ── SOCKS5 (RFC 1928) ──────────────────────────────────────────────────
Future<Socket> _connectSocks5(String targetHost, int targetPort) async {
final proxySocket = await RawSocket.connect(settings.host, settings.port);
logger.i('SOCKS5: подключено к прокси ${settings.host}:${settings.port}');
final io = _RawSocketIO(proxySocket);
try {
// 1. Greeting
final useAuth = settings.hasCredentials;
if (useAuth) {
await io.write([0x05, 0x02, 0x00, 0x02]);
} else {
await io.write([0x05, 0x01, 0x00]);
}
var response = await io.readExact(2);
if (response[0] != 0x05) {
throw SocketException(
'SOCKS5: неверная версия протокола: ${response[0]}',
);
}
final method = response[1];
if (method == 0xFF) {
throw SocketException(
'SOCKS5: сервер отклонил все методы аутентификации',
);
}
// 2. Аутентификация (RFC 1929)
if (method == 0x02) {
if (!useAuth) {
throw SocketException('SOCKS5: прокси требует аутентификацию');
}
final usernameBytes = utf8.encode(settings.username ?? '');
final passwordBytes = utf8.encode(settings.password ?? '');
final authPacket = BytesBuilder()
..addByte(0x01)
..addByte(usernameBytes.length)
..add(usernameBytes)
..addByte(passwordBytes.length)
..add(passwordBytes);
await io.write(authPacket.toBytes());
final authResponse = await io.readExact(2);
if (authResponse[1] != 0x00) {
throw SocketException('SOCKS5: аутентификация не пройдена');
}
logger.i('SOCKS5: аутентификация пройдена');
}
// 3. Connect request
final hostBytes = utf8.encode(targetHost);
final connectPacket = BytesBuilder()
..addByte(0x05) // VER
..addByte(0x01) // CMD: CONNECT
..addByte(0x00) // RSV
..addByte(0x03) // ATYP: domain
..addByte(hostBytes.length)
..add(hostBytes)
..addByte((targetPort >> 8) & 0xFF)
..addByte(targetPort & 0xFF);
await io.write(connectPacket.toBytes());
// 4. Reply
final reply = await io.readExact(4);
if (reply[0] != 0x05) {
throw SocketException('SOCKS5: неверная версия в ответе');
}
if (reply[1] != 0x00) {
throw SocketException('SOCKS5: ошибка подключения, код: ${reply[1]}');
}
// Пропускаем bind address
switch (reply[3]) {
case 0x01:
await io.readExact(4 + 2);
break;
case 0x03:
final lenBuf = await io.readExact(1);
await io.readExact(lenBuf[0] + 2);
break;
case 0x04:
await io.readExact(16 + 2);
break;
}
logger.i('SOCKS5: туннель к $targetHost:$targetPort установлен');
// Создаём локальную пару и проксируем данные
return _bridgeToFreshSocket(proxySocket, io);
} catch (e) {
io.dispose();
proxySocket.close();
rethrow;
}
}
// ── HTTP CONNECT ────────────────────────────────────────────────────────
Future<Socket> _connectHttpConnect(String targetHost, int targetPort) async {
final proxySocket = await RawSocket.connect(settings.host, settings.port);
logger.i(
'HTTP CONNECT: подключено к прокси ${settings.host}:${settings.port}',
);
final io = _RawSocketIO(proxySocket);
try {
final request = StringBuffer()
..write('CONNECT $targetHost:$targetPort HTTP/1.1\r\n')
..write('Host: $targetHost:$targetPort\r\n');
if (settings.hasCredentials) {
final credentials = base64Encode(
utf8.encode('${settings.username}:${settings.password}'),
);
request.write('Proxy-Authorization: Basic $credentials\r\n');
}
request.write('\r\n');
await io.write(utf8.encode(request.toString()));
// Читаем HTTP-ответ до \r\n\r\n
final headerBytes = <int>[];
while (true) {
final byte = await io.readExact(1);
headerBytes.add(byte[0]);
if (headerBytes.length >= 4 &&
headerBytes[headerBytes.length - 4] == 0x0D &&
headerBytes[headerBytes.length - 3] == 0x0A &&
headerBytes[headerBytes.length - 2] == 0x0D &&
headerBytes[headerBytes.length - 1] == 0x0A) {
break;
}
if (headerBytes.length > 8192) {
throw SocketException(
'HTTP CONNECT: заголовок ответа слишком большой',
);
}
}
final responseStr = utf8.decode(headerBytes, allowMalformed: true);
final statusLine = responseStr.split('\r\n').first;
final parts = statusLine.split(' ');
if (parts.length < 2) {
throw SocketException('HTTP CONNECT: некорректный ответ: $statusLine');
}
final statusCode = int.tryParse(parts[1]) ?? 0;
if (statusCode != 200) {
throw SocketException('HTTP CONNECT: прокси вернул статус $statusCode');
}
logger.i('HTTP CONNECT: туннель к $targetHost:$targetPort установлен');
return _bridgeToFreshSocket(proxySocket, io);
} catch (e) {
io.dispose();
proxySocket.close();
rethrow;
}
}
Future<Socket> _bridgeToFreshSocket(
RawSocket proxySocket,
_RawSocketIO io,
) async {
ServerSocket? server;
try {
server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
} catch (e) {
io.dispose();
proxySocket.close();
rethrow;
}
final clientFuture = Socket.connect(
InternetAddress.loopbackIPv4,
server.port,
);
final serverSide = await server.first;
final clientSide = await clientFuture;
await server.close();
io.onData = (data) {
serverSide.add(data);
};
io.onClosed = () {
serverSide.close();
};
serverSide.listen(
(data) {
unawaited(
io.write(data).catchError((Object _) {
try {
serverSide.destroy();
} catch (_) {}
}),
);
},
onError: (Object _) {
proxySocket.shutdown(SocketDirection.send);
},
onDone: () {
proxySocket.shutdown(SocketDirection.send);
},
);
// Сливаем данные, буферизованные во время handshake
io.flushBuffered();
logger.i('Прокси-мост через loopback создан');
return clientSide;
}
}
/// Обёртка над единственной подпиской [RawSocket], с буфером для чтения.
///
/// После handshake переключается в режим моста:
/// данные из proxy-сокета пересылаются через [onData] в loopback-пару.
class _RawSocketIO {
final RawSocket _socket;
late final StreamSubscription<RawSocketEvent> _sub;
final _readBuffer = <int>[];
Completer<void>? _readWaiter;
Completer<void>? _writeWaiter;
bool _closed = false;
Object? _error;
/// Коллбэк для данных в режиме моста.
void Function(Uint8List data)? onData;
/// Коллбэк закрытия в режиме моста.
void Function()? onClosed;
_RawSocketIO(this._socket) {
_sub = _socket.listen(
_onEvent,
onError: (Object err) {
_error = err;
_closed = true;
_readWaiter?.completeError(err);
_readWaiter = null;
_writeWaiter?.completeError(err);
_writeWaiter = null;
},
);
}
void _onEvent(RawSocketEvent event) {
switch (event) {
case RawSocketEvent.read:
final data = _socket.read();
if (data != null) {
if (onData != null) {
// Режим моста — пересылаем напрямую
onData!(data);
} else {
// Режим handshake — буферизуем
_readBuffer.addAll(data);
_readWaiter?.complete();
_readWaiter = null;
}
}
break;
case RawSocketEvent.write:
_writeWaiter?.complete();
_writeWaiter = null;
break;
case RawSocketEvent.readClosed:
case RawSocketEvent.closed:
_closed = true;
onClosed?.call();
_readWaiter?.completeError(SocketException('Прокси закрыл соединение'));
_readWaiter = null;
_writeWaiter?.completeError(
SocketException('Прокси закрыл соединение'),
);
_writeWaiter = null;
break;
}
}
/// Читает ровно [count] байт.
Future<Uint8List> readExact(int count) async {
while (_readBuffer.length < count) {
if (_error != null) throw _error!;
if (_closed) {
throw SocketException(
'Соединение закрыто '
'(ожидали $count байт, получили ${_readBuffer.length})',
);
}
_readWaiter = Completer<void>();
await _readWaiter!.future.timeout(
const Duration(seconds: 15),
onTimeout: () => throw SocketException('Тайм-аут при чтении от прокси'),
);
}
final result = Uint8List.fromList(_readBuffer.sublist(0, count));
_readBuffer.removeRange(0, count);
return result;
}
/// Записывает все байты.
Future<void> write(List<int> data) async {
var offset = 0;
while (offset < data.length) {
if (_error != null) throw _error!;
if (_closed) throw SocketException('Соединение закрыто при записи');
final written = _socket.write(data, offset);
if (written > 0) {
offset += written;
} else {
_writeWaiter = Completer<void>();
await _writeWaiter!.future.timeout(
const Duration(seconds: 15),
onTimeout: () =>
throw SocketException('Тайм-аут при записи в прокси'),
);
}
}
}
/// Пересылает данные, оставшиеся в буфере после handshake, в мост.
void flushBuffered() {
if (_readBuffer.isNotEmpty && onData != null) {
onData!(Uint8List.fromList(_readBuffer));
_readBuffer.clear();
}
}
void dispose() {
_sub.cancel();
}
}
lib/core/transport/receiver.dart
-74
View File
@@ -1,74 +0,0 @@
import 'dart:typed_data';
import '../protocol/packet.dart';
class ReceiverOverflowException implements Exception {
final int size;
const ReceiverOverflowException(this.size);
@override
String toString() => 'PacketReceiver: переполнение буфера ($size B)';
}
class PacketReceiver {
Uint8List _buffer = Uint8List(0);
int _start = 0;
int _end = 0;
static const int _maxBufferSize = 16 * 1024 * 1024;
List<Uint8List> feed(Uint8List data) {
_append(data);
if (_end - _start > _maxBufferSize) {
final overflow = _end - _start;
reset();
throw ReceiverOverflowException(overflow);
}
final packets = <Uint8List>[];
while (_end - _start >= headerSize) {
final bd = ByteData.view(
_buffer.buffer,
_buffer.offsetInBytes + _start,
headerSize,
);
final packedLen = bd.getUint32(6, Endian.big);
final payloadLength = packedLen & 0xFFFFFF;
final totalLength = headerSize + payloadLength;
if (_end - _start < totalLength) break;
packets.add(Uint8List.sublistView(_buffer, _start, _start + totalLength));
_start += totalLength;
}
if (_start == _end) {
_start = 0;
_end = 0;
}
return packets;
}
void _append(Uint8List data) {
final pending = _end - _start;
if (pending == 0) {
_buffer = Uint8List.fromList(data);
_start = 0;
_end = data.length;
return;
}
final total = pending + data.length;
final newBuffer = Uint8List(total);
newBuffer.setRange(0, pending, _buffer, _start);
newBuffer.setRange(pending, total, data);
_buffer = newBuffer;
_start = 0;
_end = total;
}
void reset() {
_buffer = Uint8List(0);
_start = 0;
_end = 0;
}
}
lib/core/transport/sender.dart
-27
View File
@@ -1,27 +0,0 @@
import '../protocol/packet.dart';
import '../utils/log_redact.dart';
import '../utils/logger.dart';
import 'connection.dart';
import 'traffic_monitor.dart';
class PacketSender {
int _seq = 0;
int get currentSeq => _seq;
int _nextSeq() {
_seq = (_seq + 1) % 65536;
return _seq;
}
int send(Connection connection, int opcode, Map<dynamic, dynamic> payload) {
final seq = _nextSeq();
final data = packPacket(opcode, payload, seq: seq);
connection.write(data);
TrafficMonitor.instance.recordOutgoing(opcode, payload, seq, data.length);
logger.i(
'=> {ver: 10, cmd: 0, seq: $seq, opcode: $opcode, payload: ${payloadForLog(payload)}}',
);
return seq;
}
}