tests: add tests

.github/workflows/build.yml

@@ -4,6 +4,7 @@ on:
   push:
     branches:
       - master
+      - dev
   release:
     types: ['published']
 
@@ -36,9 +37,26 @@ jobs:
           registry: ghcr.io
           username: ${{ github.actor }}
           password: ${{ github.token }}
-      - uses: docker/[email protected]
+
+      - name: Build image
+        uses: docker/[email protected]
+        with:
+          load: true
+          push: false
+          tags: pyda_tmp
+          platforms: linux/amd64
+          provenance: false
+          cache-from: type=gha
+
+      - name: Test
+        run: |
+          docker run -e PYTHONUNBUFFERED=1 --rm --workdir /opt/pyda/tests --entrypoint python3 pyda_tmp run_tests.py
+
+      - name: Push image
+        uses: docker/[email protected]
+        if: github.event_name != 'pull_request'
         with:
-          push: ${{ github.event_name != 'pull_request' }}
+          push: true
           tags: ${{ steps.meta.outputs.tags }}
           platforms: linux/amd64
           provenance: false

tests/err_hook.py

@@ -0,0 +1,32 @@
+from pyda import *
+from pwnlib.elf.elf import ELF
+from pwnlib.util.packing import u64
+import string
+import sys, time
+
+p = process()
+
+e = ELF(p.exe_path)
+e.address = p.maps[p.exe_path].base
+
+plt_map = { e.plt[x]: x for x in e.plt }
+
+counter = 0
+def lib_hook(p):
+    global counter
+    name = plt_map[p.regs.rip]
+    print(f"[thread {p.tid}] {name}")
+
+    counter += 1
+    if counter == 1000:
+        jsdkfjdsaklfadska
+
+def thread_entry(p):
+    print(f"thread_entry for {p.tid}")
+
+p.set_thread_entry(thread_entry)
+
+for x in e.plt:
+    p.hook(e.plt[x], lib_hook)
+
+p.run()

tests/err_thread_entry.py

@@ -0,0 +1,32 @@
+from pyda import *
+from pwnlib.elf.elf import ELF
+from pwnlib.util.packing import u64
+import string
+import sys, time
+
+p = process()
+
+e = ELF(p.exe_path)
+e.address = p.maps[p.exe_path].base
+
+plt_map = { e.plt[x]: x for x in e.plt }
+
+counter = 0
+def lib_hook(p):
+    name = plt_map[p.regs.rip]
+    print(f"[thread {p.tid}] {name}")
+
+def thread_entry(p):
+    global counter
+    print(f"thread_entry for {p.tid}")
+
+    counter += 1
+    if counter == 27:
+        jsdkfjdsaklfadska
+
+p.set_thread_entry(thread_entry)
+
+for x in e.plt:
+    p.hook(e.plt[x], lib_hook)
+
+p.run()

tests/run_tests.py

