subaudit 0.1.0

Subscribe and unsubscribe for specific audit events

subaudit: Subscribe and unsubscribe for specific audit events

Audit hooks in Python are called on all events, and they remain in place until the interpreter shuts down.

This library provides a higher-level interface that allows listeners to be subscribed to specific audit events, and unsubscribed from them. It provides context managers for using that interface with a convenient notation that ensures the listener is unsubscribed. The context managers are reentrant—you can nest with-statements that listen to events. By default, a single audit hook is used for any number of events and listeners.

The primary use case for this library is in writing test code.

License

subaudit is licensed under 0BSD, which is a “public-domain equivalent” license. See LICENSE.

Compatibility

The subaudit library can be used to observe audit events generated by the Python interpreter and standard library, as well as custom audit events. It requires Python 3.7 or later. It is most useful on Python 3.8 or later, because audit events were introduced in Python 3.8. On Python 3.7, subaudit uses the sysaudit library to support audit events, but the Python interpreter and standard library still do not provide any events, so only custom events can be used on Python 3.7.

To avoid the performance cost of explicit locking in the audit hook, some operations are assumed atomic. I believe these assumptions are correct for CPython, as well as PyPy and some other implementations, but there may exist Python implementations on which these assumptions don’t hold.

Installation

Install the subaudit package (PyPI) in your project’s environment.

Basic usage

The subaudit.listening context manager

The best way to use subaudit is usually the listening context manager.

import subaudit

def listen_open(path, mode, flags):
    ...  # Handle the event.

with subaudit.listening('open', listen_open):
    ...  # Do something that may raise the event.

The listener—here, listen_open—is called with the event arguments each time the event is raised. They are passed to the listener as separate positional arguments (not as an args tuple).

In tests, it is convenient to use Mock objects as listeners, because they record calls, provide a mock_calls attribute to see the calls, and provide various assert_* methods to make assertions about the calls:

from unittest.mock import ANY, Mock
import subaudit

with subaudit.listening('open', Mock()) as listener:
    ...  # Do something that may raise the event.

listener.assert_any_call('/path/to/file.txt', 'r', ANY)

Note how, when the listening context manager is entered, it returns the listener that was passed in, for convenience.

The subaudit.extracting context manager

You may want to extract some information about calls to a list:

from dataclasses import InitVar, dataclass
import subaudit

@dataclass(frozen=True)
class PathAndMode:  # Usually strings. See examples/notebooks/open_event.ipynb.
    path: str
    mode: str
    flags: InitVar = None  # Opt not to record this argument.

with subaudit.extracting('open', PathAndMode) as extracts:
    ...  # Do something that may raise the event.

assert PathAndMode('/path/to/file.txt', 'r') in extracts

The extractor—here, PathAndMode—can be any callable that accepts the event args as separate positional arguments. Entering the context manager returns an initially empty list, which will be populated with extracts gleaned from the event args. Each time the event is raised, the extractor is called and the object it returns is appended to the list.

subaudit.subscribe and subaudit.unsubscribe

Although you should usually use the listening or extracting context managers instead, you can subscribe and unsubscribe listeners without a context manager:

import subaudit

def listen_open(path, mode, flags):
    ...  # Handle the event.

subaudit.subscribe('open', listen_open)
try:
    ...  # Do something that may raise the event.
finally:
    subaudit.unsubscribe('open', listen_open)

Attempting to unsubscribe a listener that is not subscribed raises ValueError. Currently, subaudit provides no feature to make this succeed silently instead. But you can suppress the exception:

with contextlib.suppress(ValueError):
    subaudit.unsubscribe('glob.glob', possibly_subscribed_listener)

Nesting

To unsubscribe a listener from an event, it must be subscribed to the event. Subject to this restriction, calls to subscribe and unsubscribe can happen in any order, and listening and extracting may be arbitrarily nested.

listening and extracting support reentrant use with both the same event and different events. Here’s an example with three listening contexts:

