Exceptions

Decorators and utilities for prefixing exception stack traces while obscuring the exception message itself.

`PublicArgumentError`

Argument error with public message. Exceptions of this type raised under prefix_stack_trace or print_prefixed_stack_trace_and_raise will have the message prefixed with PREFIX in both the printed stack trace and the re-raised exception.

`PublicFileNotFoundError`

File not found error with public message. Exceptions of this type raised under prefix_stack_trace or print_prefixed_stack_trace_and_raise will have the message prefixed with PREFIX in both the printed stack trace and the re-raised exception.

`PublicIndexError`

Index error with public message. Exceptions of this type raised under prefix_stack_trace or print_prefixed_stack_trace_and_raise will have the message prefixed with PREFIX in both the printed stack trace and the re-raised exception.

`PublicIOError`

I/O error with public message. Exceptions of this type raised

under prefix_stack_trace or print_prefixed_stack_trace_and_raise will have the message prefixed with PREFIX in both the printed stack trace and the re-raised exception.

`PublicKeyError`

Key error with public message. Exceptions of this type raised under prefix_stack_trace or print_prefixed_stack_trace_and_raise will have the message prefixed with PREFIX in both the printed stack trace and the re-raised exception.

`PublicNotImplementedError`

Not implemented error with public message. Exceptions of this type raised under prefix_stack_trace or print_prefixed_stack_trace_and_raise will have the message prefixed with PREFIX in both the printed stack trace and the re-raised exception.

`PublicRuntimeError`

Runtime error with public message. Exceptions of this type raised under prefix_stack_trace or print_prefixed_stack_trace_and_raise will have the message prefixed with PREFIX in both the printed stack trace and the re-raised exception.

`PublicTypeError`

Type error with public message. Exceptions of this type raised under prefix_stack_trace or print_prefixed_stack_trace_and_raise will have the message prefixed with PREFIX in both the printed stack trace and the re-raised exception.

`PublicValueError`

Value error with public message. Exceptions of this type raised under prefix_stack_trace or print_prefixed_stack_trace_and_raise will have the message prefixed with PREFIX in both the printed stack trace and the re-raised exception.

`is_exception_allowed(exception, allow_list)`

Check if message is allowed, either by allow_list, or default_allow_list.

Parameters:

Name	Type	Description	Default
`exception`	`Union[BaseException, traceback.TracebackException]`	the exception to test	required
`allow_list`	`list`	list of regex expressions. If any expression matches the exception name or message, it will be considered allowed.	required

Returns:

Type	Description
`bool`	bool: True if message is allowed, False otherwise.

Source code in confidential_ml_utils/exceptions.py

def is_exception_allowed(
    exception: Union[BaseException, TracebackException], allow_list: list
) -> bool:
    """
    Check if message is allowed, either by `allow_list`, or `default_allow_list`.

    Args:
        exception (TracebackException): the exception to test
        allow_list (list): list of regex expressions. If any expression matches
            the exception name or message, it will be considered allowed.

    Returns:
        bool: True if message is allowed, False otherwise.
    """
    if not isinstance(exception, TracebackException):
        exception = TracebackException.from_exception(exception)

    # empty list means all messages are allowed
    for expr in allow_list + default_allow_list:
        if re.search(expr, getattr(exception, "_str", ""), re.IGNORECASE):
            return True
        if re.search(expr, getattr(exception.exc_type, "__name__", ""), re.IGNORECASE):
            return True
    return False

`prefix_stack_trace(file=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>, disable=False, prefix='SystemLog:', scrub_message='Exception message scrubbed', keep_message=False, allow_list=[], add_timestamp=False)`

Decorator which wraps the decorated function and prints the stack trace of exceptions which occur, prefixed with prefix and with exception messages scrubbed (replaced with scrub_message). To use this, just add @prefix_stack_trace() above your function definition, e.g.

@prefix_stack_trace()
def foo(x):
    pass

Source code in confidential_ml_utils/exceptions.py

def prefix_stack_trace(
    file: TextIO = sys.stderr,
    disable: bool = bool(sys.flags.debug),
    prefix: str = PREFIX,
    scrub_message: str = SCRUB_MESSAGE,
    keep_message: bool = False,
    allow_list: list = [],
    add_timestamp: bool = False,
) -> Callable:
    """
    Decorator which wraps the decorated function and prints the stack trace of
    exceptions which occur, prefixed with `prefix` and with exception messages
    scrubbed (replaced with `scrub_message`). To use this, just add
    `@prefix_stack_trace()` above your function definition, e.g.

        @prefix_stack_trace()
        def foo(x):
            pass
    """

    return _PrefixStackTraceWrapper(
        file, disable, prefix, scrub_message, keep_message, allow_list, add_timestamp
    )

`print_prefixed_stack_trace_and_raise(file=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>, prefix='SystemLog:', scrub_message='Exception message scrubbed', keep_message=False, allow_list=[], add_timestamp=False, err=None)`

Print the current exception and stack trace to file (usually client standard error), prefixing the stack trace with prefix.

Parameters:

Name	Type	Description	Default
`keep_message`	`bool`	if True, don't scrub message. If false, scrub (unless allowed).	`False`
`allow_list`	`list`	exception allow_list. Ignored if keep_message is True. If empty all messages will be srubbed.	`[]`
`err`	`Optional[BaseException]`	the error that was thrown. None accepted for backwards compatibility.	`None`

Source code in confidential_ml_utils/exceptions.py

