diff --git a/backend/app/api/v1/endpoints/audit.py b/backend/app/api/v1/endpoints/audit.py new file mode 100644 index 0000000..787f591 --- /dev/null +++ b/backend/app/api/v1/endpoints/audit.py @@ -0,0 +1,27 @@ +"""Audit log endpoints.""" +from typing import Optional +from fastapi import APIRouter, Depends, Query +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.database import get_db +from app.core.security import require_role +from app.models.user import User +from app.services.audit_log_service import AuditLogService + +router = APIRouter(prefix="/audit", tags=["审计日志"]) + + +@router.get("/logs") +async def list_audit_logs( + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + user_id: Optional[int] = None, + action: Optional[str] = None, + target_type: Optional[str] = None, + db: AsyncSession = Depends(get_db), + current_user: User = Depends(require_role("admin", "dba")), +): + return await AuditLogService.list_logs( + db, page=page, page_size=page_size, + user_id=user_id, action=action, target_type=target_type, + ) diff --git a/backend/app/api/v1/endpoints/instance.py b/backend/app/api/v1/endpoints/instance.py index aab09b8..dfe6876 100644 --- a/backend/app/api/v1/endpoints/instance.py +++ b/backend/app/api/v1/endpoints/instance.py @@ -10,6 +10,8 @@ from app.models.user import User from app.models.instance import Instance from app.schemas.instance import InstanceCreate, InstanceUpdate, InstanceResponse, ConnectionTestResult from app.services.instance_service import InstanceService +from app.core.crypto import encrypt_password, decrypt_password +from app.services.audit_log_service import AuditLogService router = APIRouter(prefix="/instances", tags=["数据源管理"]) @@ -42,12 +44,18 @@ async def create_instance( host=data.host, port=data.port, username=data.username, - password=data.password, + password=encrypt_password(data.password), database=data.database, env=data.env, description=data.description, ) db.add(instance) + await db.flush() + await AuditLogService.log( + db, user_id=current_user.id, action="CREATE_INSTANCE", + target_type="instance", target_id=instance.id, + detail=f"创建数据源: {data.name} ({data.db_type}@{data.host}:{data.port})", + ) await db.commit() await db.refresh(instance) return instance @@ -79,6 +87,8 @@ async def update_instance( raise HTTPException(status_code=404, detail="数据源不存在") update_data = data.model_dump(exclude_unset=True) + if "password" in update_data: + update_data["password"] = encrypt_password(update_data["password"]) for key, value in update_data.items(): setattr(instance, key, value) await db.commit() @@ -97,6 +107,11 @@ async def delete_instance( if not instance: raise HTTPException(status_code=404, detail="数据源不存在") + await AuditLogService.log( + db, user_id=current_user.id, action="DELETE_INSTANCE", + target_type="instance", target_id=instance.id, + detail=f"删除数据源: {instance.name}", + ) await db.delete(instance) await db.commit() return {"message": "数据源已删除"} @@ -113,7 +128,18 @@ async def test_connection( if not instance: raise HTTPException(status_code=404, detail="数据源不存在") - return await InstanceService.test_connection(instance) + # Decrypt password for connection test + from app.models.instance import Instance as InstModel + decrypted = instance.__class__( + name=instance.name, db_type=instance.db_type, host=instance.host, + port=instance.port, username=instance.username, + password=decrypt_password(instance.password), + database=instance.database, env=instance.env, + description=instance.description, status=instance.status, + version=instance.version, + ) + decrypted.id = instance.id + return await InstanceService.test_connection(decrypted) @router.get("/{instance_id}/databases") diff --git a/backend/app/api/v1/endpoints/query.py b/backend/app/api/v1/endpoints/query.py index cd275fb..57debdb 100644 --- a/backend/app/api/v1/endpoints/query.py +++ b/backend/app/api/v1/endpoints/query.py @@ -9,6 +9,7 @@ from app.models.instance import Instance from app.models.audit_log import AuditLog from app.schemas.query import QueryRequest, QueryResult from app.services.query_service import QueryService +from app.core.crypto import decrypt_password router = APIRouter(prefix="/query", tags=["SQL 查询"]) @@ -27,6 +28,9 @@ async def execute_query( if not instance: raise HTTPException(status_code=404, detail="数据源不存在") + # Decrypt password for query execution + instance.password = decrypt_password(instance.password) + try: query_result = await QueryService.execute_query(instance, req.database, req.sql, current_user.id) except ValueError as e: diff --git a/backend/app/api/v1/endpoints/ticket.py b/backend/app/api/v1/endpoints/ticket.py index 01e60a3..f0363b5 100644 --- a/backend/app/api/v1/endpoints/ticket.py +++ b/backend/app/api/v1/endpoints/ticket.py @@ -11,6 +11,10 @@ from app.models.user import User from app.models.ticket import Ticket from app.models.instance import Instance from app.schemas.ticket import TicketCreate, TicketReview, TicketResponse +from app.core.crypto import decrypt_password +from app.services.audit_engine import SQLAuditEngine +from app.services.audit_log_service import AuditLogService +import json router = APIRouter(prefix="/tickets", tags=["工单管理"]) @@ -104,6 +108,10 @@ async def create_ticket( if not inst.scalar_one_or_none(): raise HTTPException(status_code=400, detail="数据源不存在") + # Auto-audit SQL before creating ticket + engine = SQLAuditEngine(env="production" if data.db_type == "mysql" else "development") + report = engine.audit(data.sql_content) + ticket = Ticket( title=data.title, db_type=data.db_type, @@ -115,11 +123,23 @@ async def create_ticket( priority=data.priority, applicant_id=current_user.id, status="pending", + audit_result=json.dumps(report.to_dict(), ensure_ascii=False) if report.findings else None, ) db.add(ticket) + await db.flush() # flush to get ticket.id for audit log + + # Log ticket creation + await AuditLogService.log( + db, user_id=current_user.id, action="CREATE_TICKET", + target_type="ticket", target_id=ticket.id, + detail=f"SQL审核评分: {report.score}, 通过: {report.passed}", + ) + await db.commit() await db.refresh(ticket) - return await _enrich_ticket(ticket, db) + result = await _enrich_ticket(ticket, db) + result["audit_report"] = report.to_dict() + return result @router.get("/statistics") @@ -184,6 +204,13 @@ async def review_ticket( ticket.approver_id = current_user.id ticket.review_time = datetime.now(timezone.utc) ticket.audit_result = data.comment + + await AuditLogService.log( + db, user_id=current_user.id, action="REVIEW_TICKET", + target_type="ticket", target_id=ticket.id, + detail=f"审核动作: {data.action}, 备注: {data.comment}", + ) + await db.commit() await db.refresh(ticket) return await _enrich_ticket(ticket, db) @@ -213,6 +240,7 @@ async def execute_ticket( raise HTTPException(status_code=400, detail="数据源不存在") from app.services.query_service import QueryService + instance.password = decrypt_password(instance.password) exec_result = await QueryService.execute_dml(instance, ticket.target_database, ticket.sql_content) ticket.status = "success" diff --git a/backend/app/api/v1/endpoints/user.py b/backend/app/api/v1/endpoints/user.py index ed33686..f8e602d 100644 --- a/backend/app/api/v1/endpoints/user.py +++ b/backend/app/api/v1/endpoints/user.py @@ -8,6 +8,7 @@ from app.core.database import get_db from app.core.security import get_current_user, hash_password, require_role from app.models.user import User from app.schemas.user import UserCreate, UserUpdate, UserResponse +from app.services.audit_log_service import AuditLogService router = APIRouter(prefix="/users", tags=["用户管理"]) @@ -47,6 +48,12 @@ async def create_user( hashed_password=hash_password(data.password), ) db.add(user) + await db.flush() + await AuditLogService.log( + db, user_id=current_user.id, action="CREATE_USER", + target_type="user", target_id=user.id, + detail=f"创建用户: {data.username} ({data.role})", + ) await db.commit() await db.refresh(user) return user @@ -99,6 +106,11 @@ async def delete_user( if not user: raise HTTPException(status_code=404, detail="用户不存在") + await AuditLogService.log( + db, user_id=current_user.id, action="DELETE_USER", + target_type="user", target_id=user.id, + detail=f"删除用户: {user.username}", + ) await db.delete(user) await db.commit() return {"message": "用户已删除"} diff --git a/backend/app/api/v1/router.py b/backend/app/api/v1/router.py index f334804..c8bdeb4 100644 --- a/backend/app/api/v1/router.py +++ b/backend/app/api/v1/router.py @@ -1,6 +1,6 @@ from fastapi import APIRouter -from app.api.v1.endpoints import auth, user, instance, ticket, query, dashboard +from app.api.v1.endpoints import auth, user, instance, ticket, query, dashboard, audit api_router = APIRouter(prefix="/api/v1") @@ -10,3 +10,4 @@ api_router.include_router(instance.router) api_router.include_router(ticket.router) api_router.include_router(query.router) api_router.include_router(dashboard.router) +api_router.include_router(audit.router) diff --git a/backend/app/core/crypto.py b/backend/app/core/crypto.py new file mode 100644 index 0000000..8e86aaf --- /dev/null +++ b/backend/app/core/crypto.py @@ -0,0 +1,45 @@ +"""AES-256-GCM encryption for instance passwords.""" +import base64 +import os +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + +from app.core.config import get_settings + +_NONCE_SIZE = 12 # 96-bit nonce for AES-GCM + + +def _get_key() -> bytes: + """Derive a 32-byte AES key from settings.SECRET_KEY.""" + settings = get_settings() + # Use first 32 bytes of utf-8 encoded secret key (padded if needed) + raw = settings.SECRET_KEY.encode("utf-8") + if len(raw) >= 32: + return raw[:32] + return raw.ljust(32, b"\x00") + + +def encrypt_password(plain: str) -> str: + """Encrypt a password string, return base64-encoded ciphertext (nonce + ciphertext).""" + key = _get_key() + nonce = os.urandom(_NONCE_SIZE) + aesgcm = AESGCM(key) + ct = aesgcm.encrypt(nonce, plain.encode("utf-8"), None) + return base64.b64encode(nonce + ct).decode("ascii") + + +def decrypt_password(encrypted: str) -> str: + """Decrypt a base64-encoded encrypted password back to plaintext. + If the value is not valid base64 (e.g., legacy plaintext), return as-is. + """ + try: + key = _get_key() + raw = base64.b64decode(encrypted, validate=True) + if len(raw) <= _NONCE_SIZE: + return encrypted # too short to be encrypted + nonce = raw[:_NONCE_SIZE] + ct = raw[_NONCE_SIZE:] + aesgcm = AESGCM(key) + return aesgcm.decrypt(nonce, ct, None).decode("utf-8") + except Exception: + # Not encrypted (legacy plaintext password) — return as-is + return encrypted diff --git a/backend/app/services/audit_engine.py b/backend/app/services/audit_engine.py new file mode 100644 index 0000000..739cebd --- /dev/null +++ b/backend/app/services/audit_engine.py @@ -0,0 +1,234 @@ +"""SQL Audit Engine — rule-based SQL review for tickets. + +Checks SQL against configurable rules and returns structured results: + - level: "error" (must fix), "warning" (should review), "info" (FYI) + - rule: rule identifier + - message: human-readable description +""" +import re +from dataclasses import dataclass, field +from typing import List, Optional +from enum import Enum + + +class Severity(str, Enum): + ERROR = "error" + WARNING = "warning" + INFO = "info" + + +@dataclass +class AuditFinding: + rule: str + level: Severity + message: str + + +@dataclass +class AuditReport: + sql: str + findings: List[AuditFinding] = field(default_factory=list) + + @property + def passed(self) -> bool: + return not any(f.level == Severity.ERROR for f in self.findings) + + @property + def has_warnings(self) -> bool: + return any(f.level == Severity.WARNING for f in self.findings) + + @property + def score(self) -> int: + """100 = clean, -10 per error, -5 per warning.""" + s = 100 + for f in self.findings: + if f.level == Severity.ERROR: + s -= 10 + elif f.level == Severity.WARNING: + s -= 5 + return max(0, s) + + def to_dict(self) -> dict: + return { + "passed": self.passed, + "score": self.score, + "findings": [ + {"rule": f.rule, "level": f.level.value, "message": f.message} + for f in self.findings + ], + } + + +class SQLAuditEngine: + """Rule-based SQL audit engine.""" + + def __init__(self, env: str = "development"): + self.env = env + + def audit(self, sql: str) -> AuditReport: + report = AuditReport(sql=sql) + clean = sql.strip() + upper = clean.upper() + + # Rule 1: DDL without IF EXISTS + self._check_ddl_if_exists(clean, upper, report) + + # Rule 2: DROP TABLE danger + self._check_drop_table(upper, report) + + # Rule 3: UPDATE/DELETE without WHERE + self._check_missing_where(clean, upper, report) + + # Rule 4: SELECT * usage + self._check_select_star(clean, upper, report) + + # Rule 5: LIMIT on DML + self._check_limit_dml(clean, upper, report) + + # Rule 6: Production env extra checks + if self.env == "production": + self._check_production_rules(clean, upper, report) + + # Rule 7: Multiple statements (semicolon) + self._check_multiple_statements(clean, report) + + # Rule 8: Large IN clause + self._check_large_in(clean, upper, report) + + # Rule 9: No WHERE with LIKE '%...' + self._check_full_like(clean, upper, report) + + # Rule 10: JOIN without ON + self._check_join_without_on(upper, report) + + return report + + def _check_ddl_if_exists(self, clean: str, upper: str, report: AuditReport): + """ALTER/DROP statements should use IF EXISTS where applicable.""" + if re.search(r"\bDROP\s+TABLE\b", upper) and "IF EXISTS" not in upper: + report.findings.append(AuditFinding( + rule="DDL_IF_EXISTS", + level=Severity.WARNING, + message="DROP TABLE 建议添加 IF EXISTS 防止表不存在时报错", + )) + + def _check_drop_table(self, upper: str, report: AuditReport): + """DROP TABLE is always dangerous in production.""" + if re.search(r"\bDROP\s+TABLE\b", upper): + if self.env == "production": + report.findings.append(AuditFinding( + rule="DROP_TABLE_PRODUCTION", + level=Severity.ERROR, + message="生产环境禁止 DROP TABLE 操作", + )) + else: + report.findings.append(AuditFinding( + rule="DROP_TABLE", + level=Severity.WARNING, + message="DROP TABLE 操作不可逆，请确认已备份", + )) + + if re.search(r"\bTRUNCATE\b", upper): + report.findings.append(AuditFinding( + rule="TRUNCATE", + level=Severity.ERROR if self.env == "production" else Severity.WARNING, + message="TRUNCATE 会清空表数据且不可回滚", + )) + + def _check_missing_where(self, clean: str, upper: str, report: AuditReport): + """UPDATE/DELETE without WHERE clause affects all rows.""" + if re.match(r"\s*UPDATE\s+", upper, re.IGNORECASE): + if "WHERE" not in upper: + report.findings.append(AuditFinding( + rule="UPDATE_NO_WHERE", + level=Severity.ERROR, + message="UPDATE 缺少 WHERE 条件，将更新全表", + )) + + if re.match(r"\s*DELETE\s+FROM\s+", upper, re.IGNORECASE): + if "WHERE" not in upper: + report.findings.append(AuditFinding( + rule="DELETE_NO_WHERE", + level=Severity.ERROR, + message="DELETE 缺少 WHERE 条件，将删除全表数据", + )) + + def _check_select_star(self, clean: str, upper: str, report: AuditReport): + """SELECT * is inefficient for large tables.""" + if re.search(r"SELECT\s+\*(?:\s|,|\))", upper): + report.findings.append(AuditFinding( + rule="SELECT_STAR", + level=Severity.INFO, + message="SELECT * 可能返回过多列，建议明确指定字段", + )) + + def _check_limit_dml(self, clean: str, upper: str, report: AuditReport): + """UPDATE/DELETE with LIMIT is MySQL-specific and risky.""" + for kw in ("UPDATE", "DELETE"): + if re.match(rf"\s*{kw}\b", upper): + if "LIMIT" in upper: + report.findings.append(AuditFinding( + rule="DML_WITH_LIMIT", + level=Severity.WARNING, + message=f"{kw} 使用 LIMIT 可能导致部分数据变更，请确认意图", + )) + + def _check_production_rules(self, clean: str, upper: str, report: AuditReport): + """Extra strict rules for production.""" + if re.search(r"\bALTER\s+TABLE\b", upper): + report.findings.append(AuditFinding( + rule="ALTER_TABLE_PRODUCTION", + level=Severity.ERROR, + message="生产环境 ALTER TABLE 需 DBA 特别审批", + )) + + if re.search(r"\bCREATE\s+(TABLE|INDEX)\b", upper): + report.findings.append(AuditFinding( + rule="CREATE_DDL_PRODUCTION", + level=Severity.WARNING, + message="生产环境创建对象需确认影响范围", + )) + + def _check_multiple_statements(self, clean: str, report: AuditReport): + """Multiple SQL statements separated by semicolons.""" + # Count semicolons not at the end + stripped = clean.rstrip(";").strip() + if ";" in stripped: + report.findings.append(AuditFinding( + rule="MULTI_STATEMENT", + level=Severity.ERROR, + message="单次提交不支持多条 SQL 语句（分号分隔），请拆分为多个工单", + )) + + def _check_large_in(self, clean: str, upper: str, report: AuditReport): + """IN clause with many values is slow.""" + in_matches = re.findall(r"\bIN\s*\(([^)]+)\)", upper) + for m in in_matches: + count = m.count(",") + 1 + if count > 100: + report.findings.append(AuditFinding( + rule="LARGE_IN", + level=Severity.WARNING, + message=f"IN 子句包含 {count} 个值，建议使用临时表或 JOIN 替代", + )) + + def _check_full_like(self, clean: str, upper: str, report: AuditReport): + """LIKE '%...' cannot use index.""" + if re.search(r"\bLIKE\s+'%[^']*'", clean, re.IGNORECASE): + report.findings.append(AuditFinding( + rule="FULL_LIKE", + level=Severity.INFO, + message="LIKE '%...' 无法使用索引，大数据量下性能较差", + )) + + def _check_join_without_on(self, upper: str, report: AuditReport): + """JOIN without ON produces cartesian product.""" + # Match JOIN but not followed by ON (within reasonable distance) + if re.search(r"\bJOIN\b", upper) and "ON" not in upper: + # Could be NATURAL JOIN, allow that + if not re.search(r"\bNATURAL\s+JOIN\b", upper): + report.findings.append(AuditFinding( + rule="JOIN_NO_ON", + level=Severity.WARNING, + message="JOIN 缺少 ON 条件，可能产生笛卡尔积", + )) diff --git a/backend/app/services/audit_log_service.py b/backend/app/services/audit_log_service.py new file mode 100644 index 0000000..0e949f2 --- /dev/null +++ b/backend/app/services/audit_log_service.py @@ -0,0 +1,78 @@ +"""Audit log service — centralized logging for all operations.""" +from typing import Optional, List, Dict, Any +from sqlalchemy import select, func +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.audit_log import AuditLog + + +class AuditLogService: + @staticmethod + async def log( + db: AsyncSession, + user_id: int, + action: str, + target_type: str, + target_id: Optional[int] = None, + detail: Optional[str] = None, + ip_address: Optional[str] = None, + ) -> AuditLog: + """Create an audit log entry.""" + entry = AuditLog( + user_id=user_id, + action=action, + target_type=target_type, + target_id=target_id, + detail=detail[:500] if detail else None, + ip_address=ip_address, + ) + db.add(entry) + # Don't commit — let the caller's transaction handle it + await db.flush() + return entry + + @staticmethod + async def list_logs( + db: AsyncSession, + page: int = 1, + page_size: int = 20, + user_id: Optional[int] = None, + action: Optional[str] = None, + target_type: Optional[str] = None, + ) -> Dict[str, Any]: + """List audit logs with pagination and filters.""" + stmt = select(AuditLog) + + if user_id: + stmt = stmt.where(AuditLog.user_id == user_id) + if action: + stmt = stmt.where(AuditLog.action == action) + if target_type: + stmt = stmt.where(AuditLog.target_type == target_type) + + total_stmt = select(func.count()).select_from(stmt.subquery()) + total = (await db.execute(total_stmt)).scalar() + + stmt = stmt.order_by(AuditLog.created_at.desc()) + stmt = stmt.offset((page - 1) * page_size).limit(page_size) + result = await db.execute(stmt) + logs = result.scalars().all() + + return { + "total": total, + "page": page, + "page_size": page_size, + "items": [ + { + "id": log.id, + "user_id": log.user_id, + "action": log.action, + "target_type": log.target_type, + "target_id": log.target_id, + "detail": log.detail, + "ip_address": log.ip_address, + "created_at": log.created_at.isoformat() if log.created_at else None, + } + for log in logs + ], + } diff --git a/backend/tests/test_audit_engine.py b/backend/tests/test_audit_engine.py new file mode 100644 index 0000000..d009b01 --- /dev/null +++ b/backend/tests/test_audit_engine.py @@ -0,0 +1,194 @@ +"""Tests for SQL Audit Engine (Phase 3.2).""" +import pytest +from app.services.audit_engine import SQLAuditEngine, Severity + + +class TestSQLAuditEngine: + def setup_method(self): + self.engine = SQLAuditEngine(env="development") + + # --- DDL Rules --- + + def test_drop_table_without_if_exists(self): + report = self.engine.audit("DROP TABLE users;") + assert not report.passed or report.has_warnings + rules = [f.rule for f in report.findings] + assert "DDL_IF_EXISTS" in rules + assert "DROP_TABLE" in rules + + def test_drop_table_production_blocked(self): + engine = SQLAuditEngine(env="production") + report = engine.audit("DROP TABLE users;") + assert not report.passed + errors = [f for f in report.findings if f.level == Severity.ERROR] + assert any(f.rule == "DROP_TABLE_PRODUCTION" for f in errors) + + def test_truncate_warning(self): + report = self.engine.audit("TRUNCATE TABLE logs;") + rules = [f.rule for f in report.findings] + assert "TRUNCATE" in rules + + def test_truncate_production_error(self): + engine = SQLAuditEngine(env="production") + report = engine.audit("TRUNCATE TABLE logs;") + errors = [f for f in report.findings if f.level == Severity.ERROR] + assert any(f.rule == "TRUNCATE" for f in errors) + + # --- Missing WHERE --- + + def test_update_no_where_error(self): + report = self.engine.audit("UPDATE users SET status = 1;") + assert not report.passed + errors = [f.rule for f in report.findings if f.level == Severity.ERROR] + assert "UPDATE_NO_WHERE" in errors + + def test_delete_no_where_error(self): + report = self.engine.audit("DELETE FROM users;") + assert not report.passed + errors = [f.rule for f in report.findings if f.level == Severity.ERROR] + assert "DELETE_NO_WHERE" in errors + + def test_update_with_where_passes(self): + report = self.engine.audit("UPDATE users SET name = 'test' WHERE id = 1;") + error_rules = [f.rule for f in report.findings if f.level == Severity.ERROR] + assert "UPDATE_NO_WHERE" not in error_rules + + def test_delete_with_where_passes(self): + report = self.engine.audit("DELETE FROM users WHERE id = 1;") + error_rules = [f.rule for f in report.findings if f.level == Severity.ERROR] + assert "DELETE_NO_WHERE" not in error_rules + + # --- SELECT * --- + + def test_select_star_info(self): + report = self.engine.audit("SELECT * FROM users WHERE id = 1;") + rules = [f.rule for f in report.findings] + assert "SELECT_STAR" in rules + star_finding = next(f for f in report.findings if f.rule == "SELECT_STAR") + assert star_finding.level == Severity.INFO + + def test_select_columns_clean(self): + report = self.engine.audit("SELECT id, name FROM users WHERE id = 1;") + rules = [f.rule for f in report.findings] + assert "SELECT_STAR" not in rules + + # --- Multiple statements --- + + def test_multi_statement_error(self): + report = self.engine.audit("UPDATE users SET name = 'a'; DELETE FROM logs;") + assert not report.passed + rules = [f.rule for f in report.findings if f.level == Severity.ERROR] + assert "MULTI_STATEMENT" in rules + + def test_single_statement_ok(self): + report = self.engine.audit("SELECT id FROM users LIMIT 10;") + rules = [f.rule for f in report.findings] + assert "MULTI_STATEMENT" not in rules + + # --- Large IN clause --- + + def test_large_in_warning(self): + values = ", ".join([str(i) for i in range(150)]) + sql = f"SELECT * FROM users WHERE id IN ({values});" + report = self.engine.audit(sql) + rules = [f.rule for f in report.findings] + assert "LARGE_IN" in rules + + def test_small_in_ok(self): + report = self.engine.audit("SELECT * FROM users WHERE id IN (1, 2, 3);") + rules = [f.rule for f in report.findings] + assert "LARGE_IN" not in rules + + # --- Production extra rules --- + + def test_alter_table_production_error(self): + engine = SQLAuditEngine(env="production") + report = engine.audit("ALTER TABLE users ADD COLUMN age INT;") + assert not report.passed + errors = [f.rule for f in report.findings if f.level == Severity.ERROR] + assert "ALTER_TABLE_PRODUCTION" in errors + + def test_alter_table_dev_ok(self): + report = self.engine.audit("ALTER TABLE users ADD COLUMN age INT;") + error_rules = [f.rule for f in report.findings if f.level == Severity.ERROR] + assert "ALTER_TABLE_PRODUCTION" not in error_rules + + # --- JOIN without ON --- + + def test_join_no_on_warning(self): + report = self.engine.audit("SELECT * FROM users JOIN orders;") + rules = [f.rule for f in report.findings] + assert "JOIN_NO_ON" in rules + + def test_join_with_on_ok(self): + report = self.engine.audit("SELECT * FROM users JOIN orders ON users.id = orders.user_id;") + rules = [f.rule for f in report.findings] + assert "JOIN_NO_ON" not in rules + + # --- Score --- + + def test_clean_sql_score_100(self): + report = self.engine.audit("SELECT id, name FROM users WHERE id = 1 LIMIT 10;") + assert report.score == 100 + assert report.passed + + def test_error_reduces_score(self): + report = self.engine.audit("UPDATE users SET name = 'test';") + assert report.score < 100 + assert not report.passed + + def test_report_to_dict(self): + report = self.engine.audit("SELECT * FROM users;") + d = report.to_dict() + assert "passed" in d + assert "score" in d + assert "findings" in d + assert isinstance(d["findings"], list) + + +class TestAuditEngineIntegration: + """Test audit engine integration with ticket creation via API.""" + + @pytest.mark.asyncio + async def test_ticket_creation_returns_audit_report(self, client, db, admin_token, test_instance): + from tests.conftest import auth_header + resp = await client.post( + "/api/v1/tickets", + json={ + "title": "SQL审核测试工单", + "db_type": "mysql", + "instance_id": test_instance.id, + "target_database": "test_db", + "sql_content": "SELECT * FROM users WHERE id = 1;", + "reason": "测试审核引擎", + "priority": "medium", + }, + headers=auth_header(admin_token), + ) + assert resp.status_code == 201 + data = resp.json() + assert "audit_report" in data + assert "score" in data["audit_report"] + assert "findings" in data["audit_report"] + + @pytest.mark.asyncio + async def test_ticket_with_dangerous_sql_has_low_score(self, client, db, admin_token, test_instance): + from tests.conftest import auth_header + resp = await client.post( + "/api/v1/tickets", + json={ + "title": "危险SQL工单", + "db_type": "mysql", + "instance_id": test_instance.id, + "target_database": "test_db", + "sql_content": "UPDATE users SET status = 1;", + "reason": "全表更新测试", + "priority": "high", + }, + headers=auth_header(admin_token), + ) + assert resp.status_code == 201 + data = resp.json() + report = data["audit_report"] + assert report["passed"] is False + assert report["score"] < 100 diff --git a/backend/tests/test_audit_logs.py b/backend/tests/test_audit_logs.py new file mode 100644 index 0000000..e8039b7 --- /dev/null +++ b/backend/tests/test_audit_logs.py @@ -0,0 +1,190 @@ +"""Tests for Audit Log API (Phase 3.4).""" +import pytest +from tests.conftest import auth_header + + +class TestAuditLogAPI: + @pytest.mark.asyncio + async def test_list_audit_logs_admin(self, client, admin_token): + """Admin can list audit logs.""" + resp = await client.get( + "/api/v1/audit/logs", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + data = resp.json() + assert "items" in data + assert "total" in data + assert "page" in data + + @pytest.mark.asyncio + async def test_list_audit_logs_dba(self, client, dba_token): + """DBA can also list audit logs.""" + resp = await client.get( + "/api/v1/audit/logs", + headers=auth_header(dba_token), + ) + assert resp.status_code == 200 + + @pytest.mark.asyncio + async def test_list_audit_logs_developer_forbidden(self, client, developer_token): + """Developer cannot access audit logs.""" + resp = await client.get( + "/api/v1/audit/logs", + headers=auth_header(developer_token), + ) + assert resp.status_code == 403 + + @pytest.mark.asyncio + async def test_create_instance_logs_audit(self, client, db, admin_token): + """Creating an instance produces an audit log entry.""" + from app.models.audit_log import AuditLog + from sqlalchemy import select + + resp = await client.post( + "/api/v1/instances", + json={ + "name": "审计测试实例", + "db_type": "mysql", + "host": "10.0.0.100", + "port": 3306, + "username": "root", + "password": "test123", + "database": "test", + }, + headers=auth_header(admin_token), + ) + assert resp.status_code == 201 + + # Check audit log was created + result = await db.execute( + select(AuditLog).where(AuditLog.action == "CREATE_INSTANCE") + ) + log = result.scalar_one_or_none() + assert log is not None + assert "审计测试实例" in log.detail + + @pytest.mark.asyncio + async def test_create_user_logs_audit(self, client, db, admin_token): + """Creating a user produces an audit log entry.""" + from app.models.audit_log import AuditLog + from sqlalchemy import select + + resp = await client.post( + "/api/v1/users", + json={ + "username": "audit_test_user", + "display_name": "审计测试", + "password": "123456", + "role": "developer", + }, + headers=auth_header(admin_token), + ) + assert resp.status_code == 201 + + result = await db.execute( + select(AuditLog).where(AuditLog.action == "CREATE_USER") + ) + log = result.scalar_one_or_none() + assert log is not None + assert "audit_test_user" in log.detail + + @pytest.mark.asyncio + async def test_create_ticket_logs_audit(self, client, db, admin_token, test_instance): + """Creating a ticket produces an audit log entry with SQL score.""" + from app.models.audit_log import AuditLog + from sqlalchemy import select + + resp = await client.post( + "/api/v1/tickets", + json={ + "title": "审计日志工单测试", + "db_type": "mysql", + "instance_id": test_instance.id, + "target_database": "test_db", + "sql_content": "SELECT id FROM users LIMIT 10;", + "reason": "测试审计日志", + }, + headers=auth_header(admin_token), + ) + assert resp.status_code == 201 + + result = await db.execute( + select(AuditLog).where(AuditLog.action == "CREATE_TICKET") + ) + log = result.scalar_one_or_none() + assert log is not None + assert "SQL审核评分" in log.detail + + @pytest.mark.asyncio + async def test_audit_log_pagination(self, client, admin_token): + """Audit log pagination works.""" + resp = await client.get( + "/api/v1/audit/logs?page=1&page_size=5", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + data = resp.json() + assert data["page_size"] == 5 + + @pytest.mark.asyncio + async def test_audit_log_filter_by_action(self, client, admin_token, db, test_instance): + """Can filter audit logs by action.""" + # Create a ticket first + await client.post( + "/api/v1/tickets", + json={ + "title": "过滤测试工单", + "db_type": "mysql", + "instance_id": test_instance.id, + "target_database": "test_db", + "sql_content": "SELECT 1;", + "reason": "过滤测试用", + }, + headers=auth_header(admin_token), + ) + + resp = await client.get( + "/api/v1/audit/logs?action=CREATE_TICKET", + headers=auth_header(admin_token), + ) + assert resp.status_code == 200 + data = resp.json() + for item in data["items"]: + assert item["action"] == "CREATE_TICKET" + + @pytest.mark.asyncio + async def test_review_ticket_logs_audit(self, client, db, admin_token, dba_token, test_instance): + """Reviewing a ticket creates an audit log.""" + from app.models.audit_log import AuditLog + from sqlalchemy import select + + # Create ticket as admin + resp = await client.post( + "/api/v1/tickets", + json={ + "title": "审核日志测试", + "db_type": "mysql", + "instance_id": test_instance.id, + "target_database": "test_db", + "sql_content": "SELECT 1;", + "reason": "审核日志测试", + }, + headers=auth_header(admin_token), + ) + ticket_id = resp.json()["id"] + + # Review as DBA + resp = await client.post( + f"/api/v1/tickets/{ticket_id}/review", + json={"action": "approve", "comment": "批准"}, + headers=auth_header(dba_token), + ) + assert resp.status_code == 200 + + result = await db.execute( + select(AuditLog).where(AuditLog.action == "REVIEW_TICKET") + ) + log = result.scalar_one_or_none() + assert log is not None + assert "approve" in log.detail diff --git a/backend/tests/test_crypto.py b/backend/tests/test_crypto.py new file mode 100644 index 0000000..a289c2b --- /dev/null +++ b/backend/tests/test_crypto.py @@ -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" diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 0835be7..5f757c0 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -15,6 +15,7 @@ import WorkflowPage from '@/pages/WorkflowPage'; import AutoTaskPage from '@/pages/AutoTaskPage'; import AnnouncementPage from '@/pages/AnnouncementPage'; import ConfigPlatformPage from '@/pages/ConfigPlatformPage'; +import AuditLogsPage from '@/pages/AuditLogsPage'; // 路由守卫：未登录跳转登录页，无权限跳转仪表盘 function RouteGuard() { @@ -60,6 +61,7 @@ function AppRoutes() { } /> } /> } /> + } /> } /> } /> } /> diff --git a/frontend/src/hooks/useAuth.tsx b/frontend/src/hooks/useAuth.tsx index 51e79ef..74a33fb 100644 --- a/frontend/src/hooks/useAuth.tsx +++ b/frontend/src/hooks/useAuth.tsx @@ -46,6 +46,7 @@ const routePermissionMap: Record = { users: ['admin'], 'data-permission': ['admin', 'dba'], 'audit-rules': ['admin', 'dba'], + 'audit-logs': ['admin', 'dba'], workflow: ['admin', 'dba'], 'auto-task': ['admin'], announcement: ['admin'], diff --git a/frontend/src/lib/mock.ts b/frontend/src/lib/mock.ts index eaac1fa..19f2492 100644 --- a/frontend/src/lib/mock.ts +++ b/frontend/src/lib/mock.ts @@ -32,6 +32,7 @@ export const menuItems = [ { id: 'users', label: '用户管理', icon: 'Users', section: 'permission', roles: ['admin'] }, { id: 'data-permission', label: '数据权限', icon: 'Lock', section: 'permission', roles: ['admin', 'dba'] }, { id: 'audit-rules', label: '审核规则', icon: 'Shield', section: 'system', roles: ['admin', 'dba'] }, + { id: 'audit-logs', label: '审计日志', icon: 'FileText', section: 'system', roles: ['admin', 'dba'] }, { id: 'workflow', label: '审批流程', icon: 'GitBranch', section: 'system', roles: ['admin', 'dba'] }, { id: 'auto-task', label: '自动任务', icon: 'Clock', section: 'system', roles: ['admin'] }, { id: 'announcement', label: '公告管理', icon: 'Megaphone', section: 'system', roles: ['admin'] }, @@ -165,8 +166,8 @@ export const notificationChannels: NotificationChannel[] = [ ]; export const roles: Role[] = [ - { id: 'admin', name: '超级管理员', description: '拥有系统的所有权限，包括用户管理、系统配置等', icon: 'Crown', iconClass: 'text-purple-400 bg-purple-400/10', userCount: 2, permissionCount: 56, permissions: ['dashboard', 'mysql', 'postgres', 'redis', 'mongo', 'tickets', 'queries', 'roles', 'users', 'data-permission', 'audit-rules', 'workflow', 'auto-task', 'announcement', 'config-platform', 'notifications'] }, - { id: 'dba', name: '数据库管理员', description: '负责数据库运维、工单审批和 SQL 审核', icon: 'Database', iconClass: 'text-blue-400 bg-blue-400/10', userCount: 5, permissionCount: 32, permissions: ['dashboard', 'mysql', 'postgres', 'redis', 'tickets', 'queries', 'data-permission', 'audit-rules', 'workflow'] }, + { id: 'admin', name: '超级管理员', description: '拥有系统的所有权限，包括用户管理、系统配置等', icon: 'Crown', iconClass: 'text-purple-400 bg-purple-400/10', userCount: 2, permissionCount: 56, permissions: ['dashboard', 'mysql', 'postgres', 'redis', 'mongo', 'tickets', 'queries', 'roles', 'users', 'data-permission', 'audit-rules', 'audit-logs', 'workflow', 'auto-task', 'announcement', 'config-platform', 'notifications'] }, + { id: 'dba', name: '数据库管理员', description: '负责数据库运维、工单审批和 SQL 审核', icon: 'Database', iconClass: 'text-blue-400 bg-blue-400/10', userCount: 5, permissionCount: 32, permissions: ['dashboard', 'mysql', 'postgres', 'redis', 'tickets', 'queries', 'data-permission', 'audit-rules', 'audit-logs', 'workflow'] }, { id: 'developer', name: '开发人员', description: '可以提交变更工单和执行数据查询', icon: 'Code', iconClass: 'text-green-400 bg-green-400/10', userCount: 28, permissionCount: 12, permissions: ['dashboard', 'tickets', 'queries'] }, { id: 'guest', name: '访客', description: '仅可查看，无法执行任何操作', icon: 'Eye', iconClass: 'text-gray-400 bg-gray-400/10', userCount: 8, permissionCount: 4, permissions: ['dashboard', 'queries'] }, ]; diff --git a/frontend/src/pages/AuditLogsPage.tsx b/frontend/src/pages/AuditLogsPage.tsx new file mode 100644 index 0000000..bf12bb5 --- /dev/null +++ b/frontend/src/pages/AuditLogsPage.tsx @@ -0,0 +1,159 @@ +import { useState, useEffect } from 'react'; +import request from '../api/request'; + +interface AuditLog { + id: number; + user_id: number; + action: string; + target_type: string; + target_id: number | null; + detail: string | null; + ip_address: string | null; + created_at: string | null; +} + +interface AuditLogResponse { + total: number; + page: number; + page_size: number; + items: AuditLog[]; +} + +const ACTION_COLORS: Record = { + CREATE_TICKET: 'bg-blue-100 text-blue-800', + REVIEW_TICKET: 'bg-yellow-100 text-yellow-800', + EXECUTE_TICKET: 'bg-green-100 text-green-800', + CREATE_INSTANCE: 'bg-purple-100 text-purple-800', + DELETE_INSTANCE: 'bg-red-100 text-red-800', + CREATE_USER: 'bg-indigo-100 text-indigo-800', + DELETE_USER: 'bg-red-100 text-red-800', + QUERY: 'bg-cyan-100 text-cyan-800', +}; + +const TARGET_ICONS: Record = { + ticket: '📋', + instance: '🗄️', + user: '👤', + query: '🔍', +}; + +export default function AuditLogsPage() { + const [logs, setLogs] = useState([]); + const [total, setTotal] = useState(0); + const [page, setPage] = useState(1); + const [loading, setLoading] = useState(true); + const [filterAction, setFilterAction] = useState(''); + const pageSize = 20; + + useEffect(() => { + fetchLogs(); + }, [page, filterAction]); + + const fetchLogs = async () => { + setLoading(true); + try { + const params = new URLSearchParams({ page: String(page), page_size: String(pageSize) }); + if (filterAction) params.set('action', filterAction); + const res = await request.get(`/audit/logs?${params}`); + const data = res.data as AuditLogResponse; + setLogs(data.items || []); + setTotal(data.total); + } catch (e) { + console.error('获取审计日志失败', e); + } finally { + setLoading(false); + } + }; + + const totalPages = Math.ceil(total / pageSize); + + const formatDate = (d: string | null) => { + if (!d) return '-'; + return new Date(d).toLocaleString('zh-CN'); + }; + + return ( +
+
+

审计日志

+ 共 {total} 条记录 +
+ + {/* Filters */} +
+ {['', 'CREATE_TICKET', 'REVIEW_TICKET', 'EXECUTE_TICKET', 'CREATE_INSTANCE', 'DELETE_INSTANCE', 'QUERY'].map(action => ( + + ))} +
+ + {/* Table */} +
+ + + + + + + + + + + {loading ? ( + + ) : logs.length === 0 ? ( + + ) : ( + logs.map(log => ( + + + + + + + )) + )} + +
操作目标详情时间
加载中...
暂无审计日志
+ + {log.action} + + + {TARGET_ICONS[log.target_type] || '📌'} + {log.target_type} + {log.target_id && #{log.target_id}} + {log.detail || '-'}{formatDate(log.created_at)}
+
+ + {/* Pagination */} + {totalPages > 1 && ( +
+ + {page} / {totalPages} + +
+ )} +
+ ); +}