Execute a file instead of code fragments

.github/workflows/tests.yml

@@ -20,7 +20,7 @@ jobs:
     container: quay.io/pypa/manylinux_2_28_x86_64  # Need a relatively modern platform for the actions to work
     strategy:
       matrix:
-        python-version: [cp38-cp38, cp39-cp39, cp310-cp310, cp311-cp311]
+        python-version: [cp38-cp38, cp39-cp39, cp310-cp310, cp311-cp311, cp312-cp312]
     steps:
       - uses: actions/checkout@v4
       - name: Set up Python
@@ -50,12 +50,14 @@ jobs:
           - "python:3.9-alpine"
           - "python:3.10-alpine"
           - "python:3.11-alpine"
+          - "python:3.12-alpine"
     steps:
       - name: Install packages
         # gcc and musl-dev needed for compiling the package
         # git needed for checkout
         # patchelf needed for auditwheel
-        run: apk add gcc musl-dev git patchelf
+        # linux-headers needed for compiling injection.so
+        run: apk add gcc musl-dev git patchelf linux-headers
       - uses: actions/checkout@v4
       - name: Install dependencies
         run: pip install tox==4.16.0 tox-gh==1.3.2 auditwheel==6.0.0
@@ -74,7 +76,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: [ "3.8", "3.9", "3.10", "3.11" ]
+        python-version: [ "3.8", "3.9", "3.10", "3.11", "3.12" ]
         arch: [ x86, x64 ]
     steps:
       - uses: actions/checkout@v4
@@ -100,7 +102,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: [ "3.8", "3.9", "3.10", "3.11" ]
+        python-version: [ "3.8", "3.9", "3.10", "3.11", "3.12" ]
     steps:
       - uses: actions/checkout@v4
       - name: Set up Python

hypno/__init__.py

@@ -1 +1 @@
-from .api import inject_py, CodeTooLongException
+from .api import inject_py, inject_py_script, CodeTooLongException

hypno/__main__.py

@@ -1,19 +1,24 @@
 from argparse import Namespace, ArgumentParser
 from typing import Optional, List
+from pathlib import Path
 
-from hypno import inject_py
+from hypno import inject_py, inject_py_script
 
 
 def parse_args(args: Optional[List[str]]) -> Namespace:
     parser = ArgumentParser(description='Inject python code into a running python process.')
     parser.add_argument('pid', type=int, help='pid of the process to inject code into')
-    parser.add_argument('python_code', type=str.encode, help='python code to inject')
+    parser.add_argument('--script', action='store_true', help='Inject a script instead of code', default=False)
+    parser.add_argument('python_code', type=str, help='python code to inject')
     return parser.parse_args(args)
 
 
 def main(args: Optional[List[str]] = None) -> None:
     parsed_args = parse_args(args)
-    inject_py(parsed_args.pid, parsed_args.python_code)
+    if parsed_args.script:
+        inject_py_script(parsed_args.pid, Path(parsed_args.python_code))
+    else:
+        inject_py(parsed_args.pid, parsed_args.python_code.encode())
 
 
 if __name__ == '__main__':

hypno/api.py

@@ -8,6 +8,7 @@
 
 INJECTION_LIB_PATH = Path(find_spec('.injection', __package__).origin)
 MAGIC = b'--- hypno code start ---'
+PATH_SENTINEL = b'--- hypno script path start ---'
 WINDOWS = sys.platform == 'win32'
 
 
@@ -49,3 +50,34 @@ def inject_py(pid: int, python_code: AnyStr, permissions=0o644) -> None:
     finally:
         if path is not None and path.exists():
             path.unlink()
+
+def inject_py_script(pid: int, python_script: Path, permissions=0o644) -> None:
+    """
+    :param pid: PID of target python process
+    :param python_script: Path to python script to inject to the target process.
+    :param permissions: Permissions of the generated shared library file that will be injected to the target process.
+                        Make sure the file is readable from the target process. By default, all users can read the file.
+    """
+
+    script_path_null_terminated = str(python_script).encode() + b'\0'
+
+    lib = INJECTION_LIB_PATH.read_bytes()
+    magic_addr = lib.find(PATH_SENTINEL)
+    path_str_addr = magic_addr - 1
+
+    patched_lib = bytearray(lib)
+    patched_lib[path_str_addr:(path_str_addr + len(script_path_null_terminated))] = script_path_null_terminated
+
+    path = None
+    try:
+        # delete=False because can't delete a loaded shared library on Windows
+        with NamedTemporaryFile(prefix='hypno', suffix=INJECTION_LIB_PATH.suffix, delete=False) as temp:
+            path = Path(temp.name)
+            temp.write(patched_lib)
+        print(f"Injecting {path} into {pid}")
+        path.chmod(permissions)
+        inject(pid, str(temp.name), uninject=True)
+    finally:
+        pass
+        # if path is not None and path.exists():
+            # path.unlink()

