refactor(flake.nix): externalize pkgs, add overlay

This splits the complexity of the `flake.nix` into multiple files. At
cross-compiled and static builds at the benefit of simpler nix
expressions and generally better cross compilation compatibility.
the same time, naersk is removed; causing much slower builds for cross-
compiled packages.

This partially addresses the points mentioned in #412.

.github/workflows/nix.yaml

@@ -6,6 +6,11 @@ on:
   push:
     branches:
       - main
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
 jobs:
   i686-linux---default:
     name: Build i686-linux.default
@@ -241,30 +246,30 @@ jobs:
           authToken: ${{ secrets.CACHIX_AUTH_TOKEN }}
       - name: Build
         run: nix build .#packages.x86_64-linux.release-package --print-build-logs
-  aarch64-linux---release-package:
-    name: Build aarch64-linux.release-package
-    runs-on:
-      - ubuntu-latest
-    needs:
-      - aarch64-linux---rosenpass-oci-image
-      - aarch64-linux---rosenpass
-      - aarch64-linux---rp
-    steps:
-      - run: |
-          DEBIAN_FRONTEND=noninteractive
-          sudo apt-get update -q -y && sudo apt-get install -q -y qemu-system-aarch64 qemu-efi binfmt-support qemu-user-static
-      - uses: actions/checkout@v3
-      - uses: cachix/install-nix-action@v22
-        with:
-          nix_path: nixpkgs=channel:nixos-unstable
-          extra_nix_config: |
-            system = aarch64-linux
-      - uses: cachix/cachix-action@v12
-        with:
-          name: rosenpass
-          authToken: ${{ secrets.CACHIX_AUTH_TOKEN }}
-      - name: Build
-        run: nix build .#packages.aarch64-linux.release-package --print-build-logs
+  # aarch64-linux---release-package:
+  #   name: Build aarch64-linux.release-package
+  #   runs-on:
+  #     - ubuntu-latest
+  #   needs:
+  #     - aarch64-linux---rosenpass-oci-image
+  #     - aarch64-linux---rosenpass
+  #     - aarch64-linux---rp
+  #   steps:
+  #     - run: |
+  #         DEBIAN_FRONTEND=noninteractive
+  #         sudo apt-get update -q -y && sudo apt-get install -q -y qemu-system-aarch64 qemu-efi binfmt-support qemu-user-static
+  #     - uses: actions/checkout@v3
+  #     - uses: cachix/install-nix-action@v22
+  #       with:
+  #         nix_path: nixpkgs=channel:nixos-unstable
+  #         extra_nix_config: |
+  #           system = aarch64-linux
+  #     - uses: cachix/cachix-action@v12
+  #       with:
+  #         name: rosenpass
+  #         authToken: ${{ secrets.CACHIX_AUTH_TOKEN }}
+  #     - name: Build
+  #       run: nix build .#packages.aarch64-linux.release-package --print-build-logs
   x86_64-linux---rosenpass:
     name: Build x86_64-linux.rosenpass
     runs-on:

.github/workflows/qc.yaml

@@ -4,6 +4,10 @@ on:
   push:
     branches: [main]
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
 permissions:
   checks: write
   contents: read
@@ -195,13 +199,16 @@ jobs:
       - run: rustup component add llvm-tools-preview
       - run: |
           cargo install cargo-llvm-cov || true
-          cargo llvm-cov --lcov --output-path coverage.lcov
+          cargo llvm-cov \
+            --workspace\
+            --all-features \
+            --lcov \
+            --output-path coverage.lcov
       # If using tarapulin
       #- run: cargo install cargo-tarpaulin
       #- run: cargo tarpaulin --out Xml
       - name: Upload coverage reports to Codecov
-        uses: codecov/codecov-action@v4.0.1
+        uses: codecov/codecov-action@v4
         with:
-          token: ${{ secrets.CODECOV_TOKEN }}
           files: ./coverage.lcov
           verbose: true

.github/workflows/regressions.yml

@@ -1,9 +1,13 @@
-name: QC
+name: Regressions
 on:
   pull_request:
   push:
     branches: [main]
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
 permissions:
   checks: write
   contents: read

.prettierignore

@@ -1,4 +1,5 @@
 .direnv/
+flake.lock
 papers/whitepaper.md
-target/
 src/usage.md
+target/

Cargo.toml

@@ -32,55 +32,59 @@ rosenpass-secret-memory = { path = "secret-memory" }
 rosenpass-oqs = { path = "oqs" }
 rosenpass-wireguard-broker = { path = "wireguard-broker" }
 doc-comment = "0.3.3"
-base64ct = {version = "1.6.0", default-features=false}
+base64ct = { version = "1.6.0", default-features = false }
 zeroize = "1.8.1"
 memoffset = "0.9.1"
-thiserror = "1.0.63"
+thiserror = "1.0.64"
 paste = "1.0.15"
 env_logger = "0.10.2"
 toml = "0.7.8"
 static_assertions = "1.1.0"
 allocator-api2 = "0.2.14"
-memsec = { git="https://github.com/rosenpass/memsec.git" ,rev="aceb9baee8aec6844125bd6612f92e9a281373df", features = [ "alloc_ext", ] }
+memsec = { git = "https://github.com/rosenpass/memsec.git", rev = "aceb9baee8aec6844125bd6612f92e9a281373df", features = [
+    "alloc_ext",
+] }
 rand = "0.8.5"
 typenum = "1.17.0"
 log = { version = "0.4.22" }
-clap = { version = "4.5.13", features = ["derive"] }
-serde = { version = "1.0.204", features = ["derive"] }
+clap = { version = "4.5.19", features = ["derive"] }
+serde = { version = "1.0.210", features = ["derive"] }
 arbitrary = { version = "1.3.2", features = ["derive"] }
-anyhow = { version = "1.0.86", features = ["backtrace", "std"] }
-mio = { version = "1.0.1", features = ["net", "os-poll"] }
+anyhow = { version = "1.0.89", features = ["backtrace", "std"] }
+mio = { version = "1.0.2", features = ["net", "os-poll"] }
 oqs-sys = { version = "0.9.1", default-features = false, features = [
-  'classic_mceliece',
-  'kyber',
+    'classic_mceliece',
+    'kyber',
 ] }
 blake2 = "0.10.6"
 chacha20poly1305 = { version = "0.10.1", default-features = false, features = [
-  "std",
-  "heapless",
+    "std",
+    "heapless",
 ] }
 zerocopy = { version = "0.7.35", features = ["derive"] }
 home = "0.5.9"
-derive_builder = "0.20.0"
-tokio = { version = "1.39", features = ["macros", "rt-multi-thread"] }
-postcard= {version = "1.0.8", features = ["alloc"]}
+derive_builder = "0.20.1"
+tokio = { version = "1.40", features = ["macros", "rt-multi-thread"] }
+postcard = { version = "1.0.10", features = ["alloc"] }
 libcrux = { version = "0.0.2-pre.2" }
 hex-literal = { version = "0.4.1" }
 hex = { version = "0.4.3" }
-heck = { version = "0.5.0"  }
+heck = { version = "0.5.0" }
+libc = { version = "0.2" }
+uds = { git = "https://github.com/rosenpass/uds" }
 
 #Dev dependencies
 serial_test = "3.1.1"
 tempfile = "3"
-stacker = "0.1.15"
+stacker = "0.1.17"
 libfuzzer-sys = "0.4"
 test_bin = "0.4.0"
 criterion = "0.4.0"
 allocator-api2-tests = "0.2.15"
-procspawn = {version = "1.0.0", features= ["test-support"]}
+procspawn = { version = "1.0.1", features = ["test-support"] }
 
 
 #Broker dependencies (might need cleanup or changes)
 wireguard-uapi = { version = "3.0.0", features = ["xplatform"] }
 command-fds = "0.2.3"
-rustix = { version = "0.38.27", features = ["net", "fs"] }
+rustix = { version = "0.38.37", features = ["net", "fs"] }

ciphers/Cargo.toml

@@ -23,4 +23,4 @@ static_assertions = { workspace = true }
 zeroize = { workspace = true }
 chacha20poly1305 = { workspace = true }
 blake2 = { workspace = true }
-libcrux = { workspace = true, optional = true } 
+libcrux = { workspace = true, optional = true }

flake.lock

@@ -2,15 +2,17 @@
   "nodes": {
     "fenix": {
       "inputs": {
-        "nixpkgs": ["nixpkgs"],
+        "nixpkgs": [
+          "nixpkgs"
+        ],
         "rust-analyzer-src": "rust-analyzer-src"
       },
       "locked": {
-        "lastModified": 1712298178,
-        "narHash": "sha256-590fpCPXYAkaAeBz/V91GX4/KGzPObdYtqsTWzT6AhI=",
+        "lastModified": 1728282832,
+        "narHash": "sha256-I7AbcwGggf+CHqpyd/9PiAjpIBGTGx5woYHqtwxaV7I=",
         "owner": "nix-community",
         "repo": "fenix",
-        "rev": "569b5b5781395da08e7064e825953c548c26af76",
+        "rev": "1ec71be1f4b8f3105c5d38da339cb061fefc43f4",
         "type": "github"
       },
       "original": {
@@ -24,11 +26,11 @@
         "systems": "systems"
       },
       "locked": {
-        "lastModified": 1710146030,
-        "narHash": "sha256-SZ5L6eA7HJ/nmkzGG7/ISclqe6oZdOZTNoesiInkXPQ=",
+        "lastModified": 1726560853,
+        "narHash": "sha256-X6rJYSESBVr3hBoH0WbKE5KvhPU5bloyZ2L4K60/fPQ=",
         "owner": "numtide",
         "repo": "flake-utils",
-        "rev": "b1d9ab70662946ef0850d488da1c9019f3a9752a",
+        "rev": "c1dfcf08411b08f6b8615f7d8971a2bfa81d5e8a",
         "type": "github"
       },
       "original": {
@@ -37,36 +39,18 @@
         "type": "github"
       }
     },
-    "naersk": {
-      "inputs": {
-        "nixpkgs": ["nixpkgs"]
-      },
-      "locked": {
-        "lastModified": 1698420672,
-        "narHash": "sha256-/TdeHMPRjjdJub7p7+w55vyABrsJlt5QkznPYy55vKA=",
-        "owner": "nix-community",
-        "repo": "naersk",
-        "rev": "aeb58d5e8faead8980a807c840232697982d47b9",
-        "type": "github"
-      },
-      "original": {
-        "owner": "nix-community",
-        "repo": "naersk",
-        "type": "github"
-      }
-    },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1712168706,
-        "narHash": "sha256-XP24tOobf6GGElMd0ux90FEBalUtw6NkBSVh/RlA6ik=",
+        "lastModified": 1728193676,
+        "narHash": "sha256-PbDWAIjKJdlVg+qQRhzdSor04bAPApDqIv2DofTyynk=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "1487bdea619e4a7a53a4590c475deabb5a9d1bfb",
+        "rev": "ecbc1ca8ffd6aea8372ad16be9ebbb39889e55b6",
         "type": "github"
       },
       "original": {
         "owner": "NixOS",
-        "ref": "nixos-23.11",
+        "ref": "nixos-24.05",
         "repo": "nixpkgs",
         "type": "github"
       }
@@ -75,18 +59,17 @@
       "inputs": {
         "fenix": "fenix",
         "flake-utils": "flake-utils",
-        "naersk": "naersk",
         "nixpkgs": "nixpkgs"
       }
     },
     "rust-analyzer-src": {
       "flake": false,
       "locked": {
-        "lastModified": 1712156296,
-        "narHash": "sha256-St7ZQrkrr5lmQX9wC1ZJAFxL8W7alswnyZk9d1se3Us=",
+        "lastModified": 1728249780,
+        "narHash": "sha256-J269DvCI5dzBmPrXhAAtj566qt0b22TJtF3TIK+tMsI=",
         "owner": "rust-lang",
         "repo": "rust-analyzer",
-        "rev": "8e581ac348e223488622f4d3003cb2bd412bf27e",
+        "rev": "2b750da1a1a2c1d2c70896108d7096089842d877",
         "type": "github"
       },
       "original": {

flake.nix

@@ -1,12 +1,8 @@
 {
   inputs = {
-    nixpkgs.url = "github:NixOS/nixpkgs/nixos-23.11";
+    nixpkgs.url = "github:NixOS/nixpkgs/nixos-24.05";
     flake-utils.url = "github:numtide/flake-utils";
 
-    # for quicker rust builds
-    naersk.url = "github:nix-community/naersk";
-    naersk.inputs.nixpkgs.follows = "nixpkgs";
-
     # for rust nightly with llvm-tools-preview
     fenix.url = "github:nix-community/fenix";
     fenix.inputs.nixpkgs.follows = "nixpkgs";
@@ -15,6 +11,15 @@
   outputs = { self, nixpkgs, flake-utils, ... }@inputs:
     nixpkgs.lib.foldl (a: b: nixpkgs.lib.recursiveUpdate a b) { } [
 
+
+      #
+      ### Export the overlay.nix from this flake ###
+      #
+      {
+        overlays.default = import ./overlay.nix;
+      }
+
+
       #
       ### Actual Rosenpass Package and Docker Container Images ###
       #
@@ -30,310 +35,39 @@
       ]
         (system:
           let
-            scoped = (scope: scope.result);
-            lib = nixpkgs.lib;
-
             # normal nixpkgs
             pkgs = import nixpkgs {
               inherit system;
-            };
 
-            # parsed Cargo.toml
-            cargoToml = builtins.fromTOML (builtins.readFile ./rosenpass/Cargo.toml);
-
-            # source files relevant for rust
-            src = scoped rec {
-              # File suffices to include
-              extensions = [
-                "lock"
-                "rs"
-                "toml"
-              ];
-              # Files to explicitly include
-              files = [
-                "to/README.md"
-              ];
-
-              src = ./.;
-              filter = (path: type: scoped rec {
-                inherit (lib) any id removePrefix hasSuffix;
-                anyof = (any id);
-
-                basename = baseNameOf (toString path);
-                relative = removePrefix (toString src + "/") (toString path);
-
-                result = anyof [
-                  (type == "directory")
-                  (any (ext: hasSuffix ".${ext}" basename) extensions)
-                  (any (file: file == relative) files)
-                ];
-              });
-
-              result = pkgs.lib.sources.cleanSourceWith { inherit src filter; };
-            };
-
-            # a function to generate a nix derivation for rosenpass against any
-            # given set of nixpkgs
-            rosenpassDerivation = p:
-              let
-                # whether we want to build a statically linked binary
-                isStatic = p.targetPlatform.isStatic;
-
-                # the rust target of `p`
-                target = p.rust.toRustTargetSpec p.targetPlatform;
-
-                # convert a string to shout case
-                shout = string: builtins.replaceStrings [ "-" ] [ "_" ] (pkgs.lib.toUpper string);
-
-                # suitable Rust toolchain
-                toolchain = with inputs.fenix.packages.${system}; combine [
-                  stable.cargo
-                  stable.rustc
-                  targets.${target}.stable.rust-std
-                ];
-
-                # naersk with a custom toolchain
-                naersk = pkgs.callPackage inputs.naersk {
-                  cargo = toolchain;
-                  rustc = toolchain;
-                };
-
-                # used to trick the build.rs into believing that CMake was ran **again**
-                fakecmake = pkgs.writeScriptBin "cmake" ''
-                  #! ${pkgs.stdenv.shell} -e
-                  true
-                '';
-              in
-              naersk.buildPackage
-                {
-                  # metadata and source
-                  name = cargoToml.package.name;
-                  version = cargoToml.package.version;
-                  inherit src;
-
-                  cargoBuildOptions = x: x ++ [ "-p" "rosenpass" ];
-                  cargoTestOptions = x: x ++ [ "-p" "rosenpass" ];
-
-                  doCheck = true;
-
-                  nativeBuildInputs = with pkgs; [
-                    p.stdenv.cc
-                    cmake # for oqs build in the oqs-sys crate
-                    mandoc # for the built-in manual
-                    removeReferencesTo
-                    rustPlatform.bindgenHook # for C-bindings in the crypto libs
-                  ];
-                  buildInputs = with p; [ bash ];
-
-                  override = x: {
-                    preBuild =
-                      # nix defaults to building for aarch64 _without_ the armv8-a crypto
-                      # extensions, but liboqs depens on these
-                      (lib.optionalString (system == "aarch64-linux") ''
-                        NIX_CFLAGS_COMPILE="$NIX_CFLAGS_COMPILE -march=armv8-a+crypto"
-                      ''
-                      );
-
-                    # fortify is only compatible with dynamic linking
-                    hardeningDisable = lib.optional isStatic "fortify";
-                  };
-
-                  overrideMain = x: {
-                    # CMake detects that it was served a _foreign_ target dir, and CMake
-                    # would be executed again upon the second build step of naersk.
-                    # By adding our specially optimized CMake version, we reduce the cost
-                    # of recompilation by 99 % while, while avoiding any CMake errors.
-                    nativeBuildInputs = [ (lib.hiPrio fakecmake) ] ++ x.nativeBuildInputs;
-
-                    # make sure that libc is linked, under musl this is not the case per
-                    # default
-                    preBuild = (lib.optionalString isStatic ''
-                      NIX_CFLAGS_COMPILE="$NIX_CFLAGS_COMPILE -lc"
-                    '');
-                  };
-
-                  # We want to build for a specific target...
-                  CARGO_BUILD_TARGET = target;
-
-                  # ... which might require a non-default linker:
-                  "CARGO_TARGET_${shout target}_LINKER" =
-                    let
-                      inherit (p.stdenv) cc;
-                    in
-                    "${cc}/bin/${cc.targetPrefix}cc";
-
-                  meta = with pkgs.lib;
-                    {
-                      inherit (cargoToml.package) description homepage;
-                      license = with licenses; [ mit asl20 ];
-                      maintainers = [ maintainers.wucke13 ];
-                      platforms = platforms.all;
-                    };
-                } // (lib.mkIf isStatic {
-                # otherwise pkg-config tries to link non-existent dynamic libs
-                # documented here: https://docs.rs/pkg-config/latest/pkg_config/
-                PKG_CONFIG_ALL_STATIC = true;
-
-                # tell rust to build everything statically linked
-                CARGO_BUILD_RUSTFLAGS = "-C target-feature=+crt-static";
-              });
-            # a function to generate a nix derivation for the rp helper against any
-            # given set of nixpkgs
-            rpDerivation = p:
-              let
-                # whether we want to build a statically linked binary
-                isStatic = p.targetPlatform.isStatic;
-
-                # the rust target of `p`
-                target = p.rust.toRustTargetSpec p.targetPlatform;
-
-                # convert a string to shout case
-                shout = string: builtins.replaceStrings [ "-" ] [ "_" ] (pkgs.lib.toUpper string);
-
-                # suitable Rust toolchain
-                toolchain = with inputs.fenix.packages.${system}; combine [
-                  stable.cargo
-                  stable.rustc
-                  targets.${target}.stable.rust-std
-                ];
-
-                # naersk with a custom toolchain
-                naersk = pkgs.callPackage inputs.naersk {
-                  cargo = toolchain;
-                  rustc = toolchain;
-                };
-
-                # used to trick the build.rs into believing that CMake was ran **again**
-                fakecmake = pkgs.writeScriptBin "cmake" ''
-                  #! ${pkgs.stdenv.shell} -e
-                  true
-                '';
-              in
-              naersk.buildPackage
-                {
-                  # metadata and source
-                  name = cargoToml.package.name;
-                  version = cargoToml.package.version;
-                  inherit src;
-
-                  cargoBuildOptions = x: x ++ [ "-p" "rp" ];
-                  cargoTestOptions = x: x ++ [ "-p" "rp" ];
-
-                  doCheck = true;
-
-                  nativeBuildInputs = with pkgs; [
-                    p.stdenv.cc
-                    cmake # for oqs build in the oqs-sys crate
-                    mandoc # for the built-in manual
-                    removeReferencesTo
-                    rustPlatform.bindgenHook # for C-bindings in the crypto libs
-                  ];
-                  buildInputs = with p; [ bash ];
-
-                  override = x: {
-                    preBuild =
-                      # nix defaults to building for aarch64 _without_ the armv8-a crypto
-                      # extensions, but liboqs depens on these
-                      (lib.optionalString (system == "aarch64-linux") ''
-                        NIX_CFLAGS_COMPILE="$NIX_CFLAGS_COMPILE -march=armv8-a+crypto"
-                      ''
-                      );
-
-                    # fortify is only compatible with dynamic linking
-                    hardeningDisable = lib.optional isStatic "fortify";
-                  };
-
-                  overrideMain = x: {
-                    # CMake detects that it was served a _foreign_ target dir, and CMake
-                    # would be executed again upon the second build step of naersk.
-                    # By adding our specially optimized CMake version, we reduce the cost
-                    # of recompilation by 99 % while, while avoiding any CMake errors.
-                    nativeBuildInputs = [ (lib.hiPrio fakecmake) ] ++ x.nativeBuildInputs;
-
-                    # make sure that libc is linked, under musl this is not the case per
-                    # default
-                    preBuild = (lib.optionalString isStatic ''
-                      NIX_CFLAGS_COMPILE="$NIX_CFLAGS_COMPILE -lc"
-                    '');
-                  };
-
-                  # We want to build for a specific target...
-                  CARGO_BUILD_TARGET = target;
-
-                  # ... which might require a non-default linker:
-                  "CARGO_TARGET_${shout target}_LINKER" =
-                    let
-                      inherit (p.stdenv) cc;
-                    in
-                    "${cc}/bin/${cc.targetPrefix}cc";
-
-                  meta = with pkgs.lib;
-                    {
-                      inherit (cargoToml.package) description homepage;
-                      license = with licenses; [ mit asl20 ];
-                      maintainers = [ maintainers.wucke13 ];
-                      platforms = platforms.all;
-                    };
-                } // (lib.mkIf isStatic {
-                # otherwise pkg-config tries to link non-existent dynamic libs
-                # documented here: https://docs.rs/pkg-config/latest/pkg_config/
-                PKG_CONFIG_ALL_STATIC = true;
-
-                # tell rust to build everything statically linked
-                CARGO_BUILD_RUSTFLAGS = "-C target-feature=+crt-static";
-              });
-            # a function to generate a docker image based of rosenpass
-            rosenpassOCI = name: pkgs.dockerTools.buildImage rec {
-              inherit name;
-              copyToRoot = pkgs.buildEnv {
-                name = "image-root";
-                paths = [ self.packages.${system}.${name} ];
-                pathsToLink = [ "/bin" ];
-              };
-              config.Cmd = [ "/bin/rosenpass" ];
+              # apply our own overlay, overriding/inserting our packages as defined in ./pkgs
+              overlays = [ self.overlays.default ];
             };
           in
-          rec {
-            packages = rec {
-              default = rosenpass;
-              rosenpass = rosenpassDerivation pkgs;
-              rp = rpDerivation pkgs;
-              rosenpass-oci-image = rosenpassOCI "rosenpass";
+          {
+            packages = {
+              default = pkgs.rosenpass;
+              rosenpass = pkgs.rosenpass;
+              rosenpass-oci-image = pkgs.rosenpass-oci-image;
+              rp = pkgs.rp;
 
-              # derivation for the release
-              release-package =
-                let
-                  version = cargoToml.package.version;
-                  package =
-                    if pkgs.hostPlatform.isLinux then
-                      packages.rosenpass-static
-                    else packages.rosenpass;
-                  rp =
-                    if pkgs.hostPlatform.isLinux then
-                      packages.rp-static
-                    else packages.rp;
-                  oci-image =
-                    if pkgs.hostPlatform.isLinux then
-                      packages.rosenpass-static-oci-image
-                    else packages.rosenpass-oci-image;
-                in
-                pkgs.runCommandNoCC "lace-result" { }
-                  ''
-                    mkdir {bin,$out}
-                    tar -cvf $out/rosenpass-${system}-${version}.tar \
-                      -C ${package} bin/rosenpass \
-                      -C ${rp} bin/rp
-                    cp ${oci-image} \
-                      $out/rosenpass-oci-image-${system}-${version}.tar.gz
-                  '';
-            } // (if pkgs.stdenv.isLinux then rec {
-              rosenpass-static = rosenpassDerivation pkgs.pkgsStatic;
-              rp-static = rpDerivation pkgs.pkgsStatic;
-              rosenpass-static-oci-image = rosenpassOCI "rosenpass-static";
-            } else { });
+              release-package = pkgs.release-package;
+
+              # for good measure, we also offer to cross compile to Linux on Arm
+              aarch64-linux-rosenpass-static =
+                pkgs.pkgsCross.aarch64-multiplatform.pkgsStatic.rosenpass;
+              aarch64-linux-rp-static = pkgs.pkgsCross.aarch64-multiplatform.pkgsStatic.rp;
+            }
+            //
+            # We only offer static builds for linux, as this is not supported on OS X
+            (nixpkgs.lib.attrsets.optionalAttrs pkgs.stdenv.isLinux {
+              rosenpass-static = pkgs.pkgsStatic.rosenpass;
+              rosenpass-static-oci-image = pkgs.pkgsStatic.rosenpass-oci-image;
+              rp-static = pkgs.pkgsStatic.rp;
+            });
           }
         ))
 
+
       #
       ### Linux specifics ###
       #
@@ -341,88 +75,46 @@
         let
           pkgs = import nixpkgs {
             inherit system;
+
+            # apply our own overlay, overriding/inserting our packages as defined in ./pkgs
+            overlays = [ self.overlays.default ];
           };
-          packages = self.packages.${system};
         in
         {
-          #
-          ### Whitepaper ###
-          #
-          packages.whitepaper =
-            let
-              tlsetup = (pkgs.texlive.combine {
-                inherit (pkgs.texlive) scheme-basic acmart amsfonts ccicons
-                  csquotes csvsimple doclicense fancyvrb fontspec gobble
-                  koma-script ifmtarg latexmk lm markdown mathtools minted noto
-                  nunito pgf soul unicode-math lualatex-math paralist
-                  gitinfo2 eso-pic biblatex biblatex-trad biblatex-software
-                  xkeyval xurl xifthen biber;
-              });
-            in
-            pkgs.stdenvNoCC.mkDerivation {
-              name = "whitepaper";
-              src = ./papers;
-              nativeBuildInputs = with pkgs; [
-                ncurses # tput
-                python3Packages.pygments
-                tlsetup # custom tex live scheme
-                which
-              ];
-              buildPhase = ''
-                export HOME=$(mktemp -d)
-                latexmk -r tex/CI.rc
-              '';
-              installPhase = ''
-                mkdir -p $out
-                mv *.pdf readme.md $out/
-              '';
-            };
 
+          #
+          ### Reading materials ###
+          #
+          packages.whitepaper = pkgs.whitepaper;
 
           #
           ### Proof and Proof Tools ###
           #
-          packages.proverif-patched = pkgs.proverif.overrideAttrs (old: {
-            postInstall = ''
-              install -D -t $out/lib cryptoverif.pvl
-            '';
-          });
-          packages.proof-proverif = pkgs.stdenv.mkDerivation {
-            name = "rosenpass-proverif-proof";
-            version = "unstable";
-            src = pkgs.lib.sources.sourceByRegex ./. [
-              "analyze.sh"
-              "marzipan(/marzipan.awk)?"
-              "analysis(/.*)?"
-            ];
-            nativeBuildInputs = [ pkgs.proverif pkgs.graphviz ];
-            CRYPTOVERIF_LIB = packages.proverif-patched + "/lib/cryptoverif.pvl";
-            installPhase = ''
-              mkdir -p $out
-              bash analyze.sh -color -html $out
-            '';
-          };
+          packages.proverif-patched = pkgs.proverif-patched;
+          packages.proof-proverif = pkgs.proof-proverif;
 
 
           #
           ### Devshells ###
           #
           devShells.default = pkgs.mkShell {
-            inherit (packages.proof-proverif) CRYPTOVERIF_LIB;
-            inputsFrom = [ packages.default ];
+            inherit (pkgs.proof-proverif) CRYPTOVERIF_LIB;
+            inputsFrom = [ pkgs.rosenpass ];
             nativeBuildInputs = with pkgs; [
-              inputs.fenix.packages.${system}.complete.toolchain
-              cmake # override the fakecmake from the main step above
               cargo-release
               clippy
+              rustfmt
               nodePackages.prettier
               nushell # for the .ci/gen-workflow-files.nu script
-              packages.proverif-patched
+              proverif-patched
             ];
           };
           devShells.coverage = pkgs.mkShell {
-            inputsFrom = [ packages.default ];
-            nativeBuildInputs = with pkgs; [ inputs.fenix.packages.${system}.complete.toolchain cargo-llvm-cov ];
+            inputsFrom = [ pkgs.rosenpass ];
+            nativeBuildInputs = [
+              inputs.fenix.packages.${system}.complete.toolchain
+              pkgs.cargo-llvm-cov
+            ];
           };
 
 

manual_tests/psk_broker/peer_a.rp.config

@@ -0,0 +1,13 @@
+secret_key = "peer_a.rp.sk"
+public_key = "peer_a.rp.pk"
+listen = ["[::1]:46127"]
+verbosity = "Verbose"
+
+[api]
+listen_path = []
+listen_fd = []
+stream_fd = []
+
+[[peers]]
+public_key = "peer_b.rp.pk"
+device = "rpPskBrkTestA"

manual_tests/psk_broker/peer_b.rp.config

@@ -0,0 +1,14 @@
+secret_key = "peer_b.rp.sk"
+public_key = "peer_b.rp.pk"
+listen = []
+verbosity = "Verbose"
+
+[api]
+listen_path = []
+listen_fd = []
+stream_fd = []
+
+[[peers]]
+public_key = "peer_a.rp.pk"
+endpoint = "[::1]:46127"
+device = "rpPskBrkTestB"

manual_tests/psk_broker/run_test.sh

@@ -0,0 +1,215 @@
+#! /bin/bash
+
+set -e -o pipefail
+
+enquote() {
+  while (( "$#" > 1)); do
+    printf "%q " "$1"
+    shift
+  done
+  if (("$#" > 0)); then
+    printf "%q" "$1"
+  fi
+}
+
+CLEANUP_HOOKS=()
+hook_cleanup() {
+  local hook
+  set +e +o pipefail
+  for hook in "${CLEANUP_HOOKS[@]}"; do
+    eval "${hook}"
+  done
+}
+
+cleanup() {
+  CLEANUP_HOOKS=("$(enquote exc_with_ctx cleanup "$@")" "${CLEANUP_HOOKS[@]}")
+}
+
+cleanup_eval() {
+  cleanup eval "$*"
+}
+
+stderr() {
+  echo >&2 "$@"
+}
+
+log() {
+  local level; level="$1"; shift || fatal "USAGE: log LVL MESSAGE.."
+  stderr "[${level}]" "$@"
+}
+
+info() {
+  log "INFO" "$@"
+}
+
+debug() {
+  log "DEBUG" "$@"
+}
+
+fatal() {
+  log "FATAL" "$@"
+  exit 1
+}
+
+assert() {
+  local msg; msg="$1"; shift || fatal "USAGE: assert_cmd MESSAGE COMMAND.."
+  "$@" || fatal "${msg}"
+}
+
+abs_dir() {
+  local dir; dir="$1"; shift || fatal "USAGE: abs_dir DIR"
+  (
+    cd "${dir}"
+    pwd -P
+  )
+}
+
+exc_with_ctx() {
+  local ctx; ctx="$1"; shift || fatal "USAGE: exc_with_ctx CONTEXT COMMAND.."
+  if [[ -z "${ctx}" ]]; then
+    info '$' "$@"
+  else
+    info "${ctx}\$" "$@"
+  fi
+
+  "$@"
+}
+
+exc() {
+  exc_with_ctx "" "$@"
+}
+
+exc_eval() {
+  exc eval "$*"
+} 
+
+exc_eval_with_ctx() {
+  local ctx; ctx="$1"; shift || fatal "USAGE: exc_eval_with_ctx CONTEXT EVAL_COMMAND.."
+  exc_with_ctx "eval:${ctx}" "$*"
+} 
+
+exc_as_user() {
+  exc sudo -u "${SUDO_USER}" "$@"
+}
+
+exc_eval_as_user() {
+  exc_as_user bash -c "$*"
+}
+
+fork_eval_as_user() {
+  exc sudo -u "${SUDO_USER}" bash -c "$*" &
+  local pid; pid="$!"
+  cleanup wait "${pid}"
+  cleanup pkill -2 -P "${pid}" # Reverse ordering
+}
+
+info_success() {
+  stderr
+  stderr
+  if [[ "${SUCCESS}" = 1 ]]; then
+    stderr "  Test was a success!"
+  else
+    stderr "  !!! TEST WAS A FAILURE!!!"
+  fi
+  stderr
+}
+
+main() {
+  assert "Use as root with sudo" [ "$(id -u)" -eq 0 ]
+  assert "Use as root with sudo" [ -n "${SUDO_UID}" ]
+  assert "SUDO_UID is 0; refusing to build as root" [ "${SUDO_UID}" -ne 0 ]
+
+  cleanup info_success
+
+  trap hook_cleanup EXIT
+
+  SCRIPT="$0"
+  CFG_TEMPLATE_DIR="$(abs_dir "$(dirname "${SCRIPT}")")"
+  REPO="$(abs_dir "${CFG_TEMPLATE_DIR}/../..")"
+  BINS="${REPO}/target/debug"
+
+  # Create temp dir
+  TMP_DIR="/tmp/rosenpass-psk-broker-test-$(date +%s)-$(uuidgen)"
+  cleanup rm -rf "${TMP_DIR}"
+  exc_as_user mkdir -p "${TMP_DIR}"
+
+  # Copy config
+  CFG_DIR="${TMP_DIR}/cfg"
+  exc_as_user cp -R "${CFG_TEMPLATE_DIR}" "${CFG_DIR}"
+
+  exc umask 077
+
+  exc cd "${REPO}"
+  local build_cmd; build_cmd=(cargo build --workspace --color=always --all-features --bins --profile dev)
+  if test -e "${BINS}/rosenpass-wireguard-broker-privileged" -a -e "${BINS}/rosenpass"; then
+    info "Found the binaries rosenpass-wireguard-broker-privileged and rosenpass." \
+      "Run following commands as a regular user to recompile the binaries with the right options" \
+      "in case of an error:" '$' "${build_cmd[@]}"
+  else
+    exc_as_user "${build_cmd[@]}"
+  fi
+  exc sudo setcap CAP_NET_ADMIN=+eip "${BINS}/rosenpass-wireguard-broker-privileged"
+
+  exc cd "${CFG_DIR}"
+  exc_eval_as_user "wg genkey > peer_a.wg.sk"
+  exc_eval_as_user "wg pubkey < peer_a.wg.sk > peer_a.wg.pk"
+  exc_eval_as_user "wg genkey > peer_b.wg.sk"
+  exc_eval_as_user "wg pubkey < peer_b.wg.sk > peer_b.wg.pk"
+  exc_eval_as_user "wg genpsk > peer_a_invalid.psk"
+  exc_eval_as_user "wg genpsk > peer_b_invalid.psk"
+  exc_eval_as_user "echo $(enquote "peer = \"$(cat peer_b.wg.pk)\"") >> peer_a.rp.config"
+  exc_eval_as_user "echo $(enquote "peer = \"$(cat peer_a.wg.pk)\"") >> peer_b.rp.config"
+  exc_as_user "${BINS}"/rosenpass gen-keys peer_a.rp.config
+  exc_as_user "${BINS}"/rosenpass gen-keys peer_b.rp.config
+
+  cleanup ip l del dev rpPskBrkTestA
+  cleanup ip l del dev rpPskBrkTestB
+  exc ip l add dev rpPskBrkTestA type wireguard
+  exc ip l add dev rpPskBrkTestB type wireguard
+
+  exc wg set rpPskBrkTestA \
+    listen-port 46125 \
+    private-key peer_a.wg.sk \
+    peer "$(cat peer_b.wg.pk)" \
+      endpoint 'localhost:46126' \
+      preshared-key peer_a_invalid.psk \
+      allowed-ips fe80::2/64
+  exc wg set rpPskBrkTestB \
+    listen-port 46126 \
+    private-key peer_b.wg.sk \
+    peer "$(cat peer_a.wg.pk)" \
+      endpoint 'localhost:46125' \
+      preshared-key peer_b_invalid.psk \
+      allowed-ips fe80::1/64
+
+  exc ip l set rpPskBrkTestA up
+  exc ip l set rpPskBrkTestB up
+
+  exc ip a add fe80::1/64 dev rpPskBrkTestA
+  exc ip a add fe80::2/64 dev rpPskBrkTestB
+
+  fork_eval_as_user "\
+    RUST_LOG='info' \
+    PATH=$(enquote "${REPO}/target/debug:${PATH}") \
+    $(enquote "${BINS}/rosenpass") --psk-broker-spawn \
+      exchange-config peer_a.rp.config"
+  fork_eval_as_user "\
+    RUST_LOG='info' \
+    PATH=$(enquote "${REPO}/target/debug:${PATH}") \
+    $(enquote "${BINS}/rosenpass-wireguard-broker-socket-handler") \
+      --listen-path broker.sock"
+  fork_eval_as_user "\
+    RUST_LOG='info' \
+    PATH=$(enquote "$PWD/target/debug:${PATH}") \
+    $(enquote "${BINS}/rosenpass") --psk-broker-path broker.sock \
+      exchange-config peer_b.rp.config"
+
+  exc_as_user ping -c 2 -w 10 fe80::1%rpPskBrkTestA
+  exc_as_user ping -c 2 -w 10 fe80::2%rpPskBrkTestB
+  exc_as_user ping -c 2 -w 10 fe80::2%rpPskBrkTestA
+  exc_as_user ping -c 2 -w 10 fe80::1%rpPskBrkTestB
+
+  SUCCESS=1
+}
+
+main "$@"

