add option to create private key through makefile

.github/workflows/tests.yml

@@ -50,9 +50,9 @@ jobs:
         os: [ubuntu-latest]
         python-version: ["3.7", "3.8", "3.9", "3.10", "3.11"]
         allow-failure: [false]
-        test-case: [test-local]
+        test-case: ["create-private-key test-local"]
         # can use below option to set environment variables or makefile settings applied during test execution
-        test-option: ["MAGPIE_NETWORK_CREATE_MISSING_PEM_FILE=True"]
+        test-option: [""]
         include:
           # linter tests
           - os: ubuntu-latest
@@ -93,8 +93,7 @@ jobs:
           - os: ubuntu-20.04
             python-version: "3.6"
             allow-failure: true
-            test-case: test-local
-            test-option: "MAGPIE_NETWORK_CREATE_MISSING_PEM_FILE=True"
+            test-case: create-private-key test-local
           # test allowed failing for recent versions
           # - os: ubuntu-latest
           #   python-version: 3.12

Makefile

@@ -818,3 +818,7 @@ conda-env: conda-base	## create conda environment if missing and required
 			echo "Creating conda environment at '$(CONDA_ENV_PATH)'..." && \
 		 	"$(CONDA_BIN)" create -y -n "$(CONDA_ENV_NAME)" python=$(PYTHON_VERSION)) \
 		)
+
+.PHONY: create-private-key
+create-private-key:  ## create a private key file according to the MAGPIE_NETWORK_PEM_FILES and MAGPIE_NETWORK_PEM_PASSWORDS settings
+	@bash -c '$(CONDA_CMD) magpie_create_private_key --config "$(APP_INI)"'

magpie/cli/create_private_key.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+
+"""
+Create a private key file used to generate a JSON Web Key.
+
+This file is required when network mode is enabled in order to sign JSON Web Tokens.
+"""
+
+import argparse
+import os.path
+import sys
+from typing import TYPE_CHECKING
+
+from magpie.api.management.network.network_utils import pem_files, create_private_key
+from magpie.cli.utils import make_logging_options, setup_logger_from_options
+from magpie.constants import get_constant
+from magpie.utils import get_logger, get_settings_from_config_ini
+
+if TYPE_CHECKING:
+    from typing import Optional, Sequence
+
+    from magpie.typedefs import Str
+
+LOGGER = get_logger(__name__,
+                    message_format="%(asctime)s - %(levelname)s - %(message)s",
+                    datetime_format="%d-%b-%y %H:%M:%S", force_stdout=False)
+
+
+def make_parser():
+    # type: () -> argparse.ArgumentParser
+    parser = argparse.ArgumentParser(description="Create a private key used to generate a JSON Web Key.")
+    parser.add_argument("--config", "--ini", metavar="CONFIG", dest="ini_config",
+                        default=get_constant("MAGPIE_INI_FILE_PATH"),
+                        help="Configuration INI file to retrieve database connection settings (default: %(default)s).")
+    parser.add_argument("--key-file",
+                        help="Location to write key file to. Default is to use the first file listed in the "
+                             "MAGPIE_NETWORK_PEM_FILES variable.")
+    parser.add_argument("--password",
+                        help="Password used to encrypt the key file. Default is to not encrypt the key file unless the "
+                             "the --key-file argument is not set and there is an associated password in the "
+                             "MAGPIE_NETWORK_PEM_PASSWORDS variable.")
+    parser.add_argument("--force", action="store_true", help="Recreate the key file if it already exists.")
+    make_logging_options(parser)
+    return parser
+
+
+def main(args=None, parser=None, namespace=None):
+    # type: (Optional[Sequence[Str]], Optional[argparse.ArgumentParser], Optional[argparse.Namespace]) -> int
+    if not parser:
+        parser = make_parser()
+    args = parser.parse_args(args=args, namespace=namespace)
+    setup_logger_from_options(LOGGER, args)
+    settings_container = get_settings_from_config_ini(args.ini_config)
+
+    if args.key_file:
+        key_file = args.key_file
+    else:
+        pem_files_ = pem_files(settings_container)
+        if pem_files_:
+            key_file = pem_files_[0]
+        else:
+            LOGGER.error(
+                "No network PEM files specified. Either set MAGPIE_NETWORK_PEM_FILES or use the --key-file argument")
+            return 1
+
+    if os.path.isfile(key_file) and not args.force:
+        LOGGER.warning("File %s already exists. To overwrite this file use the --force option.", key_file)
+        return 2
+
+    password = args.password
+    if password is not None:
+        password = password.encode()
+
+    create_private_key(key_file, password=password, settings_container=settings_container)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

setup.py

@@ -255,7 +255,8 @@ def _extra_requirements(base_requirements, other_requirements):
             "magpie_run_db_migration = magpie.cli.run_db_migration:main",
             "magpie_send_email = magpie.cli.send_email:main",
             "magpie_sync_resources = magpie.cli.sync_resources:main",
-            "magpie_purge_expired_network_tokens = magpie.cli.purge_expired_network_tokens:main"
+            "magpie_purge_expired_network_tokens = magpie.cli.purge_expired_network_tokens:main",
+            "magpie_create_private_key = magpie.cli.create_private_key:main"
         ],
     }
 )

tests/test_magpie_cli.py

@@ -37,17 +37,18 @@
 ]
 
 
