feat: Phase 3 - AES encryption, SQL audit engine, audit logs (85 tests)

Phase 3.1: Instance password AES-256-GCM encryption
Phase 3.2: SQL audit rule engine (10 rules)
Phase 3.3: Auto-audit on ticket creation
Phase 3.4: Operation audit logs + GET /audit/logs
Tests: 44 -> 85 (41 new), Frontend: AuditLogsPage
This commit is contained in:
logan committed 2026-05-22 13:08:55 +08:00
1 parent 27989274f5
commit 66b09e4d85
16 files changed
+1146 -6

No files matched your search

+138
View File
@@ -0,0 +1,138 @@
"""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"