overlay.nix

@@ -0,0 +1,39 @@
+final: prev: {
+
+
+  #
+  ### Actual rosenpass software ###
+  #
+  rosenpass = final.callPackage ./pkgs/rosenpass.nix { };
+  rosenpass-oci-image = final.callPackage ./pkgs/rosenpass-oci-image.nix { };
+  rp = final.callPackage ./pkgs/rosenpass.nix { package = "rp"; };
+
+  release-package = final.callPackage ./pkgs/release-package.nix { };
+
+  #
+  ### Appendix ###
+  #
+  proverif-patched = prev.proverif.overrideAttrs (old: {
+    postInstall = ''
+      install -D -t $out/lib cryptoverif.pvl
+    '';
+  });
+
+  proof-proverif = final.stdenv.mkDerivation {
+    name = "rosenpass-proverif-proof";
+    version = "unstable";
+    src = final.lib.sources.sourceByRegex ./. [
+      "analyze.sh"
+      "marzipan(/marzipan.awk)?"
+      "analysis(/.*)?"
+    ];
+    nativeBuildInputs = [ final.proverif final.graphviz ];
+    CRYPTOVERIF_LIB = final.proverif-patched + "/lib/cryptoverif.pvl";
+    installPhase = ''
+      mkdir -p $out
+      bash analyze.sh -color -html $out
+    '';
+  };
+
+  whitepaper = final.callPackage ./pkgs/whitepaper.nix { };
+}

pkgs/release-package.nix

@@ -0,0 +1,27 @@
+{ lib, stdenvNoCC, runCommandNoCC, pkgsStatic, rosenpass, rosenpass-oci-image, rp } @ args:
+
+let
+  version = rosenpass.version;
+
+  # select static packages on Linux, default packages otherwise
+  package =
+    if stdenvNoCC.hostPlatform.isLinux then
+      pkgsStatic.rosenpass
+    else args.rosenpass;
+  rp =
+    if stdenvNoCC.hostPlatform.isLinux then
+      pkgsStatic.rp
+    else args.rp;
+  oci-image =
+    if stdenvNoCC.hostPlatform.isLinux then
+      pkgsStatic.rosenpass-oci-image
+    else args.rosenpass-oci-image;
+in
+runCommandNoCC "lace-result" { } ''
+  mkdir {bin,$out}
+  tar -cvf $out/rosenpass-${stdenvNoCC.hostPlatform.system}-${version}.tar \
+    -C ${package} bin/rosenpass \
+    -C ${rp} bin/rp
+  cp ${oci-image} \
+    $out/rosenpass-oci-image-${stdenvNoCC.hostPlatform.system}-${version}.tar.gz
+''

pkgs/rosenpass-oci-image.nix

@@ -0,0 +1,11 @@
+{ dockerTools, buildEnv, rosenpass }:
+
+dockerTools.buildImage {
+  name = rosenpass.name + "-oci";
+  copyToRoot = buildEnv {
+    name = "image-root";
+    paths = [ rosenpass ];
+    pathsToLink = [ "/bin" ];
+  };
+  config.Cmd = [ "/bin/rosenpass" ];
+}

pkgs/rosenpass.nix

@@ -0,0 +1,78 @@
+{ lib, stdenv, rustPlatform, cmake, mandoc, removeReferencesTo, bash, package ? "rosenpass" }:
+
+let
+  # whether we want to build a statically linked binary
+  isStatic = stdenv.targetPlatform.isStatic;
+
+  scoped = (scope: scope.result);
+
+  # source files relevant for rust
+  src = scoped rec {
+    # File suffices to include
+    extensions = [
+      "lock"
+      "rs"
+      "toml"
+    ];
+    # Files to explicitly include
+    files = [
+      "to/README.md"
+    ];
+
+    src = ../.;
+    filter = (path: type: scoped rec {
+      inherit (lib) any id removePrefix hasSuffix;
+      anyof = (any id);
+
+      basename = baseNameOf (toString path);
+      relative = removePrefix (toString src + "/") (toString path);
+
+      result = anyof [
+        (type == "directory")
+        (any (ext: hasSuffix ".${ext}" basename) extensions)
+        (any (file: file == relative) files)
+      ];
+    });
+
+    result = lib.sources.cleanSourceWith { inherit src filter; };
+  };
+
+  # parsed Cargo.toml
+  cargoToml = builtins.fromTOML (builtins.readFile (src + "/rosenpass/Cargo.toml"));
+in
+rustPlatform.buildRustPackage {
+  name = cargoToml.package.name;
+  version = cargoToml.package.version;
+  inherit src;
+
+  cargoBuildOptions = [ "--package" package ];
+  cargoTestOptions = [ "--package" package ];
+
+  doCheck = true;
+
+  cargoLock = {
+    lockFile = src + "/Cargo.lock";
+    outputHashes = {
+      "memsec-0.6.3" = "sha256-4ri+IEqLd77cLcul3lZrmpDKj4cwuYJ8oPRAiQNGeLw=";
+      "uds-0.4.2" = "sha256-qlxr/iJt2AV4WryePIvqm/8/MK/iqtzegztNliR93W8=";
+    };
+  };
+
+  nativeBuildInputs = [
+    stdenv.cc
+    cmake # for oqs build in the oqs-sys crate
+    mandoc # for the built-in manual
+    removeReferencesTo
+    rustPlatform.bindgenHook # for C-bindings in the crypto libs
+  ];
+  buildInputs = [ bash ];
+
+  hardeningDisable = lib.optional isStatic "fortify";
+
+  meta = {
+    inherit (cargoToml.package) description homepage;
+    license = with lib.licenses; [ mit asl20 ];
+    maintainers = [ lib.maintainers.wucke13 ];
+    platforms = lib.platforms.all;
+  };
+}

pkgs/whitepaper.nix

@@ -0,0 +1,29 @@
+{ stdenvNoCC, texlive, ncurses, python3Packages, which }:
+
+let
+  customTexLiveSetup = (texlive.combine {
+    inherit (texlive) acmart amsfonts biber biblatex biblatex-software
+      biblatex-trad ccicons csquotes csvsimple doclicense eso-pic fancyvrb
+      fontspec gitinfo2 gobble ifmtarg koma-script latexmk lm lualatex-math
+      markdown mathtools minted noto nunito paralist pgf scheme-basic soul
+      unicode-math upquote xifthen xkeyval xurl;
+  });
+in
+stdenvNoCC.mkDerivation {
+  name = "whitepaper";
+  src = ../papers;
+  nativeBuildInputs = [
+    ncurses # tput
+    python3Packages.pygments
+    customTexLiveSetup # custom tex live scheme
+    which
+  ];
+  buildPhase = ''
+    export HOME=$(mktemp -d)
+    latexmk -r tex/CI.rc
+  '';
+  installPhase = ''
+    mkdir -p $out
+    mv *.pdf readme.md $out/
+  '';
+}

rosenpass/Cargo.toml

@@ -22,6 +22,10 @@ required-features = ["experiment_api", "internal_bin_gen_ipc_msg_types"]
 name = "api-integration-tests"
 required-features = ["experiment_api", "internal_testing"]
 
+[[test]]
+name = "api-integration-tests-api-setup"
+required-features = ["experiment_api", "internal_testing"]
+
 [[bench]]
 name = "handshake"
 harness = false
@@ -47,12 +51,15 @@ mio = { workspace = true }
 rand = { workspace = true }
 zerocopy = { workspace = true }
 home = { workspace = true }
-derive_builder = {workspace = true}
-rosenpass-wireguard-broker = {workspace = true}
+derive_builder = { workspace = true }
+rosenpass-wireguard-broker = { workspace = true }
 zeroize = { workspace = true }
 hex-literal = { workspace = true, optional = true }
 hex = { workspace = true, optional = true }
 heck = { workspace = true, optional = true }
+command-fds = { workspace = true, optional = true }
+rustix = { workspace = true }
+uds = { workspace = true, optional = true, features = ["mio_1xx"] }
 
 [build-dependencies]
 anyhow = { workspace = true }
@@ -61,14 +68,21 @@ anyhow = { workspace = true }
 criterion = { workspace = true }
 test_bin = { workspace = true }
 stacker = { workspace = true }
-serial_test = {workspace = true}
-procspawn = {workspace = true}
+serial_test = { workspace = true }
+procspawn = { workspace = true }
 tempfile = { workspace = true }
+rustix = { workspace = true }
 
 [features]
-enable_broker_api = ["rosenpass-wireguard-broker/enable_broker_api"]
-experiment_memfd_secret = []
+default = ["experiment_api"]
+experiment_memfd_secret = ["rosenpass-wireguard-broker/experiment_memfd_secret"]
 experiment_libcrux = ["rosenpass-ciphers/experiment_libcrux"]
-experiment_api = ["hex-literal"]
-internal_testing  = []
+experiment_api = [
+    "hex-literal",
+    "uds",
+    "command-fds",
+    "rosenpass-util/experiment_file_descriptor_passing",
+    "rosenpass-wireguard-broker/experiment_api",
+]
+internal_testing = []
 internal_bin_gen_ipc_msg_types = ["hex", "heck"]

rosenpass/src/api/api_handler.rs

@@ -0,0 +1,295 @@
+use std::{borrow::BorrowMut, collections::VecDeque, os::fd::OwnedFd};
+
+use anyhow::Context;
+use rosenpass_to::{ops::copy_slice, To};
+use rosenpass_util::{
+    fd::FdIo,
+    functional::{run, ApplyExt},
+    io::ReadExt,
+    mem::DiscardResultExt,
+    mio::UnixStreamExt,
+    result::OkExt,
+};
+use rosenpass_wireguard_broker::brokers::mio_client::MioBrokerClient;
+
+use crate::{
+    api::{add_listen_socket_response_status, add_psk_broker_response_status},
+    app_server::AppServer,
+    protocol::BuildCryptoServer,
+};
+
+use super::{supply_keypair_response_status, Server as ApiServer};
+
+#[derive(Debug)]
+pub struct ApiHandler {
+    _dummy: (),
+}
+
+impl ApiHandler {
+    #[allow(clippy::new_without_default)]
+    pub fn new() -> Self {
+        Self { _dummy: () }
+    }
+}
+
+pub trait ApiHandlerContext {
+    fn api_handler(&self) -> &ApiHandler;
+    fn app_server(&self) -> &AppServer;
+    fn api_handler_mut(&mut self) -> &mut ApiHandler;
+    fn app_server_mut(&mut self) -> &mut AppServer;
+}
+
+#[derive(thiserror::Error, Debug)]
+#[error("Error in SupplyKeypair")]
+struct SupplyKeypairError {
+    status: u128,
+    #[source]
+    cause: anyhow::Error,
+}
+
+trait SupplyKeypairErrorExt<T> {
+    fn e_custom(self, status: u128) -> Result<T, SupplyKeypairError>;
+    fn einternal(self) -> Result<T, SupplyKeypairError>;
+    fn ealready_supplied(self) -> Result<T, SupplyKeypairError>;
+    fn einvalid_req(self) -> Result<T, SupplyKeypairError>;
+}
+
+impl<T, E: Into<anyhow::Error>> SupplyKeypairErrorExt<T> for Result<T, E> {
+    fn e_custom(self, status: u128) -> Result<T, SupplyKeypairError> {
+        self.map_err(|e| SupplyKeypairError {
+            status,
+            cause: e.into(),
+        })
+    }
+
+    fn einternal(self) -> Result<T, SupplyKeypairError> {
+        self.e_custom(supply_keypair_response_status::INTERNAL_ERROR)
+    }
+
+    fn ealready_supplied(self) -> Result<T, SupplyKeypairError> {
+        self.e_custom(supply_keypair_response_status::KEYPAIR_ALREADY_SUPPLIED)
+    }
+
+    fn einvalid_req(self) -> Result<T, SupplyKeypairError> {
+        self.e_custom(supply_keypair_response_status::INVALID_REQUEST)
+    }
+}
+
+impl<T> ApiServer for T
+where
+    T: ?Sized + ApiHandlerContext,
+{
+    fn ping(
+        &mut self,
+        req: &super::PingRequest,
+        _req_fds: &mut VecDeque<OwnedFd>,
+        res: &mut super::PingResponse,
+    ) -> anyhow::Result<()> {
+        let (req, res) = (&req.payload, &mut res.payload);
+        copy_slice(&req.echo).to(&mut res.echo);
+        Ok(())
+    }
+
+    fn supply_keypair(
+        &mut self,
+        req: &super::SupplyKeypairRequest,
+        req_fds: &mut VecDeque<OwnedFd>,
+        res: &mut super::SupplyKeypairResponse,
+    ) -> anyhow::Result<()> {
+        let outcome: Result<(), SupplyKeypairError> = run(|| {
+            // Acquire the file descriptors
+            let mut sk_io = FdIo(
+                req_fds
+                    .front()
+                    .context("First file descriptor, secret key, missing.")
+                    .einvalid_req()?,
+            );
+            let mut pk_io = FdIo(
+                req_fds
+                    .get(1)
+                    .context("Second file descriptor, public key, missing.")
+                    .einvalid_req()?,
+            );
+
+            // Actually read the secrets
+            let mut sk = crate::protocol::SSk::zero();
+            sk_io.read_exact_til_end(sk.secret_mut()).einvalid_req()?;
+
+            let mut pk = crate::protocol::SPk::zero();
+            pk_io.read_exact_til_end(pk.borrow_mut()).einvalid_req()?;
+
+            // Retrieve the construction site
+            let construction_site = self.app_server_mut().crypto_site.borrow_mut();
+
+            // Retrieve the builder
+            use rosenpass_util::build::ConstructionSite as C;
+            let maybe_builder = match construction_site {
+                C::Builder(builder) => Some(builder),
+                C::Product(_) => None,
+                C::Void => {
+                    return Err(anyhow::Error::msg("CryptoServer construction side is void"))
+                        .einternal();
+                }
+            };
+
+            // Retrieve a reference to the keypair
+            let Some(BuildCryptoServer {
+                ref mut keypair, ..
+            }) = maybe_builder
+            else {
+                return Err(anyhow::Error::msg("CryptoServer already built")).ealready_supplied();
+            };
+
+            // Supply the keypair to the CryptoServer
+            keypair
+                .insert(crate::protocol::Keypair { sk, pk })
+                .discard_result();
+
+            // Actually construct the CryptoServer
+            construction_site
+                .erect()
+                .map_err(|e| anyhow::Error::msg(format!("Error erecting the CryptoServer {e:?}")))
+                .einternal()?;
+
+            Ok(())
+        });
+
+        // Handle errors
+        use supply_keypair_response_status as status;
+        let status = match outcome {
+            Ok(()) => status::OK,
+            Err(e) => {
+                let lvl = match e.status {
+                    status::INTERNAL_ERROR => log::Level::Warn,
+                    _ => log::Level::Debug,
+                };
+
+                log::log!(
+                    lvl,
+                    "Error while processing API Request.\n    Request: {:?}\n    Error: {:?}",
+                    req,
+                    e.cause
+                );
+
+                if e.status == status::INTERNAL_ERROR {
+                    return Err(e.cause);
+                }
+
+                e.status
+            }
+        };
+
+        res.payload.status = status;
+
+        Ok(())
+    }
+
+    fn add_listen_socket(
+        &mut self,
+        _req: &super::boilerplate::AddListenSocketRequest,
+        req_fds: &mut VecDeque<OwnedFd>,
+        res: &mut super::boilerplate::AddListenSocketResponse,
+    ) -> anyhow::Result<()> {
+        // Retrieve file descriptor
+        let sock_res = run(|| -> anyhow::Result<mio::net::UdpSocket> {
+            let sock = req_fds
+                .pop_front()
+                .context("Invalid request – socket missing.")?;
+            // TODO: We need to have this outside linux
+            #[cfg(target_os = "linux")]
+            rosenpass_util::fd::GetSocketProtocol::demand_udp_socket(&sock)?;
+            let sock = std::net::UdpSocket::from(sock);
+            sock.set_nonblocking(true)?;
+            mio::net::UdpSocket::from_std(sock).ok()
+        });
+
+        let sock = match sock_res {
+            Ok(sock) => sock,
+            Err(e) => {
+                log::debug!("Error processing AddListenSocket API request: {e:?}");
+                res.payload.status = add_listen_socket_response_status::INVALID_REQUEST;
+                return Ok(());
+            }
+        };
+
+        // Register socket
+        let reg_result = self.app_server_mut().register_listen_socket(sock);
+
+        if let Err(internal_error) = reg_result {
+            log::warn!("Internal error processing AddListenSocket API request: {internal_error:?}");
+            res.payload.status = add_listen_socket_response_status::INTERNAL_ERROR;
+            return Ok(());
+        };
+
+        res.payload.status = add_listen_socket_response_status::OK;
+        Ok(())
+    }
+
+    fn add_psk_broker(
+        &mut self,
+        _req: &super::boilerplate::AddPskBrokerRequest,
+        req_fds: &mut VecDeque<OwnedFd>,
+        res: &mut super::boilerplate::AddPskBrokerResponse,
+    ) -> anyhow::Result<()> {
+        // Retrieve file descriptor
+        let sock_res = run(|| {
+            let sock = req_fds
+                .pop_front()
+                .context("Invalid request – socket missing.")?;
+            mio::net::UnixStream::from_fd(sock)
+        });
+
+        // Handle errors
+        let sock = match sock_res {
+            Ok(sock) => sock,
+            Err(e) => {
+                log::debug!(
+                    "Request found to be invalid while processing AddPskBroker API request: {e:?}"
+                );
+                res.payload.status = add_psk_broker_response_status::INVALID_REQUEST;
+                return Ok(());
+            }
+        };
+
+        // Register Socket
+        let client = Box::new(MioBrokerClient::new(sock));
+
+        // Workaround: The broker code is currently impressively overcomplicated. Brokers are
+        // stored in a hash map but the hash map key used is just a counter so a vector could
+        // have been used. Broker configuration is abstracted, different peers can have different
+        // brokers but there is no facility to add multiple brokers in practice. The broker index
+        // uses a `Public` wrapper without actually holding any cryptographic data. Even the broker
+        // configuration uses a trait abstraction for no discernible reason and a lot of the code
+        // introduces pointless, single-field wrapper structs.
+        // We should use an implement-what-is-actually-needed strategy next time.
+        // The Broker code needs to be slimmed down, the right direction to go is probably to
+        // just add event and capability support to the API and use the API to deliver OSK events.
+        //
+        // For now, we just replace the latest broker.
+        let erase_ptr = {
+            use crate::app_server::BrokerStorePtr;
+            //
+            use rosenpass_secret_memory::Public;
+            use zerocopy::AsBytes;
+            (self.app_server().brokers.store.len() - 1)
+                .apply(|x| x as u64)
+                .apply(|x| Public::from_slice(x.as_bytes()))
+                .apply(BrokerStorePtr)
+        };
+
+        let register_result = run(|| {
+            let srv = self.app_server_mut();
+            srv.unregister_broker(erase_ptr)?;
+            srv.register_broker(client)
+        });
+
+        if let Err(e) = register_result {
+            log::warn!("Internal error while processing AddPskBroker API request: {e:?}");
+            res.payload.status = add_psk_broker_response_status::INTERNAL_ERROR;
+            return Ok(());
+        }
+
+        res.payload.status = add_psk_broker_response_status::OK;
+        Ok(())
+    }
+}

rosenpass/src/api/boilerplate/byte_slice_ext.rs

@@ -4,7 +4,7 @@ use rosenpass_util::zerocopy::{RefMaker, ZerocopySliceExt};
 
 use super::{
     PingRequest, PingResponse, RawMsgType, RefMakerRawMsgTypeExt, RequestMsgType, RequestRef,
-    ResponseMsgType, ResponseRef,
+    ResponseMsgType, ResponseRef, SupplyKeypairRequest, SupplyKeypairResponse,
 };
 
 pub trait ByteSliceRefExt: ByteSlice {
@@ -111,6 +111,112 @@ pub trait ByteSliceRefExt: ByteSlice {
     fn ping_response_from_suffix(self) -> anyhow::Result<Ref<Self, PingResponse>> {
         self.zk_parse_suffix()
     }
+
+    fn supply_keypair_request(self) -> anyhow::Result<Ref<Self, SupplyKeypairRequest>> {
+        self.zk_parse()
+    }
+
+    fn supply_keypair_request_from_prefix(self) -> anyhow::Result<Ref<Self, SupplyKeypairRequest>> {
+        self.zk_parse_prefix()
+    }
+
+    fn supply_keypair_request_from_suffix(self) -> anyhow::Result<Ref<Self, SupplyKeypairRequest>> {
+        self.zk_parse_suffix()
+    }
+
+    fn supply_keypair_response_maker(self) -> RefMaker<Self, SupplyKeypairResponse> {
+        self.zk_ref_maker()
+    }
+
+    fn supply_keypair_response(self) -> anyhow::Result<Ref<Self, SupplyKeypairResponse>> {
+        self.zk_parse()
+    }
+
+    fn supply_keypair_response_from_prefix(
+        self,
+    ) -> anyhow::Result<Ref<Self, SupplyKeypairResponse>> {
+        self.zk_parse_prefix()
+    }
+
+    fn supply_keypair_response_from_suffix(
+        self,
+    ) -> anyhow::Result<Ref<Self, SupplyKeypairResponse>> {
+        self.zk_parse_suffix()
+    }
+
+    fn add_listen_socket_request(self) -> anyhow::Result<Ref<Self, super::AddListenSocketRequest>> {
+        self.zk_parse()
+    }
+
+    fn add_listen_socket_request_from_prefix(
+        self,
+    ) -> anyhow::Result<Ref<Self, super::AddListenSocketRequest>> {
+        self.zk_parse_prefix()
+    }
+
+    fn add_listen_socket_request_from_suffix(
+        self,
+    ) -> anyhow::Result<Ref<Self, super::AddListenSocketRequest>> {
+        self.zk_parse_suffix()
+    }
+
+    fn add_listen_socket_response_maker(self) -> RefMaker<Self, super::AddListenSocketResponse> {
+        self.zk_ref_maker()
+    }
+
+    fn add_listen_socket_response(
+        self,
+    ) -> anyhow::Result<Ref<Self, super::AddListenSocketResponse>> {
+        self.zk_parse()
+    }
+
+    fn add_listen_socket_response_from_prefix(
+        self,
+    ) -> anyhow::Result<Ref<Self, super::AddListenSocketResponse>> {
+        self.zk_parse_prefix()
+    }
+
+    fn add_listen_socket_response_from_suffix(
+        self,
+    ) -> anyhow::Result<Ref<Self, super::AddListenSocketResponse>> {
+        self.zk_parse_suffix()
+    }
+
+    fn add_psk_broker_request(self) -> anyhow::Result<Ref<Self, super::AddPskBrokerRequest>> {
+        self.zk_parse()
+    }
+
+    fn add_psk_broker_request_from_prefix(
+        self,
+    ) -> anyhow::Result<Ref<Self, super::AddPskBrokerRequest>> {
+        self.zk_parse_prefix()
+    }
+
+    fn add_psk_broker_request_from_suffix(
+        self,
+    ) -> anyhow::Result<Ref<Self, super::AddPskBrokerRequest>> {
+        self.zk_parse_suffix()
+    }
+
+    fn add_psk_broker_response_maker(self) -> RefMaker<Self, super::AddPskBrokerResponse> {
+        self.zk_ref_maker()
+    }
+
+    fn add_psk_broker_response(self) -> anyhow::Result<Ref<Self, super::AddPskBrokerResponse>> {
+        self.zk_parse()
+    }
+
+    fn add_psk_broker_response_from_prefix(
+        self,
+    ) -> anyhow::Result<Ref<Self, super::AddPskBrokerResponse>> {
+        self.zk_parse_prefix()
+    }
+
+    fn add_psk_broker_response_from_suffix(
+        self,
+    ) -> anyhow::Result<Ref<Self, super::AddPskBrokerResponse>> {
+        self.zk_parse_suffix()
+    }
 }
 
 impl<B: ByteSlice> ByteSliceRefExt for B {}

rosenpass/src/api/boilerplate/message_type.rs

@@ -14,6 +14,27 @@ pub const PING_REQUEST: RawMsgType =
 pub const PING_RESPONSE: RawMsgType =
     RawMsgType::from_le_bytes(hex!("4ec7 f6f0 2bbc ba64    48f1 da14 c7cf 0260"));
 
+// hash domain hash of: Rosenpass IPC API -> Rosenpass Protocol Server -> Supply Keypair Request
+const SUPPLY_KEYPAIR_REQUEST: RawMsgType =
+    RawMsgType::from_le_bytes(hex!("ac91 a5a6 4f4b 21d0    ac7f 9b55 74f7 3529"));
+// hash domain hash of: Rosenpass IPC API -> Rosenpass Protocol Server -> Supply Keypair Response
+const SUPPLY_KEYPAIR_RESPONSE: RawMsgType =
+    RawMsgType::from_le_bytes(hex!("f2dc 49bd e261 5f10    40b7 3c16 ec61 edb9"));
+
+// hash domain hash of: Rosenpass IPC API -> Rosenpass Protocol Server -> Add Listen Socket Request
+const ADD_LISTEN_SOCKET_REQUEST: RawMsgType =
+    RawMsgType::from_le_bytes(hex!("3f21 434f 87cc a08c    02c4 61e4 0816 c7da"));
+// hash domain hash of: Rosenpass IPC API -> Rosenpass Protocol Server -> Add Listen Socket Response
+const ADD_LISTEN_SOCKET_RESPONSE: RawMsgType =
+    RawMsgType::from_le_bytes(hex!("45d5 0f0d 93f0 6105    98f2 9469 5dfd 5f36"));
+
+// hash domain hash of: Rosenpass IPC API -> Rosenpass Protocol Server -> Add Psk Broker Request
+const ADD_PSK_BROKER_REQUEST: RawMsgType =
+    RawMsgType::from_le_bytes(hex!("d798 b8dc bd61 5cab    8df1 c63d e4eb a2d1"));
+// hash domain hash of: Rosenpass IPC API -> Rosenpass Protocol Server -> Add Psk Broker Response
+const ADD_PSK_BROKER_RESPONSE: RawMsgType =
+    RawMsgType::from_le_bytes(hex!("bd25 e418 ffb0 6930    248b 217e 2fae e353"));
+
 pub trait MessageAttributes {
     fn message_size(&self) -> usize;
 }
@@ -21,17 +42,26 @@ pub trait MessageAttributes {
 #[derive(Hash, PartialEq, Eq, PartialOrd, Ord, Debug, Clone, Copy)]
 pub enum RequestMsgType {
     Ping,
+    SupplyKeypair,
+    AddListenSocket,
+    AddPskBroker,
 }
 
 #[derive(Hash, PartialEq, Eq, PartialOrd, Ord, Debug, Clone, Copy)]
 pub enum ResponseMsgType {
     Ping,
+    SupplyKeypair,
+    AddListenSocket,
+    AddPskBroker,
 }
 
 impl MessageAttributes for RequestMsgType {
     fn message_size(&self) -> usize {
         match self {
             Self::Ping => std::mem::size_of::<super::PingRequest>(),
+            Self::SupplyKeypair => std::mem::size_of::<super::SupplyKeypairRequest>(),
+            Self::AddListenSocket => std::mem::size_of::<super::AddListenSocketRequest>(),
+            Self::AddPskBroker => std::mem::size_of::<super::AddPskBrokerRequest>(),
         }
     }
 }
@@ -40,6 +70,9 @@ impl MessageAttributes for ResponseMsgType {
     fn message_size(&self) -> usize {
         match self {
             Self::Ping => std::mem::size_of::<super::PingResponse>(),
+            Self::SupplyKeypair => std::mem::size_of::<super::SupplyKeypairResponse>(),
+            Self::AddListenSocket => std::mem::size_of::<super::AddListenSocketResponse>(),
+            Self::AddPskBroker => std::mem::size_of::<super::AddPskBrokerResponse>(),
         }
     }
 }
@@ -51,6 +84,9 @@ impl TryFrom<RawMsgType> for RequestMsgType {
         use RequestMsgType as E;
         Ok(match value {
             self::PING_REQUEST => E::Ping,
+            self::SUPPLY_KEYPAIR_REQUEST => E::SupplyKeypair,
+            self::ADD_LISTEN_SOCKET_REQUEST => E::AddListenSocket,
+            self::ADD_PSK_BROKER_REQUEST => E::AddPskBroker,
             _ => return Err(InvalidApiMessageType(value)),
         })
     }
@@ -61,6 +97,9 @@ impl From<RequestMsgType> for RawMsgType {
         use RequestMsgType as E;
         match val {
             E::Ping => self::PING_REQUEST,
+            E::SupplyKeypair => self::SUPPLY_KEYPAIR_REQUEST,
+            E::AddListenSocket => self::ADD_LISTEN_SOCKET_REQUEST,
+            E::AddPskBroker => self::ADD_PSK_BROKER_REQUEST,
         }
     }
 }
@@ -72,6 +111,9 @@ impl TryFrom<RawMsgType> for ResponseMsgType {
         use ResponseMsgType as E;
         Ok(match value {
             self::PING_RESPONSE => E::Ping,
+            self::SUPPLY_KEYPAIR_RESPONSE => E::SupplyKeypair,
+            self::ADD_LISTEN_SOCKET_RESPONSE => E::AddListenSocket,
+            self::ADD_PSK_BROKER_RESPONSE => E::AddPskBroker,
             _ => return Err(InvalidApiMessageType(value)),
         })
     }
@@ -82,6 +124,9 @@ impl From<ResponseMsgType> for RawMsgType {
         use ResponseMsgType as E;
         match val {
             E::Ping => self::PING_RESPONSE,
+            E::SupplyKeypair => self::SUPPLY_KEYPAIR_RESPONSE,
+            E::AddListenSocket => self::ADD_LISTEN_SOCKET_RESPONSE,
+            E::AddPskBroker => self::ADD_PSK_BROKER_RESPONSE,
         }
     }
 }