-def run_and_get_output(command, trim=True):
+def run_and_get_output(command, trim=True, expect_output=True):
     cmd = " ".join(command) if isinstance(command, (list, tuple)) else command
     env = {"PATH": os.path.expandvars(os.environ["PATH"])}  # when debugging, explicit expand of install path required
     pipe = subprocess.PIPE
     with subprocess.Popen(cmd, shell=True, env=env, universal_newlines=True, stdout=pipe) as proc:  # nosec # noqa
         out, err = proc.communicate()
     assert not err, "process returned with error code {}".format(err)
     # when no output is present, it is either because CLI was not installed correctly, or caused by some other error
-    assert out != "", "process did not execute as expected, no output available"
     out_lines = [line for line in out.splitlines() if not trim or (line and not line.startswith(" "))]
-    assert len(out_lines), "could not retrieve any console output"
+    if expect_output:
+        assert out != "", "process did not execute as expected, no output available"
+        assert len(out_lines), "could not retrieve any console output"
     return out_lines
 
 
@@ -288,6 +289,9 @@ def test_purge_expired_network_tokens_help_directly():
     assert "Delete all expired network tokens." in out_lines[1]
 
 
+@runner.MAGPIE_TEST_CLI
+@runner.MAGPIE_TEST_LOCAL
+@runner.MAGPIE_TEST_NETWORK
 def test_purge_expired_network_tokens():
     test_url = "http://localhost"
     test_username = "test_username"
@@ -307,3 +311,83 @@ def mocked_request(*args, **_kwargs):
         session_mock.assert_any_call("POST", "{}/signin".format(test_url), data=None,
                                      json={"user_name": "test_username", "password": "qwertyqwerty"})
         session_mock.assert_any_call("DELETE", "{}/network/tokens?expired_only=true".format(test_url))
+
+@runner.MAGPIE_TEST_CLI
+@runner.MAGPIE_TEST_LOCAL
+@runner.MAGPIE_TEST_NETWORK
+def test_create_private_key_help_via_magpie_helper():
+    out_lines = run_and_get_output("magpie_helper create_private_key --help")
+    assert "usage: magpie_helper create_private_key" in out_lines[0]
+    assert "Create a private key used to generate a JSON Web Key." in out_lines[1]
+
+
+@runner.MAGPIE_TEST_CLI
+@runner.MAGPIE_TEST_LOCAL
+@runner.MAGPIE_TEST_NETWORK
+def test_create_private_key_help_directly():
+    out_lines = run_and_get_output("magpie_create_private_key --help")
+    assert "usage: magpie_create_private_key" in out_lines[0]
+    assert "Create a private key used to generate a JSON Web Key." in out_lines[1]
+
+
+@runner.MAGPIE_TEST_CLI
+@runner.MAGPIE_TEST_LOCAL
+@runner.MAGPIE_TEST_NETWORK
+def test_create_private_key_create_file():
+    tmp_file = tempfile.NamedTemporaryFile(mode="w")
+    tmp_file.close()
+    try:
+        run_and_get_output("magpie_create_private_key --key-file {}".format(tmp_file.name), expect_output=False)
+        with open(tmp_file.name) as f:
+            key_content = f.read()
+            assert key_content.startswith("-----BEGIN RSA PRIVATE KEY-----")
+            assert "ENCRYPTED" not in key_content
+    finally:
+        os.unlink(tmp_file.name)
+
+
+@runner.MAGPIE_TEST_CLI
+@runner.MAGPIE_TEST_LOCAL
+@runner.MAGPIE_TEST_NETWORK
+def test_create_private_key_create_file_no_force_exists():
+    tmp_file = tempfile.NamedTemporaryFile(mode="w", delete=False)
+    tmp_file.close()
+    try:
+        out = run_and_get_output("magpie_create_private_key --key-file {}".format(tmp_file.name), expect_output=False)
+        assert "File {} already exists.".format(tmp_file.name) in out[0]
+    finally:
+        os.unlink(tmp_file.name)
+
+
+@runner.MAGPIE_TEST_CLI
+@runner.MAGPIE_TEST_LOCAL
+@runner.MAGPIE_TEST_NETWORK
+def test_create_private_key_create_file_force_exists():
+    tmp_file = tempfile.NamedTemporaryFile(mode="w")
+    tmp_file.close()
+    try:
+        run_and_get_output("magpie_create_private_key --force --key-file {}".format(tmp_file.name), expect_output=False)
+        with open(tmp_file.name) as f:
+            key_content = f.read()
+            assert key_content.startswith("-----BEGIN RSA PRIVATE KEY-----")
+            assert "ENCRYPTED" not in key_content
+
+    finally:
+        os.unlink(tmp_file.name)
+
+
+@runner.MAGPIE_TEST_CLI
+@runner.MAGPIE_TEST_LOCAL
+@runner.MAGPIE_TEST_NETWORK
+def test_create_private_key_create_file_with_password():
+    tmp_file = tempfile.NamedTemporaryFile(mode="w")
+    tmp_file.close()
+    try:
+        run_and_get_output("magpie_create_private_key --password qwertqwerty --key-file {}".format(tmp_file.name),
+                           expect_output=False)
+        with open(tmp_file.name) as f:
+            key_content = f.read()
+            assert key_content.startswith("-----BEGIN RSA PRIVATE KEY-----")
+            assert "ENCRYPTED" in key_content
+    finally:
+        os.unlink(tmp_file.name)