hypno/injection.c

@@ -1,27 +1,43 @@
 #include <Python.h>
 #include <errno.h>
 
+#if defined(_WIN32) || defined(_WIN64)
+#include <windows.h>
+#define MAX_FILE_PATH MAX_PATH
+#elif defined(__APPLE__)
+#include <sys/syslimits.h>
+#define MAX_FILE_PATH PATH_MAX
+#elif defined(__linux__)
+#include <linux/limits.h>
+#define MAX_FILE_PATH PATH_MAX
+#endif
+
 #define MAX_PYTHON_CODE_SIZE 60500
 #define STR_EXPAND(tok) #tok
 #define STR(tok) STR_EXPAND(tok)
 
 PyMODINIT_FUNC PyInit_injection(void) {return NULL;}
 
 volatile char PYTHON_CODE[MAX_PYTHON_CODE_SIZE + 1] = "\0--- hypno code start ---" STR(MAX_PYTHON_CODE_SIZE);
+volatile char PYTHON_SCRIPT_PATH[MAX_FILE_PATH] = "\0--- hypno script path start ---" STR(MAX_FILE_PATH);
 
 void run_python_code() {
-    if (PYTHON_CODE[0]) {
-        int saved_errno = errno;
-        PyGILState_STATE gstate = PyGILState_Ensure();
+    int saved_errno = errno;
+    PyGILState_STATE gstate = PyGILState_Ensure();
+    if (PYTHON_CODE[0] != '\0') {
         PyRun_SimpleString(PYTHON_CODE);
-        PyGILState_Release(gstate);
-        errno = saved_errno;
+    } else if (PYTHON_SCRIPT_PATH[0] != '\0') {
+        int fp = fopen(PYTHON_SCRIPT_PATH, "r");
+        if (fp) {
+            PyRun_SimpleFile(fp, PYTHON_SCRIPT_PATH);
+            fclose(fp);
+        }
     }
+    PyGILState_Release(gstate);
+    errno = saved_errno;
 }
 
 #ifdef _WIN32
-    #include <windows.h>
-
     BOOL WINAPI DllMain(HINSTANCE hinstDLL, DWORD fdwReason, LPVOID lpReserved) {
         switch( fdwReason ) {
             case DLL_PROCESS_ATTACH:

tests/test_hypno.py

@@ -3,8 +3,9 @@
 from pathlib import Path
 import pytest
 from pytest import mark, fixture
+from tempfile import NamedTemporaryFile
 
-from hypno import inject_py, CodeTooLongException
+from hypno import inject_py, inject_py_script, CodeTooLongException
 
 WHILE_TRUE_SCRIPT = Path(__file__).parent.resolve() / 'while_true.py'
 PROCESS_WAIT_TIMEOUT = 1.5
@@ -32,6 +33,16 @@ def process(process_loop_output, process_end_output):
     yield process
     process.kill()
 
+def write_code_to_temp_file(code: bytes, use_thread: bool) -> Path:
+    with NamedTemporaryFile(delete=False) as temp:
+        if use_thread:
+            temp.write(b"from threading import Thread\n")
+            temp.write(b"def x(): " + code + b"\n")
+            temp.write(b"\nThread(target=x).start()\n")
+        else:
+            temp.write(code)
+        temp.flush()
+        return Path(temp.name)
 
 def start_in_thread_code(code: bytes, use_thread: bool) -> bytes:
     if not use_thread:
@@ -60,6 +71,27 @@ def test_hypno(process: Popen, times: int, thread: bool, process_loop_output: st
     assert stdout == data * (times - 1), stdout
 
 
+@mark.parametrize('times', [0, 1, 2, 3])
+@mark.parametrize('thread', [True, False])
+def test_hypno_script(process: Popen, times: int, thread: bool, process_loop_output: str, process_end_output: str):
+    if thread and (sys.platform == "win32" or (sys.platform == "darwin" and sys.version_info[:2] == (3, 8))):
+        pytest.xfail("Starting a thread from injection makes inject() never return on windows, "
+                    "and output an exception on macos in python 3.8")
+    data = b'test_data_woohoo'
+    for _ in range(times - 1):
+        tmp_file = write_code_to_temp_file(b'print("' + data + b'", end="");', thread)
+        inject_py_script(process.pid, tmp_file)
+        # Making sure the process is still working
+        process.stderr.read(len(process_loop_output))
+    inject_py(process.pid, start_in_thread_code(b'__import__("__main__").should_exit = True', thread))
+    assert process.wait(PROCESS_WAIT_TIMEOUT) == 0
+
+    stdout = process.stdout.read()
+    stderr = process.stderr.read()
+    assert stderr.endswith(process_end_output.encode()), stderr
+    assert stdout == data * (times - 1), stdout
+
+
 def test_hypno_with_too_long_code():
     code = b'^' * 100000
     with pytest.raises(CodeTooLongException):