Add --unblock option for side effects

Author: pschanely

crosshair/auditwall.py

@@ -6,7 +6,17 @@
 import traceback
 from contextlib import contextmanager
 from types import ModuleType
-from typing import Callable, Dict, Generator, Iterable, Optional, Sequence, Set, Tuple
+from typing import (
+    Callable,
+    Dict,
+    Generator,
+    Iterable,
+    NoReturn,
+    Optional,
+    Sequence,
+    Set,
+    Tuple,
+)
 
 
 class SideEffectDetected(Exception):
@@ -24,13 +34,16 @@ def accept(event: str, args: Tuple) -> None:
 
 def explain(event: str, args: Tuple) -> str:
     argstr = "".join(f":{arg}" for arg in args)
-    return (
-        f"CrossHair should not be run on code with side effects. "
-        f'To allow this operation, use "--unblock={event}{argstr}" (or some colon-delimited prefix).'
-    )
+    parts = [
+        f"It's dangerous to run CrossHair on code with side effects.",
+        f'To allow this operation anyway, use "--unblock={event}{argstr}".',
+    ]
+    if args:
+        parts.append("(or some colon-delimited prefix)")
+    return " ".join(parts)
 
 
-def reject(event: str, args: Tuple) -> None:
+def reject(event: str, args: Tuple) -> NoReturn:
     raise SideEffectDetected(
         f'A "{event}" operation was detected. ' + explain(event, args)
     )
@@ -88,7 +101,7 @@ def check_subprocess(event: str, args: Tuple) -> None:
         reject(event, args)
 
 
-_SPECIAL_HANDLERS = {
+_SPECIAL_HANDLERS: Dict[str, Callable[[str, Tuple], None]] = {
     "open": check_open,
     "subprocess.Popen": check_subprocess,
     "os.posix_spawn": check_subprocess,
@@ -189,6 +202,7 @@ def opened_auditwall() -> Generator:
 
 def _make_prefix_based_handler(
     allowed_arg_prefixes: Sequence[Sequence[str]],
+    previous_handler: Optional[Callable[[str, Tuple], None]] = None,
 ) -> Callable[[str, Tuple], None]:
     trie: Dict = {}
     for prefix in allowed_arg_prefixes:
@@ -201,10 +215,14 @@ def handler(event: str, args: Tuple) -> None:
         current = trie
         for arg in map(str, args):
             if arg not in current:
-                reject(event, args)
+                break
             current = current[arg]
             if None in current:
                 return  # Found a valid prefix
+        if previous_handler:
+            previous_handler(event, args)
+        else:
+            reject(event, args)
 
     return handler
 
@@ -215,13 +233,17 @@ def _update_special_handlers(allow_prefixes: Sequence[str]) -> None:
     ):
         group = tuple(group_itr)
         if any(event == g for g in group):
-            _HANDLERS[event] = accept
+            _SPECIAL_HANDLERS[event] = accept
         else:
             args = tuple(a.split(":")[1:] for a in group)
-            _HANDLERS[event] = _make_prefix_based_handler(args)
+            _SPECIAL_HANDLERS[event] = _make_prefix_based_handler(
+                args, _SPECIAL_HANDLERS.get(event)
+            )
 
 
 def engage_auditwall(allow_prefixes: Sequence[str] = ()) -> None:
+    if "EVERYTHING" in allow_prefixes:
+        return
     _update_special_handlers(allow_prefixes)
     sys.dont_write_bytecode = True  # disable .pyc file writing
     sys.addaudithook(audithook)

crosshair/auditwall_test.py

@@ -30,6 +30,40 @@ def test_fs_write_disallowed():
     assert call([pyexec, __file__, "write_open", "withwall"]) == 10
 
 
+def test_fs_write_allowed_if_prefix_allowed():
+    try:
+        assert (
+            call(
+                [
+                    pyexec,
+                    __file__,
+                    "write_open",
+                    "withwall",
+                    "open:./auditwall.testwrite.txt:w",
+                ]
+            )
+            == 0
+        )
+        # Confirm the new handler doesn't interfere with the prior handler:
+        assert (
+            call(
+                [
+                    pyexec,
+                    __file__,
+                    "read_open",
+                    "withwall",
+                    "open:./auditwall.testwrite.txt:w",
+                ]
+            )
+            == 0
+        )
+    finally:
+        try:
+            os.unlink("./auditwall.testwrite.txt")
+        except FileNotFoundError:
+            pass
+
+
 def test_http_disallowed():
     assert call([pyexec, __file__, "http", "withwall"]) == 10
 
@@ -104,7 +138,7 @@ def test_popen_via_platform_allowed():
     "read_open": lambda: open("/dev/null", "rb"),
     "scandir": lambda: os.scandir("."),
     "import": lambda: __import__("shutil"),
-    "write_open": lambda: open("/.auditwall.testwrite.txt", "w"),
+    "write_open": lambda: open("./auditwall.testwrite.txt", "w"),
     "http": lambda: urllib.request.urlopen("http://localhost/foo"),
     "unlink": lambda: os.unlink("./delme.txt"),
     "popen": lambda: call(["echo", "hello"]),

crosshair/main.py

@@ -102,8 +102,25 @@ def command_line_parser() -> argparse.ArgumentParser:
         "--extra_plugin",
         type=str,
         nargs="+",
+        metavar="FILE",
         help="Plugin file(s) you wish to use during the current execution",
     )
