aboutsummaryrefslogtreecommitdiffstats
path: root/link
diff options
context:
space:
mode:
authoreunhhu <[email protected]>2026-09-29 07:06:27 +0100
committereunhhu <[email protected]>2026-09-29 07:06:27 +0100
commita052b95c8135cdc5f9b91f8efc991cac80fbbe8b (patch)
tree2b7b0d7a0ccd5bc6ceaa603a01eb87deb6c1f2e8 /link
parent0b4a9f7225b5873e606106b8eaf9174b254c0c0f (diff)
downloadpure-a052b95c8135cdc5f9b91f8efc991cac80fbbe8b.tar.gz
pure-a052b95c8135cdc5f9b91f8efc991cac80fbbe8b.zip
Put the links and the crypto they use in link/, a part of their own
libpure-link.a holds them; pure-serve and the M5 harness link it instead of naming its source files.
Diffstat (limited to 'link')
-rw-r--r--link/crypto-openssl.c284
-rw-r--r--link/crypto.c167
-rw-r--r--link/crypto.h49
-rw-r--r--link/link.c1862
-rw-r--r--link/link.h118
-rw-r--r--link/seal.h80
-rw-r--r--link/tls.h18
7 files changed, 2578 insertions, 0 deletions
diff --git a/link/crypto-openssl.c b/link/crypto-openssl.c
new file mode 100644
index 0000000..6bdf6fb
--- /dev/null
+++ b/link/crypto-openssl.c
@@ -0,0 +1,284 @@
+#define _GNU_SOURCE
+/* pure-crypto.so: what pure-serve asks of OpenSSL (link/crypto.h), mapped
+ * only when a server seals, opens a seal or reaches an https receiver. It
+ * exports one table, pure_crypto. */
+#include "crypto.h"
+#include <arpa/inet.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <openssl/crypto.h>
+#include <openssl/evp.h>
+#include <openssl/core_names.h>
+#include <openssl/hpke.h>
+#include <openssl/kdf.h>
+#include <openssl/ssl.h>
+#include <openssl/x509v3.h>
+#include <limits.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+/* ---- TLS: one client context, verifying the receiver's certificate against
+ * the system's CAs or the file given, and its name. */
+static SSL_CTX *tls_context;
+
+static bool tls_start(const char *ca_file) {
+ SSL_CTX *context = SSL_CTX_new(TLS_client_method());
+ if (!context) return false;
+ if (!SSL_CTX_set_min_proto_version(context, TLS1_2_VERSION) ||
+ !(ca_file ? SSL_CTX_load_verify_locations(context, ca_file, NULL) : SSL_CTX_set_default_verify_paths(context))) {
+ SSL_CTX_free(context);
+ return false;
+ }
+ SSL_CTX_set_verify(context, SSL_VERIFY_PEER, NULL);
+ SSL_CTX_set_mode(context, SSL_MODE_AUTO_RETRY);
+ tls_context = context;
+ return true;
+}
+
+/* TLS over a connected socket: the receiver's certificate must name the host
+ * (a DNS name, or an IP address written as one). */
+static PureTls *tls_open(const char *host, int fd) {
+ SSL *ssl = SSL_new(tls_context);
+ if (!ssl) return NULL;
+ unsigned char address[16];
+ bool literal = inet_pton(AF_INET, host, address) == 1 || inet_pton(AF_INET6, host, address) == 1;
+ bool named = literal ? X509_VERIFY_PARAM_set1_ip_asc(SSL_get0_param(ssl), host) == 1 :
+ SSL_set_tlsext_host_name(ssl, host) == 1 && SSL_set1_host(ssl, host) == 1;
+ if (!named || SSL_set_fd(ssl, fd) != 1 || SSL_connect(ssl) != 1 || SSL_get_verify_result(ssl) != X509_V_OK) {
+ SSL_free(ssl);
+ return NULL;
+ }
+ return (PureTls *)ssl;
+}
+
+static int tls_read(PureTls *tls, void *bytes, int size) { return SSL_read((SSL *)tls, bytes, size); }
+static int tls_write(PureTls *tls, const void *bytes, int size) { return SSL_write((SSL *)tls, bytes, size); }
+static void tls_close(PureTls *tls) { SSL_free((SSL *)tls); }
+
+/* ---- Sealing: HPKE (RFC 9180), X25519 and HKDF-SHA256, with the AEAD the
+ * suite names; auth mode when the sender seals with a key of its own. A
+ * session is one encapsulation, its seals numbered by the context. */
+struct PureSealKey { EVP_PKEY *key; };
+struct PureSealer { OSSL_HPKE_CTX *context; };
+struct PureOpener { OSSL_HPKE_CTX *context; };
+
+static const unsigned char info[] = "pure sealed send 2";
+
+static bool suite_named(const char *name, OSSL_HPKE_SUITE *out) {
+ OSSL_HPKE_SUITE suite = {OSSL_HPKE_KEM_ID_X25519, OSSL_HPKE_KDF_ID_HKDF_SHA256, 0};
+ if (!strcmp(name, "x25519-sha256-aes128gcm")) suite.aead_id = OSSL_HPKE_AEAD_ID_AES_GCM_128;
+ else if (!strcmp(name, "x25519-sha256-chacha20poly1305")) suite.aead_id = OSSL_HPKE_AEAD_ID_CHACHA_POLY1305;
+ else return false;
+ *out = suite;
+ return true;
+}
+
+static pthread_once_t secure_once = PTHREAD_ONCE_INIT;
+static void secure_heap(void) { CRYPTO_secure_malloc_init(16384, 32); }
+
+static PureSealKey *seal_key(const char *path, char public_hex[2 * PURE_SEAL_KEY + 1], char *error, size_t size) {
+ pthread_once(&secure_once, secure_heap);
+ unsigned char *raw = OPENSSL_secure_malloc(PURE_SEAL_KEY), public_key[PURE_SEAL_KEY];
+ PureSealKey *made = calloc(1, sizeof(*made));
+ EVP_PKEY *key = NULL;
+ size_t length = PURE_SEAL_KEY;
+ if (!raw || !made) { snprintf(error, size, "out of memory"); goto fail; }
+ int fd = open(path, O_RDONLY | O_NOFOLLOW | O_CLOEXEC);
+ struct stat info_;
+ if (fd >= 0) {
+ bool ok = !fstat(fd, &info_) && S_ISREG(info_.st_mode) && !(info_.st_mode & 077) &&
+ read(fd, raw, PURE_SEAL_KEY) == PURE_SEAL_KEY && info_.st_size == PURE_SEAL_KEY;
+ close(fd);
+ if (!ok) { snprintf(error, size, "%s: a key is 32 bytes that only its owner may read (0600)", path); goto fail; }
+ key = EVP_PKEY_new_raw_private_key(EVP_PKEY_X25519, NULL, raw, PURE_SEAL_KEY);
+ } else if (errno == ENOENT) {
+ /* A new key: made, written whole and synced before it is used. */
+ key = EVP_PKEY_Q_keygen(NULL, NULL, "X25519");
+ fd = open(path, O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW | O_CLOEXEC, 0600);
+ bool ok = key && fd >= 0 && EVP_PKEY_get_raw_private_key(key, raw, &length) == 1 && length == PURE_SEAL_KEY &&
+ write(fd, raw, PURE_SEAL_KEY) == PURE_SEAL_KEY && !fsync(fd);
+ if (fd >= 0 && close(fd)) ok = false;
+ if (!ok) { snprintf(error, size, "%s: cannot make a key there", path); unlink(path); goto fail; }
+ } else { snprintf(error, size, "%s: %s", path, strerror(errno)); goto fail; }
+ length = PURE_SEAL_KEY;
+ if (!key || EVP_PKEY_get_raw_public_key(key, public_key, &length) != 1 || length != PURE_SEAL_KEY) {
+ snprintf(error, size, "%s: not an X25519 key", path);
+ goto fail;
+ }
+ for (size_t i = 0; i < PURE_SEAL_KEY; ++i) snprintf(public_hex + 2 * i, 3, "%02x", public_key[i]);
+ OPENSSL_secure_clear_free(raw, PURE_SEAL_KEY);
+ made->key = key;
+ return made;
+fail:
+ EVP_PKEY_free(key);
+ if (raw) OPENSSL_secure_clear_free(raw, PURE_SEAL_KEY);
+ free(made);
+ return NULL;
+}
+
+static void seal_key_free(PureSealKey *key) {
+ if (!key) return;
+ EVP_PKEY_free(key->key);
+ free(key);
+}
+
+static PureSealer *sealer(const char *suite_name, const unsigned char peer[PURE_SEAL_KEY], const PureSealKey *own,
+ unsigned char enc[PURE_SEAL_KEY]) {
+ OSSL_HPKE_SUITE suite;
+ PureSealer *made = suite_named(suite_name, &suite) ? calloc(1, sizeof(*made)) : NULL;
+ if (!made) return NULL;
+ made->context = OSSL_HPKE_CTX_new(own ? OSSL_HPKE_MODE_AUTH : OSSL_HPKE_MODE_BASE, suite, OSSL_HPKE_ROLE_SENDER,
+ NULL, NULL);
+ size_t length = PURE_SEAL_KEY;
+ if (!made->context || (own && OSSL_HPKE_CTX_set1_authpriv(made->context, own->key) != 1) ||
+ OSSL_HPKE_encap(made->context, enc, &length, peer, PURE_SEAL_KEY, info, sizeof(info) - 1) != 1 ||
+ length != PURE_SEAL_KEY) {
+ OSSL_HPKE_CTX_free(made->context);
+ free(made);
+ return NULL;
+ }
+ return made;
+}
+
+static bool seal(PureSealer *s, const char *aad, const unsigned char *plain, size_t size, unsigned char *out,
+ uint64_t *n) {
+ size_t sealed = size + PURE_SEAL_OVERHEAD;
+ return OSSL_HPKE_CTX_get_seq(s->context, n) == 1 &&
+ OSSL_HPKE_seal(s->context, out, &sealed, (const unsigned char *)aad, strlen(aad), plain, size) == 1 &&
+ sealed == size + PURE_SEAL_OVERHEAD;
+}
+
+static void sealer_free(PureSealer *s) {
+ if (!s) return;
+ OSSL_HPKE_CTX_free(s->context);
+ free(s);
+}
+
+static PureOpener *opener(const PureSealKey *key, const char *suite_name, const unsigned char enc[PURE_SEAL_KEY],
+ const unsigned char *from) {
+ OSSL_HPKE_SUITE suite;
+ PureOpener *made = key && suite_named(suite_name, &suite) ? calloc(1, sizeof(*made)) : NULL;
+ if (!made) return NULL;
+ made->context = OSSL_HPKE_CTX_new(from ? OSSL_HPKE_MODE_AUTH : OSSL_HPKE_MODE_BASE, suite, OSSL_HPKE_ROLE_RECEIVER,
+ NULL, NULL);
+ if (!made->context || (from && OSSL_HPKE_CTX_set1_authpub(made->context, from, PURE_SEAL_KEY) != 1) ||
+ OSSL_HPKE_decap(made->context, enc, PURE_SEAL_KEY, key->key, info, sizeof(info) - 1) != 1) {
+ OSSL_HPKE_CTX_free(made->context);
+ free(made);
+ return NULL;
+ }
+ return made;
+}
+
+/* A receiver may set the number (a sender may not: it would reuse a nonce). */
+static bool unseal(PureOpener *o, uint64_t n, const char *aad, const unsigned char *sealed, size_t size,
+ unsigned char *plain) {
+ size_t opened = size - PURE_SEAL_OVERHEAD;
+ return size >= PURE_SEAL_OVERHEAD && OSSL_HPKE_CTX_set_seq(o->context, n) == 1 &&
+ OSSL_HPKE_open(o->context, plain, &opened, (const unsigned char *)aad, strlen(aad), sealed, size) == 1 &&
+ opened == size - PURE_SEAL_OVERHEAD;
+}
+
+static void opener_free(PureOpener *o) {
+ if (!o) return;
+ OSSL_HPKE_CTX_free(o->context);
+ free(o);
+}
+
+/* ---- Links: X25519 between two nodes' keys, HKDF-SHA256, AES-128-GCM with
+ * the caller's nonce (one context sealing and one opening, each keeping its
+ * key's schedule). */
+struct PureAead { EVP_CIPHER_CTX *seal, *open; };
+
+static bool link_secret(const PureSealKey *own, const unsigned char peer[PURE_SEAL_KEY], unsigned char out[32]) {
+ EVP_PKEY *theirs = EVP_PKEY_new_raw_public_key(EVP_PKEY_X25519, NULL, peer, PURE_SEAL_KEY);
+ EVP_PKEY_CTX *context = theirs && own ? EVP_PKEY_CTX_new(own->key, NULL) : NULL;
+ size_t length = 32;
+ bool ok = context && EVP_PKEY_derive_init(context) == 1 && EVP_PKEY_derive_set_peer(context, theirs) == 1 &&
+ EVP_PKEY_derive(context, out, &length) == 1 && length == 32;
+ EVP_PKEY_CTX_free(context);
+ EVP_PKEY_free(theirs);
+ unsigned char any = 0; /* a small-order key makes zeros: no secret */
+ for (size_t i = 0; ok && i < 32; ++i) any |= out[i];
+ return ok && any;
+}
+
+/* HKDF, fetched once. */
+static EVP_KDF *hkdf(void) {
+ static EVP_KDF *kept;
+ EVP_KDF *kdf = __atomic_load_n(&kept, __ATOMIC_ACQUIRE), *none = NULL;
+ if (kdf) return kdf;
+ if ((kdf = EVP_KDF_fetch(NULL, "HKDF", NULL)) &&
+ !__atomic_compare_exchange_n(&kept, &none, kdf, false, __ATOMIC_ACQ_REL, __ATOMIC_ACQUIRE)) {
+ EVP_KDF_free(kdf); /* another thread's is kept */
+ kdf = none;
+ }
+ return kdf;
+}
+
+static bool link_derive(const unsigned char secret[32], const unsigned char *info, size_t info_size, unsigned char *out,
+ size_t out_size) {
+ static const char salt[] = "pure link 0";
+ EVP_KDF *kdf = hkdf();
+ EVP_KDF_CTX *context = kdf ? EVP_KDF_CTX_new(kdf) : NULL;
+ OSSL_PARAM params[] = {OSSL_PARAM_construct_utf8_string(OSSL_KDF_PARAM_DIGEST, (char *)"SHA256", 0),
+ OSSL_PARAM_construct_octet_string(OSSL_KDF_PARAM_KEY, (void *)secret, 32),
+ OSSL_PARAM_construct_octet_string(OSSL_KDF_PARAM_SALT, (void *)salt, sizeof(salt) - 1),
+ OSSL_PARAM_construct_octet_string(OSSL_KDF_PARAM_INFO, (void *)info, info_size),
+ OSSL_PARAM_construct_end()};
+ bool ok = context && EVP_KDF_derive(context, out, out_size, params) == 1;
+ EVP_KDF_CTX_free(context);
+ return ok;
+}
+
+static void aead_free(PureAead *a) {
+ if (!a) return;
+ EVP_CIPHER_CTX_free(a->seal);
+ EVP_CIPHER_CTX_free(a->open);
+ free(a);
+}
+
+static PureAead *aead(const unsigned char key[PURE_AEAD_KEY]) {
+ PureAead *a = calloc(1, sizeof(*a));
+ if (!a) return NULL;
+ a->seal = EVP_CIPHER_CTX_new();
+ a->open = EVP_CIPHER_CTX_new();
+ if (!a->seal || !a->open || EVP_EncryptInit_ex(a->seal, EVP_aes_128_gcm(), NULL, key, NULL) != 1 ||
+ EVP_DecryptInit_ex(a->open, EVP_aes_128_gcm(), NULL, key, NULL) != 1) {
+ aead_free(a);
+ return NULL;
+ }
+ return a;
+}
+
+static bool aead_seal(PureAead *a, const unsigned char *key, const unsigned char nonce[PURE_AEAD_NONCE],
+ const unsigned char *aad, size_t aad_size, const unsigned char *plain, size_t size,
+ unsigned char *out) {
+ int n = 0, last = 0;
+ return size <= INT_MAX && EVP_EncryptInit_ex(a->seal, NULL, NULL, key, nonce) == 1 &&
+ (!aad_size || EVP_EncryptUpdate(a->seal, NULL, &n, aad, (int)aad_size) == 1) &&
+ (!size || EVP_EncryptUpdate(a->seal, out, &n, plain, (int)size) == 1) &&
+ EVP_EncryptFinal_ex(a->seal, out + size, &last) == 1 &&
+ EVP_CIPHER_CTX_ctrl(a->seal, EVP_CTRL_GCM_GET_TAG, PURE_AEAD_TAG, out + size) == 1;
+}
+
+static bool aead_open(PureAead *a, const unsigned char *key, const unsigned char nonce[PURE_AEAD_NONCE],
+ const unsigned char *aad, size_t aad_size, const unsigned char *sealed, size_t size,
+ unsigned char *plain) {
+ int n = 0, last = 0;
+ if (size < PURE_AEAD_TAG || size - PURE_AEAD_TAG > INT_MAX) return false;
+ size_t body = size - PURE_AEAD_TAG;
+ return EVP_DecryptInit_ex(a->open, NULL, NULL, key, nonce) == 1 &&
+ (!aad_size || EVP_DecryptUpdate(a->open, NULL, &n, aad, (int)aad_size) == 1) &&
+ (!body || EVP_DecryptUpdate(a->open, plain, &n, sealed, (int)body) == 1) &&
+ EVP_CIPHER_CTX_ctrl(a->open, EVP_CTRL_GCM_SET_TAG, PURE_AEAD_TAG, (void *)(sealed + body)) == 1 &&
+ EVP_DecryptFinal_ex(a->open, plain + body, &last) == 1;
+}
+
+const PureCrypto pure_crypto = {PURE_CRYPTO_VERSION, tls_start, tls_open, tls_read, tls_write, tls_close,
+ seal_key, seal_key_free, sealer, seal, sealer_free, opener, unseal, opener_free,
+ link_secret, link_derive, aead, aead_seal, aead_open, aead_free};
diff --git a/link/crypto.c b/link/crypto.c
new file mode 100644
index 0000000..2fa7d44
--- /dev/null
+++ b/link/crypto.c
@@ -0,0 +1,167 @@
+#define _GNU_SOURCE
+/* Sealing and TLS for pure-serve (link/seal.h, link/tls.h), by way of
+ * pure-crypto.so (link/crypto.h): mapped, and OpenSSL with it, the first
+ * time a server needs it. What needs no OpenSSL (a key's hex, the suite
+ * this machine seals with) is here. */
+#include "crypto.h"
+#include <dlfcn.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#if defined(__aarch64__) && defined(__linux__)
+#include <asm/hwcap.h>
+#include <sys/auxv.h>
+#endif
+
+static pthread_once_t once = PTHREAD_ONCE_INIT;
+static const PureCrypto *crypto;
+static char why[512];
+
+/* pure-crypto.so beside the executable, or where PURE_CRYPTO names it. */
+static void load(void) {
+ char path[4096];
+ const char *named = getenv("PURE_CRYPTO");
+ if (named) snprintf(path, sizeof(path), "%s", named);
+ else {
+ ssize_t length = readlink("/proc/self/exe", path, sizeof(path) - 32);
+ char *slash = length > 0 ? memrchr(path, '/', (size_t)length) : NULL;
+ if (!slash) { snprintf(why, sizeof(why), "pure-crypto.so: cannot find pure-serve's directory"); return; }
+ strcpy(slash + 1, "pure-crypto.so");
+ }
+ void *library = dlopen(path, RTLD_NOW | RTLD_LOCAL);
+ const PureCrypto *table = library ? dlsym(library, "pure_crypto") : NULL;
+ if (!table || table->version != PURE_CRYPTO_VERSION) {
+ snprintf(why, sizeof(why), "%s", !library ? dlerror() : "pure-crypto.so is not this pure-serve's");
+ return;
+ }
+ crypto = table;
+}
+
+static const PureCrypto *loaded(char *error, size_t size) {
+ pthread_once(&once, load);
+ if (!crypto && error) snprintf(error, size, "%s", why);
+ return crypto;
+}
+
+/* ---- TLS */
+
+bool pure_tls_start(const char *ca_file, char *error, size_t size) {
+ static int started; /* 1 made, -1 failed; by the thread that opens wires */
+ const PureCrypto *c = loaded(error, size);
+ if (!c) return false;
+ if (!started) started = c->tls_start(ca_file) ? 1 : -1;
+ if (started < 0) snprintf(error, size, "no TLS context: %s cannot be read", ca_file ? ca_file : "the system's CAs");
+ return started > 0;
+}
+
+/* A session exists only once the table is loaded. */
+PureTls *pure_tls_open(const char *host, int fd) { return crypto ? crypto->tls_open(host, fd) : NULL; }
+int pure_tls_read(PureTls *tls, void *bytes, int size) { return crypto->tls_read(tls, bytes, size); }
+int pure_tls_write(PureTls *tls, const void *bytes, int size) { return crypto->tls_write(tls, bytes, size); }
+void pure_tls_close(PureTls *tls) { if (tls) crypto->tls_close(tls); }
+
+/* ---- Sealing */
+
+static const char aes_name[] = "x25519-sha256-aes128gcm", chacha_name[] = "x25519-sha256-chacha20poly1305";
+
+/* AES-GCM where AES is in the CPU (two to three times ChaCha20-Poly1305 from
+ * a kilobyte up on a Cortex-A76, research 086); ChaCha20-Poly1305, fast in
+ * plain code, elsewhere. */
+static bool has_aes(void) {
+#if defined(__aarch64__) && defined(__linux__)
+ return (getauxval(AT_HWCAP) & HWCAP_AES) && (getauxval(AT_HWCAP) & HWCAP_PMULL);
+#elif defined(__x86_64__) && (defined(__GNUC__) || defined(__clang__))
+ return __builtin_cpu_supports("aes") && __builtin_cpu_supports("pclmul");
+#else
+ return false;
+#endif
+}
+
+const char *pure_seal_suite(void) { return has_aes() ? aes_name : chacha_name; }
+
+static bool hex_value(char c, unsigned *v) {
+ if (c >= '0' && c <= '9') *v = (unsigned)(c - '0');
+ else if (c >= 'a' && c <= 'f') *v = (unsigned)(c - 'a' + 10);
+ else if (c >= 'A' && c <= 'F') *v = (unsigned)(c - 'A' + 10);
+ else return false;
+ return true;
+}
+
+bool pure_seal_peer(const char *hex, unsigned char out[PURE_SEAL_KEY]) {
+ if (strlen(hex) != 2 * PURE_SEAL_KEY) return false;
+ for (size_t i = 0; i < PURE_SEAL_KEY; ++i) {
+ unsigned high, low;
+ if (!hex_value(hex[2 * i], &high) || !hex_value(hex[2 * i + 1], &low)) return false;
+ out[i] = (unsigned char)(high << 4 | low);
+ }
+ return true;
+}
+
+bool pure_seal_ready(char *error, size_t size) { return loaded(error, size) != NULL; }
+
+PureSealKey *pure_seal_key(const char *path, char public_hex[2 * PURE_SEAL_KEY + 1], char *error, size_t size) {
+ const PureCrypto *c = loaded(error, size);
+ return c ? c->seal_key(path, public_hex, error, size) : NULL;
+}
+
+/* A key or a session exists only once the table is loaded. */
+void pure_seal_key_free(PureSealKey *key) { if (key) crypto->seal_key_free(key); }
+
+PureSealer *pure_sealer(const unsigned char peer[PURE_SEAL_KEY], const PureSealKey *own,
+ unsigned char enc[PURE_SEAL_KEY]) {
+ const PureCrypto *c = loaded(NULL, 0);
+ return c ? c->sealer(pure_seal_suite(), peer, own, enc) : NULL;
+}
+
+bool pure_seal(PureSealer *s, const char *aad, const unsigned char *plain, size_t size, unsigned char *out,
+ uint64_t *n) {
+ return crypto->seal(s, aad, plain, size, out, n);
+}
+
+void pure_sealer_free(PureSealer *s) { if (s) crypto->sealer_free(s); }
+
+PureOpener *pure_opener(const PureSealKey *key, const char *suite, const unsigned char enc[PURE_SEAL_KEY],
+ const unsigned char *from) {
+ return key ? crypto->opener(key, suite, enc, from) : NULL;
+}
+
+bool pure_unseal(PureOpener *o, uint64_t n, const char *aad, const unsigned char *sealed, size_t size,
+ unsigned char *plain) {
+ return crypto->unseal(o, n, aad, sealed, size, plain);
+}
+
+void pure_opener_free(PureOpener *o) { if (o) crypto->opener_free(o); }
+
+/* ---- Links */
+
+bool pure_link_secret(const PureSealKey *own, const unsigned char peer[PURE_SEAL_KEY], unsigned char out[32]) {
+ const PureCrypto *c = loaded(NULL, 0);
+ return c && own && c->link_secret(own, peer, out);
+}
+
+bool pure_link_derive(const unsigned char secret[32], const unsigned char *info, size_t info_size, unsigned char *out,
+ size_t out_size) {
+ const PureCrypto *c = loaded(NULL, 0);
+ return c && c->link_derive(secret, info, info_size, out, out_size);
+}
+
+PureAead *pure_aead(const unsigned char key[PURE_AEAD_KEY]) {
+ const PureCrypto *c = loaded(NULL, 0);
+ return c ? c->aead(key) : NULL;
+}
+
+bool pure_aead_seal(PureAead *a, const unsigned char *key, const unsigned char nonce[PURE_AEAD_NONCE],
+ const unsigned char *aad, size_t aad_size, const unsigned char *plain, size_t size,
+ unsigned char *out) {
+ return crypto->aead_seal(a, key, nonce, aad, aad_size, plain, size, out);
+}
+
+bool pure_aead_open(PureAead *a, const unsigned char *key, const unsigned char nonce[PURE_AEAD_NONCE],
+ const unsigned char *aad, size_t aad_size, const unsigned char *sealed, size_t size,
+ unsigned char *plain) {
+ return crypto->aead_open(a, key, nonce, aad, aad_size, sealed, size, plain);
+}
+
+void pure_aead_free(PureAead *a) { if (a) crypto->aead_free(a); }
diff --git a/link/crypto.h b/link/crypto.h
new file mode 100644
index 0000000..53ac785
--- /dev/null
+++ b/link/crypto.h
@@ -0,0 +1,49 @@
+#ifndef PURE_CRYPTO_H
+#define PURE_CRYPTO_H
+#include "seal.h"
+#include "tls.h"
+
+/* What pure-serve asks of OpenSSL, as the table pure-crypto.so exports
+ * (link/crypto-openssl.c). pure-serve maps it, and OpenSSL with it, the
+ * first time a server seals, opens a seal or connects to an https receiver
+ * (link/crypto.c): one that does none of these maps no OpenSSL, 2.3 MiB
+ * less resident and 0.6 MiB less of its own. */
+enum { PURE_CRYPTO_VERSION = 4 };
+
+typedef struct {
+ unsigned version;
+ /* TLS client: the context (verifying against ca_file, or the system's
+ * CAs when NULL), sessions over connected sockets, their reads and
+ * writes (as SSL_read and SSL_write: bytes, or <= 0), and their end. */
+ bool (*tls_start)(const char *ca_file);
+ PureTls *(*tls_open)(const char *host, int fd);
+ int (*tls_read)(PureTls *, void *bytes, int size);
+ int (*tls_write)(PureTls *, const void *bytes, int size);
+ void (*tls_close)(PureTls *);
+ /* Sealing, as link/seal.h says; a sender's sessions in the suite named. */
+ PureSealKey *(*seal_key)(const char *path, char public_hex[2 * PURE_SEAL_KEY + 1], char *error, size_t size);
+ void (*seal_key_free)(PureSealKey *);
+ PureSealer *(*sealer)(const char *suite, const unsigned char peer[PURE_SEAL_KEY], const PureSealKey *own,
+ unsigned char enc[PURE_SEAL_KEY]);
+ bool (*seal)(PureSealer *, const char *aad, const unsigned char *plain, size_t size, unsigned char *out,
+ uint64_t *n);
+ void (*sealer_free)(PureSealer *);
+ PureOpener *(*opener)(const PureSealKey *, const char *suite, const unsigned char enc[PURE_SEAL_KEY],
+ const unsigned char *from);
+ bool (*unseal)(PureOpener *, uint64_t n, const char *aad, const unsigned char *sealed, size_t size,
+ unsigned char *plain);
+ void (*opener_free)(PureOpener *);
+ /* Links, as link/seal.h says. */
+ bool (*link_secret)(const PureSealKey *, const unsigned char peer[PURE_SEAL_KEY], unsigned char out[32]);
+ bool (*link_derive)(const unsigned char secret[32], const unsigned char *info, size_t info_size, unsigned char *out,
+ size_t out_size);
+ PureAead *(*aead)(const unsigned char key[PURE_AEAD_KEY]);
+ bool (*aead_seal)(PureAead *, const unsigned char *key, const unsigned char nonce[PURE_AEAD_NONCE],
+ const unsigned char *aad, size_t aad_size, const unsigned char *plain, size_t size,
+ unsigned char *out);
+ bool (*aead_open)(PureAead *, const unsigned char *key, const unsigned char nonce[PURE_AEAD_NONCE],
+ const unsigned char *aad, size_t aad_size, const unsigned char *sealed, size_t size,
+ unsigned char *plain);
+ void (*aead_free)(PureAead *);
+} PureCrypto;
+#endif
diff --git a/link/link.c b/link/link.c
new file mode 100644
index 0000000..9c6794b
--- /dev/null
+++ b/link/link.c
@@ -0,0 +1,1862 @@
+#define _GNU_SOURCE
+/* Links without connections (link/link.h, docs/design/011-pure-transport.md). */
+#include "link.h"
+#include <arpa/inet.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <malloc.h>
+#include <netdb.h>
+#include <netinet/in.h>
+#include <netinet/udp.h>
+#include <poll.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/eventfd.h>
+#include <sys/mman.h>
+#include <sys/random.h>
+#include <sys/socket.h>
+#include <sys/stat.h>
+#include <time.h>
+#include <unistd.h>
+
+enum { VERSION = 0x70, HEADER = 21, DATAGRAM = 1200, BODY = DATAGRAM - HEADER - PURE_AEAD_TAG, PIECE = 1000 };
+enum { F_PADDING = 0, F_STREAM = 1, F_RECORD = 2, F_ACK = 3, F_STORED = 4, F_PING = 5, F_RECORDS = 6, STORED_GAP = 1 };
+enum { RECORDS_HEAD = 10 }; /* type, the first number, the count */
+/* Packets sent at once with UDP GSO: each but the last is a whole datagram
+ * (a full one padded), and 54 of them fit a UDP datagram's 65,507 bytes. */
+enum { BURST = 54 };
+#ifndef UDP_SEGMENT
+#define UDP_SEGMENT 103
+#endif
+#ifndef UDP_GRO
+#define UDP_GRO 104
+#endif
+enum { RANGES = 32, RUNS = 2, UNITS = 64, CREDIT = 1024, BATCH = 256, RECORD_HEAD = 19 };
+enum { WINDOW_FIRST = 10 * DATAGRAM, WINDOW_LEAST = 2 * DATAGRAM };
+#define MS UINT64_C(1000000)
+/* RFC 9002's values, but a first RTT of 100 ms (a LAN's, not the 333 ms of
+ * the Internet at large); a probe at least every second while a receiver is
+ * away; a record received but not stored after a second is sent again. */
+static const uint64_t FIRST_RTT = 100 * MS, ACK_DELAY_MOST = 25 * MS, GRANULARITY = 1 * MS, PTO_MOST = 1000 * MS,
+ STORED_WAIT = 1000 * MS, RECORD_MOST = UINT64_C(64) << 20, HELD_MOST = UINT64_C(16) << 20;
+
+static void put16(unsigned char *p, uint32_t v) { p[0] = (unsigned char)v, p[1] = (unsigned char)(v >> 8); }
+static void put32(unsigned char *p, uint32_t v) { for (int i = 0; i < 4; ++i) p[i] = (unsigned char)(v >> 8 * i); }
+static void put64(unsigned char *p, uint64_t v) { for (int i = 0; i < 8; ++i) p[i] = (unsigned char)(v >> 8 * i); }
+static uint32_t get16(const unsigned char *p) { return (uint32_t)p[0] | (uint32_t)p[1] << 8; }
+static uint32_t get32(const unsigned char *p) {
+ return (uint32_t)p[0] | (uint32_t)p[1] << 8 | (uint32_t)p[2] << 16 | (uint32_t)p[3] << 24;
+}
+static uint64_t get64(const unsigned char *p) { return (uint64_t)get32(p) | (uint64_t)get32(p + 4) << 32; }
+/* LEB128: seven bits a byte, low first. */
+static size_t put_varint(unsigned char *p, uint64_t v) {
+ size_t n = 0;
+ for (; v >= 0x80; v >>= 7) p[n++] = (unsigned char)(v | 0x80);
+ p[n++] = (unsigned char)v;
+ return n;
+}
+static size_t varint_size(uint64_t v) {
+ size_t n = 1;
+ for (; v >= 0x80; v >>= 7) ++n;
+ return n;
+}
+/* A varint from p (size bytes): how many bytes it took, or 0. */
+static size_t get_varint(const unsigned char *p, size_t size, uint64_t *v) {
+ *v = 0;
+ for (size_t n = 0; n < size && n < 10; ++n) {
+ *v |= (uint64_t)(p[n] & 0x7f) << (7 * n);
+ if (!(p[n] & 0x80)) return n + 1;
+ }
+ return 0;
+}
+
+static uint64_t least(uint64_t a, uint64_t b) { return a < b ? a : b; }
+static uint64_t most(uint64_t a, uint64_t b) { return a > b ? a : b; }
+
+static uint64_t now_ns(void) {
+ struct timespec t;
+ clock_gettime(CLOCK_MONOTONIC, &t);
+ return (uint64_t)t.tv_sec * 1000000000u + (uint64_t)t.tv_nsec;
+}
+
+static uint32_t fnv(uint32_t h, const void *bytes, size_t size) {
+ const unsigned char *p = bytes;
+ for (size_t i = 0; i < size; ++i) h = (h ^ p[i]) * 16777619u;
+ return h;
+}
+
+static uint32_t key_id(const unsigned char key[PURE_SEAL_KEY]) { return fnv(2166136261u, key, PURE_SEAL_KEY); }
+
+/* What a message of input looks like: its fields' names, types and places. */
+static uint32_t shape_of(const PureModule *m, uint32_t input) {
+ const PureInputShape *s = &m->input[input];
+ uint32_t h = fnv(2166136261u, &s->prefix, sizeof(s->prefix));
+ for (uint32_t f = 0; f < s->field_count; ++f) {
+ const PureField *field = &m->field[s->field_start + f];
+ unsigned char fixed[7] = {field->type, (unsigned char)field->rows, (unsigned char)(field->rows >> 8)};
+ put32(fixed + 3, field->offset);
+ h = fnv(fnv(h, field->name, strlen(field->name) + 1), fixed, sizeof(fixed));
+ }
+ return h;
+}
+
+static uint64_t random64(void) {
+ uint64_t value = 0;
+ while (getrandom(&value, sizeof(value), 0) != sizeof(value) || !value) {}
+ return value;
+}
+
+/* The key and nonce base of packets from `from` (in its run from_run) to
+ * `to`, answering to_run (0 for none). */
+static bool derive(const unsigned char secret[32], const unsigned char from[PURE_SEAL_KEY],
+ const unsigned char to[PURE_SEAL_KEY], uint64_t from_run, uint64_t to_run,
+ unsigned char key[PURE_AEAD_KEY], unsigned char base[PURE_AEAD_NONCE]) {
+ unsigned char info[16 + 2 * PURE_SEAL_KEY + 16], out[PURE_AEAD_KEY + PURE_AEAD_NONCE];
+ memcpy(info, "pure link 0 keys", 16);
+ memcpy(info + 16, from, PURE_SEAL_KEY);
+ memcpy(info + 16 + PURE_SEAL_KEY, to, PURE_SEAL_KEY);
+ put64(info + 16 + 2 * PURE_SEAL_KEY, from_run);
+ put64(info + 24 + 2 * PURE_SEAL_KEY, to_run);
+ if (!pure_link_derive(secret, info, sizeof(info), out, sizeof(out))) return false;
+ memcpy(key, out, PURE_AEAD_KEY);
+ memcpy(base, out + PURE_AEAD_KEY, PURE_AEAD_NONCE);
+ explicit_bzero(out, sizeof(out));
+ return true;
+}
+
+/* The same as a context of its own, for a sender's one link. */
+static PureAead *aead_of(const unsigned char secret[32], const unsigned char from[PURE_SEAL_KEY],
+ const unsigned char to[PURE_SEAL_KEY], uint64_t from_run, uint64_t to_run,
+ unsigned char base[PURE_AEAD_NONCE]) {
+ unsigned char key[PURE_AEAD_KEY];
+ PureAead *aead = derive(secret, from, to, from_run, to_run, key, base) ? pure_aead(key) : NULL;
+ explicit_bzero(key, sizeof(key));
+ return aead;
+}
+
+static void nonce_of(unsigned char out[PURE_AEAD_NONCE], const unsigned char base[PURE_AEAD_NONCE], uint64_t pn) {
+ memcpy(out, base, PURE_AEAD_NONCE);
+ for (int i = 0; i < 8; ++i) out[4 + i] ^= (unsigned char)(pn >> (56 - 8 * i));
+}
+
+/* A packet: the header, then body sealed (with key, or the context's own
+ * when NULL); its size, or 0. */
+static size_t seal_packet(PureAead *aead, const unsigned char *key, const unsigned char base[PURE_AEAD_NONCE], uint32_t id,
+ uint64_t run, uint64_t pn, const unsigned char *body, size_t size, unsigned char *out) {
+ unsigned char nonce[PURE_AEAD_NONCE];
+ out[0] = VERSION;
+ put32(out + 1, id);
+ put64(out + 5, run);
+ put64(out + 13, pn);
+ nonce_of(nonce, base, pn);
+ return pure_aead_seal(aead, key, nonce, out, HEADER, body, size, out + HEADER) ? HEADER + size + PURE_AEAD_TAG : 0;
+}
+
+static size_t put_names(unsigned char *p, const char *source, const char *input) {
+ size_t a = strlen(source), b = strlen(input);
+ p[0] = (unsigned char)a;
+ memcpy(p + 1, source, a);
+ p[1 + a] = (unsigned char)b;
+ memcpy(p + 2 + a, input, b);
+ return 2 + a + b;
+}
+
+/* A source and an input's names from p (size bytes); how many it takes, or 0. */
+static size_t get_names(const unsigned char *p, size_t size, char source[PURE_SOURCE + 1], char input[PURE_MODULE_NAME]) {
+ if (size < 1 || !p[0] || p[0] > PURE_SOURCE || size < 2u + p[0]) return 0;
+ size_t a = p[0], b = p[1 + a];
+ if (!b || b >= PURE_MODULE_NAME || size < 2 + a + b) return 0;
+ memcpy(source, p + 1, a);
+ source[a] = 0;
+ memcpy(input, p + 2 + a, b);
+ input[b] = 0;
+ return 2 + a + b;
+}
+
+/* Datagrams taken from a socket at once (recvmmsg). */
+enum { TAKE = 64, SLOT = DATAGRAM + 64 };
+typedef struct {
+ unsigned char bytes[TAKE][SLOT];
+ struct sockaddr_storage from[TAKE];
+ struct iovec parts[TAKE];
+ struct mmsghdr messages[TAKE];
+} Taking;
+
+/* With UDP GRO the kernel hands a listener the datagrams of one sender that
+ * came one after another as one buffer, each the segment's size but the
+ * last (a burst sent with GSO comes whole, not cut up): up to GATHER such
+ * buffers at once. */
+enum { GATHER = 8, GATHERED = 65536 };
+typedef struct {
+ unsigned char bytes[GATHER][GATHERED];
+ struct sockaddr_storage from[GATHER];
+ struct iovec parts[GATHER];
+ struct mmsghdr messages[GATHER];
+ _Alignas(8) unsigned char control[GATHER][64]; /* a cmsghdr's alignment */
+} Gathering;
+
+static int gather(int fd, Gathering *g) {
+ for (unsigned k = 0; k < GATHER; ++k) {
+ g->parts[k] = (struct iovec){g->bytes[k], GATHERED};
+ g->messages[k] = (struct mmsghdr){.msg_hdr = {.msg_name = &g->from[k], .msg_namelen = sizeof(g->from[k]),
+ .msg_iov = &g->parts[k], .msg_iovlen = 1,
+ .msg_control = g->control[k],
+ .msg_controllen = sizeof(g->control[k])}};
+ }
+ int n = recvmmsg(fd, g->messages, GATHER, MSG_DONTWAIT, NULL);
+ return n > 0 ? n : 0;
+}
+
+/* The size of each datagram in a buffer gathered (all of it when one). */
+static size_t segment_of(struct msghdr *m, size_t length) {
+ for (struct cmsghdr *c = CMSG_FIRSTHDR(m); c; c = CMSG_NXTHDR(m, c))
+ if (c->cmsg_level == SOL_UDP && c->cmsg_type == UDP_GRO) {
+ int size;
+ memcpy(&size, CMSG_DATA(c), sizeof(size));
+ if (size > 0) return (size_t)size;
+ }
+ return length ? length : 1;
+}
+
+/* Up to TAKE datagrams from fd: how many (0 when none waits). */
+static int take_datagrams(int fd, Taking *t) {
+ for (unsigned k = 0; k < TAKE; ++k) {
+ t->parts[k] = (struct iovec){t->bytes[k], SLOT};
+ t->messages[k] = (struct mmsghdr){.msg_hdr = {.msg_name = &t->from[k], .msg_namelen = sizeof(t->from[k]),
+ .msg_iov = &t->parts[k], .msg_iovlen = 1}};
+ }
+ int n = recvmmsg(fd, t->messages, TAKE, MSG_DONTWAIT, NULL);
+ return n > 0 ? n : 0;
+}
+
+/* ---- The sending side */
+
+enum { UNSENT, FLYING, RECEIVED };
+typedef struct { uint64_t pn; uint8_t state; } Piece;
+/* A record queued: its bytes at `at` in the link's byte ring, its pieces'
+ * states (one kept here, more beside). */
+typedef struct { uint64_t seq; size_t at; uint32_t size, pieces; Piece *more, one; } Record;
+static Piece *piece_of(Record *r, uint32_t i) { return r->pieces == 1 ? &r->one : &r->more[i]; }
+typedef struct { uint64_t seq; uint32_t piece; } Unit;
+typedef struct {
+ uint64_t pn, sent; /* its number; when it went (ns) */
+ uint32_t bytes;
+ bool eliciting, done; /* acknowledged or lost */
+ uint16_t count;
+ Unit units[UNITS];
+} Flight;
+
+struct PureLink {
+ const PureModule *module;
+ uint32_t output, shape, own_id, peer_id;
+ char host[256], port[8], input[PURE_MODULE_NAME], source[PURE_SOURCE + 1];
+ const PureSealKey *own;
+ unsigned char own_public[PURE_SEAL_KEY], peer[PURE_SEAL_KEY], secret[32];
+ bool peered;
+ pthread_mutex_t lock;
+ Record *records; /* the sends not yet stored, oldest first (a ring of consecutive numbers) */
+ size_t head, count, capacity;
+ /* Their bytes, one after another as queued (a record's never cut by the
+ * ring's end): taken back in order as they are stored. */
+ unsigned char *ring;
+ size_t ring_capacity;
+ uint64_t queued, stored, told; /* the last queued; saved by the receiver; told the owner */
+ bool stopping, started;
+ /* The thread sleeps (in poll), and has been woken since; records were
+ * queued since it last sent. A queue wakes it only when it sleeps, once. */
+ bool sleeping, signaled, queued_new;
+ int fd, wake;
+ pthread_t thread;
+ void (*notify)(void *);
+ void *context;
+ /* The thread's own, under the lock. */
+ uint64_t run, next_pn, in_run;
+ PureAead *out, *in;
+ unsigned char out_base[PURE_AEAD_NONCE], in_base[PURE_AEAD_NONCE];
+ Flight *flights; /* one for each number sent, oldest first (a ring) */
+ size_t flight_head, flight_count, flight_capacity;
+ uint64_t in_flight, window, threshold, recovery_start, eliciting;
+ uint64_t largest_acked, latest_rtt, smoothed_rtt, rtt_var, min_rtt;
+ bool acked_any, sampled;
+ uint64_t loss_time, last_eliciting, credit, stored_changed;
+ uint64_t received; /* pieces received and not yet stored */
+ uint64_t unsent_from; /* no record before this number has a piece unsent */
+ Taking *taking; /* answers, received at once */
+ unsigned char *burst; /* packets sealed and not yet sent (BURST datagrams) */
+ size_t burst_size;
+ unsigned burst_count;
+ uint64_t burst_pn[BURST];
+ bool gso;
+ unsigned pto_count, probes;
+ PureLinkCounts counts;
+};
+
+PureLink *pure_link_open(const PureModule *m, const char *output, const char *url, const char *source,
+ const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY], char *error,
+ size_t size) {
+ int index = pure_module_input(m, output);
+ if (index < 0 || m->input[index].kind != PURE_KIND_OUTPUT) {
+ snprintf(error, size, "no output '%s'", output);
+ return NULL;
+ }
+ if (!own) {
+ snprintf(error, size, "a pure:// link needs the node's --key");
+ return NULL;
+ }
+ size_t length = strlen(source);
+ const char *host = url + 7, *slash = strchr(host, '/'), *colon = slash ? memchr(host, ':', (size_t)(slash - host)) : NULL;
+ if (!length || length > PURE_SOURCE || strncmp(url, "pure://", 7) || !slash || !colon || colon == host ||
+ (size_t)(colon - host) >= 256 || slash - colon - 1 < 1 || slash - colon - 1 > 5 || !slash[1] ||
+ strlen(slash + 1) >= PURE_MODULE_NAME) {
+ snprintf(error, size, "a link is pure://host:port/INPUT");
+ return NULL;
+ }
+ PureLink *l = calloc(1, sizeof(*l));
+ if (!l || pthread_mutex_init(&l->lock, NULL)) {
+ free(l);
+ snprintf(error, size, "out of memory");
+ return NULL;
+ }
+ l->module = m;
+ l->output = (uint32_t)index;
+ l->shape = shape_of(m, (uint32_t)index);
+ memcpy(l->host, host, (size_t)(colon - host));
+ memcpy(l->port, colon + 1, (size_t)(slash - colon - 1));
+ strcpy(l->input, slash + 1);
+ strcpy(l->source, source);
+ l->own = own;
+ memcpy(l->own_public, own_public, PURE_SEAL_KEY);
+ l->own_id = key_id(own_public);
+ l->fd = l->wake = -1;
+ return l;
+}
+
+void pure_link_peer(PureLink *l, const unsigned char peer[PURE_SEAL_KEY]) {
+ memcpy(l->peer, peer, PURE_SEAL_KEY);
+ l->peer_id = key_id(peer);
+ l->peered = true;
+}
+
+uint32_t pure_link_output(const PureLink *l) { return l->output; }
+
+static Record *record_at(PureLink *l, size_t k) { return &l->records[(l->head + k) % l->capacity]; }
+
+/* The queued record numbered seq, or NULL (stored already, or not queued). */
+static Record *record_of(PureLink *l, uint64_t seq) {
+ if (!l->count || seq < record_at(l, 0)->seq || seq - record_at(l, 0)->seq >= l->count) return NULL;
+ return record_at(l, (size_t)(seq - record_at(l, 0)->seq));
+}
+
+/* Every change of a piece's state goes through here: the link knows at once
+ * how many pieces wait to be stored, and where the unsent ones begin. */
+static void set_piece(PureLink *l, Record *r, uint32_t i, uint8_t state) {
+ uint8_t was = (*piece_of(r, i)).state;
+ if (was == state) return;
+ if (was == RECEIVED) --l->received;
+ if (state == RECEIVED) ++l->received;
+ if (state == UNSENT && r->seq < l->unsent_from) l->unsent_from = r->seq;
+ (*piece_of(r, i)).state = state;
+}
+
+static void wake(int fd) {
+ uint64_t one = 1;
+ if (fd >= 0 && write(fd, &one, sizeof(one)) != sizeof(one)) { /* full: it is readable already */ }
+}
+
+/* Where size bytes go at the end of the byte ring (after the newest record,
+ * or from its start when that is free and the end is not), or SIZE_MAX. */
+static size_t ring_place(PureLink *l, size_t size) {
+ if (!l->count) return size <= l->ring_capacity ? 0 : SIZE_MAX;
+ Record *oldest = record_at(l, 0), *newest = record_at(l, l->count - 1);
+ size_t head = oldest->at, tail = newest->at + newest->size;
+ if (newest->at >= head) /* the bytes are [head, tail): after them, or before them from the start */
+ return l->ring_capacity - tail >= size ? tail : size <= head ? 0 : SIZE_MAX;
+ return head - tail >= size ? tail : SIZE_MAX; /* around the end: between the newest and the oldest */
+}
+
+/* A ring room for size bytes more: twice as large (or more), the records
+ * moved into it one after another. */
+static bool ring_grow(PureLink *l, size_t size) {
+ size_t used = 0;
+ for (size_t k = 0; k < l->count; ++k) used += record_at(l, k)->size;
+ size_t capacity = l->ring_capacity ? 2 * l->ring_capacity : 64 << 10;
+ while (capacity < used + size + 1) capacity *= 2;
+ unsigned char *ring = malloc(capacity);
+ if (!ring) return false;
+ for (size_t k = 0, at = 0; k < l->count; ++k) {
+ Record *r = record_at(l, k);
+ if (r->size) memcpy(ring + at, l->ring + r->at, r->size);
+ r->at = at;
+ at += r->size;
+ }
+ free(l->ring);
+ l->ring = ring;
+ l->ring_capacity = capacity;
+ return true;
+}
+
+bool pure_link_queue(PureLink *l, uint64_t seq, PureSpan payload) {
+ Record r = {.seq = seq, .size = (uint32_t)payload.size};
+ r.pieces = payload.size ? (uint32_t)((payload.size + PIECE - 1) / PIECE) : 1;
+ if (payload.size > RECORD_MOST || (r.pieces > 1 && !(r.more = calloc(r.pieces, sizeof(Piece))))) return false;
+ pthread_mutex_lock(&l->lock);
+ bool ok = seq > l->queued && (!l->count || seq == l->queued + 1);
+ size_t at = ok ? ring_place(l, payload.size) : SIZE_MAX;
+ if (ok && at == SIZE_MAX) ok = ring_grow(l, payload.size) && (at = ring_place(l, payload.size)) != SIZE_MAX;
+ if (ok) {
+ r.at = at;
+ if (payload.size) memcpy(l->ring + at, payload.data, payload.size);
+ }
+ if (ok && l->count == l->capacity) {
+ size_t capacity = l->capacity ? 2 * l->capacity : 64;
+ Record *grown = malloc(capacity * sizeof(*grown));
+ if (grown) {
+ for (size_t k = 0; k < l->count; ++k) grown[k] = *record_at(l, k);
+ free(l->records);
+ l->records = grown;
+ l->capacity = capacity;
+ l->head = 0;
+ } else
+ ok = false;
+ }
+ bool signal = false;
+ if (ok) {
+ *record_at(l, l->count++) = r;
+ l->queued = seq;
+ l->queued_new = true;
+ signal = l->sleeping && !l->signaled;
+ l->signaled = l->signaled || signal;
+ }
+ pthread_mutex_unlock(&l->lock);
+ if (!ok) {
+ free(r.more);
+ return false;
+ }
+ if (signal) wake(l->wake);
+ return true;
+}
+
+uint64_t pure_link_queued(PureLink *l) {
+ pthread_mutex_lock(&l->lock);
+ uint64_t queued = l->queued;
+ pthread_mutex_unlock(&l->lock);
+ return queued;
+}
+
+uint64_t pure_link_delivered(PureLink *l) {
+ pthread_mutex_lock(&l->lock);
+ uint64_t stored = l->stored;
+ pthread_mutex_unlock(&l->lock);
+ return stored;
+}
+
+static Flight *flight_of(PureLink *l, uint64_t pn) {
+ if (!l->flight_count) return NULL;
+ uint64_t first = l->flights[l->flight_head].pn;
+ if (pn < first || pn - first >= l->flight_count) return NULL;
+ return &l->flights[(l->flight_head + (size_t)(pn - first)) % l->flight_capacity];
+}
+
+static bool push_flight(PureLink *l, const Flight *f) {
+ if (l->flight_count == l->flight_capacity) {
+ size_t capacity = l->flight_capacity ? 2 * l->flight_capacity : 256;
+ Flight *grown = malloc(capacity * sizeof(*grown));
+ if (!grown) return false;
+ for (size_t k = 0; k < l->flight_count; ++k) grown[k] = l->flights[(l->flight_head + k) % l->flight_capacity];
+ free(l->flights);
+ l->flights = grown;
+ l->flight_capacity = capacity;
+ l->flight_head = 0;
+ }
+ l->flights[(l->flight_head + l->flight_count++) % l->flight_capacity] = *f;
+ return true;
+}
+
+static void settle(PureLink *l, Flight *f) {
+ f->done = true;
+ if (f->eliciting) {
+ --l->eliciting;
+ l->in_flight -= f->bytes;
+ }
+}
+
+/* A packet lost: what it carried and was not received otherwise goes again. */
+static void lose(PureLink *l, Flight *f, uint64_t now) {
+ settle(l, f);
+ ++l->counts.lost;
+ for (uint16_t u = 0; u < f->count; ++u) {
+ Record *r = record_of(l, f->units[u].seq);
+ Piece *p = r ? &(*piece_of(r, f->units[u].piece)) : NULL;
+ if (p && p->state == FLYING && p->pn == f->pn) {
+ set_piece(l, r, f->units[u].piece, UNSENT);
+ ++l->counts.resent;
+ }
+ }
+ if (f->sent > l->recovery_start) { /* one congestion event per round trip */
+ l->recovery_start = now;
+ l->window = most(l->window / 2, WINDOW_LEAST);
+ l->threshold = l->window;
+ }
+}
+
+static void detect_lost(PureLink *l, uint64_t now) {
+ l->loss_time = 0;
+ if (!l->acked_any) return;
+ uint64_t rtt = most(l->latest_rtt, l->sampled ? l->smoothed_rtt : FIRST_RTT);
+ uint64_t delay = most(rtt * 9 / 8, GRANULARITY);
+ for (size_t k = 0; k < l->flight_count; ++k) {
+ Flight *f = &l->flights[(l->flight_head + k) % l->flight_capacity];
+ if (f->pn >= l->largest_acked) break;
+ if (f->done) continue;
+ if (f->sent + delay <= now || l->largest_acked >= f->pn + 3) lose(l, f, now);
+ else if (!l->loss_time || f->sent + delay < l->loss_time) l->loss_time = f->sent + delay;
+ }
+}
+
+static void sample_rtt(PureLink *l, uint64_t latest, uint64_t ack_delay) {
+ l->latest_rtt = latest;
+ if (!l->sampled) {
+ l->min_rtt = l->smoothed_rtt = latest;
+ l->rtt_var = latest / 2;
+ l->sampled = true;
+ return;
+ }
+ l->min_rtt = least(l->min_rtt, latest);
+ ack_delay = least(ack_delay, ACK_DELAY_MOST);
+ uint64_t adjusted = latest >= l->min_rtt + ack_delay ? latest - ack_delay : latest;
+ uint64_t difference = l->smoothed_rtt > adjusted ? l->smoothed_rtt - adjusted : adjusted - l->smoothed_rtt;
+ l->rtt_var = (3 * l->rtt_var + difference) / 4;
+ l->smoothed_rtt = (7 * l->smoothed_rtt + adjusted) / 8;
+}
+
+static void on_ack(PureLink *l, const unsigned char *p, size_t size, uint64_t now) {
+ if (size < 13 || size < 13 + 16u * p[12]) return;
+ uint64_t largest = get64(p), delay = (uint64_t)get32(p + 8) * 1000;
+ Flight *top = NULL;
+ for (unsigned k = 0; k < p[12]; ++k) {
+ uint64_t first = get64(p + 13 + 16 * k), last = get64(p + 21 + 16 * k);
+ if (first > last || !l->flight_count) continue;
+ uint64_t oldest = l->flights[l->flight_head].pn, newest = oldest + l->flight_count - 1;
+ for (uint64_t pn = most(first, oldest); pn <= least(last, newest); ++pn) {
+ Flight *f = flight_of(l, pn);
+ if (!f || f->done) continue;
+ settle(l, f);
+ ++l->counts.acknowledged;
+ for (uint16_t u = 0; u < f->count; ++u) {
+ Record *r = record_of(l, f->units[u].seq);
+ if (r && (*piece_of(r, f->units[u].piece)).state != RECEIVED) {
+ set_piece(l, r, f->units[u].piece, RECEIVED);
+ l->stored_changed = now; /* waits for its save from now */
+ }
+ }
+ if (f->eliciting && f->sent > l->recovery_start)
+ l->window += l->window < l->threshold ? f->bytes : (uint64_t)DATAGRAM * f->bytes / l->window;
+ if (pn == largest) top = f;
+ }
+ }
+ if (!l->acked_any || largest > l->largest_acked) l->largest_acked = largest;
+ l->acked_any = true;
+ if (top && top->eliciting) sample_rtt(l, now - top->sent, delay);
+ l->pto_count = 0;
+ detect_lost(l, now);
+ while (l->flight_count && l->flights[l->flight_head].done) {
+ l->flight_head = (l->flight_head + 1) % l->flight_capacity;
+ --l->flight_count;
+ }
+}
+
+static void on_stored(PureLink *l, const unsigned char *p, size_t size, uint64_t now) {
+ char source[PURE_SOURCE + 1], input[PURE_MODULE_NAME];
+ size_t names = get_names(p, size, source, input);
+ if (!names || size < names + 13 || strcmp(source, l->source) || strcmp(input, l->input)) return;
+ uint64_t heard = get64(p + names);
+ unsigned flags = p[names + 8];
+ uint32_t credit = get32(p + names + 9);
+ if (heard > l->stored) {
+ while (l->count && record_at(l, 0)->seq <= heard) {
+ Record *r = record_at(l, 0);
+ for (uint32_t i = 0; i < r->pieces; ++i) l->received -= (*piece_of(r, i)).state == RECEIVED;
+ free(r->more); /* its bytes go back with the ring's head */
+ l->head = (l->head + 1) % l->capacity;
+ --l->count;
+ }
+ l->stored = heard;
+ l->stored_changed = now;
+ }
+ if (flags & STORED_GAP) /* it did not keep what came after a gap: send it again */
+ for (size_t k = 0; k < l->count && l->received; ++k) {
+ Record *r = record_at(l, k);
+ for (uint32_t i = 0; i < r->pieces; ++i)
+ if ((*piece_of(r, i)).state == RECEIVED) set_piece(l, r, i, UNSENT);
+ }
+ l->credit = credit ? least(credit, 65536) : 1;
+}
+
+static void take_answer(PureLink *l, unsigned char *packet, ssize_t n, uint64_t now);
+
+static void take_answers(PureLink *l, uint64_t now) {
+ for (unsigned rounds = 0, n = TAKE; rounds < 4 && n == TAKE; ++rounds) {
+ n = (unsigned)take_datagrams(l->fd, l->taking);
+ for (unsigned k = 0; k < n; ++k) take_answer(l, l->taking->bytes[k], l->taking->messages[k].msg_len, now);
+ }
+}
+
+static void take_answer(PureLink *l, unsigned char *packet, ssize_t n, uint64_t now) {
+ unsigned char plain[DATAGRAM], nonce[PURE_AEAD_NONCE];
+ if (n < HEADER + PURE_AEAD_TAG || (size_t)n > DATAGRAM || packet[0] != VERSION || get32(packet + 1) != l->peer_id)
+ return; /* not ours, or an error from an earlier send (nobody listening) */
+ uint64_t run = get64(packet + 5), pn = get64(packet + 13);
+ if (!l->in || l->in_run != run) {
+ pure_aead_free(l->in);
+ l->in = aead_of(l->secret, l->peer, l->own_public, run, l->run, l->in_base);
+ l->in_run = run;
+ if (!l->in) return;
+ }
+ nonce_of(nonce, l->in_base, pn);
+ size_t size = (size_t)n - HEADER - PURE_AEAD_TAG;
+ if (!pure_aead_open(l->in, NULL, nonce, packet, HEADER, packet + HEADER, (size_t)n - HEADER, plain)) return;
+ ++l->counts.received;
+ for (size_t at = 0; at < size;) {
+ unsigned type = plain[at++];
+ if (type == F_ACK && at + 13 <= size && at + 13 + 16u * plain[at + 12] <= size) {
+ on_ack(l, plain + at, size - at, now);
+ at += 13 + 16u * plain[at + 12];
+ } else if (type == F_STORED) {
+ char source[PURE_SOURCE + 1], input[PURE_MODULE_NAME];
+ size_t names = get_names(plain + at, size - at, source, input);
+ if (!names || at + names + 13 > size) break;
+ on_stored(l, plain + at, size - at, now);
+ at += names + 13;
+ } else if (type != F_PING)
+ break; /* padding, or what this side does not know: the rest */
+ }
+}
+
+/* A packet of pieces not sent yet, in order, within the receiver's credit:
+ * its body's size, or 0 when there is none. */
+static size_t fill(PureLink *l, unsigned char *body, Flight *f) {
+ size_t at = 0;
+ body[at++] = F_STREAM;
+ at += put_names(body + at, l->source, l->input);
+ put32(body + at, l->shape);
+ at += 4;
+ /* The credit counts from the oldest record not stored: after a restart
+ * the sender does not know yet how far the receiver stored. */
+ uint64_t base = l->count ? record_at(l, 0)->seq - 1 : l->stored;
+ /* From the first record that may have an unsent piece: the ones before
+ * it are sent, and every one found sent moves that mark on. */
+ size_t k = l->count && l->unsent_from > base + 1 ? (size_t)(l->unsent_from - base - 1) : 0;
+ bool before = true;
+ /* A record none of whose pieces went yet goes whole when it fits what is
+ * left of the packet: whole records whose numbers follow on share a
+ * RECORDS frame (its count at run). Else its unsent pieces go, each in a
+ * RECORD frame. */
+ size_t run = 0;
+ uint64_t next = 0;
+ for (; k < l->count && f->count < UNITS; ++k) {
+ Record *r = record_at(l, k);
+ if (r->seq > base + l->credit) break;
+ bool unsent = false, fresh = true, joins = run && r->seq == next;
+ for (uint32_t i = 0; i < r->pieces && fresh; ++i) fresh = piece_of(r, i)->state == UNSENT;
+ size_t whole = (joins ? 0 : RECORDS_HEAD) + varint_size(r->size) + r->size;
+ if (fresh && f->count + r->pieces <= UNITS && at + whole <= BODY) {
+ if (!joins) {
+ body[at] = F_RECORDS;
+ put64(body + at + 1, r->seq);
+ body[at + 9] = 0;
+ run = at + 9;
+ at += RECORDS_HEAD;
+ }
+ at += put_varint(body + at, r->size);
+ if (r->size) memcpy(body + at, l->ring + r->at, r->size);
+ at += r->size;
+ ++body[run];
+ next = r->seq + 1;
+ for (uint32_t i = 0; i < r->pieces; ++i) {
+ set_piece(l, r, i, FLYING);
+ piece_of(r, i)->pn = f->pn;
+ f->units[f->count++] = (Unit){r->seq, i};
+ }
+ ++l->counts.records;
+ } else if (fresh && r->pieces == 1)
+ unsent = true; /* too large for what is left: the next packet */
+ else
+ for (uint32_t i = 0; i < r->pieces; ++i) {
+ if (piece_of(r, i)->state != UNSENT) continue;
+ uint32_t offset = i * PIECE, length = (uint32_t)least(PIECE, r->size - offset);
+ if (f->count == UNITS || at + RECORD_HEAD + length > BODY) {
+ unsent = true;
+ break;
+ }
+ body[at] = F_RECORD;
+ put64(body + at + 1, r->seq);
+ put32(body + at + 9, r->size);
+ put32(body + at + 13, offset);
+ put16(body + at + 17, length);
+ if (length) memcpy(body + at + RECORD_HEAD, l->ring + r->at + offset, length);
+ at += RECORD_HEAD + length;
+ run = 0;
+ set_piece(l, r, i, FLYING);
+ piece_of(r, i)->pn = f->pn;
+ f->units[f->count++] = (Unit){r->seq, i};
+ ++l->counts.records;
+ }
+ if (before && !unsent) l->unsent_from = r->seq + 1;
+ before = before && !unsent;
+ if (unsent) { /* the packet is full: whole, so it can go in a burst */
+ if (l->gso && f->count) {
+ memset(body + at, F_PADDING, BODY - at);
+ at = BODY;
+ }
+ break;
+ }
+ }
+ return f->count ? at : 0;
+}
+
+/* Seal and send f with body; its number is spent even when the socket
+ * refuses it (what it carried is then marked unsent again). */
+/* A packet the socket refused: what it carried is unsent again, and it
+ * counts as never sent. Its number is spent all the same. */
+static void refused(PureLink *l, uint64_t pn) {
+ Flight *f = flight_of(l, pn);
+ if (!f || f->done) return;
+ for (uint16_t u = 0; u < f->count; ++u) {
+ Record *r = record_of(l, f->units[u].seq);
+ if (r && (*piece_of(r, f->units[u].piece)).pn == pn) set_piece(l, r, f->units[u].piece, UNSENT);
+ }
+ settle(l, f);
+ f->eliciting = false;
+ f->count = 0;
+ --l->counts.packets;
+}
+
+static bool transient(void) { return errno == EAGAIN || errno == EWOULDBLOCK || errno == ENOBUFS || errno == EINTR; }
+
+/* The packets of the burst to the socket (each but the last a whole
+ * datagram): with GSO in one call, else one by one. */
+static void flush(PureLink *l) {
+ unsigned n = l->burst_count, sent = 0;
+ bool whole = n > 1 && l->gso;
+ if (whole && send(l->fd, l->burst, l->burst_size, 0) >= 0) sent = n;
+ else if (whole && !transient()) l->gso = false, whole = false; /* no GSO here: one by one from now on */
+ for (size_t at = 0; !whole && sent < n; ++sent) {
+ size_t size = sent + 1 < n ? DATAGRAM : l->burst_size - at;
+ if (send(l->fd, l->burst + at, size, 0) < 0 && transient()) break;
+ at += size;
+ }
+ for (unsigned k = sent; k < n; ++k) refused(l, l->burst_pn[k]);
+ l->burst_count = 0;
+ l->burst_size = 0;
+}
+
+/* Seal f with body into the burst (a whole datagram each but the last), and
+ * count it sent; its number is spent even when the socket then refuses it. */
+static bool transmit(PureLink *l, Flight *f, const unsigned char *body, size_t size, uint64_t now) {
+ if (l->burst_count && l->burst_size != (size_t)l->burst_count * DATAGRAM) flush(l); /* the last was short */
+ size_t total = seal_packet(l->out, NULL, l->out_base, l->own_id, l->run, f->pn, body, size, l->burst + l->burst_size);
+ ++l->next_pn;
+ f->sent = now;
+ f->bytes = (uint32_t)total;
+ if (!total) { /* not sealed: what it carried is unsent again */
+ for (uint16_t u = 0; u < f->count; ++u) {
+ Record *r = record_of(l, f->units[u].seq);
+ if (r && (*piece_of(r, f->units[u].piece)).pn == f->pn) set_piece(l, r, f->units[u].piece, UNSENT);
+ }
+ f->done = true;
+ f->eliciting = false;
+ f->count = 0;
+ push_flight(l, f);
+ return false;
+ }
+ if (!push_flight(l, f)) return false;
+ l->burst_pn[l->burst_count++] = f->pn;
+ l->burst_size += total;
+ ++l->counts.packets;
+ if (f->eliciting) {
+ ++l->eliciting;
+ l->in_flight += f->bytes;
+ l->last_eliciting = now;
+ }
+ if (l->burst_count == BURST) flush(l);
+ return true;
+}
+
+static void send_packets(PureLink *l, uint64_t now) {
+ unsigned char body[BODY];
+ l->queued_new = false; /* all queued is looked at now */
+ for (unsigned n = 0; n < 128; ++n) {
+ bool probe = l->probes > 0;
+ if (!probe && l->in_flight + DATAGRAM > l->window) break;
+ Flight f = {.pn = l->next_pn, .eliciting = true};
+ size_t size = fill(l, body, &f);
+ if (!size) {
+ if (!probe) break;
+ body[0] = F_PING; /* nothing to send again: a probe all the same */
+ size = 1;
+ }
+ if (!transmit(l, &f, body, size, now)) break;
+ if (probe) --l->probes;
+ }
+ flush(l);
+}
+
+static uint64_t pto_period(const PureLink *l) {
+ uint64_t smoothed = l->sampled ? l->smoothed_rtt : FIRST_RTT, var = l->sampled ? l->rtt_var : FIRST_RTT / 2;
+ uint64_t period = smoothed + most(4 * var, GRANULARITY) + ACK_DELAY_MOST;
+ for (unsigned i = 0; i < l->pto_count && period < PTO_MOST; ++i) period *= 2;
+ return least(period, PTO_MOST);
+}
+
+static bool waiting_for_save(PureLink *l) { return l->received > 0; }
+
+/* The next time a timer is due (ns), or 0. */
+static uint64_t next_timer(PureLink *l) {
+ uint64_t due = l->loss_time;
+ if (l->eliciting) {
+ uint64_t pto = l->last_eliciting + pto_period(l);
+ if (!due || pto < due) due = pto;
+ }
+ if (waiting_for_save(l)) {
+ uint64_t stored = l->stored_changed + STORED_WAIT;
+ if (!due || stored < due) due = stored;
+ }
+ return due;
+}
+
+static void run_timers(PureLink *l, uint64_t now) {
+ if (l->loss_time && l->loss_time <= now) detect_lost(l, now);
+ else if (l->eliciting && l->last_eliciting + pto_period(l) <= now) {
+ /* Nothing heard for a while: a probe, what flies longest sent again. */
+ ++l->pto_count;
+ l->probes = 1;
+ for (size_t k = 0; k < l->count; ++k) {
+ Record *r = record_at(l, k);
+ uint32_t i = 0;
+ while (i < r->pieces && (*piece_of(r, i)).state != FLYING) ++i;
+ if (i < r->pieces) {
+ set_piece(l, r, i, UNSENT);
+ ++l->counts.resent;
+ break;
+ }
+ }
+ }
+ if (waiting_for_save(l) && l->stored_changed + STORED_WAIT <= now) {
+ /* Received long ago and still not stored: the receiver lost them (it
+ * restarted) or its word got lost. Sent again: it takes a number once. */
+ for (size_t k = 0; k < l->count && l->received; ++k) {
+ Record *r = record_at(l, k);
+ for (uint32_t i = 0; i < r->pieces; ++i)
+ if ((*piece_of(r, i)).state == RECEIVED) set_piece(l, r, i, UNSENT), ++l->counts.resent;
+ }
+ l->stored_changed = now;
+ }
+}
+
+static void *run_link(void *context) {
+ PureLink *l = context;
+ for (;;) {
+ pthread_mutex_lock(&l->lock);
+ uint64_t now = now_ns(), due = next_timer(l);
+ bool stopping = l->stopping, more = l->queued_new; /* queued while it sent: no sleep */
+ l->sleeping = !more;
+ pthread_mutex_unlock(&l->lock);
+ if (stopping) break;
+ int timeout = more ? 0 : due ? (int)((due > now ? due - now : 0) / MS) + 1 : -1;
+ struct pollfd waits[2] = {{l->fd, POLLIN, 0}, {l->wake, POLLIN, 0}};
+ if (poll(waits, 2, timeout) < 0 && errno != EINTR) break;
+ if (waits[1].revents & POLLIN) {
+ uint64_t signals;
+ if (read(l->wake, &signals, sizeof(signals)) < 0) { /* cleared already */ }
+ }
+ pthread_mutex_lock(&l->lock);
+ l->sleeping = l->signaled = false;
+ now = now_ns();
+ if (waits[0].revents) take_answers(l, now);
+ run_timers(l, now);
+ send_packets(l, now);
+ bool told = l->stored != l->told;
+ l->told = l->stored;
+ pthread_mutex_unlock(&l->lock);
+ if (told && l->notify) l->notify(l->context);
+ }
+ return NULL;
+}
+
+bool pure_link_start(PureLink *l, void (*notify)(void *), void *context, char *error, size_t size) {
+ if (!l->peered) {
+ snprintf(error, size, "pure://%s:%s/%s: no receiver's key (--seal)", l->host, l->port, l->input);
+ return false;
+ }
+ struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_DGRAM}, *found = NULL;
+ int resolved = getaddrinfo(l->host, l->port, &hints, &found);
+ if (resolved) {
+ snprintf(error, size, "%s: %s", l->host, gai_strerror(resolved));
+ return false;
+ }
+ l->fd = socket(found->ai_family, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
+ bool ok = l->fd >= 0 && !connect(l->fd, found->ai_addr, found->ai_addrlen);
+ freeaddrinfo(found);
+ int buffer = 4 << 20;
+ if (ok) {
+ setsockopt(l->fd, SOL_SOCKET, SO_SNDBUF, &buffer, sizeof(buffer));
+ setsockopt(l->fd, SOL_SOCKET, SO_RCVBUF, &buffer, sizeof(buffer));
+ }
+ int segment = DATAGRAM;
+ l->gso = ok && !setsockopt(l->fd, SOL_UDP, UDP_SEGMENT, &segment, sizeof(segment));
+ ok = ok && (l->burst = malloc((size_t)BURST * DATAGRAM)) && (l->taking = malloc(sizeof(*l->taking)));
+ l->run = random64();
+ l->window = WINDOW_FIRST;
+ l->threshold = UINT64_MAX;
+ l->credit = CREDIT;
+ ok = ok && pure_link_secret(l->own, l->peer, l->secret) &&
+ (l->out = aead_of(l->secret, l->own_public, l->peer, l->run, 0, l->out_base)) &&
+ (l->wake = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC)) >= 0;
+ l->notify = notify;
+ l->context = context;
+ if (!ok || pthread_create(&l->thread, NULL, run_link, l)) {
+ snprintf(error, size, "pure://%s:%s/%s: cannot start (a socket, the keys)", l->host, l->port, l->input);
+ return false;
+ }
+ pthread_setname_np(l->thread, "pure-link"); /* a link's CPU, seen apart */
+ l->started = true;
+ return true;
+}
+
+void pure_link_counts(PureLink *l, PureLinkCounts *out) {
+ pthread_mutex_lock(&l->lock);
+ *out = l->counts;
+ out->stored = l->stored;
+ pthread_mutex_unlock(&l->lock);
+}
+
+void pure_link_close(PureLink *l) {
+ if (!l) return;
+ if (l->started) {
+ pthread_mutex_lock(&l->lock);
+ l->stopping = true;
+ pthread_mutex_unlock(&l->lock);
+ wake(l->wake);
+ pthread_join(l->thread, NULL);
+ }
+ for (size_t k = 0; k < l->count; ++k) {
+ free(record_at(l, k)->more);
+ }
+ free(l->records);
+ free(l->ring);
+ free(l->flights);
+ free(l->burst);
+ free(l->taking);
+ pure_aead_free(l->out);
+ pure_aead_free(l->in);
+ explicit_bzero(l->secret, sizeof(l->secret));
+ if (l->fd >= 0) close(l->fd);
+ if (l->wake >= 0) close(l->wake);
+ pthread_mutex_destroy(&l->lock);
+ free(l);
+}
+
+/* ---- The receiving side
+ *
+ * A sender allowed is a peer: its key and the sources it may send as, a few
+ * bytes each, found by its key's id. What a sending peer needs besides (the
+ * secret its key shares with the node's, its runs' keys and addresses, its
+ * streams) is its active state: made at its first packet (one X25519), let
+ * go once the peer has been quiet for the idle period with nothing of it
+ * waiting here. So a link at rest costs the node its allowing, and the next
+ * packet goes on without a handshake. */
+
+typedef struct { uint64_t first, last; } Range;
+typedef struct Active Active;
+typedef struct {
+ Active *active;
+ uint64_t run, seen; /* the sender's run; when it last sent */
+ bool used, ack_due, due;
+ unsigned char in_key[PURE_AEAD_KEY], out_key[PURE_AEAD_KEY], in_base[PURE_AEAD_NONCE], out_base[PURE_AEAD_NONCE];
+ uint64_t next_pn, largest_at;
+ struct sockaddr_in6 address; /* or an IPv4 one */
+ socklen_t address_size;
+ Range got[RANGES]; /* the numbers received, newest first */
+ unsigned got_count;
+} Run;
+typedef struct {
+ uint64_t seq;
+ unsigned char *bytes; /* its size, then a byte a piece: had */
+ uint32_t size, pieces, got;
+} Held;
+typedef struct Stream {
+ Active *active;
+ char source[PURE_SOURCE + 1];
+ uint32_t input;
+ uint64_t heard; /* the last number heard (stored), once known */
+ bool known, out, stored_due, gap, ready;
+ /* The line: once the last heard is known, whole records from the one
+ * after it on, in order, as a delivery carries them (u32 size, bytes),
+ * from line_head: handed over as they are, let go by moving the head.
+ * Records that come early or in pieces, or before the last heard is
+ * known, wait in held until the line reaches them. */
+ unsigned char *line;
+ size_t line_head, line_size, line_capacity;
+ uint64_t line_first;
+ unsigned line_count;
+ Held *held; /* records not yet stored, by number: held_count from held_first */
+ size_t held_first, held_count, held_capacity, held_bytes;
+ Run *run; /* the run that sent to it last: where STORED goes */
+ uint64_t run_id;
+} Stream;
+struct Active {
+ uint32_t peer; /* its place among the peers */
+ unsigned char secret[32];
+ uint64_t seen;
+ size_t place; /* among the active */
+ Run runs[RUNS];
+ Stream **streams;
+ unsigned stream_count, stream_capacity;
+};
+typedef struct {
+ unsigned char key[PURE_SEAL_KEY];
+ uint32_t id, allowed; /* allowed: the place + 1 of its first source (0: none) */
+ Active *active;
+} Peer;
+typedef struct { uint32_t name, next; } Allowed; /* name: where in names (u8 length, bytes); next: a place + 1 */
+typedef struct { Stream *stream; uint64_t heard; bool stored, gap; } Done;
+
+/* A peer quiet this long lets its active state go; the active are looked at
+ * that often; what they held goes back to the system once TRIM_AFTER have
+ * gone, or TRIM_WAIT after any went. */
+static const uint64_t IDLE_FIRST = 30000 * MS, SWEEP_EVERY = 250 * MS, TRIM_WAIT = 10000 * MS;
+enum { TRIM_AFTER = 1024 };
+
+struct PureLinkListener {
+ const PureModule *module;
+ const PureSealKey *own;
+ unsigned char own_public[PURE_SEAL_KEY];
+ uint32_t own_id;
+ unsigned port;
+ uint64_t run, idle, swept, trimmed, untrimmed;
+ PureAead *aead; /* keyed by each packet's run */
+ unsigned char open_key[PURE_AEAD_KEY], seal_key[PURE_AEAD_KEY]; /* the keys it has now */
+ bool open_keyed, seal_keyed;
+ unsigned char *memo; /* the secrets remembered (pure_link_remember), mapped */
+ size_t memo_size;
+ int fd, wake;
+ pthread_t thread;
+ bool started;
+ pthread_mutex_t lock; /* the done list and stopping */
+ Done *done;
+ size_t done_count, done_capacity;
+ bool stopping;
+ Peer *peers;
+ size_t peer_count, peer_capacity;
+ uint32_t *index; /* the peers by key id: open addressing, a place + 1 (0: empty) */
+ size_t index_size;
+ Allowed *allowed;
+ size_t allowed_count, allowed_capacity;
+ unsigned char *names;
+ size_t names_size, names_capacity;
+ Active **actives;
+ size_t active_count, active_capacity;
+ Stream **ready; /* streams that may have whole records to hand over */
+ size_t ready_count, ready_capacity;
+ Run **due; /* runs to answer */
+ size_t due_count, due_capacity;
+ bool (*deliver)(void *, const PureLinkDelivery *);
+ void *context;
+ unsigned char *batch;
+ size_t batch_capacity;
+ PureLinkCounts counts;
+};
+
+PureLinkListener *pure_link_listen(const PureModule *m, unsigned port, const PureSealKey *own,
+ const unsigned char own_public[PURE_SEAL_KEY],
+ bool (*deliver)(void *, const PureLinkDelivery *), void *context, char *error,
+ size_t size) {
+ PureLinkListener *s = own ? calloc(1, sizeof(*s)) : NULL;
+ if (!s || pthread_mutex_init(&s->lock, NULL)) {
+ free(s);
+ snprintf(error, size, own ? "out of memory" : "--link-port needs the node's --key");
+ return NULL;
+ }
+ s->module = m;
+ s->own = own;
+ memcpy(s->own_public, own_public, PURE_SEAL_KEY);
+ s->own_id = key_id(own_public);
+ s->port = port;
+ s->idle = IDLE_FIRST;
+ s->deliver = deliver;
+ s->context = context;
+ s->fd = s->wake = -1;
+ return s;
+}
+
+void pure_link_idle(PureLinkListener *s, uint64_t ms) { s->idle = ms * MS; }
+
+/* items (count used of capacity) with room for one more: the same or grown,
+ * or NULL when memory runs out (items stay as they were). */
+static void *room(void *items, size_t *capacity, size_t count, size_t item_size, size_t first) {
+ if (count < *capacity) return items;
+ size_t more = *capacity ? 2 * *capacity : first;
+ void *grown = realloc(items, more * item_size);
+ if (grown) *capacity = more;
+ return grown;
+}
+
+/* The peers by key id again, in an index twice as large. */
+static bool grow_index(PureLinkListener *s) {
+ size_t size = s->index_size ? 2 * s->index_size : 1024;
+ uint32_t *index = calloc(size, sizeof(*index));
+ if (!index) return false;
+ for (size_t p = 0; p < s->peer_count; ++p) {
+ size_t slot = s->peers[p].id & (size - 1);
+ while (index[slot]) slot = (slot + 1) & (size - 1);
+ index[slot] = (uint32_t)p + 1;
+ }
+ free(s->index);
+ s->index = index;
+ s->index_size = size;
+ return true;
+}
+
+static bool same_name(const PureLinkListener *s, uint32_t name, const char *source, size_t length) {
+ return s->names[name] == length && !memcmp(s->names + name + 1, source, length);
+}
+
+/* Keys are found by their ids, which are 32 bits: two keys may share one. */
+bool pure_link_allow(PureLinkListener *s, const char *source, const unsigned char key[PURE_SEAL_KEY], char *error,
+ size_t size) {
+ size_t length = strlen(source), p = SIZE_MAX;
+ uint32_t id = key_id(key);
+ if (!length || length > PURE_SOURCE) {
+ snprintf(error, size, "a source's name is 1 to %d bytes", PURE_SOURCE);
+ return false;
+ }
+ for (size_t slot = s->index_size ? id & (s->index_size - 1) : 0; s->index_size && s->index[slot];
+ slot = (slot + 1) & (s->index_size - 1)) {
+ const Peer *peer = &s->peers[s->index[slot] - 1];
+ if (peer->id == id && !memcmp(peer->key, key, PURE_SEAL_KEY)) {
+ p = s->index[slot] - 1;
+ break;
+ }
+ }
+ if (p == SIZE_MAX) {
+ Peer *peers = s->peer_count < UINT32_MAX - 1 && ((s->peer_count + 1) * 2 <= s->index_size || grow_index(s)) ?
+ room(s->peers, &s->peer_capacity, s->peer_count, sizeof(Peer), 64) : NULL;
+ if (!peers) goto full;
+ s->peers = peers;
+ p = s->peer_count++;
+ s->peers[p] = (Peer){.id = id};
+ memcpy(s->peers[p].key, key, PURE_SEAL_KEY);
+ size_t slot = id & (s->index_size - 1);
+ while (s->index[slot]) slot = (slot + 1) & (s->index_size - 1);
+ s->index[slot] = (uint32_t)p + 1;
+ }
+ for (uint32_t a = s->peers[p].allowed; a; a = s->allowed[a - 1].next)
+ if (same_name(s, s->allowed[a - 1].name, source, length)) return true;
+ Allowed *kept = s->allowed_count < UINT32_MAX - 1 && s->names_size + 1 + length <= UINT32_MAX ?
+ room(s->allowed, &s->allowed_capacity, s->allowed_count, sizeof(Allowed), 64) : NULL;
+ if (!kept) goto full;
+ s->allowed = kept;
+ if (s->names_size + 1 + length > s->names_capacity) {
+ size_t more = s->names_capacity ? 2 * s->names_capacity : 4096;
+ unsigned char *names = realloc(s->names, more);
+ if (!names) goto full;
+ s->names = names;
+ s->names_capacity = more;
+ }
+ s->names[s->names_size] = (unsigned char)length;
+ memcpy(s->names + s->names_size + 1, source, length);
+ s->allowed[s->allowed_count] = (Allowed){(uint32_t)s->names_size, s->peers[p].allowed};
+ s->peers[p].allowed = (uint32_t)++s->allowed_count;
+ s->names_size += 1 + length;
+ return true;
+full:
+ snprintf(error, size, "out of memory");
+ return false;
+}
+
+static Held *held_at(Stream *st, size_t k) { return &st->held[st->held_first + k]; }
+
+static uint64_t line_end(const Stream *st) { return st->line_first + st->line_count; }
+
+/* Room for more bytes at the line's end: its front let go given back
+ * first when that is at least half of it (so no byte moves twice in a
+ * doubling), else a longer line. */
+static bool line_room(Stream *st, size_t more) {
+ if (st->line_size + more <= st->line_capacity) return true;
+ if (st->line_head && st->line_head >= st->line_capacity / 2) {
+ memmove(st->line, st->line + st->line_head, st->line_size - st->line_head);
+ st->line_size -= st->line_head;
+ st->line_head = 0;
+ if (st->line_size + more <= st->line_capacity) return true;
+ }
+ size_t capacity = st->line_capacity ? st->line_capacity : 4096;
+ while (capacity < st->line_size + more) capacity *= 2;
+ unsigned char *grown = realloc(st->line, capacity);
+ if (!grown) return false;
+ st->line = grown;
+ st->line_capacity = capacity;
+ return true;
+}
+
+static bool line_append(Stream *st, const unsigned char *bytes, uint32_t size) {
+ if (!line_room(st, 4 + (size_t)size)) return false;
+ put32(st->line + st->line_size, size);
+ if (size) memcpy(st->line + st->line_size + 4, bytes, size);
+ st->line_size += 4 + (size_t)size;
+ ++st->line_count;
+ return true;
+}
+
+/* The line's records up to heard let go (all of them, when heard is past it). */
+static void line_drop(Stream *st, uint64_t heard) {
+ if (heard < st->line_first) return;
+ if (heard >= line_end(st)) {
+ st->line_head = st->line_size = 0;
+ st->line_count = 0;
+ st->line_first = heard + 1;
+ return;
+ }
+ for (uint64_t n = heard - st->line_first + 1; n; --n, --st->line_count, ++st->line_first)
+ st->line_head += 4 + (size_t)get32(st->line + st->line_head);
+}
+
+/* Records held that the line reached, moved onto it; ones behind it let go. */
+static void drop_first(Stream *st);
+static void line_pull(Stream *st) {
+ while (st->held_count) {
+ Held *h = held_at(st, 0);
+ if (h->seq < line_end(st)) drop_first(st);
+ else if (h->seq == line_end(st) && h->got == h->pieces && line_append(st, h->bytes, h->size)) drop_first(st);
+ else break;
+ }
+}
+
+/* The first record held let go: the ones after it stay where they are. */
+static void drop_first(Stream *st) {
+ st->held_bytes -= held_at(st, 0)->size;
+ free(held_at(st, 0)->bytes);
+ ++st->held_first;
+ if (!--st->held_count) st->held_first = 0;
+}
+
+/* The memo of secrets (pure_link_remember): a header, then a place a peer
+ * in the order allowed, each the first 28 bytes of its key, a check (FNV-1a
+ * of its whole key and the secret) and the secret. A place whose check
+ * fails (never made, another key, cut short by a crash) is made again. */
+enum { MEMO_HEAD = 64, MEMO_PLACE = 64 };
+static const unsigned char memo_magic[8] = {'P', 'L', 'M', '1', 0, 0, 0, 0};
+
+static uint32_t memo_check(const unsigned char key[PURE_SEAL_KEY], const unsigned char secret[32]) {
+ return fnv(fnv(2166136261u, key, PURE_SEAL_KEY), secret, 32);
+}
+
+bool pure_link_remember(PureLinkListener *s, const char *path, char *error, size_t size) {
+ int fd = open(path, O_RDWR | O_CREAT | O_NOFOLLOW | O_CLOEXEC, 0600);
+ struct stat info;
+ unsigned char head[MEMO_HEAD];
+ size_t needed = MEMO_HEAD + (s->peer_count ? s->peer_count : 1) * MEMO_PLACE;
+ bool ok = fd >= 0 && !fstat(fd, &info) && S_ISREG(info.st_mode) && !(info.st_mode & 077);
+ /* Another node's key, or none yet: begun again. */
+ bool ours = ok && pread(fd, head, MEMO_HEAD, 0) == MEMO_HEAD && !memcmp(head, memo_magic, 8) &&
+ !memcmp(head + 8, s->own_public, PURE_SEAL_KEY);
+ if (ok && !ours) {
+ memset(head, 0, MEMO_HEAD);
+ memcpy(head, memo_magic, 8);
+ memcpy(head + 8, s->own_public, PURE_SEAL_KEY);
+ ok = !ftruncate(fd, 0) && pwrite(fd, head, MEMO_HEAD, 0) == MEMO_HEAD;
+ }
+ ok = ok && (ours && (size_t)info.st_size >= needed ? true : !ftruncate(fd, (off_t)needed));
+ void *map = ok ? mmap(NULL, needed, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0) : MAP_FAILED;
+ if (fd >= 0) close(fd);
+ if (map == MAP_FAILED) {
+ snprintf(error, size, "%s: cannot keep secrets there (a file only its owner may read)", path);
+ return false;
+ }
+ s->memo = map;
+ s->memo_size = needed;
+ return true;
+}
+
+/* A peer's active state, made at its first packet: the secret its key
+ * shares with ours, remembered or made (one X25519, then remembered: the
+ * secret first, the key that makes the place its own last). */
+static Active *activate(PureLinkListener *s, Peer *peer, uint64_t now) {
+ Active *a = calloc(1, sizeof(*a)), **actives = room(s->actives, &s->active_capacity, s->active_count, sizeof(*actives), 64);
+ if (actives) s->actives = actives;
+ if (!a || !actives) {
+ free(a);
+ return NULL;
+ }
+ size_t index = (size_t)(peer - s->peers);
+ unsigned char *place = s->memo && MEMO_HEAD + (index + 1) * MEMO_PLACE <= s->memo_size ?
+ s->memo + MEMO_HEAD + index * MEMO_PLACE : NULL;
+ if (place && !memcmp(place, peer->key, 28) && get32(place + 28) == memo_check(peer->key, place + 32)) {
+ memcpy(a->secret, place + 32, 32);
+ ++s->counts.remembered;
+ } else if (pure_link_secret(s->own, peer->key, a->secret)) {
+ if (place) {
+ memset(place, 0, 28);
+ __atomic_thread_fence(__ATOMIC_RELEASE);
+ memcpy(place + 32, a->secret, 32);
+ put32(place + 28, memo_check(peer->key, a->secret));
+ __atomic_thread_fence(__ATOMIC_RELEASE);
+ memcpy(place, peer->key, 28);
+ }
+ } else {
+ explicit_bzero(a, sizeof(*a));
+ free(a);
+ return NULL;
+ }
+ a->peer = (uint32_t)(peer - s->peers);
+ a->seen = now;
+ a->place = s->active_count;
+ s->actives[s->active_count++] = a;
+ peer->active = a;
+ return a;
+}
+
+static void let_go(PureLinkListener *s, Active *a) {
+ for (unsigned k = 0; k < a->stream_count; ++k) {
+ while (a->streams[k]->held_count) drop_first(a->streams[k]);
+ free(a->streams[k]->held);
+ free(a->streams[k]->line);
+ free(a->streams[k]);
+ }
+ free(a->streams);
+ s->peers[a->peer].active = NULL;
+ s->actives[a->place] = s->actives[--s->active_count];
+ s->actives[a->place]->place = a->place;
+ explicit_bzero(a, sizeof(*a));
+ free(a);
+}
+
+/* Whether any of a peer's records or answers waits here. */
+static bool busy(const Active *a) {
+ for (unsigned k = 0; k < a->stream_count; ++k)
+ if (a->streams[k]->out || a->streams[k]->held_count || a->streams[k]->line_count || a->streams[k]->ready) return true;
+ for (unsigned i = 0; i < RUNS; ++i)
+ if (a->runs[i].due) return true;
+ return false;
+}
+
+/* Let go the active state of peers quiet for the idle period; what idle
+ * links held goes back to the system. */
+static void sweep(PureLinkListener *s, uint64_t now) {
+ if (now - s->swept < SWEEP_EVERY) return;
+ s->swept = now;
+ size_t freed = 0;
+ for (size_t k = 0; k < s->active_count;) {
+ Active *a = s->actives[k];
+ if (now - a->seen < s->idle || busy(a)) ++k;
+ else let_go(s, a), ++freed;
+ }
+ s->counts.swept += freed;
+ s->untrimmed += freed;
+ if (s->untrimmed >= TRIM_AFTER || (s->untrimmed && now - s->trimmed >= TRIM_WAIT)) {
+ malloc_trim(0);
+ s->untrimmed = 0;
+ s->trimmed = now;
+ }
+}
+
+static void mark_ready(PureLinkListener *s, Stream *st) {
+ Stream **ready = st->ready ? NULL : room(s->ready, &s->ready_capacity, s->ready_count, sizeof(*ready), 64);
+ if (!ready) return;
+ s->ready = ready;
+ s->ready[s->ready_count++] = st;
+ st->ready = true;
+}
+
+static void mark_due(PureLinkListener *s, Run *r) {
+ Run **due = r->due ? NULL : room(s->due, &s->due_capacity, s->due_count, sizeof(*due), 64);
+ if (!due) return;
+ s->due = due;
+ s->due[s->due_count++] = r;
+ r->due = true;
+}
+
+static Run *run_found(Active *a, uint64_t run) {
+ for (unsigned i = 0; i < RUNS; ++i)
+ if (a->runs[i].used && a->runs[i].run == run) return &a->runs[i];
+ return NULL;
+}
+
+/* A new run's keys, in the slot unused longest: only once its first packet
+ * opened with them, so a forged one displaces nothing. */
+static Run *install_run(Active *a, const Run *made) {
+ Run *slot = &a->runs[0];
+ for (unsigned i = 0; i < RUNS; ++i) {
+ Run *r = &a->runs[i];
+ if (!r->used || (slot->used && r->seen < slot->seen)) slot = r;
+ }
+ bool due = slot->due; /* the answer list may hold the slot */
+ *slot = *made;
+ slot->due = due;
+ return slot;
+}
+
+static bool had(const Run *r, uint64_t pn) {
+ for (unsigned i = 0; i < r->got_count; ++i)
+ if (pn >= r->got[i].first && pn <= r->got[i].last) return true;
+ return false;
+}
+
+static void note(Run *r, uint64_t pn, uint64_t now) {
+ unsigned i = 0, n = r->got_count;
+ while (i < n && r->got[i].first > pn + 1) ++i;
+ if (i < n && r->got[i].first == pn + 1) { /* joins the range above it */
+ r->got[i].first = pn;
+ if (i + 1 < n && r->got[i + 1].last + 1 == pn) {
+ r->got[i].first = r->got[i + 1].first;
+ memmove(&r->got[i + 1], &r->got[i + 2], (n - i - 2) * sizeof(Range));
+ --r->got_count;
+ }
+ } else if (i < n && r->got[i].last + 1 == pn) {
+ r->got[i].last = pn;
+ } else if (i < RANGES) {
+ if (n == RANGES) --n; /* the oldest range is forgotten */
+ memmove(&r->got[i + 1], &r->got[i], (n - i) * sizeof(Range));
+ r->got[i] = (Range){pn, pn};
+ r->got_count = n + 1;
+ }
+ if (r->got_count && r->got[0].last == pn) r->largest_at = now;
+}
+
+static bool allowed(const PureLinkListener *s, const Peer *peer, const char *source) {
+ for (uint32_t a = peer->allowed; a; a = s->allowed[a - 1].next)
+ if (same_name(s, s->allowed[a - 1].name, source, strlen(source))) return true;
+ return false;
+}
+
+/* The stream of a peer's source to input, made the first time: NULL when the
+ * source is not allowed from the peer, or input takes no such messages. */
+static Stream *stream_of(PureLinkListener *s, Active *a, const char *source, const char *input, uint32_t shape) {
+ int index = pure_module_input(s->module, input);
+ if (index < 0) return NULL;
+ for (unsigned k = 0; k < a->stream_count; ++k) {
+ Stream *st = a->streams[k];
+ if (st->input == (uint32_t)index && !strcmp(st->source, source)) return st;
+ }
+ const PureInputShape *in = &s->module->input[index];
+ if (!allowed(s, &s->peers[a->peer], source) || in->kind != PURE_KIND_MESSAGE || in->keyed ||
+ shape_of(s->module, (uint32_t)index) != shape)
+ return NULL;
+ size_t capacity = a->stream_capacity;
+ Stream **streams = room(a->streams, &capacity, a->stream_count, sizeof(*streams), 2), *st;
+ if (!streams) return NULL;
+ a->streams = streams;
+ a->stream_capacity = (unsigned)capacity;
+ if (!(st = calloc(1, sizeof(*st)))) return NULL;
+ st->active = a;
+ st->input = (uint32_t)index;
+ snprintf(st->source, sizeof(st->source), "%s", source);
+ a->streams[a->stream_count++] = st;
+ return st;
+}
+
+/* A record's piece, kept until the record is whole and handed over. */
+static void hold(Stream *st, uint64_t seq, uint32_t size, uint32_t offset, const unsigned char *bytes, uint32_t length) {
+ if (st->known && seq <= st->heard) {
+ st->stored_due = true; /* sent again: say again how far it is stored */
+ return;
+ }
+ if ((st->known && seq > st->heard + CREDIT) || size > RECORD_MOST) return;
+ if (st->known && seq < line_end(st)) return; /* on the line already */
+ if (st->known && seq == line_end(st) && !offset && length == size) { /* the next, whole */
+ if (st->held_bytes + (st->line_size - st->line_head) + size <= HELD_MOST && line_append(st, bytes, size))
+ line_pull(st);
+ return;
+ }
+ /* In order it goes last; out of order, its place is found by halving. */
+ size_t k = st->held_count;
+ if (k && held_at(st, k - 1)->seq >= seq) {
+ size_t low = 0;
+ while (low < k) {
+ size_t middle = low + (k - low) / 2;
+ if (held_at(st, middle)->seq < seq) low = middle + 1;
+ else k = middle;
+ }
+ }
+ if (k == st->held_count || held_at(st, k)->seq != seq) {
+ if (st->held_bytes + size > HELD_MOST) return;
+ if (st->held_first && st->held_first + st->held_count == st->held_capacity) { /* the front is free: used again */
+ memmove(st->held, held_at(st, 0), st->held_count * sizeof(Held));
+ st->held_first = 0;
+ }
+ Held *kept = room(st->held, &st->held_capacity, st->held_first + st->held_count, sizeof(Held), 16);
+ if (!kept) return;
+ st->held = kept;
+ Held h = {.seq = seq, .size = size, .pieces = size ? (size + PIECE - 1) / PIECE : 1};
+ if (!(h.bytes = malloc(size + h.pieces))) return;
+ memset(h.bytes + size, 0, h.pieces);
+ memmove(held_at(st, k + 1), held_at(st, k), (st->held_count - k) * sizeof(Held));
+ *held_at(st, k) = h;
+ ++st->held_count;
+ st->held_bytes += size;
+ }
+ Held *h = held_at(st, k);
+ if (h->size != size) return;
+ if (!offset && length == size) { /* whole: every piece had */
+ if (size) memcpy(h->bytes, bytes, size);
+ memset(h->bytes + size, 1, h->pieces);
+ h->got = h->pieces;
+ return;
+ }
+ uint32_t piece = offset / PIECE;
+ if (offset % PIECE || piece >= h->pieces || length != least(PIECE, size - offset) || h->bytes[size + piece]) return;
+ if (length) memcpy(h->bytes + offset, bytes, length);
+ h->bytes[size + piece] = 1;
+ ++h->got;
+}
+
+/* Whole records from the one after the last heard, in order, to the host:
+ * one delivery a stream at a time. Before the last heard is known (after a
+ * restart) they go from the first held: the host says how far it heard.
+ * False when the host could not take them now. */
+static bool hand_over(PureLinkListener *s, Stream *st) {
+ if (st->out) return true;
+ if (st->known) { /* the line as it is, up to BATCH records */
+ line_pull(st);
+ if (!st->line_count) return true;
+ unsigned n = 0;
+ size_t bytes = 0;
+ for (; n < st->line_count && n < BATCH; ++n) bytes += 4 + (size_t)get32(st->line + st->line_head + bytes);
+ PureLinkDelivery d = {st->input, st->source, st->line_first, n, st->line + st->line_head, bytes, st};
+ if (!s->deliver(s->context, &d)) return false;
+ st->out = true;
+ ++s->counts.deliveries;
+ return true;
+ }
+ if (!st->held_count || (st->known && held_at(st, 0)->seq != st->heard + 1)) return true;
+ size_t n = 0, bytes = 0;
+ while (n < st->held_count && n < BATCH && held_at(st, n)->got == held_at(st, n)->pieces &&
+ held_at(st, n)->seq == held_at(st, 0)->seq + n)
+ bytes += 4 + held_at(st, n++)->size;
+ if (!n) return true;
+ if (bytes > s->batch_capacity) {
+ unsigned char *grown = realloc(s->batch, bytes);
+ if (!grown) return false;
+ s->batch = grown;
+ s->batch_capacity = bytes;
+ }
+ size_t at = 0;
+ for (size_t k = 0; k < n; ++k) {
+ put32(s->batch + at, held_at(st, k)->size);
+ memcpy(s->batch + at + 4, held_at(st, k)->bytes, held_at(st, k)->size);
+ at += 4 + held_at(st, k)->size;
+ }
+ PureLinkDelivery d = {st->input, st->source, held_at(st, 0)->seq, (unsigned)n, s->batch, bytes, st};
+ if (!s->deliver(s->context, &d)) return false;
+ st->out = true;
+ ++s->counts.deliveries;
+ return true;
+}
+
+/* A packet from peer, if it opens with the keys of peer's run (made for a
+ * run not seen yet, and for a peer at rest its active state): its frames. */
+static bool take_from(PureLinkListener *s, Peer *peer, unsigned char *packet, size_t n,
+ const struct sockaddr_storage *from, socklen_t from_size, uint64_t now) {
+ uint64_t pn = get64(packet + 13), run = get64(packet + 5);
+ Active *a = peer->active;
+ Run *r = a ? run_found(a, run) : NULL;
+ if (r && had(r, pn)) {
+ r->ack_due = true; /* its acknowledgment was lost, perhaps */
+ mark_due(s, r);
+ return true;
+ }
+ bool woke = !a;
+ if (woke && !(a = activate(s, peer, now))) return false;
+ Run made = {.active = a, .run = run, .seen = now, .used = true};
+ bool keyed = r || (derive(a->secret, peer->key, s->own_public, run, 0, made.in_key, made.in_base) &&
+ derive(a->secret, s->own_public, peer->key, s->run, run, made.out_key, made.out_base));
+ unsigned char plain[DATAGRAM], nonce[PURE_AEAD_NONCE];
+ nonce_of(nonce, r ? r->in_base : made.in_base, pn);
+ size_t size = n - HEADER - PURE_AEAD_TAG;
+ const unsigned char *key = r ? r->in_key : made.in_key, *give = key;
+ if (s->open_keyed && !memcmp(s->open_key, key, PURE_AEAD_KEY)) give = NULL; /* the key it has */
+ else if (keyed) memcpy(s->open_key, key, PURE_AEAD_KEY), s->open_keyed = true;
+ if (!keyed || !pure_aead_open(s->aead, give, nonce, packet, HEADER, packet + HEADER, n - HEADER, plain)) {
+ explicit_bzero(&made, sizeof(made));
+ if (woke) let_go(s, a);
+ return false;
+ }
+ if (woke) ++s->counts.woken;
+ if (!r) r = install_run(a, &made);
+ a->seen = r->seen = now;
+ note(r, pn, now);
+ r->address_size = from_size <= sizeof(r->address) ? from_size : sizeof(r->address);
+ memcpy(&r->address, from, r->address_size); /* where it is now: answers go there */
+ ++s->counts.received;
+ Stream *st = NULL;
+ bool eliciting = false;
+ for (size_t at = 0; at < size;) {
+ unsigned type = plain[at++];
+ if (type == F_STREAM) {
+ char source[PURE_SOURCE + 1], input[PURE_MODULE_NAME];
+ size_t names = get_names(plain + at, size - at, source, input);
+ if (!names || at + names + 4 > size) break;
+ st = stream_of(s, a, source, input, get32(plain + at + names));
+ if (st) {
+ st->run = r;
+ st->run_id = r->run;
+ }
+ at += names + 4;
+ eliciting = true;
+ } else if (type == F_RECORD) {
+ if (at + RECORD_HEAD - 1 > size) break;
+ uint32_t length = get16(plain + at + 16);
+ if (at + RECORD_HEAD - 1 + length > size) break;
+ if (st) {
+ hold(st, get64(plain + at), get32(plain + at + 8), get32(plain + at + 12), plain + at + 18, length);
+ mark_ready(s, st);
+ }
+ at += RECORD_HEAD - 1 + length;
+ eliciting = true;
+ } else if (type == F_RECORDS) {
+ if (at + RECORDS_HEAD - 1 > size) break;
+ uint64_t first = get64(plain + at), length = 0;
+ unsigned count = plain[at + 8];
+ at += RECORDS_HEAD - 1;
+ unsigned i = 0;
+ for (size_t used; i < count && (used = get_varint(plain + at, size - at, &length)) &&
+ length <= size - at - used; ++i) {
+ if (st) hold(st, first + i, (uint32_t)length, 0, plain + at + used, (uint32_t)length);
+ at += used + length;
+ }
+ if (st) mark_ready(s, st);
+ eliciting = true;
+ if (i < count) break; /* cut short: the rest is not read */
+ } else if (type == F_PING) {
+ eliciting = true;
+ } else
+ break; /* padding, or what this side does not know: the rest */
+ }
+ if (eliciting) {
+ r->ack_due = true;
+ mark_due(s, r);
+ }
+ return true;
+}
+
+static void take_packet(PureLinkListener *s, unsigned char *packet, size_t n, const struct sockaddr_storage *from,
+ socklen_t from_size, uint64_t now) {
+ if (n < HEADER + PURE_AEAD_TAG || n > DATAGRAM || packet[0] != VERSION || !s->index_size) return;
+ uint32_t id = get32(packet + 1);
+ size_t mask = s->index_size - 1;
+ uint64_t run = get64(packet + 5);
+ /* The peers with the packet's key id: first one whose run is known (no
+ * key to make), then any. */
+ for (size_t slot = id & mask; s->index[slot]; slot = (slot + 1) & mask) {
+ Peer *peer = &s->peers[s->index[slot] - 1];
+ if (peer->id == id && peer->active && run_found(peer->active, run) &&
+ take_from(s, peer, packet, n, from, from_size, now))
+ return;
+ }
+ for (size_t slot = id & mask; s->index[slot]; slot = (slot + 1) & mask) {
+ Peer *peer = &s->peers[s->index[slot] - 1];
+ if (peer->id == id && !(peer->active && run_found(peer->active, run)) &&
+ take_from(s, peer, packet, n, from, from_size, now))
+ return;
+ }
+}
+
+/* Acknowledgments and how far each stream is stored, a packet to each run
+ * that has any. How far a stream is stored goes with every acknowledgment
+ * to its run, so a lost word is said again at the next packet. */
+static void answer(PureLinkListener *s, uint64_t now) {
+ unsigned char body[BODY], packet[DATAGRAM];
+ for (size_t k = 0; k < s->due_count; ++k) {
+ Run *r = s->due[k];
+ Active *a = r->active;
+ r->due = false;
+ if (!r->used) continue;
+ size_t at = 0;
+ if (r->ack_due && r->got_count) {
+ body[at++] = F_ACK;
+ put64(body + at, r->got[0].last);
+ put32(body + at + 8, (uint32_t)least((now - r->largest_at) / 1000, UINT32_MAX));
+ body[at + 12] = (unsigned char)r->got_count;
+ at += 13;
+ for (unsigned i = 0; i < r->got_count; ++i, at += 16) {
+ put64(body + at, r->got[i].first);
+ put64(body + at + 8, r->got[i].last);
+ }
+ }
+ for (unsigned j = 0; j < a->stream_count; ++j) {
+ Stream *st = a->streams[j];
+ if (!(st->stored_due || r->ack_due) || st->run != r || st->run_id != r->run || !st->known) continue;
+ if (at + 2 + PURE_SOURCE + PURE_MODULE_NAME + 14 > BODY) break;
+ body[at++] = F_STORED;
+ at += put_names(body + at, st->source, s->module->input_names[st->input]);
+ put64(body + at, st->heard);
+ body[at + 8] = st->gap ? STORED_GAP : 0;
+ size_t room = st->held_bytes + (st->line_size - st->line_head) < HELD_MOST ? CREDIT : 1;
+ put32(body + at + 9, (uint32_t)room);
+ at += 13;
+ st->stored_due = st->gap = false;
+ }
+ r->ack_due = false;
+ if (!at) continue;
+ const unsigned char *give = r->out_key;
+ if (s->seal_keyed && !memcmp(s->seal_key, r->out_key, PURE_AEAD_KEY)) give = NULL; /* the key it has */
+ else memcpy(s->seal_key, r->out_key, PURE_AEAD_KEY), s->seal_keyed = true;
+ size_t total = seal_packet(s->aead, give, r->out_base, s->own_id, s->run, r->next_pn++, body, at, packet);
+ if (total && sendto(s->fd, packet, total, 0, (const struct sockaddr *)&r->address, r->address_size) >= 0)
+ ++s->counts.packets;
+ }
+ s->due_count = 0;
+}
+
+static void take_done(PureLinkListener *s) {
+ pthread_mutex_lock(&s->lock);
+ Done *done = s->done;
+ size_t count = s->done_count;
+ s->done = NULL;
+ s->done_count = s->done_capacity = 0;
+ pthread_mutex_unlock(&s->lock);
+ for (size_t k = 0; k < count; ++k) {
+ Stream *st = done[k].stream;
+ st->out = false;
+ mark_ready(s, st);
+ if (!done[k].stored) continue; /* not saved: handed over again */
+ if (!st->known || done[k].heard > st->heard) st->heard = done[k].heard;
+ if (!st->known) st->line_first = st->heard + 1; /* the line begins */
+ line_drop(st, st->heard);
+ while (st->held_count && held_at(st, 0)->seq <= st->heard) drop_first(st);
+ st->known = true;
+ st->gap = done[k].gap;
+ st->stored_due = true;
+ if (st->run && st->run->used && st->run_id == st->run->run) mark_due(s, st->run);
+ s->counts.stored = st->heard;
+ }
+ free(done);
+}
+
+/* The streams ready, each handed over; one the host could not take now is
+ * tried again at the next turn. */
+static void hand_over_ready(PureLinkListener *s) {
+ size_t count = s->ready_count;
+ for (size_t k = 0; k < count; ++k) {
+ Stream *st = s->ready[k];
+ st->ready = false;
+ if (!hand_over(s, st)) mark_ready(s, st);
+ }
+ memmove(s->ready, s->ready + count, (s->ready_count - count) * sizeof(*s->ready));
+ s->ready_count -= count;
+}
+
+static void *run_listener(void *context) {
+ PureLinkListener *s = context;
+ Gathering *gathering = malloc(sizeof(*gathering));
+ if (!gathering) return NULL;
+ for (;;) {
+ struct pollfd waits[2] = {{s->fd, POLLIN, 0}, {s->wake, POLLIN, 0}};
+ int timeout = s->active_count || s->ready_count ? (int)(SWEEP_EVERY / MS) : -1;
+ if (poll(waits, 2, timeout) < 0 && errno != EINTR) break;
+ pthread_mutex_lock(&s->lock);
+ bool stopping = s->stopping;
+ pthread_mutex_unlock(&s->lock);
+ if (stopping) break;
+ if (waits[1].revents & POLLIN) {
+ uint64_t signals;
+ if (read(s->wake, &signals, sizeof(signals)) < 0) { /* cleared already */ }
+ }
+ uint64_t now = now_ns();
+ for (unsigned rounds = 0, n = GATHER; rounds < 8 && n == GATHER; ++rounds) {
+ n = (unsigned)gather(s->fd, gathering);
+ for (unsigned k = 0; k < n; ++k) {
+ struct mmsghdr *m = &gathering->messages[k];
+ size_t length = m->msg_len, step = segment_of(&m->msg_hdr, length);
+ for (size_t at = 0; at < length; at += step)
+ take_packet(s, gathering->bytes[k] + at, least(step, length - at), &gathering->from[k],
+ m->msg_hdr.msg_namelen, now);
+ }
+ }
+ take_done(s);
+ hand_over_ready(s);
+ answer(s, now);
+ sweep(s, now);
+ }
+ free(gathering);
+ return NULL;
+}
+
+bool pure_link_serve(PureLinkListener *s, char *error, size_t size) {
+ struct sockaddr_in6 six = {.sin6_family = AF_INET6, .sin6_port = htons((uint16_t)s->port), .sin6_addr = in6addr_any};
+ int off = 0, buffer = 4 << 20;
+ s->fd = socket(AF_INET6, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
+ bool bound = s->fd >= 0 && !setsockopt(s->fd, IPPROTO_IPV6, IPV6_V6ONLY, &off, sizeof(off)) &&
+ !bind(s->fd, (struct sockaddr *)&six, sizeof(six));
+ if (!bound) { /* no IPv6 here */
+ if (s->fd >= 0) close(s->fd);
+ struct sockaddr_in four = {.sin_family = AF_INET, .sin_port = htons((uint16_t)s->port), .sin_addr = {INADDR_ANY}};
+ s->fd = socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
+ bound = s->fd >= 0 && !bind(s->fd, (struct sockaddr *)&four, sizeof(four));
+ }
+ if (!bound) {
+ snprintf(error, size, "--link-port %u: %s", s->port, strerror(errno));
+ return false;
+ }
+ setsockopt(s->fd, SOL_SOCKET, SO_RCVBUF, &buffer, sizeof(buffer));
+ setsockopt(s->fd, SOL_SOCKET, SO_SNDBUF, &buffer, sizeof(buffer));
+ int one = 1; /* without GRO, a buffer is a datagram */
+ setsockopt(s->fd, SOL_UDP, UDP_GRO, &one, sizeof(one));
+ s->run = random64();
+ s->swept = s->trimmed = now_ns();
+ s->wake = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
+ unsigned char key[PURE_AEAD_KEY] = {0}; /* each packet brings its run's */
+ if (s->wake < 0 || !(s->aead = pure_aead(key)) || pthread_create(&s->thread, NULL, run_listener, s)) {
+ snprintf(error, size, "--link-port %u: cannot start", s->port);
+ return false;
+ }
+ pthread_setname_np(s->thread, "pure-listen");
+ s->started = true;
+ return true;
+}
+
+void pure_link_done(PureLinkListener *s, void *token, uint64_t heard, bool stored, bool gap) {
+ pthread_mutex_lock(&s->lock);
+ if (s->done_count == s->done_capacity) {
+ size_t capacity = s->done_capacity ? 2 * s->done_capacity : 16;
+ Done *grown = realloc(s->done, capacity * sizeof(*grown));
+ if (!grown) abort(); /* a delivery's word cannot be lost */
+ s->done = grown;
+ s->done_capacity = capacity;
+ }
+ s->done[s->done_count++] = (Done){token, heard, stored, gap};
+ pthread_mutex_unlock(&s->lock);
+ wake(s->wake);
+}
+
+void pure_link_listener_close(PureLinkListener *s, PureLinkCounts *counts) {
+ if (!s) return;
+ if (s->started) {
+ pthread_mutex_lock(&s->lock);
+ s->stopping = true;
+ pthread_mutex_unlock(&s->lock);
+ wake(s->wake);
+ pthread_join(s->thread, NULL);
+ }
+ if (counts) {
+ *counts = s->counts;
+ counts->active = s->active_count;
+ }
+ while (s->active_count) let_go(s, s->actives[0]);
+ pure_aead_free(s->aead);
+ if (s->memo) munmap(s->memo, s->memo_size);
+ free(s->actives);
+ free(s->peers);
+ free(s->index);
+ free(s->allowed);
+ free(s->names);
+ free(s->ready);
+ free(s->due);
+ free(s->batch);
+ free(s->done);
+ if (s->fd >= 0) close(s->fd);
+ if (s->wake >= 0) close(s->wake);
+ pthread_mutex_destroy(&s->lock);
+ free(s);
+}
diff --git a/link/link.h b/link/link.h
new file mode 100644
index 0000000..83cc645
--- /dev/null
+++ b/link/link.h
@@ -0,0 +1,118 @@
+#ifndef PURE_LINK_H
+#define PURE_LINK_H
+#include "module.h"
+#include "seal.h"
+
+/* Links without connections (docs/design/011-pure-transport.md): one node's
+ * saved sends to another node's input over UDP, sealed with a key the two
+ * nodes' keys share, with no handshake: the first packet carries records.
+ * A link's lasting state is the apps' own: the sender's outbox (what to send
+ * and send again) and the receiver's heard numbers (what it applied, saved
+ * with the events). What either side keeps in memory is a cache, so either
+ * may restart and the other goes on with the next packet.
+ *
+ * A packet is a header in the clear, then frames sealed with AES-128-GCM
+ * (the header as associated data):
+ * u8 0x70 (version 0), u32le the key id of the node that sent it (FNV-1a
+ * of its public key; two keys may share an id, and a receiver tries the
+ * keys it allows with it), u64le that node's run (random, new each time it
+ * starts, so no nonce repeats under a key across restarts), u64le the
+ * packet's number in its run.
+ * Keys come from the secret the two nodes' keys share (X25519) by
+ * HKDF-SHA256, with both public keys and the sending node's run: a sender's
+ * packets use its own run; a receiver's answers add the run they answer,
+ * so an answer opens only in that run. The nonce is the key's base with the
+ * packet's number in its last eight bytes (XOR, big-endian, as QUIC does).
+ *
+ * Frames (numbers little-endian):
+ * 0x01 STREAM source (u8 length, bytes), input (u8 length, bytes), u32 the
+ * shape of the messages (the sender's output and the
+ * receiver's input must have the same): the stream the
+ * records after it are of
+ * 0x02 RECORD u64 number, u32 size, u32 offset, u16 length, bytes: a piece
+ * of a saved send (pieces are 1000 bytes, the last shorter)
+ * 0x03 ACK u64 largest number, u32 delay (µs), u8 ranges, each as u64
+ * first and u64 last, newest first
+ * 0x04 STORED source, input (as STREAM), u64 the last number heard
+ * (applied and saved), u8 flags (1: a record after a gap was
+ * not taken: send what follows it again), u32 credit (records
+ * it takes past that)
+ * 0x05 PING to be acknowledged
+ * 0x06 RECORDS u64 the first one's number, u8 count, then whole records one
+ * after another (numbers following on), each its size (a
+ * varint: LEB128) and its bytes: a send that fits the packet
+ * whole, with 1 to 3 bytes of framing, not the 19 of RECORD
+ * 0x00 PADDING to the end of the packet: a full packet sent in a burst is
+ * a whole datagram
+ * Losses are detected as RFC 9002 says; congestion control is its NewReno
+ * (RFC 9002 §7). */
+
+/* The sending side: one output's saved sends to url, pure://host:port/INPUT,
+ * as source (a Pure receiver's name for it), from own (this node's key,
+ * whose public half is own_public) to the receiver's public key, given by
+ * pure_link_peer before pure_link_start. The owner queues and trims as it
+ * does a wire's (serve/deliver.h): a send counts as delivered once the
+ * receiver says it saved it. */
+typedef struct PureLink PureLink;
+PureLink *pure_link_open(const PureModule *m, const char *output, const char *url, const char *source,
+ const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY], char *error,
+ size_t size);
+void pure_link_peer(PureLink *, const unsigned char peer[PURE_SEAL_KEY]);
+uint32_t pure_link_output(const PureLink *);
+bool pure_link_queue(PureLink *, uint64_t seq, PureSpan payload);
+uint64_t pure_link_queued(PureLink *);
+/* The last number the receiver saved; any thread. */
+uint64_t pure_link_delivered(PureLink *);
+/* Send on a thread of the link's own, calling notify when more was saved. */
+bool pure_link_start(PureLink *, void (*notify)(void *), void *context, char *error, size_t size);
+void pure_link_close(PureLink *);
+
+/* The receiving side: packets on a UDP port, from the senders allowed (a
+ * source name and the key it seals with, a few bytes each however many
+ * there are: see pure_link_idle). Whole records of a stream, in
+ * order, go to deliver (on the listener's thread) as one delivery; the host
+ * applies them from the one after the source's last heard number, stops at
+ * a gap, saves, and then calls pure_link_done (any thread) with the last
+ * number heard. Records wait here until then, and only the saved ones are
+ * acknowledged as stored. */
+typedef struct {
+ uint32_t input;
+ const char *source;
+ uint64_t first;
+ unsigned count;
+ const unsigned char *records; /* count of: u32le size, bytes */
+ size_t size;
+ void *token; /* for pure_link_done */
+} PureLinkDelivery;
+typedef struct PureLinkListener PureLinkListener;
+PureLinkListener *pure_link_listen(const PureModule *m, unsigned port, const PureSealKey *own,
+ const unsigned char own_public[PURE_SEAL_KEY],
+ bool (*deliver)(void *, const PureLinkDelivery *), void *context, char *error,
+ size_t size);
+bool pure_link_allow(PureLinkListener *, const char *source, const unsigned char key[PURE_SEAL_KEY], char *error,
+ size_t size);
+/* A sender quiet for ms (30 s at first) with nothing of it waiting lets its
+ * keys, addresses and streams go: an idle link costs the listener only its
+ * allowing, and the sender's next packet makes them again (one X25519). */
+void pure_link_idle(PureLinkListener *, uint64_t ms);
+/* Remember the secret each sender's key shares with the node's in a file
+ * (path, 0600; begun again for another node's key): a place a sender in
+ * the order allowed, mapped, so a sender at rest takes no memory for it.
+ * A node that starts again then makes no X25519 for a sender it heard from
+ * before. After the senders are allowed, before serving. */
+bool pure_link_remember(PureLinkListener *, const char *path, char *error, size_t size);
+bool pure_link_serve(PureLinkListener *, char *error, size_t size);
+/* stored: the delivery's save succeeded; gap: it stopped at a gap. */
+void pure_link_done(PureLinkListener *, void *token, uint64_t heard, bool stored, bool gap);
+
+/* Counts for a host's report: packets and records sent, records sent again,
+ * packets lost and acknowledged, packets received, deliveries handed over,
+ * the last number stored; a listener's senders woken (their active state
+ * made), of them with a secret remembered (no X25519), let go when idle,
+ * and active when it closed. A listener's are given when it closes. */
+typedef struct {
+ uint64_t packets, records, resent, lost, acknowledged, received, deliveries, stored, woken, remembered, swept, active;
+} PureLinkCounts;
+void pure_link_counts(PureLink *, PureLinkCounts *);
+void pure_link_listener_close(PureLinkListener *, PureLinkCounts *counts);
+#endif
diff --git a/link/seal.h b/link/seal.h
new file mode 100644
index 0000000..9245f25
--- /dev/null
+++ b/link/seal.h
@@ -0,0 +1,80 @@
+#ifndef PURE_SEAL_H
+#define PURE_SEAL_H
+#include <stdbool.h>
+#include <stddef.h>
+#include <stdint.h>
+
+/* Sends sealed end to end (docs/research/086-sealed-sends.md, 088): a
+ * receiving node's key opens what a sender sealed to its public key, and
+ * nothing between them (a relay, a proxy, a mailbox) can read or change it.
+ * HPKE (RFC 9180): X25519, HKDF-SHA256, and AES-128-GCM where the CPU has AES
+ * instructions, else ChaCha20-Poly1305; auth mode when the sender has a key
+ * of its own, so the receiver knows who sealed. A sender keeps one session
+ * (one encapsulation) and numbers its seals in it; a receiver opens a
+ * session once and then each seal by its number. A private key lives in
+ * OpenSSL's secure heap: locked in memory, out of core dumps, cleared when
+ * freed. OpenSSL comes with pure-crypto.so (link/crypto.c), mapped the
+ * first time a key is read or a session made. */
+enum { PURE_SEAL_KEY = 32, PURE_SEAL_OVERHEAD = 16 }; /* a public key; a seal's tag */
+
+typedef struct PureSealKey PureSealKey;
+typedef struct PureSealer PureSealer; /* a sender's session */
+typedef struct PureOpener PureOpener; /* a receiver's session */
+
+/* Whether this machine can seal (pure-crypto.so maps), with why not in error. */
+bool pure_seal_ready(char *error, size_t size);
+/* A node's key: read from path, or made and written there (0600) when there
+ * is none; its public half in hex. NULL with why in error. */
+PureSealKey *pure_seal_key(const char *path, char public_hex[2 * PURE_SEAL_KEY + 1], char *error, size_t size);
+void pure_seal_key_free(PureSealKey *);
+/* A public key from its hex. */
+bool pure_seal_peer(const char *hex, unsigned char out[PURE_SEAL_KEY]);
+/* The suite this machine seals with, as the X-Pure-Seal header names it. */
+const char *pure_seal_suite(void);
+
+/* A session sealing to peer, authenticated by own when given: its
+ * encapsulated key into enc (what a receiver opens the session with). */
+PureSealer *pure_sealer(const unsigned char peer[PURE_SEAL_KEY], const PureSealKey *own,
+ unsigned char enc[PURE_SEAL_KEY]);
+/* plain sealed as the session's next seal, bound to aad: size +
+ * PURE_SEAL_OVERHEAD bytes into out, and its number into n. */
+bool pure_seal(PureSealer *, const char *aad, const unsigned char *plain, size_t size, unsigned char *out,
+ uint64_t *n);
+void pure_sealer_free(PureSealer *);
+
+/* A session a sender made to key's public half, in suite (this machine's or
+ * the other known one), from its enc; from is the sender's public key when
+ * it sealed with its own (NULL otherwise). NULL when it does not open. */
+PureOpener *pure_opener(const PureSealKey *key, const char *suite, const unsigned char enc[PURE_SEAL_KEY],
+ const unsigned char *from);
+/* The session's seal number n: size - PURE_SEAL_OVERHEAD bytes into plain;
+ * false when it does not open (another session, another aad, or changed on
+ * the way). */
+bool pure_unseal(PureOpener *, uint64_t n, const char *aad, const unsigned char *sealed, size_t size,
+ unsigned char *plain);
+void pure_opener_free(PureOpener *);
+
+/* Links (link/link.h): the secret two nodes share (X25519 of this node's
+ * private key and the other's public key: the same on both sides, and
+ * never all zeros), keys from it (HKDF-SHA256 with the info given), and
+ * AES-128-GCM under such a key with the nonce the caller gives (sealing may
+ * be in place: out may be plain). A context is used by one thread at a time;
+ * sealing or opening with a key given uses that key from then on, so one
+ * context serves many links (a key costs about as much as a small packet),
+ * and NULL keeps the key it has. */
+typedef struct PureAead PureAead;
+enum { PURE_AEAD_KEY = 16, PURE_AEAD_NONCE = 12, PURE_AEAD_TAG = 16 };
+bool pure_link_secret(const PureSealKey *own, const unsigned char peer[PURE_SEAL_KEY], unsigned char out[32]);
+bool pure_link_derive(const unsigned char secret[32], const unsigned char *info, size_t info_size, unsigned char *out,
+ size_t out_size);
+PureAead *pure_aead(const unsigned char key[PURE_AEAD_KEY]);
+/* size + PURE_AEAD_TAG bytes into out. */
+bool pure_aead_seal(PureAead *, const unsigned char *key, const unsigned char nonce[PURE_AEAD_NONCE],
+ const unsigned char *aad, size_t aad_size, const unsigned char *plain, size_t size,
+ unsigned char *out);
+/* size - PURE_AEAD_TAG bytes into plain; false when it does not open. */
+bool pure_aead_open(PureAead *, const unsigned char *key, const unsigned char nonce[PURE_AEAD_NONCE],
+ const unsigned char *aad, size_t aad_size, const unsigned char *sealed, size_t size,
+ unsigned char *plain);
+void pure_aead_free(PureAead *);
+#endif
diff --git a/link/tls.h b/link/tls.h
new file mode 100644
index 0000000..3939f21
--- /dev/null
+++ b/link/tls.h
@@ -0,0 +1,18 @@
+#ifndef PURE_TLS_H
+#define PURE_TLS_H
+#include <stdbool.h>
+#include <stddef.h>
+
+/* TLS to https receivers (link/crypto.c): one client context, verifying a
+ * receiver's certificate against the system's CAs or ca_file, and its name. */
+typedef struct PureTls PureTls;
+/* The context, made once; false with why in error when it cannot be. */
+bool pure_tls_start(const char *ca_file, char *error, size_t size);
+/* A session over a connected socket to host (a DNS name, or an IP address
+ * written as one), its certificate verified; NULL when it fails. */
+PureTls *pure_tls_open(const char *host, int fd);
+/* As SSL_read and SSL_write: the bytes moved, or <= 0. */
+int pure_tls_read(PureTls *, void *bytes, int size);
+int pure_tls_write(PureTls *, const void *bytes, int size);
+void pure_tls_close(PureTls *);
+#endif