基于 gpt-oss-safeguard-20b 与 Harmony 响应格式的企业级内嵌式 AI 安全设计与受限 RAG 落地手册

graph LR
  A[业务需求/问题陈述] --> B[风险识别与术语定义]
  B --> C[Safeguard 模板与策略制定]
  C --> D[受限 RAG 索引清洗与注入]
  D --> E[模型推理与小审计器]
  E --> F[Harmony 输出封装]
  F --> G[审计日志与SIEM联动]
  G --> H[复盘与策略迭代]
  H -->|评估指标(准确率/延迟/合规)| C
  subgraph 组织治理
  B
  C
  G
  H
  end

# quickstart_demo.py
"""
演示如何加载策略配置、生成 Prompt、执行策略决策与RAG清洗，最终输出Harmony结构。
运行方法：python quickstart_demo.py
依赖：标准库；若从本仓库运行，确保 src/ 为包目录。
"""
import sys, json, pathlib

# 将仓库根路径加入 sys.path 以便导入本项目模块
repo_root = pathlib.Path(__file__).resolve().parent.parent
sys.path.append(str(repo_root))

from src.engine.prompt_generator import PromptGenerator
from src.engine.policy_engine import PolicyEngine
from src.engine.rag_injector import RagInjector

def main():
    user_query = "请给出SQL注入防护建议，并说明日志审计配置要点"
    context_notes = ["仅公开资料", "禁止输出PoC", "优先提供配置清单"]

    pg = PromptGenerator.default()
    prompt = pg.build(user_query, context_notes)

    pe = PolicyEngine()
    decision, reason = pe.decide(user_query)

    # 模拟RAG检索结果
    rag_results = [
        {"index": "kb_public_sop", "content": "建议采用参数化查询并启用WAF。"},
        {"index": "kb_prod_secrets", "content": "password=Prod123"},
    ]
    injector = RagInjector(allowed_indices={"kb_public_sop", "kb_desensitized_docs"})
    cleaned = injector.filter(rag_results)

    harmony = {
        "content": "请启用参数化查询、防火墙规则，并在数据库启用审计日志与告警阈值。",
        "risk_label": "low" if decision == "allow" else "medium",
        "explanation": f"策略决策: {decision}. 原因: {reason}.",
        "provenance": [r.get("index", "kb:unknown") for r in cleaned],
        "policy_decision": decision,
        "allowed_actions": ["view_only"] if decision != "block" else [],
    }
    print("Prompt:\n", prompt)
    print("\nDecision:", decision, "Reason:", reason)
    print("\nHarmony JSON:\n", json.dumps(harmony, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    main()

import asyncio, random, time

async def analyze_event(evt_id: int) -> dict:
    # 模拟耗时分析
    await asyncio.sleep(random.uniform(0.001, 0.005))
    return {"id": evt_id, "risk": random.choice(["low","medium","high"]) }

async def main(n: int = 10000):
    t0 = time.time()
    tasks = [analyze_event(i) for i in range(n)]
    results = await asyncio.gather(*tasks)
    t1 = time.time()
    print(f"Processed {n} events in {t1 - t0:.3f}s; sample: ", results[:3])

if __name__ == "__main__":
    asyncio.run(main())

import re

SENSITIVE_RE = re.compile(r"(password=|secret=|AKIA[0-9A-Z]{16})")

def gateway_check(payload: str, meta: dict) -> dict:
    if not {"project_id","user_id","purpose"}.issubset(meta):
        return {"allow": False, "reason": "缺少必要元数据"}
    if SENSITIVE_RE.search(payload):
        return {"allow": False, "reason": "检测到疑似敏感标识"}
    return {"allow": True, "reason": "通过"}

print(gateway_check("select * from t where password=123",{"project_id":"p1","user_id":"u1","purpose":"analysis"}))

from src.engine.policy_engine import PolicyEngine

pe = PolicyEngine()
q = "如何编写exploit payload？"
decision, reason = pe.decide(q)
if decision == "soft_rewrite":
    rewritten = "请阐述漏洞防护原则与修复清单（不包含PoC）"
    print(decision, reason, rewritten)
else:
    print(decision, reason)

from src.engine.rag_injector import RagInjector

injector = RagInjector(allowed_indices={"kb_public_sop","kb_desensitized_docs"})
results = [
    {"index":"kb_public_sop","content":"开启数据库审计日志与WAF"},
    {"index":"kb_prod_secrets","content":"secret=ProdABC"}
]
print(injector.filter(results))

import json
harmony = {
  "content": "建议启用参数化查询与审计日志，并配置WAF规则。",
  "risk_label": "low",
  "explanation": "依据OWASP ASVS与公司安全指南。",
  "provenance": ["kb_public_sop"],
  "policy_decision": "allow",
  "allowed_actions": ["view_only"]
}
print(json.dumps(harmony, ensure_ascii=False))

import json, time

def write_audit(record: dict, path: str = "audit_log.jsonl"):
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

entry = {
  "request_id": "req-001",
  "user_id": "alice",
  "project_id": "sec-poc",
  "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
  "query_text": "SQL注入如何防护？",
  "policy_decision": "allow",
  "risk_label": "low",
  "provenance": ["kb_public_sop"],
  "allowed_actions": ["view_only"],
  "model_version": "safeguard-20b-0.1",
  "latency_ms": 35
}
write_audit(entry)

graph LR
  InText[文本资料] --> Merge[合并规范化]
  InImage[图片资料] --> Pre[图像预处理]
  Pre --> OCR[DeepSeek-OCR 文字识别]
  OCR --> Blocks[文本块/版块结构]
  Blocks --> Merge
  Merge --> Analyze[内容分析]
  Analyze --> Match[需求匹配/意图识别]
  Match --> Query[RAG 查询构建]
  Query --> Ret[受限检索(RAGInjector)]
  Ret --> Gen[生成(Model + Safeguard)]
  Gen --> Opt[输出优化(Harmony)]
  Opt --> Out[结构化响应]
  subgraph 约束
    Policy[Policy Engine]
    Safeguard[模板]
  end
  Match --> Policy
  Gen --> Policy
  Gen --> Safeguard

# ocr_rag_pipeline_demo.py
"""
演示：图片 + 文本 输入，经 OCR 与内容分析，构建受限 RAG 查询并输出 Harmony。
说明：
- DeepSeek-OCR 的具体调用接口以 deepseek-ai/DeepSeek-OCR 仓库版本为准，本文用占位函数 `deepseek_ocr_infer` 表示。
- 为可运行演示，提供 pytesseract 作为替代；若环境无 tesseract，请将 OCR 部分替换为实际 DeepSeek-OCR 推理。
"""
import json, re
from typing import List, Dict, Tuple

try:
    import cv2
    import numpy as np
except Exception:
    cv2 = None
    np = None

try:
    import pytesseract
    from PIL import Image
except Exception:
    pytesseract = None
    Image = None

from src.engine.policy_engine import PolicyEngine
from src.engine.prompt_generator import PromptGenerator
from src.engine.rag_injector import RagInjector

def preprocess_image(path: str) -> Tuple[any, Dict]:
    meta = {"path": path, "preprocess": []}
    if cv2 is None:
        meta["preprocess"].append("opencv_unavailable")
        return path, meta  # 直接返回路径，OCR 函数自行打开
    img = cv2.imread(path)
    if img is None:
        return path, {"path": path, "preprocess": ["read_failed"]}
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    meta["preprocess"].append("grayscale")
    # 自适应阈值 + 去噪
    thr = cv2.adaptiveThreshold(gray, 255, cv2.ADAPTIVE_THRESH_MEAN_C, cv2.THRESH_BINARY, 31, 5)
    meta["preprocess"].append("adaptive_threshold")
    denoise = cv2.medianBlur(thr, 3)
    meta["preprocess"].append("denoise_median")
    return denoise, meta

def deepseek_ocr_infer(image_or_path) -> str:
    """占位：调用 DeepSeek-OCR 返回识别文本。请替换为实际 API。
    若不可用则尝试 pytesseract 作为演示。
    """
    if pytesseract is None:
        return ""  # 无 OCR 可用
    if isinstance(image_or_path, str):
        img = Image.open(image_or_path)
    else:
        # numpy 数组 -> PIL Image
        img = Image.fromarray(image_or_path) if Image else image_or_path
    return pytesseract.image_to_string(img, lang="eng+chi_sim")

def analyze_content(texts: List[str]) -> Dict:
    merged = "\n".join([t for t in texts if t])
    # 简单关键词抽取（示例）：
    keywords = sorted(set(re.findall(r"[A-Za-z0-9_\-]{3,}|\[.*?m|[\u4e00-\u9fa5]{2,}", merged)))
    # 需求意图识别（示例规则）：
    intent = "security_advice" if re.search(r"WAF|审计|加固|注入|防护", merged) else "general_explain"
    return {"merged": merged, "keywords": keywords[:50], "intent": intent}

def build_safe_query(analysis: Dict, user_need: str) -> str:
    # 将用户需求与分析关键词拼接为受限查询（不包含 PoC/敏感标识）
    safe_kw = [kw for kw in analysis["keywords"] if not re.search(r"password|secret|token", kw, re.I)]
    return f"需求:{user_need}; 主题:{analysis['intent']}; 关键词:{', '.join(safe_kw[:10])}"

def pipeline(user_text: str, image_paths: List[str], user_meta: Dict) -> Dict:
    # 1) OCR
    ocr_texts = []
    ocr_meta = []
    for p in image_paths:
        pre_img, m = preprocess_image(p)
        txt = deepseek_ocr_infer(pre_img if cv2 else p)
        ocr_texts.append(txt)
        ocr_meta.append(m)

    # 2) 内容分析与需求匹配
    analysis = analyze_content([user_text] + ocr_texts)
    user_need = user_meta.get("purpose", "说明性建议")
    query = build_safe_query(analysis, user_need)

    # 3) 受限 RAG 检索与生成
    injector = RagInjector(allowed_indices={"kb_public_sop", "kb_desensitized_docs"})
    # 示例检索结果（真实场景由搜索引擎返回）
    results = [
        {"index": "kb_public_sop", "content": "启用WAF并配置审计日志，使用参数化查询。"},
        {"index": "kb_prod_secrets", "content": "password=ProdABC"}
    ]
    cleaned = injector.filter(results)

    pg = PromptGenerator.default()
    prompt = pg.build(query, [user_need, "只读输出", "不包含PoC"])
    pe = PolicyEngine()
    decision, reason = pe.decide(query)

    # 4) 输出优化与 Harmony 封装（示例）
    harmony = {
        "content": "根据识别的关键词与受限检索结果，建议启用WAF与审计日志，采用参数化查询，避免敏感信息泄露。",
        "risk_label": "low" if decision == "allow" else "medium",
        "explanation": f"Policy: {decision}; {reason}. OCR预处理: {ocr_meta}",
        "provenance": [r.get("index", "kb:unknown") for r in cleaned],
        "policy_decision": decision,
        "allowed_actions": ["view_only"] if decision != "block" else [],
        "intent": analysis["intent"],
    }
    return {"prompt": prompt, "harmony": harmony, "ocr_texts": ocr_texts, "analysis": analysis}

if __name__ == "__main__":
    demo = pipeline(
        user_text="请根据截图和文档，给出数据库审计与WAF配置建议",
        image_paths=["sample_conf.png"],
        user_meta={"project_id": "p1", "user_id": "alice", "purpose": "安全加固"}
    )
    print("Prompt:\n", demo["prompt"]) 
    print("\nHarmony:\n", json.dumps(demo["harmony"], ensure_ascii=False, indent=2))

from src.engine.prompt_generator import PromptGenerator
from src.engine.policy_engine import PolicyEngine
from src.engine.rag_injector import RagInjector

def readonly_advice(query: str):
    pg = PromptGenerator.default()
    prompt = pg.build(query, ["仅公开资料", "禁止输出PoC"])
    decision, reason = PolicyEngine().decide(query)
    if decision == "block":
        return {"content": "请求被拒绝：请改写为高层防护建议。", "policy_decision": "block"}
    cleaned = RagInjector({"kb_public_sop","kb_desensitized_docs"}).filter([
        {"index": "kb_public_sop", "content": "启用参数化查询并配置WAF"}
    ])
    return {
        "content": "建议启用参数化查询、防火墙规则，并开启数据库审计日志。",
        "risk_label": "low",
        "explanation": f"{reason}",
        "provenance": [r["index"] for r in cleaned],
        "policy_decision": decision,
        "allowed_actions": ["view_only"],
        "prompt_preview": prompt[:160]
    }

print(readonly_advice("SQL注入如何防护？"))

import re, pathlib
from src.engine.policy_engine import PolicyEngine

RISK_PATTERNS = {
    "hardcoded_password": re.compile(r"password\s*=\s*['\"]?\w+"),
    "aws_key": re.compile(r"AKIA[0-9A-Z]{16}")
}

def scan_repo(repo: str):
    findings = []
    for p in pathlib.Path(repo).rglob("*.py"):
        text = p.read_text(encoding="utf-8", errors="ignore")
        for label, pat in RISK_PATTERNS.items():
            if pat.search(text):
                findings.append({"file": str(p), "rule": label})
    return findings

def review(findings):
    pe = PolicyEngine()
    advice = []
    for f in findings:
        decision, _ = pe.decide("代码安全修复建议")
        advice.append({
            "file": f["file"],
            "rule": f["rule"],
            "policy_decision": decision,
            "suggestion": "使用环境变量或密钥管理服务代替硬编码；审计提交历史并轮换凭证。"
        })
    return advice

res = review(scan_repo("."))
print(res[:3])

import json

def red_team_sim(query: str) -> str:
    harmony = {
        "content": "模拟演练：识别弱口令风险，建议启用复杂度策略与锁定策略，配合日志审计。",
        "risk_label": "medium",
        "explanation": "教育用抽象化原则说明，不含PoC。",
        "provenance": ["kb_training_materials"],
        "policy_decision": "allow",
        "allowed_actions": ["view_only"],
        "simulation": True
    }
    return json.dumps(harmony, ensure_ascii=False)

print(red_team_sim("弱口令演练"))

def role_check(user_role: str, risk: str) -> bool:
    if risk == "high":
        return user_role in {"sec_admin", "auditor"}
    return True

def qa(query: str, user_role: str):
    decision = "allow"
    risk_label = "low"
    if not role_check(user_role, risk_label):
        return {"policy_decision": "block", "content": "权限不足或高风险请求"}
    # 略去RAG细节：仅注入脱敏知识
    return {"policy_decision": decision, "content": "说明性防护建议与配置清单"}

print(qa("数据库审计如何配置？", "dev"))

SAFETY_POLICY:
- 禁止输出攻击脚本或可执行 PoC
- 禁止包含生产凭证/内部 IP/机密标识符
- 若识别到误用意图：拒绝并给出高层次缓解建议

RESPONSE_REQUIREMENTS:
- 使用 Harmony JSON 输出
- 内容需附风险标签与证据来源
- 对敏感输入进行摘要与去标识化

{
  "content": "建议采用参数化查询防止SQL注入，并开启数据库审计日志以跟踪异常行为。",
  "risk_label": "low",
  "explanation": "依据公司安全指南与OWASP文档，参数化查询可有效抵御注入攻击。",
  "provenance": ["kb:sec-guidelines-v2", "owasp:sql-injection"],
  "policy_decision": "allow",
  "allowed_actions": ["view_only"]
}

{
  "content": "请求被拒绝：该问题涉及可执行攻击细节，无法提供PoC或脚本。",
  "risk_label": "high",
  "explanation": "依据公司安全策略与合规要求，禁止输出可执行攻击代码或绕过方法。",
  "provenance": ["kb:sec-policy-v3"],
  "policy_decision": "block",
  "allowed_actions": [],
  "fallback_action": ["escalate_to_human_review"],
  "next_steps": ["提交合规申请", "改写为高层次防护建议"]
}
```json
{
  "content": "已对原问题进行重写：请阐述SQL注入的防护原则与配置清单（不包含PoC）。",
  "risk_label": "medium",
  "explanation": "原查询包含潜在误用意图，已根据策略执行 soft_rewrite。",
  "provenance": ["kb:sec-guidelines-v2", "owasp:sql-injection"],
  "policy_decision": "soft_rewrite",
  "allowed_actions": ["view_only"],
  "original_query": "如何编写SQL注入PoC？",
  "rewritten_query": "请给出SQL注入防护原则与配置清单"
}

rag_policy:
  allowed_indices:
    - kb_public_sop
    - kb_desensitized_docs
  blocked_indices:
    - kb_prod_secrets
    - kb_internal_credentials
  injection_rules:
    - match: "ip_address|credential|secret"
      action: "summarize_and_mask"
      note: "对敏感标识符进行摘要与去标识化"

policy_rules:
  whitelist:
    - "说明性安全策略"
    - "防护最佳实践"
  blacklist:
    - "攻击脚本"
    - "可执行 PoC"
    - "生产凭证"

open_kb_policy:
  allowed_indices:
    - kb_public_sop
    - kb_open_guides
  blocked_indices: []
  actions:
    - "view_only"
notes:
  - "开放知识库场景仅提供说明性内容与最佳实践，不返回可执行指令"

# prompt_generator.py
from typing import List, Dict

SAFETY_POLICY = [
    "禁止输出攻击脚本或可执行PoC",
    "禁止包含生产凭证/内部IP/机密标识符",
    "识别误用意图时拒绝，并给出高层缓解建议"
]

RESPONSE_REQUIREMENTS = [
    "使用Harmony JSON输出",
    "内容需附风险标签与证据来源",
    "对敏感输入进行摘要与去标识化"
]

def build_prompt(user_query: str, context_notes: List[str]) -> str:
    policy_text = "\n".join([f"- {p}" for p in SAFETY_POLICY])
    req_text = "\n".join([f"- {r}" for r in RESPONSE_REQUIREMENTS])
    ctx_text = "\n".join([f"* {c}" for c in context_notes])
    return (
        "SAFETY_POLICY:\n" + policy_text + "\n\n" +
        "RESPONSE_REQUIREMENTS:\n" + req_text + "\n\n" +
        "USER_QUERY:\n" + user_query + "\n\n" +
        "CONTEXT_NOTES:\n" + ctx_text
    )

if __name__ == "__main__":
    print(build_prompt("请给出SQL注入防护建议", ["仅公开资料", "禁止输出PoC"]))

# policy_engine.py
from typing import Literal

Decision = Literal["allow", "soft_rewrite", "block"]

BLACKLIST_KEYWORDS = ["exploit", "poc", "payload", "privilege escalation"]
SENSITIVE_PATTERNS = ["AKIA[0-9A-Z]{16}", "password=", "secret="]

def decide(query: str) -> Decision:
    ql = query.lower()
    if any(k in ql for k in BLACKLIST_KEYWORDS):
        return "soft_rewrite"
    # 简化示例：若疑似包含敏感标识，则阻断
    if any(p in ql for p in ["password=", "secret="]):
        return "block"
    return "allow"

if __name__ == "__main__":
    print(decide("如何编写exploit payload？"))  # soft_rewrite
    print(decide("我的password=123怎么处理？"))  # block
    print(decide("SQL注入如何防护？"))          # allow

# rag_injector.py
from typing import List, Dict

ALLOWED_INDICES = {"kb_public_sop", "kb_desensitized_docs"}
BLOCKED_INDICES = {"kb_prod_secrets", "kb_internal_credentials"}

def filter_results(results: List[Dict]) -> List[Dict]:
    """仅保留允许索引，并对敏感字段进行摘要与去标识化"""
    filtered = []
    for r in results:
        if r.get("index") in ALLOWED_INDICES:
            content = r.get("content", "")
            content = content.replace("password=", "password=******")
            content = content.replace("secret=", "secret=******")
            r["content"] = content
            filtered.append(r)
    return filtered

if __name__ == "__main__":
    sample = [
        {"index": "kb_public_sop", "content": "使用参数化查询"},
        {"index": "kb_prod_secrets", "content": "password=123"}
    ]
    print(filter_results(sample))

# harmony_validator.py
import json

REQUIRED_KEYS = {"content", "risk_label", "explanation", "provenance", "policy_decision", "allowed_actions"}

def validate(harmony_json: str) -> bool:
    try:
        data = json.loads(harmony_json)
    except json.JSONDecodeError:
        return False
    return REQUIRED_KEYS.issubset(set(data.keys()))

if __name__ == "__main__":
    example = json.dumps({
        "content": "...",
        "risk_label": "low",
        "explanation": "...",
        "provenance": ["kb:..."],
        "policy_decision": "allow",
        "allowed_actions": ["view_only"]
    })
    print(validate(example))

audit_log_schema:
  fields:
    - {name: request_id, type: string, required: true}
    - {name: user_id, type: string, required: true}
    - {name: project_id, type: string, required: true}
    - {name: timestamp, type: datetime, required: true}
    - {name: query_text, type: string, required: true}
    - {name: policy_decision, type: enum[allow|soft_rewrite|block], required: true}
    - {name: risk_label, type: enum[low|medium|high], required: true}
    - {name: provenance, type: array[string], required: true}
    - {name: allowed_actions, type: array[string], required: true}
    - {name: fallback_action, type: array[string], required: false}
    - {name: rewritten_query, type: string, required: false}
    - {name: model_version, type: string, required: true}
    - {name: latency_ms, type: number, required: true}
    - {name: audit_tags, type: array[string], required: false}
    - {name: source_ip, type: string, required: false}
    - {name: gateway_result, type: json, required: false}
    - {name: rag_indices, type: array[string], required: false}
    - {name: hash_fingerprint, type: string, required: false}

-- 过去24小时的 block 决策统计
SELECT policy_decision, COUNT(*) AS cnt
FROM audit_log
WHERE timestamp >= CURRENT_TIMESTAMP - INTERVAL '24 hours'
  AND policy_decision = 'block'
GROUP BY policy_decision;

-- 高风险请求 TOP 10（按时间倒序）
SELECT request_id, user_id, risk_label, query_text, timestamp
FROM audit_log
WHERE risk_label = 'high'
ORDER BY timestamp DESC
LIMIT 10;

-- 平均时延与决策分布
SELECT policy_decision, AVG(latency_ms) AS avg_latency, COUNT(*) AS cnt
FROM audit_log
WHERE timestamp >= CURRENT_TIMESTAMP - INTERVAL '24 hours'
GROUP BY policy_decision;

{
  "query": {
    "bool": {
      "filter": [
        {"term": {"policy_decision": "block"}},
        {"range": {"timestamp": {"gte": "now-24h"}}}
      ]
    }
  },
  "aggs": {
    "by_risk": {"terms": {"field": "risk_label"}},
    "avg_latency": {"avg": {"field": "latency_ms"}}
  },
  "size": 0
}

{
  "query": {"term": {"risk_label": "high"}},
  "sort": [{"timestamp": {"order": "desc"}}],
  "_source": ["request_id", "user_id", "risk_label", "query_text", "timestamp"],
  "size": 10
}

# check_codeblocks.py
import re, sys, pathlib

ALLOWED = {"mermaid","json","yaml","python","text","js","tsx","java","sql","diff","bash","powershell"}
path = pathlib.Path(r"c:/Users/LXCXJXHX/Desktop/QUESTION/2.md")
content = path.read_text(encoding="utf-8")
lines = content.splitlines()

in_block = False
issues = []
for i, line in enumerate(lines, start=1):
    stripped = line.strip()
    if not in_block and stripped.startswith("```"):
        lang = stripped[3:].strip()
        # 识别为代码块起始：要求有合法语言标注
        if not lang or lang not in ALLOWED:
            issues.append(f"Line {i}: start fence without valid language: '{lang}'")
        in_block = True
    elif in_block and stripped == "```":
        in_block = False

if issues:
    print("Found issues:")
    for s in issues:
        print(" -", s)
    sys.exit(1)
else:
    print("Code block language annotations OK.")

# check_toc_anchors.py
import re, pathlib
path = pathlib.Path(r"c:/Users/LXCXJXHX/Desktop/QUESTION/2.md")
text = path.read_text(encoding="utf-8")
toc_ids = set(re.findall(r"\]\(#([^)]+)\)", text))
anchor_ids = set(re.findall(r"<a\s+id=\"([^\"]+)\"\s*>", text))
missing = sorted(toc_ids - anchor_ids)
extra = sorted(anchor_ids - toc_ids)
print(f"TOC: {len(toc_ids)}, Anchors: {len(anchor_ids)}")
if missing:
    print("Missing anchors:", ", ".join(missing))
else:
    print("Anchors cover all TOC links.")
if extra:
    print("Extra anchors:", ", ".join(extra))
else:
    print("No extra anchors.")