rosenpass/src/api/boilerplate/payload.rs

@@ -6,6 +6,7 @@ use super::{Message, RawMsgType, RequestMsgType, ResponseMsgType};
 /// Size required to fit any message in binary form
 pub const MAX_REQUEST_LEN: usize = 2500; // TODO fix this
 pub const MAX_RESPONSE_LEN: usize = 2500; // TODO fix this
+pub const MAX_REQUEST_FDS: usize = 2;
 
 #[repr(packed)]
 #[derive(Debug, Copy, Clone, Hash, AsBytes, FromBytes, FromZeroes, PartialEq, Eq)]
@@ -94,3 +95,257 @@ impl Message for PingResponse {
         self.msg_type = Self::MESSAGE_TYPE.into();
     }
 }
+
+#[repr(packed)]
+#[derive(Debug, Copy, Clone, Hash, AsBytes, FromBytes, FromZeroes, PartialEq, Eq)]
+pub struct SupplyKeypairRequestPayload {}
+
+pub type SupplyKeypairRequest = RequestEnvelope<SupplyKeypairRequestPayload>;
+
+impl Default for SupplyKeypairRequest {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl SupplyKeypairRequest {
+    pub fn new() -> Self {
+        Self::from_payload(SupplyKeypairRequestPayload {})
+    }
+}
+
+impl Message for SupplyKeypairRequest {
+    type Payload = SupplyKeypairRequestPayload;
+    type MessageClass = RequestMsgType;
+    const MESSAGE_TYPE: Self::MessageClass = RequestMsgType::SupplyKeypair;
+
+    fn from_payload(payload: Self::Payload) -> Self {
+        Self {
+            msg_type: Self::MESSAGE_TYPE.into(),
+            payload,
+        }
+    }
+
+    fn setup<B: ByteSliceMut>(buf: B) -> anyhow::Result<Ref<B, Self>> {
+        let mut r: Ref<B, Self> = buf.zk_zeroized()?;
+        r.init();
+        Ok(r)
+    }
+
+    fn init(&mut self) {
+        self.msg_type = Self::MESSAGE_TYPE.into();
+    }
+}
+
+pub mod supply_keypair_response_status {
+    pub const OK: u128 = 0;
+    pub const KEYPAIR_ALREADY_SUPPLIED: u128 = 1;
+    pub const INTERNAL_ERROR: u128 = 2;
+    pub const INVALID_REQUEST: u128 = 3;
+    pub const IO_ERROR: u128 = 4;
+}
+
+#[repr(packed)]
+#[derive(Debug, Copy, Clone, Hash, AsBytes, FromBytes, FromZeroes, PartialEq, Eq)]
+pub struct SupplyKeypairResponsePayload {
+    pub status: u128,
+}
+
+pub type SupplyKeypairResponse = ResponseEnvelope<SupplyKeypairResponsePayload>;
+
+impl SupplyKeypairResponse {
+    pub fn new(status: u128) -> Self {
+        Self::from_payload(SupplyKeypairResponsePayload { status })
+    }
+}
+
+impl Message for SupplyKeypairResponse {
+    type Payload = SupplyKeypairResponsePayload;
+    type MessageClass = ResponseMsgType;
+    const MESSAGE_TYPE: Self::MessageClass = ResponseMsgType::SupplyKeypair;
+
+    fn from_payload(payload: Self::Payload) -> Self {
+        Self {
+            msg_type: Self::MESSAGE_TYPE.into(),
+            payload,
+        }
+    }
+
+    fn setup<B: ByteSliceMut>(buf: B) -> anyhow::Result<Ref<B, Self>> {
+        let mut r: Ref<B, Self> = buf.zk_zeroized()?;
+        r.init();
+        Ok(r)
+    }
+
+    fn init(&mut self) {
+        self.msg_type = Self::MESSAGE_TYPE.into();
+    }
+}
+
+#[repr(packed)]
+#[derive(Debug, Copy, Clone, Hash, AsBytes, FromBytes, FromZeroes, PartialEq, Eq)]
+pub struct AddListenSocketRequestPayload {}
+
+pub type AddListenSocketRequest = RequestEnvelope<AddListenSocketRequestPayload>;
+
+impl Default for AddListenSocketRequest {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl AddListenSocketRequest {
+    pub fn new() -> Self {
+        Self::from_payload(AddListenSocketRequestPayload {})
+    }
+}
+
+impl Message for AddListenSocketRequest {
+    type Payload = AddListenSocketRequestPayload;
+    type MessageClass = RequestMsgType;
+    const MESSAGE_TYPE: Self::MessageClass = RequestMsgType::AddListenSocket;
+
+    fn from_payload(payload: Self::Payload) -> Self {
+        Self {
+            msg_type: Self::MESSAGE_TYPE.into(),
+            payload,
+        }
+    }
+
+    fn setup<B: ByteSliceMut>(buf: B) -> anyhow::Result<Ref<B, Self>> {
+        let mut r: Ref<B, Self> = buf.zk_zeroized()?;
+        r.init();
+        Ok(r)
+    }
+
+    fn init(&mut self) {
+        self.msg_type = Self::MESSAGE_TYPE.into();
+    }
+}
+
+pub mod add_listen_socket_response_status {
+    pub const OK: u128 = 0;
+    pub const INVALID_REQUEST: u128 = 1;
+    pub const INTERNAL_ERROR: u128 = 2;
+}
+
+#[repr(packed)]
+#[derive(Debug, Copy, Clone, Hash, AsBytes, FromBytes, FromZeroes, PartialEq, Eq)]
+pub struct AddListenSocketResponsePayload {
+    pub status: u128,
+}
+
+pub type AddListenSocketResponse = ResponseEnvelope<AddListenSocketResponsePayload>;
+
+impl AddListenSocketResponse {
+    pub fn new(status: u128) -> Self {
+        Self::from_payload(AddListenSocketResponsePayload { status })
+    }
+}
+
+impl Message for AddListenSocketResponse {
+    type Payload = AddListenSocketResponsePayload;
+    type MessageClass = ResponseMsgType;
+    const MESSAGE_TYPE: Self::MessageClass = ResponseMsgType::AddListenSocket;
+
+    fn from_payload(payload: Self::Payload) -> Self {
+        Self {
+            msg_type: Self::MESSAGE_TYPE.into(),
+            payload,
+        }
+    }
+
+    fn setup<B: ByteSliceMut>(buf: B) -> anyhow::Result<Ref<B, Self>> {
+        let mut r: Ref<B, Self> = buf.zk_zeroized()?;
+        r.init();
+        Ok(r)
+    }
+
+    fn init(&mut self) {
+        self.msg_type = Self::MESSAGE_TYPE.into();
+    }
+}
+
+#[repr(packed)]
+#[derive(Debug, Copy, Clone, Hash, AsBytes, FromBytes, FromZeroes, PartialEq, Eq)]
+pub struct AddPskBrokerRequestPayload {}
+
+pub type AddPskBrokerRequest = RequestEnvelope<AddPskBrokerRequestPayload>;
+
+impl Default for AddPskBrokerRequest {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl AddPskBrokerRequest {
+    pub fn new() -> Self {
+        Self::from_payload(AddPskBrokerRequestPayload {})
+    }
+}
+
+impl Message for AddPskBrokerRequest {
+    type Payload = AddPskBrokerRequestPayload;
+    type MessageClass = RequestMsgType;
+    const MESSAGE_TYPE: Self::MessageClass = RequestMsgType::AddPskBroker;
+
+    fn from_payload(payload: Self::Payload) -> Self {
+        Self {
+            msg_type: Self::MESSAGE_TYPE.into(),
+            payload,
+        }
+    }
+
+    fn setup<B: ByteSliceMut>(buf: B) -> anyhow::Result<Ref<B, Self>> {
+        let mut r: Ref<B, Self> = buf.zk_zeroized()?;
+        r.init();
+        Ok(r)
+    }
+
+    fn init(&mut self) {
+        self.msg_type = Self::MESSAGE_TYPE.into();
+    }
+}
+
+pub mod add_psk_broker_response_status {
+    pub const OK: u128 = 0;
+    pub const INVALID_REQUEST: u128 = 1;
+    pub const INTERNAL_ERROR: u128 = 2;
+}
+
+#[repr(packed)]
+#[derive(Debug, Copy, Clone, Hash, AsBytes, FromBytes, FromZeroes, PartialEq, Eq)]
+pub struct AddPskBrokerResponsePayload {
+    pub status: u128,
+}
+
+pub type AddPskBrokerResponse = ResponseEnvelope<AddPskBrokerResponsePayload>;
+
+impl AddPskBrokerResponse {
+    pub fn new(status: u128) -> Self {
+        Self::from_payload(AddPskBrokerResponsePayload { status })
+    }
+}
+
+impl Message for AddPskBrokerResponse {
+    type Payload = AddPskBrokerResponsePayload;
+    type MessageClass = ResponseMsgType;
+    const MESSAGE_TYPE: Self::MessageClass = ResponseMsgType::AddPskBroker;
+
+    fn from_payload(payload: Self::Payload) -> Self {
+        Self {
+            msg_type: Self::MESSAGE_TYPE.into(),
+            payload,
+        }
+    }
+
+    fn setup<B: ByteSliceMut>(buf: B) -> anyhow::Result<Ref<B, Self>> {
+        let mut r: Ref<B, Self> = buf.zk_zeroized()?;
+        r.init();
+        Ok(r)
+    }
+
+    fn init(&mut self) {
+        self.msg_type = Self::MESSAGE_TYPE.into();
+    }
+}

rosenpass/src/api/boilerplate/request_ref.rs

@@ -25,6 +25,9 @@ impl<B: ByteSlice> RequestRef<B> {
     pub fn message_type(&self) -> RequestMsgType {
         match self {
             Self::Ping(_) => RequestMsgType::Ping,
+            Self::SupplyKeypair(_) => RequestMsgType::SupplyKeypair,
+            Self::AddListenSocket(_) => RequestMsgType::AddListenSocket,
+            Self::AddPskBroker(_) => RequestMsgType::AddPskBroker,
         }
     }
 }
@@ -35,6 +38,24 @@ impl<B> From<Ref<B, PingRequest>> for RequestRef<B> {
     }
 }
 
+impl<B> From<Ref<B, super::SupplyKeypairRequest>> for RequestRef<B> {
+    fn from(v: Ref<B, super::SupplyKeypairRequest>) -> Self {
+        Self::SupplyKeypair(v)
+    }
+}
+
+impl<B> From<Ref<B, super::AddListenSocketRequest>> for RequestRef<B> {
+    fn from(v: Ref<B, super::AddListenSocketRequest>) -> Self {
+        Self::AddListenSocket(v)
+    }
+}
+
+impl<B> From<Ref<B, super::AddPskBrokerRequest>> for RequestRef<B> {
+    fn from(v: Ref<B, super::AddPskBrokerRequest>) -> Self {
+        Self::AddPskBroker(v)
+    }
+}
+
 impl<B: ByteSlice> RequestRefMaker<B> {
     fn new(buf: B) -> anyhow::Result<Self> {
         let msg_type = buf.deref().request_msg_type_from_prefix()?;
@@ -48,6 +69,15 @@ impl<B: ByteSlice> RequestRefMaker<B> {
     fn parse(self) -> anyhow::Result<RequestRef<B>> {
         Ok(match self.msg_type {
             RequestMsgType::Ping => RequestRef::Ping(self.buf.ping_request()?),
+            RequestMsgType::SupplyKeypair => {
+                RequestRef::SupplyKeypair(self.buf.supply_keypair_request()?)
+            }
+            RequestMsgType::AddListenSocket => {
+                RequestRef::AddListenSocket(self.buf.add_listen_socket_request()?)
+            }
+            RequestMsgType::AddPskBroker => {
+                RequestRef::AddPskBroker(self.buf.add_psk_broker_request()?)
+            }
         })
     }
 
@@ -82,6 +112,9 @@ impl<B: ByteSlice> RequestRefMaker<B> {
 
 pub enum RequestRef<B> {
     Ping(Ref<B, PingRequest>),
+    SupplyKeypair(Ref<B, super::SupplyKeypairRequest>),
+    AddListenSocket(Ref<B, super::AddListenSocketRequest>),
+    AddPskBroker(Ref<B, super::AddPskBrokerRequest>),
 }
 
 impl<B> RequestRef<B>
@@ -91,6 +124,9 @@ where
     pub fn bytes(&self) -> &[u8] {
         match self {
             Self::Ping(r) => r.bytes(),
+            Self::SupplyKeypair(r) => r.bytes(),
+            Self::AddListenSocket(r) => r.bytes(),
+            Self::AddPskBroker(r) => r.bytes(),
         }
     }
 }
@@ -102,6 +138,9 @@ where
     pub fn bytes_mut(&mut self) -> &[u8] {
         match self {
             Self::Ping(r) => r.bytes_mut(),
+            Self::SupplyKeypair(r) => r.bytes_mut(),
+            Self::AddListenSocket(r) => r.bytes_mut(),
+            Self::AddPskBroker(r) => r.bytes_mut(),
         }
     }
 }

rosenpass/src/api/boilerplate/request_response.rs

@@ -42,10 +42,49 @@ impl ResponseMsg for PingResponse {
     type RequestMsg = PingRequest;
 }
 
+impl RequestMsg for super::SupplyKeypairRequest {
+    type ResponseMsg = super::SupplyKeypairResponse;
+}
+
+impl ResponseMsg for super::SupplyKeypairResponse {
+    type RequestMsg = super::SupplyKeypairRequest;
+}
+
+impl RequestMsg for super::AddListenSocketRequest {
+    type ResponseMsg = super::AddListenSocketResponse;
+}
+
+impl ResponseMsg for super::AddListenSocketResponse {
+    type RequestMsg = super::AddListenSocketRequest;
+}
+
+impl RequestMsg for super::AddPskBrokerRequest {
+    type ResponseMsg = super::AddPskBrokerResponse;
+}
+
+impl ResponseMsg for super::AddPskBrokerResponse {
+    type RequestMsg = super::AddPskBrokerRequest;
+}
+
 pub type PingPair<B1, B2> = (Ref<B1, PingRequest>, Ref<B2, PingResponse>);
+pub type SupplyKeypairPair<B1, B2> = (
+    Ref<B1, super::SupplyKeypairRequest>,
+    Ref<B2, super::SupplyKeypairResponse>,
+);
+pub type AddListenSocketPair<B1, B2> = (
+    Ref<B1, super::AddListenSocketRequest>,
+    Ref<B2, super::AddListenSocketResponse>,
+);
+pub type AddPskBrokerPair<B1, B2> = (
+    Ref<B1, super::AddPskBrokerRequest>,
+    Ref<B2, super::AddPskBrokerResponse>,
+);
 
 pub enum RequestResponsePair<B1, B2> {
     Ping(PingPair<B1, B2>),
+    SupplyKeypair(SupplyKeypairPair<B1, B2>),
+    AddListenSocket(AddListenSocketPair<B1, B2>),
+    AddPskBroker(AddPskBrokerPair<B1, B2>),
 }
 
 impl<B1, B2> From<PingPair<B1, B2>> for RequestResponsePair<B1, B2> {
@@ -54,6 +93,24 @@ impl<B1, B2> From<PingPair<B1, B2>> for RequestResponsePair<B1, B2> {
     }
 }
 
+impl<B1, B2> From<SupplyKeypairPair<B1, B2>> for RequestResponsePair<B1, B2> {
+    fn from(v: SupplyKeypairPair<B1, B2>) -> Self {
+        RequestResponsePair::SupplyKeypair(v)
+    }
+}
+
+impl<B1, B2> From<AddListenSocketPair<B1, B2>> for RequestResponsePair<B1, B2> {
+    fn from(v: AddListenSocketPair<B1, B2>) -> Self {
+        RequestResponsePair::AddListenSocket(v)
+    }
+}
+
+impl<B1, B2> From<AddPskBrokerPair<B1, B2>> for RequestResponsePair<B1, B2> {
+    fn from(v: AddPskBrokerPair<B1, B2>) -> Self {
+        RequestResponsePair::AddPskBroker(v)
+    }
+}
+
 impl<B1, B2> RequestResponsePair<B1, B2>
 where
     B1: ByteSlice,
@@ -66,6 +123,21 @@ where
                 let res = ResponseRef::Ping(res.emancipate());
                 (req, res)
             }
+            Self::SupplyKeypair((req, res)) => {
+                let req = RequestRef::SupplyKeypair(req.emancipate());
+                let res = ResponseRef::SupplyKeypair(res.emancipate());
+                (req, res)
+            }
+            Self::AddListenSocket((req, res)) => {
+                let req = RequestRef::AddListenSocket(req.emancipate());
+                let res = ResponseRef::AddListenSocket(res.emancipate());
+                (req, res)
+            }
+            Self::AddPskBroker((req, res)) => {
+                let req = RequestRef::AddPskBroker(req.emancipate());
+                let res = ResponseRef::AddPskBroker(res.emancipate());
+                (req, res)
+            }
         }
     }
 
@@ -90,6 +162,21 @@ where
                 let res = ResponseRef::Ping(res.emancipate_mut());
                 (req, res)
             }
+            Self::SupplyKeypair((req, res)) => {
+                let req = RequestRef::SupplyKeypair(req.emancipate_mut());
+                let res = ResponseRef::SupplyKeypair(res.emancipate_mut());
+                (req, res)
+            }
+            Self::AddListenSocket((req, res)) => {
+                let req = RequestRef::AddListenSocket(req.emancipate_mut());
+                let res = ResponseRef::AddListenSocket(res.emancipate_mut());
+                (req, res)
+            }
+            Self::AddPskBroker((req, res)) => {
+                let req = RequestRef::AddPskBroker(req.emancipate_mut());
+                let res = ResponseRef::AddPskBroker(res.emancipate_mut());
+                (req, res)
+            }
         }
     }
 

rosenpass/src/api/boilerplate/response_ref.rs

@@ -26,6 +26,9 @@ impl<B: ByteSlice> ResponseRef<B> {
     pub fn message_type(&self) -> ResponseMsgType {
         match self {
             Self::Ping(_) => ResponseMsgType::Ping,
+            Self::SupplyKeypair(_) => ResponseMsgType::SupplyKeypair,
+            Self::AddListenSocket(_) => ResponseMsgType::AddListenSocket,
+            Self::AddPskBroker(_) => ResponseMsgType::AddPskBroker,
         }
     }
 }
@@ -36,6 +39,24 @@ impl<B> From<Ref<B, PingResponse>> for ResponseRef<B> {
     }
 }
 
+impl<B> From<Ref<B, super::SupplyKeypairResponse>> for ResponseRef<B> {
+    fn from(v: Ref<B, super::SupplyKeypairResponse>) -> Self {
+        Self::SupplyKeypair(v)
+    }
+}
+
+impl<B> From<Ref<B, super::AddListenSocketResponse>> for ResponseRef<B> {
+    fn from(v: Ref<B, super::AddListenSocketResponse>) -> Self {
+        Self::AddListenSocket(v)
+    }
+}
+
+impl<B> From<Ref<B, super::AddPskBrokerResponse>> for ResponseRef<B> {
+    fn from(v: Ref<B, super::AddPskBrokerResponse>) -> Self {
+        Self::AddPskBroker(v)
+    }
+}
+
 impl<B: ByteSlice> ResponseRefMaker<B> {
     fn new(buf: B) -> anyhow::Result<Self> {
         let msg_type = buf.deref().response_msg_type_from_prefix()?;
@@ -49,6 +70,15 @@ impl<B: ByteSlice> ResponseRefMaker<B> {
     fn parse(self) -> anyhow::Result<ResponseRef<B>> {
         Ok(match self.msg_type {
             ResponseMsgType::Ping => ResponseRef::Ping(self.buf.ping_response()?),
+            ResponseMsgType::SupplyKeypair => {
+                ResponseRef::SupplyKeypair(self.buf.supply_keypair_response()?)
+            }
+            ResponseMsgType::AddListenSocket => {
+                ResponseRef::AddListenSocket(self.buf.add_listen_socket_response()?)
+            }
+            ResponseMsgType::AddPskBroker => {
+                ResponseRef::AddPskBroker(self.buf.add_psk_broker_response()?)
+            }
         })
     }
 
@@ -83,6 +113,9 @@ impl<B: ByteSlice> ResponseRefMaker<B> {
 
 pub enum ResponseRef<B> {
     Ping(Ref<B, PingResponse>),
+    SupplyKeypair(Ref<B, super::SupplyKeypairResponse>),
+    AddListenSocket(Ref<B, super::AddListenSocketResponse>),
+    AddPskBroker(Ref<B, super::AddPskBrokerResponse>),
 }
 
 impl<B> ResponseRef<B>
@@ -92,6 +125,9 @@ where
     pub fn bytes(&self) -> &[u8] {
         match self {
             Self::Ping(r) => r.bytes(),
+            Self::SupplyKeypair(r) => r.bytes(),
+            Self::AddListenSocket(r) => r.bytes(),
+            Self::AddPskBroker(r) => r.bytes(),
         }
     }
 }
@@ -103,6 +139,9 @@ where
     pub fn bytes_mut(&mut self) -> &[u8] {
         match self {
             Self::Ping(r) => r.bytes_mut(),
+            Self::SupplyKeypair(r) => r.bytes_mut(),
+            Self::AddListenSocket(r) => r.bytes_mut(),
+            Self::AddPskBroker(r) => r.bytes_mut(),
         }
     }
 }

rosenpass/src/api/boilerplate/server.rs

@@ -1,24 +1,63 @@
+use super::{ByteSliceRefExt, Message, PingRequest, PingResponse, RequestRef, RequestResponsePair};
+use std::{collections::VecDeque, os::fd::OwnedFd};
 use zerocopy::{ByteSlice, ByteSliceMut};
 
-use super::{ByteSliceRefExt, Message, PingRequest, PingResponse, RequestRef, RequestResponsePair};
-
 pub trait Server {
-    fn ping(&mut self, req: &PingRequest, res: &mut PingResponse) -> anyhow::Result<()>;
+    fn ping(
+        &mut self,
+        req: &PingRequest,
+        req_fds: &mut VecDeque<OwnedFd>,
+        res: &mut PingResponse,
+    ) -> anyhow::Result<()>;
+
+    fn supply_keypair(
+        &mut self,
+        req: &super::SupplyKeypairRequest,
+        req_fds: &mut VecDeque<OwnedFd>,
+        res: &mut super::SupplyKeypairResponse,
+    ) -> anyhow::Result<()>;
+
+    fn add_listen_socket(
+        &mut self,
+        req: &super::AddListenSocketRequest,
+        req_fds: &mut VecDeque<OwnedFd>,
+        res: &mut super::AddListenSocketResponse,
+    ) -> anyhow::Result<()>;
+
+    fn add_psk_broker(
+        &mut self,
+        req: &super::AddPskBrokerRequest,
+        req_fds: &mut VecDeque<OwnedFd>,
+        res: &mut super::AddPskBrokerResponse,
+    ) -> anyhow::Result<()>;
 
     fn dispatch<ReqBuf, ResBuf>(
         &mut self,
         p: &mut RequestResponsePair<ReqBuf, ResBuf>,
+        req_fds: &mut VecDeque<OwnedFd>,
     ) -> anyhow::Result<()>
     where
         ReqBuf: ByteSlice,
         ResBuf: ByteSliceMut,
     {
         match p {
-            RequestResponsePair::Ping((req, res)) => self.ping(req, res),
+            RequestResponsePair::Ping((req, res)) => self.ping(req, req_fds, res),
+            RequestResponsePair::SupplyKeypair((req, res)) => {
+                self.supply_keypair(req, req_fds, res)
+            }
+            RequestResponsePair::AddListenSocket((req, res)) => {
+                self.add_listen_socket(req, req_fds, res)
+            }
+            RequestResponsePair::AddPskBroker((req, res)) => self.add_psk_broker(req, req_fds, res),
         }
     }
 
-    fn handle_message<ReqBuf, ResBuf>(&mut self, req: ReqBuf, res: ResBuf) -> anyhow::Result<usize>
+    fn handle_message<ReqBuf, ResBuf>(
+        &mut self,
+        req: ReqBuf,
+        req_fds: &mut VecDeque<OwnedFd>,
+        res: ResBuf,
+    ) -> anyhow::Result<usize>
     where
         ReqBuf: ByteSlice,
         ResBuf: ByteSliceMut,
@@ -31,10 +70,25 @@ pub trait Server {
                 res.init();
                 RequestResponsePair::Ping((req, res))
             }
+            RequestRef::SupplyKeypair(req) => {
+                let mut res = res.supply_keypair_response_from_prefix()?;
+                res.init();
+                RequestResponsePair::SupplyKeypair((req, res))
+            }
+            RequestRef::AddListenSocket(req) => {
+                let mut res = res.add_listen_socket_response_from_prefix()?;
+                res.init();
+                RequestResponsePair::AddListenSocket((req, res))
+            }
+            RequestRef::AddPskBroker(req) => {
+                let mut res = res.add_psk_broker_response_from_prefix()?;
+                res.init();
+                RequestResponsePair::AddPskBroker((req, res))
+            }
         };
-        self.dispatch(&mut pair)?;
+        self.dispatch(&mut pair, req_fds)?;
 
-        let res_len = pair.request().bytes().len();
+        let res_len = pair.response().bytes().len();
         Ok(res_len)
     }
 }

rosenpass/src/api/config.rs

@@ -38,4 +38,12 @@ impl ApiConfig {
 
         Ok(())
     }
+
+    pub fn count_api_sources(&self) -> usize {
+        self.listen_path.len() + self.listen_fd.len() + self.stream_fd.len()
+    }
+
+    pub fn has_api_sources(&self) -> bool {
+        self.count_api_sources() > 0
+    }
 }

rosenpass/src/api/crypto_server_api_handler.rs

@@ -1,44 +0,0 @@
-use rosenpass_to::{ops::copy_slice, To};
-
-use crate::protocol::CryptoServer;
-
-use super::Server as ApiServer;
-
-#[derive(Debug)]
-pub struct CryptoServerApiState {
-    _dummy: (),
-}
-
-impl CryptoServerApiState {
-    #[allow(clippy::new_without_default)]
-    pub fn new() -> Self {
-        Self { _dummy: () }
-    }
-
-    pub fn acquire_backend<'a>(
-        &'a mut self,
-        crypto: &'a mut Option<CryptoServer>,
-    ) -> CryptoServerApiHandler<'a> {
-        let state = self;
-        CryptoServerApiHandler { state, crypto }
-    }
-}
-
-pub struct CryptoServerApiHandler<'a> {
-    #[allow(unused)] // TODO: Remove
-    crypto: &'a mut Option<CryptoServer>,
-    #[allow(unused)] // TODO: Remove
-    state: &'a mut CryptoServerApiState,
-}
-
-impl<'a> ApiServer for CryptoServerApiHandler<'a> {
-    fn ping(
-        &mut self,
-        req: &super::PingRequest,
-        res: &mut super::PingResponse,
-    ) -> anyhow::Result<()> {
-        let (req, res) = (&req.payload, &mut res.payload);
-        copy_slice(&req.echo).to(&mut res.echo);
-        Ok(())
-    }
-}

rosenpass/src/api/mio/connection.rs

@@ -1,103 +1,188 @@
-use mio::{net::UnixStream, Interest};
+use std::borrow::{Borrow, BorrowMut};
+use std::collections::VecDeque;
+use std::os::fd::OwnedFd;
+
+use mio::net::UnixStream;
+use rosenpass_secret_memory::Secret;
+use rosenpass_util::mio::ReadWithFileDescriptors;
 use rosenpass_util::{
     io::{IoResultKindHintExt, TryIoResultKindHintExt},
     length_prefix_encoding::{
         decoder::{self as lpe_decoder, LengthPrefixDecoder},
         encoder::{self as lpe_encoder, LengthPrefixEncoder},
     },
+    mio::interest::RW as MIO_RW,
 };
 use zeroize::Zeroize;
 
-use crate::{api::Server, app_server::MioTokenDispenser, protocol::CryptoServer};
+use crate::api::MAX_REQUEST_FDS;
+use crate::{api::Server, app_server::AppServer};
 
