diff --git a/tests/test_deliberate_crashes_suppress_cores.py b/tests/test_deliberate_crashes_suppress_cores.py index 1350b9542d..5496ffaf50 100644 --- a/tests/test_deliberate_crashes_suppress_cores.py +++ b/tests/test_deliberate_crashes_suppress_cores.py @@ -45,6 +45,7 @@ Three things this deliberately gets right, each of which it got wrong first: from __future__ import annotations import ast +import warnings from functools import lru_cache from pathlib import Path @@ -74,18 +75,33 @@ _CRASH_MARKERS = ("string_at(0)", "os.abort()", "_sigsegv(") # idiom and must not be flagged. _SIGNAL_DIRECTED = ("raise_signal(", "os.kill(", ".kill(") _DIRECTED_NAMES = {"raise_signal", "kill"} +# Which argument carries the signal. Checking every argument read the PID in +# `os.kill(11, signal.SIGKILL)` as SIGSEGV and failed CI over a call that cannot dump. +_SIGNAL_ARG_INDEX = {"raise_signal": 0, "kill": 1} # A signal aimed at self dumps core only for these. SIGKILL, SIGTERM and SIGINT do not, # which is why SIGKILL is the recommended way to make a child vanish. _FATAL_SIGNALS = ("SIGSEGV", "SIGABRT", "SIGBUS", "SIGILL", "SIGFPE", "SIGTRAP") +_FATAL_SIGNAL_NUMBERS = {4, 5, 6, 7, 8, 11} # SIGILL SIGTRAP SIGABRT SIGBUS SIGFPE SIGSEGV _PR_SET_DUMPABLE = 4 # Raw-text prefilter, run before any parsing. The AST is a subset of the source, so a # file whose text holds none of these cannot match and parsing it is wasted work. This -# is what keeps the check cheap: 32 of ~1050 files are actually parsed, 496ms rather -# than 9s. -_PREFILTER = _CRASH_MARKERS + _SIGNAL_DIRECTED + ("abort(",) +# is what keeps the check cheap: 32 of ~1050 files are actually parsed. Measured at +# ~1.5s for the whole suite against ~9s with every file parsed. The scope and alias +# analysis added since is what accounts for the difference from the 496ms this once +# cost; the prefilter itself still admits the same 32 files. +# Deliberately looser than `_CRASH_MARKERS`: matching `string_at(0)` exactly meant +# `string_at( 0)` was skipped before it was ever parsed. The prefilter only decides what +# to parse, so it should over-match and let the AST checks be the precise ones. +_PREFILTER = ("string_at(", "abort(", "_sigsegv(") + _SIGNAL_DIRECTED +# An aliased import carries none of the call shapes above: `from os import abort as die` +# then `die()` has no "abort(" anywhere, so the prefilter skipped the file and the alias +# resolution that would have caught it never ran. The import spelling is the one piece of +# text such a file must contain, so match on that too. Costs one more scan of files that +# import these names and nothing else. +_PREFILTER += ("import abort", "import string_at", "import raise_signal", "import kill") def _test_roots(): @@ -150,29 +166,6 @@ def _prctl_clears_dumpable(node) -> bool: return cmd_ok and isinstance(value, ast.Constant) and value.value == 0 -def _scope_suppresses(scope) -> bool: - """Whether this function (or module) clears the dumpable flag anywhere in it.""" - return any(_prctl_clears_dumpable(n) for n in ast.walk(scope)) - - -def _snippet_suppresses(snippet: str) -> bool: - """Whether a child-script snippet clears the dumpable flag. - - Snippets are Python source, so parse them and ask the same question. Fragments that - do not parse fall back to a textual check. - """ - try: - return _scope_suppresses(ast.parse(snippet)) - except SyntaxError: - return "prctl(4, 0" in snippet or "PR_SET_DUMPABLE, 0" in snippet - - -def _snippet_crashes(snippet: str) -> bool: - if any(marker in snippet for marker in _CRASH_MARKERS): - return True - return any(m in snippet for m in _SIGNAL_DIRECTED) and any(s in snippet for s in _FATAL_SIGNALS) - - def _fold(node, env): """Constant-fold a string expression. Returns (value, consumed literal ids). @@ -191,53 +184,178 @@ def _fold(node, env): return None, set() -def _snippets(tree): - """Candidate child scripts: folded module constants, then unconsumed literals.""" - env, consumed = {}, set() - for node in tree.body: - if not isinstance(node, ast.Assign) or len(node.targets) != 1: - continue - target = node.targets[0] - if not isinstance(target, ast.Name): - continue - value, ids = _fold(node.value, env) - if value is not None: - env[target.id] = value - consumed |= ids - - docstrings = _docstring_ids(tree) - out = list(env.values()) - for node in ast.walk(tree): - if not isinstance(node, ast.Constant) or not isinstance(node.value, str): - continue - if id(node) in consumed or id(node) in docstrings: - continue - out.append(node.value) - return out +def _assigned_pair(node): + """`(target, value)` for a single-name assignment, else None.""" + if isinstance(node, ast.Assign) and len(node.targets) == 1: + target, value = node.targets[0], node.value + elif isinstance(node, ast.AnnAssign) and node.value is not None: + target, value = node.target, node.value + else: + return None + return (target, value) if isinstance(target, ast.Name) else None -def _docstring_ids(tree): +def _string_env(tree): + """Name to foldable string, per scope, so same-named locals cannot collide. + + One flat map let an unrelated function's `SCRIPT = "print(1)"` overwrite a + module-level `SCRIPT` that really does crash, and the executed script then went + unchecked. Functions are seeded from the module bindings and shadow them locally. + Annotated constants and strings assembled inside a test are folded either way, + so `_SAFE = _SUPPRESS + "ctypes.string_at(0)"` is still judged as what it becomes. + """ + owner = _enclosing_scopes(tree) + module_env, scoped = {}, {} + for module_pass in (True, False): + for node in ast.walk(tree): + pair = _assigned_pair(node) + if pair is None: + continue + scope = owner.get(id(node), tree) + if (scope is tree) != module_pass: + continue + env = module_env if module_pass else scoped.setdefault(id(scope), dict(module_env)) + folded, _ = _fold(pair[1], env) + if folded is not None: + env[pair[0].id] = folded + return owner, module_env, scoped + + +def _iter_executable(scope): + """Nodes that run when `scope` runs, skipping bodies that need a separate call. + + Walking everything treated a call inside an uninvoked nested `def` as having + already run, so a helper that suppresses cores blessed a crash it never covered. + """ + for child in ast.iter_child_nodes(scope): + if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda, ast.ClassDef)): + continue + yield child + yield from _iter_executable(child) + + +def _rebound_names(scope): + """Names this scope binds itself, which therefore no longer mean the import.""" out = set() - for node in ast.walk(tree): - if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): - body = getattr(node, "body", None) - if ( - body - and isinstance(body[0], ast.Expr) - and isinstance(body[0].value, ast.Constant) - and isinstance(body[0].value.value, str) - ): - out.add(id(body[0].value)) + for node in ast.walk(scope): + pair = _assigned_pair(node) + if pair is not None: + out.add(pair[0].id) + elif isinstance(node, ast.arg): + out.add(node.arg) return out -def _is_crash_call(node) -> bool: +def _sequence_env(tree, owner): + """Name to the elements of a list/tuple it is bound to. + + Separate from `_string_env` because that one folds to a string and a command + vector does not fold. Flat by name rather than per scope: a false name collision + here can only add candidate strings to read, and reading one extra string is a + cheaper mistake than missing the script a child actually runs. + """ + out = {} + for node in ast.walk(tree): + pair = _assigned_pair(node) + if pair is None: + continue + target, value = pair + if isinstance(value, (ast.List, ast.Tuple)): + out.setdefault(target.id, []).extend(value.elts) + return out + + +def _snippets(tree): + """Strings that actually reach a child interpreter. + + A string only counts if it is passed as a call argument, directly or through a + name, a list/tuple, or a concatenation. That is what `subprocess.run([exe, "-c", + SCRIPT])` looks like. Treating every string literal as a candidate script meant an + ordinary expectation such as `assert "ctypes.string_at(0)" in text` was reported as + a deliberate crash, which would fail CI over a string nothing executes. + """ + owner, module_env, scoped = _string_env(tree) + sequences = _sequence_env(tree, owner) + out = [] + + def collect(node, env): + if isinstance(node, ast.Constant) and isinstance(node.value, str): + out.append(node.value) + elif isinstance(node, ast.JoinedStr): + # An f-string reaches the child as a script like any other. Keep its + # literal parts: the interpolations cannot be known here, and dropping the + # whole thing let `f"import os; os.abort(); print({v})"` through unread. + out.append( + "".join( + part.value + for part in node.values + if isinstance(part, ast.Constant) and isinstance(part.value, str) + ) + ) + elif isinstance(node, ast.Name): + if node.id in env: + out.append(env[node.id]) + elif node.id in sequences: + # `CMD = [sys.executable, "-c", SCRIPT]` then `subprocess.run(CMD)`. + # A command vector built once and passed by name is ordinary + # subprocess usage, and folding only strings meant the child script + # inside it was never read. + for element in sequences[node.id]: + collect(element, env) + elif isinstance(node, (ast.List, ast.Tuple, ast.Set)): + for element in node.elts: + collect(element, env) + elif isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): + folded, _ = _fold(node, env) + if folded is not None: + out.append(folded) + else: + collect(node.left, env) + collect(node.right, env) + + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + env = scoped.get(id(owner.get(id(node), tree)), module_env) + for argument in list(node.args) + [kw.value for kw in node.keywords]: + collect(argument, env) + return out + + +def _crash_aliases(tree): + """Names imported directly from a crashing module, e.g. `from os import abort`. + + Needed so an aliased call is still recognised. A bare `abort()` is only fatal if it + came from `os` or `ctypes`; Playwright's `route.abort()` keeps its receiver and is + still correctly ignored. + + Keyed by the name the call site uses and valued by the name the rules are written + against, because `from os import abort as die` binds `die` and a lookup of `die` in + `_CRASH_CALLS` finds nothing. Without the mapping an aliased import was invisible. + """ + out = {} + for node in ast.walk(tree): + if not isinstance(node, ast.ImportFrom) or node.module is None: + continue + if node.module.split(".")[0] not in ("os", "ctypes", "signal", "faulthandler"): + continue + for alias in node.names: + if alias.name in _CRASH_CALLS or alias.name in _DIRECTED_NAMES: + out[alias.asname or alias.name] = alias.name + return out + + +def _is_crash_call(node, aliases = None) -> bool: """A call that deliberately takes a core-dumping signal, written as code.""" + aliases = aliases or {} if not isinstance(node, ast.Call): return False name = _called_name(node) if name is None: return False + # `from os import abort as die` binds `die`; the rules are written against `abort`. + if isinstance(node.func, ast.Name) and name in aliases: + name = aliases[name] if name in _CRASH_CALLS: func = ast.unparse(node.func) rule = _CRASH_CALLS[name] @@ -247,13 +365,25 @@ def _is_crash_call(node) -> bool: return False if "owners" in rule: receiver = func[: -len(name)] - if not any(owner in receiver for owner in rule["owners"]): + bare = isinstance(node.func, ast.Name) and node.func.id in aliases + if not bare and not any(owner in receiver for owner in rule["owners"]): return False return True if name not in _DIRECTED_NAMES: return False - rendered = ast.unparse(node) - return any(s in rendered for s in _FATAL_SIGNALS) + index = _SIGNAL_ARG_INDEX[name] + if len(node.args) <= index: + return False + # Only the signal argument, so a PID never reads as a signal. + signal_argument = node.args[index] + rendered = ast.unparse(signal_argument) + if any(s in rendered for s in _FATAL_SIGNALS): + return True + # Numeric signals. `signal.raise_signal(11)` and `os.kill(pid, 6)` dump exactly the + # same core as the named forms, so matching only symbolic names missed them. + return ( + isinstance(signal_argument, ast.Constant) and signal_argument.value in _FATAL_SIGNAL_NUMBERS + ) def _enclosing_scopes(tree): @@ -272,6 +402,171 @@ def _enclosing_scopes(tree): return owner +def _functions_by_name(tree): + return { + node.name: node + for node in ast.walk(tree) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + } + + +def _position(node): + """Source order of a node. Line alone ties on a one-line `-c` script, where + `prctl(4, 0, ...); ctypes.string_at(0)` really does suppress the crash after it.""" + return (node.lineno, node.col_offset) + + +_AFTER_EVERYTHING = (float("inf"), 0) + + +def _clears_dumpable_before(scope, position) -> bool: + """A prctl(4, 0, ...) on this scope's own path that runs before `position`. + + Order matters. Suppression placed after the fault does nothing, so accepting it + anywhere in the scope blessed a child that still dumps. + """ + return any( + _prctl_clears_dumpable(node) and _position(node) < position + for node in _iter_executable(scope) + ) + + +def _suppressed(node, scope, functions) -> bool: + """Whether this crash call is covered, directly or by a helper it calls first.""" + position = _position(node) + if _clears_dumpable_before(scope, position): + return True + # Following one level of local helper covers `suppress_core()` then the fault, + # which is the natural shape once more than one test needs this. + for called in _iter_executable(scope): + if not isinstance(called, ast.Call) or _position(called) >= position: + continue + # Bare calls only. `obj.suppress_core()` shares its trailing name with a local + # `def suppress_core`, and crediting the local one there means an object method + # that may clear nothing at all is taken as proof the fault is covered. + if not isinstance(called.func, ast.Name): + continue + target = functions.get(called.func.id) + if target is None: + continue + # Calling an `async def` builds a coroutine and runs none of its body, so the + # prctl never happens. Only an awaited one has actually cleared dumpability. + if isinstance(target, ast.AsyncFunctionDef) and not _is_awaited(called, scope): + continue + if _clears_dumpable_before(target, _AFTER_EVERYTHING): + return True + return False + + +def _is_awaited(call, scope) -> bool: + """Whether `call` is the operand of an `await`.""" + for node in ast.walk(scope): + if isinstance(node, ast.Await) and node.value is call: + return True + return False + + +def _live_aliases(aliases, scope, rebound): + """The imported crash names still in force where this call sits. + + A scope that binds the name itself no longer means the import, so a test that + does `abort = mock` before calling it is not crashing anything. + """ + if not aliases or scope is None: + return aliases + if id(scope) not in rebound: + rebound[id(scope)] = _rebound_names(scope) + shadowed = rebound[id(scope)] + return {bound: original for bound, original in aliases.items() if bound not in shadowed} + + +def _unsuppressed_crashes(tree): + """Crash calls in this tree whose core dump is not suppressed first.""" + owner = _enclosing_scopes(tree) + functions = None + aliases = _crash_aliases(tree) + rebound = {} + out = [] + for node in ast.walk(tree): + scope = owner.get(id(node), tree) + if not _is_crash_call(node, _live_aliases(aliases, scope, rebound)): + continue + if functions is None: + functions = _functions_by_name(tree) + if not _suppressed(node, scope, functions): + out.append((node, scope)) + return out + + +def _tree_crashes(tree) -> bool: + aliases = _crash_aliases(tree) + owner = _enclosing_scopes(tree) + rebound = {} + return any( + _is_crash_call(node, _live_aliases(aliases, owner.get(id(node), tree), rebound)) + for node in ast.walk(tree) + ) + + +_NESTED_EXEC = {"exec", "eval"} +_MAX_SNIPPET_DEPTH = 5 + + +def _nested_scripts(tree): + """Source a snippet hands to exec/eval, or on to another child interpreter. + + `SCRIPT = 'exec("import os; os.abort()")'` parses cleanly and holds no crash call + of its own, so without this the crash one level down was never looked at. + """ + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + if _called_name(node) in _NESTED_EXEC: + argument = node.args[0] if node.args else None + if isinstance(argument, ast.Constant) and isinstance(argument.value, str): + yield argument.value + else: + yield from _snippets_of_call(node) + + +def _snippets_of_call(node): + """Script strings passed to a nested subprocess call, e.g. a `-c` argument.""" + for argument in list(node.args) + [kw.value for kw in node.keywords]: + if isinstance(argument, (ast.List, ast.Tuple)): + for element in argument.elts: + if isinstance(element, ast.Constant) and isinstance(element.value, str): + yield element.value + + +def _snippet_state(snippet: str, depth: int = 0): + """`(crashes, violates)` for a child script. + + The snippet is Python, so parse it and reuse the same call detector rather than + matching marker substrings. Textual matching missed aliased forms such as + `from os import abort; abort()` and any spacing the markers did not anticipate. + """ + try: + with warnings.catch_warnings(): + # Snippets are other people's source. An unrelated escape-sequence warning + # from one of them must not show up as noise in this suite's output. + warnings.simplefilter("ignore") + tree = ast.parse(snippet) + except (SyntaxError, ValueError): + crashes = any(marker in snippet for marker in _CRASH_MARKERS) or ( + any(m in snippet for m in _SIGNAL_DIRECTED) + and any(s in snippet for s in _FATAL_SIGNALS) + ) + suppressed = "prctl(4, 0" in snippet or "PR_SET_DUMPABLE, 0" in snippet + return crashes, crashes and not suppressed + crashes, violates = _tree_crashes(tree), bool(_unsuppressed_crashes(tree)) + if depth < _MAX_SNIPPET_DEPTH: + for nested in _nested_scripts(tree): + inner_crashes, inner_violates = _snippet_state(nested, depth + 1) + crashes = crashes or inner_crashes + violates = violates or inner_violates + return crashes, violates + + @lru_cache(maxsize = None) def _analyze(path): """`(crashes_on_purpose, unsuppressed_reasons)` for one file. @@ -291,23 +586,15 @@ def _analyze(path): crashes, out = False, [] for snippet in _snippets(tree): - if not _snippet_crashes(snippet): - continue - crashes = True - if not _snippet_suppresses(snippet): + snippet_crashes, violates = _snippet_state(snippet) + crashes = crashes or snippet_crashes + if violates: out.append("a child script that crashes on purpose") - owner = None - for node in ast.walk(tree): - if not _is_crash_call(node): - continue - crashes = True - if owner is None: - owner = _enclosing_scopes(tree) - scope = owner.get(id(node), tree) - if not _scope_suppresses(scope): - where = getattr(scope, "name", "module level") - out.append(f"{ast.unparse(node)} at line {node.lineno} (in {where})") + for node, scope in _unsuppressed_crashes(tree): + where = getattr(scope, "name", "module level") + out.append(f"{ast.unparse(node)} at line {node.lineno} (in {where})") + crashes = crashes or _tree_crashes(tree) return crashes, tuple(dict.fromkeys(out)) @@ -362,16 +649,112 @@ _FIXTURES = { True, # the first function must not bless the second ), "helper_then_a_naked_script": ( - 'SUPPRESS = "ctypes.CDLL(None).prctl(4, 0, 0, 0, 0)\\n"\n' + "import subprocess, sys\n" + 'SUPPRESS = "import ctypes\\nctypes.CDLL(None).prctl(4, 0, 0, 0, 0)\\n"\n' 'SAFE = SUPPRESS + "ctypes.string_at(0)\\n"\n' - 'NAKED = "ctypes.string_at(0)\\n"\n', + 'NAKED = "import ctypes\\nctypes.string_at(0)\\n"\n' + 'subprocess.run([sys.executable, "-c", SAFE])\n' + 'subprocess.run([sys.executable, "-c", NAKED])\n', True, # SAFE is fine, NAKED is not, and the file must not pass on SAFE ), "concatenated_script_is_folded": ( - 'SUPPRESS = "ctypes.CDLL(None).prctl(4, 0, 0, 0, 0)\\n"\n' - 'SAFE = SUPPRESS + "ctypes.string_at(0)\\n"\n', + "import subprocess, sys\n" + 'SUPPRESS = "import ctypes\\nctypes.CDLL(None).prctl(4, 0, 0, 0, 0)\\n"\n' + 'SAFE = SUPPRESS + "ctypes.string_at(0)\\n"\n' + 'subprocess.run([sys.executable, "-c", SAFE])\n', False, # judged as the script it becomes, not as a bare literal ), + # Each of the following was a live hole found in review of the guard itself. + "spacing_the_markers_did_not_anticipate": ( + "import ctypes\ndef child():\n ctypes.string_at( 0)\n", + True, # the prefilter must not decide precision, only what to parse + ), + "suppression_placed_after_the_fault": ( + "import ctypes\n" + "def child():\n" + " ctypes.string_at(0)\n" + " ctypes.CDLL(None).prctl(4, 0, 0, 0, 0)\n", + True, # too late, the core is already written + ), + "numeric_fatal_signal": ( + "import signal\ndef child():\n signal.raise_signal(11)\n", + True, # 11 is SIGSEGV and dumps the same core as the symbolic name + ), + "annotated_script_constant": ( + "import subprocess, sys\n" + 'SUP: str = "import ctypes\\nctypes.CDLL(None).prctl(4, 0, 0, 0, 0)\\n"\n' + 'SCRIPT: str = SUP + "ctypes.string_at(0)\\n"\n' + 'subprocess.run([sys.executable, "-c", SCRIPT])\n', + False, # an annotated assignment folds like any other + ), + "expectation_string_is_not_a_script": ( + 'def test_x(script):\n assert "ctypes.string_at(0)" in script\n', + False, # nothing executes it, so failing CI on it would be wrong + ), + "aliased_crash_inside_a_script": ( + "import subprocess, sys\n" + 'SCRIPT = "from os import abort\\nabort()\\n"\n' + 'subprocess.run([sys.executable, "-c", SCRIPT])\n', + True, # same SIGABRT, spelled differently + ), + "suppression_via_a_local_helper": ( + "import ctypes\n" + "def suppress_core():\n" + " ctypes.CDLL(None).prctl(4, 0, 0, 0, 0)\n" + "def child():\n" + " suppress_core()\n" + " ctypes.string_at(0)\n", + False, # factoring the suppression out is good practice, not a violation + ), + "command_vector_bound_to_a_name": ( + "import subprocess, sys\n" + 'CMD = [sys.executable, "-c", "import ctypes; ctypes.string_at(0)"]\n' + "subprocess.run(CMD)\n", + True, # a command list built once and passed by name is ordinary subprocess use + ), + "crash_imported_under_an_alias": ( + "from os import abort as die\ndef child():\n die()\n", + True, # the bound name is `die`; the rules are written against `abort` + ), + "aliased_string_at": ( + "from ctypes import string_at as boom\ndef child():\n boom(0)\n", + True, + ), + "method_sharing_a_helper_name_is_not_suppression": ( + "import ctypes\n" + "def suppress_core():\n" + " ctypes.CDLL(None).prctl(4, 0, 0, 0, 0)\n" + "def child(obj):\n" + " obj.suppress_core()\n" + " ctypes.string_at(0)\n", + True, # the object's method may clear nothing; only a bare local call counts + ), + "async_suppressor_must_be_awaited": ( + "import ctypes\n" + "async def suppress_core():\n" + " ctypes.CDLL(None).prctl(4, 0, 0, 0, 0)\n" + "def child():\n" + " suppress_core()\n" + " ctypes.string_at(0)\n", + True, # calling it only builds a coroutine, so the prctl never runs + ), + "awaited_async_suppressor_counts": ( + "import ctypes\n" + "async def suppress_core():\n" + " ctypes.CDLL(None).prctl(4, 0, 0, 0, 0)\n" + "async def child():\n" + " await suppress_core()\n" + " ctypes.string_at(0)\n", + False, + ), + "rebound_alias_is_not_a_crash": ( + "from os import abort\ndef child():\n abort = lambda: None\n abort()\n", + False, + ), + "sigkill_is_not_a_deliberate_crash": ( + "import os, signal\ndef stop(pid):\n os.kill(pid, signal.SIGKILL)\n", + False, # SIGKILL never dumps, which is why it is the recommended alternative + ), "dumpable_set_back_to_one": ( "import ctypes\n" "def child():\n" @@ -394,6 +777,54 @@ _FIXTURES = { "def go(route, task):\n route.abort('failed')\n task.abort()\n", False, # Playwright and friends share the name and crash nothing ), + # Seven more the detector got wrong, each found by reading it rather than by a + # failing run. Four let a real core dump through; three failed CI on safe code. + "a_pid_that_looks_like_a_signal": ( + "import os, signal\ndef stop():\n os.kill(11, signal.SIGKILL)\n", + False, # 11 is the PID here, and SIGKILL never dumps + ), + "same_name_bound_in_two_scopes": ( + "import subprocess, sys\n" + 'SCRIPT = "import ctypes\\nctypes.string_at(0)\\n"\n' + "def unrelated():\n" + ' SCRIPT = "print(1)"\n' + " return SCRIPT\n" + 'subprocess.run([sys.executable, "-c", SCRIPT])\n', + True, # a local of the same name must not overwrite the script that runs + ), + "suppression_earlier_on_the_same_line": ( + "import subprocess, sys\n" + 'SCRIPT = "import ctypes; ctypes.CDLL(None).prctl(4, 0, 0, 0, 0); ' + 'ctypes.string_at(0)"\n' + 'subprocess.run([sys.executable, "-c", SCRIPT])\n', + False, # a one-line -c script still has an order, given by the column + ), + "helper_called_only_from_an_uninvoked_def": ( + "import ctypes\n" + "def suppress_core():\n" + " ctypes.CDLL(None).prctl(4, 0, 0, 0, 0)\n" + "def child():\n" + " def configure():\n" + " suppress_core()\n" + " ctypes.string_at(0)\n", + True, # configure() is never called, so nothing suppressed the fault + ), + "script_built_as_an_f_string": ( + "import subprocess, sys\n" + 'subprocess.run([sys.executable, "-c", f"import os; os.abort(); ' + 'print({sys.argv})"])\n', + True, # an f-string reaches the child as a script like any other + ), + "imported_name_rebound_before_the_call": ( + "from os import abort\ndef test_it(mock):\n abort = mock\n abort()\n", + False, # only the mock runs, so there is no crash to suppress + ), + "crash_nested_inside_an_exec": ( + "import subprocess, sys\n" + "SCRIPT = \"exec('import os; os.abort()')\"\n" + 'subprocess.run([sys.executable, "-c", SCRIPT])\n', + True, # the SIGABRT is one level down, and dumps just the same + ), }