from unittest.mock import Mock, call

listen_to = Mock()  # Let us assert calls to child mocks in a specific order.

with subaudit.listening('open', print):  # Print all open events' arguments.
    with subaudit.listening('open', listen_to.open):  # Log opening.
        with subaudit.listening('glob.glob', listen_to.glob):  # Log globbing.
            ...  # Do something that may raise the events.

assert listen_to.mock_calls == ...  # Assert a specific order of calls.

(That is written out to make the nesting clear. You could also use a single with-statement with commas.)

Here’s an example with both listening and extracting contexts:

from unittest.mock import Mock, call

def extract(*args):
    return args

with (
    subaudit.extracting('pathlib.Path.glob', extract) as glob_extracts,
    subaudit.listening('pathlib.Path.glob', Mock()) as glob_listener,
    subaudit.extracting('pathlib.Path.rglob', extract) as rglob_extracts,
    subaudit.listening('pathlib.Path.rglob', Mock()) as rglob_listener,
):
    ...  # Do something that may raise the events.

# Assert something about, or otherwise use, the mocks glob_listener and
# rglob_listener, as well as the lists glob_extracts and rglob_extracts.
...

(That example uses parenthesized context managers, which were introduced in Python 3.10.)

Specialized usage

subaudit.Hook objects

Each instance of the subaudit.Hook class represents a single audit hook that supports subscribing and unsubscribing listeners for any number of events, with methods corresponding to the four top-level functions listed above. Separate Hook instances use separate audit hooks. The Hook class exists for three purposes:

• It supplies the behavior of the top-level listening, extracting, subscribe, and unsubscribe functions, which correspond to the same-named methods on a global Hook instance.
• It allows multiple audit hooks to be used, for special cases where that might be desired.
• It facilitates customization, as detailed below.

The actual audit hook that a Hook object encapsulates is not installed until the first listener is subscribed. This happens on the first call to its subscribe method, or the first time one of its context managers (from calling its listening or extracting method) is entered. This is also true of the global Hook instance used by the top-level functions—merely importing subaudit does not install an audit hook.

Whether the top-level functions are bound methods of a Hook instance, or delegate in some other way to those methods on an instance, is currently considered an implementation detail.

Deriving from Hook

You can derive from Hook to provide custom behavior for subscribing and unsubscribing, by overriding the subscribe and unsubscribe methods. You can also override the listening and extracting methods, though that may be less useful. Overridden subscribe and unsubscribe methods are automatically used by listening and extracting.

Whether extracting uses listening, or directly calls subscribe and unsubscribe, is currently considered an implementation detail.

Locking

Consider two possible cases of race conditions:

1. Between audit hook and subscribe/unsubscribe (audit hook does not lock)

In this scenario, a Hook object’s installed audit hook runs at the same time as a listener is subscribed or unsubscribed.

This is likely to occur often and it cannot be prevented, because audit hooks are called for all audit events. For the same reason, locking in the audit hook has performance implications. Instead of having audit hooks take locks, subaudit relies on each of these operations being atomic:

• Writing an attribute reference, when it is a simple write to an instance dictionary or a slot. Writing an attribute need not be atomic when, for example, __setattr__ has been overridden.
• Writing or deleting a str key in a dictionary whose keys are all of the built-in str type. Note that the search need not be atomic, but the dictionary must always be observed to be in a valid state.

The audit hook is written, and the data structures it uses are selected, to avoid relying on more than these assumptions.

2. Between calls to subscribe/unsubscribe (by default, they lock)

In this scenario, two listeners are subscribed at a time, or unsubscribed at a time, or one listener is subscribed while another (or the same) listener is unsubscribed.

This is less likely to occur and much easier to avoid. But it is also harder to make safe without a lock. Subscribing and unsubscribing are unlikely to happen at a sustained high rate, so locking is unlikely to be a performance bottleneck. So, by default, subscribing and unsubscribing are synchronized with a threading.Lock, to ensure that shared state is not corrupted.

