initial commit

meson.build

@@ -0,0 +1,20 @@
+project('floof store-and-forward library', 'c', 'cpp',
+  default_options : ['c_std=c17', 'cpp_std=c++17'],
+  version : '0.2.0'
+  license : 'Apache-2.0',
+  meson_version : '>=0.50')
+
+threads = dependency('threads')
+glib = dependency('glib-2.0')
+rocksdb = dependency('rocksdb')
+sodium = dependency('libsodium')
+
+subdir('src')
+
+pkg_mod = import('pkgconfig')
+pkg_mod.generate(
+  libraries : [libfloof],
+  version   : meson.project_version(),
+  name      : 'libfloof',
+  filebase  : 'floof',
+  description : 'floof store-and-forward library')

shell.nix

@@ -0,0 +1,14 @@
+{ pkgs ? import <nixpkgs> {} }:
+  pkgs.mkShell {
+    nativeBuildInputs = [
+      pkgs.meson
+      pkgs.ninja
+      pkgs.pkg-config
+    ];
+
+    buildInputs = [
+      pkgs.glib
+      pkgs.rocksdb
+      pkgs.libsodium
+    ];
+  }

src/floof.c

@@ -0,0 +1,139 @@
+#include <stdbool.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <netinet/sctp.h>
+#include <arpa/inet.h>
+
+#include <glib.h>
+#include <sodium.h>
+
+#include "floof_int.h"
+#include "floof_db.h"
+
+typedef struct {
+  floof_db_t *db;
+  GThreadPool *thpool;
+  bool (*filter_packets)(unsigned char *data, size_t datalen);
+} floof_node_t;
+
+typedef struct {
+  floof_node_t * node;
+  int sock;
+} floof_conn_t;
+
+typedef struct {
+  unsigned char pubkey[crypto_sign_PUBLICKEYBYTES];
+  unsigned char sig[crypto_sign_BYTES];
+  unsigned char *data;
+  size_t datalen;
+} floof_worker_task_t;
+
+/** internal **/
+
+static void floof_free_task(void * data) {
+  floof_worker_task_t * task = (floof_worker_task_t*) data;
+
+  free(task->data);
+  free(task);
+}
+
+static void floof_worker(void * data, void * user_data) {
+  floof_worker_task_t * task = (floof_worker_task_t*) data;
+  floof_node_t * node = (floof_node_t*) user_data;
+
+  if(0 != crypto_sign_verify_detached(task->sig, task->data, task->datalen, task->pubkey)) {
+    // signature check failed, drop packet
+    goto drop;
+  }
+
+  if(!node->filter_packets(task->data, task->datalen)) {
+    // drop packet
+    goto drop;
+  }
+
+  // forward packet
+  floof_db_insert_packet(node->db, task);
+  goto cleanup;
+
+drop:
+
+cleanup:
+  floof_free_task(data);
+}
+
+/** external **/
+
+floof_node_t * floof_new_node(const char * spooldir) {
+  floof_node_t * ret = calloc(sizeof(floof_node_t), 1);
+  if(!ret) return 0;
+
+  const unsigned int cpu_count = floof_get_cpu_count();
+  ret->thpool = g_thread_pool_new(
+    floof_worker,
+    (void*) ret,
+    g_get_num_processors(),
+    FALSE,
+    0);
+
+  if(floof_db_open(&ret->db, spooldir) == -1)
+    goto error;
+
+  return ret;
+
+error:
+  floof_destroy_node(ret);
+  return 0;
+}
+
+floof_conn_t * floof_socket(floof_node_t * node, int domain) {
+  floof_conn_t * ret = calloc(sizeof(floof_conn_t), 1);
+  if(!ret) return 0;
+
+  ret->node = node;
+
+  if(-1 == (ret->sock = socket(domain, SOCK_SEQPACKET, IPPROTO_SCTP))) {
+    free(ret);
+    return 0;
+  }
+
+  return ret;
+}
+
+int floof_bind_listen(floof_conn_t * conn, struct sockaddr *addr, socklen_t addrlen) {
+  if(bind(conn->sock, addr, addrlen) < 0)
+    return -1;
+
+  if(listen(conn->sock, 10) < 0)
+    return -1;
+
+  return 0;
+}
+
+int floof_connect(floof_node_t * conn, struct sockaddr *remotename, socklen_t addrlen) {
+  if(connect(conn->sock, remotename, addrlen) < 0)
+    return -1;
+
+  return 0;
+}
+
+void floof_close(floof_conn_t * conn) {
+  if(!conn) return;
+  if(conn->sock != -1)
+    close(conn->sock);
+  free(conn);
+}
+
+void floof_destroy_node(floof_node_t * node) {
+  if(!node) return;
+
+  if(node->thpool)
+    g_thread_pool_free(node->thpool, TRUE, TRUE);
+
+  floof_db_close(&node->db);
+
+  free(node);
+}