+    common.add_argument(
+        "--unblock",
+        type=str,
+        nargs="+",
+        default=(),
+        metavar="EVENT",
+        help=textwrap.dedent(
+            """\
+        Allow specific side-effects. See the list of audit events at:
+        https://docs.python.org/3/library/audit_events.html
+        You may specify colon-delimited event arguments to narrow the unblock, e.g.:
+            --unblock subprocess.Popen:echo
+        Finally, `--unblock EVERYTHING` will disable all side-effect detection.
+        """
+        ),
+    )
     parser = argparse.ArgumentParser(
         prog="crosshair", description="CrossHair Analysis Tool"
     )
@@ -958,16 +975,18 @@ def mypy_and_check(cmd_args: Optional[List[str]] = None) -> None:
         if mypy_ret != 0:
             print(_mypy_out, file=sys.stdout)
             sys.exit(mypy_ret)
-    engage_auditwall()
+    engage_auditwall(check_args.unblock)
     debug("Running crosshair with these args:", check_args)
     sys.exit(unwalled_main(check_args))
 
 
 def main(cmd_args: Optional[List[str]] = None) -> None:
     if cmd_args is None:
         cmd_args = sys.argv[1:]
-    engage_auditwall()
-    sys.exit(unwalled_main(cmd_args))
+    parsed_args = command_line_parser().parse_args(cmd_args)
+
+    engage_auditwall(parsed_args.unblock)
+    sys.exit(unwalled_main(parsed_args))
 
 
 if __name__ == "__main__":

crosshair/options.py

@@ -50,6 +50,7 @@ class AnalysisOptionSet:
     max_iterations: Optional[int] = None
     report_all: Optional[bool] = None
     report_verbose: Optional[bool] = None
+    unblock: Optional[Sequence[str]] = None
     timeout: Optional[float] = None
     max_uninteresting_iterations: Optional[int] = None
 
@@ -106,6 +107,7 @@ def option_set_from_dict(source: Mapping[str, object]) -> AnalysisOptionSet:
         "max_uninteresting_iterations",
         "report_all",
         "report_verbose",
+        "unblock",
     ):
         arg_val = source.get(optname, None)
         if arg_val is not None:
@@ -124,6 +126,7 @@ class AnalysisOptions:
     max_iterations: int
     report_all: bool
     report_verbose: bool
+    unblock: Sequence[str]
     timeout: float
     per_path_timeout: float
     max_uninteresting_iterations: int
@@ -212,6 +215,7 @@ def incr(self, key: str):
     max_iterations=sys.maxsize,
     report_all=False,
     report_verbose=True,
+    unblock=(),
     timeout=float("inf"),
     per_path_timeout=float("NaN"),
     max_uninteresting_iterations=sys.maxsize,

crosshair/watcher.py

@@ -33,7 +33,7 @@
     run_checkables,
 )
 from crosshair.fnutil import NotFound, walk_paths