You should not usually change this. But if you want to, you can construct a Hook object by calling Hook(sub_lock_factory=...) instead of Hook, where ... is a type, or other context manager factory, to be used instead of threading.Lock. In particular, to disable locking, pass contextlib.nullcontext.

Functions related to compatibility

As noted above, Python supports audit hooks since 3.8. For Python 3.7, but not Python 3.8 or later, the subaudit library declares sysaudit as a dependency.

subaudit.addaudithook and subaudit.audit

subaudit exports addaudithook and audit functions.

• On Python 3.8 and later, they are sys.addaudithook and sys.audit.
• On Python 3.7, they are sysaudit.addaudithook and sysaudit.audit.

subaudit uses subaudit.addaudithook when it adds its own audit hook (or all its own hooks, if you use additional Hook instances besides the global one implicitly used by the top-level functions). subaudit does not itself use subaudit.audit, but it is whichever audit function corresponds to subaudit.addaudithook.

@subaudit.skip_if_unavailable

The primary use case for subaudit is in writing unit tests, to assert that particular events have been raised or not raised. Usually these are “built in” events—those raised by the Python interpreter or standard library. But the sysaudit library doesn’t backport those events, which would not really be feasible to do.

For this reason, tests that particular audit events did or didn’t occur—such as a test that a file has been opened by listening to the open event—should typically be skipped when running a test suite on Python 3.7.

When using the unittest framework, you can apply the @skip_if_unavailable decorator to a test class or test method, so it is skipped prior to Python 3.8 with a message explaining why. For example:

import unittest
from unittest.mock import ANY, Mock
import subaudit

class TestSomeThings(unittest.TestCase):
    ...

    @subaudit.skip_if_unavailable  # Skip this test if < 3.8, with a message.
    def test_file_is_opened_for_read(self):
        with subaudit.listening('open', Mock()) as listener:
            ...  # Do something that may raise the event.

        listener.assert_any_call('/path/to/file.txt', 'r', ANY)

    ...

@subaudit.skip_if_unavailable  # Skip the whole class if < 3.8, with a message.
class TestSomeMoreThings(unittest.TestCase):
    ...

It could be useful also to have a conditional xfail (expected failure) decorator for unittest—and, more so, marks for pytest providing specialized skip/skipif and xfail—but subaudit does not currently provide them. Of course, in pytest, you can still use the @pytest.mark.skip and @pytest.mark.xfail decorators, by passing sys.version_info < (3, 8) as the condition.

Overview by level of abstraction

From higher to lower level, from the perspective of the top-level listening and extracting functions:

• subaudit.extracting - context manager that listens and extracts to a list
• subaudit.listening - context manager to subscribe and unsubscribe a custom listener (usually use this)
• subaudit.subscribe and subaudit.unsubscribe - manually subscribe/unsubscribe a listener
• subaudit.Hook - abstraction around an audit hook allowing subscribing and unsubscribing for specific events, with extracting, listening, subscribe, and unsubscribe instance methods
• subaudit.addaudithook - trivial abstraction representing whether the function from sys or sysaudit is used
• sys.addaudithook or sysaudit.addaudithook - not part of subaudit - install a PEP 578 audit hook

This list is not exhaustive. For example, @skip_if_unavailable is not part of that conceptual hierarchy.

Acknowledgements

I’d like to thank:

• Brett Langdon, who wrote the sysaudit library (which subaudit uses on 3.7).

• David Vassallo, for reviewing pull requests about testing using audit hooks in a project we have collaborated on, which helped me to recognize what kinds of usage were more or less clear and that it could be good to have a library like subaudit; and for coauthoring a @skip_if_unavailable decorator that had been used there, which motivated the one here.

About the name

This library is called “subaudit” because it provides a way to effectively subscribe to and unsubscribe from a subset of audit events rather than all of them.