#define _POSIX_C_SOURCE 200809L #include "dat_session.h" #include "sleep.h" #include "sleep_crypto.h" #include #include #include #include #include #include typedef struct send_capture { uint8_t peer[DAT_PEER_ID_BYTES]; uint8_t frame[4096]; size_t length; uint8_t history[32][4096]; size_t history_length[32]; unsigned calls; } send_capture_t; static int capture_send(void *context, const uint8_t peer_id[DAT_PEER_ID_BYTES], const uint8_t *frame, size_t length) { send_capture_t *capture = context; assert(length <= sizeof(capture->frame)); memcpy(capture->peer, peer_id, sizeof(capture->peer)); memcpy(capture->frame, frame, length); capture->length = length; if (capture->calls < 32) { memcpy(capture->history[capture->calls], frame, length); capture->history_length[capture->calls] = length; } capture->calls++; return 0; } static void count_peer(void *context, const dat_session_peer_t *peer) { unsigned *count = context; assert(peer->state == DAT_SESSION_PEER_DISCOVERED); (*count)++; } static void stream_transfer(dat_session_t *receiver, const uint8_t peer_id[DAT_PEER_ID_BYTES], int source_fd, int receiver_fd, const uint8_t *frame, size_t frame_len) { size_t sent = 0, received = 0; uint8_t incoming[8192]; while (sent < frame_len) { ssize_t n = write(source_fd, frame + sent, frame_len - sent); assert(n > 0); sent += (size_t)n; } while (received < frame_len) { ssize_t n = read(receiver_fd, incoming + received, frame_len - received); assert(n > 0); received += (size_t)n; } assert(dat_session_receive(receiver, peer_id, incoming, received) == 0); } #ifndef DAT_LEGACY_HYPERCORE static int count_poll(void *context, int timeout_ms) { unsigned *count = context; assert(timeout_ms == 0); (*count)++; return 0; } #endif #ifdef DAT_LEGACY_HYPERCORE static void legacy_session_handshake(void) { cesspool_t pool; dat_session_t session; sleep_feed_t metadata, content; sleep_keypair_t metadata_keys, content_keys; send_capture_t capture; cesspool_peer_t discovered; dat_session_peer_t *peer; dat_feed_open_t feed; dat_handshake_t handshake, decoded; dat_wire_message_t wire; uint8_t archive_key[32] = {1}, local_id[32] = {2}, remote_id[32] = {3}; uint8_t remote_nonce[24] = {4}, payload[256], encoded[512], plain[512]; uint8_t frame[512]; size_t payload_len, encoded_len, content_frame_len, frame_len, consumed; unsigned discovered_count = 0; char feed_dir[] = "/tmp/datgpt-v6-XXXXXX"; memset(&capture, 0, sizeof(capture)); assert(mkdtemp(feed_dir) != NULL); assert(sleep_gen_keypair(&metadata_keys) == 0); assert(sleep_gen_keypair(&content_keys) == 0); assert(sleep_feed_open(&metadata, feed_dir, "metadata", &metadata_keys) == 0); assert(sleep_feed_open(&content, feed_dir, "content", &content_keys) == 0); assert(cesspool_init(&pool, local_id) == 0); assert(cesspool_listen(&pool, "127.0.0.1", 0) == 0); assert(cesspool_add_bootstrap(&pool, "127.0.0.1", 9) == 0); assert(dat_session_init(&session, &pool, archive_key, local_id, pool.port, &metadata, metadata_keys.pubkey, &content, content_keys.pubkey, capture_send, &capture) == 0); assert(dat_session_set_peer_callback(&session, count_peer, &discovered_count) == 0); assert(dat_session_start(&session, 1, 1) == 0); memset(&discovered, 0, sizeof(discovered)); memcpy(discovered.public_key, remote_id, 32); memcpy(discovered.discovery_topic, session.discovery_topic, 32); discovered.discovered = 1; assert(dat_session_peer_discovered(&session, &discovered) == 0); assert(discovered_count == 1); assert(dat_session_peer_secure(&session, remote_id, 1) == 0); assert(capture.calls == 3); peer = dat_session_find_peer(&session, remote_id); assert(peer && peer->tx_stream_encrypted && peer->tx_offset > 0); assert(dat_wire_decode(&wire, capture.history[0], capture.history_length[0], &consumed) == 0); assert(wire.channel == 0 && wire.type == DAT_MSG_FEED); assert(dat_feed_open_decode(&feed, wire.payload, wire.payload_len) == 0 && feed.has_nonce); assert(sleep_stream_xsalsa20_xor(plain, capture.history[1], capture.history_length[1], feed.nonce, metadata_keys.pubkey, 0) == 0); assert(dat_wire_decode(&wire, plain, capture.history_length[1], &consumed) == 0); assert(wire.channel == 0 && wire.type == DAT_MSG_HANDSHAKE); assert(dat_handshake_decode(&decoded, wire.payload, wire.payload_len) == 0); assert(decoded.has_peer_id); assert(sleep_stream_xsalsa20_xor(plain, capture.history[2], capture.history_length[2], feed.nonce, metadata_keys.pubkey, capture.history_length[1]) == 0); assert(dat_wire_decode(&wire, plain, capture.history_length[2], &consumed) == 0); assert(wire.channel == 1 && wire.type == DAT_MSG_FEED); /* The first Feed carries the responder nonce in clear; all following bytes stream-cipher. */ memset(&feed, 0, sizeof(feed)); assert(sleep_compute_discovery_key(metadata_keys.pubkey, feed.discovery_key) == 0); memcpy(feed.nonce, remote_nonce, sizeof(remote_nonce)); feed.has_nonce = 1; assert(dat_feed_open_encode(&feed, payload, sizeof(payload), &payload_len) == 0); wire.channel = 0; wire.type = DAT_MSG_FEED; wire.payload = payload; wire.payload_len = payload_len; assert(dat_wire_encode(&wire, frame, sizeof(frame), &frame_len) == 0); assert(dat_session_receive(&session, remote_id, frame, frame_len / 2) == 0); assert(peer->state == DAT_SESSION_PEER_SECURE); assert(dat_session_receive(&session, remote_id, frame + frame_len / 2, frame_len - frame_len / 2) == 0); assert(dat_handshake_init(&handshake, remote_id, 0) == 0); assert(dat_handshake_encode(&handshake, payload, sizeof(payload), &payload_len) == 0); wire.channel = 0; wire.type = DAT_MSG_HANDSHAKE; wire.payload = payload; wire.payload_len = payload_len; assert(dat_wire_encode(&wire, encoded, sizeof(encoded), &encoded_len) == 0); assert(sleep_compute_discovery_key(content_keys.pubkey, feed.discovery_key) == 0); feed.has_nonce = 0; assert(dat_feed_open_encode(&feed, payload, sizeof(payload), &payload_len) == 0); wire.channel = 1; wire.type = DAT_MSG_FEED; wire.payload = payload; wire.payload_len = payload_len; assert(dat_wire_encode(&wire, encoded + encoded_len, sizeof(encoded) - encoded_len, &content_frame_len) == 0); encoded_len += content_frame_len; assert(sleep_stream_xsalsa20_xor(frame, encoded, encoded_len, remote_nonce, metadata_keys.pubkey, 0) == 0); assert(dat_session_receive(&session, remote_id, frame, encoded_len) == 0); assert(peer->state == DAT_SESSION_PEER_READY && peer->rx_offset == encoded_len && capture.calls == 3); assert(dat_session_peer_down(&session, remote_id) == 0); assert(dat_session_peer_discovered(&session, &discovered) == 0); assert(discovered_count == 2); assert(dat_session_close(&session) == 0); assert(cesspool_destroy(&pool) == 0); sleep_feed_close(&metadata); sleep_feed_close(&content); } #endif #ifndef DAT_LEGACY_HYPERCORE static void receive_standard_data(dat_session_t *source, cesspool_t *pool, const send_capture_t *source_capture, const uint8_t archive_key[32], const uint8_t source_id[32], const uint8_t remote_id[32], uint8_t feed_keys[2][32]) { dat_session_t receiver; sleep_feed_t metadata_replica, content_replica, reopened_replica; send_capture_t receiver_capture; int p2p[2]; cesspool_peer_t discovered; uint8_t receiver_id[32] = {9}, frame[8192], payload[8192]; uint8_t handshake_bytes[128], open_bytes[128], request_bytes[64]; uint8_t *data_bytes; dat_handshake_t handshake; dat_feed_open_t feed_open; dat_wire_message_t wire; dat_range_t range = {0, 3}; size_t frame_len, payload_len, consumed, handshake_len, open_len, data_length; unsigned feed; char replica_dir[] = "/tmp/datgpt-replica-XXXXXX"; memset(&metadata_replica, 0, sizeof(metadata_replica)); memset(&content_replica, 0, sizeof(content_replica)); memset(&receiver_capture, 0, sizeof(receiver_capture)); assert(mkdtemp(replica_dir) != NULL); assert(socketpair(AF_UNIX, SOCK_STREAM, 0, p2p) == 0); assert(sleep_feed_open_remote(&metadata_replica, replica_dir, "metadata-replica", feed_keys[0]) == 0); assert(sleep_feed_open_remote(&content_replica, replica_dir, "content-replica", feed_keys[1]) == 0); assert(dat_session_init(&receiver, pool, archive_key, receiver_id, pool->port, &metadata_replica, feed_keys[0], &content_replica, feed_keys[1], capture_send, &receiver_capture) == 0); assert(dat_session_start(&receiver, 0, 1) == 0); memset(&discovered, 0, sizeof(discovered)); memcpy(discovered.public_key, source_id, 32); memcpy(discovered.discovery_topic, receiver.discovery_topic, 32); discovered.discovered = 1; assert(dat_session_peer_discovered(&receiver, &discovered) == 0); assert(dat_session_peer_secure(&receiver, source_id, 1) == 0); for (feed = 0; feed < 2; feed++) { assert(sleep_compute_discovery_key(feed_keys[feed], feed_open.discovery_key) == 0); feed_open.has_nonce = 0; assert(dat_feed_open_encode(&feed_open, open_bytes, sizeof(open_bytes), &open_len) == 0); wire.channel = feed + 1; wire.type = DAT_MSG_FEED; wire.payload = open_bytes; wire.payload_len = open_len; assert(dat_wire_encode(&wire, frame, sizeof(frame), &frame_len) == 0); stream_transfer(&receiver, source_id, p2p[0], p2p[1], frame, frame_len); } assert(dat_handshake_init(&handshake, source_id, 0) == 0); assert(dat_handshake_encode(&handshake, handshake_bytes, sizeof(handshake_bytes), &handshake_len) == 0); wire.channel = 0; wire.type = DAT_MSG_HANDSHAKE; wire.payload = handshake_bytes; wire.payload_len = handshake_len; assert(dat_wire_encode(&wire, frame, sizeof(frame), &frame_len) == 0); assert(dat_session_receive(&receiver, source_id, frame, frame_len) == 0); assert(dat_range_encode(&range, request_bytes, sizeof(request_bytes), &payload_len) == 0); wire.channel = 2; wire.type = DAT_MSG_HAVE; wire.payload = request_bytes; wire.payload_len = payload_len; assert(dat_wire_encode(&wire, frame, sizeof(frame), &frame_len) == 0); assert(dat_session_receive(&receiver, source_id, frame, frame_len) == 0); assert(dat_session_find_peer(&receiver, source_id)->replication[DAT_SESSION_CONTENT].remote_length == 3); for (feed = 0; feed < 2; feed++) { dat_data_t sent, legacy; size_t block_index = feed == 0 ? 1u : 0u; assert(dat_session_send_block(source, remote_id, DAT_SESSION_CONTENT, block_index) == 0); assert(dat_wire_decode(&wire, source_capture->frame, source_capture->length, &consumed) == 0); assert(wire.type == DAT_MSG_DATA && dat_data_decode(&sent, wire.payload, wire.payload_len) == 0); assert(sent.index == block_index); assert(sent.proof_len && sent.has_signature && sent.node_count); legacy = sent; legacy.proof = NULL; legacy.proof_len = 0; assert(dat_data_encode(&legacy, payload, sizeof(payload), &payload_len) == 0); wire.channel = 2; wire.type = DAT_MSG_DATA; wire.payload = payload; wire.payload_len = payload_len; assert(dat_wire_encode(&wire, frame, sizeof(frame), &frame_len) == 0); stream_transfer(&receiver, source_id, p2p[0], p2p[1], frame, frame_len); } for (feed = 0; feed < 2; feed++) { uint8_t *block; size_t block_len; assert(sleep_feed_read(&content_replica, feed, &block, &block_len) == 0); assert(block_len == (feed == 0 ? 4u : 3u)); assert(!memcmp(block, feed == 0 ? "zero" : "one", block_len)); free(block); } assert(content_replica.remote_only); puts("live peer byte-stream block transfer: PASS"); assert(dat_session_send_block(&receiver, source_id, DAT_SESSION_CONTENT, 0) == 0); assert(dat_wire_decode(&wire, receiver_capture.frame, receiver_capture.length, &consumed) == 0 && wire.type == DAT_MSG_DATA); { dat_data_t forwarded; sleep_proof_t proof; assert(dat_data_decode(&forwarded, wire.payload, wire.payload_len) == 0); assert(forwarded.proof_len); assert(sleep_proof_decode(&proof, forwarded.proof, forwarded.proof_len) == 0); assert(sleep_proof_verify(&proof, forwarded.value, forwarded.value_len, feed_keys[1]) == 0); } assert(dat_session_close(&receiver) == 0); close(p2p[0]); close(p2p[1]); sleep_feed_close(&metadata_replica); sleep_feed_close(&content_replica); assert(sleep_feed_open_remote(&reopened_replica, replica_dir, "content-replica", feed_keys[1]) == 0); assert(sleep_feed_has(&reopened_replica, 0) && sleep_feed_has(&reopened_replica, 1)); assert(sleep_feed_read(&reopened_replica, 1, &data_bytes, &data_length) == 0); assert(data_length == 3 && !memcmp(data_bytes, "one", 3)); free(data_bytes); sleep_feed_close(&reopened_replica); { char cache_path[1200]; FILE *cache; int byte; snprintf(cache_path, sizeof(cache_path), "%s/content-replica.remote", replica_dir); cache = fopen(cache_path, "r+b"); assert(cache && fseek(cache, -1, SEEK_END) == 0); byte = fgetc(cache); assert(byte != EOF && fseek(cache, -1, SEEK_CUR) == 0); assert(fputc(byte ^ 1, cache) != EOF && fclose(cache) == 0); assert(sleep_feed_open_remote(&reopened_replica, replica_dir, "content-replica", feed_keys[1]) != 0); } } #endif int main(void) { #ifdef DAT_LEGACY_HYPERCORE legacy_session_handshake(); return 0; #else cesspool_t pool; dat_session_t session; sleep_feed_t metadata, content; sleep_keypair_t metadata_keys, content_keys; send_capture_t capture; uint8_t archive_key[32] = {1}, local_id[32] = {2}, remote_id[32] = {3}; uint8_t feed_key[2][32], handshake_bytes[72], frame[256], request_bytes[64]; dat_handshake_t handshake; dat_feed_open_t remote_feed; dat_wire_message_t wire; dat_request_t request; cesspool_peer_t discovered; size_t handshake_len, frame_len, consumed, request_len; char feed_dir[] = "/tmp/datgpt-session-XXXXXX"; unsigned discovered_count = 0; unsigned poll_count = 0; static const uint8_t mblock[] = "metadata"; static const uint8_t cblocks[][8] = {"zero", "one", "two"}; memset(&capture, 0, sizeof(capture)); assert(mkdtemp(feed_dir) != NULL); assert(sleep_gen_keypair(&metadata_keys) == 0); assert(sleep_gen_keypair(&content_keys) == 0); memcpy(feed_key[0], metadata_keys.pubkey, 32); memcpy(feed_key[1], content_keys.pubkey, 32); assert(sleep_feed_open(&metadata, feed_dir, "metadata", &metadata_keys) == 0); assert(sleep_feed_open(&content, feed_dir, "content", &content_keys) == 0); assert(sleep_feed_append(&metadata, mblock, sizeof(mblock) - 1) == 0); assert(sleep_feed_append(&content, cblocks[0], 4) == 0); assert(sleep_feed_append(&content, cblocks[1], 3) == 0); assert(sleep_feed_append(&content, cblocks[2], 3) == 0); assert(sleep_feed_flush(&metadata) == 0); assert(sleep_feed_flush(&content) == 0); assert(cesspool_init(&pool, local_id) == 0); assert(cesspool_listen(&pool, "127.0.0.1", 0) == 0); assert(cesspool_add_bootstrap(&pool, "127.0.0.1", 9) == 0); assert(dat_session_init(&session, &pool, archive_key, local_id, 3282, &metadata, feed_key[0], &content, feed_key[1], capture_send, &capture) == 0); assert(dat_session_set_peer_callback(&session, count_peer, &discovered_count) == 0); assert(dat_session_set_transport_poll(&session, count_poll, &poll_count) == 0); assert(dat_session_start(&session, 1, 1) == 0); assert(session.state == DAT_SESSION_DISCOVERING); memset(&discovered, 0, sizeof(discovered)); memcpy(discovered.public_key, remote_id, sizeof(remote_id)); memcpy(discovered.discovery_topic, session.discovery_topic, sizeof(discovered.discovery_topic)); discovered.discovered = 1; assert(cesspool_add_peer(&pool, &discovered) == 0); assert(dat_session_peer_discovered(&session, &discovered) == 0); assert(dat_session_tick(&session, 1) == 0); assert(poll_count == 1); assert(dat_session_find_peer(&session, remote_id) != NULL); assert(discovered_count == 1); assert(dat_session_peer_secure(&session, remote_id, 1) == 0); assert(capture.calls == 3); for (unsigned feed_index = 0; feed_index < 2; feed_index++) { assert(sleep_compute_discovery_key(feed_key[feed_index], remote_feed.discovery_key) == 0); remote_feed.has_nonce = 0; assert(dat_feed_open_encode(&remote_feed, request_bytes, sizeof(request_bytes), &request_len) == 0); wire.channel = feed_index + 1; wire.type = DAT_MSG_FEED; wire.payload = request_bytes; wire.payload_len = request_len; assert(dat_wire_encode(&wire, frame, sizeof(frame), &frame_len) == 0); assert(dat_session_receive(&session, remote_id, frame, frame_len) == 0); } assert(dat_handshake_init(&handshake, remote_id, 7) == 0); assert(dat_handshake_encode(&handshake, handshake_bytes, sizeof(handshake_bytes), &handshake_len) == 0); wire.channel = 0; wire.type = DAT_MSG_HANDSHAKE; wire.payload = handshake_bytes; wire.payload_len = handshake_len; assert(dat_wire_encode(&wire, frame, sizeof(frame), &frame_len) == 0); assert(dat_session_receive(&session, remote_id, frame, frame_len / 2) == 0); assert(dat_session_receive(&session, remote_id, frame + frame_len / 2, frame_len - frame_len / 2) == 0); assert(session.state == DAT_SESSION_ACTIVE); assert(dat_session_find_peer(&session, remote_id)->state == DAT_SESSION_PEER_READY); assert(capture.calls == 5); assert(dat_wire_decode(&wire, capture.frame, capture.length, &consumed) == 0); assert(wire.type == DAT_MSG_HAVE && wire.channel == 2); { dat_range_t have; assert(dat_range_decode(&have, wire.payload, wire.payload_len) == 0); assert(have.start == 0 && have.length == 3); } { dat_range_t want = {0, 3}; assert(dat_range_encode(&want, request_bytes, sizeof(request_bytes), &request_len) == 0); wire.channel = 2; wire.type = DAT_MSG_WANT; wire.payload = request_bytes; wire.payload_len = request_len; assert(dat_wire_encode(&wire, frame, sizeof(frame), &frame_len) == 0); assert(dat_session_receive(&session, remote_id, frame, frame_len) == 0); assert(capture.calls == 6); assert(dat_wire_decode(&wire, capture.frame, capture.length, &consumed) == 0); assert(wire.channel == 2 && wire.type == DAT_MSG_HAVE); } assert(dat_session_request_block(&session, remote_id, DAT_SESSION_CONTENT, 12, 4096) == 0); assert(capture.calls == 7); assert(dat_wire_decode(&wire, capture.frame, capture.length, &consumed) == 0); assert(consumed == capture.length && wire.channel == 2 && wire.type == DAT_MSG_REQUEST); assert(dat_request_decode(&request, wire.payload, wire.payload_len) == 0); assert(request.index == 12 && request.bytes == 4096 && request.want_hash && request.want_nodes); request.index = 1; request.bytes = 0; request.want_hash = request.want_nodes = 1; assert(dat_request_encode(&request, request_bytes, sizeof(request_bytes), &request_len) == 0); wire.channel = 2; wire.type = DAT_MSG_REQUEST; wire.payload = request_bytes; wire.payload_len = request_len; assert(dat_wire_encode(&wire, frame, sizeof(frame), &frame_len) == 0); assert(dat_session_receive(&session, remote_id, frame, frame_len) == 0); assert(capture.calls == 8); assert(dat_wire_decode(&wire, capture.frame, capture.length, &consumed) == 0); assert(wire.type == DAT_MSG_DATA && wire.channel == 2); { dat_data_t data; sleep_proof_t proof; assert(dat_data_decode(&data, wire.payload, wire.payload_len) == 0); assert(data.index == 1 && data.value_len == 3 && !memcmp(data.value, "one", 3)); assert(data.has_signature && data.node_count > 0); assert(sleep_proof_decode(&proof, data.proof, data.proof_len) == 0); assert(sleep_proof_verify(&proof, data.value, data.value_len, content_keys.pubkey) == 0); { sleep_feed_t replica; sleep_proof_t first_proof, forwarded; uint8_t proof_bytes[8192], forwarded_bytes[8192]; size_t proof_len, forwarded_len; memset(&replica, 0, sizeof(replica)); strcpy(replica.directory, feed_dir); strcpy(replica.name, "content-replica"); assert(sleep_proof_create(&content, 0, &first_proof) == 0); assert(sleep_proof_encode(&first_proof, proof_bytes, sizeof(proof_bytes), &proof_len) == 0); assert(sleep_feed_append_remote(&replica, 0, cblocks[0], 4, proof_bytes, proof_len) == 0); assert(sleep_feed_append_remote(&replica, 1, data.value, data.value_len, data.proof, data.proof_len) == 0); assert(sleep_proof_create(&replica, 1, &forwarded) == 0); assert(sleep_proof_encode(&forwarded, forwarded_bytes, sizeof(forwarded_bytes), &forwarded_len) == 0); assert(forwarded_len == data.proof_len && !memcmp(forwarded_bytes, data.proof, forwarded_len)); assert(sleep_proof_verify(&forwarded, data.value, data.value_len, content_keys.pubkey) == 0); assert(sleep_feed_flush(&replica) != 0); sleep_feed_close(&replica); } receive_standard_data(&session, &pool, &capture, archive_key, local_id, remote_id, feed_key); ((uint8_t *)data.value)[0] ^= 1; assert(sleep_proof_verify(&proof, data.value, data.value_len, content_keys.pubkey) != 0); } assert(dat_session_peer_down(&session, remote_id) == 0); assert(session.state == DAT_SESSION_DISCOVERING); assert(dat_session_peer_discovered(&session, &discovered) == 0); assert(discovered_count == 2); assert(dat_session_close(&session) == 0); assert(session.state == DAT_SESSION_CLOSED); assert(cesspool_destroy(&pool) == 0); sleep_feed_close(&metadata); sleep_feed_close(&content); return 0; #endif }