def print_prefixed_stack_trace_and_raise(
    file: TextIO = sys.stderr,
    prefix: str = PREFIX,
    scrub_message: str = SCRUB_MESSAGE,
    keep_message: bool = False,
    allow_list: list = [],
    add_timestamp: bool = False,
    err: Optional[BaseException] = None,
) -> None:
    """
    Print the current exception and stack trace to `file` (usually client
    standard error), prefixing the stack trace with `prefix`.
    Args:
        keep_message (bool): if True, don't scrub message. If false, scrub (unless
            allowed).
        allow_list (list): exception allow_list. Ignored if keep_message is True. If
            empty all messages will be srubbed.
        err: the error that was thrown. None accepted for backwards compatibility.
    """
    if err is None:
        err = sys.exc_info()[1]
    scrubbed_err = scrub_exception(err, scrub_message, prefix, keep_message, allow_list)

    tb_exception = TracebackException.from_exception(scrubbed_err)  # type: ignore

    for execution in tb_exception.format():
        if "return function(*func_args, **func_kwargs)" in execution:
            # Do not show the stack trace for our decorator.
            continue
        for line in execution.splitlines():
            if add_timestamp:
                current_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
                print(f"{prefix} {current_time} {line}", file=file)
            else:
                print(f"{prefix} {line}", file=file)

    raise scrubbed_err  # type: ignore

`scrub_exception(exception, scrub_message, prefix, keep_message, allow_list, _seen=None)`

Recursively scrub all potentially private data from an exception, using the logic in _attribute_transformer.

Inspired by Dan Schwartz's closed-source implementation: https://dev.azure.com/eemo/TEE/_git/TEEGit?path=%2FOffline%2FFocusedInbox%2FComTriage%2Fcomtriage%2Futils%2Fscrubber.py&version=GBcompliant%2FComTriage&_a=content

which is closely based on the CPython implementation of the TracebackException class: https://github.com/python/cpython/blob/master/Lib/traceback.py#L478

Source code in confidential_ml_utils/exceptions.py

def scrub_exception(
    exception: Optional[BaseException],
    scrub_message: str,
    prefix: str,
    keep_message: bool,
    allow_list: list,
    _seen: Optional[Set[int]] = None,
) -> Optional[BaseException]:
    """
    Recursively scrub all potentially private data from an exception, using the
    logic in `_attribute_transformer`.

    Inspired by Dan Schwartz's closed-source implementation:
    https://dev.azure.com/eemo/TEE/_git/TEEGit?path=%2FOffline%2FFocusedInbox%2FComTriage%2Fcomtriage%2Futils%2Fscrubber.py&version=GBcompliant%2FComTriage&_a=content

    which is closely based on the CPython implementation of the
    TracebackException class:
    https://github.com/python/cpython/blob/master/Lib/traceback.py#L478
    """
    if not exception:
        return None

    # Handle loops in __cause__ or __context__ .
    if _seen is None:
        _seen = set()
    _seen.add(id(exception))

    # Gracefully handle being called with no type or value.
    if exception.__cause__ is not None and id(exception.__cause__) not in _seen:
        exception.__cause__ = scrub_exception(
            exception.__cause__,
            scrub_message,
            prefix,
            keep_message,
            allow_list,
            _seen,
        )
    if exception.__context__ is not None and id(exception.__context__) not in _seen:
        exception.__context__ = scrub_exception(
            exception.__context__,
            scrub_message,
            prefix,
            keep_message,
            allow_list,
            _seen,
        )

    keep = keep_message or is_exception_allowed(exception, allow_list)
    transformer = _attribute_transformer(prefix, scrub_message, keep)

    for attr in dir(exception):
        if attr and not attr.startswith("__"):
            try:
                value = getattr(exception, attr)
            except AttributeError:
                # In some cases, e.g. FileNotFoundError, there are attributes
                # which show up in dir(e), but for which an AttributeError is
                # thrown when attempting to access the value. See, e.g.:
                # https://stackoverflow.com/q/47775772 .
                continue
            try:
                # If unable to transform or set the attribute, replace the
                # entire exception since the attribute value is readable, but
                # we are unable to scrub it.
                new_value = transformer(value)
                setattr(exception, attr, new_value)
            except BaseException as e:
                new_exception = PublicRuntimeError(
                    f"{prefix} Obtained {type(e).__name__} when trying to scrub {attr} from {type(exception).__name__}"  # noqa: E501
                )
                new_exception.__cause__ = exception.__cause__
                new_exception.__context__ = exception.__context__
                exception = new_exception
                break

    return exception

Exceptions

PublicArgumentError

PublicFileNotFoundError

PublicIndexError

PublicIOError

PublicKeyError

PublicNotImplementedError

PublicRuntimeError

PublicTypeError

PublicValueError

is_exception_allowed(exception, allow_list)

prefix_stack_trace(file=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>, disable=False, prefix='SystemLog:', scrub_message='**Exception message scrubbed**', keep_message=False, allow_list=[], add_timestamp=False)

print_prefixed_stack_trace_and_raise(file=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>, prefix='SystemLog:', scrub_message='**Exception message scrubbed**', keep_message=False, allow_list=[], add_timestamp=False, err=None)

scrub_exception(exception, scrub_message, prefix, keep_message, allow_list, _seen=None)

`PublicArgumentError`

`PublicFileNotFoundError`

`PublicIndexError`

`PublicIOError`

`PublicKeyError`

`PublicNotImplementedError`

`PublicRuntimeError`

`PublicTypeError`

`PublicValueError`

`is_exception_allowed(exception, allow_list)`

`prefix_stack_trace(file=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>, disable=False, prefix='SystemLog:', scrub_message='Exception message scrubbed', keep_message=False, allow_list=[], add_timestamp=False)`

`print_prefixed_stack_trace_and_raise(file=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>, prefix='SystemLog:', scrub_message='Exception message scrubbed', keep_message=False, allow_list=[], add_timestamp=False, err=None)`

`scrub_exception(exception, scrub_message, prefix, keep_message, allow_list, _seen=None)`