-use super::super::{CryptoServerApiState, MAX_REQUEST_LEN, MAX_RESPONSE_LEN};
+use super::super::{ApiHandler, ApiHandlerContext};
+
+#[derive(Debug)]
+struct SecretBuffer<const N: usize>(pub Secret<N>);
+
+impl<const N: usize> SecretBuffer<N> {
+    fn new() -> Self {
+        Self(Secret::zero())
+    }
+}
+
+impl<const N: usize> Borrow<[u8]> for SecretBuffer<N> {
+    fn borrow(&self) -> &[u8] {
+        self.0.secret()
+    }
+}
+
+impl<const N: usize> BorrowMut<[u8]> for SecretBuffer<N> {
+    fn borrow_mut(&mut self) -> &mut [u8] {
+        self.0.secret_mut()
+    }
+}
+// TODO: Unfortunately, zerocopy is quite particular about alignment, hence the 4096
+type ReadBuffer = LengthPrefixDecoder<SecretBuffer<4096>>;
+type WriteBuffer = LengthPrefixEncoder<SecretBuffer<4096>>;
+type ReadFdBuffer = VecDeque<OwnedFd>;
+
+#[derive(Debug)]
+struct MioConnectionBuffers {
+    read_buffer: ReadBuffer,
+    write_buffer: WriteBuffer,
+    read_fd_buffer: ReadFdBuffer,
+}
 
 #[derive(Debug)]
 pub struct MioConnection {
     io: UnixStream,
+    mio_token: mio::Token,
     invalid_read: bool,
-    read_buffer: LengthPrefixDecoder<[u8; MAX_REQUEST_LEN]>,
-    write_buffer: LengthPrefixEncoder<[u8; MAX_RESPONSE_LEN]>,
-    api_state: CryptoServerApiState,
+    buffers: Option<MioConnectionBuffers>,
+    api_handler: ApiHandler,
 }
 
 impl MioConnection {
-    pub fn new(
-        mut io: UnixStream,
-        registry: &mio::Registry,
-        token_dispenser: &mut MioTokenDispenser, // TODO: We should actually start using tokens…
-    ) -> std::io::Result<Self> {
-        registry.register(
-            &mut io,
-            token_dispenser.dispense(),
-            Interest::READABLE | Interest::WRITABLE,
-        )?;
+    pub fn new(app_server: &mut AppServer, mut io: UnixStream) -> std::io::Result<Self> {
+        let mio_token = app_server.mio_token_dispenser.dispense();
+        app_server
+            .mio_poll
+            .registry()
+            .register(&mut io, mio_token, MIO_RW)?;
 
         let invalid_read = false;
-        let read_buffer = LengthPrefixDecoder::new([0u8; MAX_REQUEST_LEN]);
-        let write_buffer = LengthPrefixEncoder::from_buffer([0u8; MAX_RESPONSE_LEN]);
-        let api_state = CryptoServerApiState::new();
-        Ok(Self {
-            io,
-            invalid_read,
+        let read_buffer = LengthPrefixDecoder::new(SecretBuffer::new());
+        let write_buffer = LengthPrefixEncoder::from_buffer(SecretBuffer::new());
+        let read_fd_buffer = VecDeque::new();
+        let buffers = Some(MioConnectionBuffers {
             read_buffer,
             write_buffer,
-            api_state,
+            read_fd_buffer,
+        });
+        let api_state = ApiHandler::new();
+        Ok(Self {
+            io,
+            mio_token,
+            invalid_read,
+            buffers,
+            api_handler: api_state,
         })
     }
 
-    pub fn poll(&mut self, crypto: &mut Option<CryptoServer>) -> anyhow::Result<()> {
-        self.flush_write_buffer()?;
-        if self.write_buffer.exhausted() {
-            self.recv(crypto)?;
-        }
+    pub fn shoud_close(&self) -> bool {
+        let exhausted = self
+            .buffers
+            .as_ref()
+            .map(|b| b.write_buffer.exhausted())
+            .unwrap_or(false);
+        self.invalid_read && exhausted
+    }
+
+    pub fn close(mut self, app_server: &mut AppServer) -> anyhow::Result<()> {
+        app_server.mio_poll.registry().deregister(&mut self.io)?;
         Ok(())
     }
 
-    // This is *exclusively* called by recv if the read_buffer holds a message
-    fn handle_incoming_message(&mut self, crypto: &mut Option<CryptoServer>) -> anyhow::Result<()> {
-        // Unwrap is allowed because recv() confirms before the call that a message was
-        // received
-        let req = self.read_buffer.message().unwrap().unwrap();
+    pub fn mio_token(&self) -> mio::Token {
+        self.mio_token
+    }
+}
 
-        // TODO: The API should not return anyhow::Result
-        let response_len = self
-            .api_state
-            .acquire_backend(crypto)
-            .handle_message(req, self.write_buffer.buffer_bytes_mut())?;
-        self.read_buffer.zeroize(); // clear for new message to read
-        self.write_buffer
-            .restart_write_with_new_message(response_len)?;
+pub trait MioConnectionContext {
+    fn mio_connection(&self) -> &MioConnection;
+    fn app_server(&self) -> &AppServer;
+    fn mio_connection_mut(&mut self) -> &mut MioConnection;
+    fn app_server_mut(&mut self) -> &mut AppServer;
+
+    fn poll(&mut self) -> anyhow::Result<()> {
+        macro_rules! short {
+            ($e:expr) => {
+                match $e {
+                    None => return Ok(()),
+                    Some(()) => {}
+                }
+            };
+        }
+
+        // All of these functions return an error, None ("operation incomplete")
+        // or some ("operation complete, keep processing")
+        short!(self.flush_write_buffer()?); // Flush last message
+        short!(self.recv()?); // Receive new message
+        short!(self.handle_incoming_message()?); // Process new message with API
+        short!(self.flush_write_buffer()?); // Begin flushing response
 
-        self.flush_write_buffer()?;
         Ok(())
     }
 
-    fn flush_write_buffer(&mut self) -> anyhow::Result<()> {
-        if self.write_buffer.exhausted() {
-            return Ok(());
+    fn handle_incoming_message(&mut self) -> anyhow::Result<Option<()>> {
+        self.with_buffers_stolen(|this, bufs| {
+            // Acquire request & response. Caller is responsible to make sure
+            // that read buffer holds a message and that write buffer is cleared.
+            // Hence the unwraps and assertions
+            assert!(bufs.write_buffer.exhausted());
+            let req = bufs.read_buffer.message().unwrap().unwrap();
+            let req_fds = &mut bufs.read_fd_buffer;
+            let res = bufs.write_buffer.buffer_bytes_mut();
+
+            // Call API handler
+            // Transitive trait implementations: MioConnectionContext -> ApiHandlerContext -> as ApiServer
+            let response_len = this.handle_message(req, req_fds, res)?;
+
+            bufs.write_buffer
+                .restart_write_with_new_message(response_len)?;
+            bufs.read_buffer.zeroize(); // clear for new message to read
+            bufs.read_fd_buffer.clear();
+
+            Ok(Some(()))
+        })
+    }
+
+    fn flush_write_buffer(&mut self) -> anyhow::Result<Option<()>> {
+        if self.write_buf_mut().exhausted() {
+            return Ok(Some(()));
         }
 
+        use lpe_encoder::WriteToIoReturn as Ret;
+        use std::io::ErrorKind as K;
+
         loop {
-            use lpe_encoder::WriteToIoReturn as Ret;
-            use std::io::ErrorKind as K;
+            let conn = self.mio_connection_mut();
+            let bufs = conn.buffers.as_mut().unwrap();
 
-            match self
-                .write_buffer
-                .write_to_stdio(&self.io)
-                .io_err_kind_hint()
-            {
+            let sock = &conn.io;
+            let write_buf = &mut bufs.write_buffer;
+
+            match write_buf.write_to_stdio(sock).io_err_kind_hint() {
                 // Done
                 Ok(Ret { done: true, .. }) => {
-                    self.write_buffer.zeroize(); // clear for new message to write
-                    break;
+                    write_buf.zeroize(); // clear for new message to write
+                    break Ok(Some(()));
                 }
 
                 // Would block
                 Ok(Ret {
                     bytes_written: 0, ..
-                }) => break,
-                Err((_e, K::WouldBlock)) => break,
+                }) => break Ok(None),
+                Err((_e, K::WouldBlock)) => break Ok(None),
 
                 // Just continue
                 Ok(_) => continue, /* Ret { bytes_written > 0, done = false } acc. to previous cases*/
@@ -107,22 +192,31 @@ impl MioConnection {
                 Err((e, _ek)) => Err(e)?,
             }
         }
-
-        Ok(())
     }
 
-    fn recv(&mut self, crypto: &mut Option<CryptoServer>) -> anyhow::Result<()> {
-        if !self.write_buffer.exhausted() || self.invalid_read {
-            return Ok(());
+    fn recv(&mut self) -> anyhow::Result<Option<()>> {
+        if !self.write_buf_mut().exhausted() || self.mio_connection().invalid_read {
+            return Ok(None);
         }
 
-        loop {
-            use lpe_decoder::{ReadFromIoError as E, ReadFromIoReturn as Ret};
-            use std::io::ErrorKind as K;
+        use lpe_decoder::{ReadFromIoError as E, ReadFromIoReturn as Ret};
+        use std::io::ErrorKind as K;
 
-            match self
-                .read_buffer
-                .read_from_stdio(&self.io)
+        loop {
+            let conn = self.mio_connection_mut();
+            let bufs = conn.buffers.as_mut().unwrap();
+
+            let read_buf = &mut bufs.read_buffer;
+            let read_fd_buf = &mut bufs.read_fd_buffer;
+
+            let sock = &conn.io;
+            let fd_passing_sock = ReadWithFileDescriptors::<MAX_REQUEST_FDS, UnixStream, _, _>::new(
+                sock,
+                read_fd_buf,
+            );
+
+            match read_buf
+                .read_from_stdio(fd_passing_sock)
                 .try_io_err_kind_hint()
             {
                 // We actually received a proper message
@@ -130,38 +224,98 @@ impl MioConnection {
                 Ok(Ret {
                     message: Some(_msg),
                     ..
-                }) => {}
+                }) => break Ok(Some(())),
 
                 // Message does not fit in buffer
                 Err((e @ E::MessageTooLargeError { .. }, _)) => {
                     log::warn!("Received message on API that was too big to fit in our buffers; \
-                        looks like the client is broken. Stopping to process messages of the client.\n\
-                        Error: {e:?}");
-                    // TODO: We should properly close down the socket in this case, but to do that,
-                    // we need to have the facilities in the Rosenpass IO handling system to close
-                    // open connections.
-                    // Just leaving the API connections dangling for now.
-                    // This should be fixed for non-experimental use of the API.
-                    self.invalid_read = true;
-                    break;
+                            looks like the client is broken. Stopping to process messages of the client.\n\
+                            Error: {e:?}");
+                    conn.invalid_read = true; // Closed mio_manager
+                    break Ok(None);
                 }
 
                 // Would block
-                Ok(Ret { bytes_read: 0, .. }) => break,
-                Err((_, Some(K::WouldBlock))) => break,
+                Ok(Ret { bytes_read: 0, .. }) => break Ok(None),
+                Err((_, Some(K::WouldBlock))) => break Ok(None),
 
                 // Just keep going
                 Ok(Ret { bytes_read: _, .. }) => continue,
                 Err((_, Some(K::Interrupted))) => continue,
 
                 // Other IO Error (just pass on to the caller)
-                Err((E::IoError(e), _)) => Err(e)?,
+                Err((E::IoError(e), _)) => {
+                    log::warn!(
+                        "IO error while trying to read message from API socket. \
+                            The connection is broken. Stopping to process messages of the client.\n\
+                            Error: {e:?}"
+                    );
+                    conn.invalid_read = true; // closed later by mio_manager
+                    break Err(e.into());
+                }
             };
-
-            self.handle_incoming_message(crypto)?;
-            break; // Handle just one message, leave some room for other IO handlers
         }
+    }
 
-        Ok(())
+    fn mio_token(&self) -> mio::Token {
+        self.mio_connection().mio_token()
+    }
+
+    fn should_close(&self) -> bool {
+        self.mio_connection().shoud_close()
+    }
+}
+
+trait MioConnectionContextPrivate: MioConnectionContext {
+    fn steal_buffers(&mut self) -> MioConnectionBuffers {
+        self.mio_connection_mut().buffers.take().unwrap()
+    }
+
+    fn return_buffers(&mut self, buffers: MioConnectionBuffers) {
+        let opt = &mut self.mio_connection_mut().buffers;
+        assert!(opt.is_none());
+        let _ = opt.insert(buffers);
+    }
+
+    fn with_buffers_stolen<R, F: FnOnce(&mut Self, &mut MioConnectionBuffers) -> R>(
+        &mut self,
+        f: F,
+    ) -> R {
+        let mut bufs = self.steal_buffers();
+        let res = f(self, &mut bufs);
+        self.return_buffers(bufs);
+        res
+    }
+
+    fn write_buf_mut(&mut self) -> &mut WriteBuffer {
+        self.mio_connection_mut()
+            .buffers
+            .as_mut()
+            .unwrap()
+            .write_buffer
+            .borrow_mut()
+    }
+}
+
+impl<T> MioConnectionContextPrivate for T where T: ?Sized + MioConnectionContext {}
+
+impl<T> ApiHandlerContext for T
+where
+    T: ?Sized + MioConnectionContext,
+{
+    fn api_handler(&self) -> &ApiHandler {
+        &self.mio_connection().api_handler
+    }
+
+    fn app_server(&self) -> &AppServer {
+        MioConnectionContext::app_server(self)
+    }
+
+    fn api_handler_mut(&mut self) -> &mut ApiHandler {
+        &mut self.mio_connection_mut().api_handler
+    }
+
+    fn app_server_mut(&mut self) -> &mut AppServer {
+        MioConnectionContext::app_server_mut(self)
     }
 }

rosenpass/src/api/mio/manager.rs

@@ -1,82 +1,120 @@
-use std::io;
+use std::{borrow::BorrowMut, io};
 
 use mio::net::{UnixListener, UnixStream};
 
-use rosenpass_util::{io::nonblocking_handle_io_errors, mio::interest::RW as MIO_RW};
+use rosenpass_util::{
+    functional::ApplyExt, io::nonblocking_handle_io_errors, mio::interest::RW as MIO_RW,
+};
 
-use crate::{app_server::MioTokenDispenser, protocol::CryptoServer};
+use crate::app_server::{AppServer, AppServerIoSource};
 
-use super::MioConnection;
+use super::{MioConnection, MioConnectionContext};
 
 #[derive(Default, Debug)]
 pub struct MioManager {
     listeners: Vec<UnixListener>,
-    connections: Vec<MioConnection>,
+    connections: Vec<Option<MioConnection>>,
+}
+
+#[derive(Debug, PartialEq, Eq, Copy, Clone)]
+pub enum MioManagerIoSource {
+    Listener(usize),
+    Connection(usize),
 }
 
 impl MioManager {
     pub fn new() -> Self {
         Self::default()
     }
+}
+
+struct MioConnectionFocus<'a, T: ?Sized + MioManagerContext> {
+    ctx: &'a mut T,
+    conn_idx: usize,
+}
+
+impl<'a, T: ?Sized + MioManagerContext> MioConnectionFocus<'a, T> {
+    fn new(ctx: &'a mut T, conn_idx: usize) -> Self {
+        Self { ctx, conn_idx }
+    }
+}
+
+pub trait MioManagerContext {
+    fn mio_manager(&self) -> &MioManager;
+    fn mio_manager_mut(&mut self) -> &mut MioManager;
+    fn app_server(&self) -> &AppServer;
+    fn app_server_mut(&mut self) -> &mut AppServer;
+
+    fn add_listener(&mut self, mut listener: UnixListener) -> io::Result<()> {
+        let srv = self.app_server_mut();
+        let mio_token = srv.mio_token_dispenser.dispense();
+        srv.mio_poll
+            .registry()
+            .register(&mut listener, mio_token, MIO_RW)?;
+        let io_source = self
+            .mio_manager()
+            .listeners
+            .len()
+            .apply(MioManagerIoSource::Listener)
+            .apply(AppServerIoSource::MioManager);
+        self.mio_manager_mut().listeners.push(listener);
+        self.app_server_mut()
+            .register_io_source(mio_token, io_source);
 
-    pub fn add_listener(
-        &mut self,
-        mut listener: UnixListener,
-        registry: &mio::Registry,
-        token_dispenser: &mut MioTokenDispenser,
-    ) -> io::Result<()> {
-        registry.register(&mut listener, token_dispenser.dispense(), MIO_RW)?;
-        self.listeners.push(listener);
         Ok(())
     }
 
-    pub fn add_connection(
-        &mut self,
-        connection: UnixStream,
-        registry: &mio::Registry,
-        token_dispenser: &mut MioTokenDispenser,
-    ) -> io::Result<()> {
-        let connection = MioConnection::new(connection, registry, token_dispenser)?;
-        self.connections.push(connection);
+    fn add_connection(&mut self, connection: UnixStream) -> io::Result<()> {
+        let connection = MioConnection::new(self.app_server_mut(), connection)?;
+        let mio_token = connection.mio_token();
+        let conns: &mut Vec<Option<MioConnection>> =
+            self.mio_manager_mut().connections.borrow_mut();
+        let idx = conns
+            .iter_mut()
+            .enumerate()
+            .find(|(_, slot)| slot.is_some())
+            .map(|(idx, _)| idx)
+            .unwrap_or(conns.len());
+        conns.insert(idx, Some(connection));
+        let io_source = idx
+            .apply(MioManagerIoSource::Listener)
+            .apply(AppServerIoSource::MioManager);
+        self.app_server_mut()
+            .register_io_source(mio_token, io_source);
         Ok(())
     }
 
-    pub fn poll(
-        &mut self,
-        crypto: &mut Option<CryptoServer>,
-        registry: &mio::Registry,
-        token_dispenser: &mut MioTokenDispenser,
-    ) -> anyhow::Result<()> {
-        self.accept_connections(registry, token_dispenser)?;
-        self.poll_connections(crypto)?;
+    fn poll_particular(&mut self, io_source: MioManagerIoSource) -> anyhow::Result<()> {
+        use MioManagerIoSource as S;
+        match io_source {
+            S::Listener(idx) => self.accept_from(idx)?,
+            S::Connection(idx) => self.poll_particular_connection(idx)?,
+        };
         Ok(())
     }
 
-    fn accept_connections(
-        &mut self,
-        registry: &mio::Registry,
-        token_dispenser: &mut MioTokenDispenser,
-    ) -> io::Result<()> {
-        for idx in 0..self.listeners.len() {
-            self.accept_from(idx, registry, token_dispenser)?;
+    fn poll(&mut self) -> anyhow::Result<()> {
+        self.accept_connections()?;
+        self.poll_connections()?;
+        Ok(())
+    }
+
+    fn accept_connections(&mut self) -> io::Result<()> {
+        for idx in 0..self.mio_manager_mut().listeners.len() {
+            self.accept_from(idx)?;
         }
         Ok(())
     }
 
-    fn accept_from(
-        &mut self,
-        idx: usize,
-        registry: &mio::Registry,
-        token_dispenser: &mut MioTokenDispenser,
-    ) -> io::Result<()> {
+    fn accept_from(&mut self, idx: usize) -> io::Result<()> {
         // Accept connection until the socket would block or returns another error
         // TODO: This currently only adds connections--we eventually need the ability to remove
         // them as well, see the note in connection.rs
         loop {
-            match nonblocking_handle_io_errors(|| self.listeners[idx].accept())? {
+            match nonblocking_handle_io_errors(|| self.mio_manager().listeners[idx].accept())? {
                 None => break,
                 Some((conn, _addr)) => {
-                    self.add_connection(conn, registry, token_dispenser)?;
+                    self.add_connection(conn)?;
                 }
             };
         }
@@ -84,10 +122,52 @@ impl MioManager {
         Ok(())
     }
 
-    fn poll_connections(&mut self, crypto: &mut Option<CryptoServer>) -> anyhow::Result<()> {
-        for conn in self.connections.iter_mut() {
-            conn.poll(crypto)?
+    fn poll_connections(&mut self) -> anyhow::Result<()> {
+        for idx in 0..self.mio_manager().connections.len() {
+            self.poll_particular_connection(idx)?;
         }
         Ok(())
     }
+
+    fn poll_particular_connection(&mut self, idx: usize) -> anyhow::Result<()> {
+        if self.mio_manager().connections[idx].is_none() {
+            return Ok(());
+        }
+
+        let mut conn = MioConnectionFocus::new(self, idx);
+        conn.poll()?;
+
+        if conn.should_close() {
+            let conn = self.mio_manager_mut().connections[idx].take().unwrap();
+            let mio_token = conn.mio_token();
+            if let Err(e) = conn.close(self.app_server_mut()) {
+                log::warn!("Error while closing API connection {e:?}");
+            };
+            self.app_server_mut().unregister_io_source(mio_token);
+        }
+
+        Ok(())
+    }
+}
+
+impl<T: ?Sized + MioManagerContext> MioConnectionContext for MioConnectionFocus<'_, T> {
+    fn mio_connection(&self) -> &MioConnection {
+        self.ctx.mio_manager().connections[self.conn_idx]
+            .as_ref()
+            .unwrap()
+    }
+
+    fn app_server(&self) -> &AppServer {
+        self.ctx.app_server()
+    }
+
+    fn mio_connection_mut(&mut self) -> &mut MioConnection {
+        self.ctx.mio_manager_mut().connections[self.conn_idx]
+            .as_mut()
+            .unwrap()
+    }
+
+    fn app_server_mut(&mut self) -> &mut AppServer {
+        self.ctx.app_server_mut()
+    }
 }

rosenpass/src/api/mod.rs

@@ -1,8 +1,8 @@
+mod api_handler;
 mod boilerplate;
-mod crypto_server_api_handler;
 
+pub use api_handler::*;
 pub use boilerplate::*;
-pub use crypto_server_api_handler::*;
 
 pub mod cli;
 pub mod config;

rosenpass/src/app_server.rs

@@ -8,7 +8,14 @@ use mio::Interest;
 use mio::Token;
 use rosenpass_secret_memory::Public;
 use rosenpass_secret_memory::Secret;
+use rosenpass_util::build::ConstructionSite;
 use rosenpass_util::file::StoreValueB64;
+use rosenpass_util::functional::run;
+use rosenpass_util::functional::ApplyExt;
+use rosenpass_util::io::IoResultKindHintExt;
+use rosenpass_util::io::SubstituteForIoErrorKindExt;
+use rosenpass_util::option::SomeExt;
+use rosenpass_util::result::OkExt;
 use rosenpass_wireguard_broker::WireguardBrokerMio;
 use rosenpass_wireguard_broker::{WireguardBrokerCfg, WG_KEY_LEN};
 use zerocopy::AsBytes;
@@ -16,7 +23,9 @@ use zerocopy::AsBytes;
 use std::cell::Cell;
 
 use std::collections::HashMap;
+use std::collections::VecDeque;
 use std::fmt::Debug;
+use std::io;
 use std::io::stdout;
 use std::io::ErrorKind;
 use std::io::Write;
@@ -31,6 +40,7 @@ use std::slice;
 use std::time::Duration;
 use std::time::Instant;
 
+use crate::protocol::BuildCryptoServer;
 use crate::protocol::HostIdentification;
 use crate::{
     config::Verbosity,
@@ -75,7 +85,7 @@ impl MioTokenDispenser {
 
 #[derive(Debug, Default)]
 pub struct BrokerStore {
-    store: HashMap<
+    pub store: HashMap<
         Public<BROKER_ID_BYTES>,
         Box<dyn WireguardBrokerMio<Error = anyhow::Error, MioError = anyhow::Error>>,
     >,
@@ -141,15 +151,29 @@ pub struct AppServerTest {
     pub termination_handler: Option<std::sync::mpsc::Receiver<()>>,
 }
 
+#[derive(Debug, PartialEq, Eq, Copy, Clone)]
+pub enum AppServerIoSource {
+    Socket(usize),
+    #[cfg(feature = "experiment_api")]
+    PskBroker(Public<BROKER_ID_BYTES>),
+    #[cfg(feature = "experiment_api")]
+    MioManager(crate::api::mio::MioManagerIoSource),
+}
+
+const EVENT_CAPACITY: usize = 20;
+
 /// Holds the state of the application, namely the external IO
 ///
 /// Responsible for file IO, network IO
 // TODO add user control via unix domain socket and stdin/stdout
 #[derive(Debug)]
 pub struct AppServer {
-    pub crypt: Option<CryptoServer>,
+    pub crypto_site: ConstructionSite<BuildCryptoServer, CryptoServer>,
     pub sockets: Vec<mio::net::UdpSocket>,
     pub events: mio::Events,
+    pub short_poll_queue: VecDeque<mio::event::Event>,
+    pub performed_long_poll: bool,
+    pub io_source_index: HashMap<mio::Token, AppServerIoSource>,
     pub mio_poll: mio::Poll,
     pub mio_token_dispenser: MioTokenDispenser,
     pub brokers: BrokerStore,
@@ -512,15 +536,14 @@ impl HostPathDiscoveryEndpoint {
 
 impl AppServer {
     pub fn new(
-        sk: SSk,
-        pk: SPk,
+        keypair: Option<(SSk, SPk)>,
         addrs: Vec<SocketAddr>,
         verbosity: Verbosity,
         test_helpers: Option<AppServerTest>,
     ) -> anyhow::Result<Self> {
         // setup mio
         let mio_poll = mio::Poll::new()?;
-        let events = mio::Events::with_capacity(20);
+        let events = mio::Events::with_capacity(EVENT_CAPACITY);
         let mut mio_token_dispenser = MioTokenDispenser::default();
 
         // bind each SocketAddr to a socket
@@ -595,22 +618,30 @@ impl AppServer {
         }
 
         // register all sockets to mio
-        for socket in sockets.iter_mut() {
-            mio_poll.registry().register(
-                socket,
-                mio_token_dispenser.dispense(),
-                Interest::READABLE,
-            )?;
+        let mut io_source_index = HashMap::new();
+        for (idx, socket) in sockets.iter_mut().enumerate() {
+            let mio_token = mio_token_dispenser.dispense();
+            mio_poll
+                .registry()
+                .register(socket, mio_token, Interest::READABLE)?;
+            let prev = io_source_index.insert(mio_token, AppServerIoSource::Socket(idx));
+            assert!(prev.is_none());
         }
 
-        // TODO use mio::net::UnixStream together with std::os::unix::net::UnixStream for Linux
+        let crypto_site = match keypair {
+            Some((sk, pk)) => ConstructionSite::from_product(CryptoServer::new(sk, pk)),
+            None => ConstructionSite::new(BuildCryptoServer::empty()),
+        };
 
         Ok(Self {
-            crypt: Some(CryptoServer::new(sk, pk)),
+            crypto_site,
             peers: Vec::new(),
             verbosity,
             sockets,
             events,
+            short_poll_queue: Default::default(),
+            performed_long_poll: false,
+            io_source_index,
             mio_poll,
             mio_token_dispenser,
             brokers: BrokerStore::default(),
@@ -627,14 +658,14 @@ impl AppServer {
     }
 
     pub fn crypto_server(&self) -> anyhow::Result<&CryptoServer> {
-        self.crypt
-            .as_ref()
+        self.crypto_site
+            .product_ref()
             .context("Cryptography handler not initialized")
     }
 
     pub fn crypto_server_mut(&mut self) -> anyhow::Result<&mut CryptoServer> {
-        self.crypt
-            .as_mut()
+        self.crypto_site
+            .product_mut()
             .context("Cryptography handler not initialized")
     }
 
@@ -642,41 +673,57 @@ impl AppServer {
         matches!(self.verbosity, Verbosity::Verbose)
     }
 
+    pub fn register_listen_socket(&mut self, mut sock: mio::net::UdpSocket) -> anyhow::Result<()> {
+        let mio_token = self.mio_token_dispenser.dispense();
+        self.mio_poll
+            .registry()
+            .register(&mut sock, mio_token, mio::Interest::READABLE)?;
+        let io_source = self.sockets.len().apply(AppServerIoSource::Socket);
+        self.sockets.push(sock);
+        self.register_io_source(mio_token, io_source);
+        Ok(())
+    }
+
+    pub fn register_io_source(&mut self, token: mio::Token, io_source: AppServerIoSource) {
+        let prev = self.io_source_index.insert(token, io_source);
+        assert!(prev.is_none());
+    }
+
+    pub fn unregister_io_source(&mut self, token: mio::Token) {
+        let value = self.io_source_index.remove(&token);
+        assert!(value.is_some(), "Removed IO source that does not exist");
+    }
+
     pub fn register_broker(
         &mut self,
         broker: Box<dyn WireguardBrokerMio<Error = anyhow::Error, MioError = anyhow::Error>>,
     ) -> Result<BrokerStorePtr> {
         let ptr = Public::from_slice((self.brokers.store.len() as u64).as_bytes());
-
         if self.brokers.store.insert(ptr, broker).is_some() {
             bail!("Broker already registered");
         }
+
+        let mio_token = self.mio_token_dispenser.dispense();
+        let io_source = ptr.apply(AppServerIoSource::PskBroker);
         //Register broker
         self.brokers
             .store
             .get_mut(&ptr)
             .ok_or(anyhow::format_err!("Broker wasn't added to registry"))?
-            .register(
-                self.mio_poll.registry(),
-                self.mio_token_dispenser.dispense(),
-            )?;
+            .register(self.mio_poll.registry(), mio_token)?;
+        self.register_io_source(mio_token, io_source);
 
         Ok(BrokerStorePtr(ptr))
     }
 
     pub fn unregister_broker(&mut self, ptr: BrokerStorePtr) -> Result<()> {
-        //Unregister broker
-        self.brokers
-            .store
-            .get_mut(&ptr.0)
-            .ok_or_else(|| anyhow::anyhow!("Broker not found"))?
-            .unregister(self.mio_poll.registry())?;
-
-        //Remove broker from store
-        self.brokers
+        let mut broker = self
+            .brokers
             .store
             .remove(&ptr.0)
-            .ok_or_else(|| anyhow::anyhow!("Broker not found"))?;
+            .context("Broker not found")?;
+        self.unregister_io_source(broker.mio_token().unwrap());
+        broker.unregister(self.mio_poll.registry())?;
         Ok(())
     }
 
@@ -688,8 +735,13 @@ impl AppServer {
         broker_peer: Option<BrokerPeer>,
         hostname: Option<String>,
     ) -> anyhow::Result<AppPeerPtr> {
-        let PeerPtr(pn) = self.crypto_server_mut()?.add_peer(psk, pk)?;
+        let PeerPtr(pn) = match &mut self.crypto_site {
+            ConstructionSite::Void => bail!("Crypto server construction site is void"),
+            ConstructionSite::Builder(builder) => builder.add_peer(psk, pk),
+            ConstructionSite::Product(srv) => srv.add_peer(psk, pk)?,
+        };
         assert!(pn == self.peers.len());
+
         let initial_endpoint = hostname
             .map(Endpoint::discovery_from_hostname)
             .transpose()?;
@@ -731,7 +783,7 @@ impl AppServer {
             );
             if tries_left > 0 {
                 error!("re-initializing networking in {sleep}! {tries_left} tries left.");
-                std::thread::sleep(self.crypto_server_mut()?.timebase.dur(sleep));
+                std::thread::sleep(Duration::from_secs_f64(sleep));
                 continue;
             }
 
@@ -774,16 +826,31 @@ impl AppServer {
                 }
             }
 
-            match self.poll(&mut *rx)? {
-                #[allow(clippy::redundant_closure_call)]
-                SendInitiation(peer) => tx_maybe_with!(peer, || self
+            enum CryptoSrv {
+                Avail,
+                Missing,
+            }
+
+            let poll_result = self.poll(&mut *rx)?;
+            let have_crypto = match self.crypto_site.is_available() {
+                true => CryptoSrv::Avail,
+                false => CryptoSrv::Missing,
+            };
+
+            #[allow(clippy::redundant_closure_call)]
+            match (have_crypto, poll_result) {
+                (CryptoSrv::Missing, SendInitiation(_)) => {}
+                (CryptoSrv::Avail, SendInitiation(peer)) => tx_maybe_with!(peer, || self
                     .crypto_server_mut()?
                     .initiate_handshake(peer.lower(), &mut *tx))?,
-                #[allow(clippy::redundant_closure_call)]
-                SendRetransmission(peer) => tx_maybe_with!(peer, || self
+
+                (CryptoSrv::Missing, SendRetransmission(_)) => {}
+                (CryptoSrv::Avail, SendRetransmission(peer)) => tx_maybe_with!(peer, || self
                     .crypto_server_mut()?
                     .retransmit_handshake(peer.lower(), &mut *tx))?,
-                DeleteKey(peer) => {
+
+                (CryptoSrv::Missing, DeleteKey(_)) => {}
+                (CryptoSrv::Avail, DeleteKey(peer)) => {
                     self.output_key(peer, Stale, &SymKey::random())?;
 
                     // There was a loss of connection apparently; restart host discovery
@@ -797,7 +864,8 @@ impl AppServer {
                     );
                 }
 
-                ReceivedMessage(len, endpoint) => {
+                (CryptoSrv::Missing, ReceivedMessage(_, _)) => {}
+                (CryptoSrv::Avail, ReceivedMessage(len, endpoint)) => {
                     let msg_result = match self.under_load {
                         DoSOperation::UnderLoad => {
                             self.handle_msg_under_load(&endpoint, &rx[..len], &mut *tx)
@@ -910,17 +978,32 @@ impl AppServer {
     pub fn poll(&mut self, rx_buf: &mut [u8]) -> anyhow::Result<AppPollResult> {
         use crate::protocol::PollResult as C;
         use AppPollResult as A;
-        loop {
-            return Ok(match self.crypto_server_mut()?.poll()? {
-                C::DeleteKey(PeerPtr(no)) => A::DeleteKey(AppPeerPtr(no)),
-                C::SendInitiation(PeerPtr(no)) => A::SendInitiation(AppPeerPtr(no)),
-                C::SendRetransmission(PeerPtr(no)) => A::SendRetransmission(AppPeerPtr(no)),
-                C::Sleep(timeout) => match self.try_recv(rx_buf, timeout)? {
-                    Some((len, addr)) => A::ReceivedMessage(len, addr),
-                    None => continue,
-                },
-            });
-        }
+        let res = loop {
+            // Call CryptoServer's poll (if available)
+            let crypto_poll = self
+                .crypto_site
+                .product_mut()
+                .map(|crypto| crypto.poll())
+                .transpose()?;
+
+            // Map crypto server's poll result to our poll result
+            let io_poll_timeout = match crypto_poll {
+                Some(C::DeleteKey(PeerPtr(no))) => break A::DeleteKey(AppPeerPtr(no)),
+                Some(C::SendInitiation(PeerPtr(no))) => break A::SendInitiation(AppPeerPtr(no)),
+                Some(C::SendRetransmission(PeerPtr(no))) => {
+                    break A::SendRetransmission(AppPeerPtr(no))
+                }
+                Some(C::Sleep(timeout)) => timeout, // No event from crypto-server, do IO
+                None => crate::protocol::UNENDING,  // Crypto server is uninitialized, do IO
+            };
+
+            // Perform IO (look for a message)
+            if let Some((len, addr)) = self.try_recv(rx_buf, io_poll_timeout)? {
+                break A::ReceivedMessage(len, addr);
+            }
+        };
+
+        Ok(res)
     }
 
     /// Tries to receive a new message
@@ -958,22 +1041,33 @@ impl AppServer {
         // readiness event seems to be good enough™ for now.
 
         // only poll if we drained all sockets before
-        if self.all_sockets_drained {
-            //Non blocked polling
-            self.mio_poll
-                .poll(&mut self.events, Some(Duration::from_secs(0)))?;
-
-            if self.events.iter().peekable().peek().is_none() {
-                // if there are no events, then add to blocking poll count
-                self.blocking_polls_count += 1;
-                //Execute blocking poll
-                self.mio_poll.poll(&mut self.events, Some(timeout))?;
-            } else {
-                self.non_blocking_polls_count += 1;
+        run(|| -> anyhow::Result<()> {
+            if !self.all_sockets_drained || !self.short_poll_queue.is_empty() {
+                self.unpolled_count += 1;
+                return Ok(());
             }
-        } else {
-            self.unpolled_count += 1;
-        }
+
+            self.perform_mio_poll_and_register_events(Duration::from_secs(0))?; // Non-blocking poll
+            if !self.short_poll_queue.is_empty() {
+                // Got some events in non-blocking mode
+                self.non_blocking_polls_count += 1;
+                return Ok(());
+            }
+
+            if !self.performed_long_poll {
+                // pass – go perform a full long poll before we enter blocking poll mode
+                // to make sure our experimental short poll feature did not miss any events
+                // due to being buggy.
+                return Ok(());
+            }
+
+            // Perform and register blocking poll
+            self.blocking_polls_count += 1;
+            self.perform_mio_poll_and_register_events(timeout)?;
+            self.performed_long_poll = false;
+
+            Ok(())
+        })?;
 
         if let Some(AppServerTest {
             enable_dos_permanently: true,
@@ -1008,26 +1102,58 @@ impl AppServer {
             }
         }
 
+        // Focused polling – i.e. actually using mio::Token – is experimental for now.
+        // The reason for this is that we need to figure out how to integrate load detection
+        // and focused polling for one. Mio event-based polling also does not play nice with
+        // the current function signature and its reentrant design which is focused around receiving UDP socket packages
+        // for processing by the crypto protocol server.
+        // Besides that, there are also some parts of the code which intentionally block
+        // despite available data. This is the correct behavior; e.g. api::mio::Connection blocks
+        // further reads from its unix socket until the write buffer is flushed. In other words
+        // the connection handler makes sure that there is a buffer to put the response in while
+        // before reading further request.
+        // The potential problem with this behavior is that we end up ignoring instructions from
+        // epoll() to read from the particular sockets, so epoll will return information about that
+        // particular – blocked – file descriptor every call. We have only so many event slots and
+        // in theory, the event array could fill up entirely with intentionally blocked sockets.
+        // We need to figure out how to deal with this situation.
+        // Mio uses uses epoll in level-triggered mode, so we could handle taint-tracking for ignored
+        // sockets ourselves. The facilities are available in epoll and Mio, but we need to figure out how mio uses those
+        // facilities and how we can integrate them here.
+        // This will involve rewriting a lot of IO code and we should probably have integration
+        // tests  before we approach that.
+        //
+        // This hybrid approach is not without merit though; the short poll implementation covers
+        // all our IO sources, so under contention, rosenpass should generally not hit the long
+        // poll mode below. We keep short polling and calling epoll() in non-blocking mode (timeout
+        // of zero) until we run out of IO events processed. Then, just before we would perform a
+        // blocking poll, we go through all available IO sources to see if we missed anything.
+        {
+            while let Some(ev) = self.short_poll_queue.pop_front() {
+                if let Some(v) = self.try_recv_from_mio_token(buf, ev.token())? {
+                    return Ok(Some(v));
+                }
+            }
+        }
+
         // drain all sockets
         let mut would_block_count = 0;
-        for (sock_no, socket) in self.sockets.iter_mut().enumerate() {
-            match socket.recv_from(buf) {
-                Ok((n, addr)) => {
+        for sock_no in 0..self.sockets.len() {
+            match self
+                .try_recv_from_listen_socket(buf, sock_no)
+                .io_err_kind_hint()
+            {
+                Ok(None) => continue,
+                Ok(Some(v)) => {
                     // at least one socket was not drained...
                     self.all_sockets_drained = false;
-                    return Ok(Some((
-                        n,
-                        Endpoint::SocketBoundAddress(SocketBoundEndpoint::new(
-                            SocketPtr(sock_no),
-                            addr,
-                        )),
-                    )));
+                    return Ok(Some(v));
                 }
-                Err(e) if e.kind() == ErrorKind::WouldBlock => {
+                Err((_, ErrorKind::WouldBlock)) => {
                     would_block_count += 1;
                 }
                 // TODO if one socket continuously returns an error, then we never poll, thus we never wait for a timeout, thus we have a spin-lock
-                Err(e) => return Err(e.into()),
+                Err((e, _)) => return Err(e)?,
             }
         }
 
@@ -1042,30 +1168,123 @@ impl AppServer {
         // API poll
 
         #[cfg(feature = "experiment_api")]
-        self.api_manager.poll(
-            &mut self.crypt,
-            self.mio_poll.registry(),
-            &mut self.mio_token_dispenser,
-        )?;
+        {
+            use crate::api::mio::MioManagerContext;
+            MioManagerFocus(self).poll()?;
+        }
+
+        self.performed_long_poll = true;
 
         Ok(None)
     }
 
+    fn perform_mio_poll_and_register_events(&mut self, timeout: Duration) -> io::Result<()> {
+        self.mio_poll.poll(&mut self.events, Some(timeout))?;
+        // Fill the short poll buffer with the acquired events
+        self.events
+            .iter()
+            .cloned()
+            .for_each(|v| self.short_poll_queue.push_back(v));
+        Ok(())
+    }
+
+    fn try_recv_from_mio_token(
+        &mut self,
+        buf: &mut [u8],
+        token: mio::Token,
+    ) -> anyhow::Result<Option<(usize, Endpoint)>> {
+        let io_source = match self.io_source_index.get(&token) {
+            Some(io_source) => *io_source,
+            None => {
+                log::warn!("No IO source assiociated with mio token ({token:?}). Polling using mio tokens directly is an experimental feature and IO handler should recover when all available io sources are polled. This is a developer error. Please report it.");
+                return Ok(None);
+            }
+        };
+
+        self.try_recv_from_io_source(buf, io_source)
+    }
+
+    fn try_recv_from_io_source(
+        &mut self,
+        buf: &mut [u8],
+        io_source: AppServerIoSource,
+    ) -> anyhow::Result<Option<(usize, Endpoint)>> {
+        use crate::api::mio::MioManagerContext;
+
+        match io_source {
+            AppServerIoSource::Socket(idx) => self
+                .try_recv_from_listen_socket(buf, idx)
+                .substitute_for_ioerr_wouldblock(None)?
+                .ok(),
+
+            #[cfg(feature = "experiment_api")]
+            AppServerIoSource::PskBroker(key) => self
+                .brokers
+                .store
+                .get_mut(&key)
+                .with_context(|| format!("No PSK broker under key {key:?}"))?
+                .process_poll()
+                .map(|_| None),
+
+            #[cfg(feature = "experiment_api")]
+            AppServerIoSource::MioManager(mmio_src) => MioManagerFocus(self)
+                .poll_particular(mmio_src)
+                .map(|_| None),
+        }
+    }
+
+    fn try_recv_from_listen_socket(
+        &mut self,
+        buf: &mut [u8],
+        idx: usize,
+    ) -> io::Result<Option<(usize, Endpoint)>> {
+        use std::io::ErrorKind as K;
+        let (n, addr) = loop {
+            match self.sockets[idx].recv_from(buf).io_err_kind_hint() {
+                Ok(v) => break v,
+                Err((_, K::Interrupted)) => continue,
+                Err((e, _)) => return Err(e)?,
+            }
+        };
+        SocketPtr(idx)
+            .apply(|sp| SocketBoundEndpoint::new(sp, addr))
+            .apply(Endpoint::SocketBoundAddress)
+            .apply(|ep| (n, ep))
+            .some()
+            .ok()
+    }
+
     #[cfg(feature = "experiment_api")]
     pub fn add_api_connection(&mut self, connection: mio::net::UnixStream) -> std::io::Result<()> {
-        self.api_manager.add_connection(
-            connection,
-            self.mio_poll.registry(),
-            &mut self.mio_token_dispenser,
-        )
+        use crate::api::mio::MioManagerContext;
+        MioManagerFocus(self).add_connection(connection)
     }
 
     #[cfg(feature = "experiment_api")]
     pub fn add_api_listener(&mut self, listener: mio::net::UnixListener) -> std::io::Result<()> {
-        self.api_manager.add_listener(
-            listener,
-            self.mio_poll.registry(),
-            &mut self.mio_token_dispenser,
-        )
+        use crate::api::mio::MioManagerContext;
+        MioManagerFocus(self).add_listener(listener)
+    }
+}
+
+#[cfg(feature = "experiment_api")]
+struct MioManagerFocus<'a>(&'a mut AppServer);
+
+#[cfg(feature = "experiment_api")]
+impl crate::api::mio::MioManagerContext for MioManagerFocus<'_> {
+    fn mio_manager(&self) -> &crate::api::mio::MioManager {
+        &self.0.api_manager
+    }
+
+    fn mio_manager_mut(&mut self) -> &mut crate::api::mio::MioManager {
+        &mut self.0.api_manager
+    }
+
+    fn app_server(&self) -> &AppServer {
+        self.0
+    }
+
+    fn app_server_mut(&mut self) -> &mut AppServer {
+        self.0
     }
 }

rosenpass/src/bin/gen-ipc-msg-types.rs

@@ -76,6 +76,12 @@ fn main() -> Result<()> {
             vec![
                 Tree::Leaf("Ping Request".to_owned()),
                 Tree::Leaf("Ping Response".to_owned()),
+                Tree::Leaf("Supply Keypair Request".to_owned()),
+                Tree::Leaf("Supply Keypair Response".to_owned()),
+                Tree::Leaf("Add Listen Socket Request".to_owned()),
+                Tree::Leaf("Add Listen Socket Response".to_owned()),
+                Tree::Leaf("Add Psk Broker Request".to_owned()),
+                Tree::Leaf("Add Psk Broker Response".to_owned()),
             ],
         )],
     );

rosenpass/src/cli.rs

@@ -1,4 +1,4 @@
-use anyhow::{bail, ensure};
+use anyhow::{bail, ensure, Context};
 use clap::{Parser, Subcommand};
 use rosenpass_cipher_traits::Kem;
 use rosenpass_ciphers::kem::StaticKem;
@@ -16,6 +16,29 @@ use crate::protocol::{SPk, SSk, SymKey};
 
 use super::config;
 
+#[cfg(feature = "experiment_api")]
+use {
+    command_fds::{CommandFdExt, FdMapping},
+    log::{error, info},
+    mio::net::UnixStream,
+    rosenpass_util::fd::claim_fd,
+    rosenpass_wireguard_broker::brokers::mio_client::MioBrokerClient,
+    rosenpass_wireguard_broker::WireguardBrokerMio,
+    rustix::fd::AsRawFd,
+    rustix::net::{socketpair, AddressFamily, SocketFlags, SocketType},
+    std::os::unix::net,
+    std::process::Command,
+    std::thread,
+};
+
+/// enum representing a choice of interface to a WireGuard broker
+#[derive(Debug)]
+pub enum BrokerInterface {
+    Socket(PathBuf),
+    FileDescriptor(i32),
+    SocketPair,
+}
+
 /// struct holding all CLI arguments for `clap` crate to parse
 #[derive(Parser, Debug)]
 #[command(author, version, about, long_about)]
@@ -36,6 +59,26 @@ pub struct CliArgs {
     #[cfg(feature = "experiment_api")]
     api: crate::api::cli::ApiCli,
 
+    /// path of the wireguard_psk broker socket to connect to
+    #[cfg(feature = "experiment_api")]
+    #[arg(long, group = "psk-broker-specs")]
+    psk_broker_path: Option<PathBuf>,
+
+    /// fd of the wireguard_spk broker socket to connect to
+    ///
+    /// when this command is called from another process, the other process can open and bind the
+    /// Unix socket for the psk broker connection to use themselves, passing it to this process --
+    /// in Rust this can be achieved using the
+    /// [command-fds](https://docs.rs/command-fds/latest/command_fds/) crate
+    #[cfg(feature = "experiment_api")]
+    #[arg(long, group = "psk-broker-specs")]
+    psk_broker_fd: Option<i32>,
+
+    /// spawn a psk broker locally using a socket pair
+    #[cfg(feature = "experiment_api")]
+    #[arg(short, long, group = "psk-broker-specs")]
+    psk_broker_spawn: bool,
+
     #[command(subcommand)]
     pub command: CliCommand,
 }
@@ -58,13 +101,35 @@ impl CliArgs {
             return Some(log::LevelFilter::Info);
         }
         if self.quiet {
-            return Some(log::LevelFilter::Error);
+            return Some(log::LevelFilter::Warn);
         }
         if let Some(level_filter) = self.log_level {
             return Some(level_filter);
         }
         None
     }
+
+    #[cfg(feature = "experiment_api")]
+    /// returns the broker interface set by CLI args
+    /// returns `None` if the `experiment_api` feature isn't enabled
+    pub fn get_broker_interface(&self) -> Option<BrokerInterface> {
+        if let Some(path_ref) = self.psk_broker_path.as_ref() {
+            Some(BrokerInterface::Socket(path_ref.to_path_buf()))
+        } else if let Some(fd) = self.psk_broker_fd {
+            Some(BrokerInterface::FileDescriptor(fd))
+        } else if self.psk_broker_spawn {
+            Some(BrokerInterface::SocketPair)
+        } else {
+            None
+        }
+    }
+
+    #[cfg(not(feature = "experiment_api"))]
+    /// returns the broker interface set by CLI args
+    /// returns `None` if the `experiment_api` feature isn't enabled
+    pub fn get_broker_interface(&self) -> Option<BrokerInterface> {
+        None
+    }
 }
 
 /// represents a command specified via CLI
@@ -164,7 +229,11 @@ impl CliArgs {
     ///
     /// ## TODO
     /// - This method consumes the [`CliCommand`] value. It might be wise to use a reference...
-    pub fn run(self, test_helpers: Option<AppServerTest>) -> anyhow::Result<()> {
+    pub fn run(
+        self,
+        broker_interface: Option<BrokerInterface>,
+        test_helpers: Option<AppServerTest>,
+    ) -> anyhow::Result<()> {
         use CliCommand::*;
         match &self.command {
             Man => {
@@ -234,8 +303,11 @@ impl CliArgs {
                         );
 
                         let config = config::Rosenpass::load(config_file)?;
+                        let keypair = config
+                            .keypair
+                            .context("Config file present, but no keypair is specified.")?;
 
-                        (config.public_key, config.secret_key)
+                        (keypair.public_key, keypair.secret_key)
                     }
                     (_, Some(pkf), Some(skf)) => (pkf.clone(), skf.clone()),
                     _ => {
@@ -247,12 +319,14 @@ impl CliArgs {
                 let mut problems = vec![];
                 if !force && pkf.is_file() {
                     problems.push(format!(
-                        "public-key file {pkf:?} exist, refusing to overwrite it"
+                        "public-key file {:?} exists, refusing to overwrite",
+                        std::fs::canonicalize(&pkf)?,
                     ));
                 }
                 if !force && skf.is_file() {
                     problems.push(format!(
-                        "secret-key file {skf:?} exist, refusing to overwrite it"
+                        "secret-key file {:?} exists, refusing to overwrite",
+                        std::fs::canonicalize(&skf)?,
                     ));
                 }
                 if !problems.is_empty() {
@@ -272,8 +346,9 @@ impl CliArgs {
                 let mut config = config::Rosenpass::load(config_file)?;
                 config.validate()?;
                 self.apply_to_config(&mut config)?;
+                config.check_usefullness()?;
 
-                Self::event_loop(config, test_helpers)?;
+                Self::event_loop(config, broker_interface, test_helpers)?;
             }
 
             Exchange {
@@ -292,8 +367,9 @@ impl CliArgs {
                 }
                 config.validate()?;
                 self.apply_to_config(&mut config)?;
+                config.check_usefullness()?;
 
-                Self::event_loop(config, test_helpers)?;
+                Self::event_loop(config, broker_interface, test_helpers)?;
             }
 
             Validate { config_files } => {
@@ -317,18 +393,25 @@ impl CliArgs {
 
     fn event_loop(
         config: config::Rosenpass,
+        broker_interface: Option<BrokerInterface>,
         test_helpers: Option<AppServerTest>,
     ) -> anyhow::Result<()> {
         const MAX_PSK_SIZE: usize = 1000;
 
         // load own keys
-        let sk = SSk::load(&config.secret_key)?;
-        let pk = SPk::load(&config.public_key)?;
+        let keypair = config
+            .keypair
+            .as_ref()
+            .map(|kp| -> anyhow::Result<_> {
+                let sk = SSk::load(&kp.secret_key)?;
+                let pk = SPk::load(&kp.public_key)?;
+                Ok((sk, pk))
+            })
+            .transpose()?;
 
         // start an application server
         let mut srv = std::boxed::Box::<AppServer>::new(AppServer::new(
-            sk,
-            pk,
+            keypair,
             config.listen.clone(),
             config.verbosity,
             test_helpers,
@@ -336,7 +419,8 @@ impl CliArgs {
 
         config.apply_to_app_server(&mut srv)?;
 
-        let broker_store_ptr = srv.register_broker(Box::new(NativeUnixBroker::new()))?;
+        let broker = Self::create_broker(broker_interface)?;
+        let broker_store_ptr = srv.register_broker(broker)?;
 
         fn cfg_err_map(e: NativeUnixBrokerConfigBaseBuilderError) -> anyhow::Error {
             anyhow::Error::msg(format!("NativeUnixBrokerConfigBaseBuilderError: {:?}", e))
@@ -373,6 +457,83 @@ impl CliArgs {
 
         srv.event_loop()
     }
+
+    #[cfg(feature = "experiment_api")]
+    fn create_broker(
+        broker_interface: Option<BrokerInterface>,
+    ) -> Result<
+        Box<dyn WireguardBrokerMio<MioError = anyhow::Error, Error = anyhow::Error>>,
+        anyhow::Error,
+    > {
+        if let Some(interface) = broker_interface {
+            let socket = Self::get_broker_socket(interface)?;
+            Ok(Box::new(MioBrokerClient::new(socket)))
+        } else {
+            Ok(Box::new(NativeUnixBroker::new()))
+        }
+    }
+
+    #[cfg(not(feature = "experiment_api"))]
+    fn create_broker(
+        _broker_interface: Option<BrokerInterface>,
+    ) -> Result<Box<NativeUnixBroker>, anyhow::Error> {
+        Ok(Box::new(NativeUnixBroker::new()))
+    }
+
+    #[cfg(feature = "experiment_api")]
+    fn get_broker_socket(broker_interface: BrokerInterface) -> Result<UnixStream, anyhow::Error> {
+        // Connect to the psk broker unix socket if one was specified
+        // OR OTHERWISE spawn the psk broker and use socketpair(2) to connect with them
+        match broker_interface {
+            BrokerInterface::Socket(broker_path) => Ok(UnixStream::connect(broker_path)?),
+            BrokerInterface::FileDescriptor(broker_fd) => {
+                // mio::net::UnixStream doesn't implement From<OwnedFd>, so we have to go through std
+                let sock = net::UnixStream::from(claim_fd(broker_fd)?);
+                sock.set_nonblocking(true)?;
+                Ok(UnixStream::from_std(sock))
+            }
+            BrokerInterface::SocketPair => {
+                // Form a socketpair for communicating to the broker
+                let (ours, theirs) = socketpair(
+                    AddressFamily::UNIX,
+                    SocketType::STREAM,
+                    SocketFlags::empty(),
+                    None,
+                )?;
+
+                // Setup our end of the socketpair
+                let ours = net::UnixStream::from(ours);
+                ours.set_nonblocking(true)?;
+
+                // Start the PSK broker
+                let mut child = Command::new("rosenpass-wireguard-broker-socket-handler")
+                    .args(["--stream-fd", "3"])
+                    .fd_mappings(vec![FdMapping {
+                        parent_fd: theirs.as_raw_fd(),
+                        child_fd: 3,
+                    }])?
+                    .spawn()?;
+
+                // Handle the PSK broker crashing
+                thread::spawn(move || {
+                    let status = child.wait();
+
+                    if let Ok(status) = status {
+                        if status.success() {
+                            // Maybe they are doing double forking?
+                            info!("PSK broker exited.");
+                        } else {
+                            error!("PSK broker exited with an error ({status:?})");
+                        }
+                    } else {
+                        error!("Wait on PSK broker process failed ({status:?})");
+                    }
+                });
+
+                Ok(UnixStream::from_std(ours))
+            }
+        }
+    }
 }
 
 /// generate secret and public keys, store in files according to the paths passed as arguments

rosenpass/src/config.rs

@@ -21,16 +21,25 @@ use serde::{Deserialize, Serialize};
 
 use crate::app_server::AppServer;
 
+#[cfg(feature = "experiment_api")]
+fn empty_api_config() -> crate::api::config::ApiConfig {
+    crate::api::config::ApiConfig {
+        listen_path: Vec::new(),
+        listen_fd: Vec::new(),
+        stream_fd: Vec::new(),
+    }
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 pub struct Rosenpass {
-    /// path to the public key file
-    pub public_key: PathBuf,
-
-    /// path to the secret key file
-    pub secret_key: PathBuf,
+    // TODO: Raise error if secret key or public key alone is set during deserialization
+    // SEE: https://github.com/serde-rs/serde/issues/2793
+    #[serde(flatten)]
+    pub keypair: Option<Keypair>,
 
     /// Location of the API listen sockets
     #[cfg(feature = "experiment_api")]
+    #[serde(default = "empty_api_config")]
     pub api: crate::api::config::ApiConfig,
 
     /// list of [`SocketAddr`] to listen on
@@ -58,6 +67,26 @@ pub struct Rosenpass {
     pub config_file_path: PathBuf,
 }
 
+#[derive(Debug, Deserialize, Serialize, PartialEq, Eq, Clone)]
+pub struct Keypair {
+    /// path to the public key file
+    pub public_key: PathBuf,
+
+    /// path to the secret key file
+    pub secret_key: PathBuf,
+}
+
+impl Keypair {
+    pub fn new<Pk: AsRef<Path>, Sk: AsRef<Path>>(public_key: Pk, secret_key: Sk) -> Self {
+        let public_key = public_key.as_ref().to_path_buf();
+        let secret_key = secret_key.as_ref().to_path_buf();
+        Self {
+            public_key,
+            secret_key,
+        }
+    }
+}
+
 /// ## TODO
 /// - replace this type with [`log::LevelFilter`], also see <https://github.com/rosenpass/rosenpass/pull/246>
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Copy, Clone)]
@@ -113,6 +142,12 @@ pub struct WireGuard {
     pub extra_params: Vec<String>,
 }
 
+impl Default for Rosenpass {
+    fn default() -> Self {
+        Self::empty()
+    }
+}
+
 impl Rosenpass {
     /// load configuration from a TOML file
     ///
@@ -128,8 +163,10 @@ impl Rosenpass {
 
         // resolve `~` (see https://github.com/rosenpass/rosenpass/issues/237)
         use util::resolve_path_with_tilde;
-        resolve_path_with_tilde(&mut config.public_key);
-        resolve_path_with_tilde(&mut config.secret_key);
+        if let Some(ref mut keypair) = config.keypair {
+            resolve_path_with_tilde(&mut keypair.public_key);
+            resolve_path_with_tilde(&mut keypair.secret_key);
+        }
         for peer in config.peers.iter_mut() {
             resolve_path_with_tilde(&mut peer.public_key);
             if let Some(ref mut psk) = &mut peer.pre_shared_key {
@@ -175,19 +212,21 @@ impl Rosenpass {
     /// - check that files do not just exist but are also readable
     /// - warn if neither out_key nor exchange_command of a peer is defined (v.i.)
     pub fn validate(&self) -> anyhow::Result<()> {
-        // check the public key file exists
-        ensure!(
-            self.public_key.is_file(),
-            "could not find public-key file {:?}: no such file",
-            self.public_key
-        );
+        if let Some(ref keypair) = self.keypair {
+            // check the public key file exists
+            ensure!(
+                keypair.public_key.is_file(),
+                "could not find public-key file {:?}: no such file",
+                keypair.public_key
+            );
 
-        // check the secret-key file exists
-        ensure!(
-            self.secret_key.is_file(),
-            "could not find secret-key file {:?}: no such file",
-            self.secret_key
-        );
+            // check the secret-key file exists
+            ensure!(
+                keypair.secret_key.is_file(),
+                "could not find secret-key file {:?}: no such file",
+                keypair.secret_key
+            );
+        }
 
         for (i, peer) in self.peers.iter().enumerate() {
             // check peer's public-key file exists
@@ -212,11 +251,33 @@ impl Rosenpass {
         Ok(())
     }
 
+    pub fn check_usefullness(&self) -> anyhow::Result<()> {
+        #[cfg(not(feature = "experiment_api"))]
+        ensure!(self.keypair.is_some(), "Server keypair missing.");
+
+        #[cfg(feature = "experiment_api")]
+        ensure!(
+            self.keypair.is_some() || self.api.has_api_sources(),
+            "{}{}",
+            "Specify a server keypair or some API connections to configure the keypair with.",
+            "Without a keypair, rosenpass can not operate."
+        );
+
+        Ok(())
+    }
+
+    pub fn empty() -> Self {
+        Self::new(None)
+    }
+
+    pub fn from_sk_pk<Sk: AsRef<Path>, Pk: AsRef<Path>>(sk: Sk, pk: Pk) -> Self {
+        Self::new(Some(Keypair::new(pk, sk)))
+    }
+
     /// Creates a new configuration
-    pub fn new<P1: AsRef<Path>, P2: AsRef<Path>>(public_key: P1, secret_key: P2) -> Self {
+    pub fn new(keypair: Option<Keypair>) -> Self {
         Self {
-            public_key: PathBuf::from(public_key.as_ref()),
-            secret_key: PathBuf::from(secret_key.as_ref()),
+            keypair,
             listen: vec![],
             #[cfg(feature = "experiment_api")]
             api: crate::api::config::ApiConfig::default(),
@@ -242,7 +303,7 @@ impl Rosenpass {
     /// from chaotic args
     /// Quest: the grammar is undecideable, what do we do here?
     pub fn parse_args(args: Vec<String>) -> anyhow::Result<Self> {
-        let mut config = Self::new("", "");
+        let mut config = Self::new(Some(Keypair::new("", "")));
 
         #[derive(Debug, Hash, PartialEq, Eq)]
         enum State {
@@ -303,7 +364,7 @@ impl Rosenpass {
                         already_set.insert(OwnPublicKey),
                         "public-key was already set"
                     );
-                    config.public_key = pk.into();
+                    config.keypair.as_mut().unwrap().public_key = pk.into();
                     Own
                 }
                 (OwnSecretKey, sk, None) => {
@@ -311,7 +372,7 @@ impl Rosenpass {
                         already_set.insert(OwnSecretKey),
                         "secret-key was already set"
                     );
-                    config.secret_key = sk.into();
+                    config.keypair.as_mut().unwrap().secret_key = sk.into();
                     Own
                 }
                 (OwnListen, l, None) => {
@@ -446,10 +507,12 @@ impl Rosenpass {
         };
 
         Self {
-            public_key: "/path/to/rp-public-key".into(),
-            secret_key: "/path/to/rp-secret-key".into(),
+            keypair: Some(Keypair {
+                public_key: "/path/to/rp-public-key".into(),
+                secret_key: "/path/to/rp-secret-key".into(),
+            }),
             peers: vec![peer],
-            ..Self::new("", "")
+            ..Self::new(None)
         }
     }
 }
@@ -462,13 +525,119 @@ impl Default for Verbosity {
 
 #[cfg(test)]
 mod test {
+
     use super::*;
-    use std::net::IpAddr;
+    use std::{borrow::Borrow, net::IpAddr};
+
+    fn toml_des<S: Borrow<str>>(s: S) -> Result<toml::Table, toml::de::Error> {
+        toml::from_str(s.borrow())
+    }
+
+    fn toml_ser<S: Serialize>(s: S) -> Result<toml::Table, toml::ser::Error> {
+        toml::Table::try_from(s)
+    }
+
+    fn assert_toml<L: Serialize, R: Borrow<str>>(l: L, r: R, info: &str) -> anyhow::Result<()> {
+        fn lines_prepend(prefix: &str, s: &str) -> anyhow::Result<String> {
+            use std::fmt::Write;
+
+            let mut buf = String::new();
+            for line in s.lines() {
+                writeln!(&mut buf, "{prefix}{line}")?;
+            }
+            Ok(buf)
+        }
+
+        let l = toml_ser(l)?;
+        let r = toml_des(r.borrow())?;
+        ensure!(
+            l == r,
+            "{}{}TOML value mismatch.\n  Have:\n{}\n  Expected:\n{}",
+            info,
+            if info.is_empty() { "" } else { ": " },
+            lines_prepend("    ", &toml::to_string_pretty(&l)?)?,
+            lines_prepend("    ", &toml::to_string_pretty(&r)?)?
+        );
+        Ok(())
+    }
+
+    fn assert_toml_round<'de, L: Serialize + Deserialize<'de>, R: Borrow<str>>(
+        l: L,
+        r: R,
+    ) -> anyhow::Result<()> {
+        let l = toml_ser(l)?;
+        assert_toml(&l, r.borrow(), "Straight deserialization")?;
+
+        let l: L = l.try_into().unwrap();
+        let l = toml_ser(l).unwrap();
+        assert_toml(l, r.borrow(), "Roundtrip deserialization")?;
+
+        Ok(())
+    }
 
     fn split_str(s: &str) -> Vec<String> {
         s.split(' ').map(|s| s.to_string()).collect()
     }
 
+    #[test]
+    fn toml_serialization() -> anyhow::Result<()> {
+        #[cfg(feature = "experiment_api")]
+        assert_toml_round(
+            Rosenpass::empty(),
+            r#"
+            listen = []
+            verbosity = "Quiet"
+            peers = []
+
+            [api]
+            listen_path = []
+            listen_fd = []
+            stream_fd = []
+        "#,
+        )?;
+
+        #[cfg(not(feature = "experiment_api"))]
+        assert_toml_round(
+            Rosenpass::empty(),
+            r#"
+            listen = []
+            verbosity = "Quiet"
+            peers = []
+        "#,
+        )?;
+
+        #[cfg(feature = "experiment_api")]
+        assert_toml_round(
+            Rosenpass::from_sk_pk("/my/sk", "/my/pk"),
+            r#"
+            public_key = "/my/pk"
+            secret_key = "/my/sk"
+            listen = []
+            verbosity = "Quiet"
+            peers = []
+
+            [api]
+            listen_path = []
+            listen_fd = []
+            stream_fd = []
+        "#,
+        )?;
+
+        #[cfg(not(feature = "experiment_api"))]
+        assert_toml_round(
+            Rosenpass::from_sk_pk("/my/sk", "/my/pk"),
+            r#"
+            public_key = "/my/pk"
+            secret_key = "/my/sk"
+            listen = []
+            verbosity = "Quiet"
+            peers = []
+        "#,
+        )?;
+
+        Ok(())
+    }
+
     #[test]
     fn test_simple_cli_parse() {
         let args = split_str(
@@ -479,8 +648,10 @@ mod test {
 
         let config = Rosenpass::parse_args(args).unwrap();
 
-        assert_eq!(config.public_key, PathBuf::from("/my/public-key"));
-        assert_eq!(config.secret_key, PathBuf::from("/my/secret-key"));
+        assert_eq!(
+            config.keypair,
+            Some(Keypair::new("/my/public-key", "/my/secret-key"))
+        );
         assert_eq!(config.verbosity, Verbosity::Verbose);
         assert_eq!(
             &config.listen,
@@ -509,8 +680,10 @@ mod test {
 
         let config = Rosenpass::parse_args(args).unwrap();
 
-        assert_eq!(config.public_key, PathBuf::from("/my/public-key"));
-        assert_eq!(config.secret_key, PathBuf::from("/my/secret-key"));
+        assert_eq!(
+            config.keypair,
+            Some(Keypair::new("/my/public-key", "/my/secret-key"))
+        );
         assert_eq!(config.verbosity, Verbosity::Verbose);
         assert!(&config.listen.is_empty());
         assert_eq!(

rosenpass/src/main.rs

@@ -34,7 +34,8 @@ pub fn main() {
         // error!("error dummy");
     }
 
-    match args.run(None) {
+    let broker_interface = args.get_broker_interface();
+    match args.run(broker_interface, None) {
         Ok(_) => {}
         Err(e) => {
             error!("{e:?}");

rosenpass/src/protocol/build_crypto_server.rs

@@ -0,0 +1,127 @@
+use rosenpass_util::{
+    build::Build,
+    mem::{DiscardResultExt, SwapWithDefaultExt},
+    result::ensure_or,
+};
+use thiserror::Error;
+
+use super::{CryptoServer, PeerPtr, SPk, SSk, SymKey};
+
+#[derive(Debug, Clone)]
+pub struct Keypair {
+    pub sk: SSk,
+    pub pk: SPk,
+}
+
+// TODO: We need a named tuple derive
+impl Keypair {
+    pub fn new(sk: SSk, pk: SPk) -> Self {
+        Self { sk, pk }
+    }
+
+    pub fn zero() -> Self {
+        Self::new(SSk::zero(), SPk::zero())
+    }
+
+    pub fn random() -> Self {
+        Self::new(SSk::random(), SPk::random())
+    }
+
+    pub fn from_parts(parts: (SSk, SPk)) -> Self {
+        Self::new(parts.0, parts.1)
+    }
+
+    pub fn into_parts(self) -> (SSk, SPk) {
+        (self.sk, self.pk)
+    }
+}
+
+#[derive(Error, Debug)]
+#[error("PSK already set in BuildCryptoServer")]
+pub struct PskAlreadySet;
+
+#[derive(Error, Debug)]
+#[error("Keypair already set in BuildCryptoServer")]
+pub struct KeypairAlreadySet;
+
+#[derive(Error, Debug)]
+#[error("Can not construct CryptoServer: Missing keypair")]
+pub struct MissingKeypair;
+
+#[derive(Debug, Default)]
+pub struct BuildCryptoServer {
+    pub keypair: Option<Keypair>,
+    pub peers: Vec<PeerParams>,
+}
+
+impl Build<CryptoServer> for BuildCryptoServer {
+    type Error = anyhow::Error;
+
+    fn build(self) -> Result<CryptoServer, Self::Error> {
+        let Some(Keypair { sk, pk }) = self.keypair else {
+            return Err(MissingKeypair)?;
+        };
+
+        let mut srv = CryptoServer::new(sk, pk);
+
+        for (idx, PeerParams { psk, pk }) in self.peers.into_iter().enumerate() {
+            let PeerPtr(idx2) = srv.add_peer(psk, pk)?;
+            assert!(idx == idx2, "Peer id changed during CryptoServer construction from {idx} to {idx2}. This is a developer error.")
+        }
+
+        Ok(srv)
+    }
+}
+
+#[derive(Debug)]
+pub struct PeerParams {
+    pub psk: Option<SymKey>,
+    pub pk: SPk,
+}
+
+impl BuildCryptoServer {
+    pub fn new(keypair: Option<Keypair>, peers: Vec<PeerParams>) -> Self {
+        Self { keypair, peers }
+    }
+
+    pub fn empty() -> Self {
+        Self::new(None, Vec::new())
+    }
+
+    pub fn from_parts(parts: (Option<Keypair>, Vec<PeerParams>)) -> Self {
+        Self {
+            keypair: parts.0,
+            peers: parts.1,
+        }
+    }
+
+    pub fn take_parts(&mut self) -> (Option<Keypair>, Vec<PeerParams>) {
+        (self.keypair.take(), self.peers.swap_with_default())
+    }
+
+    pub fn into_parts(mut self) -> (Option<Keypair>, Vec<PeerParams>) {
+        self.take_parts()
+    }
+
+    pub fn with_keypair(&mut self, keypair: Keypair) -> Result<&mut Self, KeypairAlreadySet> {
+        ensure_or(self.keypair.is_none(), KeypairAlreadySet)?;
+        self.keypair.insert(keypair).discard_result();
+        Ok(self)
+    }
+
+    pub fn with_added_peer(&mut self, psk: Option<SymKey>, pk: SPk) -> &mut Self {
+        // TODO: Check here already whether peer was already added
+        self.peers.push(PeerParams { psk, pk });
+        self
+    }
+
+    pub fn add_peer(&mut self, psk: Option<SymKey>, pk: SPk) -> PeerPtr {
+        let id = PeerPtr(self.peers.len());
+        self.with_added_peer(psk, pk);
+        id
+    }
+
+    pub fn emancipate(&mut self) -> Self {
+        Self::from_parts(self.take_parts())
+    }
+}

rosenpass/src/protocol/mod.rs

@@ -0,0 +1,6 @@
+mod build_crypto_server;
+#[allow(clippy::module_inception)]
+mod protocol;
+
+pub use build_crypto_server::*;
+pub use protocol::*;

rosenpass/src/protocol/protocol.rs

@@ -91,19 +91,13 @@ use rosenpass_ciphers::kem::{EphemeralKem, StaticKem};
 use rosenpass_ciphers::{aead, xaead, KEY_LEN};
 use rosenpass_constant_time as constant_time;
 use rosenpass_secret_memory::{Public, PublicBox, Secret};
-use rosenpass_util::{cat, mem::cpy_min, ord::max_usize, time::Timebase};
+use rosenpass_util::{cat, mem::cpy_min, time::Timebase};
 use zerocopy::{AsBytes, FromBytes, Ref};
 
 use crate::{hash_domains, msgs::*, RosenpassError};
 
 // CONSTANTS & SETTINGS //////////////////////////
 
-/// Size required to fit any message in binary form
-pub const RTX_BUFFER_SIZE: usize = max_usize(
-    size_of::<Envelope<InitHello>>(),
-    size_of::<Envelope<InitConf>>(),
-);
-
 /// A type for time, e.g. for backoff before re-tries
 pub type Timing = f64;
 
@@ -140,11 +134,10 @@ pub const PEER_COOKIE_VALUE_EPOCH: Timing = 120.0;
 // decryption for a second epoch
 pub const BISCUIT_EPOCH: Timing = 300.0;
 
-// Retransmission pub constants; will retransmit for up to _ABORT ms; starting with a delay of
-// _DELAY_BEG ms and increasing the delay exponentially by a factor of
-// _DELAY_GROWTH up to _DELAY_END. An additional jitter factor of ±_DELAY_JITTER
-// is added.
-pub const RETRANSMIT_ABORT: Timing = 120.0;
+// Retransmission pub constants; will retransmit for up to _ABORT seconds;
+// starting with a delay of _DELAY_BEGIN seconds and increasing the delay
+// exponentially by a factor of _DELAY_GROWTH up to _DELAY_END.
+// An additional jitter factor of ±_DELAY_JITTER is added.
 pub const RETRANSMIT_DELAY_GROWTH: Timing = 2.0;
 pub const RETRANSMIT_DELAY_BEGIN: Timing = 0.5;
 pub const RETRANSMIT_DELAY_END: Timing = 10.0;
@@ -1479,7 +1472,7 @@ impl IniHsPtr {
                         .min(ih.tx_count as f64),
                 )
                 * RETRANSMIT_DELAY_JITTER
-                * (rand::random::<f64>() + 1.0); // TODO: Replace with the rand crate
+                * (rand::random::<f64>() + 1.0);
         ih.tx_count += 1;
         Ok(())
     }
@@ -2016,8 +2009,7 @@ impl CryptoServer {
 
         // Send ack – Implementing sending the empty acknowledgement here
         // instead of a generic PeerPtr::send(&Server, Option<&[u8]>) -> Either<EmptyData, Data>
-        // because data transmission is a stub currently. This software is supposed to be used
-        // as a key exchange service feeding a PSK into some classical (i.e. non post quantum)
+        // because data transmission is a stub currently.
         let ses = peer
             .session()
             .get_mut(self)

rosenpass/tests/api-integration-tests-api-setup.rs

@@ -0,0 +1,332 @@
+use std::{
+    borrow::Borrow,
+    io::{BufRead, BufReader, Write},
+    os::unix::net::UnixStream,
+    process::Stdio,
+    thread::sleep,
+    time::Duration,
+};
+
+use anyhow::{bail, Context};
+use command_fds::{CommandFdExt, FdMapping};
+use hex_literal::hex;
+use rosenpass::api::{
+    self, add_listen_socket_response_status, add_psk_broker_response_status,
+    supply_keypair_response_status,
+};
+use rosenpass_util::{
+    b64::B64Display,
+    file::LoadValueB64,
+    io::IoErrorKind,
+    length_prefix_encoding::{decoder::LengthPrefixDecoder, encoder::LengthPrefixEncoder},
+    mem::{DiscardResultExt, MoveExt},
+    mio::WriteWithFileDescriptors,
+    zerocopy::ZerocopySliceExt,
+};
+use rustix::fd::{AsFd, AsRawFd};
+use tempfile::TempDir;
+use zerocopy::AsBytes;
+
+use rosenpass::protocol::SymKey;
+
+struct KillChild(std::process::Child);
+
+impl Drop for KillChild {
+    fn drop(&mut self) {
+        self.0.kill().discard_result();
+        self.0.wait().discard_result()
+    }
+}
+
+#[test]
+fn api_integration_api_setup() -> anyhow::Result<()> {
+    rosenpass_secret_memory::policy::secret_policy_use_only_malloc_secrets();
+
+    let dir = TempDir::with_prefix("rosenpass-api-integration-test")?;
+
+    macro_rules! tempfile {
+        ($($lst:expr),+) => {{
+            let mut buf =  dir.path().to_path_buf();
+            $(buf.push($lst);)*
+            buf
+        }}
+    }
+
+    let peer_a_endpoint = "[::1]:0";
+    let peer_a_listen = std::net::UdpSocket::bind(peer_a_endpoint)?;
+    let peer_a_endpoint = format!("{}", peer_a_listen.local_addr()?);
+    let peer_a_keypair = config::Keypair::new(tempfile!("a.pk"), tempfile!("a.sk"));
+
+    let peer_b_osk = tempfile!("b.osk");
+    let peer_b_wg_device = "mock_device";
+    let peer_b_wg_peer_id = hex!(
+        "
+        93 0f ee 77 0c 6b 54 7e  13 5f 13 92 21 97 26 53
+        7d 77 4a 6a 0f 6c eb 1a  dd 6e 5b c4 1b 92 cd 99
+    "
+    );
+
+    use rosenpass::config;
+    let peer_a = config::Rosenpass {
+        config_file_path: tempfile!("a.config"),
+        keypair: None,
+        listen: vec![], // TODO: This could collide by accident
+        verbosity: config::Verbosity::Verbose,
+        api: api::config::ApiConfig {
+            listen_path: vec![tempfile!("a.sock")],
+            listen_fd: vec![],
+            stream_fd: vec![],
+        },
+        peers: vec![config::RosenpassPeer {
+            public_key: tempfile!("b.pk"),
+            key_out: None,
+            endpoint: None,
+            pre_shared_key: None,
+            wg: Some(config::WireGuard {
+                device: peer_b_wg_device.to_string(),
+                peer: format!("{}", peer_b_wg_peer_id.fmt_b64::<8129>()),
+                extra_params: vec![],
+            }),
+        }],
+    };
+
+    let peer_b_keypair = config::Keypair::new(tempfile!("b.pk"), tempfile!("b.sk"));
+    let peer_b = config::Rosenpass {
+        config_file_path: tempfile!("b.config"),
+        keypair: Some(peer_b_keypair.clone()),
+        listen: vec![],
+        verbosity: config::Verbosity::Verbose,
+        api: api::config::ApiConfig {
+            listen_path: vec![tempfile!("b.sock")],
+            listen_fd: vec![],
+            stream_fd: vec![],
+        },
+        peers: vec![config::RosenpassPeer {
+            public_key: tempfile!("a.pk"),
+            key_out: Some(peer_b_osk.clone()),
+            endpoint: Some(peer_a_endpoint.to_owned()),
+            pre_shared_key: None,
+            wg: None,
+        }],
+    };
+
+    // Generate the keys
+    rosenpass::cli::testing::generate_and_save_keypair(
+        peer_a_keypair.secret_key.clone(),
+        peer_a_keypair.public_key.clone(),
+    )?;
+    rosenpass::cli::testing::generate_and_save_keypair(
+        peer_b_keypair.secret_key.clone(),
+        peer_b_keypair.public_key.clone(),
+    )?;
+
+    // Write the configuration files
+    peer_a.commit()?;
+    peer_b.commit()?;
+
+    let (deliberate_fail_api_client, deliberate_fail_api_server) =
+        std::os::unix::net::UnixStream::pair()?;
+    let deliberate_fail_child_fd = 3;
+
+    // Start peer a
+    let _proc_a = KillChild(
+        std::process::Command::new(env!("CARGO_BIN_EXE_rosenpass"))
+            .args(["--api-stream-fd", &deliberate_fail_child_fd.to_string()])
+            .fd_mappings(vec![FdMapping {
+                parent_fd: deliberate_fail_api_server.move_here().as_raw_fd(),
+                child_fd: 3,
+            }])?
+            .args([
+                "exchange-config",
+                peer_a.config_file_path.to_str().context("")?,
+            ])
+            .stdin(Stdio::null())
+            .stdout(Stdio::null())
+            .spawn()?,
+    );
+
+    // Start peer b
+    let mut proc_b = KillChild(
+        std::process::Command::new(env!("CARGO_BIN_EXE_rosenpass"))
+            .args([
+                "exchange-config",
+                peer_b.config_file_path.to_str().context("")?,
+            ])
+            .stdin(Stdio::null())
+            .stderr(Stdio::null())
+            .stdout(Stdio::piped())
+            .spawn()?,
+    );
+
+    // Acquire stdout
+    let mut out_b = BufReader::new(proc_b.0.stdout.take().context("")?).lines();
+
+    // Now connect to the peers
+    let api_path = peer_a.api.listen_path[0].as_path();
+
+    // Wait for the socket to be created
+    let attempt = 0;
+    while !api_path.exists() {
+        sleep(Duration::from_millis(200));
+        assert!(
+            attempt < 50,
+            "Api failed to be created even after 50 seconds"
+        );
+    }
+
+    let api = UnixStream::connect(api_path)?;
+    let (psk_broker_sock, psk_broker_server_sock) = UnixStream::pair()?;
+
+    // Send AddListenSocket request
+    {
+        let fd = peer_a_listen.as_fd();
+
+        let mut fds = vec![&fd].into();
+        let mut api = WriteWithFileDescriptors::<UnixStream, _, _, _>::new(&api, &mut fds);
+        LengthPrefixEncoder::from_message(api::AddListenSocketRequest::new().as_bytes())
+            .write_all_to_stdio(&mut api)?;
+        assert!(fds.is_empty(), "Failed to write all file descriptors");
+        std::mem::forget(peer_a_listen);
+    }
+
+    // Read response
+    {
+        let mut decoder = LengthPrefixDecoder::new([0u8; api::MAX_RESPONSE_LEN]);
+        let res = decoder.read_all_from_stdio(&api)?;
+        let res = res.zk_parse::<api::AddListenSocketResponse>()?;
+        assert_eq!(
+            *res,
+            api::AddListenSocketResponse::new(add_listen_socket_response_status::OK)
+        );
+    }
+
+    // Deliberately break API connection given via FD; this checks that the
+    // API connections are closed when invalid data is received and it also
+    // implicitly checks that other connections are unaffected
+    {
+        use std::io::ErrorKind as K;
+        let client = deliberate_fail_api_client;
+        let err = loop {
+            if let Err(e) = client.borrow().write(&[0xffu8; 16]) {
+                break e;
+            }
+        };
+        // NotConnected happens on Mac
+        assert!(matches!(
+            err.io_error_kind(),
+            K::ConnectionReset | K::BrokenPipe | K::NotConnected
+        ));
+    }
+
+    // Send SupplyKeypairRequest
+    {
+        use rustix::fs::{open, Mode, OFlags};
+        let sk = open(peer_a_keypair.secret_key, OFlags::RDONLY, Mode::empty())?;
+        let pk = open(peer_a_keypair.public_key, OFlags::RDONLY, Mode::empty())?;
+
+        let mut fds = vec![&sk, &pk].into();
+        let mut api = WriteWithFileDescriptors::<UnixStream, _, _, _>::new(&api, &mut fds);
+        LengthPrefixEncoder::from_message(api::SupplyKeypairRequest::new().as_bytes())
+            .write_all_to_stdio(&mut api)?;
+        assert!(fds.is_empty(), "Failed to write all file descriptors");
+    }
+
+    // Read response
+    {
+        let mut decoder = LengthPrefixDecoder::new([0u8; api::MAX_RESPONSE_LEN]);
+        let res = decoder.read_all_from_stdio(&api)?;
+        let res = res.zk_parse::<api::SupplyKeypairResponse>()?;
+        assert_eq!(
+            *res,
+            api::SupplyKeypairResponse::new(supply_keypair_response_status::OK)
+        );
+    }
+
+    // Send AddPskBroker request
+    {
+        let mut fds = vec![psk_broker_server_sock.as_fd()].into();
+        let mut api = WriteWithFileDescriptors::<UnixStream, _, _, _>::new(&api, &mut fds);
+        LengthPrefixEncoder::from_message(api::AddPskBrokerRequest::new().as_bytes())
+            .write_all_to_stdio(&mut api)?;
+        assert!(fds.is_empty(), "Failed to write all file descriptors");
+    }
+
+    // Read response
+    {
+        let mut decoder = LengthPrefixDecoder::new([0u8; api::MAX_RESPONSE_LEN]);
+        let res = decoder.read_all_from_stdio(&api)?;
+        let res = res.zk_parse::<api::AddPskBrokerResponse>()?;
+        assert_eq!(
+            *res,
+            api::AddPskBrokerResponse::new(add_psk_broker_response_status::OK)
+        );
+    }
+
+    // Wait for the keys to successfully exchange a key
+    let mut attempt = 0;
+    loop {
+        // Read OSK generated by A
+        let osk_a = {
+            use rosenpass_wireguard_broker::api::msgs as M;
+            type SetPskReqPkg = M::Envelope<M::SetPskRequest>;
+            type SetPskResPkg = M::Envelope<M::SetPskResponse>;
+
+            // Receive request
+            let mut decoder = LengthPrefixDecoder::new([0u8; M::REQUEST_MSG_BUFFER_SIZE]);
+            let req = decoder.read_all_from_stdio(&psk_broker_sock)?;
+
+            let req = req.zk_parse::<SetPskReqPkg>()?;
+            assert_eq!(req.msg_type, M::MsgType::SetPsk as u8);
+            assert_eq!(req.payload.peer_id, peer_b_wg_peer_id);
+            assert_eq!(req.payload.iface()?, peer_b_wg_device);
+
+            // Send response
+            let res = SetPskResPkg {
+                msg_type: M::MsgType::SetPsk as u8,
+                reserved: [0u8; 3],
+                payload: M::SetPskResponse {
+                    return_code: M::SetPskResponseReturnCode::Success as u8,
+                },
+            };
+            LengthPrefixEncoder::from_message(res.as_bytes())
+                .write_all_to_stdio(&psk_broker_sock)?;
+
+            SymKey::from_slice(&req.payload.psk)
+        };
+
+        // Read OSK generated by B
+        let osk_b = {
+            let line = out_b.next().context("")??;
+            let words = line.split(' ').collect::<Vec<_>>();
+
+            // FIXED     FIXED PEER-ID                                      FIXED    FILENAME       STATUS
+            // output-key peer KZqXTZ4l2aNnkJtLPhs4D8JxHTGmRSL9w3Qr+X8JxFk= key-file "client-A-osk" exchanged
+            let peer_id = words
+                .get(2)
+                .with_context(|| format!("Bad rosenpass output: `{line}`"))?;
+            assert_eq!(
+                line,
+                format!(
+                    "output-key peer {peer_id} key-file \"{}\" exchanged",
+                    peer_b_osk.to_str().context("")?
+                )
+            );
+
+            SymKey::load_b64::<64, _>(peer_b_osk.clone())?
+        };
+
+        // TODO: This may be flaky. Both rosenpass instances are not guaranteed to produce
+        // the same number of output events; they merely guarantee eventual consistency of OSK.
+        // Correctly, we should use tokio to read any number of generated OSKs and indicate
+        // success on consensus
+        match osk_a.secret() == osk_b.secret() {
+            true => break,
+            false if attempt > 10 => bail!("Peers did not produce a matching key even after ten attempts. Something is wrong with the key exchange!"),
+            false => {},
+        };
+
+        attempt += 1;
+    }
+
+    Ok(())
+}

rosenpass/tests/api-integration-tests.rs

@@ -8,16 +8,25 @@ use std::{
 use anyhow::{bail, Context};
 use rosenpass::api;
 use rosenpass_to::{ops::copy_slice_least_src, To};
-use rosenpass_util::zerocopy::ZerocopySliceExt;
 use rosenpass_util::{
     file::LoadValueB64,
     length_prefix_encoding::{decoder::LengthPrefixDecoder, encoder::LengthPrefixEncoder},
 };
+use rosenpass_util::{mem::DiscardResultExt, zerocopy::ZerocopySliceExt};
 use tempfile::TempDir;
 use zerocopy::AsBytes;
 
 use rosenpass::protocol::SymKey;
 
+struct KillChild(std::process::Child);
+
+impl Drop for KillChild {
+    fn drop(&mut self) {
+        self.0.kill().discard_result();
+        self.0.wait().discard_result()
+    }
+}
+
 #[test]
 fn api_integration_test() -> anyhow::Result<()> {
     rosenpass_secret_memory::policy::secret_policy_use_only_malloc_secrets();
@@ -37,10 +46,11 @@ fn api_integration_test() -> anyhow::Result<()> {
     let peer_b_osk = tempfile!("b.osk");
 
     use rosenpass::config;
+
+    let peer_a_keypair = config::Keypair::new(tempfile!("a.pk"), tempfile!("a.sk"));
     let peer_a = config::Rosenpass {
         config_file_path: tempfile!("a.config"),
-        secret_key: tempfile!("a.sk"),
-        public_key: tempfile!("a.pk"),
+        keypair: Some(peer_a_keypair.clone()),
         listen: peer_a_endpoint.to_socket_addrs()?.collect(), // TODO: This could collide by accident
         verbosity: config::Verbosity::Verbose,
         api: api::config::ApiConfig {
@@ -57,10 +67,10 @@ fn api_integration_test() -> anyhow::Result<()> {
         }],
     };
 
+    let peer_b_keypair = config::Keypair::new(tempfile!("b.pk"), tempfile!("b.sk"));
     let peer_b = config::Rosenpass {
         config_file_path: tempfile!("b.config"),
-        secret_key: tempfile!("b.sk"),
-        public_key: tempfile!("b.pk"),
+        keypair: Some(peer_b_keypair.clone()),
         listen: vec![],
         verbosity: config::Verbosity::Verbose,
         api: api::config::ApiConfig {
@@ -79,12 +89,12 @@ fn api_integration_test() -> anyhow::Result<()> {
 
     // Generate the keys
     rosenpass::cli::testing::generate_and_save_keypair(
-        peer_a.secret_key.clone(),
-        peer_a.public_key.clone(),
+        peer_a_keypair.secret_key.clone(),
+        peer_a_keypair.public_key.clone(),
     )?;
     rosenpass::cli::testing::generate_and_save_keypair(
-        peer_b.secret_key.clone(),
-        peer_b.public_key.clone(),
+        peer_b_keypair.secret_key.clone(),
+        peer_b_keypair.public_key.clone(),
     )?;
 
     // Write the configuration files
@@ -92,28 +102,32 @@ fn api_integration_test() -> anyhow::Result<()> {
     peer_b.commit()?;
 
     // Start peer a
-    let proc_a = std::process::Command::new(env!("CARGO_BIN_EXE_rosenpass"))
-        .args([
-            "exchange-config",
-            peer_a.config_file_path.to_str().context("")?,
-        ])
-        .stdin(Stdio::null())
-        .stdout(Stdio::piped())
-        .spawn()?;
+    let mut proc_a = KillChild(
+        std::process::Command::new(env!("CARGO_BIN_EXE_rosenpass"))
+            .args([
+                "exchange-config",
+                peer_a.config_file_path.to_str().context("")?,
+            ])
+            .stdin(Stdio::null())
+            .stdout(Stdio::piped())
+            .spawn()?,
+    );
 
     // Start peer b
-    let proc_b = std::process::Command::new(env!("CARGO_BIN_EXE_rosenpass"))
-        .args([
-            "exchange-config",
-            peer_b.config_file_path.to_str().context("")?,
-        ])
-        .stdin(Stdio::null())
-        .stdout(Stdio::piped())
-        .spawn()?;
+    let mut proc_b = KillChild(
+        std::process::Command::new(env!("CARGO_BIN_EXE_rosenpass"))
+            .args([
+                "exchange-config",
+                peer_b.config_file_path.to_str().context("")?,
+            ])
+            .stdin(Stdio::null())
+            .stdout(Stdio::piped())
+            .spawn()?,
+    );
 
     // Acquire stdout
-    let mut out_a = BufReader::new(proc_a.stdout.context("")?).lines();
-    let mut out_b = BufReader::new(proc_b.stdout.context("")?).lines();
+    let mut out_a = BufReader::new(proc_a.0.stdout.take().context("")?).lines();
+    let mut out_b = BufReader::new(proc_b.0.stdout.take().context("")?).lines();
 
     // Wait for the keys to successfully exchange a key
     let mut attempt = 0;

rosenpass/tests/integration_test.rs

@@ -108,7 +108,7 @@ fn run_server_client_exchange(
             .termination_handler(Some(server_terminate_rx))
             .build()
             .unwrap();
-        cli.run(Some(test_helpers)).unwrap();
+        cli.run(None, Some(test_helpers)).unwrap();
     });
 
     let cli = CliArgs::try_parse_from(
@@ -123,7 +123,7 @@ fn run_server_client_exchange(
             .termination_handler(Some(client_terminate_rx))
             .build()
             .unwrap();
-        cli.run(Some(test_helpers)).unwrap();
+        cli.run(None, Some(test_helpers)).unwrap();
     });
 
     // give them some time to do the key exchange under load
@@ -293,6 +293,7 @@ struct MockBrokerInner {
 #[derive(Debug, Default)]
 struct MockBroker {
     inner: Arc<Mutex<MockBrokerInner>>,
+    mio_token: Option<mio::Token>,
 }
 
 impl WireguardBrokerMio for MockBroker {
@@ -301,8 +302,9 @@ impl WireguardBrokerMio for MockBroker {
     fn register(
         &mut self,
         _registry: &mio::Registry,
-        _token: mio::Token,
+        token: mio::Token,
     ) -> Result<(), Self::MioError> {
+        self.mio_token = Some(token);
         Ok(())
     }
 
@@ -311,8 +313,13 @@ impl WireguardBrokerMio for MockBroker {
     }
 
     fn unregister(&mut self, _registry: &mio::Registry) -> Result<(), Self::MioError> {
+        self.mio_token = None;
         Ok(())
     }
+
+    fn mio_token(&self) -> Option<mio::Token> {
+        self.mio_token
+    }
 }
 
 impl rosenpass_wireguard_broker::WireGuardBroker for MockBroker {

rp/Cargo.toml

@@ -20,9 +20,9 @@ rosenpass-ciphers = { workspace = true }
 rosenpass-cipher-traits = { workspace = true }
 rosenpass-secret-memory = { workspace = true }
 rosenpass-util = { workspace = true }
-rosenpass-wireguard-broker = {workspace = true}
+rosenpass-wireguard-broker = { workspace = true }
 
-tokio = {workspace = true}
+tokio = { workspace = true }
 
 [target.'cfg(any(target_os = "linux", target_os = "freebsd"))'.dependencies]
 ctrlc-async = "3.2"
@@ -35,8 +35,8 @@ netlink-packet-generic = "0.3"
 netlink-packet-wireguard = "0.2"
 
 [dev-dependencies]
-tempfile = {workspace = true}
-stacker = {workspace = true}
+tempfile = { workspace = true }
+stacker = { workspace = true }
 
 [features]
 experiment_memfd_secret = []

rp/src/exchange.rs

@@ -188,8 +188,7 @@ pub async fn exchange(options: ExchangeOptions) -> Result<()> {
     let pk = SPk::load(&pqpk)?;
 
     let mut srv = Box::new(AppServer::new(
-        sk,
-        pk,
+        Some((sk, pk)),
         if let Some(listen) = options.listen {
             vec![listen]
         } else {

secret-memory/Cargo.toml

@@ -21,6 +21,6 @@ log = { workspace = true }
 
 [dev-dependencies]
 allocator-api2-tests = { workspace = true }
-tempfile = {workspace = true}
-base64ct = {workspace = true}
-procspawn = {workspace = true}
+tempfile = { workspace = true }
+base64ct = { workspace = true }
+procspawn = { workspace = true }

util/Cargo.toml

@@ -16,8 +16,14 @@ base64ct = { workspace = true }
 anyhow = { workspace = true }
 typenum = { workspace = true }
 static_assertions = { workspace = true }
-rustix = {workspace = true}
-zeroize = {workspace = true}
+rustix = { workspace = true }
+zeroize = { workspace = true }
 zerocopy = { workspace = true }
 thiserror = { workspace = true }
 mio = { workspace = true }
+tempfile = { workspace = true }
+uds = { workspace = true, optional = true, features = ["mio_1xx"] }
+
+
+[features]
+experiment_file_descriptor_passing = ["uds"]

util/src/build.rs

@@ -0,0 +1,169 @@
+use crate::{
+    functional::ApplyExt,
+    mem::{SwapWithDefaultExt, SwapWithExt},
+};
+
+#[derive(thiserror::Error, Debug)]
+pub enum ConstructionSiteErectError<E> {
+    #[error("Construction site is void")]
+    IsVoid,
+    #[error("Construction is already built")]
+    AlreadyBuilt,
+    #[error("Other construction site error {0:?}")]
+    Other(#[from] E),
+}
+
+pub trait Build<T>: Sized {
+    type Error;
+    fn build(self) -> Result<T, Self::Error>;
+}
+
+#[derive(Debug)]
+pub enum ConstructionSite<Builder, T>
+where
+    Builder: Build<T>,
+{
+    Void,
+    Builder(Builder),
+    Product(T),
+}
+
+impl<Builder, T> Default for ConstructionSite<Builder, T>
+where
+    Builder: Build<T>,
+{
+    fn default() -> Self {
+        Self::Void
+    }
+}
+
+impl<Builder, T> ConstructionSite<Builder, T>
+where
+    Builder: Build<T>,
+{
+    pub fn void() -> Self {
+        Self::Void
+    }
+
+    pub fn new(builder: Builder) -> Self {
+        Self::Builder(builder)
+    }
+
+    pub fn from_product(value: T) -> Self {
+        Self::Product(value)
+    }
+
+    pub fn take(&mut self) -> Self {
+        self.swap_with_default()
+    }
+
+    pub fn modify_taken_with_return<R, F>(&mut self, f: F) -> R
+    where
+        F: FnOnce(Self) -> (Self, R),
+    {
+        let (site, res) = self.take().apply(f);
+        self.swap_with(site);
+        res
+    }
+
+    pub fn modify_taken<F>(&mut self, f: F)
+    where
+        F: FnOnce(Self) -> Self,
+    {
+        self.take().apply(f).swap_with_mut(self)
+    }
+
+    #[allow(clippy::result_unit_err)]
+    pub fn erect(&mut self) -> Result<(), ConstructionSiteErectError<Builder::Error>> {
+        self.modify_taken_with_return(|site| {
+            let builder = match site {
+                site @ Self::Void => return (site, Err(ConstructionSiteErectError::IsVoid)),
+                site @ Self::Product(_) => {
+                    return (site, Err(ConstructionSiteErectError::AlreadyBuilt))
+                }
+                Self::Builder(builder) => builder,
+            };
+
+            let product = match builder.build() {
+                Err(e) => {
+                    return (Self::void(), Err(ConstructionSiteErectError::Other(e)));
+                }
+                Ok(p) => p,
+            };
+
+            (Self::from_product(product), Ok(()))
+        })
+    }
+
+    /// Returns `true` if the construction site is [`Void`].
+    ///
+    /// [`Void`]: ConstructionSite::Void
+    #[must_use]
+    pub fn is_void(&self) -> bool {
+        matches!(self, Self::Void)
+    }
+
+    /// Returns `true` if the construction site is [`InProgress`].
+    ///
+    /// [`InProgress`]: ConstructionSite::InProgress
+    #[must_use]
+    pub fn in_progess(&self) -> bool {
+        matches!(self, Self::Builder(..))
+    }
+
+    /// Returns `true` if the construction site is [`Done`].
+    ///
+    /// [`Done`]: ConstructionSite::Done
+    #[must_use]
+    pub fn is_available(&self) -> bool {
+        matches!(self, Self::Product(..))
+    }
+
+    pub fn into_builder(self) -> Option<Builder> {
+        use ConstructionSite as S;
+        match self {
+            S::Builder(v) => Some(v),
+            _ => None,
+        }
+    }
+
+    pub fn builder_ref(&self) -> Option<&Builder> {
+        use ConstructionSite as S;
+        match self {
+            S::Builder(v) => Some(v),
+            _ => None,
+        }
+    }
+
+    pub fn builder_mut(&mut self) -> Option<&mut Builder> {
+        use ConstructionSite as S;
+        match self {
+            S::Builder(v) => Some(v),
+            _ => None,
+        }
+    }
+
+    pub fn into_product(self) -> Option<T> {
+        use ConstructionSite as S;
+        match self {
+            S::Product(v) => Some(v),
+            _ => None,
+        }
+    }
+
+    pub fn product_ref(&self) -> Option<&T> {
+        use ConstructionSite as S;
+        match self {
+            S::Product(v) => Some(v),
+            _ => None,
+        }
+    }
+
+    pub fn product_mut(&mut self) -> Option<&mut T> {
+        use ConstructionSite as S;
+        match self {
+            S::Product(v) => Some(v),
+            _ => None,
+        }
+    }
+}

util/src/controlflow.rs

@@ -0,0 +1,149 @@
+/// A collection of control flow utility macros
+
+#[macro_export]
+/// A simple for loop to repeat a $body a number of times
+macro_rules! repeat {
+    ($times:expr, $body:expr) => {
+        for _ in 0..($times) {
+            $body
+        }
+    };
+}
+
+#[macro_export]
+/// Return unless the condition $cond is true, with return value $val, if given.
+macro_rules! return_unless {
+    ($cond:expr) => {
+        if !($cond) {
+            return;
+        }
+    };
+    ($cond:expr, $val:expr) => {
+        if !($cond) {
+            return $val;
+        }
+    };
+}
+
+#[macro_export]
+/// Return if the condition $cond is true, with return value $val, if given.
+macro_rules! return_if {
+    ($cond:expr) => {
+        if $cond {
+            return;
+        }
+    };
+    ($cond:expr, $val:expr) => {
+        if $cond {
+            return $val;
+        }
+    };
+}
+
+#[macro_export]
+/// Break unless the condition is true, from the loop with label $val, if given.
+macro_rules! break_if {
+    ($cond:expr) => {
+        if $cond {
+            break;
+        }
+    };
+    ($cond:expr, $val:tt) => {
+        if $cond {
+            break $val;
+        }
+    };
+}
+
+#[macro_export]
+/// Continue if the condition is true, in the loop with label $val, if given.
+macro_rules! continue_if {
+    ($cond:expr) => {
+        if $cond {
+            continue;
+        }
+    };
+    ($cond:expr, $val:tt) => {
+        if $cond {
+            continue $val;
+        }
+    };
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn test_repeat() {
+        let mut sum = 0;
+        repeat!(10, {
+            sum += 1;
+        });
+        assert_eq!(sum, 10);
+    }
+
+    #[test]
+    fn test_return_unless() {
+        fn test_fn() -> i32 {
+            return_unless!(true, 1);
+            0
+        }
+        assert_eq!(test_fn(), 0);
+
+        fn test_fn2() -> i32 {
+            return_unless!(false, 1);
+            0
+        }
+        assert_eq!(test_fn2(), 1);
+    }
+
+    #[test]
+    fn test_return_if() {
+        fn test_fn() -> i32 {
+            return_if!(true, 1);
+            0
+        }
+        assert_eq!(test_fn(), 1);
+
+        fn test_fn2() -> i32 {
+            return_if!(false, 1);
+            0
+        }
+        assert_eq!(test_fn2(), 0);
+    }
+
+    #[test]
+    fn test_break_if() {
+        let mut sum = 0;
+        for i in 0..10 {
+            break_if!(i == 5);
+            sum += 1;
+        }
+        assert_eq!(sum, 5);
+
+        let mut sum = 0;
+        'one: for _ in 0..10 {
+            for j in 0..20 {
+                break_if!(j == 5, 'one);
+                sum += 1;
+            }
+        }
+        assert_eq!(sum, 5);
+    }
+
+    #[test]
+    fn test_continue_if() {
+        let mut sum = 0;
+        for i in 0..10 {
+            continue_if!(i == 5);
+            sum += 1;
+        }
+        assert_eq!(sum, 9);
+
+        let mut sum = 0;
+        'one: for i in 0..10 {
+            continue_if!(i == 5, 'one);
+            sum += 1;
+        }
+        assert_eq!(sum, 9);
+    }
+}

util/src/fd.rs

@@ -1,12 +1,14 @@
+use anyhow::bail;
 use rustix::{
     fd::{AsFd, BorrowedFd, FromRawFd, OwnedFd, RawFd},
-    io::{fcntl_dupfd_cloexec, DupFlags},
+    io::fcntl_dupfd_cloexec,
 };
 
-use crate::mem::Forgetting;
+use crate::{mem::Forgetting, result::OkExt};
 
 /// Prepare a file descriptor for use in Rust code.
 ///
+
 /// Checks if the file descriptor is valid and duplicates it to a new file descriptor.
 /// The old file descriptor is masked to avoid potential use after free (on file descriptor)
 /// in case the given file descriptor is still used somewhere
@@ -16,6 +18,18 @@ pub fn claim_fd(fd: RawFd) -> rustix::io::Result<OwnedFd> {
     Ok(new)
 }
 
+/// Prepare a file descriptor for use in Rust code.
+///
+/// Checks if the file descriptor is valid.
+///
+/// Unlike [claim_fd], this will reuse the same file descriptor identifier instead of masking it.
+pub fn claim_fd_inplace(fd: RawFd) -> rustix::io::Result<OwnedFd> {
+    let mut new = unsafe { OwnedFd::from_raw_fd(fd) };
+    let tmp = clone_fd_cloexec(&new)?;
+    clone_fd_to_cloexec(tmp, &mut new)?;
+    Ok(new)
+}
+
 pub fn mask_fd(fd: RawFd) -> rustix::io::Result<()> {
     // Safety: because the OwnedFd resulting from OwnedFd::from_raw_fd is wrapped in a Forgetting,
     // it never gets dropped, meaning that fd is never closed and thus outlives the OwnedFd
@@ -30,7 +44,7 @@ pub fn clone_fd_cloexec<Fd: AsFd>(fd: Fd) -> rustix::io::Result<OwnedFd> {
 
 #[cfg(target_os = "linux")]
 pub fn clone_fd_to_cloexec<Fd: AsFd>(fd: Fd, new: &mut OwnedFd) -> rustix::io::Result<()> {
-    use rustix::io::dup3;
+    use rustix::io::{dup3, DupFlags};
     dup3(fd, new, DupFlags::CLOEXEC)
 }
 
@@ -48,3 +62,239 @@ pub fn open_nullfd() -> rustix::io::Result<OwnedFd> {
     // TODO: Add tests showing that this will throw errors on use
     open("/dev/null", OFlags::CLOEXEC, Mode::empty())
 }
+
+/// Convert low level errors into std::io::Error
+pub trait IntoStdioErr {
+    type Target;
+    fn into_stdio_err(self) -> Self::Target;
+}
+
+impl IntoStdioErr for rustix::io::Errno {
+    type Target = std::io::Error;
+
+    fn into_stdio_err(self) -> Self::Target {
+        std::io::Error::from_raw_os_error(self.raw_os_error())
+    }
+}
+
+impl<T> IntoStdioErr for rustix::io::Result<T> {
+    type Target = std::io::Result<T>;
+
+    fn into_stdio_err(self) -> Self::Target {
+        self.map_err(IntoStdioErr::into_stdio_err)
+    }
+}
+
+/// Read and write directly from a file descriptor
+pub struct FdIo<Fd: AsFd>(pub Fd);
+
+impl<Fd: AsFd> std::io::Read for FdIo<Fd> {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        rustix::io::read(&self.0, buf).into_stdio_err()
+    }
+}
+
+impl<Fd: AsFd> std::io::Write for FdIo<Fd> {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        rustix::io::write(&self.0, buf).into_stdio_err()
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+pub trait StatExt {
+    fn is_socket(&self) -> bool;
+}
+
+impl StatExt for rustix::fs::Stat {
+    fn is_socket(&self) -> bool {
+        use rustix::fs::FileType;
+        let ft = FileType::from_raw_mode(self.st_mode);
+        matches!(ft, FileType::Socket)
+    }
+}
+
+pub trait TryStatExt {
+    type Error;
+    fn is_socket(&self) -> Result<bool, Self::Error>;
+}
+
+impl<T> TryStatExt for T
+where
+    T: AsFd,
+{
+    type Error = rustix::io::Errno;
+
+    fn is_socket(&self) -> Result<bool, Self::Error> {
+        rustix::fs::fstat(self)?.is_socket().ok()
+    }
+}
+
+pub trait GetSocketType {
+    type Error;
+    fn socket_type(&self) -> Result<rustix::net::SocketType, Self::Error>;
+    fn is_datagram_socket(&self) -> Result<bool, Self::Error> {
+        use rustix::net::SocketType;
+        matches!(self.socket_type()?, SocketType::DGRAM).ok()
+    }
+    fn is_stream_socket(&self) -> Result<bool, Self::Error> {
+        Ok(self.socket_type()? == rustix::net::SocketType::STREAM)
+    }
+}
+
+impl<T> GetSocketType for T
+where
+    T: AsFd,
+{
+    type Error = rustix::io::Errno;
+
+    fn socket_type(&self) -> Result<rustix::net::SocketType, Self::Error> {
+        rustix::net::sockopt::get_socket_type(self)
+    }
+}
+
+#[cfg(target_os = "linux")]
+pub trait GetSocketDomain {
+    type Error;
+    fn socket_domain(&self) -> Result<rustix::net::AddressFamily, Self::Error>;
+    fn socket_address_family(&self) -> Result<rustix::net::AddressFamily, Self::Error> {
+        self.socket_domain()
+    }
+    fn is_unix_socket(&self) -> Result<bool, Self::Error> {
+        Ok(self.socket_domain()? == rustix::net::AddressFamily::UNIX)
+    }
+}
+
+#[cfg(target_os = "linux")]
+impl<T> GetSocketDomain for T
+where
+    T: AsFd,
+{
+    type Error = rustix::io::Errno;
+
+    fn socket_domain(&self) -> Result<rustix::net::AddressFamily, Self::Error> {
+        rustix::net::sockopt::get_socket_domain(self)
+    }
+}
+
+#[cfg(target_os = "linux")]
+pub trait GetUnixSocketType {
+    type Error;
+    fn is_unix_stream_socket(&self) -> Result<bool, Self::Error>;
+    fn demand_unix_stream_socket(&self) -> anyhow::Result<()>;
+}
+
+#[cfg(target_os = "linux")]
+impl<T> GetUnixSocketType for T
+where
+    T: GetSocketType + GetSocketDomain<Error = <T as GetSocketType>::Error>,
+    anyhow::Error: From<<T as GetSocketType>::Error>,
+{
+    type Error = <T as GetSocketType>::Error;
+
+    fn is_unix_stream_socket(&self) -> Result<bool, Self::Error> {
+        Ok(self.is_unix_socket()? && self.is_stream_socket()?)
+    }
+
+    fn demand_unix_stream_socket(&self) -> anyhow::Result<()> {
+        use rustix::net::AddressFamily as SA;
+        use rustix::net::SocketType as ST;
+        match (self.socket_domain()?, self.socket_type()?) {
+            (SA::UNIX, ST::STREAM) => Ok(()),
+            (SA::UNIX, mode) => bail!("Expected unix socket in stream mode, but mode is {mode:?}"),
+            (domain, _) => bail!("Expected unix socket, but socket domain is {domain:?} instead"),
+        }
+    }
+}
+
+#[cfg(target_os = "linux")]
+pub trait GetSocketProtocol {
+    fn socket_protocol(&self) -> Result<Option<rustix::net::Protocol>, rustix::io::Errno>;
+    fn is_udp_socket(&self) -> Result<bool, rustix::io::Errno> {
+        self.socket_protocol()?
+            .map(|p| p == rustix::net::ipproto::UDP)
+            .unwrap_or(false)
+            .ok()
+    }
+    fn demand_udp_socket(&self) -> anyhow::Result<()> {
+        match self.socket_protocol() {
+            Ok(Some(rustix::net::ipproto::UDP)) => Ok(()),
+            Ok(Some(other_proto)) => {
+                bail!("Not a udp socket, instead socket protocol is: {other_proto:?}")
+            }
+            Ok(None) => bail!("getsockopt() returned empty value"),
+            Err(errno) => Err(errno.into_stdio_err())?,
+        }
+    }
+}
+
+#[cfg(target_os = "linux")]
+impl<T> GetSocketProtocol for T
+where
+    T: AsFd,
+{
+    fn socket_protocol(&self) -> Result<Option<rustix::net::Protocol>, rustix::io::Errno> {
+        rustix::net::sockopt::get_socket_protocol(self)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::fs::{read_to_string, File};
+    use std::io::{Read, Write};
+    use std::os::fd::IntoRawFd;
+    use tempfile::tempdir;
+
+    #[test]
+    fn test_claim_fd() {
+        let tmp_dir = tempdir().unwrap();
+        let path = tmp_dir.path().join("test");
+        let file = File::create(path.clone()).unwrap();
+        let fd: RawFd = file.into_raw_fd();
+        let owned_fd = claim_fd(fd).unwrap();
+        let mut file = unsafe { File::from_raw_fd(owned_fd.into_raw_fd()) };
+        file.write_all(b"Hello, World!").unwrap();
+
+        let message = read_to_string(path).unwrap();
+        assert_eq!(message, "Hello, World!");
+    }
+
+    #[test]
+    #[should_panic(expected = "fd != u32::MAX as RawFd")]
+    fn test_claim_fd_invalid_neg() {
+        let fd: RawFd = -1;
+        let _ = claim_fd(fd);
+    }
+
+    #[test]
+    #[should_panic(expected = "fd != u32::MAX as RawFd")]
+    fn test_claim_fd_invalid_max() {
+        let fd: RawFd = i64::MAX as RawFd;
+        let _ = claim_fd(fd);
+    }
+
+    #[test]
+    fn test_open_nullfd_write() {
+        let nullfd = open_nullfd().unwrap();
+        let mut file = unsafe { File::from_raw_fd(nullfd.into_raw_fd()) };
+        let res = file.write_all(b"Hello, World!");
+        assert!(res.is_err());
+        assert_eq!(
+            res.unwrap_err().to_string(),
+            "Bad file descriptor (os error 9)"
+        );
+    }
+
+    #[test]
+    fn test_open_nullfd_read() {
+        let nullfd = open_nullfd().unwrap();
+        let mut file = unsafe { File::from_raw_fd(nullfd.into_raw_fd()) };
+        let mut buffer = [0; 10];
+        let res = file.read_exact(&mut buffer);
+        assert!(res.is_err());
+        assert_eq!(res.unwrap_err().to_string(), "failed to fill whole buffer");
+    }
+}

util/src/file.rs

@@ -114,3 +114,96 @@ pub trait DisplayValueB64 {
 
     fn display_b64<'o>(&self, output: &'o mut [u8]) -> Result<&'o str, Self::Error>;
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::fs::File;
+    use std::io::Write;
+    use std::os::unix::fs::PermissionsExt;
+    use tempfile::tempdir;
+
+    #[test]
+    fn test_fopen_w_public() {
+        let tmp_dir = tempdir().unwrap();
+        let path = tmp_dir.path().join("test");
+        let mut file = fopen_w(path, Visibility::Public).unwrap();
+        file.write_all(b"test").unwrap();
+        let metadata = file.metadata().unwrap();
+        let permissions = metadata.permissions();
+        assert_eq!(permissions.mode(), 0o100644);
+    }
+
+    #[test]
+    fn test_fopen_w_secret() {
+        let tmp_dir = tempdir().unwrap();
+        let path = tmp_dir.path().join("test");
+        let mut file = fopen_w(path, Visibility::Secret).unwrap();
+        file.write_all(b"test").unwrap();
+        let metadata = file.metadata().unwrap();
+        let permissions = metadata.permissions();
+        assert_eq!(permissions.mode(), 0o100600);
+    }
+
+    #[test]
+    fn test_fopen_r() {
+        let tmp_dir = tempdir().unwrap();
+        let path = tmp_dir.path().join("test");
+        let mut file = File::create(path.clone()).unwrap();
+        file.write_all(b"test").unwrap();
+        let mut contents = String::new();
+        let mut file = fopen_r(path).unwrap();
+        file.read_to_string(&mut contents).unwrap();
+        assert_eq!(contents, "test");
+    }
+
+    #[test]
+    fn test_read_slice_to_end() {
+        let tmp_dir = tempdir().unwrap();
+        let path = tmp_dir.path().join("test");
+        let mut file = File::create(path.clone()).unwrap();
+        file.write_all(b"test").unwrap();
+        let mut buf = [0u8; 4];
+        let mut file = fopen_r(path).unwrap();
+        file.read_slice_to_end(&mut buf).unwrap();
+        assert_eq!(buf, [116, 101, 115, 116]);
+    }
+
+    #[test]
+    fn test_read_exact_to_end() {
+        let tmp_dir = tempdir().unwrap();
+        let path = tmp_dir.path().join("test");
+        let mut file = File::create(path.clone()).unwrap();
+        file.write_all(b"test").unwrap();
+        let mut buf = [0u8; 4];
+        let mut file = fopen_r(path).unwrap();
+        file.read_exact_to_end(&mut buf).unwrap();
+        assert_eq!(buf, [116, 101, 115, 116]);
+    }
+
+    #[test]
+    fn test_read_exact_to_end_to_long() {
+        let tmp_dir = tempdir().unwrap();
+        let path = tmp_dir.path().join("test");
+        let mut file = File::create(path.clone()).unwrap();
+        file.write_all(b"test").unwrap();
+        let mut buf = [0u8; 3];
+        let mut file = fopen_r(path).unwrap();
+        let result = file.read_exact_to_end(&mut buf);
+        assert!(result.is_err());
+        assert_eq!(result.unwrap_err().to_string(), "File too long!");
+    }
+
+    #[test]
+    fn test_read_slice_to_end_to_long() {
+        let tmp_dir = tempdir().unwrap();
+        let path = tmp_dir.path().join("test");
+        let mut file = File::create(path.clone()).unwrap();
+        file.write_all(b"test").unwrap();
+        let mut buf = [0u8; 3];
+        let mut file = fopen_r(path).unwrap();
+        let result = file.read_slice_to_end(&mut buf);
+        assert!(result.is_err());
+        assert_eq!(result.unwrap_err().to_string(), "File too long!");
+    }
+}

util/src/functional.rs

@@ -6,6 +6,32 @@ where
     v
 }
 
+pub trait MutatingExt {
+    fn mutating<F>(self, f: F) -> Self
+    where
+        F: Fn(&mut Self);
+    fn mutating_mut<F>(&mut self, f: F) -> &mut Self
+    where
+        F: Fn(&mut Self);
+}
+
+impl<T> MutatingExt for T {
+    fn mutating<F>(self, f: F) -> Self
+    where
+        F: Fn(&mut Self),
+    {
+        mutating(self, f)
+    }
+
+    fn mutating_mut<F>(&mut self, f: F) -> &mut Self
+    where
+        F: Fn(&mut Self),
+    {
+        f(self);
+        self
+    }
+}
+
 pub fn sideeffect<T, F>(v: T, f: F) -> T
 where
     F: Fn(&T),
@@ -14,6 +40,58 @@ where
     v
 }
 
+pub trait SideffectExt {
+    fn sideeffect<F>(self, f: F) -> Self
+    where
+        F: Fn(&Self);
+    fn sideeffect_ref<F>(&self, f: F) -> &Self
+    where
+        F: Fn(&Self);
+    fn sideeffect_mut<F>(&mut self, f: F) -> &mut Self
+    where
+        F: Fn(&Self);
+}
+
+impl<T> SideffectExt for T {
+    fn sideeffect<F>(self, f: F) -> Self
+    where
+        F: Fn(&Self),
+    {
+        sideeffect(self, f)
+    }
+
+    fn sideeffect_ref<F>(&self, f: F) -> &Self
+    where
+        F: Fn(&Self),
+    {
+        f(self);
+        self
+    }
+
+    fn sideeffect_mut<F>(&mut self, f: F) -> &mut Self
+    where
+        F: Fn(&Self),
+    {
+        f(self);
+        self
+    }
+}
+
 pub fn run<R, F: FnOnce() -> R>(f: F) -> R {
     f()
 }
+
+pub trait ApplyExt: Sized {
+    fn apply<R, F>(self, f: F) -> R
+    where
+        F: FnOnce(Self) -> R;
+}
+
+impl<T: Sized> ApplyExt for T {
+    fn apply<R, F>(self, f: F) -> R
+    where
+        F: FnOnce(Self) -> R,
+    {
+        f(self)
+    }
+}

util/src/io.rs

@@ -1,5 +1,7 @@
 use std::{borrow::Borrow, io};
 
+use anyhow::ensure;
+
 pub trait IoErrorKind {
     fn io_error_kind(&self) -> io::ErrorKind;
 }
@@ -50,6 +52,56 @@ impl<T, E: TryIoErrorKind> TryIoResultKindHintExt<T> for Result<T, E> {
     }
 }
 
+pub trait SubstituteForIoErrorKindExt<T>: Sized {
+    type Error;
+    fn substitute_for_ioerr_kind_with<F: FnOnce() -> T>(
+        self,
+        kind: io::ErrorKind,
+        f: F,
+    ) -> Result<T, Self::Error>;
+    fn substitute_for_ioerr_kind(self, kind: io::ErrorKind, v: T) -> Result<T, Self::Error> {
+        self.substitute_for_ioerr_kind_with(kind, || v)
+    }
+
+    fn substitute_for_ioerr_interrupted_with<F: FnOnce() -> T>(
+        self,
+        f: F,
+    ) -> Result<T, Self::Error> {
+        self.substitute_for_ioerr_kind_with(io::ErrorKind::Interrupted, f)
+    }
+
+    fn substitute_for_ioerr_interrupted(self, v: T) -> Result<T, Self::Error> {
+        self.substitute_for_ioerr_interrupted_with(|| v)
+    }
+
+    fn substitute_for_ioerr_wouldblock_with<F: FnOnce() -> T>(
+        self,
+        f: F,
+    ) -> Result<T, Self::Error> {
+        self.substitute_for_ioerr_kind_with(io::ErrorKind::WouldBlock, f)
+    }
+
+    fn substitute_for_ioerr_wouldblock(self, v: T) -> Result<T, Self::Error> {
+        self.substitute_for_ioerr_wouldblock_with(|| v)
+    }
+}
+
+impl<T, E: TryIoErrorKind> SubstituteForIoErrorKindExt<T> for Result<T, E> {
+    type Error = E;
+
+    fn substitute_for_ioerr_kind_with<F: FnOnce() -> T>(
+        self,
+        kind: io::ErrorKind,
+        f: F,
+    ) -> Result<T, Self::Error> {
+        match self.try_io_err_kind_hint() {
+            Ok(v) => Ok(v),
+            Err((_, Some(k))) if k == kind => Ok(f()),
+            Err((e, _)) => Err(e),
+        }
+    }
+}
+
 /// Automatically handles `std::io::ErrorKind::Interrupted`.
 ///
 /// - If there is no error (i.e. on `Ok(r)`), the function will return `Ok(Some(r))`
@@ -108,3 +160,21 @@ impl<T: io::Read> ReadNonblockingWithBoringErrorsHandledExt for T {
         nonblocking_handle_io_errors(|| self.read(buf))
     }
 }
+
+pub trait ReadExt {
+    fn read_exact_til_end(&mut self, buf: &mut [u8]) -> anyhow::Result<()>;
+}
+
+impl<T> ReadExt for T
+where
+    T: std::io::Read,
+{
+    fn read_exact_til_end(&mut self, buf: &mut [u8]) -> anyhow::Result<()> {
+        self.read_exact(buf)?;
+        ensure!(
+            self.read(&mut [0u8; 8])? == 0,
+            "Read source longer than buffer"
+        );
+        Ok(())
+    }
+}

util/src/lib.rs

@@ -1,6 +1,8 @@
 #![recursion_limit = "256"]
 
 pub mod b64;
+pub mod build;
+pub mod controlflow;
 pub mod fd;
 pub mod file;
 pub mod functional;
@@ -8,7 +10,7 @@ pub mod io;
 pub mod length_prefix_encoding;
 pub mod mem;
 pub mod mio;
-pub mod ord;
+pub mod option;
 pub mod result;
 pub mod time;
 pub mod typenum;

util/src/mem.rs

@@ -92,3 +92,61 @@ impl<T> Drop for Forgetting<T> {
         forget(value)
     }
 }
+
+pub trait DiscardResultExt {
+    fn discard_result(self);
+}
+
+impl<T> DiscardResultExt for T {
+    fn discard_result(self) {}
+}
+
+pub trait ForgetExt {
+    fn forget(self);
+}
+
+impl<T> ForgetExt for T {
+    fn forget(self) {
+        std::mem::forget(self)
+    }
+}
+
+pub trait SwapWithExt {
+    fn swap_with(&mut self, other: Self) -> Self;
+    fn swap_with_mut(&mut self, other: &mut Self);
+}
+
+impl<T> SwapWithExt for T {
+    fn swap_with(&mut self, mut other: Self) -> Self {
+        self.swap_with_mut(&mut other);
+        other
+    }
+
+    fn swap_with_mut(&mut self, other: &mut Self) {
+        std::mem::swap(self, other)
+    }
+}
+
+pub trait SwapWithDefaultExt {
+    fn swap_with_default(&mut self) -> Self;
+}
+
+impl<T: Default> SwapWithDefaultExt for T {
+    fn swap_with_default(&mut self) -> Self {
+        self.swap_with(Self::default())
+    }
+}
+
+pub trait MoveExt {
+    /// Deliberately move the value
+    ///
+    /// Usually employed to enforce an object being
+    /// dropped after use.
+    fn move_here(self) -> Self;
+}
+
+impl<T: Sized> MoveExt for T {
+    fn move_here(self) -> Self {
+        self
+    }
+}

util/src/mio/mio.rs

@@ -1,7 +1,10 @@
 use mio::net::{UnixListener, UnixStream};
-use rustix::fd::RawFd;
+use rustix::fd::{OwnedFd, RawFd};
 
-use crate::fd::claim_fd;
+use crate::{
+    fd::{claim_fd, claim_fd_inplace},
+    result::OkExt,
+};
 
 pub mod interest {
     use mio::Interest;
@@ -25,15 +28,26 @@ impl UnixListenerExt for UnixListener {
 }
 
 pub trait UnixStreamExt: Sized {
+    fn from_fd(fd: OwnedFd) -> anyhow::Result<Self>;
     fn claim_fd(fd: RawFd) -> anyhow::Result<Self>;
+    fn claim_fd_inplace(fd: RawFd) -> anyhow::Result<Self>;
 }
 
 impl UnixStreamExt for UnixStream {
-    fn claim_fd(fd: RawFd) -> anyhow::Result<Self> {
+    fn from_fd(fd: OwnedFd) -> anyhow::Result<Self> {
         use std::os::unix::net::UnixStream as StdUnixStream;
-
-        let sock = StdUnixStream::from(claim_fd(fd)?);
+        #[cfg(target_os = "linux")] // TODO: We should support this on other plattforms
+        crate::fd::GetUnixSocketType::demand_unix_stream_socket(&fd)?;
+        let sock = StdUnixStream::from(fd);
         sock.set_nonblocking(true)?;
-        Ok(UnixStream::from_std(sock))
+        UnixStream::from_std(sock).ok()
+    }
+
+    fn claim_fd(fd: RawFd) -> anyhow::Result<Self> {
+        Self::from_fd(claim_fd(fd)?)
+    }
+
+    fn claim_fd_inplace(fd: RawFd) -> anyhow::Result<Self> {
+        Self::from_fd(claim_fd_inplace(fd)?)
     }
 }

util/src/mio/mod.rs

@@ -0,0 +1,13 @@
+#[allow(clippy::module_inception)]
+mod mio;
+pub use mio::*;
+
+#[cfg(feature = "experiment_file_descriptor_passing")]
+mod uds_send_fd;
+#[cfg(feature = "experiment_file_descriptor_passing")]
+pub use uds_send_fd::*;
+
+#[cfg(feature = "experiment_file_descriptor_passing")]
+mod uds_recv_fd;
+#[cfg(feature = "experiment_file_descriptor_passing")]
+pub use uds_recv_fd::*;

util/src/mio/uds_recv_fd.rs

@@ -0,0 +1,123 @@
+use std::{
+    borrow::{Borrow, BorrowMut},
+    collections::VecDeque,
+    io::Read,
+    marker::PhantomData,
+    os::fd::OwnedFd,
+};
+use uds::UnixStreamExt as FdPassingExt;
+
+use crate::fd::{claim_fd_inplace, IntoStdioErr};
+
+pub struct ReadWithFileDescriptors<const MAX_FDS: usize, Sock, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    BorrowSock: Borrow<Sock>,
+    BorrowFds: BorrowMut<VecDeque<OwnedFd>>,
+{
+    socket: BorrowSock,
+    fds: BorrowFds,
+    _sock_dummy: PhantomData<Sock>,
+}
+
+impl<const MAX_FDS: usize, Sock, BorrowSock, BorrowFds>
+    ReadWithFileDescriptors<MAX_FDS, Sock, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    BorrowSock: Borrow<Sock>,
+    BorrowFds: BorrowMut<VecDeque<OwnedFd>>,
+{
+    pub fn new(socket: BorrowSock, fds: BorrowFds) -> Self {
+        let _sock_dummy = PhantomData;
+        Self {
+            socket,
+            fds,
+            _sock_dummy,
+        }
+    }
+
+    pub fn into_parts(self) -> (BorrowSock, BorrowFds) {
+        let Self { socket, fds, .. } = self;
+        (socket, fds)
+    }
+
+    pub fn socket(&self) -> &Sock {
+        self.socket.borrow()
+    }
+
+    pub fn fds(&self) -> &VecDeque<OwnedFd> {
+        self.fds.borrow()
+    }
+
+    pub fn fds_mut(&mut self) -> &mut VecDeque<OwnedFd> {
+        self.fds.borrow_mut()
+    }
+}
+
+impl<const MAX_FDS: usize, Sock, BorrowSock, BorrowFds>
+    ReadWithFileDescriptors<MAX_FDS, Sock, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    BorrowSock: BorrowMut<Sock>,
+    BorrowFds: BorrowMut<VecDeque<OwnedFd>>,
+{
+    pub fn socket_mut(&mut self) -> &mut Sock {
+        self.socket.borrow_mut()
+    }
+}
+
+impl<const MAX_FDS: usize, Sock, BorrowSock, BorrowFds> Read
+    for ReadWithFileDescriptors<MAX_FDS, Sock, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    BorrowSock: Borrow<Sock>,
+    BorrowFds: BorrowMut<VecDeque<OwnedFd>>,
+{
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        // Calculate space for additional file descriptors
+        let have_fds_before_read = self.fds().len();
+        let free_fd_slots = MAX_FDS.saturating_sub(have_fds_before_read);
+
+        // Allocate a buffer for file descriptors
+        let mut fd_buf = [0; MAX_FDS];
+        let fd_buf = &mut fd_buf[..free_fd_slots];
+
+        // Read from the unix socket
+        let (bytes_read, fds_read) = self.socket.borrow().recv_fds(buf, fd_buf)?;
+        let fd_buf = &fd_buf[..fds_read];
+
+        // Process the file descriptors
+        let mut fd_iter = fd_buf.iter();
+
+        // Try claiming all the file descriptors
+        let mut claim_fd_result = Ok(bytes_read);
+        self.fds_mut().reserve(fd_buf.len());
+        for fd in fd_iter.by_ref() {
+            match claim_fd_inplace(*fd) {
+                Ok(owned) => self.fds_mut().push_back(owned),
+                Err(e) => {
+                    // Abort on error and pass to error handler
+                    // Note that claim_fd_inplace is responsible for closing this particular
+                    // file descriptor if claiming it fails
+                    claim_fd_result = Err(e.into_stdio_err());
+                    break;
+                }
+            }
+        }
+
+        // Return if we where able to claim all file descriptors
+        if claim_fd_result.is_ok() {
+            return claim_fd_result;
+        };
+
+        // An error occurred while claiming fds
+        self.fds_mut().truncate(have_fds_before_read); // Close fds successfully claimed
+
+        // Close the remaining fds
+        for fd in fd_iter {
+            unsafe { rustix::io::close(*fd) };
+        }
+
+        claim_fd_result
+    }
+}

util/src/mio/uds_send_fd.rs

@@ -0,0 +1,121 @@
+use rustix::fd::{AsFd, AsRawFd};
+use std::{
+    borrow::{Borrow, BorrowMut},
+    cmp::min,
+    collections::VecDeque,
+    io::Write,
+    marker::PhantomData,
+};
+use uds::UnixStreamExt as FdPassingExt;
+
+use crate::{repeat, return_if};
+
+pub struct WriteWithFileDescriptors<Sock, Fd, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    Fd: AsFd,
+    BorrowSock: Borrow<Sock>,
+    BorrowFds: BorrowMut<VecDeque<Fd>>,
+{
+    socket: BorrowSock,
+    fds: BorrowFds,
+    _sock_dummy: PhantomData<Sock>,
+    _fd_dummy: PhantomData<Fd>,
+}
+
+impl<Sock, Fd, BorrowSock, BorrowFds> WriteWithFileDescriptors<Sock, Fd, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    Fd: AsFd,
+    BorrowSock: Borrow<Sock>,
+    BorrowFds: BorrowMut<VecDeque<Fd>>,
+{
+    pub fn new(socket: BorrowSock, fds: BorrowFds) -> Self {
+        let _sock_dummy = PhantomData;
+        let _fd_dummy = PhantomData;
+        Self {
+            socket,
+            fds,
+            _sock_dummy,
+            _fd_dummy,
+        }
+    }
+
+    pub fn into_parts(self) -> (BorrowSock, BorrowFds) {
+        let Self { socket, fds, .. } = self;
+        (socket, fds)
+    }
+
+    pub fn socket(&self) -> &Sock {
+        self.socket.borrow()
+    }
+
+    pub fn fds(&self) -> &VecDeque<Fd> {
+        self.fds.borrow()
+    }
+
+    pub fn fds_mut(&mut self) -> &mut VecDeque<Fd> {
+        self.fds.borrow_mut()
+    }
+}
+
+impl<Sock, Fd, BorrowSock, BorrowFds> WriteWithFileDescriptors<Sock, Fd, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    Fd: AsFd,
+    BorrowSock: BorrowMut<Sock>,
+    BorrowFds: BorrowMut<VecDeque<Fd>>,
+{
+    pub fn socket_mut(&mut self) -> &mut Sock {
+        self.socket.borrow_mut()
+    }
+}
+
+impl<Sock, Fd, BorrowSock, BorrowFds> Write
+    for WriteWithFileDescriptors<Sock, Fd, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    Fd: AsFd,
+    BorrowSock: Borrow<Sock>,
+    BorrowFds: BorrowMut<VecDeque<Fd>>,
+{
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        // At least one byte of real data should be sent when sending ancillary data. -- unix(7)
+        return_if!(buf.is_empty(), Ok(0));
+
+        // The kernel constant SCM_MAX_FD defines a limit on the number of file descriptors
+        // in the array.  Attempting to  send  an  array  larger  than  this  limit  causes
+        // sendmsg(2)  to fail with the error EINVAL.  SCM_MAX_FD has the value 253 (or 255
+        // before Linux 2.6.38).
+        // -- unix(7)
+        const SCM_MAX_FD: usize = 253;
+        let buf = match self.fds().len() <= SCM_MAX_FD {
+            false => &buf[..1], // Force caller to immediately call write() again to send its data
+            true => buf,
+        };
+
+        // Allocate the buffer for the file descriptor array
+        let fd_no = min(SCM_MAX_FD, self.fds().len());
+        let mut fd_buf = [0; SCM_MAX_FD]; // My kingdom for alloca(3)
+        let fd_buf = &mut fd_buf[..fd_no];
+
+        // Fill the file descriptor array
+        for (raw, fancy) in fd_buf.iter_mut().zip(self.fds().iter()) {
+            *raw = fancy.as_fd().as_raw_fd();
+        }
+
+        // Send data and file descriptors
+        let bytes_written = self.socket().send_fds(buf, fd_buf)?;
+
+        // Drop the file descriptors from the Deque
+        repeat!(fd_no, {
+            self.fds_mut().pop_front();
+        });
+
+        Ok(bytes_written)
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}

util/src/option.rs

@@ -0,0 +1,7 @@
+pub trait SomeExt: Sized {
+    fn some(self) -> Option<Self> {
+        Some(self)
+    }
+}
+
+impl<T> SomeExt for T {}

util/src/ord.rs

@@ -1,8 +0,0 @@
-// TODO remove this once std::cmp::max becomes const
-pub const fn max_usize(a: usize, b: usize) -> usize {
-    if a > b {
-        a
-    } else {
-        b
-    }
-}

util/src/result.rs

@@ -8,6 +8,16 @@ macro_rules! attempt {
     };
 }
 
+pub trait OkExt<E>: Sized {
+    fn ok(self) -> Result<Self, E>;
+}
+
+impl<T, E> OkExt<E> for T {
+    fn ok(self) -> Result<Self, E> {
+        Ok(self)
+    }
+}
+
 /// Trait for container types that guarantee successful unwrapping.
 ///
 /// The `.guaranteed()` function can be used over unwrap to show that
@@ -25,6 +35,24 @@ pub trait GuaranteedValue {
     fn guaranteed(self) -> Self::Value;
 }
 
+pub trait FinallyExt {
+    fn finally<F: FnOnce(&mut Self)>(self, f: F) -> Self;
+}
+
+impl<T, E> FinallyExt for Result<T, E> {
+    fn finally<F: FnOnce(&mut Self)>(mut self, f: F) -> Self {
+        f(&mut self);
+        self
+    }
+}
+
+impl<T> FinallyExt for Option<T> {
+    fn finally<F: FnOnce(&mut Self)>(mut self, f: F) -> Self {
+        f(&mut self);
+        self
+    }
+}
+
 /// A result type that never contains an error.
 ///
 /// This is mostly useful in generic contexts.

util/src/time.rs

@@ -1,20 +1,53 @@
-use std::time::{Duration, Instant};
+use std::time::Instant;
+
+/// A timebase.
+///
+/// This is a simple wrapper around `std::time::Instant` that provides a
+/// convenient way to get the seconds elapsed since the creation of the
+/// `Timebase` instance.
 
 #[derive(Clone, Debug)]
 pub struct Timebase(Instant);
 
 impl Default for Timebase {
+    // TODO: Implement new()?
     fn default() -> Self {
         Self(Instant::now())
     }
 }
 
 impl Timebase {
+    /// Returns the seconds elapsed since the creation of the `Timebase`
     pub fn now(&self) -> f64 {
         self.0.elapsed().as_secs_f64()
     }
+}
 
-    pub fn dur(&self, t: f64) -> Duration {
-        Duration::from_secs_f64(t)
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::thread::sleep;
+    use std::time::Duration;
+
+    #[test]
+    fn test_timebase() {
+        let timebase = Timebase::default();
+        let now = timebase.now();
+        assert!(now > 0.0);
+    }
+
+    #[test]
+    fn test_timebase_clone() {
+        let timebase = Timebase::default();
+        let timebase_clone = timebase.clone();
+        assert_eq!(timebase.0, timebase_clone.0);
+    }
+
+    #[test]
+    fn test_timebase_sleep() {
+        let timebase = Timebase::default();
+        sleep(Duration::from_secs(1));
+        let now = timebase.now();
+        assert!(now > 1.0);
     }
 }

wireguard-broker/Cargo.toml

@@ -12,44 +12,49 @@ readme = "readme.md"
 [dependencies]
 thiserror = { workspace = true }
 zerocopy = { workspace = true }
-rosenpass-secret-memory = {workspace = true}
+rosenpass-secret-memory = { workspace = true }
 
 # Privileged only
 wireguard-uapi = { workspace = true }
 
 # Socket handler only
 rosenpass-to = { workspace = true }
-tokio = { version = "1.39.1", features = ["sync", "full", "mio"] }
+tokio = { version = "1.40.0", features = ["sync", "full", "mio"] }
 anyhow = { workspace = true }
 clap = { workspace = true }
 env_logger = { workspace = true }
 log = { workspace = true }
-derive_builder = {workspace = true}
-postcard = {workspace = true}
+derive_builder = { workspace = true }
+postcard = { workspace = true }
+# Problem in CI, unknown reasons: dependency (libc) specified without providing a local path, Git repository, version, or workspace dependency to use
+# Maybe something about the combination of features and optional crates?
+rustix = { version = "0.38.37", optional = true }
+libc = { version = "0.2", optional = true }
 
 # Mio broker client
 mio = { workspace = true }
 rosenpass-util = { workspace = true }
 
 [dev-dependencies]
-rand = {workspace = true}
-procspawn = {workspace = true}
+rand = { workspace = true }
+procspawn = { workspace = true }
 
 [features]
-enable_broker_api=[]
+experiment_api = ["rustix", "libc"]
+experiment_memfd_secret = []
 
 [[bin]]
 name = "rosenpass-wireguard-broker-privileged"
 path = "src/bin/priviledged.rs"
 test = false
 doc = false
-required-features=["enable_broker_api"]
-cfg = { target_os = "linux" } 
+required-features = ["experiment_api"]
+cfg = { target_os = "linux" }
 
 [[bin]]
 name = "rosenpass-wireguard-broker-socket-handler"
 test = false
 path = "src/bin/socket_handler.rs"
 doc = false
-required-features=["enable_broker_api"]
-cfg = { target_os = "linux" } 
+required-features = ["experiment_api"]
+cfg = { target_os = "linux" }

wireguard-broker/src/api/config.rs

@@ -2,7 +2,7 @@ use crate::{SerializedBrokerConfig, WG_KEY_LEN, WG_PEER_LEN};
 use derive_builder::Builder;
 use rosenpass_secret_memory::{Public, Secret};
 
-#[derive(Builder)]
+#[derive(Builder, Debug)]
 #[builder(pattern = "mutable")]
 //TODO: Use generics for iface, add additional params
 pub struct NetworkBrokerConfig<'a> {

wireguard-broker/src/api/msgs.rs

@@ -20,8 +20,8 @@ pub struct Envelope<M: AsBytes + FromBytes> {
 #[repr(packed)]
 #[derive(AsBytes, FromBytes, FromZeroes)]
 pub struct SetPskRequest {
-    pub peer_id: [u8; 32],
     pub psk: [u8; 32],
+    pub peer_id: [u8; 32],
     pub iface_size: u8, // TODO: We should have variable length strings in lenses
     pub iface_buf: [u8; 255],
 }

wireguard-broker/src/api/server.rs

@@ -36,6 +36,7 @@ impl<Err, Inner> BrokerServer<Err, Inner>
 where
     Inner: WireGuardBroker<Error = Err>,
     msgs::SetPskError: From<Err>,
+    Err: std::fmt::Debug,
 {
     pub fn new(inner: Inner) -> Self {
         Self { inner }
@@ -56,9 +57,9 @@ where
             .ok_or(BrokerServerError::InvalidMessage)?;
         let mut res = zerocopy::Ref::<&mut [u8], Envelope<SetPskResponse>>::new(res)
             .ok_or(BrokerServerError::InvalidMessage)?;
-
-        res.payload.return_code = msgs::MsgType::SetPsk as u8;
+        res.msg_type = msgs::MsgType::SetPsk as u8;
         self.handle_set_psk(&req.payload, &mut res.payload)?;
+
         Ok(res.bytes().len())
     }
 
@@ -83,6 +84,10 @@ where
             .build()
             .unwrap();
         let r: Result<(), Err> = self.inner.borrow_mut().set_psk(config.into());
+        if let Err(e) = &r {
+            eprintln!("Error setting PSK: {e:?}"); // TODO: Use rust log
+        }
+
         let r: msgs::SetPskResult = r.map_err(|e| e.into());
         let r: msgs::SetPskResponseReturnCode = r.into();
         res.return_code = r as u8;

wireguard-broker/src/bin/priviledged.rs

@@ -27,6 +27,14 @@ pub mod linux {
     }
 
     pub fn main() -> Result<(), BrokerAppError> {
+        {
+            use rosenpass_secret_memory as SM;
+            #[cfg(feature = "experiment_memfd_secret")]
+            SM::secret_policy_try_use_memfd_secrets();
+            #[cfg(not(feature = "experiment_memfd_secret"))]
+            SM::secret_policy_use_only_malloc_secrets();
+        }
+
         let mut broker = BrokerServer::new(wg::NetlinkWireGuardBroker::new()?);
 
         let mut stdin = stdin().lock();

wireguard-broker/src/bin/socket_handler.rs

@@ -148,6 +148,14 @@ async fn listen_for_clients(queue: mpsc::Sender<BrokerRequest>, sock: UnixListen
 async fn on_accept(queue: mpsc::Sender<BrokerRequest>, mut stream: UnixStream) -> Result<()> {
     let mut req_buf = Vec::new();
 
+    {
+        use rosenpass_secret_memory as SM;
+        #[cfg(feature = "experiment_memfd_secret")]
+        SM::secret_policy_try_use_memfd_secrets();
+        #[cfg(not(feature = "experiment_memfd_secret"))]
+        SM::secret_policy_use_only_malloc_secrets();
+    }
+
     loop {
         stream.readable().await?;
 

wireguard-broker/src/brokers/mio_client.rs

@@ -1,58 +1,83 @@
-use anyhow::{bail, ensure};
+use anyhow::{bail, Context};
 use mio::Interest;
-use std::collections::VecDeque;
-use std::io::{ErrorKind, Read, Write};
-
-use crate::{SerializedBrokerConfig, WireGuardBroker, WireguardBrokerMio};
+use rosenpass_secret_memory::Secret;
+use rosenpass_to::{ops::copy_slice_least_src, To};
+use rosenpass_util::io::{IoResultKindHintExt, TryIoResultKindHintExt};
+use rosenpass_util::length_prefix_encoding::decoder::LengthPrefixDecoder;
+use rosenpass_util::length_prefix_encoding::encoder::LengthPrefixEncoder;
+use rustix::fd::AsFd;
+use std::borrow::{Borrow, BorrowMut};
 
 use crate::api::client::{
     BrokerClient, BrokerClientIo, BrokerClientPollResponseError, BrokerClientSetPskError,
 };
-use crate::api::msgs::{self, RESPONSE_MSG_BUFFER_SIZE};
+use crate::{SerializedBrokerConfig, WireGuardBroker, WireguardBrokerMio};
 
 #[derive(Debug)]
 pub struct MioBrokerClient {
     inner: BrokerClient<MioBrokerClientIo>,
+    mio_token: Option<mio::Token>,
 }
 
-const LEN_SIZE: usize = 8;
-const RECV_BUF_SIZE: usize = RESPONSE_MSG_BUFFER_SIZE;
+#[derive(Debug)]
+struct SecretBuffer<const N: usize>(pub Secret<N>);
+
+impl<const N: usize> SecretBuffer<N> {
+    fn new() -> Self {
+        Self(Secret::zero())
+    }
+}
+
+impl<const N: usize> Borrow<[u8]> for SecretBuffer<N> {
+    fn borrow(&self) -> &[u8] {
+        self.0.secret()
+    }
+}
+
+impl<const N: usize> BorrowMut<[u8]> for SecretBuffer<N> {
+    fn borrow_mut(&mut self) -> &mut [u8] {
+        self.0.secret_mut()
+    }
+}
+
+type ReadBuffer = LengthPrefixDecoder<SecretBuffer<4096>>;
+type WriteBuffer = LengthPrefixEncoder<SecretBuffer<4096>>;
 
 #[derive(Debug)]
 struct MioBrokerClientIo {
     socket: mio::net::UnixStream,
-    send_buf: VecDeque<u8>,
-    recv_state: RxState,
-    expected_state: RxState,
-    recv_buf: [u8; RECV_BUF_SIZE],
-}
-
-#[derive(Debug, Clone, Copy)]
-enum RxState {
-    //Recieving size with buffer offset
-    RxSize(usize),
-    RxBuffer(usize),
+    read_buffer: ReadBuffer,
+    write_buffer: WriteBuffer,
 }
 
 impl MioBrokerClient {
     pub fn new(socket: mio::net::UnixStream) -> Self {
+        let read_buffer = LengthPrefixDecoder::new(SecretBuffer::new());
+        let write_buffer = LengthPrefixEncoder::from_buffer(SecretBuffer::new());
         let io = MioBrokerClientIo {
             socket,
-            send_buf: VecDeque::new(),
-            recv_state: RxState::RxSize(0),
-            recv_buf: [0u8; RECV_BUF_SIZE],
-            expected_state: RxState::RxSize(LEN_SIZE),
+            read_buffer,
+            write_buffer,
         };
         let inner = BrokerClient::new(io);
-        Self { inner }
+        Self {
+            inner,
+            mio_token: None,
+        }
     }
 
-    fn poll(&mut self) -> anyhow::Result<Option<msgs::SetPskResult>> {
+    fn poll(&mut self) -> anyhow::Result<()> {
         self.inner.io_mut().flush()?;
 
         // This sucks
-        match self.inner.poll_response() {
-            Ok(res) => Ok(res),
+        let res = self.inner.poll_response();
+        match res {
+            Ok(None) => Ok(()),
+            Ok(Some(Ok(()))) => Ok(()),
+            Ok(Some(Err(e))) => {
+                log::warn!("Error from PSK broker: {e:?}");
+                Ok(())
+            }
             Err(BrokerClientPollResponseError::IoError(e)) => Err(e),
             Err(BrokerClientPollResponseError::InvalidMessage) => bail!("Invalid message"),
         }
@@ -83,6 +108,7 @@ impl WireguardBrokerMio for MioBrokerClient {
         registry: &mio::Registry,
         token: mio::Token,
     ) -> Result<(), Self::MioError> {
+        self.mio_token = Some(token);
         registry.register(
             &mut self.inner.io_mut().socket,
             token,
@@ -97,9 +123,14 @@ impl WireguardBrokerMio for MioBrokerClient {
     }
 
     fn unregister(&mut self, registry: &mio::Registry) -> Result<(), Self::MioError> {
+        self.mio_token = None;
         registry.deregister(&mut self.inner.io_mut().socket)?;
         Ok(())
     }
+
+    fn mio_token(&self) -> Option<mio::Token> {
+        self.mio_token
+    }
 }
 
 impl BrokerClientIo for MioBrokerClientIo {
@@ -107,147 +138,101 @@ impl BrokerClientIo for MioBrokerClientIo {
     type RecvError = anyhow::Error;
 
     fn send_msg(&mut self, buf: &[u8]) -> Result<(), Self::SendError> {
-        self.flush()?;
-        self.send_or_buffer(&(buf.len() as u64).to_le_bytes())?;
-        self.send_or_buffer(buf)?;
+        // Clear write buffer (blocking write)
+        self.flush_blocking()?;
+        assert!(self.write_buffer.exhausted(), "flush_blocking() should have put the write buffer in exhausted state. Developer error!");
+
+        // Emplace new message in write buffer
+        copy_slice_least_src(buf).to(self.write_buffer.buffer_bytes_mut());
+        self.write_buffer
+            .restart_write_with_new_message(buf.len())?;
+
+        // Give the write buffer a chance to clear
         self.flush()?;
 
         Ok(())
     }
 
     fn recv_msg(&mut self) -> Result<Option<&[u8]>, Self::RecvError> {
+        use std::io::ErrorKind as K;
         loop {
-            match (self.recv_state, self.expected_state) {
-                //Stale Buffer state or recieved everything
-                (RxState::RxSize(x), RxState::RxSize(y))
-                | (RxState::RxBuffer(x), RxState::RxBuffer(y))
-                    if x == y =>
-                {
-                    match self.recv_state {
-                        RxState::RxSize(s) => {
-                            let len: &[u8; LEN_SIZE] = self.recv_buf[0..s].try_into().unwrap();
-                            let len: usize = u64::from_le_bytes(*len) as usize;
+            match self
+                .read_buffer
+                .read_from_stdio(&self.socket)
+                .try_io_err_kind_hint()
+            {
+                Ok(_) => {} // Moved down in the loop
+                Err((_, Some(K::WouldBlock))) => break Ok(None),
+                Err((_, Some(K::Interrupted))) => continue,
+                Err((e, _)) => break Err(e)?,
+            }
 
-                            ensure!(
-                                len <= msgs::RESPONSE_MSG_BUFFER_SIZE,
-                                "Oversized buffer ({len}) in psk buffer response."
-                            );
-
-                            self.recv_state = RxState::RxBuffer(0);
-                            self.expected_state = RxState::RxBuffer(len);
-                            continue;
-                        }
-                        RxState::RxBuffer(s) => {
-                            self.recv_state = RxState::RxSize(0);
-                            self.expected_state = RxState::RxSize(LEN_SIZE);
-                            return Ok(Some(&self.recv_buf[0..s]));
-                        }
-                    }
-                }
-
-                //Recieve if x < y
-                (RxState::RxSize(x), RxState::RxSize(y))
-                | (RxState::RxBuffer(x), RxState::RxBuffer(y))
-                    if x < y =>
-                {
-                    let bytes = raw_recv(&self.socket, &mut self.recv_buf[x..y])?;
-
-                    if x + bytes == y {
-                        return Ok(Some(&self.recv_buf[0..y]));
-                    }
-                    //We didn't recieve everything so let's assume something went wrong
-                    self.recv_state = RxState::RxSize(0);
-                    self.expected_state = RxState::RxSize(LEN_SIZE);
-                    bail!("Invalid state");
-                }
-                _ => {
-                    //Reset states
-                    self.recv_state = RxState::RxSize(0);
-                    self.expected_state = RxState::RxSize(LEN_SIZE);
-                    bail!("Invalid state");
-                }
-            };
+            // OK case moved here to appease borrow checker
+            break Ok(self.read_buffer.message()?);
         }
     }
 }
 
 impl MioBrokerClientIo {
-    fn flush(&mut self) -> anyhow::Result<()> {
-        let (fst, snd) = self.send_buf.as_slices();
+    fn flush_blocking(&mut self) -> anyhow::Result<()> {
+        self.flush()?;
+        if self.write_buffer.exhausted() {
+            return Ok(());
+        }
 
-        let (written, res) = match raw_send(&self.socket, fst) {
-            Ok(w1) if w1 >= fst.len() => match raw_send(&self.socket, snd) {
-                Ok(w2) => (w1 + w2, Ok(())),
-                Err(e) => (w1, Err(e)),
-            },
-            Ok(w1) => (w1, Ok(())),
-            Err(e) => (0, Err(e)),
+        log::warn!("Could not flush PSK broker write buffer in non-blocking mode. Flushing in blocking mode!");
+        use rustix::io::{fcntl_getfd, fcntl_setfd, FdFlags};
+
+        // Build O_NONBLOCK
+        let o_nonblock = {
+            let v = libc::O_NONBLOCK;
+            let v = v.try_into().context(
+                "Could not cast O_NONBLOCK (`{v}`) from libc int (i32?) to rustix int (u32?)",
+            )?;
+            FdFlags::from_bits(v).context(
+                "Could not cast O_NONBLOCK (`{v}`) from rustix int to rustix::io::FdFlags",
+            )?
         };
 
-        self.send_buf.drain(..written);
+        // Determine previous and new file descriptor flags
+        let flags_orig = fcntl_getfd(self.socket.as_fd())?;
+        let mut flags_blocking = flags_orig;
+        flags_blocking.insert(o_nonblock);
 
-        self.socket.try_io(|| (&self.socket).flush())?;
+        // Set file descriptor flags
+        fcntl_setfd(self.socket.as_fd(), flags_blocking)?;
 
-        res
+        // Blocking write
+        let res = loop {
+            if self.write_buffer.exhausted() {
+                break Ok(());
+            }
+
+            match self.flush() {
+                Ok(_) => {}
+                Err(e) => break Err(e),
+            }
+        };
+
+        // Restore file descriptor flags
+        fcntl_setfd(self.socket.as_fd(), flags_orig)?;
+
+        Ok(res?)
     }
 
-    fn send_or_buffer(&mut self, buf: &[u8]) -> anyhow::Result<()> {
-        let mut off = 0;
-
-        if self.send_buf.is_empty() {
-            off += raw_send(&self.socket, buf)?;
+    fn flush(&mut self) -> std::io::Result<()> {
+        use std::io::ErrorKind as K;
+        loop {
+            match self
+                .write_buffer
+                .write_to_stdio(&self.socket)
+                .io_err_kind_hint()
+            {
+                Ok(_) => break Ok(()),
+                Err((_, K::WouldBlock)) => break Ok(()),
+                Err((_, K::Interrupted)) => continue,
+                Err((e, _)) => return Err(e)?,
+            }
         }
-
-        self.send_buf.extend(buf[off..].iter());
-
-        Ok(())
     }
 }
-
-fn raw_send(mut socket: &mio::net::UnixStream, data: &[u8]) -> anyhow::Result<usize> {
-    let mut off = 0;
-
-    socket.try_io(|| {
-        loop {
-            if off == data.len() {
-                return Ok(());
-            }
-            match socket.write(&data[off..]) {
-                Ok(n) => {
-                    off += n;
-                }
-                Err(e) if e.kind() == ErrorKind::Interrupted => {
-                    // pass – retry
-                }
-                Err(e) if off > 0 || e.kind() == ErrorKind::WouldBlock => return Ok(()),
-                Err(e) => return Err(e),
-            }
-        }
-    })?;
-
-    Ok(off)
-}
-
-fn raw_recv(mut socket: &mio::net::UnixStream, out: &mut [u8]) -> anyhow::Result<usize> {
-    let mut off = 0;
-
-    socket.try_io(|| {
-        loop {
-            if off == out.len() {
-                return Ok(());
-            }
-            match socket.read(&mut out[off..]) {
-                Ok(n) => {
-                    off += n;
-                }
-                Err(e) if e.kind() == ErrorKind::Interrupted => {
-                    // pass – retry
-                }
-                Err(e) if off > 0 || e.kind() == ErrorKind::WouldBlock => return Ok(()),
-                Err(e) => return Err(e),
-            }
-        }
-    })?;
-
-    Ok(off)
-}

wireguard-broker/src/brokers/mod.rs

@@ -1,6 +1,6 @@
-#[cfg(feature = "enable_broker_api")]
+#[cfg(feature = "experiment_api")]
 pub mod mio_client;
-#[cfg(all(feature = "enable_broker_api", target_os = "linux"))]
+#[cfg(all(feature = "experiment_api", target_os = "linux"))]
 pub mod netlink;
 
 pub mod native_unix;

wireguard-broker/src/brokers/native_unix.rs

@@ -16,7 +16,9 @@ const MAX_B64_KEY_SIZE: usize = WG_KEY_LEN * 5 / 3;
 const MAX_B64_PEER_ID_SIZE: usize = WG_PEER_LEN * 5 / 3;
 
 #[derive(Debug)]
-pub struct NativeUnixBroker {}
+pub struct NativeUnixBroker {
+    mio_token: Option<mio::Token>,
+}
 
 impl Default for NativeUnixBroker {
     fn default() -> Self {
@@ -26,7 +28,7 @@ impl Default for NativeUnixBroker {
 
 impl NativeUnixBroker {
     pub fn new() -> Self {
-        Self {}
+        Self { mio_token: None }
     }
 }
 
@@ -88,8 +90,9 @@ impl WireguardBrokerMio for NativeUnixBroker {
     fn register(
         &mut self,
         _registry: &mio::Registry,
-        _token: mio::Token,
+        token: mio::Token,
     ) -> Result<(), Self::MioError> {
+        self.mio_token = Some(token);
         Ok(())
     }
 
@@ -98,8 +101,13 @@ impl WireguardBrokerMio for NativeUnixBroker {
     }
 
     fn unregister(&mut self, _registry: &mio::Registry) -> Result<(), Self::MioError> {
+        self.mio_token = None;
         Ok(())
     }
+
+    fn mio_token(&self) -> Option<mio::Token> {
+        self.mio_token
+    }
 }
 
 #[derive(Debug, Builder)]

wireguard-broker/src/brokers/netlink.rs

@@ -79,6 +79,7 @@ impl WireGuardBroker for NetlinkWireGuardBroker {
     fn set_psk(&mut self, config: SerializedBrokerConfig) -> Result<(), Self::Error> {
         let config: NetworkBrokerConfig = config
             .try_into()
+            // TODO: I think this is the wrong error
             .map_err(|_e| SetPskError::NoSuchInterface)?;
         // Ensure that the peer exists by querying the device configuration
         // TODO: Use InvalidInterfaceError

wireguard-broker/src/lib.rs

@@ -28,13 +28,15 @@ pub trait WireguardBrokerMio: WireGuardBroker {
         registry: &mio::Registry,
         token: mio::Token,
     ) -> Result<(), Self::MioError>;
+    fn mio_token(&self) -> Option<mio::Token>;
+
     /// Run after a mio::poll operation
     fn process_poll(&mut self) -> Result<(), Self::MioError>;
 
     fn unregister(&mut self, registry: &mio::Registry) -> Result<(), Self::MioError>;
 }
 
-#[cfg(feature = "enable_broker_api")]
+#[cfg(feature = "experiment_api")]
 pub mod api;
 
 pub mod brokers;

wireguard-broker/tests/integration.rs

@@ -1,141 +0,0 @@
-#[cfg(feature = "enable_broker_api")]
-#[cfg(test)]
-mod integration_tests {
-
-    use rand::Rng;
-    use rosenpass_secret_memory::{Public, Secret};
-    use rosenpass_wireguard_broker::api::msgs::{
-        SetPskError, REQUEST_MSG_BUFFER_SIZE, RESPONSE_MSG_BUFFER_SIZE,
-    };
-    use rosenpass_wireguard_broker::api::server::{BrokerServer, BrokerServerError};
-    use rosenpass_wireguard_broker::brokers::mio_client::MioBrokerClient;
-    use rosenpass_wireguard_broker::WG_KEY_LEN;
-    use rosenpass_wireguard_broker::WG_PEER_LEN;
-    use rosenpass_wireguard_broker::{SerializedBrokerConfig, WireGuardBroker};
-    use std::io::Read;
-    use std::sync::{Arc, Mutex};
-
-    #[derive(Default, Debug)]
-    struct MockServerBrokerInner {
-        psk: Option<Secret<WG_KEY_LEN>>,
-        peer_id: Option<Public<WG_PEER_LEN>>,
-        interface: Option<String>,
-    }
-
-    #[derive(Debug)]
-    struct MockServerBroker {
-        inner: Arc<Mutex<MockServerBrokerInner>>,
-    }
-
-    impl MockServerBroker {
-        fn new(inner: Arc<Mutex<MockServerBrokerInner>>) -> Self {
-            Self { inner }
-        }
-    }
-
-    impl WireGuardBroker for MockServerBroker {
-        type Error = SetPskError;
-
-        #[allow(clippy::clone_on_copy)]
-        fn set_psk(&mut self, config: SerializedBrokerConfig) -> Result<(), Self::Error> {
-            loop {
-                let mut lock = self.inner.try_lock();
-
-                if let Ok(ref mut mutex) = lock {
-                    **mutex = MockServerBrokerInner {
-                        psk: Some(config.psk.clone()),
-                        peer_id: Some(config.peer_id.clone()),
-                        interface: Some(std::str::from_utf8(config.interface).unwrap().to_string()),
-                    };
-                    break;
-                }
-            }
-            Ok(())
-        }
-    }
-
-    procspawn::enable_test_support!();
-
-    #[test]
-    fn test_psk_exchanges() {
-        const TEST_RUNS: usize = 100;
-
-        use rosenpass_secret_memory::test_spawn_process_provided_policies;
-
-        test_spawn_process_provided_policies!({
-            let server_broker_inner = Arc::new(Mutex::new(MockServerBrokerInner::default()));
-            // Create a mock BrokerServer
-            let server_broker = MockServerBroker::new(server_broker_inner.clone());
-
-            let mut server = BrokerServer::<SetPskError, MockServerBroker>::new(server_broker);
-
-            let (client_socket, mut server_socket) = mio::net::UnixStream::pair().unwrap();
-
-            // Spawn a new thread to connect to the unix socket
-            let handle = std::thread::spawn(move || {
-                for _ in 0..TEST_RUNS {
-                    // Wait for 8 bytes of length to come in
-                    let mut length_buffer = [0; 8];
-
-                    while let Err(_err) = server_socket.read_exact(&mut length_buffer) {}
-
-                    let length = u64::from_le_bytes(length_buffer) as usize;
-
-                    // Read the amount of length bytes into a buffer
-                    let mut data_buffer = [0; REQUEST_MSG_BUFFER_SIZE];
-                    while let Err(_err) = server_socket.read_exact(&mut data_buffer[0..length]) {}
-
-                    let mut response = [0; RESPONSE_MSG_BUFFER_SIZE];
-                    server.handle_message(&data_buffer[0..length], &mut response)?;
-                }
-                Ok::<(), BrokerServerError>(())
-            });
-
-            // Create a MioBrokerClient and send a psk
-            let mut client = MioBrokerClient::new(client_socket);
-
-            for _ in 0..TEST_RUNS {
-                //Create psk of random 32 bytes
-                let psk = Secret::random();
-                let peer_id = Public::random();
-                let interface = "test";
-                let config = SerializedBrokerConfig {
-                    psk: &psk,
-                    peer_id: &peer_id,
-                    interface: interface.as_bytes(),
-                    additional_params: &[],
-                };
-                client.set_psk(config).unwrap();
-
-                //Sleep for a while to allow the server to process the message
-                std::thread::sleep(std::time::Duration::from_millis(
-                    rand::thread_rng().gen_range(100..500),
-                ));
-
-                let psk = psk.secret().to_owned();
-
-                loop {
-                    let mut lock = server_broker_inner.try_lock();
-
-                    if let Ok(ref mut inner) = lock {
-                        // Check if the psk is received by the server
-                        let received_psk = &inner.psk;
-                        assert_eq!(
-                            received_psk.as_ref().map(|psk| psk.secret().to_owned()),
-                            Some(psk)
-                        );
-
-                        let recieved_peer_id = inner.peer_id;
-                        assert_eq!(recieved_peer_id, Some(peer_id));
-
-                        let target_interface = &inner.interface;
-                        assert_eq!(target_interface.as_deref(), Some(interface));
-
-                        break;
-                    }
-                }
-            }
-            handle.join().unwrap().unwrap();
-        });
-    }
-}