derivepassphrase.git

@@ -5,12 +5,15 @@

     matter the level.

     This is a command-line only change.

-  - In `derivepassphrase vault`, accept falsy values everywhere `vault` does:

-    depending on the setting, they are equivalent to zero, the empty string, or

-    "not set". ([#17])

-    This is a command-line only change.  The API provides a new function to

-    normalize falsy settings, but still otherwise requires settings to be of

-    the correct type.

+  - In `derivepassphrase vault`, when importing settings, accept falsy values

+    everywhere `vault` does, with a warning.  Depending on the setting, they

+    are equivalent to zero, the empty string, or "not set".  ([#17])

+

+    This is a command-line only change, and only affects importing.  The API

+    provides a new function to normalize falsy settings, but still otherwise

+    requires settings to be of the correct type.  Storing a malformed

+    configuration with such falsy values will still generate errors when

+    `derivepassphrase vault` loads the settings from disk.

 [#17]: https://github.com/the-13th-letter/derivepassphrase/issues/17

@@ -6,7 +6,9 @@

 from __future__ import annotations

+import collections

 import enum

+import json

 import math

 from typing import TYPE_CHECKING

@@ -17,7 +19,7 @@ from typing_extensions import (

 )

 if TYPE_CHECKING:

-    from collections.abc import Sequence

+    from collections.abc import MutableSequence, Sequence

     from typing import Literal

     from typing_extensions import (

@@ -136,6 +138,58 @@ class VaultConfig(TypedDict, _VaultConfig, total=False):

     services: Required[dict[str, VaultConfigServicesSettings]]

+def json_path(path: Sequence[str | int], /) -> str:

+    r"""Transform a series of keys and indices into a JSONPath selector.

+

+    The resulting JSONPath selector conforms to RFC 9535, is always

+    rooted at the JSON root node (i.e., starts with `$`), and only

+    contains name and index selectors (in shorthand dot notation, where

+    possible).

+

+    Args:

+        path:

+            A sequence of object keys or array indices to navigate to

+            the desired JSON value, starting from the root node.

+

+    Returns:

+        A valid JSONPath selector (a string) identifying the desired

+        JSON value.

+

+    Examples:

+        >>> json_path(['global', 'phrase'])

+        '$.global.phrase'

+        >>> json_path(['services', 'service name with spaces', 'length'])

+        '$.services["service name with spaces"].length'

+        >>> json_path(['services', 'special\u000acharacters', 'notes'])

+        '$.services["special\\ncharacters"].notes'

+        >>> print(json_path(['services', 'special\u000acharacters', 'notes']))

+        $.services["special\ncharacters"].notes

+        >>> json_path(['custom_array', 2, 0])

+        '$.custom_array[2][0]'

+

+    """

+

+    def needs_longhand(x: str | int) -> bool:

+        initial = (

+            frozenset('abcdefghijklmnopqrstuvwxyz')

+            | frozenset('ABCDEFGHIJKLMNOPQRSTUVWXYZ')

+            | frozenset('_')

+        )

+        chars = initial | frozenset('0123456789')

+        return not (

+            isinstance(x, str)

+            and x

+            and set(x).issubset(chars)

+            and x[:1] in initial

+        )

+

+    chunks = ['$']

+    chunks.extend(

+        f'[{json.dumps(x)}]' if needs_longhand(x) else f'.{x}' for x in path

+    )

+    return ''.join(chunks)

+

+

 def validate_vault_config(  # noqa: C901,PLR0912,PLR0915

     obj: Any,  # noqa: ANN401

     /,

@@ -162,54 +216,34 @@ def validate_vault_config(  # noqa: C901,PLR0912,PLR0915

             disallowed value.

     """

-

-    def maybe_quote(x: str) -> str:

-        chars = (

-            frozenset('abcdefghijklmnopqrstuvwxyz')

-            | frozenset('ABCDEFGHIJKLMNOPQRSTUVWXYZ')

-            | frozenset('0123456789')

-            | frozenset('_')

-        )

-        initial = (

-            frozenset('abcdefghijklmnopqrstuvwxyz')

-            | frozenset('ABCDEFGHIJKLMNOPQRSTUVWXYZ')

-            | frozenset('_')

-        )

-        return (

-            x if x and set(x).issubset(chars) and x[:1] in initial else repr(x)

-        )

-

-    def as_json_path_string(json_path: Sequence[str], /) -> str:

-        return ''.join('.' + maybe_quote(x) for x in json_path)

-

     err_obj_not_a_dict = 'vault config is not a dict'

     err_non_str_service_name = (

         'vault config contains non-string service name {!r}'

     )

-    def err_not_a_dict(json_path: Sequence[str], /) -> str:

-        json_path_str = as_json_path_string(json_path)

+    def err_not_a_dict(path: Sequence[str], /) -> str:

+        json_path_str = json_path(path)

         return f'vault config entry {json_path_str} is not a dict'

-    def err_not_a_string(json_path: Sequence[str], /) -> str:

-        json_path_str = as_json_path_string(json_path)

+    def err_not_a_string(path: Sequence[str], /) -> str:

+        json_path_str = json_path(path)

         return f'vault config entry {json_path_str} is not a string'

-    def err_not_an_int(json_path: Sequence[str], /) -> str:

-        json_path_str = as_json_path_string(json_path)

+    def err_not_an_int(path: Sequence[str], /) -> str:

+        json_path_str = json_path(path)

         return f'vault config entry {json_path_str} is not an integer'

     def err_derivepassphrase_extension(

-        key: str, json_path: Sequence[str], /

+        key: str, path: Sequence[str], /

     ) -> str:

-        json_path_str = as_json_path_string(json_path)

+        json_path_str = json_path(path)

         return (

             f'vault config entry {json_path_str} uses '

             f'`derivepassphrase` extension {key!r}'

         )

-    def err_unknown_setting(key: str, json_path: Sequence[str], /) -> str:

-        json_path_str = as_json_path_string(json_path)

+    def err_unknown_setting(key: str, path: Sequence[str], /) -> str:

+        json_path_str = json_path(path)

         return (

             f'vault config entry {json_path_str} uses '

             f'unknown setting {key!r}'

@@ -217,12 +251,12 @@ def validate_vault_config(  # noqa: C901,PLR0912,PLR0915

     def err_bad_number(

         key: str,

-        json_path: Sequence[str],

+        path: Sequence[str],

         /,

         *,

         strictly_positive: bool = False,

     ) -> str:

-        json_path_str = as_json_path_string((*json_path, key))

+        json_path_str = json_path((*path, key))

         return f'vault config entry {json_path_str} is ' + (

             'not positive' if strictly_positive else 'negative'

         )

@@ -344,7 +378,36 @@ def js_truthiness(value: Any, /) -> bool:  # noqa: ANN401

     return not (isinstance(value, float) and math.isnan(value))

-def clean_up_falsy_vault_config_values(obj: Any) -> None:  # noqa: ANN401,C901,PLR0912

+class CleanupStep(NamedTuple):

+    """A single executed step during vault config cleanup.

+

+    Attributes:

+        path:

+            A sequence of object keys or array indices to navigate to

+            the JSON value that was cleaned up.

+        old_value:

+            The old value.

+        action:

+            Either `'replace'` if `old_value` was replaced with

+            `new_value`, or `'remove'` if `old_value` was removed.

+        new_value:

+            The new value.

+

+    """

+

+    path: Sequence[str | int]

+    """"""

+    old_value: Any

+    """"""

+    action: Literal['replace', 'remove']

+    """"""

+    new_value: Any

+    """"""

+

+

+def clean_up_falsy_vault_config_values(  # noqa: C901,PLR0912

+    obj: Any,  # noqa: ANN401

+) -> Sequence[CleanupStep] | None:

     """Convert falsy values in a vault config to correct types, in-place.

     Needed for compatibility with vault(1), which sometimes uses only

@@ -358,6 +421,19 @@ def clean_up_falsy_vault_config_values(obj: Any) -> None:  # noqa: ANN401,C901,P

             A presumed valid vault configuration save for using falsy

             values of the wrong type.

+    Returns:

+        A list of 4-tuples `(key_tup, old_value, action, new_value)`,

+        indicating the cleanup actions performed.  `key_tup` is

+        a sequence of object keys and/or array indices indicating the

+        JSON path to the leaf value that was cleaned up, `old_value` is

+        the old value, `new_value` is the new value, and `action` is

+        either `replace` (`old_value` was replaced with `new_value`) or

+        `remove` (`old_value` was removed, and `new_value` is

+        meaningless).

+

+        If cleanup was never attempted because of an obviously invalid

+        vault configuration, then `None` is returned, directly.

+

     """

     if (  # pragma: no cover

         not isinstance(obj, dict)

@@ -365,27 +441,43 @@ def clean_up_falsy_vault_config_values(obj: Any) -> None:  # noqa: ANN401,C901,P

         or not isinstance(obj['services'], dict)

     ):

         # config is invalid

-        return

-    service_objects = list(obj['services'].values())

-    if not all(  # pragma: no cover

-        isinstance(service_obj, dict) for service_obj in service_objects

-    ):

-        # config is invalid

-        return

+        return None

+    service_objects: MutableSequence[

+        tuple[Sequence[str | int], dict[str, Any]]

+    ] = collections.deque()

     if 'global' in obj:

         if isinstance(obj['global'], dict):

-            service_objects.append(obj['global'])

+            service_objects.append((['global'], obj['global']))

         else:  # pragma: no cover

             # config is invalid

-            return

-    for service_obj in service_objects:

+            return None

+    service_objects.extend(

+        (['services', sv], val) for sv, val in obj['services'].items()

+    )

+    if not all(  # pragma: no cover

+        isinstance(service_obj, dict) for _, service_obj in service_objects

+    ):

+        # config is invalid

+        return None

+    cleanup_completed: MutableSequence[CleanupStep] = collections.deque()

+    for path, service_obj in service_objects:

         for key, value in list(service_obj.items()):

             # Use match/case here once Python 3.9 becomes unsupported.

             if key == 'phrase':

-                if not js_truthiness(value):

+                if not js_truthiness(value) and value != '':  # noqa: PLC1901

+                    cleanup_completed.append(

+                        CleanupStep(

+                            (*path, key), service_obj[key], 'replace', ''

+                        )

+                    )

                     service_obj[key] = ''

             elif key in {'notes', 'key', 'length', 'repeat'}:

                 if not js_truthiness(value):

+                    cleanup_completed.append(

+                        CleanupStep(

+                            (*path, key), service_obj[key], 'remove', None

+                        )

+                    )

                     service_obj.pop(key)

             elif key in {  # noqa: SIM102

                 'lower',

@@ -396,7 +488,13 @@ def clean_up_falsy_vault_config_values(obj: Any) -> None:  # noqa: ANN401,C901,P

                 'symbol',

             }:

                 if not js_truthiness(value) and value != 0:

+                    cleanup_completed.append(

+                        CleanupStep(

+                            (*path, key), service_obj[key], 'replace', 0

+                        )

+                    )

                     service_obj.pop(key)

+    return cleanup_completed

 class KeyCommentPair(NamedTuple):

@@ -10,6 +10,7 @@ import base64

 import collections

 import contextlib

 import copy

+import enum

 import importlib

 import inspect

 import json

@@ -716,29 +717,29 @@ def _prompt_for_passphrase() -> str:

     )

+class _ORIGIN(enum.Enum):

+    INTERACTIVE: str = 'interactive'

+

+

 def _check_for_misleading_passphrase(

-    key: tuple[str, ...],

+    key: tuple[str, ...] | _ORIGIN,

     value: dict[str, Any],

     *,

     form: Literal['NFC', 'NFD', 'NFKC', 'NFKD'] = 'NFC',

 ) -> None:

-    def is_json_identifier(x: str) -> bool:

-        return not x.startswith(tuple('0123456789')) and not any(

-            c.lower() not in set('0123456789abcdefghijklmnopqrstuvwxyz_')

-            for c in x

-        )

-

     if 'phrase' in value:

         phrase = value['phrase']

         if not unicodedata.is_normalized(form, phrase):

-            key_path = '.'.join(

-                x if is_json_identifier(x) else repr(x) for x in key

+            formatted_key = (

+                key.value

+                if isinstance(key, _ORIGIN)

+                else _types.json_path(key)

             )

             click.echo(

                 (

-                    f'{PROG_NAME}: Warning: the {key_path} passphrase '

-                    f'is not {form}-normalized. Make sure to double-check '

-                    f'this is really the passphrase you want.'

+                    f'{PROG_NAME}: Warning: the {formatted_key} '

+                    f'passphrase is not {form}-normalized. Make sure to '

+                    f'double-check this is really the passphrase you want.'

                 ),

                 err=True,

             )

@@ -1410,8 +1411,28 @@ def derivepassphrase_vault(  # noqa: C901,PLR0912,PLR0913,PLR0914,PLR0915

             err(f'Cannot load config: cannot decode JSON: {e}')

         except OSError as e:

             err(f'Cannot load config: {e.strerror}: {e.filename!r}')

-        _types.clean_up_falsy_vault_config_values(maybe_config)

-        if _types.is_vault_config(maybe_config):

+        cleaned = _types.clean_up_falsy_vault_config_values(maybe_config)

+        if not _types.is_vault_config(maybe_config):

+            err(f'Cannot load config: {_INVALID_VAULT_CONFIG}')

+        assert cleaned is not None

+        for step in cleaned:

+            # These are never fatal errors, because the semantics of

+            # vault upon encountering these settings are ill-specified,

+            # but not ill-defined.

+            if step.action == 'replace':

+                err_msg = (

+                    f'{PROG_NAME}: Warning: Replacing invalid value '

+                    f'{json.dumps(step.old_value)} for key '

+                    f'{_types.json_path(step.path)} with '

+                    f'{json.dumps(step.new_value)}.'

+                )

+            else:

+                err_msg = (

+                    f'{PROG_NAME}: Warning: Removing ineffective setting '

+                    f'{_types.json_path(step.path)} = '

+                    f'{json.dumps(step.old_value)}.'

+                )

+            click.echo(err_msg, err=True)

         form = cast(

             Literal['NFC', 'NFD', 'NFKC', 'NFKD'],

             maybe_config.get('global', {}).get(

@@ -1431,8 +1452,6 @@ def derivepassphrase_vault(  # noqa: C901,PLR0912,PLR0913,PLR0914,PLR0915

                 form=form,

             )

         put_config(maybe_config)

-        else:

-            err(f'Cannot load config: {_INVALID_VAULT_CONFIG}')

     elif export_settings:

         configuration = get_config()

         try:

@@ -1570,7 +1589,7 @@ def derivepassphrase_vault(  # noqa: C901,PLR0912,PLR0913,PLR0914,PLR0915

                 )

                 assert form in {'NFC', 'NFD', 'NFKC', 'NFKD'}

                 _check_for_misleading_passphrase(

-                    ('interactive',), {'phrase': phrase}, form=form

+                    _ORIGIN.INTERACTIVE, {'phrase': phrase}, form=form

                 )

             # If either --key or --phrase are given, use that setting.

@@ -642,8 +642,19 @@ class TestCLI:

             ) as infile:

                 config3 = json.load(infile)

         result = tests.ReadableResult.parse(_result)

-        assert result.clean_exit(empty_stderr=True), 'expected clean exit'

+

+        def expected_warning_line(line: str) -> bool:

+            return (

+                ' Warning: Replacing invalid value ' in line

+                or ' Warning: Removing ineffective setting ' in line

+            )

+

+        assert result.clean_exit(empty_stderr=False), 'expected clean exit'

         assert config3 == config2, 'config not imported correctly'

+        assert not result.stderr or all(

+            expected_warning_line(line)

+            for line in result.stderr.splitlines(True)

+        ), 'unexpected error output'

     def test_213b_import_bad_config_not_vault_config(

         self,

@@ -1277,7 +1288,7 @@ contents go here

                     'global': {'phrase': 'Du\u0308sseldorf'},

                     'services': {},

                 }),

-                'the global passphrase is not NFC-normalized',

+                'the $.global passphrase is not NFC-normalized',

                 id='global-NFC',

             ),

             pytest.param(

@@ -1289,7 +1300,7 @@ contents go here

                     }

                 }),

                 (

-                    "the services.'weird entry name' passphrase "

+                    'the $.services["weird entry name"] passphrase '

                     'is not NFC-normalized'

                 ),

                 id='service-weird-name-NFC',

@@ -1298,7 +1309,7 @@ contents go here

                 ['--config', '-p', DUMMY_SERVICE],

                 'Du\u0308sseldorf',

                 (

-                    f'the services.{DUMMY_SERVICE} passphrase '

+                    f'the $.services.{DUMMY_SERVICE} passphrase '

                     f'is not NFC-normalized'

                 ),

                 id='config-NFC',

@@ -1318,7 +1329,7 @@ contents go here

                     },

                     'services': {},

                 }),

-                'the global passphrase is not NFD-normalized',

+                'the $.global passphrase is not NFD-normalized',

                 id='global-NFD',

             ),

             pytest.param(

@@ -1333,7 +1344,7 @@ contents go here

                     },

                 }),

                 (

-                    "the services.'weird entry name' passphrase "

+                    'the $.services["weird entry name"] passphrase '

                     'is not NFD-normalized'

                 ),

                 id='service-weird-name-NFD',

@@ -99,14 +99,17 @@ def test_200_is_vault_config(test_config: tests.VaultTestConfig) -> None:

 def test_200a_is_vault_config_smudged(

     test_config: tests.VaultTestConfig,

 ) -> None:

-    obj, comment, _ = test_config

-    obj = copy.deepcopy(obj)

-    _types.clean_up_falsy_vault_config_values(obj)

+    _obj, comment, _ = test_config

+    obj = copy.deepcopy(_obj)

+    did_cleanup = _types.clean_up_falsy_vault_config_values(obj)

     assert _types.is_vault_config(obj) == (not comment), (

         'failed to complain about: ' + comment

         if comment

         else 'failed on valid example'

     )

+    assert did_cleanup is None or bool(did_cleanup) == (

+        obj != _obj

+    ), 'mismatched report on cleanup work'

 @pytest.mark.parametrize(

@@ -148,12 +151,12 @@ def test_400_validate_vault_config(test_config: tests.VaultTestConfig) -> None:

 def test_400a_validate_vault_config_smudged(

     test_config: tests.VaultTestConfig,

 ) -> None:

-    obj, comment, validation_settings = test_config

+    _obj, comment, validation_settings = test_config

     allow_unknown_settings, allow_derivepassphrase_extensions = (

         validation_settings or (True, True)

     )

-    obj = copy.deepcopy(obj)

-    _types.clean_up_falsy_vault_config_values(obj)

+    obj = copy.deepcopy(_obj)

+    did_cleanup = _types.clean_up_falsy_vault_config_values(obj)

     if comment:

         with pytest.raises((TypeError, ValueError)):

             _types.validate_vault_config(

@@ -170,3 +173,6 @@ def test_400a_validate_vault_config_smudged(

             )

         except (TypeError, ValueError) as exc:  # pragma: no cover

             assert not exc, 'failed to validate valid example'  # noqa: PT017

+    assert did_cleanup is None or bool(did_cleanup) == (

+        obj != _obj

+    ), 'mismatched report on cleanup work'