diff --git a/docs/api.rst b/docs/api.rst index 9ff57dd6..dd618990 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -83,14 +83,51 @@ This machinery is used by the notebook web application to record which notebooks are *trusted*, and may show dynamic output as soon as they're loaded. See :ref:`jupyter-server:server_security` for more information. +A :class:`NotebookNotary` holds the configuration -- the secret, the hashing +algorithm, and where signatures are stored -- and hands out sessions. Signing +happens through a session, which opens a signature store and closes it when the +block exits:: + + from nbformat.sign import NotebookNotary + + notary = NotebookNotary() + with notary.open_session() as session: + session.sign(nb) + +A notary can open as many sessions as you like, one after another or at the same +time; each session has its own store. + +.. versionadded:: 5.12 + :meth:`NotebookNotary.open_session` and :class:`NotarySession`. Signing + through the notary itself still works, but leaves its store open until + :meth:`NotebookNotary.close` is called, and warns to that effect. + .. autoclass:: NotebookNotary + .. automethod:: open_session + + .. automethod:: close + + .. automethod:: sign + + .. automethod:: unsign + + .. automethod:: check_signature + + .. automethod:: mark_cells + + .. automethod:: check_cells + +.. autoclass:: NotarySession + .. automethod:: sign .. automethod:: unsign .. automethod:: check_signature + .. automethod:: compute_signature + .. automethod:: mark_cells .. automethod:: check_cells diff --git a/nbformat/sign.py b/nbformat/sign.py index 40180f4f..caaf1a8d 100644 --- a/nbformat/sign.py +++ b/nbformat/sign.py @@ -7,6 +7,7 @@ import hashlib import os import sys +import threading import typing as t import warnings from collections import OrderedDict @@ -333,13 +334,194 @@ def signature_removed(nb): nb["metadata"]["signature"] = save_signature +def _compute_signature(nb: t.Any, secret: bytes, digestmod: t.Any) -> str: + """Compute a notebook's signature + + by hashing the entire contents of the notebook via HMAC digest. + """ + hmac = HMAC(secret, digestmod=digestmod) + # don't include the previous hash in the content to hash + with signature_removed(nb): + # sign the whole thing + for b in yield_everything(nb): + hmac.update(b) + + return hmac.hexdigest() + + +def _mark_cells(nb, trusted): + """Mark cells as trusted if the notebook's signature can be verified + + Sets ``cell.metadata.trusted = True | False`` on all code cells, + depending on the *trusted* parameter. + + This function is the inverse of check_cells. + """ + if nb.nbformat < 3: + return + + for cell in yield_code_cells(nb): + cell["metadata"]["trusted"] = trusted + + +def _check_cell(cell, nbformat_version): + """Do we trust an individual cell? + + Return True if: + + - cell is explicitly trusted + - cell has no potentially unsafe rich output + + If a cell has no output, or only simple print statements, + it will always be trusted. + """ + # explicitly trusted + if cell["metadata"].pop("trusted", False): + return True + + # explicitly safe output + if nbformat_version >= 4: + unsafe_output_types = ["execute_result", "display_data"] + safe_keys = {"output_type", "execution_count", "metadata"} + else: # v3 + unsafe_output_types = ["pyout", "display_data"] + safe_keys = {"output_type", "prompt_number", "metadata"} + + for output in cell["outputs"]: + output_type = output["output_type"] + if output_type in unsafe_output_types: + # if there are any data keys not in the safe whitelist + output_keys = set(output) + if output_keys.difference(safe_keys): + return False + + return True + + +def _check_cells(nb: t.Any, check_cell: t.Callable[[t.Any, int], bool] = _check_cell) -> bool: + """Return whether all code cells are trusted. + + A cell is trusted if the 'trusted' field in its metadata is truthy, or + if it has no potentially unsafe outputs. + If there are no code cells, return True. + + This function is the inverse of mark_cells. + """ + if nb.nbformat < 3: + return False + trusted = True + for cell in yield_code_cells(nb): + # only distrust a cell if it actually has some output to distrust + if not check_cell(cell, nb.nbformat): + trusted = False + + return trusted + + +class NotarySession: + """Signing operations over a signature store that the session owns. + + A session is what :meth:`NotebookNotary.open_session` hands out:: + + with notary.open_session() as session: + session.sign(nb) + + The session opens its store (by calling the *store_factory* it is given) + and closes it when the ``with`` block exits: the store's lifetime is the + session's lifetime. A session therefore has no ``close()`` of its own -- + it is not meant to be kept alive past the block, which is the whole point + of the split. (Holding on to one and using it afterwards is not prevented, + but with the default SQLite store it fails, since the store is closed.) + + A session is self-contained: it holds the secret and algorithm it was + created with, and does not consult the notary afterwards. Changing the + notary's configuration therefore affects later sessions, not open ones. + + .. versionadded:: 5.12 + """ + + def __init__( + self, + store_factory: t.Callable[[], SignatureStore], + secret: bytes, + algorithm: str, + digestmod: t.Any = None, + ) -> None: + """Open a session. Use :meth:`NotebookNotary.open_session` instead of calling this.""" + self._secret = secret + self._algorithm = algorithm + self._digestmod = digestmod if digestmod is not None else getattr(hashlib, algorithm) + self._store = store_factory() + + def compute_signature(self, nb: t.Any) -> str: + """Compute a notebook's signature.""" + return _compute_signature(nb, self._secret, self._digestmod) + + def check_signature(self, nb: t.Any) -> bool: + """Check a notebook's stored signature. + + If a signature is stored in the notebook's metadata, a new signature is + computed and compared with the stored value. + + Returns True if the signature is found and matches, False otherwise. + + The following conditions must all be met for a notebook to be trusted: + - a signature is stored in the form 'scheme:hexdigest' + - the stored scheme matches the requested scheme + - the requested scheme is available from hashlib + - the computed hash from notebook_signature matches the stored hash + """ + if nb.nbformat < 3: + return False + signature = self.compute_signature(nb) + return bool(self._store.check_signature(signature, self._algorithm)) + + def sign(self, nb: t.Any) -> None: + """Sign a notebook, indicating that its output is trusted on this machine. + + Stores hash algorithm and hmac digest in a local database of trusted notebooks. + """ + if nb.nbformat < 3: + return + signature = self.compute_signature(nb) + self._store.store_signature(signature, self._algorithm) + + def unsign(self, nb: t.Any) -> None: + """Ensure that a notebook is untrusted + + by removing its signature from the trusted database, if present. + """ + signature = self.compute_signature(nb) + self._store.remove_signature(signature, self._algorithm) + + def mark_cells(self, nb: t.Any, trusted: bool) -> None: + """Mark cells as trusted if the notebook's signature can be verified. + + This method is the inverse of check_cells. + """ + _mark_cells(nb, trusted) + + def check_cells(self, nb: t.Any) -> bool: + """Return whether all code cells are trusted. + + This method is the inverse of mark_cells. + """ + return bool(_check_cells(nb)) + + def _close(self) -> None: + """Close the store. Called for you when the ``with`` block exits.""" + self._store.close() + + class NotebookNotaryContext(t.Protocol): - """The operations available on a :class:`NotebookNotary` used as a context manager. + """The signing operations, as a structural type. + + This protocol was introduced in nbformat 5.11 to type what + ``NotebookNotary.__enter__`` returns. It is kept so that annotations + written against it keep working: both :class:`NotarySession` and + :class:`NotebookNotary` satisfy it structurally. - This is what ``NotebookNotary.__enter__`` returns. It deliberately omits - ``close()``, ``__enter__``, and ``__exit__`` so that a type checker flags - attempts to close or re-enter the notary from within its own ``with`` - block. + .. versionadded:: 5.11 """ def compute_signature(self, nb: t.Any) -> str: @@ -458,47 +640,202 @@ def _secret_default(self): def __init__(self, **kwargs): """Initialize the notary.""" super().__init__(**kwargs) - self.store = self.store_factory() - self._used_as_context_manager = False - self._warned_not_context_manager = False + self._store: SignatureStore | None = None + # whether the store was created by us (and is therefore ours to close) + self._store_owned = True + self._store_lock = threading.Lock() + self._closed = False + self._in_context = False + self._warned_prefer_session = False + + @contextmanager + def open_session(self) -> t.Iterator[NotarySession]: + """Open a :class:`NotarySession`, which owns a signature store. + + This is the way to use a notary:: + + with notary.open_session() as session: + session.sign(nb) + + Each call opens its own store, so sessions are independent: they may be + nested, used one after another, or held by different threads. The store + is closed when the block exits. + + .. versionadded:: 5.12 + """ + session = NotarySession(self.store_factory, self.secret, self.algorithm, self.digestmod) + try: + yield session + finally: + session._close() + + @property + def store(self) -> SignatureStore: + """The shared signature store used by the direct (non-session) API. + + Sessions returned by :meth:`open_session` have their own store; this one + backs direct calls such as ``notary.sign(nb)``. It is created lazily, and + re-created on demand after an internal close (such as leaving a ``with`` + block), so that code which keeps using the notary keeps working. + + Once :meth:`close` has been called explicitly, it is not re-created: + accessing it raises :exc:`RuntimeError`. + + .. versionchanged:: 5.12 + Created on first use rather than in ``__init__``, and re-created + after an internal close. A store assigned here is never closed by + the notary. + """ + with self._store_lock: + if self._closed: + msg = ( + "This NotebookNotary has been closed. Use a new notary, or " + "`with notary.open_session() as session:` for a store whose " + "lifetime is the block." + ) + raise RuntimeError(msg) + if self._store is None: + self._store = self.store_factory() + self._store_owned = True + return self._store + + @store.setter + def store(self, store: SignatureStore) -> None: + # A store assigned by the caller is owned by the caller: we hand out + # that same object for the notary's whole lifetime and never close it. + with self._store_lock: + self._store = store + self._store_owned = False def close(self): - """Close the notary's signature store. + """Close the notary's shared signature store, for good. + + Should be called when the notary is no longer needed, to release any + resources (e.g. database connections) held by the store. The store is + not re-created afterwards: using the notary's direct API again raises + :exc:`RuntimeError`. Sessions opened by :meth:`open_session` are + unaffected, since they own their stores. + + A store that was assigned to :attr:`store` by the caller is left open, + since its lifetime belongs to whoever created it. - Should be called when the notary is no longer needed, to release - any resources (e.g. database connections) held by the store. + .. versionadded:: 5.11 + + .. versionchanged:: 5.12 + The store is no longer re-created afterwards, and a caller-assigned + store is left open. """ - self.store.close() + self._close_store() + self._closed = True + + def _close_store(self): + """Close the shared store, leaving the notary usable. + + This is the internal close, used when leaving a ``with`` block: code + written before the notary was a context manager may go on using it + afterwards, and should keep working. + """ + with self._store_lock: + if not self._store_owned: + return + store, self._store = self._store, None + if store is None: + return + try: + store.close() + except BaseException: + # keep hold of a store we failed to close, so a retry can reach it + with self._store_lock: + if self._store is None: + self._store = store + raise + + def __enter__(self) -> NotebookNotary: # noqa: PYI034 (typing.Self needs 3.11) + """Enter the notary's context, as introduced in nbformat 5.11. + + This returns the notary itself, and leaving the block closes the shared + store, exactly as it did in 5.11. - def __enter__(self) -> NotebookNotaryContext: - """Enter the notary's context, marking it as used within a `with` block.""" - self._used_as_context_manager = True + Open a session instead -- a session is the thing with a lifetime, and it + can be opened as many times as you like:: + + with notary.open_session() as session: + session.sign(nb) + + The notary is not re-entrant: it holds one shared store, so nesting + ``with notary:`` blocks closes that store when the inner block exits. + + .. versionadded:: 5.11 + + .. versionchanged:: 5.12 + Warns, recommending :meth:`open_session`. Leaving the block no + longer makes the notary unusable. + """ + self._warn_prefer_session( + "Using a NotebookNotary itself as a context manager is discouraged. " + "Open a session instead:" + ) + self._in_context = True return self def __exit__(self, *exc_info): - """Exit the notary's context, closing its signature store.""" - self.close() + """Exit the notary's context, closing the shared store. + + The notary stays usable afterwards: unlike an explicit :meth:`close`, + leaving a block is not the caller saying they are done with it. - def _warn_if_not_context_manager(self): - """Warn once if the store is accessed without using this notary as a context manager. + .. versionadded:: 5.11 - Using ``NotebookNotary`` outside of a ``with`` block is deprecated as - of nbformat 5.11, since it makes it easy to forget to release the - resources (e.g. database connections) held by the store. + .. versionchanged:: 5.12 + Leaves the notary usable instead of closing it for good. """ - if self._used_as_context_manager or self._warned_not_context_manager: + self._in_context = False + self._close_store() + + def _session(self) -> NotarySession: + """A session over the shared store, for the non-session API. + + Unlike :meth:`open_session`, this borrows the store held by the notary + rather than opening one, so the direct methods keep sharing a single + store across calls the way they always have. + """ + session = NotarySession(lambda: self.store, self.secret, self.algorithm, self.digestmod) + # a subclass may override compute_signature; keep honouring that here + session.compute_signature = self.compute_signature # type:ignore[method-assign] + return session + + def _warn_prefer_session(self, lead, stacklevel=4): + """Point at :meth:`open_session`, once per notary.""" + if self._warned_prefer_session: return - self._warned_not_context_manager = True + self._warned_prefer_session = True warnings.warn( - "Using NotebookNotary without a `with` block is deprecated as of " - "nbformat 5.11. Use it as a context manager instead, e.g.:\n\n" - " with NotebookNotary() as notary:\n" - " notary.sign(nb)\n\n" - "so that the underlying signature store is properly closed.", - PendingDeprecationWarning, - stacklevel=3, + f"{lead}\n\n" + " with NotebookNotary().open_session() as session:\n" + " session.sign(nb)\n\n" + "A session closes the underlying signature store when its block " + "exits, so nothing is left open for you to remember to close.", + FutureWarning, + stacklevel=stacklevel, ) + def _warn_if_no_session(self): + """Warn once if a store operation is performed without a session. + + Calling the signing methods on the notary itself leaves the store open + until :meth:`close` is called, which is easy to forget, so we point at + :meth:`open_session` instead. + """ + if self._in_context: + return + self._warn_prefer_session( + "Prefer opening a session over calling " + "NotebookNotary.{sign,unsign,check_signature} directly:" + ) + + # Alias for the 5.11 name. + _warn_if_not_context_manager = _warn_if_no_session + def _write_secret_file(self, secret): """write my secret to my secret_file""" self.log.info("Writing notebook-signing key to %s", self.secret_file) @@ -514,15 +851,11 @@ def compute_signature(self, nb): """Compute a notebook's signature by hashing the entire contents of the notebook via HMAC digest. - """ - hmac = HMAC(self.secret, digestmod=self.digestmod) - # don't include the previous hash in the content to hash - with signature_removed(nb): - # sign the whole thing - for b in yield_everything(nb): - hmac.update(b) - return hmac.hexdigest() + This needs no store, so a session is not needed; it is also available as + :meth:`NotarySession.compute_signature`. + """ + return _compute_signature(nb, self.secret, self.digestmod) def check_signature(self, nb): """Check a notebook's stored signature @@ -537,32 +870,46 @@ def check_signature(self, nb): - the stored scheme matches the requested scheme - the requested scheme is available from hashlib - the computed hash from notebook_signature matches the stored hash + + .. note:: + Prefer ``with notary.open_session() as session: + session.check_signature(nb)``, which closes the store for you. + + .. versionchanged:: 5.12 + Warns once, recommending :meth:`open_session`. """ - self._warn_if_not_context_manager() - if nb.nbformat < 3: - return False - signature = self.compute_signature(nb) - return self.store.check_signature(signature, self.algorithm) + self._warn_if_no_session() + return self._session().check_signature(nb) def sign(self, nb): """Sign a notebook, indicating that its output is trusted on this machine Stores hash algorithm and hmac digest in a local database of trusted notebooks. + + .. note:: + Prefer ``with notary.open_session() as session: session.sign(nb)``, + which closes the store for you. + + .. versionchanged:: 5.12 + Warns once, recommending :meth:`open_session`. """ - self._warn_if_not_context_manager() - if nb.nbformat < 3: - return - signature = self.compute_signature(nb) - self.store.store_signature(signature, self.algorithm) + self._warn_if_no_session() + self._session().sign(nb) def unsign(self, nb): """Ensure that a notebook is untrusted by removing its signature from the trusted database, if present. + + .. note:: + Prefer ``with notary.open_session() as session: session.unsign(nb)``, + which closes the store for you. + + .. versionchanged:: 5.12 + Warns once, recommending :meth:`open_session`. """ - self._warn_if_not_context_manager() - signature = self.compute_signature(nb) - self.store.remove_signature(signature, self.algorithm) + self._warn_if_no_session() + self._session().unsign(nb) def mark_cells(self, nb, trusted): """Mark cells as trusted if the notebook's signature can be verified @@ -571,13 +918,12 @@ def mark_cells(self, nb, trusted): depending on the *trusted* parameter. This will typically be the return value from ``self.check_signature(nb)``. - This function is the inverse of check_cells - """ - if nb.nbformat < 3: - return + This function is the inverse of check_cells. - for cell in yield_code_cells(nb): - cell["metadata"]["trusted"] = trusted + This needs no store, so a session is not needed; it is also available as + :meth:`NotarySession.mark_cells`. + """ + _mark_cells(nb, trusted) def _check_cell(self, cell, nbformat_version): """Do we trust an individual cell? @@ -590,27 +936,7 @@ def _check_cell(self, cell, nbformat_version): If a cell has no output, or only simple print statements, it will always be trusted. """ - # explicitly trusted - if cell["metadata"].pop("trusted", False): - return True - - # explicitly safe output - if nbformat_version >= 4: - unsafe_output_types = ["execute_result", "display_data"] - safe_keys = {"output_type", "execution_count", "metadata"} - else: # v3 - unsafe_output_types = ["pyout", "display_data"] - safe_keys = {"output_type", "prompt_number", "metadata"} - - for output in cell["outputs"]: - output_type = output["output_type"] - if output_type in unsafe_output_types: - # if there are any data keys not in the safe whitelist - output_keys = set(output) - if output_keys.difference(safe_keys): - return False - - return True + return _check_cell(cell, nbformat_version) def check_cells(self, nb): """Return whether all code cells are trusted. @@ -620,16 +946,13 @@ def check_cells(self, nb): If there are no code cells, return True. This function is the inverse of mark_cells. - """ - if nb.nbformat < 3: - return False - trusted = True - for cell in yield_code_cells(nb): - # only distrust a cell if it actually has some output to distrust - if not self._check_cell(cell, nb.nbformat): - trusted = False - return trusted + This needs no store, so a session is not needed; it is also available as + :meth:`NotarySession.check_cells`. + """ + # dispatch through self._check_cell, so that subclasses overriding it + # keep working + return _check_cells(nb, self._check_cell) trust_flags: dict[str, t.Any] = { @@ -677,22 +1000,22 @@ def _config_file_name_default(self): def _notary_default(self): return NotebookNotary(parent=self, data_dir=self.data_dir) - def sign_notebook_file(self, notebook_path): - """Sign a notebook from the filesystem""" + def sign_notebook_file(self, notebook_path, *, session): + """Sign a notebook from the filesystem, using an open session""" if not Path(notebook_path).exists(): self.log.error("Notebook missing: %s", notebook_path) self.exit(1) with Path(notebook_path).open(encoding="utf8") as f: nb = read(f, NO_CONVERT) - self.sign_notebook(nb, notebook_path) + self.sign_notebook(nb, notebook_path, session=session) - def sign_notebook(self, nb, notebook_path=""): - """Sign a notebook that's been loaded""" - if self.notary.check_signature(nb): + def sign_notebook(self, nb, notebook_path="", *, session): + """Sign a notebook that's been loaded, using an open session""" + if session.check_signature(nb): print("Notebook already signed: %s" % notebook_path) # noqa: T201 else: print("Signing notebook: %s" % notebook_path) # noqa: T201 - self.notary.sign(nb) + session.sign(nb) def generate_new_key(self): """Generate a new notebook signature key""" @@ -701,22 +1024,23 @@ def generate_new_key(self): def start(self): """Start the trust notebook app.""" - with self.notary: - if self.reset: - if Path(self.notary.db_file).exists(): - print("Removing trusted signature cache: %s" % self.notary.db_file) # noqa: T201 - Path(self.notary.db_file).unlink() - self.generate_new_key() - return + if self.reset: + # don't open a store we are about to delete + if Path(self.notary.db_file).exists(): + print("Removing trusted signature cache: %s" % self.notary.db_file) # noqa: T201 + Path(self.notary.db_file).unlink() + self.generate_new_key() + return + with self.notary.open_session() as session: if not self.extra_args: self.log.debug("Reading notebook from stdin") nb_s = sys.stdin.read() assert isinstance(nb_s, str) nb = reads(nb_s, NO_CONVERT) - self.sign_notebook(nb, "") + self.sign_notebook(nb, "", session=session) else: for notebook_path in self.extra_args: - self.sign_notebook_file(notebook_path) + self.sign_notebook_file(notebook_path, session=session) main = TrustNotebookApp.launch_instance diff --git a/tests/test_sign.py b/tests/test_sign.py index 84cc03e3..a6368d3e 100644 --- a/tests/test_sign.py +++ b/tests/test_sign.py @@ -8,6 +8,7 @@ import copy import os import shutil +import sqlite3 import sys import tempfile import time @@ -32,7 +33,11 @@ def setUp(self): secret=b"secret", data_dir=self.data_dir, ) - self.notary.__enter__() + # most of these tests exercise the direct (non-session) API; entering + # the notary keeps that from warning on every call + with warnings.catch_warnings(): + warnings.simplefilter("ignore", FutureWarning) + self.notary.__enter__() with self.fopen("test3.ipynb", "r") as f: self.nb = read(f, as_version=4) with self.fopen("test3.ipynb", "r") as f: @@ -47,24 +52,26 @@ def test_invalid_db_file(self): with open(invalid_sql_file, "w", encoding="utf-8") as tempfile: tempfile.write("[invalid data]") - with sign.NotebookNotary( + invalid_notary = sign.NotebookNotary( db_file=invalid_sql_file, secret=b"secret", - ) as invalid_notary: - invalid_notary.sign(self.nb) + ) + with invalid_notary.open_session() as session: + session.sign(self.nb) + invalid_notary.close() testpath.assert_isfile(os.path.join(self.data_dir, invalid_sql_file)) testpath.assert_isfile(os.path.join(self.data_dir, invalid_sql_file + ".bak")) - def test_not_using_context_manager_warns(self): - """Using the store without entering the notary as a context manager warns once.""" + def test_not_using_a_session_warns(self): + """Signing without a session warns once, pointing at open_session().""" notary = sign.NotebookNotary( db_file=":memory:", secret=b"secret", data_dir=self.data_dir, ) try: - with pytest.warns(PendingDeprecationWarning, match="context manager"): + with pytest.warns(FutureWarning, match="Prefer opening a session"): notary.sign(self.nb) # only warns once per instance with warnings.catch_warnings(): @@ -73,16 +80,253 @@ def test_not_using_context_manager_warns(self): finally: notary.close() - def test_using_context_manager_does_not_warn(self): + def test_using_notary_as_context_manager_warns(self): + """The notary is usable as a context manager, but points at sessions.""" + notary = sign.NotebookNotary( + db_file=":memory:", + secret=b"secret", + data_dir=self.data_dir, + ) + with pytest.warns(FutureWarning, match="itself as a context manager"): + ctx = notary.__enter__() + with warnings.catch_warnings(): + # having warned on the way in, it does not warn again per call + warnings.simplefilter("error") + self.assertIs(ctx, notary) + notary.sign(self.nb) + self.assertTrue(notary.check_signature(self.nb)) + notary.__exit__(None, None, None) + + def test_using_a_session_does_not_warn(self): with warnings.catch_warnings(): warnings.simplefilter("error") - with sign.NotebookNotary( + notary = sign.NotebookNotary( db_file=":memory:", secret=b"secret", data_dir=self.data_dir, - ) as notary: - notary.sign(self.nb) + ) + with notary.open_session() as session: + session.sign(self.nb) + self.assertTrue(session.check_signature(self.nb)) + + def test_open_session(self): + db_file = os.path.join(self.data_dir, "sessions.db") + notary = sign.NotebookNotary(db_file=db_file, secret=b"secret", data_dir=self.data_dir) + with warnings.catch_warnings(): + warnings.simplefilter("error") + with notary.open_session() as session: + self.assertFalse(session.check_signature(self.nb)) + session.sign(self.nb) + self.assertTrue(session.check_signature(self.nb)) + # the notary is reusable: a later session sees the persisted signature + with notary.open_session() as session: + self.assertTrue(session.check_signature(self.nb)) + notary.close() + + def test_sessions_are_independent(self): + """Each session owns its own store; closing one leaves the others alone.""" + db_file = os.path.join(self.data_dir, "nested.db") + notary = sign.NotebookNotary(db_file=db_file, secret=b"secret", data_dir=self.data_dir) + with notary.open_session() as outer: + outer.sign(self.nb) + with notary.open_session() as inner: + self.assertIsNot(inner._store, outer._store) + self.assertTrue(inner.check_signature(self.nb)) + # closing the inner session must not disturb the outer one + self.assertTrue(outer.check_signature(self.nb)) + notary.close() + + def test_session_public_api(self): + """A session exposes its operations and nothing else: no store, no close.""" + notary = sign.NotebookNotary(db_file=":memory:", secret=b"secret", data_dir=self.data_dir) + with notary.open_session() as session: + self.assertEqual( + sorted(name for name in dir(session) if not name.startswith("_")), + [ + "check_cells", + "check_signature", + "compute_signature", + "mark_cells", + "sign", + "unsign", + ], + ) + notary.close() + + def test_session_snapshots_secret(self): + """A session keeps the secret it was opened with; the notary can change under it.""" + notary = sign.NotebookNotary(db_file=":memory:", secret=b"secret", data_dir=self.data_dir) + with notary.open_session() as session: + before = session.compute_signature(self.nb) + notary.secret = b"different" + self.assertEqual(session.compute_signature(self.nb), before) + # a later session picks the new secret up + with notary.open_session() as session: + self.assertNotEqual(session.compute_signature(self.nb), before) + notary.close() + + def test_session_snapshots_algorithm(self): + """The algorithm a session was opened with drives its digest.""" + notary = sign.NotebookNotary(db_file=":memory:", secret=b"secret", data_dir=self.data_dir) + with notary.open_session() as session: + sha256 = session.compute_signature(self.nb) + notary.algorithm = "sha512" + self.assertEqual(session.compute_signature(self.nb), sha256) + with notary.open_session() as session: + sha512 = session.compute_signature(self.nb) + self.assertEqual(len(sha256), 64) + self.assertEqual(len(sha512), 128) + notary.close() + + def test_notary_context_manager_still_works(self): + """The 5.11 `with notary as ...` spelling keeps working, and closes the store.""" + stores = [] + + def factory(): + stores.append(sign.SQLiteSignatureStore(":memory:")) + return stores[-1] + + notary = sign.NotebookNotary( + secret=b"secret", + data_dir=self.data_dir, + store_factory=factory, + ) + with warnings.catch_warnings(): + warnings.simplefilter("ignore", FutureWarning) + with notary as entered: + # 5.11 yielded the notary itself, and code inside the block may + # rely on that, both for signing and for its configuration + self.assertIs(entered, notary) + self.assertEqual(entered.algorithm, "sha256") + entered.sign(self.nb) self.assertTrue(notary.check_signature(self.nb)) + # reusable: a second block opens a fresh store + with notary: + notary.sign(self.nb) + self.assertEqual(len(stores), 2) + for store in stores: + with pytest.raises(sqlite3.ProgrammingError): + store.check_signature("abc", "sha256") + + def test_session_cell_helpers(self): + """mark_cells/check_cells are available on the session and actually work.""" + notary = sign.NotebookNotary(db_file=":memory:", secret=b"secret", data_dir=self.data_dir) + with notary.open_session() as session: + session.mark_cells(self.nb, False) + self.assertFalse(session.check_cells(self.nb)) + session.mark_cells(self.nb, True) + self.assertTrue(session.check_cells(self.nb)) + notary.close() + + def test_direct_calls_share_one_store(self): + """The non-session API keeps using a single store across calls.""" + stores = [] + + def factory(): + stores.append(sign.MemorySignatureStore()) + return stores[-1] + + notary = sign.NotebookNotary( + secret=b"secret", data_dir=self.data_dir, store_factory=factory + ) + with warnings.catch_warnings(): + warnings.simplefilter("ignore", FutureWarning) + notary.sign(self.nb) + self.assertTrue(notary.check_signature(self.nb)) + notary.unsign(self.nb) + notary.close() + self.assertEqual(len(stores), 1) + + def test_subclass_hooks_are_honoured(self): + """Subclasses overriding compute_signature/_check_cell still take effect.""" + calls = [] + + class Sub(sign.NotebookNotary): + def compute_signature(self, nb): + calls.append("compute_signature") + return super().compute_signature(nb) + + def _check_cell(self, cell, nbformat_version): + calls.append("_check_cell") + return super()._check_cell(cell, nbformat_version) + + notary = Sub(db_file=":memory:", secret=b"secret", data_dir=self.data_dir) + with warnings.catch_warnings(): + warnings.simplefilter("ignore", FutureWarning) + notary.sign(self.nb) + self.assertTrue(notary.check_signature(self.nb)) + notary.unsign(self.nb) + notary.check_cells(self.nb) + notary.close() + self.assertEqual(calls.count("compute_signature"), 3) + self.assertIn("_check_cell", calls) + + def test_assigned_store_is_not_closed(self): + """A store assigned by the caller is owned by the caller.""" + # a real store, whose close() is observable (unlike MemorySignatureStore's) + store = sign.SQLiteSignatureStore(":memory:") + notary = sign.NotebookNotary( + db_file=":memory:", + secret=b"secret", + data_dir=self.data_dir, + ) + notary.store = store + with warnings.catch_warnings(): + warnings.simplefilter("ignore", FutureWarning) + notary.sign(self.nb) + notary.close() + # even an explicit close leaves a caller-owned store alone + self.assertTrue(store.check_signature(notary.compute_signature(self.nb), "sha256")) + store.close() + + def test_reusable_after_internal_close(self): + """Leaving a `with` block closes the store but leaves the notary working.""" + notary = sign.NotebookNotary( + db_file=":memory:", + secret=b"secret", + data_dir=self.data_dir, + ) + with warnings.catch_warnings(): + warnings.simplefilter("ignore", FutureWarning) + with notary: + notary.sign(self.nb) + self.assertIsNone(notary._store) + # code that goes on using the notary keeps working: a fresh store + # is created on demand + self.assertFalse(notary.check_signature(self.nb)) + notary.sign(self.nb) + self.assertTrue(notary.check_signature(self.nb)) + notary.close() + + def test_explicit_close_is_final(self): + """An explicit close() means done: the store is not re-created.""" + notary = sign.NotebookNotary( + db_file=":memory:", + secret=b"secret", + data_dir=self.data_dir, + ) + with warnings.catch_warnings(): + warnings.simplefilter("ignore", FutureWarning) + notary.sign(self.nb) + notary.close() + with pytest.raises(RuntimeError, match="has been closed"): + notary.check_signature(self.nb) + with pytest.raises(RuntimeError, match="has been closed"): + _ = notary.store + # closing twice is harmless, and sessions are unaffected + notary.close() + with notary.open_session() as session: + self.assertFalse(session.check_signature(self.nb)) + + def test_store_free_methods_need_no_session(self): + """compute_signature/mark_cells/check_cells open no store and do not warn.""" + notary = sign.NotebookNotary(db_file=":memory:", secret=b"secret", data_dir=self.data_dir) + with warnings.catch_warnings(): + warnings.simplefilter("error") + notary.compute_signature(self.nb) + notary.mark_cells(self.nb, True) + notary.check_cells(self.nb) + self.assertIsNone(notary._store) def test_algorithms(self): last_sig = "" @@ -270,6 +514,25 @@ def sign_stdin(nb): out = sign_stdin(self.nb3) self.assertIn("already signed: ", out) + def test_trust_reset(self): + """`jupyter trust --reset` deletes the cache without reopening it.""" + app = sign.TrustNotebookApp(data_dir=self.data_dir) + app.notary = sign.NotebookNotary( + db_file=os.path.join(self.data_dir, "reset.db"), + secret_file=os.path.join(self.data_dir, "reset_secret"), + data_dir=self.data_dir, + ) + with app.notary.open_session() as session: + session.sign(self.nb) + testpath.assert_isfile(app.notary.db_file) + + app.reset = True + app.start() + + testpath.assert_not_path_exists(app.notary.db_file) + # the store must not have been reopened, which would recreate the file + self.assertIsNone(app.notary._store) + def test_config_store(): store = sign.MemorySignatureStore()