-from crosshair.options import AnalysisOptionSet
+from crosshair.options import DEFAULT_OPTIONS, AnalysisOptionSet
 from crosshair.util import (
     CrossHairInternal,
     ErrorDuringImport,
@@ -137,7 +137,7 @@ def pool_worker_main() -> None:
             # Windows, where the nice function does not exist:
             os.nice(10)  # type: ignore
         set_debug(False)
-        engage_auditwall()
+        engage_auditwall(DEFAULT_OPTIONS.overlay(item[1]).unblock)
         (stats, messages) = pool_worker_process_item(item)
         output: WorkItemOutput = (filename, stats, messages)
         print(serialize(output))

crosshair/watcher_test.py

@@ -5,6 +5,8 @@
 
 import pytest
 
+from crosshair.main import command_line_parser
+from crosshair.options import option_set_from_dict
 from crosshair.statespace import MessageType
 from crosshair.test_util import simplefs
 from crosshair.watcher import Watcher
@@ -49,6 +51,16 @@ def foofn(x: int) -> int:
 """
 }
 
+CHATTY_FOO = {
+    "foo.py": """
+from subprocess import Popen
+def foofn(x: int) -> int:
+  ''' post: _ == x '''
+  Popen(['echo', 'hello']).communicate()
+  return x
+"""
+}
+
 EMPTY_BAR = {
     "bar.py": """
 # Nothing here
@@ -105,3 +117,27 @@ def test_removed_file_given_as_argument(tmp_path: Path):
     assert watcher.check_changed()
     (tmp_path / "foo.py").unlink()
     assert watcher.check_changed()
+
+
+@pytest.mark.parametrize(
+    "unblock,expected_state",
+    [
+        ("--unblock=subprocess.Popen:echo", MessageType.CONFIRMED),
+        ("--unblock=subprocess.Popen:xyz", MessageType.EXEC_ERR),
+    ],
+)
+def test_auditwall_unblock(unblock: str, expected_state: MessageType, tmp_path: Path):
+    parsed_args, _ = command_line_parser().parse_known_args(
+        ["watch", unblock, "-v", "dummy"]
+    )
+    optionset = option_set_from_dict(parsed_args.__dict__)
+    simplefs(tmp_path, CHATTY_FOO)
+    watcher = Watcher([tmp_path], optionset)
+
+    watcher._next_file_check = time.time() - 1
+    # Detect file (yields empty result to wake up the loop)
+    assert list(watcher.run_iteration()) == [(Counter(), [])]
+    # real results in new file after restart:
+    results = list(watcher.run_iteration())
+    assert len(results) == 1
+    assert [m.state for m in results[0][1]] == [expected_state]

doc/source/changelog.rst

@@ -9,6 +9,14 @@ Next Version
   * Nothing yet!
 
 
+Version 0.0.101
+---------------
+
+  * Support user-defined exceptions for CrossHair's side-effect detection
+    mechanism with a new ``--unblock`` command-line option.
+    (resolves `#383 <https://github.com/pschanely/CrossHair/issues/383>`__)
+
+
 Version 0.0.100
 ---------------
 

doc/source/contracts.rst

@@ -97,9 +97,8 @@ Type Ctrl-C to stop this command.
 .. Help starts: crosshair watch --help
 .. code-block:: text
 
-    usage: crosshair watch [-h] [--verbose]
-                           [--extra_plugin EXTRA_PLUGIN [EXTRA_PLUGIN ...]]
-                           [--analysis_kind KIND]
+    usage: crosshair watch [-h] [--verbose] [--extra_plugin FILE [FILE ...]]
+                           [--unblock EVENT [EVENT ...]] [--analysis_kind KIND]
                            TARGET [TARGET ...]
 
     The watch command continuously looks for contract counterexamples.
@@ -112,8 +111,14 @@ Type Ctrl-C to stop this command.
     options:
       -h, --help            show this help message and exit
       --verbose, -v         Output additional debugging information on stderr
-      --extra_plugin EXTRA_PLUGIN [EXTRA_PLUGIN ...]
+      --extra_plugin FILE [FILE ...]
                             Plugin file(s) you wish to use during the current execution
+      --unblock EVENT [EVENT ...]
+                            Allow specific side-effects. See the list of audit events at:
+                            https://docs.python.org/3/library/audit_events.html
+                            You may specify colon-delimited event arguments to narrow the unblock, e.g.:
+                                --unblock subprocess.Popen:echo
+                            Finally, `--unblock EVERYTHING` will disable all side-effect detection.
       --analysis_kind KIND  Kind of contract to check.
                             By default, the PEP316, deal, and icontract kinds are all checked.
                             Multiple kinds (comma-separated) may be given.
@@ -137,9 +142,9 @@ It is more customizable than ``watch`` and produces machine-readable output.
 .. Help starts: crosshair check --help
 .. code-block:: text
 
-    usage: crosshair check [-h] [--verbose]
-                           [--extra_plugin EXTRA_PLUGIN [EXTRA_PLUGIN ...]]
-                           [--report_all] [--report_verbose]
+    usage: crosshair check [-h] [--verbose] [--extra_plugin FILE [FILE ...]]
+                           [--unblock EVENT [EVENT ...]] [--report_all]
+                           [--report_verbose]
                            [--max_uninteresting_iterations MAX_UNINTERESTING_ITERATIONS]
                            [--per_path_timeout FLOAT]
                            [--per_condition_timeout FLOAT] [--analysis_kind KIND]
@@ -164,8 +169,14 @@ It is more customizable than ``watch`` and produces machine-readable output.
     options:
       -h, --help            show this help message and exit
       --verbose, -v         Output additional debugging information on stderr
-      --extra_plugin EXTRA_PLUGIN [EXTRA_PLUGIN ...]
+      --extra_plugin FILE [FILE ...]
                             Plugin file(s) you wish to use during the current execution
+      --unblock EVENT [EVENT ...]
+                            Allow specific side-effects. See the list of audit events at:
+                            https://docs.python.org/3/library/audit_events.html
+                            You may specify colon-delimited event arguments to narrow the unblock, e.g.:
+                                --unblock subprocess.Popen:echo
+                            Finally, `--unblock EVERYTHING` will disable all side-effect detection.
       --report_all          Output analysis results for all postconditions (not just failing ones)
       --report_verbose      Output context and stack traces for counterexamples
       --max_uninteresting_iterations MAX_UNINTERESTING_ITERATIONS
@@ -287,3 +298,5 @@ directly or indirectly cause side-effects.
 CrossHair puts some protections in place (via ``sys.addaudithook``) to prevent disk
 and network access, but this protection is not perfect. (notably, it will not
 prevent actions taken by C-based modules)
+You can bypass CrossHair's internal protections with the ``--unblock`` command-line
+option, which may be appropriate in a containerized or otherwise isolated environment.