diff --git a/src/nabit/bin/cli.py b/src/nabit/bin/cli.py index 0d1beaf..8d15fcd 100644 --- a/src/nabit/bin/cli.py +++ b/src/nabit/bin/cli.py @@ -5,7 +5,7 @@ from .utils import assert_file_exists, assert_url, cli_validate, CaptureCommand from ..lib.archive import package -from ..lib.sign import KNOWN_TSAS +from ..lib.sign import KNOWN_TSAS, is_encrypted_key from ..lib.backends.base import CollectionTask, CollectionError from ..lib.backends.path import PathCollectionTask @@ -166,9 +166,12 @@ def archive( raise click.BadParameter(f'Sign must be in "cert_chain:key_file" format, got "{value}"') assert_file_exists(key) assert_file_exists(cert_chain) + params = {'key': key, 'cert_chain': cert_chain} + if is_encrypted_key(key): + params['password'] = click.prompt(f'Enter password for {key}', hide_input=True) signatures.append({ 'action': 'sign', - 'params': {'key': key, 'cert_chain': cert_chain}, + 'params': params, }) else: # Convert timestamp list of " | :" strings into a list of timestamp operations diff --git a/src/nabit/lib/sign.py b/src/nabit/lib/sign.py index b0a7252..4a35f1c 100644 --- a/src/nabit/lib/sign.py +++ b/src/nabit/lib/sign.py @@ -29,14 +29,15 @@ }, } -def run_openssl(args: list[str | Path]) -> subprocess.CompletedProcess: +def run_openssl(args: list[str | Path], env: dict = None) -> subprocess.CompletedProcess: """Run openssl subprocess and handle errors.""" command = ["openssl"] + args try: result = subprocess.run( command, capture_output=True, - check=True + check=True, + env=env ) return result except subprocess.CalledProcessError as e: @@ -44,6 +45,11 @@ def run_openssl(args: list[str | Path]) -> subprocess.CompletedProcess: print(f"OpenSSL error: {command_str}\n{e.stderr}", file=sys.stderr) raise +def is_encrypted_key(key_path: Path | str) -> bool: + """Check if a private key is encrypted.""" + contents = Path(key_path).read_text() + return any(s in contents for s in ["ENCRYPTED PRIVATE KEY", "Proc-Type: 4,ENCRYPTED"]) + def timestamp(file_path: str, output_path: str, url: str, cert_chain: str) -> None: """ Create a timestamp for a file. @@ -90,7 +96,7 @@ def verify_timestamp(timestamp_file: Path, file_to_verify: Path, pem_file: Path) "-CAfile", pem_file, ]) -def sign(file_path: Path, output_path: Path, key: str, cert_chain: Path) -> None: +def sign(file_path: Path, output_path: Path, key: str, cert_chain: Path, password: str | None = None) -> None: """Create a detached signature with full chain in PEM format.""" # Validate the certificate chain is in PEM format try: @@ -140,10 +146,15 @@ def sign(file_path: Path, output_path: Path, key: str, cert_chain: Path) -> None # in order to make the signature comply with the requirements for a CAdES Basic Electronic Signature (CAdES-BES)." "-cades", ] + env = None if include_chain: args.extend(["-certfile", cert_chain_file.name]) - - return run_openssl(args) + if password is not None: + env = os.environ.copy() + env['OPENSSL_PASS'] = password + args.extend(["-passin", "env:OPENSSL_PASS"]) + + return run_openssl(args, env=env) def verify_signature(signature_file: Path, file_to_verify: Path) -> None: diff --git a/tests/fixtures/generate_certs.sh b/tests/fixtures/generate_certs.sh index d23c504..b4c5b9b 100644 --- a/tests/fixtures/generate_certs.sh +++ b/tests/fixtures/generate_certs.sh @@ -41,6 +41,9 @@ keyUsage=critical,digitalSignature,keyCertSign,cRLSign") # Generate ECDSA private key for domain signing certificate openssl ecparam -name prime256v1 -genkey -noout -out tests/fixtures/pki/domain-signing.key +# Create encrypted version of domain signing key (PKCS#8 format) +openssl pkcs8 -topk8 -in tests/fixtures/pki/domain-signing.key -out tests/fixtures/pki/domain-signing-enc.key -passout pass:password + # Create CSR for domain signing certificate openssl req -new \ -key tests/fixtures/pki/domain-signing.key \ diff --git a/tests/fixtures/pki/domain-signing-enc.key b/tests/fixtures/pki/domain-signing-enc.key new file mode 100644 index 0000000..6798e8b --- /dev/null +++ b/tests/fixtures/pki/domain-signing-enc.key @@ -0,0 +1,8 @@ +-----BEGIN ENCRYPTED PRIVATE KEY----- +MIH0MF8GCSqGSIb3DQEFDTBSMDEGCSqGSIb3DQEFDDAkBBAV/OaoY5FSUFCHPVzP +VDsrAgIIADAMBggqhkiG9w0CCQUAMB0GCWCGSAFlAwQBKgQQronAn+/5DupHg19B +QRJLwQSBkAeWXFqhhhiDTTMkoK2Ahc5A3yBmkfpxVHHD6TQmBeOdKo6MxI9HQzub +LJk556rDL3tyvT2u/w+pEb90RrMbwqSPwz5ZWLzcdJu7aTUpHBPICoJLEpVcjNAw +8K3mZ7wsfkqbRMm0f/g82LrYWRsqj2wDmPJ4EWWuw31ukdVZH6sE6aTGrFAZ/HyC +4IqyKG0uXg== +-----END ENCRYPTED PRIVATE KEY----- diff --git a/tests/test_cli.py b/tests/test_cli.py index e999cd3..1b4b298 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -5,13 +5,12 @@ import json import re import pytest - from .utils import validate_passing, validate_failing, filter_str ### helpers -def run(runner, args, exit_code=0, output="Package created"): - result = runner.invoke(main, args, catch_exceptions=False) +def run(runner, args, exit_code=0, output="Package created", input=None, catch_exceptions=False): + result = runner.invoke(main, args, catch_exceptions=catch_exceptions, input=input) assert result.exit_code == exit_code, f"Expected exit code {exit_code}, got {result.exit_code} with output: {result.output}" if output: assert output in result.output @@ -137,6 +136,29 @@ def test_signatures(runner, tmp_path, test_files, root_ca): SUCCESS: signature /signatures/tagmanifest-sha256.txt.p7s.p7s verified SUCCESS: Timestamp /signatures/tagmanifest-sha256.txt.p7s.p7s.tsr verified\ """) + +def test_signatures_encrypted_key(runner, tmp_path, test_files, root_ca): + bag_path = tmp_path / 'bag' + run(runner, [ + 'archive', + str(bag_path), + '-p', str(test_files["payload"][0]), + '-s', 'tests/fixtures/pki/domain-chain.pem:tests/fixtures/pki/domain-signing-enc.key', + ], input='password\n') + assert validate_passing(bag_path) == snapshot("""\ +WARNING: No headers.warc found; archive lacks request and response metadata +SUCCESS: bag format is valid +SUCCESS: signature /signatures/tagmanifest-sha256.txt.p7s verified +WARNING: No timestamps found\ +""") + + # with wrong password, should fail + run(runner, [ + 'archive', + str(tmp_path / 'bag2'), + '-p', str(test_files["payload"][0]), + '-s', 'tests/fixtures/pki/domain-chain.pem:tests/fixtures/pki/domain-signing-enc.key', + ], catch_exceptions=True, exit_code=1, input='wrong\n', output='maybe wrong password') def test_just_timestamp_no_signatures(runner, tmp_path, test_files, root_ca): bag_path = tmp_path / 'bag'