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/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/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/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/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/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/src/genkey/clu_genkey.c b/src/genkey/clu_genkey.c index b9c512e8..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) @@ -121,7 +128,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 +670,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 +856,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 +1187,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 +1435,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 +1589,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; } } @@ -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/pkcs/clu_pkcs12.c b/src/pkcs/clu_pkcs12.c index 29b08219..167c6c63 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"); @@ -227,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 9c4c7049..21b3fb19 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); @@ -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; } @@ -258,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; @@ -270,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/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/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/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..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; @@ -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/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/src/tools/clu_funcs.c b/src/tools/clu_funcs.c index 71c74c8a..bf187543 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 @@ -292,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); @@ -333,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; } @@ -431,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; @@ -887,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 " @@ -909,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"); @@ -935,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); @@ -1112,6 +1123,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) @@ -1237,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; } @@ -1251,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_cert_setup.c b/src/x509/clu_cert_setup.c index 1f05998b..9564c03e 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; } @@ -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) { 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/src/x509/clu_request_setup.c b/src/x509/clu_request_setup.c index 6cea40c0..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 { @@ -1110,7 +1114,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()); @@ -1131,9 +1135,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/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() 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() diff --git a/tests/encrypt/enc-test.py b/tests/encrypt/enc-test.py index 5fa48c7f..63c94060 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 @@ -463,6 +464,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.""" @@ -912,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): @@ -1094,6 +1194,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/genkey_sign_ver/genkey-sign-ver-test.py b/tests/genkey_sign_ver/genkey-sign-ver-test.py index 14a777df..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") @@ -532,6 +623,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/tests/pkcs/pkcs12-test.py b/tests/pkcs/pkcs12-test.py index 7d067f1c..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") @@ -85,14 +86,97 @@ def test_out_bad_path_fails(self): "-out", os.path.join("no-such-dir", "out.pem")) self.assertNotEqual(r.returncode, 0) - def test_nocerts_with_passout(self): + 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): + """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) + 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") + @unittest.skipUnless(HAVE_PTY, "pty not available") + def test_out_encrypted_key_owner_only(self): + # 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)) + + @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_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 2a23aa3d..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 @@ -8,7 +9,39 @@ 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) + +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): @@ -72,6 +105,143 @@ 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).""" + 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) @@ -116,6 +286,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/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() 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() 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() 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() 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-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 82b1153e..bce9b25f 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): @@ -49,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 @@ -146,6 +150,75 @@ 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 + + +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.""" @@ -189,6 +262,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. @@ -397,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") @@ -611,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.""" @@ -668,6 +900,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.""" @@ -781,6 +1027,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. @@ -989,6 +1255,60 @@ 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): + """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): 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 */ 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 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 /**