118 lines
3.6 KiB
Python
118 lines
3.6 KiB
Python
"""kilostar.utils.crypto 加解密模块测试。"""
|
|
|
|
import os
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from cryptography.fernet import Fernet
|
|
|
|
from kilostar.utils.crypto import (
|
|
CryptoError,
|
|
decrypt_dict_secrets,
|
|
decrypt_secret,
|
|
encrypt_dict_secrets,
|
|
encrypt_secret,
|
|
is_encrypted,
|
|
_is_sensitive_key,
|
|
_get_fernet,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _set_secret_key(monkeypatch):
|
|
key = Fernet.generate_key().decode()
|
|
monkeypatch.setenv("KILOSTAR_SECRET_KEY", key)
|
|
_get_fernet.cache_clear()
|
|
yield
|
|
_get_fernet.cache_clear()
|
|
|
|
|
|
class TestEncryptDecrypt:
|
|
def test_round_trip(self):
|
|
plain = "my-secret-token-12345"
|
|
cipher = encrypt_secret(plain)
|
|
assert cipher.startswith("v1:")
|
|
assert decrypt_secret(cipher) == plain
|
|
|
|
def test_empty_string_passthrough(self):
|
|
assert encrypt_secret("") == ""
|
|
assert decrypt_secret("") == ""
|
|
|
|
def test_non_encrypted_passthrough(self):
|
|
assert decrypt_secret("plain-text") == "plain-text"
|
|
|
|
def test_is_encrypted(self):
|
|
cipher = encrypt_secret("hello")
|
|
assert is_encrypted(cipher) is True
|
|
assert is_encrypted("hello") is False
|
|
assert is_encrypted("") is False
|
|
|
|
|
|
class TestDictSecrets:
|
|
def test_encrypt_dict_secrets_targets_sensitive_keys(self):
|
|
data = {"api_key": "abc123", "url": "https://x"}
|
|
encrypted = encrypt_dict_secrets(data)
|
|
assert is_encrypted(encrypted["api_key"])
|
|
assert encrypted["url"] == "https://x"
|
|
|
|
def test_decrypt_dict_secrets_round_trip(self):
|
|
data = {"token": "secret", "name": "foo"}
|
|
encrypted = encrypt_dict_secrets(data)
|
|
decrypted = decrypt_dict_secrets(encrypted)
|
|
assert decrypted == data
|
|
|
|
def test_already_encrypted_not_double_encrypted(self):
|
|
data = {"api_key": "abc123"}
|
|
enc1 = encrypt_dict_secrets(data)
|
|
enc2 = encrypt_dict_secrets(enc1)
|
|
assert enc1["api_key"] == enc2["api_key"]
|
|
|
|
def test_non_dict_passthrough(self):
|
|
assert encrypt_dict_secrets("not a dict") == "not a dict"
|
|
assert decrypt_dict_secrets(42) == 42
|
|
|
|
|
|
class TestSensitiveKeyDetection:
|
|
@pytest.mark.parametrize(
|
|
"key,expected",
|
|
[
|
|
("api_key", True),
|
|
("API_KEY", True),
|
|
("provider_apikey", True),
|
|
("token", True),
|
|
("access_token", True),
|
|
("secret", True),
|
|
("password", True),
|
|
("db_password", True),
|
|
("url", False),
|
|
("name", False),
|
|
("transport", False),
|
|
],
|
|
)
|
|
def test_is_sensitive_key(self, key, expected):
|
|
assert _is_sensitive_key(key) is expected
|
|
|
|
|
|
class TestMissingKey:
|
|
def test_raises_when_key_not_set(self, monkeypatch):
|
|
monkeypatch.delenv("KILOSTAR_SECRET_KEY", raising=False)
|
|
_get_fernet.cache_clear()
|
|
with pytest.raises(CryptoError, match="KILOSTAR_SECRET_KEY"):
|
|
encrypt_secret("hello")
|
|
|
|
def test_raises_on_invalid_key(self, monkeypatch):
|
|
monkeypatch.setenv("KILOSTAR_SECRET_KEY", "not-a-valid-fernet-key")
|
|
_get_fernet.cache_clear()
|
|
with pytest.raises(CryptoError, match="格式无效"):
|
|
encrypt_secret("hello")
|
|
|
|
|
|
class TestDecryptWithWrongKey:
|
|
def test_wrong_key_raises(self, monkeypatch):
|
|
cipher = encrypt_secret("hello")
|
|
new_key = Fernet.generate_key().decode()
|
|
monkeypatch.setenv("KILOSTAR_SECRET_KEY", new_key)
|
|
_get_fernet.cache_clear()
|
|
with pytest.raises(CryptoError, match="解密失败"):
|
|
decrypt_secret(cipher)
|