from datetime import datetime, timezone
from typing import Any
from app.config import security_config
from app.security.input_guard import process_input
from app.security.jailbreak_guard import detect_jailbreak
from app.security.output_guard import filter_output
from app.security.pipeline import STEP_LABELS, make_step, skipped_step
from app.security.task_lock import check_task
class SecurityLayer:
def __init__(self) -> None:
self.audit_logs: list[dict[str, Any]] = []
def _audit(
self,
*,
scenario: str,
mode: str,
message: str,
attack_success: bool,
blocked_by: str | None,
consequences: list[str],
) -> dict[str, Any]:
entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"scenario": scenario,
"mode": mode,
"message_preview": message[:80] + ("..." if len(message) > 80 else ""),
"attack_success": attack_success,
"blocked_by": blocked_by,
"consequences": consequences,
}
if security_config.audit_enabled:
self.audit_logs.append(entry)
if len(self.audit_logs) > 200:
self.audit_logs = self.audit_logs[-200:]
return entry
def _build_input_step(self, input_result, cfg) -> dict[str, Any]:
if not cfg.enable_input_isolation and not cfg.enable_injection_detection:
return make_step(
"input_guard",
STEP_LABELS["input_guard"],
status="skipped",
enabled=False,
summary="输入隔离与注入检测均已关闭",
)
parts = []
if cfg.enable_input_isolation:
parts.append("已将用户输入包裹进 <user_input> 标签，与系统指令隔离")
if cfg.enable_injection_detection and input_result.injection_detected:
parts.append(f"检测到注入特征（风险等级：{input_result.risk_level}）")
elif cfg.enable_injection_detection:
parts.append("未检测到明显注入特征")
status = "warning" if input_result.injection_detected else "passed"
summary = "；".join(parts)
detail = None
if cfg.enable_input_isolation:
preview = input_result.wrapped_prompt
if len(preview) > 400:
preview = preview[:400] + "\n…（已截断）"
detail = f"发送给模型的安全提示词：\n\n{preview}"
return make_step(
"input_guard",
STEP_LABELS["input_guard"],
status=status,
enabled=True,
summary=summary,
detail=detail,
)
def process(
self,
user_input: str,
*,
scenario_id: str,
task: str = "translate",
) -> dict[str, Any]:
audit_entries: list[dict[str, Any]] = []
pipeline: list[dict[str, Any]] = []
cfg = security_config
input_result = process_input(
user_input,
task=task,
enable_isolation=cfg.enable_input_isolation,
enable_detection=cfg.enable_injection_detection,
)
pipeline.append(self._build_input_step(input_result, cfg))
if cfg.enable_jailbreak_detection:
jb = detect_jailbreak(user_input)
if jb.is_jailbreak:
pipeline.append(
make_step(
"jailbreak_guard",
STEP_LABELS["jailbreak_guard"],
status="blocked",
enabled=True,
summary=f"识别到越狱特征：{', '.join(jb.matched_reasons)}",
detail="高风险请求在调用模型前被标准化拒绝，不进入 LLM。",
)
)
pipeline.extend([
skipped_step("task_lock", STEP_LABELS["task_lock"], "前序步骤已拦截，跳过"),
skipped_step("llm_call", STEP_LABELS["llm_call"], "前序步骤已拦截，跳过"),
skipped_step("output_guard", STEP_LABELS["output_guard"], "前序步骤已拦截，跳过"),
])
entry = self._audit(
scenario=scenario_id,
mode="secure",
message=user_input,
attack_success=False,
blocked_by="jailbreak_guard",
consequences=["合规违规（已拦截）"],
)
audit_entries.append(entry)
return {
"blocked": True,
"blocked_by": "jailbreak_guard",
"output": cfg.refusal_message,
"defense_explanation": (
f"输入端识别到越狱特征：{', '.join(jb.matched_reasons)}。"
"高风险请求在调用模型前被标准化拒绝。"
),
"attack_success": False,
"consequences": [],
"risk_level": jb.risk_level,
"prompt_for_llm": None,
"injection_detected": input_result.injection_detected,
"audit_entries": audit_entries,
"pipeline": pipeline,
}
pipeline.append(
make_step(
"jailbreak_guard",
STEP_LABELS["jailbreak_guard"],
status="passed",
enabled=True,
summary="未检测到越狱或高风险伪装",
)
)
else:
pipeline.append(
make_step(
"jailbreak_guard",
STEP_LABELS["jailbreak_guard"],
status="skipped",
enabled=False,
summary="越狱检测已关闭",
)
)
task_result = check_task(
user_input,
whitelist=cfg.task_whitelist,
scenario_task=task,
)
if not task_result.allowed:
pipeline.append(
make_step(
"task_lock",
STEP_LABELS["task_lock"],
status="blocked",
enabled=True,
summary=task_result.reason or "任务超出白名单范围",
detail=f"当前白名单：{', '.join(cfg.task_whitelist)}",
)
)
pipeline.extend([
skipped_step("llm_call", STEP_LABELS["llm_call"], "前序步骤已拦截，跳过"),
skipped_step("output_guard", STEP_LABELS["output_guard"], "前序步骤已拦截，跳过"),
])
entry = self._audit(
scenario=scenario_id,
mode="secure",
message=user_input,
attack_success=False,
blocked_by="task_lock",
consequences=["越权访问（已拦截）"],
)
audit_entries.append(entry)
return {
"blocked": True,
"blocked_by": "task_lock",
"output": cfg.refusal_message,
"defense_explanation": task_result.reason,
"attack_success": False,
"consequences": [],
"risk_level": "high",
"prompt_for_llm": None,
"injection_detected": input_result.injection_detected,
"audit_entries": audit_entries,
"pipeline": pipeline,
}
pipeline.append(
make_step(
"task_lock",
STEP_LABELS["task_lock"],
status="passed",
enabled=True,
summary=f"任务「{task_result.detected_task or task}」在白名单内，允许继续",
detail=f"白名单：{', '.join(cfg.task_whitelist)}",
)
)
prompt = input_result.wrapped_prompt if cfg.enable_input_isolation else user_input
return {
"blocked": False,
"blocked_by": None,
"output": None,
"defense_explanation": None,
"attack_success": False,
"consequences": [],
"risk_level": input_result.risk_level,
"prompt_for_llm": prompt,
"injection_detected": input_result.injection_detected,
"audit_entries": audit_entries,
"pipeline": pipeline,
"input_guard_note": (
"用户输入已用 <user_input> 标签隔离，其中指令不会被执行为新任务。"
if cfg.enable_input_isolation and input_result.injection_detected
else None
),
}
def build_llm_step(self, *, called: bool, prompt_preview: str | None) -> dict[str, Any]:
if not called:
return skipped_step("llm_call", STEP_LABELS["llm_call"], "前序步骤已拦截，未调用模型")
preview = prompt_preview or ""
if len(preview) > 300:
preview = preview[:300] + "…"
return make_step(
"llm_call",
STEP_LABELS["llm_call"],
status="passed",
enabled=True,
summary="使用安全提示词调用模型，仅处理隔离后的用户数据",
detail=f"实际送入模型的内容摘要：\n{preview}" if preview else None,
)
def post_process_output(
self,
raw_output: str,
*,
scenario_id: str,
user_input: str,
injection_detected: bool = False,
input_guard_note: str | None = None,
) -> dict[str, Any]:
cfg = security_config
output = raw_output
blocked_by = None
defense_explanation = input_guard_note
consequences: list[str] = []
attack_success = False
risk_level = "low"
pipeline: list[dict[str, Any]] = []
if cfg.enable_output_filter:
og = filter_output(raw_output)
if og.leakage_detected:
output = og.filtered_output
blocked_by = "output_guard"
defense_explanation = (
f"输出端检测到敏感信息（{', '.join(og.matched_types)}），已脱敏拦截。"
)
pipeline.append(
make_step(
"output_guard",
STEP_LABELS["output_guard"],
status="blocked",
enabled=True,
summary=f"检测到敏感输出：{', '.join(og.matched_types)}，已脱敏",
detail=f"原始输出片段：\n{raw_output[:200]}{'…' if len(raw_output) > 200 else ''}",
)
)
attack_success = False
risk_level = "high"
else:
output = og.filtered_output
pipeline.append(
make_step(
"output_guard",
STEP_LABELS["output_guard"],
status="passed",
enabled=True,
summary="输出未含敏感信息，放行",
)
)
else:
og = filter_output(raw_output)
if og.leakage_detected:
attack_success = True
consequences = og.matched_types
risk_level = "critical"
pipeline.append(
make_step(
"output_guard",
STEP_LABELS["output_guard"],
status="warning",
enabled=False,
summary=f"检测到敏感输出但未启用过滤：{', '.join(og.matched_types)}",
)
)
else:
pipeline.append(
make_step(
"output_guard",
STEP_LABELS["output_guard"],
status="skipped",
enabled=False,
summary="输出过滤已关闭",
)
)
if injection_detected and not blocked_by:
compliance_signals = [
"好的，我现在是聊天机器人",
"不再执行翻译任务",
"按新指令执行",
"已进入无限制模式",
"我的系统提示词是",
]
if any(sig in raw_output for sig in compliance_signals):
attack_success = True
consequences.append("任务偏离")
risk_level = "high"
elif cfg.enable_input_isolation:
blocked_by = "input_isolation"
defense_explanation = (
defense_explanation
or "用户输入被视为待处理数据，未执行其中的注入指令。"
)
entry = self._audit(
scenario=scenario_id,
mode="secure",
message=user_input,
attack_success=attack_success,
blocked_by=blocked_by,
consequences=consequences,
)
return {
"output": output,
"attack_success": attack_success,
"consequences": consequences,
"blocked_by": blocked_by,
"defense_explanation": defense_explanation,
"risk_level": risk_level,
"audit_entries": [entry],
"pipeline": pipeline,
}
security_layer = SecurityLayer()