From 7bb76c4b10cc6c6af7cb7f366957a97110fbdbee Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 12:26:37 +0000 Subject: [PATCH 01/21] Create genkey private key files with owner-only permissions Private keys were opened with fopen/BIO_new_file, so under umask 022 they were world-readable. Add wolfCLU_FileOpenOwner/wolfCLU_BioOpenOwner, which create new files with mode 0600 on POSIX, and use them for all genkey private outputs and the XMSS state file. WOLFCLU_POSIX_FILE enables the POSIX path on the platforms where wolfSSL uses it for wc_fopen_owner_only(), and only when wolfSSL defines XFDOPEN and XCLOSE. open() is called directly: wolfSSL's open wrappers are WOLFSSL_LOCAL and not exported from the library. F-8096 --- src/genkey/clu_genkey.c | 14 ++--- src/tools/clu_funcs.c | 55 ++++++++++++++++ tests/genkey_sign_ver/genkey-sign-ver-test.py | 63 +++++++++++++++++++ wolfclu/clu_header_main.h | 24 +++++++ 4 files changed, 149 insertions(+), 7 deletions(-) diff --git a/src/genkey/clu_genkey.c b/src/genkey/clu_genkey.c index b9c512e8..781b14da 100644 --- a/src/genkey/clu_genkey.c +++ b/src/genkey/clu_genkey.c @@ -121,7 +121,7 @@ int wolfCLU_genKey_ED25519(WC_RNG* rng, char* fOutNm, int directive, int format) /* open the file for writing the private key */ if (ret == 0) { - file = XFOPEN(finalOutFNm, "wb"); + file = wolfCLU_FileOpenOwner(finalOutFNm); if (!file) { ret = OUTPUT_FILE_ERROR; } @@ -663,7 +663,7 @@ int wolfCLU_GenAndOutput_ECC(WC_RNG* rng, char* fName, int directive, fOutNameBuf[fNameSz + fExtSz] = '\0'; WOLFCLU_LOG(WOLFCLU_L0, "Private key file = %s", fOutNameBuf); - bioPri = wolfSSL_BIO_new_file(fOutNameBuf, "wb"); + bioPri = wolfCLU_BioOpenOwner(fOutNameBuf); if (bioPri == NULL) { wolfCLU_LogError("unable to read outfile %s", fOutNameBuf); @@ -849,7 +849,7 @@ int wolfCLU_genKey_RSA(WC_RNG* rng, char* fName, int directive, int fmt, int /* open the file for writing the private key */ if (ret == WOLFCLU_SUCCESS) { - file = XFOPEN(fOutNameBuf, "wb"); + file = wolfCLU_FileOpenOwner(fOutNameBuf); if (!file) { ret = OUTPUT_FILE_ERROR; } @@ -1180,7 +1180,7 @@ int wolfCLU_genKey_Dilithium(WC_RNG* rng, char* fName, int directive, int fmt, /* open file and write Private key */ if (ret == WOLFCLU_SUCCESS) { - file = XFOPEN(fOutNameBuf, "wb"); + file = wolfCLU_FileOpenOwner(fOutNameBuf); if (file == XBADFILE) { wolfCLU_LogError("unable to open file %s", fOutNameBuf); @@ -1428,7 +1428,7 @@ int wolfCLU_genKey_ML_DSA(WC_RNG* rng, char* fName, int directive, int fmt, /* open file and write Private key */ if (ret == WOLFCLU_SUCCESS) { - file = XFOPEN(fOutNameBuf, "wb"); + file = wolfCLU_FileOpenOwner(fOutNameBuf); if (file == XBADFILE) { wolfCLU_LogError("unable to open file %s", fOutNameBuf); @@ -1582,9 +1582,9 @@ enum wc_XmssRc wolfCLU_XmssKey_WriteCb(const byte * priv, file = fopen(filename, "rb+"); if (!file) { /* Create the file if it didn't exist. */ - file = fopen(filename, "wb+"); + file = wolfCLU_FileOpenOwner(filename); if (!file) { - fprintf(stderr, "error: fopen(%s, \"w+\") failed.\n", filename); + fprintf(stderr, "error: unable to create %s\n", filename); return WC_XMSS_RC_WRITE_FAIL; } } diff --git a/src/tools/clu_funcs.c b/src/tools/clu_funcs.c index 71c74c8a..ae987c73 100644 --- a/src/tools/clu_funcs.c +++ b/src/tools/clu_funcs.c @@ -34,6 +34,11 @@ #include #include +#ifdef WOLFCLU_POSIX_FILE + #include + #include +#endif + #define SALT_SIZE 8 #define DES3_BLOCK_SIZE 24 @@ -1112,6 +1117,56 @@ void wolfCLU_ForceZero(void* mem, unsigned int len) #endif } +#ifndef NO_FILESYSTEM +/* Open a file for binary write. On POSIX a new file is created with owner + * only access (0600). An existing file is truncated and keeps its mode. */ +XFILE wolfCLU_FileOpenOwner(const char* fileName) +{ + XFILE file; +#ifdef WOLFCLU_POSIX_FILE + int fd; +#endif + + if (fileName == NULL) { + return XBADFILE; + } + +#ifdef WOLFCLU_POSIX_FILE + /* wolfSSL's open() wrappers are internal, so open() is called here */ + fd = open(fileName, O_WRONLY | O_CREAT | O_TRUNC, S_IRUSR | S_IWUSR); + if (fd < 0) { + return XBADFILE; + } + file = XFDOPEN(fd, "wb"); + if (file == XBADFILE) { + XCLOSE(fd); + } +#else + file = XFOPEN(fileName, "wb"); +#endif + + return file; +} + +/* BIO version of wolfCLU_FileOpenOwner. The BIO closes the file when freed. */ +WOLFSSL_BIO* wolfCLU_BioOpenOwner(const char* fileName) +{ + XFILE file; + WOLFSSL_BIO* bio; + + file = wolfCLU_FileOpenOwner(fileName); + if (file == XBADFILE) { + return NULL; + } + + bio = wolfSSL_BIO_new_fp(file, BIO_CLOSE); + if (bio == NULL) { + XFCLOSE(file); + } + return bio; +} +#endif /* !NO_FILESYSTEM */ + #ifndef WOLFCLU_NO_TERM_SUPPORT int wolfCLU_GetPassword(char* password, int* passwordSz, char* arg) diff --git a/tests/genkey_sign_ver/genkey-sign-ver-test.py b/tests/genkey_sign_ver/genkey-sign-ver-test.py index 14a777df..9250c32b 100644 --- a/tests/genkey_sign_ver/genkey-sign-ver-test.py +++ b/tests/genkey_sign_ver/genkey-sign-ver-test.py @@ -532,6 +532,69 @@ def test_xmssmt_missing_height_arg(self): "crash)") +@unittest.skipIf(os.name == "nt", "POSIX file permissions only") +class GenkeyPermissionsTest(_GenkeySignVerifyBase): + """Private key files must be owner-only under umask 022 (F-8096).""" + + def setUp(self): + old_umask = os.umask(0o022) + self.addCleanup(os.umask, old_umask) + + def _check_modes(self, algo, keybase, fmt, extra_args=None): + # Start from new files so the create mode applies. + _cleanup_files([keybase + ".priv", keybase + ".pub"]) + priv, pub = self._genkey(algo, keybase, fmt, extra_args, + use_output_flag=True) + priv_mode = os.stat(priv).st_mode & 0o777 + pub_mode = os.stat(pub).st_mode & 0o777 + self.assertEqual(priv_mode, 0o600, + "{} {} private key mode is {:o}, expected 600".format( + algo, fmt, priv_mode)) + self.assertEqual(pub_mode, 0o644, + "{} {} public key mode is {:o}, expected 644".format( + algo, fmt, pub_mode)) + + def test_rsa_key_modes(self): + for fmt in ["der", "pem"]: + with self.subTest(fmt=fmt): + self._check_modes("rsa", "perm-rsa-" + fmt, fmt) + + def test_ecc_key_modes(self): + for fmt in ["der", "pem"]: + with self.subTest(fmt=fmt): + self._check_modes("ecc", "perm-ecc-" + fmt, fmt) + + def test_ed25519_key_modes(self): + for fmt in ["raw", "der", "pem"]: + with self.subTest(fmt=fmt): + self._check_modes("ed25519", "perm-ed25519-" + fmt, fmt) + + def test_dilithium_key_modes(self): + if not _has_algorithm("dilithium"): + self.skipTest("dilithium not available") + for algo in ["dilithium", "ml-dsa"]: + with self.subTest(algo=algo): + self._check_modes(algo, "perm-" + algo, "der", + ["-level", "2"]) + + def test_xmss_key_modes(self): + if not _has_algorithm("xmss"): + self.skipTest("xmss not available") + self._check_modes("xmss", "perm-xmss", "raw", ["-height", "10"]) + + def test_priv_only_key_mode(self): + keybase = "perm-ecc-privonly" + priv = keybase + ".priv" + self._track(priv, keybase + ".pub") + _cleanup_files([priv]) + r = run_wolfssl("-genkey", "ecc", "-out", keybase, "-outform", "der", + "-output", "priv") + self.assertEqual(r.returncode, 0, r.stderr) + mode = os.stat(priv).st_mode & 0o777 + self.assertEqual(mode, 0o600, + "private key mode is {:o}, expected 600".format(mode)) + + class SignVerifySetupArgsTest(unittest.TestCase): """Argument-parsing branches in clu_sign_verify_setup.c. diff --git a/wolfclu/clu_header_main.h b/wolfclu/clu_header_main.h index b3fa9c93..b1bf3a8e 100644 --- a/wolfclu/clu_header_main.h +++ b/wolfclu/clu_header_main.h @@ -150,6 +150,15 @@ extern "C" { #define WOLFCLU_NO_TERM_SUPPORT #endif +/* POSIX file APIs, on the platforms where wolfSSL uses them for its own + * wc_fopen_owner_only(). XFDOPEN and XCLOSE come from wolfSSL. */ +#if !defined(NO_FILESYSTEM) && defined(XFDOPEN) && defined(XCLOSE) && \ + (defined(__unix__) || defined(__APPLE__)) && \ + !defined(WOLFSSL_KERNEL_MODE) && !defined(WOLFSSL_ZEPHYR) && \ + !defined(WOLFSSL_SGX) + #define WOLFCLU_POSIX_FILE +#endif + /* @VERSION * Update every time library change, * functionality shift, @@ -614,6 +623,21 @@ int wolfCLU_PKCS12(int argc, char** argv); */ void wolfCLU_ForceZero(void* mem, unsigned int len); +#ifndef NO_FILESYSTEM +/** + * @brief open a file for writing private data. On POSIX a new file is + * created with mode 0600. An existing file is truncated and keeps its mode. + * @return file handle on success, XBADFILE on failure + */ +XFILE wolfCLU_FileOpenOwner(const char* fileName); + +/** + * @brief BIO version of wolfCLU_FileOpenOwner, the BIO closes the file + * @return new BIO on success, NULL on failure + */ +WOLFSSL_BIO* wolfCLU_BioOpenOwner(const char* fileName); +#endif + /** * @brief example client */ From 4fbddc9faceb0fe61db5f7dd1b33ff7e5fd5e179 Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 12:50:31 +0000 Subject: [PATCH 02/21] Create pkcs8 output files with owner-only permissions pkcs8 -out always holds a private key but was opened with BIO_new_file, so under umask 022 it was world-readable. Use wolfCLU_BioOpenOwner so new files are created 0600 on POSIX. F-9854 --- src/pkcs/clu_pkcs8.c | 2 +- tests/pkcs/pkcs8-test.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/src/pkcs/clu_pkcs8.c b/src/pkcs/clu_pkcs8.c index 9c4c7049..2d789176 100644 --- a/src/pkcs/clu_pkcs8.c +++ b/src/pkcs/clu_pkcs8.c @@ -109,7 +109,7 @@ int wolfCLU_PKCS8(int argc, char** argv) break; case WOLFCLU_OUTFILE: - bioOut = wolfSSL_BIO_new_file(optarg, "wb"); + bioOut = wolfCLU_BioOpenOwner(optarg); if (bioOut == NULL) { wolfCLU_LogError("Unable to open output file %s", optarg); diff --git a/tests/pkcs/pkcs8-test.py b/tests/pkcs/pkcs8-test.py index 2a23aa3d..4ae6cba5 100644 --- a/tests/pkcs/pkcs8-test.py +++ b/tests/pkcs/pkcs8-test.py @@ -72,6 +72,36 @@ def test_decrypt_and_convert(self): pkcs1_pem, shallow=False), "server-key.pem -traditional check failed") + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_out_file_owner_only(self): + """-out holds a private key, so it must be created 0600 (F-9854).""" + old_umask = os.umask(0o022) + self.addCleanup(os.umask, old_umask) + + key = os.path.join(CERTS_DIR, "server-key.pem") + cases = [ + ("perm-pkcs8.pem", ["-in", key, "-outform", "PEM"]), + ("perm-pkcs8.der", ["-in", key, "-outform", "DER"]), + ("perm-pkcs1.pem", ["-in", key, "-traditional"]), + ] + if not self.is_fips: + cases.append(("perm-pkcs8-dec.pem", + ["-in", os.path.join(CERTS_DIR, "server-keyEnc.pem"), + "-passin", "pass:yassl123"])) + + for out, args in cases: + with self.subTest(out=out): + self._cleanup(out) + # Start from a new file so the create mode applies. + if os.path.exists(out): + os.remove(out) + r = run_wolfssl("pkcs8", *(args + ["-out", out])) + self.assertEqual(r.returncode, 0, r.stderr) + mode = os.stat(out).st_mode & 0o777 + self.assertEqual(mode, 0o600, + "{} mode is {:o}, expected 600".format( + out, mode)) + def test_help(self): for flag in ("-help", "-h"): r = run_wolfssl("pkcs8", flag) From 6f169f14cfda0a5713878a4e4b809ee01676f05e Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 12:52:00 +0000 Subject: [PATCH 03/21] Create rsa private key output with owner-only permissions rsa opened -out before knowing if a private key would be written, so under umask 022 the key file was world-readable. Open -out after option parsing and use wolfCLU_BioOpenOwner unless the output is public only. -out without a file name fails instead of falling back to stdout. F-9855 --- src/pkey/clu_rsa.c | 22 +++++++++++++++---- tests/pkey/rsa-test.py | 49 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 4 deletions(-) diff --git a/src/pkey/clu_rsa.c b/src/pkey/clu_rsa.c index fa748e41..d337f5ea 100644 --- a/src/pkey/clu_rsa.c +++ b/src/pkey/clu_rsa.c @@ -77,6 +77,7 @@ int wolfCLU_RSA(int argc, char** argv) int noOut = 0; int option; int longIndex = 1; + char *outFile = NULL; WOLFSSL_BIO *bioIn = NULL; WOLFSSL_BIO *bioOut = NULL; WOLFSSL_RSA *rsa = NULL; @@ -101,10 +102,9 @@ int wolfCLU_RSA(int argc, char** argv) break; case WOLFCLU_OUTFILE: - bioOut = wolfSSL_BIO_new_file(optarg, "wb"); - if (bioOut == NULL) { - wolfCLU_LogError("unable to open out file %s", - optarg); + outFile = optarg; + if (outFile == NULL) { + wolfCLU_LogError("-out requires a file name"); ret = WOLFCLU_FATAL_ERROR; } break; @@ -201,6 +201,20 @@ int wolfCLU_RSA(int argc, char** argv) } } + /* open -out once options are known. A private key file is owner-only */ + if (ret == WOLFCLU_SUCCESS && outFile != NULL) { + if (pubOut || noOut) { + bioOut = wolfSSL_BIO_new_file(outFile, "wb"); + } + else { + bioOut = wolfCLU_BioOpenOwner(outFile); + } + if (bioOut == NULL) { + wolfCLU_LogError("unable to open out file %s", outFile); + ret = WOLFCLU_FATAL_ERROR; + } + } + /* print to stdout if no -out was used */ if (ret == WOLFCLU_SUCCESS && bioOut == NULL) { bioOut = wolfSSL_BIO_new(wolfSSL_BIO_s_file()); diff --git a/tests/pkey/rsa-test.py b/tests/pkey/rsa-test.py index 3e58b275..7266655d 100644 --- a/tests/pkey/rsa-test.py +++ b/tests/pkey/rsa-test.py @@ -179,6 +179,55 @@ def test_pubout_from_private(self): self.assertEqual(r.returncode, 0, r.stderr) self.assertEqual(r.stdout.strip(), RSA_PUBKEY_PEM) + def _out_mode(self, out, key, *args): + """Write a new -out file under umask 022 and return its mode.""" + self._cleanup(out) + if os.path.exists(out): + os.remove(out) + old_umask = os.umask(0o022) + try: + r = run_wolfssl("rsa", "-in", os.path.join(CERTS_DIR, key), + "-out", out, *args) + finally: + os.umask(old_umask) + self.assertEqual(r.returncode, 0, r.stderr) + return os.stat(out).st_mode & 0o777 + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_private_out_mode(self): + """Private key output must be owner-only (F-9855).""" + for fmt in ("PEM", "DER"): + with self.subTest(outform=fmt): + mode = self._out_mode("test-rsa-perm-priv." + fmt.lower(), + "server-key.pem", "-outform", fmt) + self.assertEqual(mode, 0o600, + "private key mode is {:o}, expected 600" + .format(mode)) + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_public_out_mode(self): + """Public only output keeps the umask default mode.""" + cases = [ + ("test-rsa-perm-pubout.pem", "server-key.pem", "-pubout"), + ("test-rsa-perm-pubin.pem", "server-keyPub.pem", "-pubin"), + ("test-rsa-perm-modulus.txt", "server-key.pem", "-noout", + "-modulus"), + ] + for out, key, *args in cases: + with self.subTest(args=args): + mode = self._out_mode(out, key, *args) + self.assertEqual(mode, 0o644, + "public output mode is {:o}, expected 644" + .format(mode)) + + + def test_out_missing_file_name(self): + """A trailing -out must fail, not print the key to stdout.""" + r = run_wolfssl("rsa", "-in", + os.path.join(CERTS_DIR, "server-key.pem"), "-out") + self.assertNotEqual(r.returncode, 0) + self.assertNotIn("PRIVATE KEY", r.stdout) + if __name__ == "__main__": test_main() From cb79142b0ff1179efa5d11a4165e024bba023b61 Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 12:50:52 +0000 Subject: [PATCH 04/21] Create dsaparam -genkey output with owner-only permissions dsaparam -genkey wrote the DSA private key through wolfSSL_BIO_new_file, so under umask 022 the file was world-readable. Open the output with wolfCLU_BioOpenOwner when -genkey is given. Params-only output is unchanged. F-9856 --- src/dsa/clu_dsa.c | 8 ++++- tests/dsa/dsa-test.py | 71 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 1 deletion(-) diff --git a/src/dsa/clu_dsa.c b/src/dsa/clu_dsa.c index be6d9eb2..9a8424ea 100644 --- a/src/dsa/clu_dsa.c +++ b/src/dsa/clu_dsa.c @@ -197,7 +197,13 @@ int wolfCLU_DsaParamSetup(int argc, char** argv) WOLFCLU_LOG(WOLFCLU_E0, "No filesystem support. Unable to open input file"); ret = WOLFCLU_FATAL_ERROR; #else - bioOut = wolfSSL_BIO_new_file(out, "wb"); + /* a generated key is private, so create the file owner-only */ + if (genKey) { + bioOut = wolfCLU_BioOpenOwner(out); + } + else { + bioOut = wolfSSL_BIO_new_file(out, "wb"); + } if (bioOut == NULL) { wolfCLU_LogError("Unable to open output file %s", optarg); diff --git a/tests/dsa/dsa-test.py b/tests/dsa/dsa-test.py index 2f4b9c8d..5c7896f8 100644 --- a/tests/dsa/dsa-test.py +++ b/tests/dsa/dsa-test.py @@ -93,6 +93,77 @@ def test_bad_input_fails(self): "-genkey", "-noout") self.assertNotEqual(r.returncode, 0) + def _new_file(self, name): + # Start from no file so the create mode applies. + if os.path.exists(name): + os.remove(name) + self.addCleanup(lambda: os.remove(name) + if os.path.exists(name) else None) + return name + + def _mode(self, name): + return os.stat(name).st_mode & 0o777 + + def _use_umask_022(self): + old_umask = os.umask(0o022) + self.addCleanup(os.umask, old_umask) + + def _gen_params(self, params_file): + r = run_wolfssl("dsaparam", "-out", self._new_file(params_file), + "1024") + self.assertEqual(r.returncode, 0, r.stderr) + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_dsaparam_genkey_out_mode(self): + """-genkey output holds a private key so it must be owner-only.""" + params_file = "dsa-perm-genkey.params" + key_file = self._new_file("dsa-perm-genkey.key") + self._use_umask_022() + self._gen_params(params_file) + + r = run_wolfssl("dsaparam", "-in", params_file, "-genkey", "-noout", + "-out", key_file) + self.assertEqual(r.returncode, 0, r.stderr) + with open(key_file, "r") as f: + self.assertIn("-----BEGIN DSA PRIVATE KEY-----", f.read()) + mode = self._mode(key_file) + self.assertEqual(mode, 0o600, + "DSA private key mode is {:o}, expected 600".format( + mode)) + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_dsaparam_genkey_with_params_out_mode(self): + """Params and key in one file: the file must be owner-only.""" + params_file = "dsa-perm-both.params" + key_file = self._new_file("dsa-perm-both.key") + self._use_umask_022() + self._gen_params(params_file) + + r = run_wolfssl("dsaparam", "-in", params_file, "-genkey", + "-out", key_file) + self.assertEqual(r.returncode, 0, r.stderr) + with open(key_file, "r") as f: + data = f.read() + self.assertIn("-----BEGIN DSA PARAMETERS-----", data) + self.assertIn("-----BEGIN DSA PRIVATE KEY-----", data) + mode = self._mode(key_file) + self.assertEqual(mode, 0o600, + "DSA private key mode is {:o}, expected 600".format( + mode)) + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_dsaparam_params_out_mode(self): + """Parameter-only output is public and keeps default permissions.""" + params_file = "dsa-perm-params.params" + copy_file = self._new_file("dsa-perm-params-copy.params") + self._use_umask_022() + self._gen_params(params_file) + self.assertEqual(self._mode(params_file), 0o644) + + r = run_wolfssl("dsaparam", "-in", params_file, "-out", copy_file) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(self._mode(copy_file), 0o644) + if __name__ == "__main__": test_main() From 22e3f0fe6585dbc48a2e57492b76df6250a4fa14 Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 13:03:25 +0000 Subject: [PATCH 05/21] Create pkcs12 -out files with owner-only permissions pkcs12 opened -out with BIO_new_file, so extracted private keys were world-readable under umask 022. Defer opening until options are parsed and use wolfCLU_BioOpenOwner unless -nokeys is given. -out without a file name fails instead of falling back to stdout. F-9861 --- src/pkcs/clu_pkcs12.c | 22 +++++++++++++---- tests/pkcs/pkcs12-test.py | 50 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 4 deletions(-) diff --git a/src/pkcs/clu_pkcs12.c b/src/pkcs/clu_pkcs12.c index 29b08219..d41a4227 100644 --- a/src/pkcs/clu_pkcs12.c +++ b/src/pkcs/clu_pkcs12.c @@ -74,6 +74,7 @@ int wolfCLU_PKCS12(int argc, char** argv) WOLF_STACK_OF(WOLFSSL_X509) *extra = NULL; WOLFSSL_BIO *bioIn = NULL; WOLFSSL_BIO *bioOut = NULL; + char *outFile = NULL; opterr = 0; /* do not display unrecognized options */ optind = 0; /* start at indent 0 */ @@ -115,10 +116,9 @@ int wolfCLU_PKCS12(int argc, char** argv) break; case WOLFCLU_OUTFILE: - bioOut = wolfSSL_BIO_new_file(optarg, "wb"); - if (bioOut == NULL) { - wolfCLU_LogError("Unable to open output file %s", - optarg); + outFile = optarg; + if (outFile == NULL) { + wolfCLU_LogError("-out requires a file name"); ret = WOLFCLU_FATAL_ERROR; } break; @@ -140,6 +140,20 @@ int wolfCLU_PKCS12(int argc, char** argv) } } + /* output with a key, even encrypted, is created owner only */ + if (ret == WOLFCLU_SUCCESS && outFile != NULL) { + if (printKeys) { + bioOut = wolfCLU_BioOpenOwner(outFile); + } + else { + bioOut = wolfSSL_BIO_new_file(outFile, "wb"); + } + if (bioOut == NULL) { + wolfCLU_LogError("Unable to open output file %s", outFile); + ret = WOLFCLU_FATAL_ERROR; + } + } + /* with currently only supporting PKCS12 parsing, an input file is expected */ if (ret == WOLFCLU_SUCCESS && bioIn == NULL) { wolfCLU_LogError("No input file set"); diff --git a/tests/pkcs/pkcs12-test.py b/tests/pkcs/pkcs12-test.py index 7d067f1c..63911261 100644 --- a/tests/pkcs/pkcs12-test.py +++ b/tests/pkcs/pkcs12-test.py @@ -85,6 +85,56 @@ def test_out_bad_path_fails(self): "-out", os.path.join("no-such-dir", "out.pem")) self.assertNotEqual(r.returncode, 0) + def test_out_missing_file_name(self): + """A trailing -out must fail, not print the key to stdout.""" + r = run_wolfssl("pkcs12", "-nodes", "-passin", 'pass:wolfSSL test', + "-passout", "pass:", "-in", P12_FILE, "-out") + self.assertNotEqual(r.returncode, 0) + self.assertNotIn("PRIVATE KEY", r.stdout) + + def _out_mode(self, out, *args, stdin_data=None): + """Run pkcs12 -out under umask 022. Return the new file's mode and + contents.""" + old_umask = os.umask(0o022) + self.addCleanup(os.umask, old_umask) + self.addCleanup(lambda: os.remove(out) if os.path.exists(out) else None) + # Start from a new file so the create mode applies. + if os.path.exists(out): + os.remove(out) + r = run_wolfssl("pkcs12", "-passin", 'pass:wolfSSL test', + "-passout", "pass:", "-in", P12_FILE, "-out", out, + *args, stdin_data=stdin_data) + self.assertEqual(r.returncode, 0, r.stderr) + with open(out, "r") as f: + content = f.read() + return os.stat(out).st_mode & 0o777, content + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_out_nodes_owner_only(self): + mode, content = self._out_mode("pkcs12-perm-nodes.pem", "-nodes") + self.assertIn("PRIVATE KEY", content) + self.assertEqual(mode, 0o600, + "-nodes output mode is {:o}, expected 600".format( + mode)) + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_out_encrypted_key_owner_only(self): + mode, content = self._out_mode("pkcs12-perm-enc.pem", + stdin_data="wolfSSL test\n") + self.assertIn("ENCRYPTED PRIVATE KEY", content) + self.assertEqual(mode, 0o600, + "encrypted key output mode is {:o}, " + "expected 600".format(mode)) + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_out_nokeys_default_mode(self): + mode, content = self._out_mode("pkcs12-perm-nokeys.pem", "-nokeys") + self.assertNotIn("KEY", content) + self.assertIn("CERTIFICATE", content) + self.assertEqual(mode, 0o644, + "-nokeys output mode is {:o}, expected 644".format( + mode)) + def test_nocerts_with_passout(self): r = subprocess.run( [WOLFSSL_BIN, "pkcs12", "-passin", "stdin", "-passout", "pass:", From 73543d4d9a8d8d366a1830bf08fa129198e935cf Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 12:50:41 +0000 Subject: [PATCH 06/21] Create req -keyout private key with owner-only permissions req wrote the generated private key with BIO_new_file, so under umask 022 it was world-readable. Open -keyout with wolfCLU_BioOpenOwner so a new key file gets mode 0600, encrypted or not. F-9860 --- src/x509/clu_request_setup.c | 2 +- tests/x509/x509-req-test.py | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/src/x509/clu_request_setup.c b/src/x509/clu_request_setup.c index 6cea40c0..469cf68c 100644 --- a/src/x509/clu_request_setup.c +++ b/src/x509/clu_request_setup.c @@ -1110,7 +1110,7 @@ int wolfCLU_requestSetup(int argc, char** argv) WOLFSSL_BIO* keyOutBio; if (keyOut != NULL) { - keyOutBio = wolfSSL_BIO_new_file(keyOut, "wb"); + keyOutBio = wolfCLU_BioOpenOwner(keyOut); } else { keyOutBio = wolfSSL_BIO_new(wolfSSL_BIO_s_file()); diff --git a/tests/x509/x509-req-test.py b/tests/x509/x509-req-test.py index 82b1153e..259df852 100644 --- a/tests/x509/x509-req-test.py +++ b/tests/x509/x509-req-test.py @@ -990,6 +990,40 @@ def test_newkey_with_passout_keyout(self): self.assertEqual(r.returncode, 0, r.stderr) +@unittest.skipIf(os.name == "nt", "POSIX file permissions only") +class TestReqKeyoutPermissions(unittest.TestCase): + """req -keyout must be owner-only under umask 022 (F-9860).""" + + def setUp(self): + old_umask = os.umask(0o022) + self.addCleanup(os.umask, old_umask) + + def _new_key(self, name, *extra): + key = _tmp(name + ".pem") + csr = _tmp(name + ".csr") + # Start from new files so the create mode applies. + _cleanup(key, csr) + self.addCleanup(_cleanup, key, csr) + r = run_wolfssl("req", "-new", "-newkey", "rsa:2048", + "-keyout", key, "-out", csr, + "-subj", "O=wolfSSL/C=US/CN=keyout-perm", *extra) + self.assertEqual(r.returncode, 0, r.stderr) + key_mode = os.stat(key).st_mode & 0o777 + csr_mode = os.stat(csr).st_mode & 0o777 + self.assertEqual(key_mode, 0o600, + "keyout mode is {:o}, expected 600".format(key_mode)) + self.assertEqual(csr_mode, 0o644, + "csr mode is {:o}, expected 644".format(csr_mode)) + + def test_nodes_keyout_mode(self): + self._new_key("test_req_perm_nodes", "-nodes") + + def test_encrypted_keyout_mode(self): + if is_fips(): + self.skipTest("FIPS build") + self._new_key("test_req_perm_enc", "-passout", + "pass:123456789wolfssl") + class TestReqHashAndKeyAlgos(unittest.TestCase): """Test hash and key algorithm options for req.""" From 66069244e5bcf4f50576bc7046af2d4e8ca5a64f Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 12:29:20 +0000 Subject: [PATCH 07/21] Fail s_server on compiled-out protocol versions server_test continued with a NULL method and ctx when the requested version was not compiled in. Treat a NULL method or ctx as fatal and propagate server_test's return code so s_server exits non-zero. F-12925 --- src/server/clu_server_setup.c | 5 ++++ src/server/server.c | 14 ++++++--- tests/server/server-test.py | 53 +++++++++++++++++++++++++++++++++++ 3 files changed, 68 insertions(+), 4 deletions(-) diff --git a/src/server/clu_server_setup.c b/src/server/clu_server_setup.c index 467ceb23..2f826c62 100644 --- a/src/server/clu_server_setup.c +++ b/src/server/clu_server_setup.c @@ -200,6 +200,11 @@ int wolfCLU_Server(int argc, char** argv) args.argv = (char**)serverArgv; args.argc = serverArgc; server_test(&args); + + if (args.return_code != 0) { + wolfCLU_LogError("s_server failed (%d).", args.return_code); + ret = WOLFCLU_FATAL_ERROR; + } } exit: diff --git a/src/server/server.c b/src/server/server.c index 5b9c70dd..c40ee0e6 100644 --- a/src/server/server.c +++ b/src/server/server.c @@ -2505,8 +2505,11 @@ THREAD_RETURN WOLFSSL_THREAD server_test(void* args) fprintf(stderr, "Bad SSL version\n"); } - if (method == NULL) - fprintf(stderr, "unable to get method\n"); + if (method == NULL) { + release(ctx, ssl, "unable to get method"); + ((func_args*)args)->return_code = VERSION_ERROR; + goto exit; + } #ifdef WOLFSSL_STATIC_MEMORY #ifdef DEBUG_WOLFSSL @@ -2538,8 +2541,11 @@ THREAD_RETURN WOLFSSL_THREAD server_test(void* args) wolfSSL_CTX_set_msg_callback(ctx, msgDebugCb); #endif #endif /* WOLFSSL_STATIC_MEMORY */ - if (ctx == NULL) - fprintf(stderr, "unable to get ctx\n"); + if (ctx == NULL) { + release(ctx, ssl, "unable to get ctx"); + ((func_args*)args)->return_code = SSL_ERROR_SSL; + goto exit; + } if (minVersion != SERVER_INVALID_VERSION) { #ifdef WOLFSSL_DTLS diff --git a/tests/server/server-test.py b/tests/server/server-test.py index 4e795f05..1cd1fe84 100644 --- a/tests/server/server-test.py +++ b/tests/server/server-test.py @@ -89,6 +89,59 @@ def test_server_client(self): server.kill() server.wait() + def _run_server_version(self, version): + """Start s_server with -version and wait for it to exit or listen. + + Returns (returncode, stderr). A server that starts listening is + killed and returncode is None. + """ + readyfile = "readyfile-version-" + version + if os.path.exists(readyfile): + os.remove(readyfile) + self.addCleanup( + lambda: os.path.exists(readyfile) and os.remove(readyfile)) + + server = subprocess.Popen( + [WOLFSSL_BIN, "s_server", "-port", str(find_free_port()), + "-key", os.path.join(CERTS_DIR, "server-key.pem"), + "-cert", os.path.join(CERTS_DIR, "server-cert.pem"), + "-version", version, "-noVerify", "-readyFile", readyfile], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, + stdin=subprocess.DEVNULL, text=True, + ) + try: + deadline = time.time() + 30 + while True: + try: + _, err = server.communicate(timeout=0.1) + return server.returncode, err + except subprocess.TimeoutExpired: + if os.path.exists(readyfile): + server.kill() + _, err = server.communicate() + return None, err + if time.time() > deadline: + self.fail("s_server neither exited nor listened") + finally: + if server.poll() is None: + server.kill() + server.communicate() + + def test_unsupported_version(self): + """s_server fails when the protocol version is not compiled in.""" + # 0: SSLv3, 1: TLS 1.0, 2: TLS 1.1 + for version in ("0", "1", "2"): + with self.subTest(version=version): + rc, err = self._run_server_version(version) + if rc is None: + self.assertNotIn("unable to get method", err, + "s_server kept running without a " + "method: " + err) + self.skipTest("version {} compiled in".format(version)) + self.assertNotEqual(rc, 0, err) + self.assertIn("unable to get method", err) + self.assertNotIn("listening on port", err) + if __name__ == "__main__": test_main() From 86cec9ce2d5e1cc423686d09fa387acb9ef643c8 Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 12:54:57 +0000 Subject: [PATCH 08/21] Bound s_server throughput remainder before narrowing to int ServerEchoData cast the size_t throughput remainder to int before min(). Remainders that are multiples of 2^32 truncated to 0 and the echo loop spun forever. Compare against the block size as size_t first. F-9845 --- src/server/server.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/server/server.c b/src/server/server.c index c40ee0e6..30540dc6 100644 --- a/src/server/server.c +++ b/src/server/server.c @@ -393,8 +393,8 @@ static int ServerEchoData(SSL* ssl, int clientfd, int echoData, int block, int select_ret = tcp_select(clientfd, 1); /* Timeout=1 second */ if (select_ret == TEST_RECV_READY) { - if (throughput) - len = min(block, (int)(throughput - xfer_bytes)); + if (throughput && throughput - xfer_bytes < (size_t)block) + len = (int)(throughput - xfer_bytes); else len = block; rx_pos = 0; From 576e8fe088bc08802597af6ba73f7927664da411 Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 12:29:11 +0000 Subject: [PATCH 09/21] Reject abbreviated or unsupported cipher modes in parseAlgo Mode prefixes like "c" or "cb" passed validation but set no cipher, so encrypt/decrypt used an uninitialised alg and wrote garbage output. Require an exact mode, start alg at WOLFCLU_ALGO_NONE and fail if no cipher was chosen. Initialise alg in wolfCLU_setup too. F-11081 --- src/crypto/clu_crypto_setup.c | 2 +- src/tools/clu_funcs.c | 8 ++++- tests/encrypt/enc-test.py | 62 +++++++++++++++++++++++++++++++++++ wolfclu/clu_optargs.h | 3 ++ 4 files changed, 73 insertions(+), 2 deletions(-) diff --git a/src/crypto/clu_crypto_setup.c b/src/crypto/clu_crypto_setup.c index d375a78a..ce0113ae 100644 --- a/src/crypto/clu_crypto_setup.c +++ b/src/crypto/clu_crypto_setup.c @@ -219,7 +219,7 @@ int wolfCLU_setup(int argc, char** argv, char action) char outNameDec[256]; /* default outfile for decrypt */ char inName[256]; /* name of the in File if not provided */ - int alg; /* algorithm from name */ + int alg = WOLFCLU_ALGO_NONE; /* algorithm from name */ char* mode = NULL; /* mode from name */ char* out = NULL; /* default output file name */ char* in = inName; /* default in data */ diff --git a/src/tools/clu_funcs.c b/src/tools/clu_funcs.c index ae987c73..422e7c91 100644 --- a/src/tools/clu_funcs.c +++ b/src/tools/clu_funcs.c @@ -297,6 +297,7 @@ static int wolfCLU_parseAlgo(char* name, int* alg, char** mode, int* size) wolfCLU_LogError("null input to get algo function"); return WOLFCLU_FATAL_ERROR; } + *alg = WOLFCLU_ALGO_NONE; /* gets name after first '-' and before the second */ tmpAlg = strtok_r(name, "-", &end); @@ -338,7 +339,7 @@ static int wolfCLU_parseAlgo(char* name, int* alg, char** mode, int* size) } for (i = 0; i < (int) (sizeof(acceptMode)/sizeof(acceptMode[0])); i++) { - if (XSTRNCMP(tmpMode, acceptMode[i], XSTRLEN(tmpMode)) == 0) + if (XSTRCMP(tmpMode, acceptMode[i]) == 0) modeCheck = 1; } @@ -436,6 +437,11 @@ static int wolfCLU_parseAlgo(char* name, int* alg, char** mode, int* size) ret = WOLFCLU_FATAL_ERROR; } + if (ret >= 0 && *alg == WOLFCLU_ALGO_NONE) { + wolfCLU_LogError("Invalid mode %s for algorithm %s", tmpMode, tmpAlg); + ret = WOLFCLU_FATAL_ERROR; + } + if (ret >= 0) { int s; diff --git a/tests/encrypt/enc-test.py b/tests/encrypt/enc-test.py index 5fa48c7f..04884c04 100644 --- a/tests/encrypt/enc-test.py +++ b/tests/encrypt/enc-test.py @@ -463,6 +463,68 @@ def test_legacy_aes_cbc_128_roundtrip(self): self._roundtrip("-aes-cbc-128", "-aes-cbc-128", "legacy aes-cbc-128 round trip failed") + # Modes must match exactly. A truncated mode (or one the algorithm does + # not support) selects no cipher, so the command must fail. + BAD_MODE_NAMES = ["aes-c-128", "aes-cb-256", "aes-ct-192", "aes-128-c", + "aes-256-cb", "camellia-c-128", "camellia-cb-256", + "camellia-ctr-128"] + + def test_abbreviated_mode_encrypt_rejected(self): + orig = os.path.join(CERTS_DIR, "crl.der") + for i, name in enumerate(self.BAD_MODE_NAMES): + for j, args in enumerate((["-encrypt", name], + ["enc", "-" + name])): + enc = "test-abbrev-{}-{}.enc".format(i, j) + self._cleanup(enc) + with self.subTest(args=args): + r = run_enc(*args, "-in", orig, "-out", enc, + password="test password") + self.assertNotEqual(r.returncode, 0, + "{} must be rejected".format(name)) + self.assertRegex(r.stderr, "Invalid (entry|mode)", + "{} not rejected when parsed".format( + name)) + self.assertFalse(os.path.exists(enc), + "{} left an output file".format(name)) + + def test_abbreviated_mode_decrypt_rejected(self): + src = os.path.join(CERTS_DIR, "crl.der.enc") + for i, name in enumerate(self.BAD_MODE_NAMES): + dec = "test-abbrev-{}.dec".format(i) + self._cleanup(dec) + with self.subTest(name=name): + r = run_enc("-decrypt", name, "-in", src, "-out", dec, + password="") + self.assertNotEqual(r.returncode, 0, + "{} must be rejected".format(name)) + self.assertRegex(r.stderr, "Invalid (entry|mode)", + "{} not rejected when parsed".format(name)) + self.assertFalse(os.path.exists(dec), + "{} left an output file".format(name)) + + def test_full_mode_names_roundtrip(self): + names = ["aes-cbc-128", "aes-256-cbc"] + r = run_wolfssl("-encrypt", "-help") + if "aes-ctr-128" in r.stdout + r.stderr: + names.append("aes-ctr-192") + if _camellia_available(): + names.append("camellia-cbc-128") + + orig = os.path.join(CERTS_DIR, "crl.der") + for name in names: + enc = "test-fullmode-{}.enc".format(name) + dec = "test-fullmode-{}.dec".format(name) + self._cleanup(enc, dec) + with self.subTest(name=name): + r = run_enc("-encrypt", name, "-in", orig, "-out", enc, + password="test password") + self.assertEqual(r.returncode, 0, r.stderr) + r = run_enc("-decrypt", name, "-in", enc, "-out", dec, + password="test password") + self.assertEqual(r.returncode, 0, r.stderr) + self.assertTrue(filecmp.cmp(orig, dec, shallow=False), + "{} round trip failed".format(name)) + def _camellia_available(): """Check if Camellia support is enabled in the wolfssl binary.""" diff --git a/wolfclu/clu_optargs.h b/wolfclu/clu_optargs.h index ef77a8fd..0b275bc9 100644 --- a/wolfclu/clu_optargs.h +++ b/wolfclu/clu_optargs.h @@ -174,6 +174,9 @@ enum { /* 3des */ #define WOLFCLU_DESCBC 2018 +/* no cipher selected */ +#define WOLFCLU_ALGO_NONE 0 + /* PQC signeture */ #define WOLFCLU_DILITHIUM 2019 #define WOLFCLU_XMSS 2020 From d01da476dcf4513575427c382c38ddc73e08a26e Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 12:55:20 +0000 Subject: [PATCH 10/21] Fail cleanly on EOF at the interactive password prompt wolfCLU_GetStdinPassword ran strlen on the buffer after fgets failed, reading uninitialised caller buffers. It now returns an empty password of size 0 on failure. pkcs12, pkcs8 and req check the result instead of encrypting or decrypting with a stale or garbage password. F-12121 --- src/pkcs/clu_pkcs12.c | 12 +++++-- src/pkcs/clu_pkcs8.c | 5 ++- src/tools/clu_funcs.c | 10 ++++-- src/x509/clu_request_setup.c | 7 ++-- tests/encrypt/enc-test.py | 18 +++++++++- tests/pkcs/pkcs12-test.py | 50 ++++++++++++++++++++++----- tests/pkcs/pkcs8-test.py | 18 +++++++++- tests/wolfclu_test.py | 67 ++++++++++++++++++++++++++++++++++++ tests/x509/x509-req-test.py | 23 ++++++++++++- 9 files changed, 191 insertions(+), 19 deletions(-) diff --git a/src/pkcs/clu_pkcs12.c b/src/pkcs/clu_pkcs12.c index d41a4227..167c6c63 100644 --- a/src/pkcs/clu_pkcs12.c +++ b/src/pkcs/clu_pkcs12.c @@ -241,9 +241,15 @@ int wolfCLU_PKCS12(int argc, char** argv) if (ret == WOLFCLU_SUCCESS && pkey != NULL && printKeys) { if (useDES) { passwordSz = MAX_PASSWORD_SIZE; - wolfCLU_GetStdinPassword((byte*)password, (word32*)&passwordSz); - ret = wolfCLU_pKeyPEMtoPriKeyEnc(bioOut, pkey, DES3b, - (byte*)password, passwordSz); + ret = wolfCLU_GetStdinPassword((byte*)password, + (word32*)&passwordSz); + if (ret != WOLFCLU_SUCCESS) { + wolfCLU_LogError("Unable to get password from stdin"); + } + else { + ret = wolfCLU_pKeyPEMtoPriKeyEnc(bioOut, pkey, DES3b, + (byte*)password, passwordSz); + } } else { ret = wolfCLU_pKeyPEMtoPriKey(bioOut, pkey); diff --git a/src/pkcs/clu_pkcs8.c b/src/pkcs/clu_pkcs8.c index 2d789176..699e8a49 100644 --- a/src/pkcs/clu_pkcs8.c +++ b/src/pkcs/clu_pkcs8.c @@ -213,7 +213,10 @@ int wolfCLU_PKCS8(int argc, char** argv) } if (ret == WOLFCLU_SUCCESS && pass == NULL && pkey == NULL) { - wolfCLU_GetStdinPassword((byte*)password, (word32*)&passwordSz); + ret = wolfCLU_GetStdinPassword((byte*)password, (word32*)&passwordSz); + if (ret != WOLFCLU_SUCCESS) { + wolfCLU_LogError("Unable to get password from stdin"); + } pass = (byte*)password; } diff --git a/src/tools/clu_funcs.c b/src/tools/clu_funcs.c index 422e7c91..a18675ce 100644 --- a/src/tools/clu_funcs.c +++ b/src/tools/clu_funcs.c @@ -1298,7 +1298,7 @@ int wolfCLU_GetStdinPassword(byte* password, word32* passwordSz) DWORD originalTerm; #endif - if (password == NULL || passwordSz == NULL) { + if (password == NULL || passwordSz == NULL || *passwordSz == 0) { return WOLFCLU_FATAL_ERROR; } @@ -1312,10 +1312,16 @@ int wolfCLU_GetStdinPassword(byte* password, word32* passwordSz) char* c = strpbrk((char*)password, "\r\n"); if (c != NULL) *c = '\0'; + *passwordSz = (word32)XSTRLEN((const char*)password); } - *passwordSz = (word32)XSTRLEN((const char*)password); ShowEcho(&originalTerm); } + + /* On EOF or error the buffer holds no valid string. */ + if (ret != WOLFCLU_SUCCESS) { + password[0] = '\0'; + *passwordSz = 0; + } return ret; } #endif diff --git a/src/x509/clu_request_setup.c b/src/x509/clu_request_setup.c index 469cf68c..05cd1693 100644 --- a/src/x509/clu_request_setup.c +++ b/src/x509/clu_request_setup.c @@ -1131,9 +1131,12 @@ int wolfCLU_requestSetup(int argc, char** argv) if (useDes) { if (!passout) { byte pass[MAX_PASSWORD_SIZE]; - wolfCLU_GetStdinPassword(pass, (word32*)&passwordSz); + ret = wolfCLU_GetStdinPassword(pass, (word32*)&passwordSz); - if (pass[0] == '\0') { + if (ret != WOLFCLU_SUCCESS) { + wolfCLU_LogError("Unable to get password from stdin"); + } + else if (pass[0] == '\0') { wolfCLU_LogError("Please enter a password"); ret = WOLFCLU_FATAL_ERROR; } diff --git a/tests/encrypt/enc-test.py b/tests/encrypt/enc-test.py index 04884c04..3242b8f6 100644 --- a/tests/encrypt/enc-test.py +++ b/tests/encrypt/enc-test.py @@ -11,7 +11,8 @@ import unittest sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from wolfclu_test import CERTS_DIR, WOLFSSL_BIN, run_wolfssl, test_main +from wolfclu_test import (CERTS_DIR, WOLFSSL_BIN, run_wolfssl, + run_wolfssl_pty, test_main) # The interactive password prompt only reads from stdin when stdin is a real # terminal (wolfCLU_GetStdinPassword -> tcgetattr fails on a pipe), so driving @@ -1156,6 +1157,21 @@ def test_default_kdf_stdin_decrypts_with_pass(self): self.assertEqual(f.read(), self.PLAINTEXT, "decrypted plaintext mismatch") + def test_password_prompt_eof_fails(self): + """EOF at the password prompt must fail and write no output.""" + plain = "f12121_eof_in.txt" + cipher = "f12121_eof.bin" + self._cleanup(plain, cipher) + self._write_plaintext(plain) + + code, out = run_wolfssl_pty("encrypt", "aes-cbc-256", + "-in", plain, "-out", cipher, + reply=b"\x04") + self.assertIn(b"Input Password", out) + self.assertGreater(code, 0, out) + self.assertFalse(os.path.exists(cipher), out) + self.assertNotIn(b"AddressSanitizer", out) + if __name__ == "__main__": test_main() diff --git a/tests/pkcs/pkcs12-test.py b/tests/pkcs/pkcs12-test.py index 63911261..53ba635d 100644 --- a/tests/pkcs/pkcs12-test.py +++ b/tests/pkcs/pkcs12-test.py @@ -7,7 +7,8 @@ import unittest sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from wolfclu_test import WOLFSSL_BIN, CERTS_DIR, is_fips, run_wolfssl, test_main +from wolfclu_test import (WOLFSSL_BIN, CERTS_DIR, HAVE_PTY, is_fips, + run_wolfssl, run_wolfssl_pty, test_main) P12_FILE = os.path.join(CERTS_DIR, "test-servercert.p12") @@ -92,7 +93,7 @@ def test_out_missing_file_name(self): self.assertNotEqual(r.returncode, 0) self.assertNotIn("PRIVATE KEY", r.stdout) - def _out_mode(self, out, *args, stdin_data=None): + def _out_mode(self, out, *args): """Run pkcs12 -out under umask 022. Return the new file's mode and contents.""" old_umask = os.umask(0o022) @@ -103,7 +104,7 @@ def _out_mode(self, out, *args, stdin_data=None): os.remove(out) r = run_wolfssl("pkcs12", "-passin", 'pass:wolfSSL test', "-passout", "pass:", "-in", P12_FILE, "-out", out, - *args, stdin_data=stdin_data) + *args) self.assertEqual(r.returncode, 0, r.stderr) with open(out, "r") as f: content = f.read() @@ -118,10 +119,23 @@ def test_out_nodes_owner_only(self): mode)) @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + @unittest.skipUnless(HAVE_PTY, "pty not available") def test_out_encrypted_key_owner_only(self): - mode, content = self._out_mode("pkcs12-perm-enc.pem", - stdin_data="wolfSSL test\n") - self.assertIn("ENCRYPTED PRIVATE KEY", content) + # The key password is read from a terminal. + out = "pkcs12-perm-enc.pem" + old_umask = os.umask(0o022) + self.addCleanup(os.umask, old_umask) + self.addCleanup(lambda: os.remove(out) if os.path.exists(out) else None) + if os.path.exists(out): + os.remove(out) + code, output = run_wolfssl_pty("pkcs12", "-passin", + "pass:wolfSSL test", "-passout", "pass:", + "-in", P12_FILE, "-out", out, + reply=b"wolfSSL test\n") + self.assertEqual(code, 0, output) + with open(out, "r") as f: + self.assertIn("ENCRYPTED PRIVATE KEY", f.read()) + mode = os.stat(out).st_mode & 0o777 self.assertEqual(mode, 0o600, "encrypted key output mode is {:o}, " "expected 600".format(mode)) @@ -135,14 +149,34 @@ def test_out_nokeys_default_mode(self): "-nokeys output mode is {:o}, expected 644".format( mode)) - def test_nocerts_with_passout(self): + def test_nocerts_with_passout_no_terminal_fails(self): + """Without -nodes the key password is read from a terminal. With no + terminal the command must fail and write no key. + + It used to succeed with the key encrypted under the whole 256-byte + password buffer, which no typed password matches.""" r = subprocess.run( [WOLFSSL_BIN, "pkcs12", "-passin", "stdin", "-passout", "pass:", "-in", P12_FILE, "-nocerts"], input=b"wolfSSL test\n", capture_output=True, text=False, timeout=60, ) - self.assertEqual(r.returncode, 0, r.stderr) + self.assertNotEqual(r.returncode, 0) + self.assertNotIn(b"PRIVATE KEY", r.stdout) + + @unittest.skipUnless(HAVE_PTY, "pty not available") + def test_key_password_prompt_eof_fails(self): + """EOF at the key password prompt must fail and write no key. + + The prompt buffer still held the -passin password, so the key was + written encrypted under it and the command succeeded.""" + code, out = run_wolfssl_pty("pkcs12", "-nocerts", + "-passin", "pass:wolfSSL test", + "-in", P12_FILE, reply=b"\x04") + self.assertIn(b"Input Password", out) + self.assertGreater(code, 0, out) + self.assertNotIn(b"PRIVATE KEY", out) + self.assertNotIn(b"AddressSanitizer", out) if __name__ == "__main__": diff --git a/tests/pkcs/pkcs8-test.py b/tests/pkcs/pkcs8-test.py index 4ae6cba5..6e84614d 100644 --- a/tests/pkcs/pkcs8-test.py +++ b/tests/pkcs/pkcs8-test.py @@ -8,7 +8,8 @@ import unittest sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from wolfclu_test import WOLFSSL_BIN, CERTS_DIR, is_fips, run_wolfssl, test_main +from wolfclu_test import (WOLFSSL_BIN, CERTS_DIR, HAVE_PTY, is_fips, + run_wolfssl, run_wolfssl_pty, test_main) class Pkcs8Test(unittest.TestCase): @@ -146,6 +147,21 @@ def test_fail_wrong_format(self): "-inform", "DER", "-passin", "pass:yassl123") self.assertNotEqual(r.returncode, 0) + @unittest.skipUnless(HAVE_PTY, "pty not available") + def test_password_prompt_eof_fails(self): + """EOF at the password prompt for an encrypted key must fail cleanly. + + The password buffer was measured with strlen after the failed read, + although nothing had been written to it.""" + code, out = run_wolfssl_pty( + "pkcs8", "-in", os.path.join(CERTS_DIR, "server-keyEnc.pem"), + reply=b"\x04") + self.assertIn(b"Input Password", out) + self.assertIn(b"Unable to get password from stdin", out) + self.assertGreater(code, 0, out) + self.assertNotIn(b"PRIVATE KEY", out) + self.assertNotIn(b"AddressSanitizer", out) + if __name__ == "__main__": test_main() diff --git a/tests/wolfclu_test.py b/tests/wolfclu_test.py index 6909c2e3..003c1ac7 100644 --- a/tests/wolfclu_test.py +++ b/tests/wolfclu_test.py @@ -88,6 +88,73 @@ def run_wolfssl(*args, stdin_data=None, timeout=60): return subprocess.run(cmd, **kwargs) +try: + import pty + HAVE_PTY = True +except ImportError: + # POSIX only; not available on Windows. + HAVE_PTY = False + + +def run_wolfssl_pty(*args, reply=b"", prompt=b"Input Password", timeout=30): + """Run the wolfssl binary on a pseudo-terminal. + + Interactive password prompts need a terminal on stdin. Once `prompt` + appears in the output, `reply` is written to the terminal (b"\\x04" sends + EOF). Returns (returncode, output), where output is the combined terminal + output as bytes and a negative returncode is the signal that killed it. + """ + import select + import signal + import time + + cmd = [WOLFSSL_BIN] + list(args) + pid, fd = pty.fork() + if pid == 0: + try: + os.execv(cmd[0], cmd) + finally: + os._exit(127) + + output = b"" + sent = False + deadline = time.monotonic() + timeout + try: + while time.monotonic() < deadline: + ready, _, _ = select.select([fd], [], [], 0.5) + if not ready: + continue + try: + data = os.read(fd, 4096) + except OSError: + break # EIO once the child has closed the terminal + if not data: + break + output += data + if not sent and prompt in output: + os.write(fd, reply) + sent = True + finally: + os.close(fd) + + # Bounded reap so a child stuck at the prompt cannot hang the suite. + status = None + end = time.monotonic() + 5 + while status is None and time.monotonic() < end: + wpid, st = os.waitpid(pid, os.WNOHANG) + if wpid == pid: + status = st + else: + time.sleep(0.05) + if status is None: + os.kill(pid, signal.SIGKILL) + _, status = os.waitpid(pid, 0) + + if os.WIFEXITED(status): + return os.WEXITSTATUS(status), output + return -os.WTERMSIG(status), output + + def is_fips(): """True when linked against a FIPS wolfSSL build (per `wolfssl -v`).""" r = run_wolfssl("-v") diff --git a/tests/x509/x509-req-test.py b/tests/x509/x509-req-test.py index 259df852..7dd3a6f4 100644 --- a/tests/x509/x509-req-test.py +++ b/tests/x509/x509-req-test.py @@ -9,7 +9,8 @@ import unittest sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from wolfclu_test import WOLFSSL_BIN, CERTS_DIR, is_fips, run_wolfssl, test_main +from wolfclu_test import (WOLFSSL_BIN, CERTS_DIR, HAVE_PTY, is_fips, + run_wolfssl, run_wolfssl_pty, test_main) def _tmp(name): @@ -989,6 +990,26 @@ def test_newkey_with_passout_keyout(self): stdin_data="long test password\n") self.assertEqual(r.returncode, 0, r.stderr) + @unittest.skipUnless(HAVE_PTY, "pty not available") + def test_newkey_password_prompt_eof_fails(self): + """EOF at the -newkey key password prompt must fail cleanly. + + The uninitialised password buffer was measured with strlen after + the failed read.""" + tmp = _tmp("test_req_prompt_eof.cert") + key = _tmp("test_req_prompt_eof.pem") + self._clean(tmp, key) + code, out = run_wolfssl_pty("req", "-new", "-newkey", "rsa:2048", + "-keyout", key, "-config", self.conf_file, + "-x509", "-out", tmp, reply=b"\x04") + self.assertIn(b"Input Password", out) + self.assertIn(b"Unable to get password from stdin", out) + self.assertGreater(code, 0, out) + self.assertNotIn(b"AddressSanitizer", out) + if os.path.exists(key): + with open(key, "rb") as f: + self.assertNotIn(b"PRIVATE KEY", f.read()) + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") class TestReqKeyoutPermissions(unittest.TestCase): From 8f799e73076b9b5acb4aa5a4681801ea0107522c Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 13:02:47 +0000 Subject: [PATCH 11/21] Mark padding in legacy encrypt header for explicit key/IV With -key/-inkey and -iv, the legacy (Camellia) encrypt path wrote an all-zero salt, so decrypt left the block padding in the output. Set salt[0] to the pad count so decrypt strips it. Old files decrypt as before. F-11082 --- src/crypto/clu_encrypt.c | 5 +++++ tests/encrypt/enc-test.py | 37 +++++++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+) diff --git a/src/crypto/clu_encrypt.c b/src/crypto/clu_encrypt.c index 750a6ca8..1bd516c7 100644 --- a/src/crypto/clu_encrypt.c +++ b/src/crypto/clu_encrypt.c @@ -144,6 +144,11 @@ int wolfCLU_encrypt(int alg, char* mode, byte* pwdKey, byte* key, int size, key[i] = pwdKey[i]; } } + else { + /* explicit key: salt is unused, but a non-zero salt[0] tells + * decrypt to strip the padding */ + salt[0] = (byte)padCounter; + } /* open the outFile in write mode */ outFile = XFOPEN(out, "wb"); diff --git a/tests/encrypt/enc-test.py b/tests/encrypt/enc-test.py index 3242b8f6..63c94060 100644 --- a/tests/encrypt/enc-test.py +++ b/tests/encrypt/enc-test.py @@ -975,6 +975,43 @@ def test_rand_hex_to_inkey_workflow(self): self.assertTrue(filecmp.cmp(self._orig(), dec, shallow=False), "rand-hex -> -inkey workflow did not round-trip") + def test_legacy_cipher_key_iv_roundtrip(self): + """The non-EVP cipher path with -key/-iv must strip its padding. + + Camellia uses wolfCLU_encrypt/wolfCLU_decrypt, which pad the input + to a whole block and record in the salt header whether they did. + Sizes 27 and 1025 need padding; 32 and 2048 are block aligned. + """ + if not _camellia_available(): + self.skipTest("Camellia not compiled in") + + for size in (27, 32, 1025, 2048): + orig = "legacy_key_iv_{}.bin".format(size) + enc = "legacy_key_iv_{}.enc".format(size) + dec = "legacy_key_iv_{}.dec".format(size) + self._cleanup(orig, enc, dec) + data = bytes(i % 251 for i in range(size)) + with open(orig, "wb") as f: + f.write(data) + + with self.subTest(size=size): + r = run_wolfssl("-encrypt", "camellia-cbc-256", + "-in", orig, "-out", enc, + "-key", self.KEY_HEX, "-iv", self.IV_HEX) + self.assertEqual(r.returncode, 0, r.stderr) + + r = run_wolfssl("-decrypt", "camellia-cbc-256", + "-in", enc, "-out", dec, + "-key", self.KEY_HEX, "-iv", self.IV_HEX) + self.assertEqual(r.returncode, 0, r.stderr) + + with open(dec, "rb") as f: + got = f.read() + self.assertEqual(len(got), size, + "decrypted length differs from original") + self.assertEqual(got, data, + "decrypted data differs from original") + @unittest.skipUnless(HAVE_PTY, "pty not available (non-POSIX)") class EncStdinPasswordTest(unittest.TestCase): From 83767e79eb3e69bfd1c7568290575659bbde5aa1 Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 12:29:30 +0000 Subject: [PATCH 12/21] x509: only force version 3 when signing a CSR wolfCLU_certSetup set v3 on every parsed cert before checking -req, so -text on a v1 cert reported "Version: 3". Check reqFlag first. certs/server-ecc-v1-cert.pem made with OpenSSL 3.0.13: openssl x509 -req (no extensions) from an ecc-key.pem CSR, signed by ca-ecc-cert.pem. F-12924 --- certs/server-ecc-v1-cert.pem | 13 ++++++++ src/x509/clu_cert_setup.c | 8 ++--- tests/x509/x509-process-test.py | 57 +++++++++++++++++++++++++++++++++ tests/x509/x509-req-test.py | 14 ++++++++ 4 files changed, 88 insertions(+), 4 deletions(-) create mode 100644 certs/server-ecc-v1-cert.pem diff --git a/certs/server-ecc-v1-cert.pem b/certs/server-ecc-v1-cert.pem new file mode 100644 index 00000000..66a02109 --- /dev/null +++ b/certs/server-ecc-v1-cert.pem @@ -0,0 +1,13 @@ +-----BEGIN CERTIFICATE----- +MIIB8DCCAZcCAWUwCgYIKoZIzj0EAwIwgZcxCzAJBgNVBAYTAlVTMRMwEQYDVQQI +DApXYXNoaW5ndG9uMRAwDgYDVQQHDAdTZWF0dGxlMRAwDgYDVQQKDAd3b2xmU1NM +MRQwEgYDVQQLDAtEZXZlbG9wbWVudDEYMBYGA1UEAwwPd3d3LndvbGZzc2wuY29t +MR8wHQYJKoZIhvcNAQkBFhBpbmZvQHdvbGZzc2wuY29tMB4XDTI2MDkyMzEyMTc1 +MFoXDTI5MDYxOTEyMTc1MFowcTELMAkGA1UEBhMCVVMxEDAOBgNVBAgMB01vbnRh +bmExEDAOBgNVBAcMB0JvemVtYW4xEDAOBgNVBAoMB3dvbGZTU0wxEjAQBgNVBAsM +CVZlcnNpb24gMTEYMBYGA1UEAwwPd3d3LndvbGZzc2wuY29tMFkwEwYHKoZIzj0C +AQYIKoZIzj0DAQcDQgAEuzOsTCdQSsZKpQTDPN6fNttyLc6U6iv6yyAJOSwW6GEC +6a9N0wKTmjFbl5Ihf/DPGNqREQI0huggWDMLgDSJ2DAKBggqhkjOPQQDAgNHADBE +AiBU/5sbyAc2PPZmKswh5G7wy8vONxP1Jm2Is99fyJ1eWwIgHYNyWJIW8ehRk51A +a2aKxD5x4CAF/vh3A6KR+5w+ktw= +-----END CERTIFICATE----- diff --git a/src/x509/clu_cert_setup.c b/src/x509/clu_cert_setup.c index 1f05998b..259b2ddd 100644 --- a/src/x509/clu_cert_setup.c +++ b/src/x509/clu_cert_setup.c @@ -455,10 +455,10 @@ int wolfCLU_certSetup(int argc, char **argv) wolfSSL_NCONF_free(conf); } - /*default to version 3 which supports extensions */ - if (ret == WOLFCLU_SUCCESS && - wolfSSL_X509_set_version(x509, WOLFSSL_X509_V3) != WOLFSSL_SUCCESS && - reqFlag) { + /* A certificate made from a CSR takes the CSR's version (v1), which + * cannot carry extensions. Issue it as v3. */ + if (ret == WOLFCLU_SUCCESS && reqFlag && + wolfSSL_X509_set_version(x509, WOLFSSL_X509_V3) != WOLFSSL_SUCCESS) { wolfCLU_LogError("Unable to set version 3 for cert"); ret = WOLFCLU_FATAL_ERROR; } diff --git a/tests/x509/x509-process-test.py b/tests/x509/x509-process-test.py index 3e63ed86..436e6ed8 100644 --- a/tests/x509/x509-process-test.py +++ b/tests/x509/x509-process-test.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 """Tests for wolfssl x509 processing (converted from x509-process-test.sh).""" +import base64 import os import shutil import subprocess @@ -518,6 +519,62 @@ def test_5a_malformed_subj_argument(self): self.assertGreater(len(r.stderr), 0) +class TestX509V1Cert(unittest.TestCase): + """Regression: inspecting a v1 certificate must not change its version.""" + + V1_CERT = os.path.join(CERTS_DIR, "server-ecc-v1-cert.pem") + + def _clean(self, *files): + for f in files: + self.addCleanup(lambda p=f: _cleanup(p)) + + def _v1_der(self): + with open(self.V1_CERT) as f: + lines = f.read().splitlines() + b64 = "".join(line for line in lines + if line and not line.startswith("-----")) + return base64.b64decode(b64) + + def _assert_v1_text(self, r): + self.assertEqual(r.returncode, 0, r.stderr) + self.assertIn("Version: 1 (0x0)", r.stdout) + self.assertNotIn("Version: 3", r.stdout) + + def test_v1_pem_text(self): + r = run_wolfssl("x509", "-in", self.V1_CERT, "-text", "-noout") + self._assert_v1_text(r) + + def test_v1_der_text(self): + der = "test_v1_text_in.der" + self._clean(der) + with open(der, "wb") as f: + f.write(self._v1_der()) + r = run_wolfssl("x509", "-inform", "der", "-in", der, + "-text", "-noout") + self._assert_v1_text(r) + + def test_v1_pem_to_der_unchanged(self): + out = "test_v1_out.der" + self._clean(out) + r = run_wolfssl("x509", "-in", self.V1_CERT, "-outform", "der", + "-out", out) + self.assertEqual(r.returncode, 0, r.stderr) + with open(out, "rb") as f: + self.assertEqual(f.read(), self._v1_der()) + + def test_v1_der_to_pem_unchanged(self): + der = "test_v1_pem_in.der" + out = "test_v1_out.pem" + self._clean(der, out) + with open(der, "wb") as f: + f.write(self._v1_der()) + r = run_wolfssl("x509", "-inform", "der", "-in", der, + "-outform", "pem", "-out", out) + self.assertEqual(r.returncode, 0, r.stderr) + with open(out) as f1, open(self.V1_CERT) as f2: + self.assertEqual(f1.read(), f2.read()) + + class TestX509ModulusNoout(unittest.TestCase): """Regression: x509 -modulus -noout must not crash.""" diff --git a/tests/x509/x509-req-test.py b/tests/x509/x509-req-test.py index 7dd3a6f4..465d240c 100644 --- a/tests/x509/x509-req-test.py +++ b/tests/x509/x509-req-test.py @@ -669,6 +669,20 @@ def test_x509_req_signkey_succeeds(self): "-out", out) self.assertEqual(r.returncode, 0, r.stderr) + def test_x509_req_signkey_sets_v3(self): + """x509 -req issues a v3 certificate from the v1 CSR.""" + out = _tmp("tmp_x509req_v3.cert") + self._clean(out) + r = run_wolfssl("x509", "-req", "-in", self.csr, "-days", "3650", + "-signkey", + os.path.join(CERTS_DIR, "server-key.pem"), + "-out", out) + self.assertEqual(r.returncode, 0, r.stderr) + + r2 = run_wolfssl("x509", "-in", out, "-text", "-noout") + self.assertEqual(r2.returncode, 0, r2.stderr) + self.assertIn("Version: 3 (0x2)", r2.stdout) + class TestX509ReqHashAlgorithms(unittest.TestCase): """Test hash algorithm options for x509 -req.""" From 3d11ff79e50653b0c4355e747770b6e2a391ba0a Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 12:55:08 +0000 Subject: [PATCH 13/21] x509: fail -fingerprint when SHA-1 or hashing is unavailable wc_HashGetDigestSize's negative result was stored unsigned, and wc_Hash or missing-DER failures never set ret, so -fingerprint exited 0 with no output. Keep the digest size signed, reject <= 0, and return WOLFCLU_FATAL_ERROR with a message on each failure. F-12122 --- src/x509/clu_cert_setup.c | 22 ++++++++++++++++++---- 1 file changed, 18 insertions(+), 4 deletions(-) diff --git a/src/x509/clu_cert_setup.c b/src/x509/clu_cert_setup.c index 259b2ddd..9564c03e 100644 --- a/src/x509/clu_cert_setup.c +++ b/src/x509/clu_cert_setup.c @@ -631,15 +631,29 @@ int wolfCLU_certSetup(int argc, char **argv) int derSz; const unsigned char *der; byte digest[WC_MAX_DIGEST_SIZE]; - word32 digestSz = WC_MAX_DIGEST_SIZE; + int digestSz; enum wc_HashType digestType = WC_HASH_TYPE_SHA; der = wolfSSL_X509_get_der(x509, &derSz); - if (der != NULL) { + if (der == NULL || derSz <= 0) { + wolfCLU_LogError("Unable to get certificate DER for fingerprint"); + ret = WOLFCLU_FATAL_ERROR; + } + else { digestSz = wc_HashGetDigestSize(digestType); - if (wc_Hash(digestType, der, derSz, digest, digestSz) == 0) { + if (digestSz <= 0 || digestSz > WC_MAX_DIGEST_SIZE) { + wolfCLU_LogError("SHA-1 not available, unable to print " + "fingerprint"); + ret = WOLFCLU_FATAL_ERROR; + } + else if (wc_Hash(digestType, der, (word32)derSz, digest, + (word32)digestSz) != 0) { + wolfCLU_LogError("Unable to hash certificate for fingerprint"); + ret = WOLFCLU_FATAL_ERROR; + } + else { char txt[MAX_TERM_WIDTH]; - word32 i; + int i; XSNPRINTF(txt, MAX_TERM_WIDTH, "SHA1 of cert. DER : "); if (wolfSSL_BIO_write(out, txt, (int)XSTRLEN(txt)) <= 0) { From cf023bf17758b7f12eacac89a3ed5160bb376f6d Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 12:29:22 +0000 Subject: [PATCH 14/21] Reset -subj entry encoding after countryName wolfCLU_ParseX509NameString kept CTC_PRINTABLE after a C= entry, so later entries (e.g. a CN with '_' or '@') were encoded as PrintableString. Choose the encoding per entry: C is PrintableString, others UTF8String. Declare the per-entry variables inside the loop so no value can carry over from a previous entry. F-9842 --- src/tools/clu_funcs.c | 14 +++++------ tests/x509/x509-req-test.py | 50 +++++++++++++++++++++++++++++++++++++ 2 files changed, 57 insertions(+), 7 deletions(-) diff --git a/src/tools/clu_funcs.c b/src/tools/clu_funcs.c index a18675ce..bf187543 100644 --- a/src/tools/clu_funcs.c +++ b/src/tools/clu_funcs.c @@ -898,14 +898,9 @@ void wolfCLU_AddNameEntry(WOLFSSL_X509_NAME* name, int type, int nid, char* str) * returns a newly created WOLFSSL_X509_NAME on success */ WOLFSSL_X509_NAME* wolfCLU_ParseX509NameString(const char* n, int nSz) { - int encoding = CTC_UTF8; - int tagSz = 0; - int nid; char* word, *end; char* deli = (char*)"/"; - char* entry = NULL; WOLFSSL_X509_NAME* ret = NULL; - char tag[5]; if (n == NULL || nSz <= 0) { wolfCLU_LogError("unexpected null argument or size with parsing " @@ -920,6 +915,9 @@ WOLFSSL_X509_NAME* wolfCLU_ParseX509NameString(const char* n, int nSz) } for (word = strtok_r((char*)n, deli, &end); word != NULL; word = strtok_r(NULL, deli, &end)) { + int tagSz; + char tag[5]; + tagSz = (int)strcspn(word, "="); if (tagSz <= 0 || word[tagSz] != '=') { wolfCLU_LogError("error finding '=' char in name"); @@ -946,8 +944,10 @@ WOLFSSL_X509_NAME* wolfCLU_ParseX509NameString(const char* n, int nSz) } if (ret != NULL) { - entry = &word[tagSz+1]; - nid = wolfSSL_OBJ_sn2nid(tag); + char* entry = &word[tagSz+1]; + int encoding = CTC_UTF8; + int nid = wolfSSL_OBJ_sn2nid(tag); + if (nid == 0) { /* try using old tag value */ char oldTag[8]; tagSz = (int)XSTRLEN(tag); diff --git a/tests/x509/x509-req-test.py b/tests/x509/x509-req-test.py index 465d240c..3ab71f7f 100644 --- a/tests/x509/x509-req-test.py +++ b/tests/x509/x509-req-test.py @@ -147,6 +147,38 @@ def _flip_last_der_byte(src, dst): f.write(data) +def _der_tlv(data, pos): + """Return (tag, value_start, value_end) of the DER element at pos.""" + tag = data[pos] + length = data[pos + 1] + pos += 2 + if length & 0x80: + n = length & 0x7F + length = int.from_bytes(data[pos:pos + n], "big") + pos += n + return tag, pos, pos + length + + +def _csr_subject_string_tags(der): + """Return [(attribute OID bytes, string tag), ...] for a DER CSR subject. + + CertificationRequest -> CertificationRequestInfo -> version, subject. + Name is a SEQUENCE of SET of SEQUENCE { OID, string }.""" + _, pos, _ = _der_tlv(der, 0) + _, pos, _ = _der_tlv(der, pos) + _, _, pos = _der_tlv(der, pos) + _, pos, end = _der_tlv(der, pos) + out = [] + while pos < end: + _, set_start, set_end = _der_tlv(der, pos) + _, atv, _ = _der_tlv(der, set_start) + _, oid_start, oid_end = _der_tlv(der, atv) + val_tag, _, _ = _der_tlv(der, oid_end) + out.append((bytes(der[oid_start:oid_end]), val_tag)) + pos = set_end + return out + + class TestReqNew(unittest.TestCase): """Test req -new with various options.""" @@ -190,6 +222,24 @@ def test_req_new_with_subj(self): self.assertEqual(subject_line, expected, "Got: {!r}".format(subject_line)) + def test_req_new_subj_country_encoding_not_reused(self): + """-subj encodes C as PrintableString and the entries after it as + UTF8String. '_' and '@' are not valid PrintableString characters.""" + tmp = _tmp("test_req_subj_encoding.csr") + self._clean(tmp) + r = run_wolfssl("req", "-new", + "-key", os.path.join(CERTS_DIR, "server-key.pem"), + "-subj", "/C=US/CN=user_name@example/O=wolfSSL", + "-outform", "der", "-out", tmp) + self.assertEqual(r.returncode, 0, r.stderr) + + with open(tmp, "rb") as f: + tags = _csr_subject_string_tags(f.read()) + printable, utf8 = 0x13, 0x0C + self.assertEqual(tags, [(b"\x55\x04\x06", printable), + (b"\x55\x04\x03", utf8), + (b"\x55\x04\x0a", utf8)]) + def test_req_new_interactive_name(self): """req -new with no -subj/-config reads the name fields from stdin. From cc95ce37cac384790cbc4ddf6879c91441b27675 Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 12:56:35 +0000 Subject: [PATCH 15/21] Parse basicConstraints as comma separated NAME:VALUE pairs "CA:TRUE, pathlen:0" was split on ':' only, so CA stayed FALSE and pathlen was dropped without an error. Split on ',' then at the first ':', trim whitespace, set pathlen the way wolfSSL_X509_add_ext reads it, and fail on unknown or malformed entries instead of emitting a wrong extension. F-11080 --- src/x509/clu_config.c | 177 +++++++++++++++++++++++++++--------- tests/x509/x509-req-test.py | 128 ++++++++++++++++++++++++++ 2 files changed, 261 insertions(+), 44 deletions(-) diff --git a/src/x509/clu_config.c b/src/x509/clu_config.c index 780e56db..0ae91652 100644 --- a/src/x509/clu_config.c +++ b/src/x509/clu_config.c @@ -76,72 +76,154 @@ WOLFSSL_ASN1_OBJECT* wolfCLU_extenstionGetObjectNID(WOLFSSL_X509_EXTENSION *ext, return obj; } -static WOLFSSL_X509_EXTENSION* wolfCLU_parseBasicConstraint(char* in, int crit) +/* skip leading and cut trailing white space, returns the new start */ +static char* wolfCLU_trimSpace(char* str) { - int idx = 0; /* offset into string */ - char* word, *end, *str = in; - char* deli = (char*)":"; - WOLFSSL_X509_EXTENSION *ext; - WOLFSSL_ASN1_OBJECT *obj; + size_t len; - if (str == NULL) { - return NULL; + while (*str == ' ' || *str == '\t' || *str == '\r' || *str == '\n') { + str++; + } + len = XSTRLEN(str); + while (len > 0 && (str[len - 1] == ' ' || str[len - 1] == '\t' || + str[len - 1] == '\r' || str[len - 1] == '\n')) { + str[--len] = '\0'; } + return str; +} - /* if critical key word was found, then advance string pointer past - * 'critical,' */ - if (crit) { - int inSz = (int)XSTRLEN(in); - for (idx = 0; idx < inSz; idx++) { - if (str[idx] == ',') break; - } +/* largest pathlen wolfSSL will encode */ +#ifdef WOLFSSL_MAX_PATH_LEN + #define WOLFCLU_MAX_PATH_LEN WOLFSSL_MAX_PATH_LEN +#else + #define WOLFCLU_MAX_PATH_LEN 127 +#endif - if (idx + 1 >= inSz) { - WOLFCLU_LOG(WOLFCLU_E0, "bad basic constraint string in conf file"); - return NULL; - } +/* parse a decimal pathlen from 0 to WOLFCLU_MAX_PATH_LEN + * return WOLFCLU_SUCCESS on success */ +static int wolfCLU_parsePathLen(const char* str, int* pathLen) +{ + int val = 0; - /* advance past any white spaces */ - for (idx = idx + 1; idx < inSz; idx++) { - if (str[idx] != ' ') break; + if (*str == '\0') { + return WOLFCLU_FATAL_ERROR; + } + for (; *str != '\0'; str++) { + if (*str < '0' || *str > '9') { + return WOLFCLU_FATAL_ERROR; } + val = val * 10 + (*str - '0'); + if (val > WOLFCLU_MAX_PATH_LEN) { + return WOLFCLU_FATAL_ERROR; + } + } + *pathLen = val; + return WOLFCLU_SUCCESS; +} + + +/* Parse an OpenSSL style basicConstraints value: comma separated NAME:VALUE + * pairs with an optional leading "critical", e.g. + * "critical, CA:TRUE, pathlen:0". Returns NULL on a bad entry. */ +static WOLFSSL_X509_EXTENSION* wolfCLU_parseBasicConstraint(char* in, int crit) +{ + int ret = WOLFCLU_SUCCESS; + int first = 1; + int inSz, pathLen; + char *str, *cur, *next, *name, *value; + WOLFSSL_X509_EXTENSION *ext; + WOLFSSL_ASN1_OBJECT *obj; + + if (in == NULL) { + return NULL; } + /* tokenize a copy, the config value must stay intact */ + inSz = (int)XSTRLEN(in); + str = (char*)XMALLOC(inSz + 1, NULL, DYNAMIC_TYPE_TMP_BUFFER); + if (str == NULL) { + wolfCLU_LogError("out of memory parsing basicConstraints"); + return NULL; + } + XMEMCPY(str, in, inSz + 1); + ext = wolfSSL_X509_EXTENSION_new(); obj = wolfCLU_extenstionGetObjectNID(ext, NID_basic_constraints, crit); if (obj == NULL) { + wolfCLU_LogError("error creating basicConstraints extension"); + XFREE(str, NULL, DYNAMIC_TYPE_TMP_BUFFER); return NULL; } + for (cur = str; ret == WOLFCLU_SUCCESS && cur != NULL; cur = next) { + next = XSTRSTR(cur, ","); + if (next != NULL) { + *next++ = '\0'; + } - for (word = XSTRTOK(str + idx, deli, &end); word != NULL; - word = XSTRTOK(NULL, deli, &end)) { - if (word != NULL && XSTRCMP(word, "CA") == 0) { - word = XSTRTOK(NULL, deli, &end); - if (word != NULL) { - int z, wordSz; - - wordSz = (int)XSTRLEN(word); - for (z = 0; z < wordSz; z++) - word[z] = toupper(word[z]); - if (XSTRCMP(word, "TRUE") == 0) { - obj->ca = 1; - } - } + /* split at the first colon */ + value = XSTRSTR(cur, ":"); + if (value != NULL) { + *value++ = '\0'; + value = wolfCLU_trimSpace(value); } + name = wolfCLU_trimSpace(cur); - if (word != NULL && XSTRCMP(word, "pathlen") == 0) { - word = XSTRTOK(NULL, deli, &end); - if (word != NULL) { - if (obj->pathlen != NULL) - wolfSSL_ASN1_INTEGER_free(obj->pathlen); - obj->pathlen = wolfSSL_ASN1_INTEGER_new(); - wolfSSL_ASN1_INTEGER_set(obj->pathlen, XATOI(word)); + if (first && value == NULL && next != NULL && + XSTRCMP(name, "critical") == 0) { + /* crit is already set by the caller */ + } + else if (value != NULL && XSTRCMP(name, "CA") == 0) { + if (XSTRCASECMP(value, "TRUE") == 0 || + XSTRCASECMP(value, "YES") == 0 || + XSTRCASECMP(value, "Y") == 0) { + obj->ca = 1; + } + else if (XSTRCASECMP(value, "FALSE") == 0 || + XSTRCASECMP(value, "NO") == 0 || + XSTRCASECMP(value, "N") == 0) { + obj->ca = 0; + } + else { + wolfCLU_LogError("bad basicConstraints CA value \"%s\"", + value); + ret = WOLFCLU_FATAL_ERROR; } } + else if (value != NULL && XSTRCMP(name, "pathlen") == 0) { + if (wolfCLU_parsePathLen(value, &pathLen) != WOLFCLU_SUCCESS) { + wolfCLU_LogError("bad basicConstraints pathlen \"%s\", " + "expected 0 to %d", value, WOLFCLU_MAX_PATH_LEN); + ret = WOLFCLU_FATAL_ERROR; + } + else { + if (obj->pathlen == NULL) { + obj->pathlen = wolfSSL_ASN1_INTEGER_new(); + } + if (obj->pathlen == NULL) { + wolfCLU_LogError("out of memory parsing basicConstraints"); + ret = WOLFCLU_FATAL_ERROR; + } + else { + /* wolfSSL_X509_add_ext() reads the path length from the + * length field */ + obj->pathlen->length = pathLen; + } + } + } + else { + wolfCLU_LogError("bad basicConstraints entry \"%s\"", name); + ret = WOLFCLU_FATAL_ERROR; + } + first = 0; } + XFREE(str, NULL, DYNAMIC_TYPE_TMP_BUFFER); + if (ret != WOLFCLU_SUCCESS) { + wolfSSL_X509_EXTENSION_free(ext); + ext = NULL; + } return ext; } @@ -303,6 +385,9 @@ static int wolfCLU_parseExtension(WOLFSSL_X509* x509, char* str, int nid, switch (nid) { case NID_basic_constraints: ext = wolfCLU_parseBasicConstraint(str, crit); + if (ext == NULL) { + return WOLFCLU_FATAL_ERROR; + } break; case NID_subject_key_identifier: ext = wolfCLU_parseSubjectKeyID(str, crit, x509); @@ -626,7 +711,11 @@ int wolfCLU_setExtensions(WOLFSSL_X509* x509, WOLFSSL_CONF* conf, char* sect) current = wolfSSL_NCONF_get_string(conf, sect, "basicConstraints"); if (current != NULL) { - wolfCLU_parseExtension(x509, current, NID_basic_constraints, &idx); + ret = wolfCLU_parseExtension(x509, current, NID_basic_constraints, + &idx); + if (ret != WOLFCLU_SUCCESS) { + return ret; + } } current = wolfSSL_NCONF_get_string(conf, sect, "subjectKeyIdentifier"); diff --git a/tests/x509/x509-req-test.py b/tests/x509/x509-req-test.py index 3ab71f7f..7f3fbd43 100644 --- a/tests/x509/x509-req-test.py +++ b/tests/x509/x509-req-test.py @@ -50,6 +50,9 @@ def _tmp(name): basicConstraints = CA:TRUE keyUsage = digitalSignature subjectAltName = @alt_names_full_skip +[ v3_ca_pathlen ] +basicConstraints = critical, CA:TRUE, pathlen:1 +keyUsage = keyCertSign, cRLSign [alt_names] DNS.1 = extraName DNS.2 = alt-name @@ -179,6 +182,43 @@ def _csr_subject_string_tags(der): return out +def _cert_basic_constraints(der): + """Return (critical, cA, pathLen) of a DER certificate's + basicConstraints extension, or None if it has none. pathLen is None + when absent. + + Certificate -> TBSCertificate -> [3] Extensions. Each Extension is + SEQUENCE { OID, critical BOOLEAN OPTIONAL, OCTET STRING } and the + OCTET STRING holds SEQUENCE { cA BOOLEAN OPTIONAL, INTEGER OPTIONAL }.""" + _, pos, _ = _der_tlv(der, 0) + _, pos, end = _der_tlv(der, pos) + while pos < end: + tag, start, pos = _der_tlv(der, pos) + if tag != 0xA3: + continue + _, ext_pos, exts_end = _der_tlv(der, start) + while ext_pos < exts_end: + _, field, ext_pos = _der_tlv(der, ext_pos) + _, oid_start, field = _der_tlv(der, field) + if bytes(der[oid_start:field]) != b"\x55\x1d\x13": + continue + critical = False + tag, val_start, val_end = _der_tlv(der, field) + if tag == 0x01: + critical = der[val_start] != 0 + _, val_start, val_end = _der_tlv(der, val_end) + _, bc_pos, bc_end = _der_tlv(der, val_start) + ca, pathlen = False, None + while bc_pos < bc_end: + tag, val_start, bc_pos = _der_tlv(der, bc_pos) + if tag == 0x01: + ca = der[val_start] != 0 + elif tag == 0x02: + pathlen = int.from_bytes(der[val_start:bc_pos], "big") + return critical, ca, pathlen + return None + + class TestReqNew(unittest.TestCase): """Test req -new with various options.""" @@ -448,6 +488,74 @@ def test_req_inline_subjectaltname_trims_whitespace(self): self.assertIn("IP Address:10.0.0.1", san_line, "IP SAN not applied/trimmed from inline config form") + def _req_basic_constraints(self, value, name): + """Run req -new -x509 with a config setting basicConstraints to value. + Return the result and the path of the DER certificate.""" + conf = _tmp(name + ".conf") + out = _tmp(name + ".der") + self._clean(conf, out) + with open(conf, "w", encoding="utf-8", newline="\n") as f: + f.write( + "[ req ]\n" + "distinguished_name = dn\n" + "prompt = no\n" + "x509_extensions = v3_bc\n" + "[ dn ]\n" + "commonName = basic-constraints-test\n" + "[ v3_bc ]\n" + "basicConstraints = " + value + "\n") + r = run_wolfssl("req", "-new", "-x509", + "-key", os.path.join(CERTS_DIR, "server-key.pem"), + "-config", conf, "-outform", "der", "-out", out) + if "not compiled with cert extensions" in r.stdout + r.stderr: + self.skipTest("cert extensions not compiled in") + return r, out + + def test_req_config_basic_constraints(self): + """A config basicConstraints value is a comma separated list of + NAME:VALUE pairs with an optional leading "critical", as in OpenSSL. + CA and pathlen must both reach the certificate.""" + cases = [ + ("CA:TRUE, pathlen:0", (False, True, 0)), + ("CA:TRUE,pathlen:3", (False, True, 3)), + ("critical, CA:TRUE, pathlen:0", (True, True, 0)), + ("critical,CA:TRUE", (True, True, None)), + ("CA:FALSE", (False, False, None)), + ("pathlen:2 , CA : true", (False, True, 2)), + ] + for i, (value, expected) in enumerate(cases): + with self.subTest(value=value): + r, out = self._req_basic_constraints( + value, "test_req_bc_{}".format(i)) + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + with open(out, "rb") as f: + got = _cert_basic_constraints(f.read()) + self.assertEqual(got, expected, + "(critical, CA, pathlen) mismatch") + + def test_req_config_basic_constraints_bad_fails(self): + """Unknown or malformed basicConstraints entries fail instead of + producing a certificate with the wrong constraints.""" + values = [ + "CA:TRUE, bogus:1", + "CA:maybe", + "CA", + "CA:TRUE,", + "CA:TRUE, pathlen:abc", + "CA:TRUE, pathlen:-1", + "CA:TRUE, pathlen:1000", + "CA:TRUE, critical", + "critical", + ] + for i, value in enumerate(values): + with self.subTest(value=value): + r, out = self._req_basic_constraints( + value, "test_req_bc_bad_{}".format(i)) + self.assertNotEqual(r.returncode, 0, + "bad basicConstraints accepted") + self.assertFalse(os.path.exists(out), + "certificate written for bad input") + def test_req_x509_addext_subject_alt_name(self): """req -x509 -addext subjectAltName adds IP and DNS alt names.""" crt = _tmp("test_req_addext.crt") @@ -846,6 +954,26 @@ def test_extfile_v3_alt_ca(self): self.assertEqual(r2.returncode, 0, r2.stderr) self.assertIn("CA:TRUE", r2.stdout) + def test_extfile_basic_constraints_pathlen(self): + """x509 -req -extfile applies "critical, CA:TRUE, pathlen:1". + wolfSSL drops pathlen unless keyUsage has keyCertSign, so the + section also replaces the CSR's keyUsage.""" + out = _tmp("tmp_ext_pathlen.der") + self._clean(out) + r = run_wolfssl("x509", "-req", "-in", self.csr, "-days", "3650", + "-extfile", self.conf_file, + "-extensions", "v3_ca_pathlen", + "-signkey", + os.path.join(CERTS_DIR, "server-key.pem"), + "-outform", "der", "-out", out) + if "not compiled with cert extensions" in r.stdout + r.stderr: + self.skipTest("cert extensions not compiled in") + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + with open(out, "rb") as f: + got = _cert_basic_constraints(f.read()) + self.assertEqual(got, (True, True, 1), + "(critical, CA, pathlen) mismatch") + class TestX509ReqLargeExtensions(unittest.TestCase): """Test x509 -req when the added extensions outgrow the input PEM. From d1df4a0a83b4b553f1cd0c58918a6be6c4efc609 Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 13:29:39 +0000 Subject: [PATCH 16/21] Write PKCS#8 DER for pkcs8 -topk8 -outform DER The DER branch always used the traditional encoder, so -topk8 -nocrypt -outform DER wrote RSAPrivateKey/ECPrivateKey. For -topk8, re-encode the key and wrap it with wolfSSL_i2d_PKCS8_PKEY, so PKCS#8 DER input is not double wrapped. Zero the key buffers before freeing. F-9841 --- src/pkcs/clu_pkcs8.c | 25 ++++++- tests/pkcs/pkcs8-test.py | 139 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 163 insertions(+), 1 deletion(-) diff --git a/src/pkcs/clu_pkcs8.c b/src/pkcs/clu_pkcs8.c index 699e8a49..21b3fb19 100644 --- a/src/pkcs/clu_pkcs8.c +++ b/src/pkcs/clu_pkcs8.c @@ -261,7 +261,27 @@ int wolfCLU_PKCS8(int argc, char** argv) unsigned char *der = NULL; int derSz = 0; - if ((derSz = wolfCLU_pKeytoPriKey(pkey, &der)) <= 0) { + derSz = wolfCLU_pKeytoPriKey(pkey, &der); + + /* -topk8 wraps the key in a PKCS#8 PrivateKeyInfo. Wrap the + * re-encoded key, since the cached input may already be PKCS#8. */ + if (derSz > 0 && toPkcs8 == 1 && traditional == 0) { + const unsigned char *p = der; + WOLFSSL_EVP_PKEY *tradKey; + + tradKey = wolfSSL_d2i_PrivateKey(wolfSSL_EVP_PKEY_id(pkey), + NULL, &p, derSz); + wolfCLU_ForceZero(der, (unsigned int)derSz); + XFREE(der, HEAP_HINT, DYNAMIC_TYPE_OPENSSL); + der = NULL; + derSz = 0; + if (tradKey != NULL) { + derSz = wolfSSL_i2d_PKCS8_PKEY(tradKey, &der); + wolfSSL_EVP_PKEY_free(tradKey); + } + } + + if (derSz <= 0) { WOLFCLU_LOG(WOLFCLU_E0, "Error converting private key to der"); ret = WOLFCLU_FATAL_ERROR; @@ -273,6 +293,9 @@ int wolfCLU_PKCS8(int argc, char** argv) } if (der != NULL) { + if (derSz > 0) { + wolfCLU_ForceZero(der, (unsigned int)derSz); + } XFREE(der, HEAP_HINT, DYNAMIC_TYPE_OPENSSL); } } diff --git a/tests/pkcs/pkcs8-test.py b/tests/pkcs/pkcs8-test.py index 6e84614d..71f1978b 100644 --- a/tests/pkcs/pkcs8-test.py +++ b/tests/pkcs/pkcs8-test.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 """PKCS8 tests for wolfCLU.""" +import base64 import filecmp import os import subprocess @@ -11,6 +12,37 @@ from wolfclu_test import (WOLFSSL_BIN, CERTS_DIR, HAVE_PTY, is_fips, run_wolfssl, run_wolfssl_pty, test_main) +RSA_OID = bytes.fromhex("2a864886f70d010101") # rsaEncryption +EC_OID = bytes.fromhex("2a8648ce3d0201") # id-ecPublicKey +P256_OID = bytes.fromhex("2a8648ce3d030107") # prime256v1 + + +def der_tlv(data, pos=0): + """Return (tag, value, end) of the DER element at pos.""" + tag = data[pos] + length = data[pos + 1] + pos += 2 + if length & 0x80: + n = length & 0x7F + length = int.from_bytes(data[pos:pos + n], "big") + pos += n + return tag, data[pos:pos + length], pos + length + + +def der_enc(tag, value): + """Return the DER element tag, length, value.""" + n = len(value) + if n < 0x80: + return bytes([tag, n]) + value + raw = n.to_bytes((n.bit_length() + 7) // 8, "big") + return bytes([tag, 0x80 | len(raw)]) + raw + value + + +def pkcs8_der(key, oid, params): + """Wrap a traditional private key in a PKCS#8 PrivateKeyInfo.""" + alg = der_enc(0x30, der_enc(0x06, oid) + params) + return der_enc(0x30, der_enc(0x02, b"\x00") + alg + der_enc(0x04, key)) + class Pkcs8Test(unittest.TestCase): @@ -73,6 +105,113 @@ def test_decrypt_and_convert(self): pkcs1_pem, shallow=False), "server-key.pem -traditional check failed") + def _read(self, path): + with open(path, "rb") as f: + return f.read() + + def _assert_pkcs8_der(self, der, oid, params): + """Check der is a PKCS#8 PrivateKeyInfo and return its privateKey.""" + tag, body, end = der_tlv(der) + self.assertEqual((tag, end), (0x30, len(der))) + tag, version, pos = der_tlv(body) + self.assertEqual((tag, version), (0x02, b"\x00")) + tag, alg, pos = der_tlv(body, pos) + self.assertEqual(tag, 0x30, "no AlgorithmIdentifier, not PKCS#8") + tag, alg_oid, alg_end = der_tlv(alg) + self.assertEqual((tag, alg_oid), (0x06, oid)) + self.assertEqual(alg[alg_end:], params) + tag, key, pos = der_tlv(body, pos) + self.assertEqual(tag, 0x04) + return key + + def test_topk8_nocrypt_der_is_pkcs8(self): + """-topk8 -nocrypt -outform DER writes PKCS#8 (F-9841).""" + rsa_pem = os.path.join(CERTS_DIR, "server-key.pem") + rsa_der = os.path.join(CERTS_DIR, "server-key.der") + enc_pem = os.path.join(CERTS_DIR, "server-keyEnc.pem") + ecc_pem = os.path.join(CERTS_DIR, "ecc-key.pem") + rsa_alg = (RSA_OID, b"\x05\x00") + ecc_alg = (EC_OID, b"\x06\x08" + P256_OID) + + p8_pem = "topk8-in-pkcs8.pem" + self._cleanup(p8_pem) + r = run_wolfssl("pkcs8", "-in", rsa_pem, "-out", p8_pem) + self.assertEqual(r.returncode, 0, r.stderr) + + # PKCS#8 DER inputs: server-key.der wrapped here, and the body of the + # PKCS#8 PEM file ca-ecc-key.pem. + rsa_p8 = pkcs8_der(self._read(rsa_der), *rsa_alg) + with open(os.path.join(CERTS_DIR, "ca-ecc-key.pem"), "r") as f: + ecc_p8 = base64.b64decode("".join( + l for l in f.read().splitlines() if not l.startswith("-"))) + rsa_p8_der = "topk8-in-rsa-p8.der" + ecc_p8_der = "topk8-in-ecc-p8.der" + self._cleanup(rsa_p8_der, ecc_p8_der) + for path, data in ((rsa_p8_der, rsa_p8), (ecc_p8_der, ecc_p8)): + with open(path, "wb") as f: + f.write(data) + + cases = [ + ("topk8-rsa-pem.der", ["-in", rsa_pem], rsa_alg), + ("topk8-rsa-der.der", ["-in", rsa_der, "-inform", "DER"], rsa_alg), + ("topk8-rsa-p8.der", ["-in", p8_pem], rsa_alg), + ("topk8-rsa-p8der.der", ["-in", rsa_p8_der, "-inform", "DER"], + rsa_alg), + ("topk8-ecc-pem.der", ["-in", ecc_pem], ecc_alg), + ("topk8-ecc-p8der.der", ["-in", ecc_p8_der, "-inform", "DER"], + ecc_alg), + ] + if not self.is_fips: + cases.append(("topk8-rsa-enc.der", + ["-in", enc_pem, "-passin", "pass:yassl123"], + rsa_alg)) + for out, args, (oid, params) in cases: + with self.subTest(out=out): + trad = "trad-" + out + self._cleanup(out, trad) + r = run_wolfssl("pkcs8", *(args + [ + "-topk8", "-nocrypt", "-outform", "DER", "-out", out])) + self.assertEqual(r.returncode, 0, r.stderr) + r = run_wolfssl("pkcs8", *(args + [ + "-traditional", "-outform", "DER", "-out", trad])) + self.assertEqual(r.returncode, 0, r.stderr) + + key = self._assert_pkcs8_der(self._read(out), oid, params) + self.assertEqual(key, self._read(trad)) + # Every RSA case is server-key, so the output is the same + # PKCS#8 DER that openssl writes. + if oid == RSA_OID: + self.assertEqual(self._read(out), rsa_p8) + + # The PKCS#8 DER output reads back as the original key. + outs = ["topk8-rsa-pem.der"] + if not self.is_fips: + outs.append("topk8-rsa-enc.der") + for out in outs: + with self.subTest(read_back=out): + pkcs1_pem = "back-" + out + ".pem" + self._cleanup(pkcs1_pem) + r = run_wolfssl("pkcs8", "-in", out, "-inform", "DER", + "-traditional", "-out", pkcs1_pem) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(self._read(pkcs1_pem), self._read(rsa_pem)) + + def test_der_without_topk8_is_traditional(self): + """Without -topk8, DER output stays traditional, as in openssl.""" + rsa_pem = os.path.join(CERTS_DIR, "server-key.pem") + expected = self._read(os.path.join(CERTS_DIR, "server-key.der")) + cases = [ + ("notopk8-trad.der", ["-traditional"]), + ("notopk8-plain.der", []), + ] + for out, args in cases: + with self.subTest(out=out): + self._cleanup(out) + r = run_wolfssl("pkcs8", "-in", rsa_pem, *(args + [ + "-outform", "DER", "-out", out])) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(self._read(out), expected) + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") def test_out_file_owner_only(self): """-out holds a private key, so it must be created 0600 (F-9854).""" From 324e0c2ceae57cc2cf154cff06b73a09a8eab812 Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 13:18:10 +0000 Subject: [PATCH 17/21] Create pkey private key output with owner-only permissions pkey opened -out before knowing if a private key would be written, so under umask 022 the converted key file was world-readable. Open -out after option parsing and use wolfCLU_BioOpenOwner unless -pubout/-pubin. -out without a file name fails instead of falling back to stdout. F-9859 --- src/pkey/clu_pkey.c | 22 +++++++++++++++---- tests/pkey/pkey-test.py | 47 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 4 deletions(-) diff --git a/src/pkey/clu_pkey.c b/src/pkey/clu_pkey.c index f171d05c..8928522d 100644 --- a/src/pkey/clu_pkey.c +++ b/src/pkey/clu_pkey.c @@ -421,6 +421,7 @@ int wolfCLU_pKeySetup(int argc, char** argv) int pubOut = 0; int option; int longIndex = 1; + char *outFile = NULL; WOLFSSL_EVP_PKEY *pkey = NULL; WOLFSSL_BIO *bioIn = NULL; WOLFSSL_BIO *bioOut = NULL; @@ -453,10 +454,9 @@ int wolfCLU_pKeySetup(int argc, char** argv) break; case WOLFCLU_OUTFILE: - bioOut = wolfSSL_BIO_new_file(optarg, "wb"); - if (bioOut == NULL) { - wolfCLU_LogError("Unable to open output file %s", - optarg); + outFile = optarg; + if (outFile == NULL) { + wolfCLU_LogError("-out requires a file name"); ret = WOLFCLU_FATAL_ERROR; } break; @@ -509,6 +509,20 @@ int wolfCLU_pKeySetup(int argc, char** argv) } } + /* open -out once options are known. A private key file is owner-only */ + if (ret == WOLFCLU_SUCCESS && outFile != NULL) { + if (pubOut) { + bioOut = wolfSSL_BIO_new_file(outFile, "wb"); + } + else { + bioOut = wolfCLU_BioOpenOwner(outFile); + } + if (bioOut == NULL) { + wolfCLU_LogError("Unable to open output file %s", outFile); + ret = WOLFCLU_FATAL_ERROR; + } + } + if (ret == WOLFCLU_SUCCESS && bioOut == NULL) { bioOut = wolfSSL_BIO_new(wolfSSL_BIO_s_file()); if (bioOut == NULL) { diff --git a/tests/pkey/pkey-test.py b/tests/pkey/pkey-test.py index 91b528d1..d74c18ca 100644 --- a/tests/pkey/pkey-test.py +++ b/tests/pkey/pkey-test.py @@ -107,6 +107,53 @@ def test_out_to_file(self): with open(out, "r") as f: self.assertIn("BEGIN PUBLIC KEY", f.read()) + def _out_mode(self, out, key, *args): + """Write a new -out file under umask 022 and return its mode.""" + self._cleanup(out) + if os.path.exists(out): + os.remove(out) + old_umask = os.umask(0o022) + try: + r = run_wolfssl("pkey", "-in", os.path.join(CERTS_DIR, key), + "-out", out, *args) + finally: + os.umask(old_umask) + self.assertEqual(r.returncode, 0, r.stderr) + return os.stat(out).st_mode & 0o777 + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_private_out_mode(self): + """Private key output must be owner-only (F-9859).""" + for fmt in ("PEM", "DER"): + with self.subTest(outform=fmt): + mode = self._out_mode("test-pkey-perm-priv." + fmt.lower(), + "ecc-key.pem", "-outform", fmt) + self.assertEqual(mode, 0o600, + "private key mode is {:o}, expected 600" + .format(mode)) + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_public_out_mode(self): + """Public only output keeps the umask default mode.""" + cases = [ + ("test-pkey-perm-pubout.pem", "ecc-key.pem", "-pubout"), + ("test-pkey-perm-pubin.pem", "ecc-keyPub.pem", "-pubin"), + ] + for out, key, *args in cases: + with self.subTest(args=args): + mode = self._out_mode(out, key, *args) + self.assertEqual(mode, 0o644, + "public output mode is {:o}, expected 644" + .format(mode)) + + + def test_out_missing_file_name(self): + """A trailing -out must fail, not print the key to stdout.""" + r = run_wolfssl("pkey", "-in", + os.path.join(CERTS_DIR, "ecc-key.pem"), "-out") + self.assertNotEqual(r.returncode, 0) + self.assertNotIn("PRIVATE KEY", r.stdout) + if __name__ == "__main__": test_main() From 815bd896cf7e6a4176a4b6e43752cdec1c0ad519 Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 13:17:46 +0000 Subject: [PATCH 18/21] Create dhparam -genkey output with owner-only permissions dhparam -genkey wrote the DH private key through wolfSSL_BIO_new_file, so under umask 022 the file was world-readable. Open the output with wolfCLU_BioOpenOwner when -genkey is given. Params-only output is unchanged. F-9857 --- src/dh/clu_dh.c | 8 ++++- tests/dh/dh-test.py | 71 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 1 deletion(-) diff --git a/src/dh/clu_dh.c b/src/dh/clu_dh.c index 373ccb1d..60c7b358 100644 --- a/src/dh/clu_dh.c +++ b/src/dh/clu_dh.c @@ -534,7 +534,13 @@ int wolfCLU_DhParamSetup(int argc, char** argv) WOLFCLU_LOG(WOLFCLU_E0, "No filesystem support. Unable to open output file"); ret = WOLFCLU_FATAL_ERROR; #else - bioOut = wolfSSL_BIO_new_file(out, "wb"); + /* a generated key is private, so create the file owner-only */ + if (genKey) { + bioOut = wolfCLU_BioOpenOwner(out); + } + else { + bioOut = wolfSSL_BIO_new_file(out, "wb"); + } if (bioOut == NULL) { wolfCLU_LogError("Unable to open output file %s", optarg); diff --git a/tests/dh/dh-test.py b/tests/dh/dh-test.py index 71cb6575..ce7bbbd9 100644 --- a/tests/dh/dh-test.py +++ b/tests/dh/dh-test.py @@ -106,6 +106,77 @@ def test_bad_input_fails(self): "-genkey", "-noout") self.assertNotEqual(r.returncode, 0) + def _new_file(self, name): + # Start from no file so the create mode applies. + if os.path.exists(name): + os.remove(name) + self.addCleanup(lambda: os.remove(name) + if os.path.exists(name) else None) + return name + + def _mode(self, name): + return os.stat(name).st_mode & 0o777 + + def _use_umask_022(self): + old_umask = os.umask(0o022) + self.addCleanup(os.umask, old_umask) + + def _gen_params(self, params_file): + r = run_wolfssl("dhparam", "-out", self._new_file(params_file), + "1024") + self.assertEqual(r.returncode, 0, r.stderr) + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_dhparam_genkey_out_mode(self): + """-genkey output holds a private key so it must be owner-only.""" + params_file = "dh-perm-genkey.params" + key_file = self._new_file("dh-perm-genkey.key") + self._use_umask_022() + self._gen_params(params_file) + + r = run_wolfssl("dhparam", "-in", params_file, "-genkey", "-noout", + "-out", key_file) + self.assertEqual(r.returncode, 0, r.stderr) + with open(key_file, "r") as f: + self.assertIn("-----BEGIN PRIVATE KEY-----", f.read()) + mode = self._mode(key_file) + self.assertEqual(mode, 0o600, + "DH private key mode is {:o}, expected 600".format( + mode)) + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_dhparam_genkey_with_params_out_mode(self): + """Params and key in one file: the file must be owner-only.""" + params_file = "dh-perm-both.params" + key_file = self._new_file("dh-perm-both.key") + self._use_umask_022() + self._gen_params(params_file) + + r = run_wolfssl("dhparam", "-in", params_file, "-genkey", + "-out", key_file) + self.assertEqual(r.returncode, 0, r.stderr) + with open(key_file, "r") as f: + data = f.read() + self.assertIn("-----BEGIN DH PARAMETERS-----", data) + self.assertIn("-----BEGIN PRIVATE KEY-----", data) + mode = self._mode(key_file) + self.assertEqual(mode, 0o600, + "DH private key mode is {:o}, expected 600".format( + mode)) + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_dhparam_params_out_mode(self): + """Parameter-only output is public and keeps default permissions.""" + params_file = "dh-perm-params.params" + copy_file = self._new_file("dh-perm-params-copy.params") + self._use_umask_022() + self._gen_params(params_file) + self.assertEqual(self._mode(params_file), 0o644) + + r = run_wolfssl("dhparam", "-in", params_file, "-out", copy_file) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(self._mode(copy_file), 0o644) + if __name__ == "__main__": test_main() From e0131698872284a8cc0cbd78ebc06666b1495c97 Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 13:22:46 +0000 Subject: [PATCH 19/21] Create ecparam -genkey output with owner-only permissions ecparam -genkey wrote the EC private key through wolfSSL_BIO_new_file, so under umask 022 the file was world-readable. Open the output with wolfCLU_BioOpenOwner when -genkey is given. Params-only output is unchanged. F-9858 --- src/ecparam/clu_ecparam.c | 8 ++++- tests/pkey/ecparam-test.py | 63 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 1 deletion(-) diff --git a/src/ecparam/clu_ecparam.c b/src/ecparam/clu_ecparam.c index 387210a6..973bf6a0 100644 --- a/src/ecparam/clu_ecparam.c +++ b/src/ecparam/clu_ecparam.c @@ -221,7 +221,13 @@ int wolfCLU_ecparam(int argc, char** argv) WOLFCLU_LOG(WOLFCLU_E0, "No filesystem support. Unable to open input file"); ret = WOLFCLU_FATAL_ERROR; #else - bioOut = wolfSSL_BIO_new_file(out, "wb"); + /* a generated key is private, so create the file owner-only */ + if (genKey) { + bioOut = wolfCLU_BioOpenOwner(out); + } + else { + bioOut = wolfSSL_BIO_new_file(out, "wb"); + } if (bioOut == NULL) { ret = WOLFCLU_FATAL_ERROR; } diff --git a/tests/pkey/ecparam-test.py b/tests/pkey/ecparam-test.py index 0579df35..75d028bb 100644 --- a/tests/pkey/ecparam-test.py +++ b/tests/pkey/ecparam-test.py @@ -173,6 +173,69 @@ def test_all_curves_genkey(self): self.assertIn(name, text, f"curve name {name} not in text output") + def _new_file(self, name): + # Start from no file so the create mode applies. + if os.path.exists(name): + os.remove(name) + self._cleanup(name) + return name + + def _mode(self, name): + return os.stat(name).st_mode & 0o777 + + def _use_umask_022(self): + old_umask = os.umask(0o022) + self.addCleanup(os.umask, old_umask) + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_genkey_out_mode(self): + """-genkey output holds a private key so it must be owner-only.""" + key_file = self._new_file("ecparam-perm-genkey.key") + self._use_umask_022() + + r = run_wolfssl("ecparam", "-genkey", "-name", "secp384r1", + "-out", key_file) + self.assertEqual(r.returncode, 0, r.stderr) + with open(key_file, "r") as f: + data = f.read() + self.assertIn("-----BEGIN EC PARAMETERS-----", data) + self.assertIn("-----BEGIN EC PRIVATE KEY-----", data) + mode = self._mode(key_file) + self.assertEqual(mode, 0o600, + "EC private key mode is {:o}, expected 600".format( + mode)) + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_genkey_der_out_mode(self): + """DER -genkey output is also a private key and must be owner-only.""" + key_file = self._new_file("ecparam-perm-genkey.der") + self._use_umask_022() + + r = run_wolfssl("ecparam", "-genkey", "-name", "secp256r1", + "-outform", "der", "-out", key_file) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertGreater(os.path.getsize(key_file), 0) + mode = self._mode(key_file) + self.assertEqual(mode, 0o600, + "EC private key mode is {:o}, expected 600".format( + mode)) + + @unittest.skipIf(os.name == "nt", "POSIX file permissions only") + def test_params_out_mode(self): + """Parameter-only output is public and keeps default permissions.""" + params_file = self._new_file("ecparam-perm-params.pem") + self._use_umask_022() + + r = run_wolfssl("ecparam", "-in", + os.path.join(CERTS_DIR, "ecc-key.pem"), + "-out", params_file) + self.assertEqual(r.returncode, 0, r.stderr) + with open(params_file, "r") as f: + data = f.read() + self.assertIn("-----BEGIN EC PARAMETERS-----", data) + self.assertNotIn("PRIVATE KEY", data) + self.assertEqual(self._mode(params_file), 0o644) + if __name__ == "__main__": test_main() From 425bba5c032771144f3b4bf31655f43c35c6e740 Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 13:17:58 +0000 Subject: [PATCH 20/21] Verify req -x509 output as a certificate, not a CSR req -verify always called wolfSSL_X509_REQ_verify, so a certificate made with -x509 was parsed as a CSR and failed. Use wolfSSL_X509_verify for generated certs, and verify with the req/cert public key, since wolfSSL cannot verify with an RSA private key object. F-9846 --- src/x509/clu_request_setup.c | 26 +++++++------ tests/x509/x509-req-test.py | 73 ++++++++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 11 deletions(-) diff --git a/src/x509/clu_request_setup.c b/src/x509/clu_request_setup.c index 05cd1693..eb70f97a 100644 --- a/src/x509/clu_request_setup.c +++ b/src/x509/clu_request_setup.c @@ -605,9 +605,7 @@ int wolfCLU_requestSetup(int argc, char** argv) byte reSign = 0; /* flag for if resigning req is needed */ byte noOut = 0; byte useDes = 1; -#ifdef NO_WOLFSSL_REQ_PRINT byte isCSR = 1; -#endif /* Multiple -addext is not yet supported. Detect it up front and fail * instead of silently dropping the extension and exiting success. */ { @@ -1019,9 +1017,7 @@ int wolfCLU_requestSetup(int argc, char** argv) /* sign the req/cert */ if (ret == WOLFCLU_SUCCESS && (reqIn == NULL || reSign)) { if (genX509) { -#ifdef NO_WOLFSSL_REQ_PRINT isCSR = 0; -#endif /* default to version 3 which supports extensions */ if (wolfSSL_X509_set_version(x509, WOLFSSL_X509_V3) != WOLFSSL_SUCCESS) { @@ -1049,18 +1045,26 @@ int wolfCLU_requestSetup(int argc, char** argv) } if (ret == WOLFCLU_SUCCESS && doVerify) { + WOLFSSL_EVP_PKEY* pubKey; + int verifyRet; - /* get public key from req if not passed in */ - if (pkey == NULL) { - pkey = wolfSSL_X509_get_pubkey(x509); - } - - if (pkey == NULL) { + /* Verify with the req/cert public key. -key and -newkey set it, and + * wolfSSL can not verify with an RSA private key object. */ + pubKey = wolfSSL_X509_get_pubkey(x509); + if (pubKey == NULL) { wolfCLU_LogError("Error getting the public key to verify"); ret = WOLFCLU_FATAL_ERROR; } else { - if (wolfSSL_X509_REQ_verify(x509, pkey) == 1) { + if (isCSR) { + verifyRet = wolfSSL_X509_REQ_verify(x509, pubKey); + } + else { + verifyRet = wolfSSL_X509_verify(x509, pubKey); + } + wolfSSL_EVP_PKEY_free(pubKey); + + if (verifyRet == 1) { WOLFCLU_LOG(WOLFCLU_L0, "verify OK"); } else { diff --git a/tests/x509/x509-req-test.py b/tests/x509/x509-req-test.py index 7f3fbd43..bce9b25f 100644 --- a/tests/x509/x509-req-test.py +++ b/tests/x509/x509-req-test.py @@ -770,6 +770,79 @@ def test_verify_tampered_csr_no_output(self): "tampered CSR verify should fail") self.assertNotIn("BEGIN CERTIFICATE REQUEST", r.stdout) + def _req_verify(self, out, header, *args): + """Run req with -verify into out and expect verify OK and a PEM + of the given type.""" + self._clean(out) + r = run_wolfssl("req", *args, "-verify", "-out", out) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertIn("verify OK", r.stdout + r.stderr) + with open(out) as f: + self.assertIn("-----BEGIN {}-----".format(header), f.read()) + + def _new_x509_verify(self, key, out): + """req -new -x509 -verify checks the certificate it made (F-9846).""" + self._req_verify(out, "CERTIFICATE", "-new", "-x509", "-days", "30", + "-key", os.path.join(CERTS_DIR, key), + "-subj", "/O=wolfSSL/C=US/CN=verify-test") + r = run_wolfssl("x509", "-in", out, "-noout", "-subject") + self.assertEqual(r.returncode, 0, r.stderr) + + def test_verify_new_x509_rsa(self): + """req -new -x509 -verify with an RSA key (F-9846).""" + self._new_x509_verify("server-key.pem", + _tmp("test_req_verify_x509_rsa.pem")) + + def test_verify_new_x509_ecc(self): + """req -new -x509 -verify with an ECC key (F-9846).""" + self._new_x509_verify("ecc-key.pem", + _tmp("test_req_verify_x509_ecc.pem")) + + def test_verify_newkey_x509(self): + """req -newkey -x509 -verify checks the new certificate (F-9846).""" + key = _tmp("test_req_verify_newkey.key") + self._clean(key) + self._req_verify(_tmp("test_req_verify_newkey.pem"), "CERTIFICATE", + "-new", "-x509", "-days", "30", + "-newkey", "rsa:2048", "-nodes", "-keyout", key, + "-subj", "/O=wolfSSL/C=US/CN=verify-test") + + def test_verify_in_csr_x509(self): + """req -in csr -x509 -verify checks the re-signed cert (F-9846).""" + self._req_verify(_tmp("test_req_verify_in_x509.pem"), "CERTIFICATE", + "-in", self.csr_pem, "-x509", "-days", "30", + "-key", self.key) + + def test_verify_new_csr_rsa(self): + """req -new -verify still checks a new CSR (F-9846).""" + self._req_verify(_tmp("test_req_verify_new_rsa.csr"), + "CERTIFICATE REQUEST", "-new", "-key", self.key, + "-subj", "/O=wolfSSL/C=US/CN=verify-test") + + def test_verify_new_csr_ecc(self): + """req -new -verify still checks a new ECC CSR (F-9846).""" + self._req_verify(_tmp("test_req_verify_new_ecc.csr"), + "CERTIFICATE REQUEST", "-new", + "-key", os.path.join(CERTS_DIR, "ecc-key.pem"), + "-subj", "/O=wolfSSL/C=US/CN=verify-test") + + def test_verify_in_csr_key(self): + """req -in csr -key -verify still checks the CSR (F-9846).""" + self._req_verify(_tmp("test_req_verify_in_key.csr"), + "CERTIFICATE REQUEST", "-in", self.csr_pem, + "-key", self.key) + + def test_verify_in_csr_wrong_key_fails(self): + """req -in csr -verify with a -key that did not sign it fails.""" + r = run_wolfssl("req", "-in", self.csr_pem, "-noout", "-verify", + "-key", os.path.join(CERTS_DIR, "ecc-key.pem")) + self.assertNotEqual(r.returncode, 0, + "verify with the wrong key should fail") + self.assertGreaterEqual(r.returncode, 0, + "verify crashed with signal " + "{}".format(r.returncode)) + self.assertNotIn("verify OK", r.stdout + r.stderr) + class TestX509ReqSign(unittest.TestCase): """Test x509 -req -signkey signing.""" From eda40ac0ce7328ae21c5a785f903b1621efe6a81 Mon Sep 17 00:00:00 2001 From: Juliusz Sosinowicz Date: Wed, 23 Sep 2026 13:15:25 +0000 Subject: [PATCH 21/21] Serialize XMSS signing with a lock on the private state file Concurrent signers could reload the same XMSS state and reuse a one-time key. Hold an flock() on the .priv file (opened read-write, as NFS needs) from reload until the new state is saved and fsynced. On CIFS mounts without nobrl the lock is mandatory and signing fails closed. flock(), fsync() and fileno() are called directly: wolfSSL has no wrappers for them. F-12944 --- src/genkey/clu_genkey.c | 75 +++++++++++++++ src/sign-verify/clu_sign.c | 22 +++++ tests/genkey_sign_ver/genkey-sign-ver-test.py | 91 +++++++++++++++++++ wolfclu/genkey/clu_genkey.h | 18 ++++ 4 files changed, 206 insertions(+) diff --git a/src/genkey/clu_genkey.c b/src/genkey/clu_genkey.c index 781b14da..fd13d61d 100644 --- a/src/genkey/clu_genkey.c +++ b/src/genkey/clu_genkey.c @@ -32,6 +32,13 @@ #include #include /* PER_FORM/DER_FORM */ +#if defined(WOLFSSL_HAVE_XMSS) && defined(WOLFCLU_POSIX_FILE) + #include + #include + #include + #define WOLFCLU_XMSS_POSIX +#endif + #ifdef HAVE_ED25519 /* return WOLFCLU_SUCCESS on success */ int wolfCLU_genKey_ED25519(WC_RNG* rng, char* fOutNm, int directive, int format) @@ -1598,6 +1605,26 @@ enum wc_XmssRc wolfCLU_XmssKey_WriteCb(const byte * priv, return WC_XMSS_RC_WRITE_FAIL; } + /* The new state must reach the disk before the signature is used. */ + err = XFFLUSH(file); +#ifdef WOLFCLU_XMSS_POSIX + if (err == 0) { + err = fsync(fileno(file)); + } +#ifdef F_FULLFSYNC + /* macOS fsync() does not flush the drive cache. Not all file systems + * support this, so a failure is ignored. */ + if (err == 0) { + (void)fcntl(fileno(file), F_FULLFSYNC); + } +#endif +#endif + if (err) { + fprintf(stderr, "error: flushing %s failed\n", filename); + fclose(file); + return WC_XMSS_RC_WRITE_FAIL; + } + err = fclose(file); if (err) { fprintf(stderr, "error: fclose returned %d\n", err); @@ -1683,6 +1710,54 @@ enum wc_XmssRc wolfCLU_XmssKey_ReadCb(byte * priv, return WC_XMSS_RC_READ_TO_MEMORY; } + +/* Lock the private key file so only one process loads, signs and saves the + * one-time key state at a time. flock() is used because, unlike fcntl() + * locks, it is not released when the callbacks close their own handles. + * No lock is taken on other platforms. */ +int wolfCLU_XmssKey_Lock(const char* fileName, XFILE* lockFile) +{ +#ifdef WOLFCLU_XMSS_POSIX + XFILE file; + int err; +#endif + + if (fileName == NULL || lockFile == NULL) { + return WOLFCLU_FATAL_ERROR; + } + *lockFile = XBADFILE; + +#ifdef WOLFCLU_XMSS_POSIX + /* NFS only allows an exclusive flock() on a file open for writing. */ + file = XFOPEN(fileName, "r+b"); + if (file == XBADFILE) { + wolfCLU_LogError("Unable to open %s: %s", fileName, strerror(errno)); + return WOLFCLU_FATAL_ERROR; + } + + /* wolfSSL has no file lock wrapper */ + do { + err = flock(fileno(file), LOCK_EX); + } while (err != 0 && errno == EINTR); + + if (err != 0) { + wolfCLU_LogError("Unable to lock %s: %s", fileName, strerror(errno)); + XFCLOSE(file); + return WOLFCLU_FATAL_ERROR; + } + *lockFile = file; +#endif + + return WOLFCLU_SUCCESS; +} + +void wolfCLU_XmssKey_Unlock(XFILE lockFile) +{ + /* Closing the file releases the lock. */ + if (lockFile != XBADFILE) { + XFCLOSE(lockFile); + } +} #endif /* WOLFSSL_HAVE_XMSS */ int wolfCLU_genKey_XMSS(WC_RNG* rng, char* fName, diff --git a/src/sign-verify/clu_sign.c b/src/sign-verify/clu_sign.c index 63217f9f..63e60381 100644 --- a/src/sign-verify/clu_sign.c +++ b/src/sign-verify/clu_sign.c @@ -920,6 +920,7 @@ int wolfCLU_sign_data_xmss(byte* data, char* out, int fSz, char* privKey) word32 outBufSz = 0; /* signature buffer size */ char* paramStr = NULL; /* parameter string */ int paramLen = XMSS_NAME_LEN + 1; /* parameter string length */ + XFILE lockFile = XBADFILE; /* private key file lock */ #ifdef WOLFSSL_SMALL_STACK XmssKey *key = (XmssKey*)XMALLOC(sizeof(XmssKey), @@ -1022,6 +1023,15 @@ int wolfCLU_sign_data_xmss(byte* data, char* out, int fSz, char* privKey) } } + /* lock the private key file until the new state is saved */ + if (ret == 0) { + if (wolfCLU_XmssKey_Lock(privKey, &lockFile) != WOLFCLU_SUCCESS) { + ret = WOLFCLU_FATAL_ERROR; + wolfCLU_LogError("Failed to lock XMSS private key file %s.", + privKey); + } + } + /* reload XMSS key to be signable state */ if (ret == 0) { ret = wc_XmssKey_Reload(key); @@ -1057,6 +1067,7 @@ int wolfCLU_sign_data_xmss(byte* data, char* out, int fSz, char* privKey) } /* clena up allocated memory */ + wolfCLU_XmssKey_Unlock(lockFile); if (outFile != NULL) { XFCLOSE(outFile); } @@ -1095,6 +1106,7 @@ int wolfCLU_sign_data_xmssmt(byte* data, char* out, int fSz, char* privKey) int paramLen = 0; /* parameter string length */ int privKeyLen = 0; /* private key file name length */ int fileHeadLen = 7; /* file header(XMSSMT-) length */ + XFILE lockFile = XBADFILE; /* private key file lock */ if (privKey == NULL) { return BAD_FUNC_ARG; @@ -1215,6 +1227,15 @@ int wolfCLU_sign_data_xmssmt(byte* data, char* out, int fSz, char* privKey) } } + /* lock the private key file until the new state is saved */ + if (ret == 0) { + if (wolfCLU_XmssKey_Lock(privKey, &lockFile) != WOLFCLU_SUCCESS) { + ret = WOLFCLU_FATAL_ERROR; + wolfCLU_LogError("Failed to lock XMSS^MT private key file %s.", + privKey); + } + } + /* reload XMSS^MT key to be signable state */ if (ret == 0) { ret = wc_XmssKey_Reload(key); @@ -1251,6 +1272,7 @@ int wolfCLU_sign_data_xmssmt(byte* data, char* out, int fSz, char* privKey) } /* clena up allocated memory */ + wolfCLU_XmssKey_Unlock(lockFile); if (outFile != NULL) { XFCLOSE(outFile); } diff --git a/tests/genkey_sign_ver/genkey-sign-ver-test.py b/tests/genkey_sign_ver/genkey-sign-ver-test.py index 9250c32b..85592ed1 100644 --- a/tests/genkey_sign_ver/genkey-sign-ver-test.py +++ b/tests/genkey_sign_ver/genkey-sign-ver-test.py @@ -2,9 +2,15 @@ """Key generation, signing, and verification tests for wolfCLU.""" import os +import subprocess import sys import unittest +try: + import fcntl +except ImportError: + fcntl = None + sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from wolfclu_test import (WOLFSSL_BIN, CERTS_DIR, not_compiled_in, run_wolfssl, test_main) @@ -133,6 +139,62 @@ def _verify_pub(self, algo, pub_key, fmt, sig_file, out_file=None): self.assertEqual(r.returncode, 0, f"public verify {algo} failed: {r.stderr}") + def _start_sign(self, algo, priv_key, sig_file): + """Start a raw-format sign in the background.""" + self._track(sig_file) + proc = subprocess.Popen( + [WOLFSSL_BIN, f"-{algo}", "-sign", "-inkey", priv_key, + "-inform", "raw", "-in", self.SIGN_FILE, "-out", sig_file], + stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True) + self.addCleanup(self._stop_process, proc) + return proc + + @staticmethod + def _stop_process(proc): + if proc.poll() is None: + proc.kill() + proc.communicate() + + @staticmethod + def _xmss_sig_index(sig_file, idx_len): + """The XMSS/XMSS^MT signature starts with the big-endian leaf index.""" + with open(sig_file, "rb") as f: + return int.from_bytes(f.read(idx_len), "big") + + def _check_sign_waits_for_lock(self, algo, keybase, genkey_args, + idx_len): + """Signing must hold an exclusive lock on the private state file + from reload until the new state is saved (F-12944).""" + if fcntl is None: + self.skipTest("fcntl.flock not available") + priv, pub = self._genkey(algo, keybase, "raw", genkey_args, + use_output_flag=True) + first_sig = keybase + "-lock-1.sig" + second_sig = keybase + "-lock-2.sig" + self._sign(algo, priv, "raw", first_sig) + + lock_file = open(priv, "rb+") + self.addCleanup(lock_file.close) + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + + proc = self._start_sign(algo, priv, second_sig) + try: + proc.wait(timeout=1.5) + except subprocess.TimeoutExpired: + pass + self.assertIsNone(proc.returncode, + "{} sign did not wait for the lock on {}".format( + algo, priv)) + + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + _, err = proc.communicate(timeout=60) + self.assertEqual(proc.returncode, 0, f"sign {algo} failed: {err}") + self.assertNotEqual(self._xmss_sig_index(first_sig, idx_len), + self._xmss_sig_index(second_sig, idx_len), + "{} signature index was reused".format(algo)) + self._verify_pub(algo, pub, "raw", second_sig) + def _gen_sign_verify(self, algo, keybase, sig_file, fmt, extra_genkey_args=None, skip_priv_verify=False, rsa_verify_out=None, use_output_flag=False): @@ -482,6 +544,30 @@ def test_xmss_raw(self): extra_genkey_args=["-height", "10"], skip_priv_verify=True, use_output_flag=True) + def test_xmss_sign_waits_for_state_lock(self): + self._check_sign_waits_for_lock("xmss", "XMSS-SHA2_10_256", + ["-height", "10"], 4) + + def test_xmss_concurrent_sign_unique_index(self): + """Concurrent signers must each use a new one-time key (F-12944).""" + if fcntl is None: + self.skipTest("no file locking on this platform") + priv, _ = self._genkey("xmss", "XMSS-SHA2_10_256", "raw", + ["-height", "10"], use_output_flag=True) + signers = [] + for i in range(6): + sig_file = "xmss-concurrent-{}.sig".format(i) + signers.append((self._start_sign("xmss", priv, sig_file), + sig_file)) + indices = [] + for proc, sig_file in signers: + _, err = proc.communicate(timeout=60) + self.assertEqual(proc.returncode, 0, + "concurrent xmss sign failed: {}".format(err)) + indices.append(self._xmss_sig_index(sig_file, 4)) + self.assertEqual(len(set(indices)), len(indices), + "xmss signature index reused: {}".format(indices)) + def test_xmss_missing_height_value(self): """-height with no value must fail gracefully (no crash).""" self._track("xmss-bad.priv", "xmss-bad.pub") @@ -514,6 +600,11 @@ def test_xmssmt_raw(self): extra_genkey_args=["-height", "20"], skip_priv_verify=True, use_output_flag=True) + def test_xmssmt_sign_waits_for_state_lock(self): + # XMSS^MT with height 20 uses a 3-byte index. + self._check_sign_waits_for_lock("xmssmt", "XMSSMT-SHA2_20-2_256", + ["-height", "20"], 3) + def test_xmssmt_missing_height_value(self): """-height with no value must fail gracefully (no crash).""" self._track("xmss-bad.priv", "xmss-bad.pub") diff --git a/wolfclu/genkey/clu_genkey.h b/wolfclu/genkey/clu_genkey.h index a670c5ef..8a42a1ad 100644 --- a/wolfclu/genkey/clu_genkey.h +++ b/wolfclu/genkey/clu_genkey.h @@ -150,6 +150,24 @@ int wolfCLU_genKey_ML_DSA(WC_RNG* rng, char* fName, int directive, int fmt, #ifdef WOLFSSL_HAVE_XMSS enum wc_XmssRc wolfCLU_XmssKey_WriteCb(const byte* priv, word32 privSz, void* context); enum wc_XmssRc wolfCLU_XmssKey_ReadCb(byte* priv, word32 privSz, void* context); + +/** + * take an exclusive lock on a XMSS private key file, waiting for other + * signers to release it + * + * @param fileName the private key file + * @param lockFile set to the handle to pass to wolfCLU_XmssKey_Unlock + * + * return WOLFCLU_SUCCESS on success +*/ +int wolfCLU_XmssKey_Lock(const char* fileName, XFILE* lockFile); + +/** + * release a lock taken by wolfCLU_XmssKey_Lock + * + * @param lockFile the handle set by wolfCLU_XmssKey_Lock +*/ +void wolfCLU_XmssKey_Unlock(XFILE lockFile); #endif /**