src/floof_db.cxx

@@ -0,0 +1,50 @@
+#include <string.h>
+
+#include <rocksdb/db.h>
+#include <rocksdb/slice.h>
+#include <rocksdb/write_batch.h>
+
+#include <sodium.h>
+#include <glib.h>
+
+#include <string>
+
+typedef rocksdb::DB floof_db_t;
+
+int floof_db_open(floof_db_t ** db, const char * path) {
+  rocksdb::Options options;
+  options.create_if_missing = true;
+  rocksdb::Status status;
+
+  status = rocksdb::DB::Open(options, path, &*db);
+  if(status.ok()) {
+    return 0;
+  } else {
+    return -1;
+  }
+}
+
+void floof_db_close(floof_db_t ** db) {
+  if(!*db) return;
+  delete (*db);
+  *db = 0;
+}
+
+void floof_db_insert_packet(floof_db_t * db, floof_worker_task_t * packet) {
+  rocksdb::WriteBatch batch;
+
+  unsigned char sigblock[crypto_sign_PUBLICKEYBYTES + crypto_sign_BYTES];
+  memcpy(&sigblock[0], packet->pubkey, crypto_sign_PUBLICKEYBYTES);
+  memcpy(&sigblock[crypto_sign_PUBLICKEYBYTES], packet->sig, crypto_sign_BYTES);
+
+  batch.Put(rocksdb::Slice(sigblock, crypto_sign_PUBLICKEYBYTES + crypto_sign_BYTES),
+    rocksdb::Slice(packet->data, packet->datalen));
+
+  rocksdb::Status status;
+  status = db->Write(rocksdb::WriteOptions(), &batch);
+
+  if(!status.ok()) {
+    const std::string tmp = status.ToString();
+    g_warning("Floof failed to insert packet into database: %s", tmp.c_str());
+  }
+}

src/floof_db.h

@@ -0,0 +1,6 @@
+#pragma once
+#include "floof_int.h"
+struct floof_db_t;
+int floof_db_open(floof_db_t ** db, const char * path);
+void floof_db_close(floof_db_t ** db);
+void floof_db_insert_packet(floof_db_t * db, floof_worker_task_t * packet);

src/floof_int.h

@@ -0,0 +1,10 @@
+#pragma once
+#include <stddef.h>
+#include <sodium.h>
+
+typedef struct {
+  unsigned char pubkey[crypto_sign_PUBLICKEYBYTES];
+  unsigned char sig[crypto_sign_BYTES];
+  unsigned char *data;
+  size_t datalen;
+} floof_worker_task_t;

src/meson.build

@@ -0,0 +1,12 @@
+install_headers('floof.h')
+
+libfloof_sources = files('floof.c', 'floof_db.cxx')
+
+libfloof = shared_library('floof',
+  libfloof_sources,
+  version : '0.0.0',
+  c_args : ['-fno-plt'],
+  cpp_args : ['-fno-plt', '-fno-rtti']
+  include_directories : include_directories('.')
+  dependencies : [threads, glib, rocksdb, sodium],
+  install : true)