derivepassphrase.git

@@ -16,8 +16,10 @@ Warning:

 from __future__ import annotations

 import base64

+import contextlib

 import copy

 import enum

+import hashlib

 import json

 import logging

 import os

@@ -159,6 +161,7 @@ def shell_complete_service(

 config_filename_table = {

     None: '.',

+    'write lock': '',

     'vault': 'vault.json',

     'user configuration': 'config.toml',

     # TODO(the-13th-letter): Remove the old settings.json file.

@@ -167,6 +170,129 @@ config_filename_table = {

     'notes backup': 'old-notes.txt',

 }

+LOCK_SIZE = 4096

+"""

+The size of the record to lock at the beginning of the file, for locking

+implementations that lock byte ranges instead of whole files.

+

+While POSIX specifies that [`fcntl`][] locks shall support a size of zero to

+denote "any conceivable file size", the locking system available in

+[`msvcrt`][] does not support this, and requires an explicit size.

+"""

+

+

+@contextlib.contextmanager

+def configuration_mutex() -> Iterator[None]:

+    """Enter a mutually exclusive context for configuration writes.

+

+    Within this context, no other cooperating instance of

+    `derivepassphrase` will attempt to write to its configuration

+    directory.  We achieve this by locking a specific temporary file

+    (whose name depends on the location of the configuration directory)

+    for the duration of the context.

+

+    Note: Locking specifics

+        The directory for the lock file is determined via

+        [`get_tempdir`][].  The lock filename is

+        `derivepassphrase-lock-<hash>.txt`, where `<hash>` is computed

+        as follows.  First, canonicalize the path to the configuration

+        directory with [`pathlib.Path.resolve`][].  Then encode the

+        result as per the filesystem encoding ([`os.fsencode`][]), and

+        hash it with SHA256.  Finally, convert the result to standard

+        base32 and use the first twelve characters, in lowercase, as

+        `<hash>`.

+

+        We use [`msvcrt.locking`][] on Windows platforms (`sys.platform

+        == "win32"`) and [`fcntl.flock`][] on all others.  All locks are

+        exclusive locks.  If the locking system requires a byte range,

+        we lock the first [`LOCK_SIZE`][] bytes.  For maximum

+        portability between locking implementations, we first open the

+        lock file for writing, which is sometimes necessary to lock

+        a file exclusively.  Thus locking will fail if we lack

+        permission to write to an already-existing lockfile.

+

+    """

+    lock_func: Callable[[int], None]

+    unlock_func: Callable[[int], None]

+    if sys.platform == 'win32':  # pragma: no cover

+        import msvcrt  # noqa: PLC0415

+

+        locking = msvcrt.locking

+        LK_LOCK = msvcrt.LK_LOCK  # noqa: N806

+        LK_UNLCK = msvcrt.LK_UNLCK  # noqa: N806

+

+        def lock_func(fileobj: int) -> None:

+            locking(fileobj, LK_LOCK, LOCK_SIZE)

+

+        def unlock_func(fileobj: int) -> None:

+            locking(fileobj, LK_UNLCK, LOCK_SIZE)

+

+    else:

+        import fcntl  # noqa: PLC0415

+

+        flock = fcntl.flock

+        LOCK_EX = fcntl.LOCK_EX  # noqa: N806

+        LOCK_UN = fcntl.LOCK_UN  # noqa: N806

+

+        def lock_func(fileobj: int) -> None:

+            flock(fileobj, LOCK_EX)

+

+        def unlock_func(fileobj: int) -> None:

+            flock(fileobj, LOCK_UN)

+

+    write_lock_file = config_filename('write lock')

+    write_lock_file.touch()

+    with write_lock_file.open('wb') as lock_fileobj:

+        lock_func(lock_fileobj.fileno())

+        try:

+            yield

+        finally:

+            unlock_func(lock_fileobj.fileno())

+

+

+def get_tempdir() -> pathlib.Path:

+    """Return a suitable temporary directory.

+

+    We implement the same algorithm as [`tempfile.gettempdir`][], except

+    that we default to the `derivepassphrase` configuration directory

+    instead of the current directory if no other choice is suitable, and

+    that we return [`pathlib.Path`][] objects directly.

+

+    """

+    paths_to_try: list[pathlib.PurePath] = []

+    env_paths_to_try = [

+        os.getenv('TMPDIR'),

+        os.getenv('TEMP'),

+        os.getenv('TMP'),

+    ]

+    paths_to_try.extend(

+        pathlib.PurePath(p) for p in env_paths_to_try if p is not None

+    )

+    posix_paths_to_try = [

+        pathlib.PurePosixPath('/tmp'),  # noqa: S108

+        pathlib.PurePosixPath('/var/tmp'),  # noqa: S108

+        pathlib.PurePosixPath('/usr/tmp'),

+    ]

+    windows_paths_to_try = [

+        pathlib.PureWindowsPath(r'C:\TEMP'),

+        pathlib.PureWindowsPath(r'C:\TMP'),

+        pathlib.PureWindowsPath(r'\TEMP'),

+        pathlib.PureWindowsPath(r'\TMP'),

+    ]

+    paths_to_try.extend(

+        windows_paths_to_try if sys.platform == 'win32' else posix_paths_to_try

+    )

+    for p in paths_to_try:

+        path = pathlib.Path(p)

+        try:

+            points_to_dir = path.is_dir()

+        except OSError:

+            continue

+        else:

+            if points_to_dir:

+                return path.resolve(strict=True)

+    return config_filename(subsystem=None)

+

 def config_filename(

     subsystem: str | None = 'old settings.json',

@@ -201,6 +327,14 @@ def config_filename(

         os.getenv(PROG_NAME.upper() + '_PATH')

         or click.get_app_dir(PROG_NAME, force_posix=True)

     )

+    if subsystem == 'write lock':

+        path_hash = base64.b32encode(

+            hashlib.sha256(os.fsencode(path.resolve())).digest()

+        )

+        path_hash_text = path_hash[:12].lower().decode('ASCII')

+        temp_path = get_tempdir()

+        filename_ = f'derivepassphrase-lock-{path_hash_text}.txt'

+        return temp_path / filename_

     try:

         filename = config_filename_table[subsystem]

     except (KeyError, TypeError):  # pragma: no cover

@@ -10,6 +10,7 @@ from __future__ import annotations

 import base64

 import collections

+import contextlib

 import functools

 import json

 import logging

@@ -34,6 +35,7 @@ from derivepassphrase._internals import cli_messages as _msg

 if TYPE_CHECKING:

     from collections.abc import (

+        Callable,

         Sequence,

     )

@@ -1029,7 +1031,21 @@ def derivepassphrase_vault(  # noqa: C901,PLR0912,PLR0913,PLR0914,PLR0915

             extra={'color': ctx.color},

         )

-    if delete_service_settings:  # noqa: PLR1702

+    readwrite_ops = [

+        delete_service_settings,

+        delete_globals,

+        clear_all_settings,

+        import_settings,

+        store_config_only,

+    ]

+    mutex: Callable[[], contextlib.AbstractContextManager[None]] = (

+        cli_helpers.configuration_mutex

+        if any(readwrite_ops)

+        else contextlib.nullcontext

+    )

+

+    with mutex():  # noqa: PLR1702

+        if delete_service_settings:

             assert service is not None

             configuration = get_config()

             if service in configuration['services']:

@@ -1231,7 +1247,9 @@ def derivepassphrase_vault(  # noqa: C901,PLR0912,PLR0913,PLR0914,PLR0915

                             this_ctx.parent is not None

                             and this_ctx.parent.info_name is not None

                         ):

-                        prog_name_pieces.appendleft(this_ctx.parent.info_name)

+                            prog_name_pieces.appendleft(

+                                this_ctx.parent.info_name

+                            )

                             this_ctx = this_ctx.parent

                         cli_helpers.print_config_as_sh_script(

                             configuration,

@@ -1281,7 +1299,9 @@ def derivepassphrase_vault(  # noqa: C901,PLR0912,PLR0913,PLR0914,PLR0915

                 },

                 cast(

                     'dict[str, Any]',

-                configuration['services'].get(service, {}) if service else {},

+                    configuration['services'].get(service, {})

+                    if service

+                    else {},

                 ),

                 cast('dict[str, Any]', configuration.get('global', {})),

             )

@@ -1356,7 +1376,9 @@ def derivepassphrase_vault(  # noqa: C901,PLR0912,PLR0913,PLR0914,PLR0915

                 view = (

                     collections.ChainMap(*settings.maps[:2])

                     if service

-                else collections.ChainMap(settings.maps[0], settings.maps[2])

+                    else collections.ChainMap(

+                        settings.maps[0], settings.maps[2]

+                    )

                 )

                 if use_key:

                     view['key'] = key

@@ -1446,7 +1468,9 @@ def derivepassphrase_vault(  # noqa: C901,PLR0912,PLR0913,PLR0914,PLR0915

                             old_notes_value,

                         ])

                     else:

-                    text = old_notes_value or str(notes_legacy_instructions)

+                        text = old_notes_value or str(

+                            notes_legacy_instructions

+                        )

                     notes_value = click.edit(text=text, require_save=False)

                     assert notes_value is not None

                     if (

@@ -1456,7 +1480,9 @@ def derivepassphrase_vault(  # noqa: C901,PLR0912,PLR0913,PLR0914,PLR0915

                         backup_file = cli_helpers.config_filename(

                             subsystem='notes backup'

                         )

-                    backup_file.write_text(old_notes_value, encoding='UTF-8')

+                        backup_file.write_text(

+                            old_notes_value, encoding='UTF-8'

+                        )

                         logger.warning(

                             _msg.TranslatedString(

                                 _msg.WarnMsgTemplate.LEGACY_EDITOR_INTERFACE_NOTES_BACKUP,

@@ -1539,7 +1565,9 @@ def derivepassphrase_vault(  # noqa: C901,PLR0912,PLR0913,PLR0914,PLR0915

                     click.echo(f'{service_notes}\n', err=True, color=ctx.color)

                 click.echo(result.decode('ASCII'), color=ctx.color)

                 if not print_notes_before and service_notes.strip():

-                click.echo(f'\n{service_notes}\n', err=True, color=ctx.color)

+                    click.echo(

+                        f'\n{service_notes}\n', err=True, color=ctx.color

+                    )

 if __name__ == '__main__':

@@ -18,6 +18,7 @@ import re

 import shlex

 import shutil

 import socket

+import tempfile

 import textwrap

 import types

 import warnings

@@ -4886,6 +4887,111 @@ Boo.

         assert _types.is_vault_config(config)

         return self.export_as_sh_helper(config)

+    @hypothesis.given(

+        env_var=strategies.sampled_from(['TMPDIR', 'TEMP', 'TMP']),

+        suffix=strategies.text(

+            tuple(' 0123456789abcdefghijklmnopqrstuvwxyz'),

+            min_size=12,

+            max_size=12,

+        ),

+    )

+    @hypothesis.example(env_var='', suffix='.')

+    def test_140a_get_tempdir(

+        self,

+        env_var: str,

+        suffix: str,

+    ) -> None:

+        """[`cli_helpers.get_tempdir`][] returns a temporary directory.

+

+        If it is not the same as the temporary directory determined by

+        [`tempfile.gettempdir`][], then assert that

+        `tempfile.gettempdir` returned the current directory and

+        `cli_helpers.get_tempdir` returned the configuration directory.

+

+        """

+        runner = click.testing.CliRunner(mix_stderr=False)

+        # TODO(the-13th-letter): Rewrite using parenthesized

+        # with-statements.

+        # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9

+        with contextlib.ExitStack() as stack:

+            monkeypatch = stack.enter_context(pytest.MonkeyPatch.context())

+            stack.enter_context(

+                tests.isolated_vault_config(

+                    monkeypatch=monkeypatch,

+                    runner=runner,

+                    vault_config={'services': {}},

+                )

+            )

+            monkeypatch.delenv('TMPDIR', raising=False)

+            monkeypatch.delenv('TEMP', raising=False)

+            monkeypatch.delenv('TMP', raising=False)

+            if env_var:

+                monkeypatch.setenv(env_var, str(pathlib.Path.cwd() / suffix))

+            system_tempdir = os.fsdecode(tempfile.gettempdir())

+            our_tempdir = cli_helpers.get_tempdir()

+            assert system_tempdir == os.fsdecode(our_tempdir) or (

+                # TODO(the-13th-letter): `tests.isolated_config`

+                # guarantees that `Path.cwd() == config_filename(None)`.

+                # So this sub-branch ought to never trigger in our

+                # tests.

+                system_tempdir == os.getcwd()  # noqa: PTH109

+                and our_tempdir == cli_helpers.config_filename(subsystem=None)

+            )

+

+    def test_140b_get_tempdir_force_default(self) -> None:

+        """[`cli_helpers.get_tempdir`][] returns a temporary directory.

+

+        If all candidates are mocked to fail for the standard temporary

+        directory choices, then we return the `derivepassphrase`

+        configuration directory.

+

+        """

+        runner = click.testing.CliRunner(mix_stderr=False)

+        # TODO(the-13th-letter): Rewrite using parenthesized

+        # with-statements.

+        # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9

+        with contextlib.ExitStack() as stack:

+            monkeypatch = stack.enter_context(pytest.MonkeyPatch.context())

+            stack.enter_context(

+                tests.isolated_vault_config(

+                    monkeypatch=monkeypatch,

+                    runner=runner,

+                    vault_config={'services': {}},

+                )

+            )

+            monkeypatch.delenv('TMPDIR', raising=False)

+            monkeypatch.delenv('TEMP', raising=False)

+            monkeypatch.delenv('TMP', raising=False)

+            config_dir = cli_helpers.config_filename(subsystem=None)

+

+            def is_dir_false(

+                self: pathlib.Path,

+                /,

+                *,

+                follow_symlinks: bool = False,

+            ) -> bool:

+                del self, follow_symlinks

+                return False

+

+            def is_dir_error(

+                self: pathlib.Path,

+                /,

+                *,

+                follow_symlinks: bool = False,

+            ) -> bool:

+                del follow_symlinks

+                raise OSError(

+                    errno.EACCES,

+                    os.strerror(errno.EACCES),

+                    str(self),

+                )

+

+            monkeypatch.setattr(pathlib.Path, 'is_dir', is_dir_false)

+            assert cli_helpers.get_tempdir() == config_dir

+

+            monkeypatch.setattr(pathlib.Path, 'is_dir', is_dir_error)

+            assert cli_helpers.get_tempdir() == config_dir

+

     @Parametrize.DELETE_CONFIG_INPUT

     def test_203_repeated_config_deletion(

         self,