diff --git a/backend/app/api/v1/endpoints/ticket.py b/backend/app/api/v1/endpoints/ticket.py index be1438c..11912f9 100644 --- a/backend/app/api/v1/endpoints/ticket.py +++ b/backend/app/api/v1/endpoints/ticket.py @@ -56,6 +56,129 @@ def _generate_ticket_no(ticket_id: int) -> str: seq = str(ticket_id).zfill(3)[-3:] # ID 补齐到3位 return f"{base}{ms}{seq}" + +def _classify_sql_type(sql: str) -> str: + """根据单条 SQL 语句判断类型。""" + s = sql.strip().upper() + if s.startswith(("SELECT", "SHOW", "EXPLAIN", "DESC", "DESCRIBE")): + return "DQL" + if s.startswith(("INSERT", "UPDATE", "DELETE", "REPLACE")): + return "DML" + if s.startswith(("CREATE", "ALTER", "DROP", "RENAME", "TRUNCATE")): + return "DDL" + if s.startswith(("GRANT", "REVOKE")): + return "DCL" + if s.startswith(("SET", "GET", "HSET", "HMSET", "LPUSH", "RPUSH", "SADD", "ZADD", "DEL", "FLUSHDB", "FLUSHALL")): + return "SET" + return "OTHER" + + +async def _update_sql_items_from_report(db, ticket_id: int, report) -> None: + """根据执行报告更新 SQL items 状态。""" + from app.models.ticket_sql_item import TicketSqlItem + + result = await db.execute( + select(TicketSqlItem) + .where(TicketSqlItem.ticket_id == ticket_id) + .order_by(TicketSqlItem.seq) + ) + items = result.scalars().all() + if not items: + return + + if report.status == "success": + for item in items: + item.status = "success" + item.finished_at = datetime.now(timezone.utc) + else: + # 尝试从 raw_output 解析逐条结果 + exec_errors = getattr(report, "errors", []) + error_stmt = "" + if exec_errors: + err_msg = str(exec_errors[0]) if exec_errors else "" + # 找到出错的语句 + for item in items: + if item.status == "pending": + if not error_stmt: + item.status = "failed" + item.error_message = err_msg + item.finished_at = datetime.now(timezone.utc) + error_stmt = item.sql_text[:50] + else: + item.status = "skipped" + + await db.flush() + + +async def _mark_sql_items_failed(db, ticket_id: int, error_msg: str) -> None: + """标记所有 pending 的 SQL items 为 failed。""" + from app.models.ticket_sql_item import TicketSqlItem + + result = await db.execute( + select(TicketSqlItem) + .where( + TicketSqlItem.ticket_id == ticket_id, + TicketSqlItem.status == "pending", + ) + ) + items = result.scalars().all() + for item in items: + item.status = "failed" + item.error_message = error_msg[:500] + item.finished_at = datetime.now(timezone.utc) + await db.flush() + + +async def _check_maintenance_window(db) -> Optional[str]: + """检查当前是否在维护窗口内。 + + Returns: + None = 允许执行（无窗口限制或在窗口内） + str = 窗口描述（表示不在窗口内，不允许执行） + """ + from app.models.system_setting import SystemSetting + result = await db.execute( + select(SystemSetting).where(SystemSetting.key == "maintenance_window") + ) + setting = result.scalar_one_or_none() + if not setting or not setting.value: + return None # 无配置，不限制 + + import json as _json + try: + window = _json.loads(setting.value) if isinstance(setting.value, str) else setting.value + except Exception: + return None + + enabled = window.get("enabled", False) + if not enabled: + return None + + # 格式: {"enabled": true, "start": "22:00", "end": "06:00", "days": ["mon","tue","wed","thu","fri"]} + start_time = window.get("start", "22:00") + end_time = window.get("end", "06:00") + days = window.get("days", []) # 空=每天 + + now = datetime.now() + current_day = now.strftime("%a").lower()[:3] + current_time = now.strftime("%H:%M") + + # 检查星期 + if days and current_day not in [d.lower()[:3] for d in days]: + return f"{start_time}-{end_time} ({'/'.join(days)})" + + # 检查时间（支持跨午夜） + if start_time <= end_time: + if not (start_time <= current_time <= end_time): + return f"{start_time}-{end_time}" + else: + # 跨午夜: 22:00-06:00 意味着 22:00~24:00 或 00:00~06:00 + if not (current_time >= start_time or current_time <= end_time): + return f"{start_time}-{end_time}（次日）" + + return None # 在窗口内，允许执行 + + if TYPE_CHECKING: from app.services.ticket_executor import ExecutionReport @@ -144,6 +267,8 @@ async def _enrich_ticket(ticket: Ticket, db: AsyncSession) -> Dict[str, Any]: "rollback_sql": ticket.rollback_sql, "status": ticket.status, "priority": ticket.priority, + "risk_level": getattr(ticket, "risk_level", "low"), + "workflow_id": getattr(ticket, "workflow_id", None), "applicant_id": ticket.applicant_id, "approver_id": ticket.approver_id, "reason": ticket.reason, @@ -373,27 +498,57 @@ async def create_ticket( # 3. 验证工单内容安全性 _validate_ticket_content(data) - # 4. 自动审核 SQL + # 4. 自动审核 SQL（带超时保护） env = "production" if instance.env == "production" else "development" v2_engine = SQLAuditEngineV2(db_session=db, env=env) + import asyncio as _asyncio try: - v2_report = await v2_engine.audit(data.sql_content, db_type=data.db_type) + v2_report = await _asyncio.wait_for( + v2_engine.audit(data.sql_content, db_type=data.db_type, instance_id=data.instance_id), + timeout=10.0, + ) + except _asyncio.TimeoutError: + logger.warning("SQL 审核超时(10s)，跳过审核: ticket title=%s", data.title) + v2_report = None except Exception as e: logger.error("SQL 审核失败: %s", e) v2_report = None - # 5. 检查是否有默认审批流 - default_wf_result = await db.execute( - select(WorkflowDefinition).where( - WorkflowDefinition.is_default == True, - WorkflowDefinition.is_active == True, - ) + # 5. 计算风险等级 + from app.services.risk_evaluator import evaluate_risk_level + audit_findings_raw = v2_report.to_dict().get("findings", []) if v2_report else [] + import sqlparse as _sqlparse + stmt_list = [s.strip() for s in _sqlparse.split(data.sql_content) if s.strip()] + risk_level = evaluate_risk_level( + sql_content=data.sql_content, + db_type=data.db_type, + sql_type=data.sql_type or "", + audit_findings=audit_findings_raw, + statement_count=len(stmt_list), ) - default_workflow = default_wf_result.scalar_one_or_none() - initial_status = TicketStatus.REVIEWING.value if default_workflow else TicketStatus.PENDING.value + # 6. 匹配审批流 — 优先按 instance.workflow_id，其次按 risk_level 匹配默认流 + workflow = None + if instance.workflow_id: + wf_result = await db.execute( + select(WorkflowDefinition).where( + WorkflowDefinition.id == instance.workflow_id, + WorkflowDefinition.is_active == True, + ) + ) + workflow = wf_result.scalar_one_or_none() + if not workflow: + default_wf_result = await db.execute( + select(WorkflowDefinition).where( + WorkflowDefinition.is_default == True, + WorkflowDefinition.is_active == True, + ) + ) + workflow = default_wf_result.scalar_one_or_none() - # 6. 创建工单 + initial_status = TicketStatus.REVIEWING.value if workflow else TicketStatus.PENDING.value + + # 7. 创建工单 ticket = Ticket( title=data.title, db_type=data.db_type, @@ -403,6 +558,8 @@ async def create_ticket( rollback_sql=data.rollback_sql, reason=data.reason, priority=data.priority, + risk_level=risk_level, + workflow_id=workflow.id if workflow else None, applicant_id=current_user.id, status=initial_status, audit_result=json.dumps(v2_report.to_dict(), ensure_ascii=False) if v2_report and v2_report.findings else None, @@ -416,10 +573,10 @@ async def create_ticket( await db.flush() # 7. 创建审批步骤（多级审批流） - if default_workflow: + if workflow: wf_steps_result = await db.execute( select(WorkflowStep) - .where(WorkflowStep.workflow_id == default_workflow.id) + .where(WorkflowStep.workflow_id == workflow.id) .order_by(WorkflowStep.step_number) ) wf_steps = wf_steps_result.scalars().all() @@ -434,7 +591,21 @@ async def create_ticket( db.add(approval) await db.flush() - # 8. 记录审计日志 + # 8. 创建 SQL 逐条执行记录 + from app.models.ticket_sql_item import TicketSqlItem + for i, stmt in enumerate(stmt_list, start=1): + stmt_type = _classify_sql_type(stmt) + sql_item = TicketSqlItem( + ticket_id=ticket.id, + seq=i, + sql_text=stmt, + sql_type=stmt_type, + status="pending", + ) + db.add(sql_item) + await db.flush() + + # 9. 记录审计日志 audit_detail = "" if v2_report: audit_detail = f"SQL审核评分: {v2_report.score}, 通过: {v2_report.passed}, 规则数: {len(v2_report.findings)}" @@ -776,6 +947,51 @@ async def get_ticket_timeline( return {"ticket_id": ticket_id, "timeline": timeline} +@router.get("/{ticket_id}/sql-items") +async def get_ticket_sql_items( + ticket_id: int, + db: AsyncSession = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """获取工单 SQL 逐条执行进度。""" + from app.models.ticket_sql_item import TicketSqlItem + + result = await db.execute( + select(TicketSqlItem) + .where(TicketSqlItem.ticket_id == ticket_id) + .order_by(TicketSqlItem.seq) + ) + items = result.scalars().all() + + return { + "ticket_id": ticket_id, + "total": len(items), + "items": [ + { + "id": item.id, + "seq": item.seq, + "sql_text": item.sql_text, + "sql_type": item.sql_type, + "status": item.status, + "affected_rows": item.affected_rows, + "duration_ms": item.duration_ms, + "error_message": item.error_message, + "started_at": item.started_at.isoformat() if item.started_at else None, + "finished_at": item.finished_at.isoformat() if item.finished_at else None, + } + for item in items + ], + "progress": { + "total": len(items), + "success": sum(1 for i in items if i.status == "success"), + "failed": sum(1 for i in items if i.status == "failed"), + "running": sum(1 for i in items if i.status == "running"), + "pending": sum(1 for i in items if i.status == "pending"), + "skipped": sum(1 for i in items if i.status == "skipped"), + }, + } + + @router.post("/{ticket_id}/cancel") async def cancel_ticket( ticket_id: int, @@ -910,12 +1126,32 @@ async def execute_ticket( if confirm and not confirm.confirmed: raise HTTPException(status_code=400, detail="未确认执行，操作已取消") + # 3.1 高危操作强制二次确认 + risk_level = getattr(ticket, "risk_level", "low") + if risk_level in ("high", "critical"): + if not confirm or not confirm.confirmed: + from app.services.risk_evaluator import RISK_LEVEL_CONFIG + risk_cfg = RISK_LEVEL_CONFIG.get(risk_level, {}) + raise HTTPException( + status_code=409, + detail=f"该工单风险等级为「{risk_cfg.get('label', risk_level)}」，需要二次确认后才能执行。请传入 confirmed=true", + ) + # 4. 查询数据源实例 instance = await _get_instance(db, ticket.instance_id) # 5. 权限检查 await check_instance_permission(current_user, instance, "ticket:execute", db) + # 5.1 维护窗口检查 — 非定时工单检查是否在维护窗口内 + if not ticket.scheduled_at: + maintenance_window = await _check_maintenance_window(db) + if maintenance_window: + raise HTTPException( + status_code=403, + detail=f"当前不在维护窗口内，允许执行时间: {maintenance_window}", + ) + # 6. 定时执行：如果 scheduled_at 在未来，改为 scheduled 状态 if ticket.scheduled_at: now = datetime.now(timezone.utc) @@ -971,6 +1207,9 @@ async def execute_ticket( ticket.execute_result = report.to_json() ticket.execute_time = datetime.now(timezone.utc) + # 8.1 更新 SQL 逐条执行记录 + await _update_sql_items_from_report(db, ticket.id, report) + # 9. 记录执行日志 log_detail = ( f"执行结果: {report.status}, 影响行数: {report.affected_rows}, " @@ -1006,6 +1245,9 @@ async def execute_ticket( }, ensure_ascii=False) ticket.execute_time = datetime.now(timezone.utc) + # 标记所有 pending SQL items 为 failed + await _mark_sql_items_failed(db, ticket.id, str(e)) + await AuditLogService.log( db, user_id=current_user.id, diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 933fbce..8639b5e 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -1,3 +1,4 @@ from app.models.instance_group import InstanceGroup, user_groups, instance_group_members # noqa: F401 from app.models.system_setting import SystemSetting # noqa: F401 from app.models.query_history import QueryHistory, QuickQueryModel # noqa: F401 +from app.models.ticket_sql_item import TicketSqlItem # noqa: F401 diff --git a/backend/app/models/ticket.py b/backend/app/models/ticket.py index e13266d..930f119 100644 --- a/backend/app/models/ticket.py +++ b/backend/app/models/ticket.py @@ -16,6 +16,8 @@ class Ticket(Base): rollback_sql = Column(Text, nullable=True) status = Column(String(20), default="pending") # pending / reviewing / approved / rejected / executing / success / failed priority = Column(String(10), default="medium") # high / medium / low + risk_level = Column(String(20), default="low", comment="自动风险等级: low/medium/high/critical") + workflow_id = Column(Integer, ForeignKey("workflow_definitions.id"), nullable=True, comment="关联审批流") applicant_id = Column(Integer, ForeignKey("users.id"), nullable=False) approver_id = Column(Integer, ForeignKey("users.id"), nullable=True) reason = Column(Text, nullable=True) diff --git a/backend/app/models/ticket_sql_item.py b/backend/app/models/ticket_sql_item.py new file mode 100644 index 0000000..5bb4afd --- /dev/null +++ b/backend/app/models/ticket_sql_item.py @@ -0,0 +1,28 @@ +"""工单 SQL 逐条执行记录 — 支持执行进度透明展示。 + +每条 SQL 单独记录执行状态、耗时、影响行数、错误信息。 +""" + +from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, func +from sqlalchemy.orm import relationship + +from app.core.database import Base + + +class TicketSqlItem(Base): + __tablename__ = "ticket_sql_items" + + id = Column(Integer, primary_key=True, autoincrement=True) + ticket_id = Column(Integer, ForeignKey("tickets.id"), nullable=False, index=True) + seq = Column(Integer, nullable=False, comment="SQL 顺序号，从 1 开始") + sql_text = Column(Text, nullable=False, comment="单条 SQL 文本") + sql_type = Column(String(20), nullable=True, comment="SQL 类型: DDL/DML/DQL/SET 等") + status = Column(String(20), default="pending", comment="pending/running/success/failed/skipped") + affected_rows = Column(Integer, default=0) + duration_ms = Column(Integer, default=0, comment="执行耗时（毫秒）") + error_message = Column(Text, nullable=True) + started_at = Column(DateTime, nullable=True) + finished_at = Column(DateTime, nullable=True) + created_at = Column(DateTime, server_default=func.now()) + + ticket = relationship("Ticket", backref="sql_items") diff --git a/backend/app/schemas/ticket.py b/backend/app/schemas/ticket.py index 66f95e7..7d74ae9 100644 --- a/backend/app/schemas/ticket.py +++ b/backend/app/schemas/ticket.py @@ -9,6 +9,7 @@ class TicketCreate(BaseModel): instance_id: int target_database: str = Field(..., min_length=1) sql_content: str = Field(..., min_length=1) + sql_type: Optional[str] = None rollback_sql: Optional[str] = None reason: str = Field(..., min_length=5) priority: Literal["high", "medium", "low"] = "medium" @@ -32,6 +33,8 @@ class TicketResponse(BaseModel): rollback_sql: Optional[str] = None status: str priority: str + risk_level: Optional[str] = "low" + workflow_id: Optional[int] = None applicant_id: int applicant_name: Optional[str] = None approver_id: Optional[int] = None diff --git a/backend/app/services/audit_engine_v2.py b/backend/app/services/audit_engine_v2.py index 56252b3..15a5441 100644 --- a/backend/app/services/audit_engine_v2.py +++ b/backend/app/services/audit_engine_v2.py @@ -116,6 +116,7 @@ class AuditReportV2: sql: str findings: List[AuditFindingV2] = field(default_factory=list) db_type: str = "mysql" + extra_info: Dict[str, Any] = field(default_factory=dict) @property def passed(self) -> bool: @@ -1633,7 +1634,9 @@ class SQLAuditEngineV2: return rules return [(r, f) for r, f in rules if r.db_type in ("all", db_type)] - async def audit(self, sql: str, db_type: str = "mysql") -> AuditReportV2: + async def audit( + self, sql: str, db_type: str = "mysql", instance_id: Optional[int] = None + ) -> AuditReportV2: """执行 SQL 审核。 根据 SQL 语句类型，匹配对应的审核规则进行检查。 @@ -1641,6 +1644,7 @@ class SQLAuditEngineV2: Args: sql: 待审核的 SQL 语句 db_type: 目标数据库类型 (mysql/postgres/redis) + instance_id: 目标实例 ID（用于 EXPLAIN 预检、上下文感知规则等） Returns: AuditReportV2 审核报告 @@ -1649,14 +1653,24 @@ class SQLAuditEngineV2: sql_upper = sql_clean.upper() report = AuditReportV2(sql=sql_clean, db_type=db_type) + report.extra_info["instance_id"] = instance_id # 加载规则 rules = await self._load_rules(db_type) + # 构建上下文（用于上下文感知规则） + context: Dict[str, Any] = {"instance_id": instance_id, "db_session": self.db_session} + # 执行每条规则的检查 for rule, check_func in rules: try: - check_func(sql_clean, sql_upper, report, rule) + check_func(sql_clean, sql_upper, report, rule, context=context) + except TypeError: + # 兼容旧签名 (sql, sql_upper, report, rule) + try: + check_func(sql_clean, sql_upper, report, rule) + except Exception: + continue except Exception: # 规则检查失败不影响整体审核 continue diff --git a/backend/app/services/notification_sender.py b/backend/app/services/notification_sender.py index 03e20c4..7b18ed0 100644 --- a/backend/app/services/notification_sender.py +++ b/backend/app/services/notification_sender.py @@ -15,6 +15,37 @@ from urllib.parse import quote class NotificationSender: """通知发送器，支持钉钉、飞书、邮件。""" + # 敏感字段关键词（用于通知脱敏） + _SENSITIVE_PATTERNS = [ + (r"password\s*=\s*'[^']*'", "password='***'"), + (r"password\s*=\s*\"[^\"]*\"", 'password="***"'), + (r"secret\s*=\s*'[^']*'", "secret='***'"), + (r"token\s*=\s*'[^']*'", "token='***'"), + (r"DROP\s+TABLE\s+\w+", "DROP TABLE ***"), + (r"TRUNCATE\s+TABLE?\s*\w+", "TRUNCATE ***"), + (r"FLUSHALL", "FLUSHALL"), + (r"FLUSHDB", "FLUSHDB"), + ] + + @staticmethod + def _mask_sensitive_data(data: Dict[str, Any]) -> Dict[str, Any]: + """对通知数据中的敏感字段进行脱敏。""" + import re + masked = dict(data) + # sql_preview 截断到前100字符 + 脱敏 + if "sql_preview" in masked and masked["sql_preview"]: + preview = str(masked["sql_preview"])[:100] + for pattern, replacement in NotificationSender._SENSITIVE_PATTERNS: + preview = re.sub(pattern, replacement, preview, flags=re.IGNORECASE) + masked["sql_preview"] = preview + ("..." if len(str(data.get("sql_preview", ""))) > 100 else "") + # error 字段脱敏 + if "error" in masked and masked["error"]: + err = str(masked["error"])[:200] + for pattern, replacement in NotificationSender._SENSITIVE_PATTERNS: + err = re.sub(pattern, replacement, err, flags=re.IGNORECASE) + masked["error"] = err + return masked + # 内置消息模板 TEMPLATES = { "ticket_created": { @@ -91,6 +122,9 @@ class NotificationSender: event_type: 事件类型 data: 模板数据 """ + # 通知脱敏：隐藏 SQL 中的敏感内容 + data = self._mask_sensitive_data(data) + try: if channel_type == "dingtalk": return await self._send_dingtalk(config, event_type, data) diff --git a/backend/app/services/risk_evaluator.py b/backend/app/services/risk_evaluator.py new file mode 100644 index 0000000..11e4242 --- /dev/null +++ b/backend/app/services/risk_evaluator.py @@ -0,0 +1,189 @@ +"""工单风险等级自动评估。 + +根据 SQL 类型 + 操作对象 + 审核结果，自动计算 risk_level: + - low: SELECT / 简单 DML (有 WHERE + LIMIT) + - medium: 普通 DML / 低风险 DDL (CREATE TABLE) + - high: 大表 ALTER / 无 WHERE 的 UPDATE/DELETE / 多表 JOIN + - critical: DROP TABLE / TRUNCATE / FLUSHDB / FLUSHALL / dropDatabase + +评估时机：工单创建时自动计算，写入 ticket.risk_level。 +""" + +import re +from typing import Optional + + +def evaluate_risk_level( + sql_content: str, + db_type: str = "mysql", + sql_type: str = "", + audit_findings: Optional[list] = None, + statement_count: int = 1, +) -> str: + """评估 SQL 的风险等级。 + + Args: + sql_content: SQL 文本（可能包含多条语句） + db_type: 数据库类型 mysql/postgres/redis/mongodb + sql_type: 前端传入的 SQL 类型标注 DDL/DML/DQL 等 + audit_findings: 审核引擎返回的 findings 列表 + statement_count: SQL 语句条数 + + Returns: + risk_level: low / medium / high / critical + """ + sql_upper = sql_content.strip().upper() + + # ── Redis 特殊处理 ── + if db_type == "redis": + return _evaluate_redis_risk(sql_upper) + + # ── MongoDB 特殊处理 ── + if db_type == "mongodb": + return _evaluate_mongo_risk(sql_upper) + + # ── MySQL / PostgreSQL ── + + # 1. critical 级别 — 不可逆危险操作 + critical_patterns = [ + r"\bDROP\s+TABLE\b", + r"\bDROP\s+DATABASE\b", + r"\bDROP\s+SCHEMA\b", + r"\bTRUNCATE\s", + ] + for pattern in critical_patterns: + if re.search(pattern, sql_upper): + return "critical" + + # 2. high 级别 — 高风险操作 + high_indicators = 0 + + # ALTER 大操作（但不包括 ADD INDEX 这类较小操作） + if re.search(r"\bALTER\s+TABLE\b", sql_upper): + alter_high = [ + r"\bDROP\s+COLUMN\b", + r"\bMODIFY\s+COLUMN\b", + r"\bCHANGE\s+COLUMN\b", + r"\bDROP\s+INDEX\b", + r"\bDROP\s+PRIMARY\s+KEY\b", + r"\bCONVERT\s+TO\s+CHARACTER\s+SET\b", + r"\bRENAME\s+(TO|AS)\b", + ] + for p in alter_high: + if re.search(p, sql_upper): + high_indicators += 1 + break + + # 无 WHERE 的 UPDATE / DELETE + if re.search(r"\bUPDATE\s", sql_upper) and not re.search(r"\bWHERE\b", sql_upper): + high_indicators += 1 + if re.search(r"\bDELETE\s+FROM\b", sql_upper) and not re.search(r"\bWHERE\b", sql_upper): + high_indicators += 1 + + # 大量语句（>5条） + if statement_count > 5: + high_indicators += 1 + + # 审核结果有 ERROR 级别 + if audit_findings: + errors = [f for f in audit_findings if f.get("level") == "error"] + if len(errors) >= 2: + high_indicators += 1 + + if high_indicators > 0: + return "high" + + # 3. medium 级别 — 常规变更 + medium_types = {"DDL", "DML", "SET", "HSET", "HMSET"} + if sql_type in medium_types: + return "medium" + + # 有 WHERE 但没有 LIMIT 的 DELETE + if re.search(r"\bDELETE\s+FROM\b", sql_upper): + if not re.search(r"\bLIMIT\b", sql_upper): + return "medium" + + # 任何 DDL + ddl_keywords = ["CREATE ", "ALTER ", "RENAME ", "GRANT ", "REVOKE ", "CREATE INDEX", "CREATE UNIQUE"] + for kw in ddl_keywords: + if kw in sql_upper: + return "medium" + + # 4. low 级别 — 只读 / 安全操作 + return "low" + + +def _evaluate_redis_risk(sql_upper: str) -> str: + """Redis 命令风险评估。""" + critical_cmds = ["FLUSHDB", "FLUSHALL"] + for cmd in critical_cmds: + if sql_upper.startswith(cmd): + return "critical" + + # 危险写操作 + high_cmds = ["DEL ", "UNLINK ", "RENAME ", "FLUSHDB", "FLUSHALL"] + for cmd in high_cmds: + if sql_upper.startswith(cmd): + return "high" + + # 一般写操作 + write_cmds = ["SET ", "HSET ", "HMSET ", "LPUSH ", "RPUSH ", "SADD ", "ZADD ", + "INCR", "DECR", "EXPIRE ", "PEXPIRE "] + for cmd in write_cmds: + if sql_upper.startswith(cmd): + return "medium" + + # 只读命令 + return "low" + + +def _evaluate_mongo_risk(sql_upper: str) -> str: + """MongoDB 操作风险评估。""" + sql_lower = sql_upper.lower() + critical_ops = ["dropdatabase", "drop database"] + for op in critical_ops: + if op in sql_lower: + return "critical" + + high_ops = ["drop(", "dropcollection", "remove(", "deletemany"] + for op in high_ops: + if op in sql_lower: + return "high" + + write_ops = ["insertone", "insertmany", "updateone", "updatemany", "deleteone", + "createindex", "create_collection"] + for op in write_ops: + if op in sql_lower: + return "medium" + + return "low" + + +# ── 风险等级配置（用于前端展示 / 审批流匹配）── + +RISK_LEVEL_CONFIG = { + "low": { + "label": "低风险", + "color": "#52c41a", + "description": "只读查询 / 安全操作", + "min_approvers": 1, + }, + "medium": { + "label": "中风险", + "color": "#faad14", + "description": "常规数据变更 / DDL 操作", + "min_approvers": 1, + }, + "high": { + "label": "高风险", + "color": "#fa8c16", + "description": "无 WHERE 删除/更新 / 大表 ALTER / 批量操作", + "min_approvers": 2, + }, + "critical": { + "label": "极高风险", + "color": "#f5222d", + "description": "DROP TABLE / TRUNCATE / FLUSHDB 等不可逆操作", + "min_approvers": 2, + }, +} diff --git a/backend/tests/test_phase1_features.py b/backend/tests/test_phase1_features.py new file mode 100644 index 0000000..7520576 --- /dev/null +++ b/backend/tests/test_phase1_features.py @@ -0,0 +1,280 @@ +"""风险等级评估 + 维护窗口 + 通知脱敏 + SQL items 测试。""" + +import json +import pytest +from unittest.mock import patch, MagicMock + +# ===== 风险等级评估测试 ===== + +class TestRiskEvaluator: + """风险等级自动计算测试。""" + + def test_select_is_low(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("SELECT * FROM users WHERE id = 1") == "low" + + def test_show_is_low(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("SHOW TABLES") == "low" + + def test_explain_is_low(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("EXPLAIN SELECT * FROM users") == "low" + + def test_create_table_is_medium(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("CREATE TABLE t1 (id INT)") == "medium" + + def test_alter_table_is_medium(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("ALTER TABLE users ADD COLUMN age INT") == "medium" + + def test_insert_with_where_is_medium(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level( + "INSERT INTO users (name) VALUES ('test')", + sql_type="DML", + ) == "medium" + + def test_update_without_where_is_high(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("UPDATE users SET status = 1") == "high" + + def test_delete_without_where_is_high(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("DELETE FROM users") == "high" + + def test_alter_drop_column_is_high(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("ALTER TABLE users DROP COLUMN age") == "high" + + def test_alter_modify_column_is_high(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("ALTER TABLE users MODIFY COLUMN age BIGINT") == "high" + + def test_drop_table_is_critical(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("DROP TABLE users") == "critical" + + def test_truncate_is_critical(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("TRUNCATE TABLE users") == "critical" + + def test_drop_database_is_critical(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("DROP DATABASE mydb") == "critical" + + def test_many_statements_is_high(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level( + "SELECT 1", statement_count=10 + ) == "high" + + def test_multiple_audit_errors_is_high(self): + from app.services.risk_evaluator import evaluate_risk_level + findings = [ + {"level": "error", "rule_id": "R1"}, + {"level": "error", "rule_id": "R2"}, + ] + assert evaluate_risk_level( + "UPDATE users SET x=1 WHERE id=1", + sql_type="DML", + audit_findings=findings, + ) == "high" + + def test_delete_with_where_no_limit_is_medium(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level( + "DELETE FROM users WHERE id > 100" + ) == "medium" + + def test_delete_with_where_and_limit_is_low(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level( + "DELETE FROM users WHERE id > 100 LIMIT 10" + ) == "low" + + # Redis 风险 + def test_redis_flushall_is_critical(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("FLUSHALL", db_type="redis") == "critical" + + def test_redis_flushdb_is_critical(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("FLUSHDB", db_type="redis") == "critical" + + def test_redis_set_is_medium(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("SET key value", db_type="redis") == "medium" + + def test_redis_get_is_low(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("GET key", db_type="redis") == "low" + + # MongoDB 风险 + def test_mongo_drop_database_is_critical(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("db.dropDatabase()", db_type="mongodb") == "critical" + + def test_mongo_drop_database_upper_is_critical(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("DROP DATABASE mydb", db_type="mongodb") == "critical" + + def test_mongo_insert_is_medium(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("db.users.insertOne({name: 'test'})", db_type="mongodb") == "medium" + + def test_mongo_find_is_low(self): + from app.services.risk_evaluator import evaluate_risk_level + assert evaluate_risk_level("db.users.find({})", db_type="mongodb") == "low" + + +# ===== 风险等级配置测试 ===== + +class TestRiskLevelConfig: + """风险等级配置验证。""" + + def test_all_levels_have_config(self): + from app.services.risk_evaluator import RISK_LEVEL_CONFIG + for level in ["low", "medium", "high", "critical"]: + assert level in RISK_LEVEL_CONFIG + cfg = RISK_LEVEL_CONFIG[level] + assert "label" in cfg + assert "color" in cfg + assert "min_approvers" in cfg + + def test_high_and_critical_require_2_approvers(self): + from app.services.risk_evaluator import RISK_LEVEL_CONFIG + assert RISK_LEVEL_CONFIG["high"]["min_approvers"] >= 2 + assert RISK_LEVEL_CONFIG["critical"]["min_approvers"] >= 2 + + +# ===== 通知脱敏测试 ===== + +class TestNotificationMasking: + """通知内容脱敏测试。""" + + def test_mask_password_in_sql_preview(self): + from app.services.notification_sender import NotificationSender + data = {"sql_preview": "SET password='secret123' WHERE id=1"} + masked = NotificationSender._mask_sensitive_data(data) + assert "secret123" not in masked["sql_preview"] + assert "***" in masked["sql_preview"] + + def test_mask_drop_table_in_sql_preview(self): + from app.services.notification_sender import NotificationSender + data = {"sql_preview": "DROP TABLE users"} + masked = NotificationSender._mask_sensitive_data(data) + assert "***" in masked["sql_preview"] + + def test_mask_error_field(self): + from app.services.notification_sender import NotificationSender + data = {"error": "Connection failed: password='mysecret'"} + masked = NotificationSender._mask_sensitive_data(data) + assert "mysecret" not in masked["error"] + + def test_truncate_long_sql_preview(self): + from app.services.notification_sender import NotificationSender + long_sql = "SELECT * FROM users WHERE " + "x=1 AND " * 50 + data = {"sql_preview": long_sql} + masked = NotificationSender._mask_sensitive_data(data) + assert len(masked["sql_preview"]) <= 110 # 100 + "..." + + def test_non_sensitive_data_unchanged(self): + from app.services.notification_sender import NotificationSender + data = {"title": "测试工单", "applicant": "张三"} + masked = NotificationSender._mask_sensitive_data(data) + assert masked["title"] == "测试工单" + assert masked["applicant"] == "张三" + + +# ===== 维护窗口测试 ===== + +class TestMaintenanceWindow: + """维护窗口逻辑测试。""" + + def test_no_config_allows_execution(self): + """无配置时允许执行。""" + from app.api.v1.endpoints.ticket import _check_maintenance_window + from unittest.mock import AsyncMock + import asyncio + + mock_db = AsyncMock() + mock_result = MagicMock() + mock_result.scalar_one_or_none.return_value = None + mock_db.execute.return_value = mock_result + + async def run(): + result = await _check_maintenance_window(mock_db) + assert result is None + + asyncio.get_event_loop().run_until_complete(run()) + + def test_disabled_window_allows_execution(self): + """disabled 的窗口允许执行。""" + from app.api.v1.endpoints.ticket import _check_maintenance_window + from app.models.system_setting import SystemSetting + from unittest.mock import AsyncMock + import asyncio + + setting = SystemSetting(key="maintenance_window", value='{"enabled": false}') + mock_db = AsyncMock() + mock_result = MagicMock() + mock_result.scalar_one_or_none.return_value = setting + mock_db.execute.return_value = mock_result + + async def run(): + result = await _check_maintenance_window(mock_db) + assert result is None + + asyncio.get_event_loop().run_until_complete(run()) + + +# ===== SQL 类型分类测试 ===== + +class TestClassifySqlType: + """SQL 类型分类辅助函数测试。""" + + def test_select_is_dql(self): + from app.api.v1.endpoints.ticket import _classify_sql_type + assert _classify_sql_type("SELECT * FROM t") == "DQL" + + def test_insert_is_dml(self): + from app.api.v1.endpoints.ticket import _classify_sql_type + assert _classify_sql_type("INSERT INTO t VALUES (1)") == "DML" + + def test_update_is_dml(self): + from app.api.v1.endpoints.ticket import _classify_sql_type + assert _classify_sql_type("UPDATE t SET x=1") == "DML" + + def test_delete_is_dml(self): + from app.api.v1.endpoints.ticket import _classify_sql_type + assert _classify_sql_type("DELETE FROM t") == "DML" + + def test_create_is_ddl(self): + from app.api.v1.endpoints.ticket import _classify_sql_type + assert _classify_sql_type("CREATE TABLE t (id INT)") == "DDL" + + def test_alter_is_ddl(self): + from app.api.v1.endpoints.ticket import _classify_sql_type + assert _classify_sql_type("ALTER TABLE t ADD COLUMN x INT") == "DDL" + + def test_drop_is_ddl(self): + from app.api.v1.endpoints.ticket import _classify_sql_type + assert _classify_sql_type("DROP TABLE t") == "DDL" + + def test_grant_is_dcl(self): + from app.api.v1.endpoints.ticket import _classify_sql_type + assert _classify_sql_type("GRANT SELECT ON t TO user") == "DCL" + + def test_redis_set(self): + from app.api.v1.endpoints.ticket import _classify_sql_type + assert _classify_sql_type("SET key value") == "SET" + + def test_redis_get(self): + from app.api.v1.endpoints.ticket import _classify_sql_type + assert _classify_sql_type("GET key") == "SET" + + def test_unknown(self): + from app.api.v1.endpoints.ticket import _classify_sql_type + assert _classify_sql_type("UNKNOWN COMMAND") == "OTHER"