完善不同加密方法的使用
This commit is contained in:
yun
@@ -0,0 +1,812 @@
|
||||
package gpgx_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"code.yun.ink/pkg/encryptx/gpgx"
|
||||
|
||||
"github.com/ProtonMail/go-crypto/openpgp"
|
||||
"github.com/ProtonMail/go-crypto/openpgp/armor"
|
||||
)
|
||||
|
||||
// 测试密钥统一在内存中生成一次（Ed25519，速度快），不依赖仓库里的任何
|
||||
// 固定密钥文件。生成失败直接 Fatal，保证后续用例不会在 nil 上继续。
|
||||
var (
|
||||
testKeysOnce sync.Once
|
||||
|
||||
alicePrivateArmor []byte
|
||||
alicePublicArmor []byte
|
||||
alicePrivate openpgp.EntityList
|
||||
alicePublic openpgp.EntityList
|
||||
|
||||
bobPrivateArmor []byte
|
||||
bobPublicArmor []byte
|
||||
bobPublic openpgp.EntityList
|
||||
|
||||
// 带口令的 Ed25519 密钥。
|
||||
passphrase = []byte("correct horse battery staple")
|
||||
lockedPrivate []byte
|
||||
lockedPublicArmor []byte
|
||||
|
||||
testKeyErr error
|
||||
)
|
||||
|
||||
func initTestKeys(t *testing.T) {
|
||||
t.Helper()
|
||||
testKeysOnce.Do(func() {
|
||||
alicePrivateArmor, alicePublicArmor, testKeyErr = gpgx.GenerateKeyPair(
|
||||
"Alice Tester", "alice@example.com", "gpgx test", nil, 0)
|
||||
if testKeyErr != nil {
|
||||
return
|
||||
}
|
||||
alicePrivate, testKeyErr = gpgx.ReadArmoredKeyRing(bytes.NewReader(alicePrivateArmor))
|
||||
if testKeyErr != nil {
|
||||
return
|
||||
}
|
||||
alicePublic, testKeyErr = gpgx.ReadArmoredKeyRing(bytes.NewReader(alicePublicArmor))
|
||||
if testKeyErr != nil {
|
||||
return
|
||||
}
|
||||
|
||||
bobPrivateArmor, bobPublicArmor, testKeyErr = gpgx.GenerateKeyPair(
|
||||
"Bob Tester", "bob@example.com", "gpgx test", nil, 0)
|
||||
if testKeyErr != nil {
|
||||
return
|
||||
}
|
||||
bobPublic, testKeyErr = gpgx.ReadArmoredKeyRing(bytes.NewReader(bobPublicArmor))
|
||||
if testKeyErr != nil {
|
||||
return
|
||||
}
|
||||
|
||||
lockedPrivate, lockedPublicArmor, testKeyErr = gpgx.GenerateKeyPair(
|
||||
"Carol Locked", "carol@example.com", "gpgx test", passphrase, 0)
|
||||
})
|
||||
if testKeyErr != nil {
|
||||
t.Fatalf("初始化测试密钥失败: %v", testKeyErr)
|
||||
}
|
||||
}
|
||||
|
||||
// requireKeys 声明用例对已生成测试密钥的依赖。
|
||||
func requireKeys(t *testing.T) {
|
||||
t.Helper()
|
||||
if len(alicePrivate) == 0 || len(alicePublic) == 0 {
|
||||
t.Fatal("Alice 测试密钥不可用")
|
||||
}
|
||||
}
|
||||
|
||||
// mustNotPanic 断言 fn 既不 panic 也返回 error（用于所有错误路径）。
|
||||
func mustNotPanic(t *testing.T, name string, fn func() error) {
|
||||
t.Helper()
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
t.Fatalf("%s: 发生 panic: %v", name, r)
|
||||
}
|
||||
}()
|
||||
if err := fn(); err == nil {
|
||||
t.Fatalf("%s: 期望返回 error，实际返回 nil", name)
|
||||
}
|
||||
}
|
||||
|
||||
// generatePlaintext 生成确定性的测试明文。
|
||||
func generatePlaintext(size int) []byte {
|
||||
out := make([]byte, size)
|
||||
for i := range out {
|
||||
out[i] = byte((i * 7) % 251)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// tamperByte 翻转 data 中第 idx 个字节；若 idx 越界则返回原数据的副本。
|
||||
func tamperByte(data []byte, idx int) []byte {
|
||||
out := append([]byte(nil), data...)
|
||||
if idx >= 0 && idx < len(out) {
|
||||
out[idx] ^= 0x01
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------ 测试用例
|
||||
|
||||
func TestGenerateKeyPairAndReadKeyRing(t *testing.T) {
|
||||
initTestKeys(t)
|
||||
requireKeys(t)
|
||||
|
||||
if !strings.HasPrefix(string(alicePrivateArmor), "-----BEGIN PGP PRIVATE KEY BLOCK-----") {
|
||||
t.Fatalf("私钥 armor 头不正确: %q", firstLine(alicePrivateArmor))
|
||||
}
|
||||
if !strings.HasPrefix(string(alicePublicArmor), "-----BEGIN PGP PUBLIC KEY BLOCK-----") {
|
||||
t.Fatalf("公钥 armor 头不正确: %q", firstLine(alicePublicArmor))
|
||||
}
|
||||
|
||||
// armored 往返（私钥）。
|
||||
private, err := gpgx.ReadArmoredKeyRing(bytes.NewReader(alicePrivateArmor))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadArmoredKeyRing(私钥) 失败: %v", err)
|
||||
}
|
||||
if len(private) != 1 {
|
||||
t.Fatalf("期望 1 个实体，实际 %d", len(private))
|
||||
}
|
||||
if private[0].PrivateKey == nil {
|
||||
t.Fatal("私钥密钥环中缺少私钥材料")
|
||||
}
|
||||
if private[0].PrivateKey.Encrypted {
|
||||
t.Fatal("无口令私钥不应处于加密状态")
|
||||
}
|
||||
if private[0].PrimaryIdentity() == nil {
|
||||
t.Fatal("密钥环中的实体缺少 User ID")
|
||||
}
|
||||
|
||||
// armored 往返（公钥）：公钥密钥环不应带私钥材料。
|
||||
public, err := gpgx.ReadArmoredKeyRing(bytes.NewReader(alicePublicArmor))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadArmoredKeyRing(公钥) 失败: %v", err)
|
||||
}
|
||||
if len(public) != 1 {
|
||||
t.Fatalf("期望 1 个实体，实际 %d", len(public))
|
||||
}
|
||||
if public[0].PrivateKey != nil {
|
||||
t.Fatal("公钥密钥环不应包含私钥材料")
|
||||
}
|
||||
if public[0].PrimaryKey.KeyId != private[0].PrimaryKey.KeyId {
|
||||
t.Fatal("公私钥 KeyId 不一致")
|
||||
}
|
||||
|
||||
// ReadKeyRing 自动识别 binary-armored / armor 两种输入。
|
||||
auto, err := gpgx.ReadKeyRing(bytes.NewReader(alicePrivateArmor))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadKeyRing(armored) 失败: %v", err)
|
||||
}
|
||||
if len(auto) != 1 || auto[0].PrimaryKey.KeyId != private[0].PrimaryKey.KeyId {
|
||||
t.Fatal("ReadKeyRing 未正确识别 armored 输入")
|
||||
}
|
||||
|
||||
// 二进制形式：armored 私钥去掉 armor 后应能被 ReadKeyRing 解析。
|
||||
binaryKeyRing := decodeArmor(t, alicePrivateArmor)
|
||||
fromBinary, err := gpgx.ReadKeyRing(bytes.NewReader(binaryKeyRing))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadKeyRing(二进制) 失败: %v", err)
|
||||
}
|
||||
if len(fromBinary) != 1 || fromBinary[0].PrimaryKey.KeyId != private[0].PrimaryKey.KeyId {
|
||||
t.Fatal("ReadKeyRing 未正确识别二进制输入")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadKeyRingFile(t *testing.T) {
|
||||
initTestKeys(t)
|
||||
requireKeys(t)
|
||||
|
||||
dir := t.TempDir()
|
||||
armoredPath := filepath.Join(dir, "alice.asc")
|
||||
if err := os.WriteFile(armoredPath, alicePrivateArmor, 0o600); err != nil {
|
||||
t.Fatalf("写入 armored 密钥文件失败: %v", err)
|
||||
}
|
||||
entities, err := gpgx.ReadKeyRingFile(armoredPath)
|
||||
if err != nil {
|
||||
t.Fatalf("ReadKeyRingFile(armored) 失败: %v", err)
|
||||
}
|
||||
if len(entities) != 1 || entities[0].PrivateKey == nil {
|
||||
t.Fatal("ReadKeyRingFile 未读回私钥材料")
|
||||
}
|
||||
|
||||
binaryPath := filepath.Join(dir, "alice.pgp")
|
||||
if err := os.WriteFile(binaryPath, decodeArmor(t, alicePrivateArmor), 0o600); err != nil {
|
||||
t.Fatalf("写入二进制密钥文件失败: %v", err)
|
||||
}
|
||||
if _, err := gpgx.ReadKeyRingFile(binaryPath); err != nil {
|
||||
t.Fatalf("ReadKeyRingFile(二进制) 失败: %v", err)
|
||||
}
|
||||
|
||||
// 文件不存在。
|
||||
mustNotPanic(t, "ReadKeyRingFile(不存在)", func() error {
|
||||
_, err := gpgx.ReadKeyRingFile(filepath.Join(dir, "missing.asc"))
|
||||
return err
|
||||
})
|
||||
|
||||
// 空文件。
|
||||
emptyPath := filepath.Join(dir, "empty.asc")
|
||||
if err := os.WriteFile(emptyPath, nil, 0o600); err != nil {
|
||||
t.Fatalf("写入空文件失败: %v", err)
|
||||
}
|
||||
mustNotPanic(t, "ReadKeyRingFile(空文件)", func() error {
|
||||
_, err := gpgx.ReadKeyRingFile(emptyPath)
|
||||
return err
|
||||
})
|
||||
|
||||
// 垃圾内容。
|
||||
garbagePath := filepath.Join(dir, "garbage.asc")
|
||||
if err := os.WriteFile(garbagePath, []byte("this is not a pgp key\n"), 0o600); err != nil {
|
||||
t.Fatalf("写入垃圾文件失败: %v", err)
|
||||
}
|
||||
mustNotPanic(t, "ReadKeyRingFile(垃圾内容)", func() error {
|
||||
_, err := gpgx.ReadKeyRingFile(garbagePath)
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
func TestEncryptDecryptRoundTrip(t *testing.T) {
|
||||
initTestKeys(t)
|
||||
requireKeys(t)
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
plaintext []byte
|
||||
}{
|
||||
{"空明文", []byte{}},
|
||||
{"普通明文", []byte("hello, openpgp! 你好，世界")},
|
||||
{"10KB 明文", generatePlaintext(10 * 1024)},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run("armored/"+tc.name, func(t *testing.T) {
|
||||
ciphertext, err := gpgx.EncryptArmored(alicePublic, tc.plaintext)
|
||||
if err != nil {
|
||||
t.Fatalf("EncryptArmored 失败: %v", err)
|
||||
}
|
||||
if !strings.HasPrefix(string(ciphertext), "-----BEGIN PGP MESSAGE-----") {
|
||||
t.Fatalf("密文 armor 头不正确: %q", firstLine(ciphertext))
|
||||
}
|
||||
|
||||
plaintext, err := gpgx.Decrypt(alicePrivate, ciphertext)
|
||||
if err != nil {
|
||||
t.Fatalf("Decrypt 失败: %v", err)
|
||||
}
|
||||
if !bytes.Equal(plaintext, tc.plaintext) {
|
||||
t.Fatalf("解密结果不匹配: 期望 %d 字节，实际 %d 字节", len(tc.plaintext), len(plaintext))
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("binary/"+tc.name, func(t *testing.T) {
|
||||
ciphertext, err := gpgx.Encrypt(alicePublic, tc.plaintext)
|
||||
if err != nil {
|
||||
t.Fatalf("Encrypt 失败: %v", err)
|
||||
}
|
||||
if bytes.HasPrefix(ciphertext, []byte("-----BEGIN")) {
|
||||
t.Fatal("Encrypt 应返回二进制报文，而不是 armor 文本")
|
||||
}
|
||||
|
||||
plaintext, err := gpgx.Decrypt(alicePrivate, ciphertext)
|
||||
if err != nil {
|
||||
t.Fatalf("Decrypt(二进制) 失败: %v", err)
|
||||
}
|
||||
if !bytes.Equal(plaintext, tc.plaintext) {
|
||||
t.Fatalf("解密结果不匹配: 期望 %d 字节，实际 %d 字节", len(tc.plaintext), len(plaintext))
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// 多次加密同一明文应产生不同密文（会话密钥随机）。
|
||||
first, err := gpgx.EncryptArmored(alicePublic, []byte("same input"))
|
||||
if err != nil {
|
||||
t.Fatalf("EncryptArmored 失败: %v", err)
|
||||
}
|
||||
second, err := gpgx.EncryptArmored(alicePublic, []byte("same input"))
|
||||
if err != nil {
|
||||
t.Fatalf("EncryptArmored 失败: %v", err)
|
||||
}
|
||||
if bytes.Equal(first, second) {
|
||||
t.Fatal("两次加密产生了完全相同的密文，会话密钥可能未随机化")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecryptErrors(t *testing.T) {
|
||||
initTestKeys(t)
|
||||
requireKeys(t)
|
||||
|
||||
ciphertext, err := gpgx.EncryptArmored(alicePublic, []byte("top secret payload"))
|
||||
if err != nil {
|
||||
t.Fatalf("EncryptArmored 失败: %v", err)
|
||||
}
|
||||
|
||||
// 用另一个密钥环解密。
|
||||
mustNotPanic(t, "错误密钥环解密", func() error {
|
||||
_, err := gpgx.Decrypt(bobPublic, ciphertext)
|
||||
return err
|
||||
})
|
||||
|
||||
// 篡改密文 1 字节。
|
||||
mustNotPanic(t, "篡改密文解密", func() error {
|
||||
_, err := gpgx.Decrypt(alicePrivate, tamperByte(ciphertext, len(ciphertext)/2))
|
||||
return err
|
||||
})
|
||||
|
||||
// 丢掉 armor 尾部（截断密文）。
|
||||
mustNotPanic(t, "截断密文解密", func() error {
|
||||
_, err := gpgx.Decrypt(alicePrivate, ciphertext[:len(ciphertext)/2])
|
||||
return err
|
||||
})
|
||||
|
||||
// 空密文与垃圾字节。
|
||||
mustNotPanic(t, "空密文解密", func() error {
|
||||
_, err := gpgx.Decrypt(alicePrivate, nil)
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "垃圾密文解密", func() error {
|
||||
_, err := gpgx.Decrypt(alicePrivate, []byte("not a pgp message at all"))
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "随机字节解密", func() error {
|
||||
_, err := gpgx.Decrypt(alicePrivate, generatePlaintext(256))
|
||||
return err
|
||||
})
|
||||
|
||||
// 空密钥环。
|
||||
mustNotPanic(t, "空密钥环解密", func() error {
|
||||
_, err := gpgx.Decrypt(nil, ciphertext)
|
||||
return err
|
||||
})
|
||||
|
||||
// 加密侧的参数校验。
|
||||
mustNotPanic(t, "空接收方加密", func() error {
|
||||
_, err := gpgx.Encrypt(nil, []byte("x"))
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "空接收方加密(armored)", func() error {
|
||||
_, err := gpgx.EncryptArmored(openpgp.EntityList{}, []byte("x"))
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "仅有私钥的接收方加密", func() error {
|
||||
_, err := gpgx.Encrypt(openpgp.EntityList{}, []byte("x"))
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
func TestSignVerifyRoundTrip(t *testing.T) {
|
||||
initTestKeys(t)
|
||||
requireKeys(t)
|
||||
|
||||
cases := [][]byte{
|
||||
{},
|
||||
[]byte("message to be signed"),
|
||||
generatePlaintext(4096),
|
||||
}
|
||||
for _, plaintext := range cases {
|
||||
signed, err := gpgx.Sign(alicePrivate, plaintext, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("Sign 失败: %v", err)
|
||||
}
|
||||
got, err := gpgx.Verify(alicePublic, signed)
|
||||
if err != nil {
|
||||
t.Fatalf("Verify 失败: %v", err)
|
||||
}
|
||||
if !bytes.Equal(got, plaintext) {
|
||||
t.Fatalf("验签返回的明文不匹配: 期望 %d 字节，实际 %d 字节", len(plaintext), len(got))
|
||||
}
|
||||
}
|
||||
|
||||
// 私钥环也可以验签（其中包含公钥）。
|
||||
signed, err := gpgx.Sign(alicePrivate, []byte("verify with private keyring"), nil)
|
||||
if err != nil {
|
||||
t.Fatalf("Sign 失败: %v", err)
|
||||
}
|
||||
if _, err := gpgx.Verify(alicePrivate, signed); err != nil {
|
||||
t.Fatalf("使用私钥环 Verify 失败: %v", err)
|
||||
}
|
||||
|
||||
// 分离签名。
|
||||
data := []byte("detached signature payload")
|
||||
signature, err := gpgx.SignDetachedArmored(alicePrivate, data, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("SignDetachedArmored 失败: %v", err)
|
||||
}
|
||||
if !strings.HasPrefix(string(signature), "-----BEGIN PGP SIGNATURE-----") {
|
||||
t.Fatalf("分离签名 armor 头不正确: %q", firstLine(signature))
|
||||
}
|
||||
signer, err := gpgx.VerifyDetached(alicePublic, data, signature)
|
||||
if err != nil {
|
||||
t.Fatalf("VerifyDetached 失败: %v", err)
|
||||
}
|
||||
if signer == nil || signer.PrimaryKey.KeyId != alicePublic[0].PrimaryKey.KeyId {
|
||||
t.Fatal("VerifyDetached 返回的签名者不正确")
|
||||
}
|
||||
|
||||
// 二进制分离签名同样可验证。
|
||||
binarySignature := decodeArmor(t, signature)
|
||||
if _, err := gpgx.VerifyDetached(alicePublic, data, binarySignature); err != nil {
|
||||
t.Fatalf("VerifyDetached(二进制签名) 失败: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestVerifyTamperedMessageAndWrongKey(t *testing.T) {
|
||||
initTestKeys(t)
|
||||
requireKeys(t)
|
||||
|
||||
data := []byte("important signed statement")
|
||||
signature, err := gpgx.SignDetachedArmored(alicePrivate, data, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("SignDetachedArmored 失败: %v", err)
|
||||
}
|
||||
|
||||
// 篡改被签名的消息。
|
||||
mustNotPanic(t, "篡改消息后验签", func() error {
|
||||
_, err := gpgx.VerifyDetached(alicePublic, tamperByte(data, 3), signature)
|
||||
return err
|
||||
})
|
||||
|
||||
// 篡改签名本身。
|
||||
mustNotPanic(t, "篡改签名后验签", func() error {
|
||||
_, err := gpgx.VerifyDetached(alicePublic, data, tamperByte(signature, len(signature)/2))
|
||||
return err
|
||||
})
|
||||
|
||||
// 用错误的公钥环验签。
|
||||
mustNotPanic(t, "错误公钥环验签", func() error {
|
||||
_, err := gpgx.VerifyDetached(bobPublic, data, signature)
|
||||
return err
|
||||
})
|
||||
|
||||
// 空/垃圾输入。
|
||||
mustNotPanic(t, "空签名验签", func() error {
|
||||
_, err := gpgx.VerifyDetached(alicePublic, data, nil)
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "垃圾签名验签", func() error {
|
||||
_, err := gpgx.VerifyDetached(alicePublic, data, []byte("garbage signature"))
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "空密钥环验签", func() error {
|
||||
_, err := gpgx.VerifyDetached(nil, data, signature)
|
||||
return err
|
||||
})
|
||||
|
||||
// 内联签名报文的错误路径。
|
||||
signed, err := gpgx.Sign(alicePrivate, data, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("Sign 失败: %v", err)
|
||||
}
|
||||
mustNotPanic(t, "篡改内联签名报文", func() error {
|
||||
_, err := gpgx.Verify(alicePublic, tamperByte(signed, len(signed)/2))
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "错误密钥环验证内联签名", func() error {
|
||||
_, err := gpgx.Verify(bobPublic, signed)
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "空报文验签", func() error {
|
||||
_, err := gpgx.Verify(alicePublic, nil)
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "垃圾报文验签", func() error {
|
||||
_, err := gpgx.Verify(alicePublic, []byte("definitely not signed"))
|
||||
return err
|
||||
})
|
||||
// 未签名的加密报文不能被当作签名报文验证。
|
||||
ciphertext, err := gpgx.EncryptArmored(alicePublic, data)
|
||||
if err != nil {
|
||||
t.Fatalf("EncryptArmored 失败: %v", err)
|
||||
}
|
||||
mustNotPanic(t, "加密报文当作签名报文", func() error {
|
||||
_, err := gpgx.Verify(alicePublic, ciphertext)
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
func TestPassphraseProtectedKey(t *testing.T) {
|
||||
initTestKeys(t)
|
||||
requireKeys(t)
|
||||
if len(lockedPrivate) == 0 {
|
||||
t.Fatal("带口令的测试密钥不可用")
|
||||
}
|
||||
|
||||
// 口令必须真的生效：读回来的私钥应处于加密状态。
|
||||
locked, err := gpgx.ReadArmoredKeyRing(bytes.NewReader(lockedPrivate))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadArmoredKeyRing(带口令私钥) 失败: %v", err)
|
||||
}
|
||||
if locked[0].PrivateKey == nil || !locked[0].PrivateKey.Encrypted {
|
||||
t.Fatal("GenerateKeyPair 未加密私钥")
|
||||
}
|
||||
lockedPublic, err := gpgx.ReadArmoredKeyRing(bytes.NewReader(lockedPublicArmor))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadArmoredKeyRing(带口令公钥) 失败: %v", err)
|
||||
}
|
||||
|
||||
data := []byte("passphrase protected payload")
|
||||
|
||||
// 正确口令：签名 + 验签。
|
||||
signature, err := gpgx.SignDetachedArmored(locked, data, passphrase)
|
||||
if err != nil {
|
||||
t.Fatalf("使用正确口令签名失败: %v", err)
|
||||
}
|
||||
if _, err := gpgx.VerifyDetached(lockedPublic, data, signature); err != nil {
|
||||
t.Fatalf("验签失败: %v", err)
|
||||
}
|
||||
|
||||
// 正确口令：内联签名。
|
||||
signed, err := gpgx.Sign(locked, data, passphrase)
|
||||
if err != nil {
|
||||
t.Fatalf("使用正确口令生成内联签名失败: %v", err)
|
||||
}
|
||||
if got, err := gpgx.Verify(lockedPublic, signed); err != nil || !bytes.Equal(got, data) {
|
||||
t.Fatalf("内联签名往返失败: err=%v", err)
|
||||
}
|
||||
|
||||
// 正确口令：解密。
|
||||
ciphertext, err := gpgx.EncryptArmored(lockedPublic, data)
|
||||
if err != nil {
|
||||
t.Fatalf("EncryptArmored 失败: %v", err)
|
||||
}
|
||||
plaintext, err := gpgx.DecryptWithPassphrase(locked, ciphertext, passphrase)
|
||||
if err != nil {
|
||||
t.Fatalf("使用正确口令解密失败: %v", err)
|
||||
}
|
||||
if !bytes.Equal(plaintext, data) {
|
||||
t.Fatal("使用正确口令解密的结果不匹配")
|
||||
}
|
||||
|
||||
// 错误口令：签名与解密都必须失败且不 panic。
|
||||
wrong := []byte("totally wrong passphrase")
|
||||
mustNotPanic(t, "错误口令签名", func() error {
|
||||
_, err := gpgx.Sign(locked, data, wrong)
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "错误口令生成分离签名", func() error {
|
||||
_, err := gpgx.SignDetachedArmored(locked, data, wrong)
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "错误口令解密", func() error {
|
||||
_, err := gpgx.DecryptWithPassphrase(locked, ciphertext, wrong)
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "未提供口令解密", func() error {
|
||||
_, err := gpgx.DecryptWithPassphrase(locked, ciphertext, nil)
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "未提供口令签名", func() error {
|
||||
_, err := gpgx.Sign(locked, data, nil)
|
||||
return err
|
||||
})
|
||||
|
||||
// 关键回归断言：上面的失败尝试不得破坏调用方传入的密钥环
|
||||
// （实现应在副本上解密私钥）。
|
||||
plaintext, err = gpgx.DecryptWithPassphrase(locked, ciphertext, passphrase)
|
||||
if err != nil {
|
||||
t.Fatalf("错误口令尝试后，正确口令解密失败（密钥环被污染）: %v", err)
|
||||
}
|
||||
if !bytes.Equal(plaintext, data) {
|
||||
t.Fatal("错误口令尝试后，正确口令解密的结果不匹配")
|
||||
}
|
||||
if _, err := gpgx.Sign(locked, data, passphrase); err != nil {
|
||||
t.Fatalf("错误口令尝试后，正确口令签名失败（密钥环被污染）: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRSAKeyPairRoundTrip 覆盖 RSA 密钥路径（GenerateKeyPair 的 bits > 0 分支）。
|
||||
func TestRSAKeyPairRoundTrip(t *testing.T) {
|
||||
rsaPrivateArmor, rsaPublicArmor, err := gpgx.GenerateKeyPair(
|
||||
"RSA Tester", "rsa@example.com", "gpgx rsa", nil, gpgx.MinRSABits)
|
||||
if err != nil {
|
||||
t.Fatalf("GenerateKeyPair(RSA %d) 失败: %v", gpgx.MinRSABits, err)
|
||||
}
|
||||
|
||||
private, err := gpgx.ReadArmoredKeyRing(bytes.NewReader(rsaPrivateArmor))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadArmoredKeyRing(RSA 私钥) 失败: %v", err)
|
||||
}
|
||||
public, err := gpgx.ReadArmoredKeyRing(bytes.NewReader(rsaPublicArmor))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadArmoredKeyRing(RSA 公钥) 失败: %v", err)
|
||||
}
|
||||
|
||||
data := []byte("rsa payload")
|
||||
ciphertext, err := gpgx.EncryptArmored(public, data)
|
||||
if err != nil {
|
||||
t.Fatalf("EncryptArmored(RSA) 失败: %v", err)
|
||||
}
|
||||
plaintext, err := gpgx.Decrypt(private, ciphertext)
|
||||
if err != nil {
|
||||
t.Fatalf("Decrypt(RSA) 失败: %v", err)
|
||||
}
|
||||
if !bytes.Equal(plaintext, data) {
|
||||
t.Fatal("RSA 加解密结果不匹配")
|
||||
}
|
||||
|
||||
signature, err := gpgx.SignDetachedArmored(private, data, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("SignDetachedArmored(RSA) 失败: %v", err)
|
||||
}
|
||||
if _, err := gpgx.VerifyDetached(public, data, signature); err != nil {
|
||||
t.Fatalf("VerifyDetached(RSA) 失败: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestKeyRingNotMutatedByFailedUnlock 断言解锁失败不会污染调用方传入的密钥环。
|
||||
func TestKeyRingNotMutatedByFailedUnlock(t *testing.T) {
|
||||
initTestKeys(t)
|
||||
requireKeys(t)
|
||||
if len(lockedPrivate) == 0 {
|
||||
t.Fatal("带口令的测试密钥不可用")
|
||||
}
|
||||
|
||||
locked, err := gpgx.ReadArmoredKeyRing(bytes.NewReader(lockedPrivate))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadArmoredKeyRing 失败: %v", err)
|
||||
}
|
||||
if !locked[0].PrivateKey.Encrypted {
|
||||
t.Fatal("测试前提不成立：私钥应为加密状态")
|
||||
}
|
||||
|
||||
// 未提供口令解密必须失败。
|
||||
mustNotPanic(t, "加密私钥无口令解密", func() error {
|
||||
_, err := gpgx.Decrypt(locked, []byte("whatever"))
|
||||
return err
|
||||
})
|
||||
// 失败后调用方密钥环中的私钥仍应保持加密状态。
|
||||
if !locked[0].PrivateKey.Encrypted {
|
||||
t.Fatal("失败的解密尝试修改了调用方密钥环（私钥被解锁）")
|
||||
}
|
||||
|
||||
// 后续用正确口令仍可正常使用。
|
||||
public, err := gpgx.ReadArmoredKeyRing(bytes.NewReader(lockedPublicArmor))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadArmoredKeyRing(公钥) 失败: %v", err)
|
||||
}
|
||||
data := []byte("still usable after failures")
|
||||
ciphertext, err := gpgx.EncryptArmored(public, data)
|
||||
if err != nil {
|
||||
t.Fatalf("EncryptArmored 失败: %v", err)
|
||||
}
|
||||
plaintext, err := gpgx.DecryptWithPassphrase(locked, ciphertext, passphrase)
|
||||
if err != nil {
|
||||
t.Fatalf("DecryptWithPassphrase 失败: %v", err)
|
||||
}
|
||||
if !bytes.Equal(plaintext, data) {
|
||||
t.Fatal("解密结果不匹配")
|
||||
}
|
||||
}
|
||||
|
||||
// TestSymmetricPassphraseRoundTrip 覆盖 gpg -c 风格的口令对称加解密。
|
||||
func TestSymmetricPassphraseRoundTrip(t *testing.T) {
|
||||
initTestKeys(t)
|
||||
requireKeys(t)
|
||||
|
||||
data := []byte("symmetrically encrypted payload 对称加密")
|
||||
ciphertext, err := gpgx.EncryptWithPassphrase(passphrase, data)
|
||||
if err != nil {
|
||||
t.Fatalf("EncryptWithPassphrase 失败: %v", err)
|
||||
}
|
||||
if bytes.HasPrefix(ciphertext, []byte("-----BEGIN")) {
|
||||
t.Fatal("EncryptWithPassphrase 应返回二进制报文")
|
||||
}
|
||||
|
||||
// 对称加密的密文可以被任意密钥环（甚至空密钥环之外的调用）解密，
|
||||
// 只要口令正确。
|
||||
got, err := gpgx.DecryptWithPassphrase(alicePrivate, ciphertext, passphrase)
|
||||
if err != nil {
|
||||
t.Fatalf("DecryptWithPassphrase(对称) 失败: %v", err)
|
||||
}
|
||||
if !bytes.Equal(got, data) {
|
||||
t.Fatal("对称加解密结果不匹配")
|
||||
}
|
||||
|
||||
mustNotPanic(t, "错误口令对称解密", func() error {
|
||||
_, err := gpgx.DecryptWithPassphrase(alicePrivate, ciphertext, []byte("wrong passphrase"))
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "空口令对称解密", func() error {
|
||||
_, err := gpgx.DecryptWithPassphrase(alicePrivate, ciphertext, nil)
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "空口令对称加密", func() error {
|
||||
_, err := gpgx.EncryptWithPassphrase(nil, data)
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "篡改对称密文", func() error {
|
||||
_, err := gpgx.DecryptWithPassphrase(alicePrivate, tamperByte(ciphertext, len(ciphertext)/2), passphrase)
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
func TestGenerateKeyPairUnencryptedWithPassphrase(t *testing.T) {
|
||||
initTestKeys(t)
|
||||
requireKeys(t)
|
||||
|
||||
// 私钥未加密时传入口令应被忽略，而不是报错。
|
||||
data := []byte("unencrypted key with passphrase argument")
|
||||
signature, err := gpgx.SignDetachedArmored(alicePrivate, data, []byte("irrelevant"))
|
||||
if err != nil {
|
||||
t.Fatalf("未加密私钥传入口令时签名失败: %v", err)
|
||||
}
|
||||
if _, err := gpgx.VerifyDetached(alicePublic, data, signature); err != nil {
|
||||
t.Fatalf("验签失败: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadKeyRingInvalidInput(t *testing.T) {
|
||||
inputs := map[string][]byte{
|
||||
"nil 输入": nil,
|
||||
"空输入": {},
|
||||
"垃圾字节": []byte("garbage not a keyring"),
|
||||
"伪 armor": []byte("-----BEGIN PGP PUBLIC KEY BLOCK-----\n\nnot base64!!!\n-----END PGP PUBLIC KEY BLOCK-----\n"),
|
||||
"随机字节": generatePlaintext(512),
|
||||
}
|
||||
|
||||
for name, input := range inputs {
|
||||
input := input
|
||||
mustNotPanic(t, "ReadKeyRing("+name+")", func() error {
|
||||
_, err := gpgx.ReadKeyRing(bytes.NewReader(input))
|
||||
return err
|
||||
})
|
||||
mustNotPanic(t, "ReadArmoredKeyRing("+name+")", func() error {
|
||||
_, err := gpgx.ReadArmoredKeyRing(bytes.NewReader(input))
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
// ReadArmoredKeyRing 应拒绝二进制密钥环。
|
||||
initTestKeys(t)
|
||||
requireKeys(t)
|
||||
mustNotPanic(t, "ReadArmoredKeyRing(二进制密钥环)", func() error {
|
||||
_, err := gpgx.ReadArmoredKeyRing(bytes.NewReader(decodeArmor(t, alicePrivateArmor)))
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
func TestGenerateKeyPairParameterValidation(t *testing.T) {
|
||||
// 弱 RSA 密钥必须被拒绝，而不是生成。
|
||||
mustNotPanic(t, "GenerateKeyPair(1024)", func() error {
|
||||
_, _, err := gpgx.GenerateKeyPair("Weak", "weak@example.com", "", nil, 1024)
|
||||
return err
|
||||
})
|
||||
// bits <= 0：按文档生成 Ed25519（这是合法调用，不应报错）。
|
||||
if _, _, err := gpgx.GenerateKeyPair("Negative", "neg@example.com", "", nil, -100); err != nil {
|
||||
t.Fatalf("GenerateKeyPair(bits=-100) 应生成 Ed25519 密钥，实际报错: %v", err)
|
||||
}
|
||||
// 非法 User ID 字符必须报错而不是 panic。
|
||||
mustNotPanic(t, "GenerateKeyPair(非法 User ID)", func() error {
|
||||
_, _, err := gpgx.GenerateKeyPair("Bad (Name)", "bad@example.com", "", nil, 0)
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
// TestNoGoroutineOrFileLeak 做一个轻量的泄漏哨兵：所有操作都应在有限时间内
|
||||
// 结束（若存在未关闭的管道/句柄导致阻塞，此测试会超时失败）。
|
||||
func TestNoGoroutineOrFileLeak(t *testing.T) {
|
||||
initTestKeys(t)
|
||||
requireKeys(t)
|
||||
|
||||
for i := 0; i < 5; i++ {
|
||||
ciphertext, err := gpgx.EncryptArmored(alicePublic, generatePlaintext(1024))
|
||||
if err != nil {
|
||||
t.Fatalf("第 %d 轮 EncryptArmored 失败: %v", i, err)
|
||||
}
|
||||
if _, err := gpgx.Decrypt(alicePrivate, ciphertext); err != nil {
|
||||
t.Fatalf("第 %d 轮 Decrypt 失败: %v", i, err)
|
||||
}
|
||||
signed, err := gpgx.Sign(alicePrivate, []byte("leak check"), nil)
|
||||
if err != nil {
|
||||
t.Fatalf("第 %d 轮 Sign 失败: %v", i, err)
|
||||
}
|
||||
if _, err := gpgx.Verify(alicePublic, signed); err != nil {
|
||||
t.Fatalf("第 %d 轮 Verify 失败: %v", i, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------- 辅助
|
||||
|
||||
func firstLine(data []byte) string {
|
||||
if idx := bytes.IndexByte(data, '\n'); idx >= 0 {
|
||||
return string(data[:idx])
|
||||
}
|
||||
return string(data)
|
||||
}
|
||||
|
||||
// decodeArmor 把 armored 文本转成底层二进制数据（仅用于测试）。
|
||||
func decodeArmor(t *testing.T, armored []byte) []byte {
|
||||
t.Helper()
|
||||
block, err := armor.Decode(bytes.NewReader(armored))
|
||||
if err != nil {
|
||||
t.Fatalf("解析 armor 失败: %v", err)
|
||||
}
|
||||
raw, err := io.ReadAll(block.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("读取 armor body 失败: %v", err)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
Reference in New Issue
Block a user