From 55f7bb3eef63a536f4b171e323a00db56efd4c40 Mon Sep 17 00:00:00 2001 From: Quentin Dufour Date: Tue, 27 Aug 2019 14:31:27 +0200 Subject: [PATCH] Implement pad --- src/algo_thunder.c | 33 +++++++++++++++++++++++++-------- src/packet.c | 14 +++++++++----- src/packet.h | 12 +++++++----- 3 files changed, 41 insertions(+), 18 deletions(-) diff --git a/src/algo_thunder.c b/src/algo_thunder.c index 2facadf..0573e9f 100644 --- a/src/algo_thunder.c +++ b/src/algo_thunder.c @@ -5,6 +5,9 @@ #include "proxy.h" #include "timer.h" +// A Tor cell size is 512 bytes but handle only 498 bytes of data +#define TOR_CELL_SIZE 498 + struct thunder_ctx { uint16_t recv_id; uint16_t emit_id; @@ -21,7 +24,7 @@ uint64_t compute_delta(struct timespec* prev_time, uint64_t max) { int secs, nsecs; uint64_t mili_sec; - // 4. We compute the time difference + // 1. We compute the time difference if (clock_gettime(CLOCK_MONOTONIC, &curr) == -1){ perror("clock_gettime error"); exit(EXIT_FAILURE); @@ -39,9 +42,13 @@ void prepare(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct bu struct algo_ctx* app_ctx = fdinfo->cat->app_ctx; struct thunder_ctx* thunderc = app_ctx->misc; + // 1. Put the raw buffer in cache + thunderc->emit_id++; + uint64_t ref = 0l + thunderc->emit_id; + dup_buffer_toa (&app_ctx->br, bp, (void *)ref); + uint64_t delta_pkt = compute_delta(&thunderc->prev_packet_time, 200); - thunderc->emit_id++; union abstract_packet metadata = { .fmt.headers.cmd = CMD_UDP_METADATA_THUNDER, .fmt.headers.size = sizeof(metadata.fmt.headers) + sizeof(metadata.fmt.content.udp_metadata_thunder), @@ -57,18 +64,28 @@ void pad(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer struct thunder_ctx* thunderc = app_ctx->misc; uint64_t ref = 0l + thunderc->emit_id; - dup_buffer_toa (&app_ctx->br, bp, (void *)ref); - + // 1. Clean old buffers if (ref > 10 && get_app_buffer (&app_ctx->br, (void *)ref - 10l)) { mv_buffer_atof (&app_ctx->br, (void *)ref - 10l); } - //@FIXME we must add the delta t of the current packet - for (uint64_t add_ref = ref; add_ref >= 0 && get_app_buffer (&app_ctx->br, (void *)add_ref); add_ref--) { - struct buffer_packet *bpa = get_app_buffer (&app_ctx->br, (void *)add_ref); + // 2. Append abstract packets stored in our buffers + uint64_t add_ref = ref; + while(1) { + if (add_ref < 1) break; + add_ref--; + struct buffer_packet *bp_iter = get_app_buffer (&app_ctx->br, (void *)add_ref); + if (bp_iter == NULL) break; + union abstract_packet *ap = buffer_first_ap (bp_iter); + if (ap->fmt.headers.cmd != CMD_UDP_ENCAPSULATED || ap->fmt.headers.flags & FLAG_READ_NEXT) { + fprintf(stderr, "Invalid buffer!\n"); + exit(EXIT_FAILURE); + } + if (buffer_full_size (bp) + ap->fmt.headers.size > TOR_CELL_SIZE - thunderc->monit_pkt_size) break; + + buffer_append_ap (bp, ap); } - } int schedule(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) { diff --git a/src/packet.c b/src/packet.c index 85989a7..1727fc8 100644 --- a/src/packet.c +++ b/src/packet.c @@ -7,8 +7,12 @@ union abstract_packet* ap_next(union abstract_packet* ap) { return NULL; } +union abstract_packet* buffer_first_ap(struct buffer_packet* bp) { + return (union abstract_packet*) &bp->ip; +} + union abstract_packet* buffer_last_ap(struct buffer_packet* bp) { - union abstract_packet* ap = (union abstract_packet*) &bp->ip, *apn = NULL; + union abstract_packet* ap = buffer_first_ap (bp), *apn = NULL; while ((apn = ap_next(ap)) != NULL) ap = apn; return ap; @@ -28,7 +32,7 @@ size_t buffer_count_ap(struct buffer_packet* bp) { return s; } -size_t get_full_size(struct buffer_packet* bp) { +size_t buffer_full_size(struct buffer_packet* bp) { return &(buffer_free_ap (bp))->raw - &bp->ip[0]; } @@ -93,8 +97,8 @@ enum FD_STATE write_packet_to_tcp(struct evt_core_fdinfo* fdinfo, struct buffer_ //dump_buffer_packet (bp); if (bp->mode != BP_WRITING) return FDS_ERR; - while (bp->awrite < get_full_size(bp)) { - nwrite = send(fdinfo->fd, &(ap->raw) + bp->awrite, get_full_size(bp) - bp->awrite, 0); + while (bp->awrite < buffer_full_size(bp)) { + nwrite = send(fdinfo->fd, &(ap->raw) + bp->awrite, buffer_full_size(bp) - bp->awrite, 0); if (nwrite == -1 && errno == EAGAIN) return FDS_AGAIN; if (nwrite == -1) return FDS_ERR; bp->awrite += nwrite; @@ -190,7 +194,7 @@ enum FD_STATE read_packet_from_udp (struct evt_core_fdinfo* fdinfo, struct buffe void dump_buffer_packet(struct buffer_packet* bp) { printf("\n"); - printf(" mode=%d, aread=%d, awrite=%d, ap_count=%d, usage=%ld/%ld\n", bp->mode, bp->aread, bp->awrite, bp->ap_count, get_full_size (bp), sizeof(bp->ip)); + printf(" mode=%d, aread=%d, awrite=%d, ap_count=%d, usage=%ld/%ld\n", bp->mode, bp->aread, bp->awrite, bp->ap_count, buffer_full_size (bp), sizeof(bp->ip)); union abstract_packet* ap = (union abstract_packet*) &bp->ip; for (int i = 0; i < bp->ap_count; i++) { dump_abstract_packet(ap); diff --git a/src/packet.h b/src/packet.h index 6b1530f..94216d3 100644 --- a/src/packet.h +++ b/src/packet.h @@ -32,9 +32,9 @@ enum BP_MODE { }; enum PKT_CMD { - CMD_UDP_ENCAPSULATED, - CMD_LINK_MONITORING_THUNDER, - CMD_UDP_METADATA_THUNDER, + CMD_UDP_ENCAPSULATED = 1, + CMD_LINK_MONITORING_THUNDER = 2, + CMD_UDP_METADATA_THUNDER = 3, }; enum PKT_FLAGS { @@ -89,8 +89,10 @@ struct udp_target { size_t get_full_size(struct buffer_packet* bp); union abstract_packet* buffer_append_ap(struct buffer_packet* bp, union abstract_packet* ap); -union abstract_packet* buffer_free_ptr(struct buffer_packet* bp); -union abstract_packet* buffer_last_ptr(struct buffer_packet* bp); +union abstract_packet* buffer_free_ap(struct buffer_packet* bp); +union abstract_packet* buffer_first_ap(struct buffer_packet* bp); +union abstract_packet* buffer_last_ap(struct buffer_packet* bp); +size_t buffer_full_size(struct buffer_packet* bp); union abstract_packet* ap_next(union abstract_packet* ap); enum FD_STATE read_packet_from_tcp(struct evt_core_fdinfo* fd, struct buffer_packet* bp);