Split auth into multiple subparsers

Author: njgheorghita

ethpm_cli/_utils/input.py

@@ -0,0 +1,13 @@
+from ethpm_cli._utils.logger import cli_logger
+
+
+def parse_bool_flag(question: str) -> bool:
+    while True:
+        response = input(f"{question} (y/n) ")
+        if response.lower() == "y":
+            return True
+        elif response.lower() == "n":
+            return False
+        else:
+            cli_logger.info(f"Invalid response: {response}.")
+            continue

ethpm_cli/commands/auth.py

@@ -1,23 +1,104 @@
+import json
 from pathlib import Path
-import tempfile
 from typing import Any, Dict
 
 import eth_keyfile
 from eth_typing import Address
-from eth_utils import to_bytes
+from eth_utils import add_0x_prefix, to_bytes
 
+from ethpm_cli._utils.filesystem import atomic_replace
+from ethpm_cli._utils.input import parse_bool_flag
+from ethpm_cli._utils.logger import cli_logger
 from ethpm_cli._utils.xdg import get_xdg_ethpmcli_root
 from ethpm_cli.constants import KEYFILE_PATH
 from ethpm_cli.exceptions import AuthorizationError
 
+PRIVATE_KEY_WARNING = (
+    "Please be careful when using your private key, it is a sensitive piece of information.\n",
+    "~ ~ ~ Not your keys, not your crypto. ~ ~ ~\n",
+    "ethPM doesn't save your private key, but it is best practice to re-use an already encrypted ",
+    "keyfile and link it with `ethpm auth link` rather than regenerating a new encrypted keyfile.",
+)
+
+
+def link_keyfile(keyfile_path: Path) -> None:
+    if valid_keyfile_exists():
+        xdg_keyfile = get_keyfile_path()
+        cli_logger.info(
+            f"Keyfile detected at {xdg_keyfile}. Please use `ethpm auth unlink` to delete this "
+            "keyfile before linking a new one."
+        )
+    else:
+        import_keyfile(keyfile_path)
+        address = get_authorized_address()
+        cli_logger.info(
+            f"Keyfile stored for address: {address}\n"
+            "It's now available for use when its password is passed in with the "
+            "`--keyfile-password` flag."
+        )
+
+
+def unlink_keyfile() -> None:
+    if not valid_keyfile_exists():
+        cli_logger.info("Unable to unlink keyfile: empty keyfile found.")
+    else:
+        keyfile_path = get_keyfile_path()
+        address = get_authorized_address()
+        keyfile_path.write_text("")
+        cli_logger.info(f"Keyfile removed for address: {address}")
+
+
+def init_keyfile() -> None:
+    if valid_keyfile_exists():
+        cli_logger.info(
+            f"Keyfile detected. Please use `ethpm auth unlink` to delete this "
+            "keyfile before initializing a new one."
+        )
+        return
+
+    cli_logger.info(PRIVATE_KEY_WARNING)
+    agreement = parse_bool_flag(
+        "Are you sure you want to proceed with initializing a keyfile? "
+    )
+    if not agreement:
+        cli_logger.info("Aborting keyfile initialization.")
+        return
+
+    private_key = to_bytes(text=input("Please enter your 32-length private key: "))
+    validate_private_key_length(private_key)
+    password = to_bytes(
+        text=input(
+            "Please enter a password to encrypt your keyfile with (Don't forget this password!): "
+        )
+    )
+    keyfile_json = eth_keyfile.create_keyfile_json(private_key, password)
+    ethpm_xdg_root = get_xdg_ethpmcli_root()
+    ethpm_cli_keyfile_path = ethpm_xdg_root / KEYFILE_PATH
+    with atomic_replace(ethpm_cli_keyfile_path) as file:
+        file.write(json.dumps(keyfile_json))
+    address = get_authorized_address()
+    cli_logger.info(f"Encrypted keyfile saved for address: {address}")
+
+
+def valid_keyfile_exists() -> bool:
+    try:
+        get_authorized_address()
+    except AuthorizationError:
+        return False
+    return True
+
+
+def validate_private_key_length(private_key: str) -> None:
+    if len(private_key) != 32:
+        raise AuthorizationError(f"{private_key} is not 32 long")
+
 
 def import_keyfile(keyfile_path: Path) -> None:
     validate_keyfile(keyfile_path)
     ethpm_xdg_root = get_xdg_ethpmcli_root()
     ethpm_cli_keyfile_path = ethpm_xdg_root / KEYFILE_PATH
