evgeny
4 days ago
3 changed files with 129 additions and 2 deletions
@ -0,0 +1,82 @@
|
||||
#include <string.h> |
||||
|
||||
#include "../../lib/debug_config.h" |
||||
#include "../transport_layer/etcp_api.h" |
||||
#include "stream_wire.h" |
||||
|
||||
/* Заголовок одинаков у raw и router payload; cmd raw сохраняется внутри routed-кодограммы. */ |
||||
int stream_wire_parse(const uint8_t *wire, size_t length, struct stream_message *m) { |
||||
if (!wire || !m || length < STREAM_WIRE_HEADER || wire[0] != ETCP_ID_STREAM || wire[1] != STREAM_WIRE_VERSION |
||||
|| wire[2] < STREAM_CMD_INFO_REQ || wire[2] > STREAM_CMD_REVERSE_REQ) { |
||||
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream wire header invalid bytes=%zu", length); return STREAM_ERR_FORMAT; |
||||
} |
||||
struct stream_message result = {.command = wire[2], .subblock = wire[3], .payload = wire + STREAM_WIRE_HEADER, |
||||
.length = length - STREAM_WIRE_HEADER}; |
||||
result.descriptor.group_id = stream_read_be(wire + 4, 8); |
||||
result.descriptor.source_node_id = stream_read_be(wire + 12, 8); |
||||
result.descriptor.source_epoch = stream_read_be(wire + 20, 8); |
||||
result.sender_epoch = stream_read_be(wire + 28, 8); result.receiver_epoch = stream_read_be(wire + 36, 8); |
||||
result.handle = stream_read_be(wire + 44, 8); |
||||
if (!result.descriptor.group_id || !result.descriptor.source_node_id || !result.descriptor.source_epoch || !result.sender_epoch) { |
||||
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream wire zero key/epoch command=%u", result.command); return STREAM_ERR_FORMAT; |
||||
} |
||||
*m = result; return STREAM_OK; |
||||
} |
||||
|
||||
void stream_wire_header(uint8_t *wire, const struct stream_message *m) { |
||||
wire[0] = ETCP_ID_STREAM; wire[1] = STREAM_WIRE_VERSION; wire[2] = m->command; wire[3] = m->subblock; |
||||
stream_write_be(wire + 4, m->descriptor.group_id, 8); stream_write_be(wire + 12, m->descriptor.source_node_id, 8); |
||||
stream_write_be(wire + 20, m->descriptor.source_epoch, 8); stream_write_be(wire + 28, m->sender_epoch, 8); |
||||
stream_write_be(wire + 36, m->receiver_epoch, 8); stream_write_be(wire + 44, m->handle, 8); |
||||
} |
||||
|
||||
/* История блока проверяется отдельно от BGP-пути; получатель не должен уже встречаться в ней. */ |
||||
int stream_path_validate(const struct stream_path *path, uint64_t source, uint64_t sender, uint64_t forbidden) { |
||||
if (!path || !path->count || path->count > STREAM_PATH_MAX || path->nodes[0] != source |
||||
|| path->nodes[path->count - 1] != sender) goto invalid; |
||||
for (unsigned i = 0; i < path->count; i++) { |
||||
if (!path->nodes[i] || (forbidden && path->nodes[i] == forbidden)) goto invalid; |
||||
for (unsigned j = 0; j < i; j++) if (path->nodes[i] == path->nodes[j]) goto invalid; |
||||
} |
||||
return STREAM_OK; |
||||
invalid: |
||||
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream path invalid/loop source=%llx sender=%llx receiver=%llx count=%u", |
||||
(unsigned long long)source, (unsigned long long)sender, (unsigned long long)forbidden, path ? path->count : 0); |
||||
return STREAM_ERR_FORMAT; |
||||
} |
||||
|
||||
int stream_path_parse(const uint8_t *wire, size_t length, struct stream_path *path, size_t *used) { |
||||
if (!wire || !path || !used || !length || !wire[0] || wire[0] > STREAM_PATH_MAX || length < 1u + wire[0] * 8u) { |
||||
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream path size invalid bytes=%zu", length); return STREAM_ERR_FORMAT; |
||||
} |
||||
struct stream_path result = {.count = wire[0]}; |
||||
for (unsigned i = 0; i < result.count; i++) result.nodes[i] = stream_read_be(wire + 1 + i * 8, 8); |
||||
*path = result; *used = 1 + result.count * 8; return STREAM_OK; |
||||
} |
||||
|
||||
size_t stream_path_write(uint8_t *wire, const struct stream_path *path) { |
||||
wire[0] = path->count; |
||||
for (unsigned i = 0; i < path->count; i++) stream_write_be(wire + 1 + i * 8, path->nodes[i], 8); |
||||
return 1 + path->count * 8; |
||||
} |
||||
|
||||
/* END имеет собственный domain и подписывает ключ запуска + последний индекс; UINT64_MAX означает пустой поток. */ |
||||
int stream_end_sign(const struct stream_descriptor *d, uint64_t last, const uint8_t key[32], uint8_t wire[STREAM_END_SIZE]) { |
||||
if (!d || !key || !wire) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream END sign invalid arguments"); return STREAM_ERR_ARGUMENT; } |
||||
memcpy(wire, "stream_end", 10); stream_write_be(wire + 10, d->group_id, 8); |
||||
stream_write_be(wire + 18, d->source_node_id, 8); stream_write_be(wire + 26, d->source_epoch, 8); |
||||
stream_write_be(wire + 34, last, 8); |
||||
if (sc_ed25519_sign(key, wire, 42, wire + 42) != SC_OK) { |
||||
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream END signature failed"); return STREAM_ERR_CRYPTO; |
||||
} |
||||
return STREAM_OK; |
||||
} |
||||
|
||||
int stream_end_verify(const struct stream_descriptor *d, const uint8_t key[32], const uint8_t *wire, size_t length, uint64_t *last) { |
||||
if (!d || !key || !wire || !last || length != STREAM_END_SIZE || memcmp(wire, "stream_end", 10) |
||||
|| stream_read_be(wire + 10, 8) != d->group_id || stream_read_be(wire + 18, 8) != d->source_node_id |
||||
|| stream_read_be(wire + 26, 8) != d->source_epoch || sc_ed25519_verify(key, wire, 42, wire + 42) != SC_OK) { |
||||
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream END invalid signature/key bytes=%zu", length); return STREAM_ERR_SIGNATURE; |
||||
} |
||||
*last = stream_read_be(wire + 34, 8); return STREAM_OK; |
||||
} |
||||
@ -0,0 +1,43 @@
|
||||
/* Прямой stream-протокол: канонический big-endian заголовок, пути и подписанный END.
|
||||
* IDs ETCP определены только в etcp_api.h. Указатели parser заимствованы до освобождения кодограммы. */ |
||||
#ifndef STREAM_WIRE_H |
||||
#define STREAM_WIRE_H |
||||
|
||||
#include "stream_proto.h" |
||||
|
||||
#define STREAM_WIRE_VERSION 1 |
||||
#define STREAM_WIRE_HEADER 52 |
||||
#define STREAM_PATH_MAX 32 |
||||
#define STREAM_END_SIZE 106 |
||||
#define STREAM_INFO_SIZE 21 |
||||
|
||||
enum stream_command { |
||||
STREAM_CMD_INFO_REQ = 1, STREAM_CMD_INFO_RESP, STREAM_CMD_SUBSCRIBE, STREAM_CMD_ACCEPT, |
||||
STREAM_CMD_REJECT, STREAM_CMD_RENEW, STREAM_CMD_UNSUBSCRIBE, STREAM_CMD_CLOSED, |
||||
STREAM_CMD_DATA, STREAM_CMD_GAP, STREAM_CMD_END, STREAM_CMD_REVERSE_REQ |
||||
}; |
||||
enum stream_reject { |
||||
STREAM_REASON_STOPPED = 1, STREAM_REASON_POSITION, STREAM_REASON_EXPIRED, STREAM_REASON_STALE, |
||||
STREAM_REASON_CONFLICT, STREAM_REASON_LIMIT, STREAM_REASON_LOOP, STREAM_REASON_TRANSPORT, |
||||
STREAM_REASON_LEASE, STREAM_REASON_MEMBER, STREAM_REASON_PROTOCOL |
||||
}; |
||||
struct stream_path { uint64_t nodes[STREAM_PATH_MAX]; uint8_t count; }; |
||||
struct stream_message { |
||||
struct stream_descriptor descriptor; |
||||
uint64_t sender_epoch, receiver_epoch, handle; |
||||
uint8_t command, subblock; |
||||
const uint8_t *payload; |
||||
size_t length; |
||||
}; |
||||
|
||||
/* Проверяет размер, версию, ненулевые ключи/epoch и известный command; payload разбирает диспетчер. */ |
||||
int stream_wire_parse(const uint8_t *wire, size_t length, struct stream_message *message); |
||||
void stream_wire_header(uint8_t *wire, const struct stream_message *message); |
||||
/* Путь без нулей/повторов, начинается источником и заканчивается отправителем; forbidden не входит в него. */ |
||||
int stream_path_validate(const struct stream_path *path, uint64_t source, uint64_t sender, uint64_t forbidden); |
||||
int stream_path_parse(const uint8_t *wire, size_t length, struct stream_path *path, size_t *used); |
||||
size_t stream_path_write(uint8_t *wire, const struct stream_path *path); |
||||
int stream_end_sign(const struct stream_descriptor *descriptor, uint64_t last, const uint8_t key[32], uint8_t wire[STREAM_END_SIZE]); |
||||
int stream_end_verify(const struct stream_descriptor *descriptor, const uint8_t key[32], const uint8_t *wire, size_t length, |
||||
uint64_t *last); |
||||
#endif |
||||
Loading…
Reference in new issue