logan
88e5f4e7ee
- 新增 crypto.go: AES-256-GCM 加解密，与 Python backend 兼容 - auth.go: Login 添加 LDAP fallback（本地优先→LDAP绑定→自动创建本地用户） - ldap.go: 实现真实 LDAP 连接/用户搜索/密码验证/OU列表/用户列表 - 所有 LDAP 操作解密 admin_password 后使用 - 新增/更新密码存储时加密 - router.go: 补充 delete-user 路由 - 添加 go-ldap/v3 依赖
139 lines
5.0 KiB
Python
139 lines
5.0 KiB
Python
"""Tests for AES encryption (Phase 3.1)."""
|
|
import pytest
|
|
from app.core.crypto import encrypt_password, decrypt_password
|
|
|
|
|
|
class TestEncryption:
|
|
def test_encrypt_decrypt_roundtrip(self):
|
|
"""Encrypt then decrypt returns original."""
|
|
plain = "my_secret_password_123"
|
|
encrypted = encrypt_password(plain)
|
|
assert encrypted != plain
|
|
assert decrypt_password(encrypted) == plain
|
|
|
|
def test_encrypt_produces_different_ciphertext(self):
|
|
"""Each encryption uses a random nonce, so ciphertext differs."""
|
|
plain = "same_password"
|
|
e1 = encrypt_password(plain)
|
|
e2 = encrypt_password(plain)
|
|
assert e1 != e2 # different nonces
|
|
assert decrypt_password(e1) == plain
|
|
assert decrypt_password(e2) == plain
|
|
|
|
def test_encrypt_empty_string(self):
|
|
"""Can encrypt and decrypt empty string."""
|
|
encrypted = encrypt_password("")
|
|
assert decrypt_password(encrypted) == ""
|
|
|
|
def test_encrypt_special_characters(self):
|
|
"""Handles special characters in passwords."""
|
|
plain = 'p@ssw0rd!#$%^&*()_+-=[]{}|;\':",./<>?'
|
|
encrypted = encrypt_password(plain)
|
|
assert decrypt_password(encrypted) == plain
|
|
|
|
def test_encrypt_unicode(self):
|
|
"""Handles unicode characters."""
|
|
plain = "密码测试🔑"
|
|
encrypted = encrypt_password(plain)
|
|
assert decrypt_password(encrypted) == plain
|
|
|
|
def test_decrypt_invalid_data_returns_as_is(self):
|
|
"""Invalid ciphertext returns original string (backward compat)."""
|
|
result = decrypt_password("not_valid_base64!!!")
|
|
assert result == "not_valid_base64!!!"
|
|
|
|
|
|
class TestInstanceEncryption:
|
|
"""Test that instance passwords are stored encrypted via API."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_instance_encrypts_password(self, client, admin_token):
|
|
"""POST /instances stores password as encrypted ciphertext."""
|
|
from tests.conftest import auth_header
|
|
resp = await client.post(
|
|
"/api/v1/instances",
|
|
json={
|
|
"name": "加密测试",
|
|
"db_type": "mysql",
|
|
"host": "10.0.0.1",
|
|
"port": 3306,
|
|
"username": "root",
|
|
"password": "plain_text_pw",
|
|
"database": "test",
|
|
"env": "development",
|
|
},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
# Response should NOT contain password at all (InstanceResponse excludes it)
|
|
assert "password" not in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_password_not_stored_as_plaintext(self, client, db, admin_token):
|
|
"""Verify the database actually stores encrypted, not plain text."""
|
|
from tests.conftest import auth_header
|
|
from app.models.instance import Instance
|
|
from sqlalchemy import select
|
|
|
|
plain_pw = "super_secret_123"
|
|
resp = await client.post(
|
|
"/api/v1/instances",
|
|
json={
|
|
"name": "明文检查",
|
|
"db_type": "mysql",
|
|
"host": "10.0.0.2",
|
|
"port": 3306,
|
|
"username": "root",
|
|
"password": plain_pw,
|
|
"database": "test",
|
|
"env": "development",
|
|
},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 201
|
|
instance_id = resp.json()["id"]
|
|
|
|
# Read raw from database
|
|
result = await db.execute(select(Instance).where(Instance.id == instance_id))
|
|
instance = result.scalar_one()
|
|
assert instance.password != plain_pw # Not stored as plaintext
|
|
assert len(instance.password) > 20 # Encrypted is longer
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_instance_re_encrypts_password(self, client, db, admin_token):
|
|
"""PUT /instances/:id re-encrypts password when updated."""
|
|
from tests.conftest import auth_header
|
|
from app.models.instance import Instance
|
|
from sqlalchemy import select
|
|
|
|
# Create
|
|
resp = await client.post(
|
|
"/api/v1/instances",
|
|
json={
|
|
"name": "更新加密",
|
|
"db_type": "mysql",
|
|
"host": "10.0.0.3",
|
|
"port": 3306,
|
|
"username": "root",
|
|
"password": "old_password",
|
|
"database": "test",
|
|
},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
iid = resp.json()["id"]
|
|
|
|
# Update password
|
|
resp = await client.put(
|
|
f"/api/v1/instances/{iid}",
|
|
json={"password": "new_password"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
# Verify encrypted
|
|
result = await db.execute(select(Instance).where(Instance.id == iid))
|
|
inst = result.scalar_one()
|
|
assert inst.password != "new_password"
|
|
assert inst.password != "old_password"
|