Test for memory leaks of secrets

crates/memory-testing/.gitignore

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

crates/memory-testing/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "memory-testing"
+version = "0.1.0"
+edition = "2021"
+publish = false
+
+[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", "./memory-testing", "/output" ]

crates/memory-testing/Dockerfile.dockerignore

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

crates/memory-testing/analyze_dumps.py

@@ -0,0 +1,116 @@
+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"
+)
+
+# ---------------------------------------------------------------------------
+
+if len(argv) < 2:
+    print("Usage: python3 test.py <output_dir>")
+    exit(1)
+
+output_dir = argv[1]
+print("Memory testing script started in", output_dir)
+
+print("------------- Processing initial core dump -------------")
+
+initial_core = read_file_to_byte_array(output_dir + "/initial_dump.bin")
+
+key_initial_matches = find_subarrays(SYMMETRIC_KEY, initial_core)
+mac_initial_matches = find_subarrays(SYMMETRIC_MAC, initial_core)
+test_initial_matches = find_subarrays(TEST_STRING, initial_core)
+
+print("-------------- Processing final core dump --------------")
+
+final_core = read_file_to_byte_array(output_dir + "/final_dump.bin")
+
+key_final_matches = find_subarrays(SYMMETRIC_KEY, final_core)
+mac_final_matches = find_subarrays(SYMMETRIC_MAC, final_core)
+test_final_matches = find_subarrays(TEST_STRING, final_core)
+
+
+debug = True
+if debug:
+    print("-------------- Printing matches for debug --------------")
+    print("Initial matches")
+    print("    Key:", key_initial_matches)
+    print("    MAC:", mac_initial_matches)
+    print("    Test:", test_initial_matches)
+    print("Final matches")
+    print("    Key:", key_final_matches)
+    print("    MAC:", mac_final_matches)
+    print("    Test:", test_final_matches)
+
+print("------------------ Checking for leaks  -----------------")
+
+error = False
+
+if len(test_initial_matches) == 0:
+    print("ERROR: Test string not found in initial core dump")
+    error = True
+
+if len(test_final_matches) > len(test_initial_matches):
+    print(
+        "ERROR: Test string found more times in final core dump than in initial core dump"
+    )
+    error = True
+
+if len(key_final_matches) > 0:
+    print(
+        "ERROR: Symmetric key found in final core dump at positions:", key_final_matches
+    )
+    error = True
+
+if len(mac_final_matches) > 0:
+    print(
+        "ERROR: Symmetric MAC found in final core dump at positions:", mac_final_matches
+    )
+    error = True
+
+if error:
+    print("Memory testing script finished with errors")
+    exit(1)
+else:
+    print("Memory testing script finished successfully")
+    exit(0)

crates/memory-testing/capture_dumps.py

@@ -0,0 +1,55 @@
+from os import remove
+from shutil import copy2
+from sys import argv
+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
+
+
+if len(argv) < 3:
+    print("Usage: python3 capture_dumps.py <binary_path> <output_dir>")
+    exit(1)
+
+binary_path = argv[1]
+output_dir = argv[2]
+
+print("Memory dump capture script started")
+
+proc = Popen(binary_path, 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(1)
+
+# Dump the process before the variables are freed
+initial_core = dump_process_to_bytearray(proc.pid, output_dir + "/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(1)
+
+# Dump the process after the variables are freed
+final_core = dump_process_to_bytearray(proc.pid, output_dir + "/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,18 @@
+# Move to the root of the repository
+cd "$(dirname "$0")"
+cd ../../
+
+OUTPUT_DIR="./crates/memory-testing/output"
+
+mkdir -p $OUTPUT_DIR
+rm $OUTPUT_DIR/*
+
+if [ "$1" = "no-docker" ]; then
+    cargo build -p memory-testing
+    python3 ./crates/memory-testing/capture_dumps.py ./target/debug/memory-testing $OUTPUT_DIR
+else
+    docker build -f crates/memory-testing/Dockerfile -t bitwarden/memory-testing .
+    docker run --rm -it -v $OUTPUT_DIR:/output bitwarden/memory-testing 
+fi
+
+python3 ./crates/memory-testing/analyze_dumps.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!")
+}