@@ -0,0 +1,128 @@
+import subprocess
+from dataclasses import dataclass
+from typing import Optional, Callable
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+@dataclass
+class ExpectedResult:
+    retcode: Optional[int] = None
+
+    # checker(stdout, stderr) -> bool
+    checkers: list[Callable[[bytes, bytes], bool]] = list
+
+Res = ExpectedResult
+
+def output_checker(stdout: bytes, stderr: bytes) -> bool:
+    try:
+        stdout.decode()
+        stderr.decode()
+    except:
+        return False
+
+    return True
+
+def main():
+    res = True
+
+    # thread_1000.c tests whether we can handle a large number of threads
+    # with concurrent hooks
+    res &= run_test(
+        "thread_1000.c", "../examples/ltrace_multithreaded.py",
+        ExpectedResult(
+            retcode=0,
+            checkers=[
+                output_checker,
+                lambda o, e: o.count(b"malloc") == 20000,
+                lambda o, e: o.count(b"free") == 20000,
+                lambda o, e: all((o.count(f"[thread {i}]".encode('utf-8')) == 40 for i in range(2, 1002))),
+            ]
+        )
+    )
+
+    # thread_nojoin.c tests whether we can handle a large number of threads
+    # that do not get waited on (i.e. they are not joined). Mostly
+    # we just care about the return code and termination here.
+    res &= run_test(
+        "thread_nojoin.c", "../examples/ltrace_multithreaded.py",
+        ExpectedResult(
+            retcode=0,
+            checkers=[
+                output_checker,
+            ]
+        )
+    )
+
+    # err_hook.py tests the case where a hook throws an exception
+    # NOTE: Hooks intentionally fail 'gracefully' and do not abort
+    res &= run_test(
+        "thread_1000.c", "err_hook.py",
+        ExpectedResult(
+            retcode=0,
+            checkers=[
+                output_checker,
+                lambda o, e: e.count(b"[Pyda] ERROR:") == 1,
+            ]
+        )
+    )
+
+    # err_thread_entry.py tests the case where a hook throws an exception
+    # NOTE: Hooks intentionally fail 'gracefully' and do not abort
+    res &= run_test(
+        "thread_1000.c", "err_thread_entry.py",
+        ExpectedResult(
+            retcode=0,
+            checkers=[
+                output_checker,
+                lambda o, e: e.count(b"[Pyda] ERROR:") == 1,
+            ]
+        )
+    )
+
+    if not res:
+        exit(1)
+
+def run_test(c_file, python_file, expected_result):
+    # Compile to temporary directory
+    with TemporaryDirectory() as tmpdir:
+        tmpdir = Path(tmpdir)
+        c_path = Path(c_file)
+        p_path = Path(python_file)
+
+        c_exe = tmpdir / c_path.stem
+        compile_res = subprocess.run(['gcc', '-o', c_exe, c_path], capture_output=True)
+        if compile_res.returncode != 0:
+            print(f"Failed to compile {c_file}")
+            print(compile_res.stderr)
+            raise RuntimeError("Failed to compile test")
+
+        result_str = ""
+        try:
+            result = subprocess.run(f"pyda {p_path.resolve()} -- {c_exe.resolve()}", shell=True, timeout=10, capture_output=True)
+        except subprocess.TimeoutExpired:
+            result_str += " Timeout occurred. Did the test hang?\n"
+            result = None
+
+
+        if result:
+            # Check the results
+            if expected_result.retcode is not None:
+                if result.returncode != expected_result.retcode:
+                    result_str += f"  Expected return code {expected_result.retcode}, got {result.returncode}\n"
+
+            for (i, checker) in enumerate(expected_result.checkers):
+                if not checker(result.stdout, result.stderr):
+                    result_str += f"  Checker {i} failed\n"
+
+
+        if len(result_str) > 0:
+            print(f"[FAIL] {c_file} {python_file}")
+            print(result_str)
+            return False
+        else:
+            print(f"[OK] {c_file} {python_file}")
+            return True
+
+
+if __name__ == '__main__':
+    main()

tests/thread_1000.c

@@ -0,0 +1,35 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <pthread.h>
+
+#define NTHREAD 1000
+
+void* thread(void* arg) {
+    void *allocs[10];
+    for (int i=0; i<10; i++) {
+        allocs[i] = malloc(0x100);
+    }
+    for (int i=0; i<10; i++) {
+        free(allocs[i]);
+    }
+    for (int i=0; i<10; i++) {
+        allocs[i] = malloc(0x100);
+    }
+    for (int i=0; i<10; i++) {
+        free(allocs[i]);
+    }
+}
+int main(int argc, char** argv) {
+    pthread_t threads[NTHREAD];
+    for (int i=0; i<NTHREAD; i++) {
+        pthread_create(&threads[i], 0, thread, 0);
+    }
+
+    for (int i=0; i<NTHREAD; i++) {
+        void *ret;
+        pthread_join(threads[i], &ret);
+    }
+
+    // thread(NULL);
+    return 0;
+}

tests/thread_nojoin.c

@@ -0,0 +1,30 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <pthread.h>
+
+#define NTHREAD 1000
+
+void* thread(void* arg) {
+    void *allocs[10];
+    for (int i=0; i<10; i++) {
+        allocs[i] = malloc(0x100);
+    }
+    for (int i=0; i<10; i++) {
+        free(allocs[i]);
+    }
+    for (int i=0; i<10; i++) {
+        allocs[i] = malloc(0x100);
+    }
+    for (int i=0; i<10; i++) {
+        free(allocs[i]);
+    }
+}
+int main(int argc, char** argv) {
+    pthread_t threads[NTHREAD];
+    for (int i=0; i<NTHREAD; i++) {
+        pthread_create(&threads[i], 0, thread, 0);
+    }
+
+    // thread(NULL);
+    return 0;
+}