-    tmp_keyfile = Path(tempfile.NamedTemporaryFile().name)
-    tmp_keyfile.write_text(keyfile_path.read_text())
-    tmp_keyfile.replace(ethpm_cli_keyfile_path)
+    with atomic_replace(ethpm_cli_keyfile_path) as file:
+        file.write(keyfile_path.read_text())
 
 
 def get_keyfile_path() -> Path:
@@ -49,7 +130,7 @@ def get_authorized_address() -> Address:
     Returns the address associated with stored keyfile. No password required.
     """
     keyfile = get_keyfile_data()
-    return keyfile["address"]
+    return add_0x_prefix(keyfile["address"])
 
 
 def get_authorized_private_key(password: str) -> str:

ethpm_cli/commands/manifest.py

@@ -11,6 +11,7 @@
 from ethpm.validation.package import validate_package_name
 from web3 import Web3
 
+from ethpm_cli._utils.input import parse_bool_flag
 from ethpm_cli._utils.logger import cli_logger
 from ethpm_cli._utils.shellart import bold_blue
 from ethpm_cli._utils.solc import (
@@ -450,18 +451,6 @@ def gen_links() -> Optional[Callable[..., Manifest]]:
     return None
 
 
-def parse_bool_flag(question: str) -> bool:
-    while True:
-        response = input(f"{question} (y/n) ")
-        if response.lower() == "y":
-            return True
-        elif response.lower() == "n":
-            return False
-        else:
-            cli_logger.info(f"Invalid response: {response}.")
-            continue
-
-
 def write_manifest_to_disk(manifest: Manifest, project_dir: Path) -> None:
     while True:
         filename = input("Please enter a filename for your manifest. ")

ethpm_cli/commands/package.py

@@ -93,7 +93,7 @@ def resolve_install_uri(args: Namespace) -> ResolvedInstallURI:
         )
         registry_address = None
     elif registry_backend.can_translate_uri(args.uri):
-        registry_address, _, _, _, _ = parse_registry_uri(args.uri)
+        registry_address, _, _, _, _, _ = parse_registry_uri(args.uri)
         manifest_uri = registry_backend.fetch_uri_contents(args.uri)
     else:
         manifest_uri = args.uri

ethpm_cli/config.py

@@ -16,7 +16,7 @@
 from ethpm_cli._utils.ipfs import get_ipfs_backend
 from ethpm_cli._utils.logger import cli_logger
 from ethpm_cli._utils.xdg import get_xdg_ethpmcli_root
-from ethpm_cli.commands.auth import get_authorized_private_key, import_keyfile
+from ethpm_cli.commands.auth import get_authorized_private_key
 from ethpm_cli.constants import (
     ETHPM_DIR_ENV_VAR,
     ETHPM_PACKAGES_DIR,
@@ -63,16 +63,14 @@ def __init__(self, args: Namespace) -> None:
         else:
             chain_id = 1
 
-        if "keyfile_path" in args and args.keyfile_path:
-            import_keyfile(args.keyfile_path)
-
+        # setup auth
         if "keyfile_password" in args and args.keyfile_password:
             self.private_key = get_authorized_private_key(args.keyfile_password)
         self.w3 = setup_w3(chain_id, self.private_key)
 
         # Setup xdg ethpm dir
         self.xdg_ethpmcli_root = get_xdg_ethpmcli_root()
-        setup_xdg_ethpm_dir(self.xdg_ethpmcli_root, self.w3)
+        validate_xdg_ethpm_dir(self.xdg_ethpmcli_root, self.w3)
 
         # Setup projects dir
         if "project_dir" in args and args.project_dir:
@@ -114,7 +112,7 @@ def setup_w3(chain_id: int, private_key: str = None) -> Web3:
     return w3
 
 
-def setup_xdg_ethpm_dir(xdg_ethpmcli_root: Path, w3: Web3) -> None:
+def validate_xdg_ethpm_dir(xdg_ethpmcli_root: Path, w3: Web3) -> None:
     if not xdg_ethpmcli_root.is_dir():
         initialize_xdg_ethpm_dir(xdg_ethpmcli_root, w3)