macOS keyring.yaml support (#8292)

* Added 'service' as a Keychain ctor param. Removed 'testing'

* Detect existing keys in the Mac Keychain

* Fix to allow migration of keys on macOS

* Added dump_keyring.py tool to show decrypted contents of keyring.yaml

* Prompt to save passphrase to macOS keychain

* Master passphrase retrieval/removal from the macOS Keychain.
Fixed typos.

* Warn if errSecInteractionNotAllowed is detected when accessing the macOS Keychain

* Fixed file_keyring synchronization test failures on macOS.
Fixed sporadic test failures on macOS when fsevents are delivered for the keyring after deletion.
TempKeyring-based tests now patch supports_os_passphrase_storage() to return False.

* TempKeyring mocks-out legacy_keyring setup to allow tests to succeed on macOS (which could find existing keys in the Keychain)

* Fixed pylint error

* Use with_name instead of with_stem (which is new to Python 3.9)

* Fixed keychain tests that started prompting for the keyring passphrase.

* Fixed LGTM issues

* Re-added the cleaning up temp keychain statement. This is being removed in a separate PR.

* Linter fixes

* Fixed keyring assignment on macOS when passphrase support is disabled.

* Include 'can_save_passphrase' flag in keyring_status response

* More linter fixes

* Fixed determination of the user_passphrase_is_set flag. This was returning true for a newly created keyring without any keys (or passphrase set)

* Removed the tidy_passphrase function per feedback

* Added some comments based on feedback

* Update chia/cmds/passphrase_funcs.py

Co-authored-by: Adam Kelly <[email protected]>

Co-authored-by: Adam Kelly <[email protected]>

Author: paninaro

chia/cmds/passphrase_funcs.py

@@ -2,8 +2,9 @@
 import sys
 
 from chia.daemon.client import acquire_connection_to_daemon
-from chia.util.keychain import Keychain, obtain_current_passphrase
+from chia.util.keychain import Keychain, obtain_current_passphrase, supports_os_passphrase_storage
 from chia.util.keyring_wrapper import DEFAULT_PASSPHRASE_IF_NO_MASTER_PASSPHRASE
+from chia.util.misc import prompt_yes_no
 from chia.util.ws_message import WsRpcMessage
 from getpass import getpass
 from io import TextIOWrapper
@@ -42,33 +43,56 @@ def verify_passphrase_meets_requirements(
         raise Exception("Unexpected passphrase verification case")
 
 
-def tidy_passphrase(passphrase: str) -> str:
-    """
-    Perform any string processing we want to apply to the entered passphrase.
-    Currently we strip leading/trailing whitespace.
-    """
-    return passphrase.strip()
+def prompt_to_save_passphrase() -> bool:
+    save: bool = False
+
+    try:
+        if supports_os_passphrase_storage():
+            location: Optional[str] = None
 
+            if sys.platform == "darwin":
+                location = "macOS Keychain"
 
-def prompt_for_new_passphrase() -> str:
+            if location is None:
+                raise ValueError("OS-specific credential store not specified")
+
+            print(
+                "\n"
+                "Your passphrase can be stored in your system's secure credential store. "
+                "Other Chia processes will be able to access your keys without prompting for your passphrase."
+            )
+            save = prompt_yes_no(f"Would you like to save your passphrase to the {location} (y/n) ")
+
+    except Exception as e:
+        print(f"Caught exception: {e}")
+        return False
+
+    return save
+
+
+def prompt_for_new_passphrase() -> Tuple[str, bool]:
     min_length: int = Keychain.minimum_passphrase_length()
     if min_length > 0:
         n = min_length
         print(f"\nPassphrases must be {n} or more characters in length")  # lgtm [py/clear-text-logging-sensitive-data]
     while True:
-        passphrase = tidy_passphrase(getpass("New Passphrase: "))
-        confirmation = tidy_passphrase(getpass("Confirm Passphrase: "))
+        passphrase: str = getpass("New Passphrase: ")
+        confirmation: str = getpass("Confirm Passphrase: ")
+        save_passphrase: bool = False
 
         valid_passphrase, error_msg = verify_passphrase_meets_requirements(passphrase, confirmation)
 
         if valid_passphrase:
-            return passphrase
+            if supports_os_passphrase_storage():
+                save_passphrase = prompt_to_save_passphrase()
+
+            return passphrase, save_passphrase
         elif error_msg:
             print(f"{error_msg}\n")  # lgtm [py/clear-text-logging-sensitive-data]
 
 
 def read_passphrase_from_file(passphrase_file: TextIOWrapper) -> str:
-    passphrase = tidy_passphrase(passphrase_file.read())
+    passphrase = passphrase_file.read()
     passphrase_file.close()
     return passphrase
 
@@ -82,14 +106,18 @@ def initialize_passphrase() -> None:
     # We'll rely on Keyring initialization to leverage the cached passphrase for
     # bootstrapping the keyring encryption process
     print("Setting keyring passphrase")
-    passphrase = None
+    passphrase: Optional[str] = None
+    # save_passphrase indicates whether the passphrase should be saved in the
+    # macOS Keychain or Windows Credential Manager
+    save_passphrase: bool = False
+
     if Keychain.has_cached_passphrase():
         passphrase = Keychain.get_cached_master_passphrase()
 
     if not passphrase or passphrase == default_passphrase():
-        passphrase = prompt_for_new_passphrase()
+        passphrase, save_passphrase = prompt_for_new_passphrase()
 
-    Keychain.set_master_passphrase(current_passphrase=None, new_passphrase=passphrase)
+    Keychain.set_master_passphrase(current_passphrase=None, new_passphrase=passphrase, save_passphrase=save_passphrase)
 
 
 def set_or_update_passphrase(passphrase: Optional[str], current_passphrase: Optional[str]) -> bool:
@@ -106,17 +134,21 @@ def set_or_update_passphrase(passphrase: Optional[str], current_passphrase: Opti
                 print(f"Unable to confirm current passphrase: {e}")
                 sys.exit(1)
 
-    success = False
-    new_passphrase = passphrase
+    success: bool = False
+    new_passphrase: Optional[str] = passphrase
+    save_passphrase: bool = False
+
     try:
         # Prompt for the new passphrase, if necessary
-        if not new_passphrase:
-            new_passphrase = prompt_for_new_passphrase()
+        if new_passphrase is None:
+            new_passphrase, save_passphrase = prompt_for_new_passphrase()
 
         if new_passphrase == current_passphrase:
             raise ValueError("passphrase is unchanged")
 
-        Keychain.set_master_passphrase(current_passphrase=current_passphrase, new_passphrase=new_passphrase)
+        Keychain.set_master_passphrase(
+            current_passphrase=current_passphrase, new_passphrase=new_passphrase, save_passphrase=save_passphrase
+        )
         success = True
     except Exception as e:
         print(f"Unable to set or update passphrase: {e}")

chia/daemon/keychain_proxy.py

@@ -55,7 +55,7 @@ def __init__(
         ssl_context: Optional[ssl.SSLContext] = None,
         local_keychain: Optional[Keychain] = None,
         user: str = None,
-        testing: bool = False,
+        service: str = None,
     ):
         self.log = log
         if local_keychain:
@@ -65,7 +65,7 @@ def __init__(
         else:
             self.keychain = None  # type: ignore
         self.keychain_user = user
-        self.keychain_testing = testing
+        self.keychain_service = service
         super().__init__(uri or "", ssl_context)
 
     def use_local_keychain(self) -> bool:
@@ -81,9 +81,9 @@ def format_request(self, command: str, data: Dict[str, Any]) -> WsRpcMessage:
         if data is None:
             data = {}
 
-        if self.keychain_user or self.keychain_testing:
+        if self.keychain_user or self.keychain_service:
             data["kc_user"] = self.keychain_user
-            data["kc_testing"] = self.keychain_testing
+            data["kc_service"] = self.keychain_service
 
         return super().format_request(command, data)
 
@@ -304,14 +304,14 @@ async def connect_to_keychain(
     ssl_context: Optional[ssl.SSLContext],
     log: logging.Logger,
     user: str = None,
-    testing: bool = False,
+    service: str = None,
 ) -> KeychainProxy:
     """
     Connect to the local daemon.
     """
 
     client = KeychainProxy(
-        uri=f"wss://{self_hostname}:{daemon_port}", ssl_context=ssl_context, log=log, user=user, testing=testing
+        uri=f"wss://{self_hostname}:{daemon_port}", ssl_context=ssl_context, log=log, user=user, service=service
     )
     # Connect to the service if the proxy isn't using a local keychain
     if not client.use_local_keychain():
@@ -320,7 +320,11 @@ async def connect_to_keychain(
 
 
 async def connect_to_keychain_and_validate(
-    root_path: Path, log: logging.Logger, *, user: str = None, testing: bool = False
+    root_path: Path,
+    log: logging.Logger,
+    *,
+    user: str = None,
+    service: str = None,
 ) -> Optional[KeychainProxy]:
     """
     Connect to the local daemon and do a ping to ensure that something is really
@@ -334,7 +338,7 @@ async def connect_to_keychain_and_validate(
         ca_key_path = root_path / net_config["private_ssl_ca"]["key"]
         ssl_context = ssl_context_for_client(ca_crt_path, ca_key_path, crt_path, key_path, log=log)
         connection = await connect_to_keychain(
-            net_config["self_hostname"], net_config["daemon_port"], ssl_context, log, user, testing
+            net_config["self_hostname"], net_config["daemon_port"], ssl_context, log, user, service
         )
 
         # If proxying to a local keychain, don't attempt to ping

chia/daemon/keychain_server.py

@@ -36,22 +36,22 @@ def __init__(self):
 
     def get_keychain_for_request(self, request: Dict[str, Any]):
         """
-        Keychain instances can have a user and testing flag associated with them.
+        Keychain instances can have user and service strings associated with them.
         The keychain backends ultimately point to the same data stores, but the user
-        and testing flags are used to partition those data stores. We attempt to
-        maintain a mapping of user/testing pairs to their corresponding Keychain.
+        and service strings are used to partition those data stores. We attempt to
+        maintain a mapping of user/service pairs to their corresponding Keychain.
         """
         keychain = None
         user = request.get("kc_user", self._default_keychain.user)
-        testing = request.get("kc_testing", self._default_keychain.testing)
-        if user == self._default_keychain.user and testing == self._default_keychain.testing:
+        service = request.get("kc_service", self._default_keychain.service)
+        if user == self._default_keychain.user and service == self._default_keychain.service:
             keychain = self._default_keychain
         else:
-            key = (user or "unnamed") + ("test" if testing else "")
+            key = (user or "unnamed") + (service or "")
             if key in self._alt_keychains:
                 keychain = self._alt_keychains[key]
             else:
-                keychain = Keychain(user=user, testing=testing)
+                keychain = Keychain(user=user, service=service)
                 self._alt_keychains[key] = keychain
         return keychain
 

chia/daemon/server.py

@@ -31,6 +31,7 @@
     KeyringRequiresMigration,
     passphrase_requirements,
     supports_keyring_passphrase,
+    supports_os_passphrase_storage,
 )
 from chia.util.path import mkdir
 from chia.util.service_groups import validate_service
@@ -341,14 +342,16 @@ async def is_keyring_locked(self) -> Dict[str, Any]:
 
     async def keyring_status(self) -> Dict[str, Any]:
         passphrase_support_enabled: bool = supports_keyring_passphrase()
-        user_passphrase_is_set: bool = not using_default_passphrase()
+        can_save_passphrase: bool = supports_os_passphrase_storage()
+        user_passphrase_is_set: bool = Keychain.has_master_passphrase() and not using_default_passphrase()
         locked: bool = Keychain.is_keyring_locked()
         needs_migration: bool = Keychain.needs_migration()
         requirements: Dict[str, Any] = passphrase_requirements()
         response: Dict[str, Any] = {
             "success": True,
             "is_keyring_locked": locked,
             "passphrase_support_enabled": passphrase_support_enabled,
+            "can_save_passphrase": can_save_passphrase,
             "user_passphrase_is_set": user_passphrase_is_set,
             "needs_migration": needs_migration,
             "passphrase_requirements": requirements,

chia/util/dump_keyring.py

@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+
+import click
+import colorama
+import threading
+import yaml
+
+from chia.cmds.passphrase_funcs import read_passphrase_from_file
+from chia.util.default_root import DEFAULT_KEYS_ROOT_PATH
+from chia.util.file_keyring import FileKeyring
+from chia.util.keyring_wrapper import DEFAULT_PASSPHRASE_IF_NO_MASTER_PASSPHRASE
+from cryptography.exceptions import InvalidTag
+from getpass import getpass
+from io import TextIOWrapper
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+DEFAULT_KEYRING_YAML = DEFAULT_KEYS_ROOT_PATH / "keyring.yaml"
+
+
+class DumpKeyring(FileKeyring):  # lgtm [py/missing-call-to-init]
+    def __init__(self, keyring_file: Path):
+        self.keyring_path = keyring_file
+        self.payload_cache = {}
+        self.load_keyring_lock = threading.RLock()
+        # We don't call super().__init__() to avoid side-effects
+
+
+def get_passphrase_prompt(keyring_file: str) -> str:
+    prompt = (
+        colorama.Fore.YELLOW
+        + colorama.Style.BRIGHT
+        + "(Unlock Keyring: "
+        + colorama.Fore.MAGENTA
+        + keyring_file
+        + colorama.Style.RESET_ALL
+        + colorama.Fore.YELLOW
+        + colorama.Style.BRIGHT
+        + ")"
+        + colorama.Style.RESET_ALL
+        + " Passphrase: "
+    )  # noqa: E501
+    return prompt
+
+
+@click.command()
+@click.argument("keyring_file", nargs=1, default=DEFAULT_KEYRING_YAML)
+@click.option(
+    "--full-payload", is_flag=True, default=False, help="Print the full keyring contents, including plaintext"
+)
+@click.option("--passphrase-file", type=click.File("r"), help="File or descriptor to read the passphrase from")
+@click.option("--pretty-print", is_flag=True, default=False)
+def dump(keyring_file, full_payload: bool, passphrase_file: Optional[TextIOWrapper], pretty_print: bool):
+    passphrase: str = DEFAULT_PASSPHRASE_IF_NO_MASTER_PASSPHRASE
+    prompt: str = get_passphrase_prompt(str(keyring_file))
+    data: Dict[str, Any] = {}
+
+    print(f"Attempting to dump contents of keyring file: {keyring_file}\n")
+
+    if passphrase_file is not None:
+        passphrase = read_passphrase_from_file(passphrase_file)
+
+    keyring = DumpKeyring(keyring_file)
+
+    if full_payload:
+        keyring.load_outer_payload()
+        data = keyring.outer_payload_cache
+
+    for i in range(5):
+        try:
+            keyring.load_keyring(passphrase)
+            if len(data) > 0:
+                data["data"] = keyring.payload_cache
+            else:
+                data = keyring.payload_cache
+
+            if pretty_print:
+                print(yaml.dump(data))
+            else:
+                print(data)
+            break
+        except (ValueError, InvalidTag):
+            passphrase = getpass(prompt)
+        except Exception as e:
+            print(f"Unhandled exception: {e}")
+            break
+
+
+def main():
+    colorama.init()
+    dump()  # pylint: disable=no-value-for-parameter
+
+
+if __name__ == "__main__":
+    main()