Test for memory leaks of secrets

crates/memory-testing/.gitignore

@@ -0,0 +1 @@
+output

crates/memory-testing/Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "memory-testing"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+bitwarden-crypto = { path = "../bitwarden-crypto", version = "=0.1.0" }

crates/memory-testing/Dockerfile

@@ -0,0 +1,26 @@
+###############################################
+#                 Build stage                 #
+###############################################
+FROM rust:1.76 AS build
+
+# Copy required project files
+COPY . /app
+
+# Build project
+WORKDIR /app
+RUN ls -la /
+RUN cargo build -p memory-testing
+# RUN cargo build -p memory-testing --release
+
+###############################################
+#                  App stage                  #
+###############################################
+FROM debian:bookworm-slim
+
+RUN apt-get update && apt-get install -y --no-install-recommends python3 gdb && apt-get clean && rm -rf /var/lib/apt/lists/*
+
+# Copy built project from the build stage
+COPY --from=build /app/target/debug/memory-testing .
+COPY --from=build /app/crates/memory-testing/capture_dumps.py .
+
+CMD [ "python3", "/capture_dumps.py" ]

crates/memory-testing/Dockerfile.dockerignore

@@ -0,0 +1,4 @@
+*
+!crates/*
+!Cargo.toml
+!Cargo.lock

crates/memory-testing/capture_dumps.py

@@ -0,0 +1,47 @@
+from os import remove
+from shutil import copy2
+from subprocess import Popen, run, PIPE, STDOUT
+from time import sleep
+
+
+def read_file_to_byte_array(file_path):
+    with open(file_path, "rb") as file:
+        byte_array = bytearray(file.read())
+    return byte_array
+
+
+def dump_process_to_bytearray(pid, output):
+    run(["gcore", "-a", str(pid)], capture_output=True, check=True)
+    core_file = "core." + str(pid)
+    core = read_file_to_byte_array(core_file)
+    copy2(core_file, output)
+    remove(core_file)
+    return core
+
+
+print("Memory dump capture script started")
+
+proc = Popen("./memory-testing", stdout=PIPE, stderr=STDOUT, stdin=PIPE, text=True)
+print("Started memory testing process with PID:", proc.pid)
+
+# Wait a bit for it to process
+sleep(5)
+
+# Dump the process before the variables are freed
+initial_core = dump_process_to_bytearray(proc.pid, "/output/initial_dump.bin")
+print("Initial core dump file size:", len(initial_core))
+
+proc.stdin.write(".")
+proc.stdin.flush()
+
+# Wait a bit for it to process
+sleep(5)
+
+# Dump the process after the variables are freed
+final_core = dump_process_to_bytearray(proc.pid, "/output/final_dump.bin")
+print("Final core dump file size:", len(final_core))
+
+# Wait for the process to finish and print the output
+stdout_data, _ = proc.communicate(input=".")
+print("STDOUT:", repr(stdout_data))
+print("Return code:", proc.wait())

crates/memory-testing/run_test.sh

@@ -0,0 +1,16 @@
+# Move to the root of the repository
+cd "$(dirname "$0")"
+cd ../../
+
+OUTPUT_DIR="./crates/memory-testing/output"
+
+mkdir -p $OUTPUT_DIR
+rm $OUTPUT_DIR/*
+
+docker build -f crates/memory-testing/Dockerfile -t bitwarden/memory-testing .
+docker run --rm -it -v $OUTPUT_DIR:/output bitwarden/memory-testing 
+
+xxd $OUTPUT_DIR/initial_dump.bin $OUTPUT_DIR/initial_dump.hex
+xxd $OUTPUT_DIR/final_dump.bin $OUTPUT_DIR/final_dump.hex
+
+python3 ./crates/memory-testing/test.py $OUTPUT_DIR

crates/memory-testing/src/main.rs

@@ -0,0 +1,41 @@
+use std::{io::Read, str::FromStr};
+
+use bitwarden_crypto::{AsymmetricCryptoKey, SymmetricCryptoKey};
+
+fn main() {
+    let now = std::time::Instant::now();
+
+    let mut test_string = String::new();
+    test_string.push_str("THIS IS USED TO CHECK THAT ");
+    test_string.push_str("THE MEMORY IS DUMPED CORRECTLY");
+
+    // In HEX:
+    // KEY: 15f8 5554 ff1f 9852 1963 55a6 46cc cf99 1995 0b15 cd59 5709 7df3 eb6e 4cb0 4cfb
+    // MAC: 4136 481f 8581 93f8 3f6c 5468 b361 7acf 7dfb a3db 2a32 5aa3 3017 d885 e5a3 1085
+    let symm_key = SymmetricCryptoKey::from_str(
+        "FfhVVP8fmFIZY1WmRszPmRmVCxXNWVcJffPrbkywTPtBNkgfhYGT+D9sVGizYXrPffuj2yoyWqMwF9iF5aMQhQ==",
+    )
+    .unwrap();
+
+    let symm_key_vec = symm_key.to_vec();
+
+    // Make a memory dump before the variables are freed
+    println!("Waiting for initial dump at {:?} ...", now.elapsed());
+    std::io::stdin().read_exact(&mut [1u8]).unwrap();
+    println!("Dumped at {:?}!", now.elapsed());
+
+    // Use all the variables so the compiler doesn't decide to remove them
+    println!("{test_string} {symm_key:?} {symm_key_vec:?}");
+
+    drop(test_string); // Note that this won't clear anything from the memory
+
+    drop(symm_key);
+    drop(symm_key_vec);
+
+    // After the variables are dropped, we want to make another dump
+    println!("Waiting for final dump at {:?} ...", now.elapsed());
+    std::io::stdin().read_exact(&mut [1u8]).unwrap();
+    println!("Dumped at {:?}!", now.elapsed());
+
+    println!("Done!")
+}

crates/memory-testing/test.py

@@ -0,0 +1,74 @@
+from sys import argv
+from typing import *
+
+
+def find_subarrays(needle: bytearray, haystack: bytearray) -> List[int]:
+    needle_len, haystack_len = len(needle), len(haystack)
+    subarrays = []
+
+    if needle_len == 0 or haystack_len == 0 or needle_len > haystack_len:
+        return []
+
+    for i in range(haystack_len - needle_len + 1):
+        if haystack[i : i + needle_len] == needle:
+            subarrays.append(i)
+
+    return subarrays
+
+
+# Check that I implemented this correctly lol
+assert find_subarrays([1, 2, 3], [1, 2, 3, 4, 5]) == [0]
+assert find_subarrays([1, 2, 3], [1, 2, 3, 4, 1, 2, 3, 5]) == [0, 4]
+assert find_subarrays([1, 2, 3], [1, 2, 3]) == [0]
+assert find_subarrays([1, 2, 3], [1, 2, 4, 3, 5]) == []
+
+
+def find_subarrays_batch(needles: List[Tuple[bytearray, str]], haystack: bytearray):
+    for needle, name in needles:
+        print(f"Subarrays of {name}:", find_subarrays(needle, haystack))
+
+
+def read_file_to_byte_array(file_path: str) -> bytearray:
+    with open(file_path, "rb") as file:
+        return bytearray(file.read())
+
+
+TEST_STRING = b"THIS IS USED TO CHECK THAT THE MEMORY IS DUMPED CORRECTLY"
+SYMMETRIC_KEY = bytearray.fromhex(
+    "15f8 5554 ff1f 9852 1963 55a6 46cc cf99 1995 0b15 cd59 5709 7df3 eb6e 4cb0 4cfb"
+)
+SYMMETRIC_MAC = bytearray.fromhex(
+    "4136 481f 8581 93f8 3f6c 5468 b361 7acf 7dfb a3db 2a32 5aa3 3017 d885 e5a3 1085"
+)
+TEST_VEC = bytearray([1, 77, 43, 124, 192, 250, 0, 78, 92])
+
+output_dir = argv[1]
+print("Memory testing script started in", output_dir)
+
+print("\n-------------------------------\n")
+
+initial_core = read_file_to_byte_array(output_dir + "/initial_dump.bin")
+print("Processing initial core dump")
+find_subarrays_batch(
+    [
+        (SYMMETRIC_KEY, "key"),
+        (SYMMETRIC_MAC, "mac"),
+        (TEST_STRING, "hello world"),
+        (TEST_VEC, "test vec"),
+    ],
+    initial_core,
+)
+
+print("\n-------------------------------\n")
+
+final_core = read_file_to_byte_array(output_dir + "/final_dump.bin")
+print("Processing final core dump")
+find_subarrays_batch(
+    [
+        (SYMMETRIC_KEY, "key"),
+        (SYMMETRIC_MAC, "mac"),
+        (TEST_STRING, "hello world"),
+        (TEST_VEC, "test vec"),
+    ],
+    final_core,
+)