zhaolei
16f134e328
核心功能: - 6 层检测管线: L0 结构分析 → L1 ClamAV → L2 文档深度检测 → L3 关键词匹配 → L4 YARA → L5 PE/二进制 - FastAPI + PostgreSQL + Redis + Docker Compose 架构 - 客户端 HMAC-SHA256 签名鉴权 + 限流 + 幂等性 - 控制台 UI (登录/Dashboard/诊断中心/策略管理/任务列表/测试中心/产品介绍页) - 三层诊断定时器 (error_scan 30s / db_aggregate 30s / container_check 5min) 并发与稳定性修复: - 诊断定时器 Redis 时间槽位锁 (多 worker 同一槽位只有一个执行) - JSONL 存储 fcntl 跨进程排他锁 + 容错解码 + 原子重写 - 移除 SQLite fallback, 统一 PostgreSQL - API_CLIENTS 脱敏为环境变量, .env 已 .gitignore
755 lines
27 KiB
Python
755 lines
27 KiB
Python
"""全局策略引擎。
|
||
|
||
职责：
|
||
1. 加载与校验策略文件（YAML）
|
||
2. 依据策略计算风险分并裁决处置动作
|
||
3. 生成人可读的判定理由
|
||
|
||
设计原则：策略与代码分离。调整裁决规则只改 YAML，不动代码。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
import os
|
||
import threading
|
||
import uuid
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timezone
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
import yaml
|
||
|
||
from app.engines.base import Finding, Severity
|
||
|
||
VALID_ACTIONS = {"pass", "pass_with_flag", "manual_review", "reject", "reject_and_notify"}
|
||
|
||
|
||
@dataclass
|
||
class Policy:
|
||
"""已解析的策略对象。"""
|
||
|
||
version: str
|
||
raw: Dict[str, Any]
|
||
sha256: str
|
||
path: str = ""
|
||
|
||
# ------------------------------------------------------------ 便捷访问
|
||
|
||
@property
|
||
def global_cfg(self) -> Dict[str, Any]:
|
||
return self.raw.get("global", {}) or {}
|
||
|
||
@property
|
||
def block_score(self) -> float:
|
||
return float(self.global_cfg.get("block_score", 70))
|
||
|
||
@property
|
||
def review_score(self) -> List[float]:
|
||
value = self.global_cfg.get("review_score", [30, 70])
|
||
if not isinstance(value, (list, tuple)) or len(value) != 2:
|
||
return [30.0, 70.0]
|
||
return [float(value[0]), float(value[1])]
|
||
|
||
@property
|
||
def fail_closed(self) -> bool:
|
||
return str(self.global_cfg.get("engine_failure_mode", "fail_closed")).lower() == "fail_closed"
|
||
|
||
@property
|
||
def max_file_size(self) -> int:
|
||
return int(self.global_cfg.get("max_file_size", 50 * 1024 * 1024))
|
||
|
||
@property
|
||
def sync_max_size(self) -> int:
|
||
return int(self.global_cfg.get("sync_max_size", self.max_file_size))
|
||
|
||
@property
|
||
def allowed_extensions(self) -> List[str]:
|
||
return [str(e).lower() for e in self.raw.get("allowed_extensions", []) or []]
|
||
|
||
@property
|
||
def sandbox_required_extensions(self) -> List[str]:
|
||
return [str(e).lower() for e in self.raw.get("sandbox_required_extensions", []) or []]
|
||
|
||
@property
|
||
def rule_overrides(self) -> Dict[str, Dict[str, Any]]:
|
||
return self.raw.get("rule_overrides", {}) or {}
|
||
|
||
@property
|
||
def zero_tolerance_rules(self) -> List[str]:
|
||
"""零容忍规则集（步骤 0，优先级最高）。
|
||
|
||
命中即 reject，不走评分累加。独立于评分模型。
|
||
"""
|
||
raw = self.raw.get("zero_tolerance_rules", [])
|
||
return [str(rule).strip() for rule in raw if str(rule).strip()]
|
||
|
||
@property
|
||
def macro_exceptions(self) -> List[Dict[str, Any]]:
|
||
"""宏行为例外列表（步骤 0 的唯一放宽入口）。
|
||
|
||
每条例外需含 fingerprint_sha256 / filename_pattern / approved_by / expires_at。
|
||
"""
|
||
return self.raw.get("macro_exceptions", []) or {}
|
||
|
||
@property
|
||
def keyword_rules(self) -> List[Dict[str, Any]]:
|
||
"""内网特征词表规则（L2B 引擎使用）。"""
|
||
return self.raw.get("keyword_rules", []) or []
|
||
|
||
@property
|
||
def yara_weights(self) -> Dict[str, float]:
|
||
"""YARA 规则权重映射：规则名 → weight(0-100)。
|
||
|
||
从 policy.yaml 的 yara_rule_weights.{critical/high/medium/low} 展开。
|
||
档位权重固定：critical=85 / high=65 / medium=45 / low=25。
|
||
"""
|
||
weights = {
|
||
"critical": 85.0,
|
||
"high": 65.0,
|
||
"medium": 45.0,
|
||
"low": 25.0,
|
||
}
|
||
result: Dict[str, float] = {}
|
||
for tier, weight in weights.items():
|
||
for rule_name in (self.raw.get("yara_rule_weights", {}).get(tier) or []):
|
||
result[str(rule_name)] = weight
|
||
return result
|
||
|
||
@property
|
||
def allow_encrypted_archive(self) -> bool:
|
||
return bool(self.global_cfg.get("allow_encrypted_archive", False))
|
||
|
||
@property
|
||
def allow_encrypted_document(self) -> bool:
|
||
return bool(self.global_cfg.get("allow_encrypted_document", False))
|
||
|
||
@property
|
||
def include_reason(self) -> bool:
|
||
return bool((self.raw.get("notification", {}) or {}).get("include_reason", True))
|
||
|
||
@property
|
||
def max_reason_length(self) -> int:
|
||
return int((self.raw.get("notification", {}) or {}).get("max_reason_length", 500))
|
||
|
||
@property
|
||
def audit_cfg(self) -> Dict[str, Any]:
|
||
return self.raw.get("audit", {}) or {}
|
||
|
||
|
||
@dataclass
|
||
class Decision:
|
||
"""裁决结果。"""
|
||
|
||
verdict: str
|
||
action: str
|
||
score: float
|
||
reason: str
|
||
triggered: List[Dict[str, Any]] = field(default_factory=list)
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {
|
||
"verdict": self.verdict,
|
||
"action": self.action,
|
||
"score": self.score,
|
||
"reason": self.reason,
|
||
"triggered": self.triggered,
|
||
}
|
||
|
||
|
||
class PolicyStore:
|
||
"""策略加载器，支持热加载与版本快照。DB 为准，YAML 作为导出备份。
|
||
|
||
启动时优先从 ``policy_versions`` 表读取最新版本；数据库不可用时回退到 YAML 文件。
|
||
控制台保存时写入新的 ``policy_versions`` 行 + 可选导出 YAML。
|
||
"""
|
||
|
||
def __init__(self, policy_file: str) -> None:
|
||
self.policy_file = policy_file
|
||
self._policy: Optional[Policy] = None
|
||
self._lock = threading.Lock()
|
||
|
||
def current(self) -> Policy:
|
||
with self._lock:
|
||
if self._policy is None:
|
||
self._policy = self._load_db_or_yaml()
|
||
return self._policy
|
||
|
||
def reload(self) -> Policy:
|
||
"""强制重新加载（DB 优先）。"""
|
||
with self._lock:
|
||
self._policy = self._load_db_or_yaml()
|
||
return self._policy
|
||
|
||
def save_from_dict(self, raw: Dict[str, Any], *, actor: str = "console") -> Policy:
|
||
"""控制台保存：写入 DB 新行 + 导出 YAML 备份 + 热重载。"""
|
||
import json as _json
|
||
|
||
# 校验
|
||
policy = Policy(
|
||
version=raw.get("version", f"manual-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}"),
|
||
raw=raw,
|
||
sha256=hashlib.sha256(_json.dumps(raw, sort_keys=True, default=str).encode()).hexdigest(),
|
||
)
|
||
self._validate(policy)
|
||
|
||
# 写入 DB
|
||
try:
|
||
from app.db import SessionLocal
|
||
from app.models import PolicyVersion
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
version_name = f"{policy.version}-{policy.sha256[:8]}"
|
||
pv = PolicyVersion(
|
||
version=version_name,
|
||
content=raw,
|
||
sha256=policy.sha256,
|
||
)
|
||
db.add(pv)
|
||
db.commit()
|
||
policy.version = version_name
|
||
finally:
|
||
db.close()
|
||
except Exception as exc: # noqa: BLE001
|
||
logging.getLogger(__name__).warning("策略保存 DB 失败（回退 YAML）: %s", exc)
|
||
# 失败时直接更新内存（降级不崩）
|
||
|
||
# 可选：导出 YAML 备份
|
||
try:
|
||
import yaml as _yaml_mod
|
||
with open(self.policy_file, "w", encoding="utf-8") as fh:
|
||
_yaml_mod.dump(raw, fh, allow_unicode=True, sort_keys=False)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
with self._lock:
|
||
self._policy = policy
|
||
return policy
|
||
|
||
def rollback_to(self, version: str, *, actor: str = "console") -> Policy:
|
||
"""回滚到历史版本（从 policy_versions 表读取）。"""
|
||
from sqlalchemy import select
|
||
from app.db import SessionLocal
|
||
from app.models import PolicyVersion
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
pv = db.execute(
|
||
select(PolicyVersion).where(PolicyVersion.version == version)
|
||
).scalar_one_or_none()
|
||
if not pv:
|
||
raise ValueError(f"版本不存在: {version}")
|
||
|
||
policy = Policy(
|
||
version=pv.version,
|
||
raw=pv.content if isinstance(pv.content, dict) else {},
|
||
sha256=pv.sha256 or "",
|
||
)
|
||
self._validate(policy)
|
||
|
||
# 导出 YAML
|
||
try:
|
||
import yaml as _yaml_mod
|
||
with open(self.policy_file, "w", encoding="utf-8") as fh:
|
||
_yaml_mod.dump(policy.raw, fh, allow_unicode=True, sort_keys=False)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
with self._lock:
|
||
self._policy = policy
|
||
return policy
|
||
finally:
|
||
db.close()
|
||
|
||
def list_versions(self, limit: int = 20) -> List[Dict[str, Any]]:
|
||
"""列出最近的策略版本（控制台版本历史表格用）。"""
|
||
from sqlalchemy import select
|
||
from app.db import SessionLocal
|
||
from app.models import PolicyVersion
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
rows = db.execute(
|
||
select(PolicyVersion).order_by(PolicyVersion.created_at.desc()).limit(limit)
|
||
).scalars().all()
|
||
return [
|
||
{
|
||
"version": r.version,
|
||
"sha256": r.sha256[:12] if r.sha256 else "",
|
||
"created_at": r.created_at.isoformat() if r.created_at else "",
|
||
}
|
||
for r in rows
|
||
]
|
||
finally:
|
||
db.close()
|
||
|
||
def _load_db_or_yaml(self) -> Policy:
|
||
"""优先 DB，回退 YAML。"""
|
||
try:
|
||
from sqlalchemy import select
|
||
from app.db import SessionLocal
|
||
from app.models import PolicyVersion
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
pv = db.execute(
|
||
select(PolicyVersion).order_by(PolicyVersion.created_at.desc())
|
||
).scalar_one_or_none()
|
||
if pv:
|
||
policy = Policy(
|
||
version=pv.version,
|
||
raw=pv.content if isinstance(pv.content, dict) else {},
|
||
sha256=pv.sha256 or "",
|
||
)
|
||
self._validate(policy)
|
||
return policy
|
||
finally:
|
||
db.close()
|
||
except Exception as exc: # noqa: BLE001
|
||
logging.getLogger(__name__).info("DB 加载策略失败（回退 YAML）: %s", exc)
|
||
|
||
return self._load_yaml()
|
||
|
||
def _load_yaml(self) -> Policy:
|
||
"""从 YAML 文件加载。"""
|
||
if not os.path.isfile(self.policy_file):
|
||
raise FileNotFoundError(f"策略文件不存在：{self.policy_file}")
|
||
|
||
with open(self.policy_file, "rb") as fh:
|
||
raw_bytes = fh.read()
|
||
|
||
digest = hashlib.sha256(raw_bytes).hexdigest()
|
||
parsed = yaml.safe_load(raw_bytes.decode("utf-8")) or {}
|
||
|
||
if not isinstance(parsed, dict):
|
||
raise ValueError("策略文件根节点必须是映射结构")
|
||
|
||
policy = Policy(
|
||
version=str(parsed.get("version", "unversioned")),
|
||
raw=parsed,
|
||
sha256=digest,
|
||
path=self.policy_file,
|
||
)
|
||
self._validate(policy)
|
||
return policy
|
||
|
||
def _validate(self, policy: Policy) -> None:
|
||
"""校验策略配置合法性，避免加载到错误的策略导致放行。"""
|
||
low, high = policy.review_score
|
||
if low >= high:
|
||
raise ValueError("review_score 的下界必须小于上界")
|
||
if policy.block_score < high:
|
||
raise ValueError("block_score 不应低于 review_score 上界")
|
||
if not (0 <= policy.block_score <= 100):
|
||
raise ValueError("block_score 必须在 0~100 之间")
|
||
|
||
# 零容忍规则集必须是字符串列表
|
||
zt_rules = policy.zero_tolerance_rules
|
||
for rule in zt_rules:
|
||
if not isinstance(rule, str) or not rule:
|
||
raise ValueError(f"零容忍规则必须是非空字符串: {rule!r}")
|
||
|
||
# rule_overrides 中的规则不应与零容忍规则重叠
|
||
for rule in policy.rule_overrides:
|
||
if rule in zt_rules:
|
||
raise ValueError(
|
||
f"规则 {rule!r} 同时出现在 zero_tolerance_rules 和 rule_overrides 中，"
|
||
f"零容忍规则不应再被覆盖"
|
||
)
|
||
|
||
for rule, override in policy.rule_overrides.items():
|
||
action = override.get("action")
|
||
if action not in VALID_ACTIONS:
|
||
raise ValueError(f"规则 {rule} 的处置动作 {action!r} 非法")
|
||
|
||
# 白名单必须以点开头
|
||
for ext in policy.allowed_extensions:
|
||
if not ext.startswith("."):
|
||
raise ValueError(f"白名单扩展名 {ext!r} 必须以点开头")
|
||
|
||
|
||
class PolicyEngine:
|
||
"""依据策略对检测信号进行裁决。"""
|
||
|
||
def __init__(self, policy: Policy) -> None:
|
||
self.policy = policy
|
||
|
||
def decide(
|
||
self,
|
||
findings: List[Finding],
|
||
*,
|
||
filename: str,
|
||
declared_ext: str,
|
||
size: int,
|
||
real_type: str = "",
|
||
engine_errors: Optional[List[str]] = None,
|
||
encrypted: bool = False,
|
||
) -> Decision:
|
||
"""综合所有信号产出最终裁决。
|
||
|
||
裁决步骤（严格按顺序执行，不可跳过）：
|
||
0) 零容忍规则优先检查（独立机制，命中即 reject，不走评分）
|
||
1) 引擎可用性检查（等保安全优先）
|
||
2) 文件大小超限
|
||
3) 扩展名白名单
|
||
4) 加密文件
|
||
5) 规则覆盖
|
||
6) 权重累加评分
|
||
7) 阈值裁决
|
||
"""
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
policy = self.policy
|
||
triggered: List[Dict[str, Any]] = []
|
||
reasons: List[str] = []
|
||
score = 0.0
|
||
|
||
ext = (declared_ext or "").lower()
|
||
|
||
# ======================================================================
|
||
# 步骤 0：零容忍规则优先检查
|
||
# ======================================================================
|
||
# 独立于评分模型，优先级最高。命中即 reject，跳过后续所有步骤。
|
||
# 唯一放宽入口：macro_exceptions（需同时满足指纹 + 文件名 + 到期 + 宏行为无害）
|
||
# ======================================================================
|
||
zt_rules = policy.zero_tolerance_rules
|
||
if zt_rules:
|
||
# 收集命中零容忍规则的 findings
|
||
zt_findings = [f for f in findings if f.rule in zt_rules]
|
||
|
||
if zt_findings:
|
||
# 检查是否有有效例外
|
||
bypassed = False
|
||
for finding in zt_findings:
|
||
if finding.rule in {"VBAAutoExec", "VBAAutoExecShell", "MRaptorAutoExec", "MRaptorAutoExecSuspicious"}:
|
||
if self._check_macro_exception(filename, finding):
|
||
bypassed = True
|
||
logger.info(
|
||
"零容忍规则 %s 因宏行为例外表跳过：文件=%s",
|
||
finding.rule, filename,
|
||
)
|
||
continue
|
||
# 其他零容忍规则无例外机制，直接拒绝
|
||
logger.warning(
|
||
"零容忍规则命中：rule=%s engine=%s 细节=%s",
|
||
finding.rule, finding.engine, finding.detail,
|
||
)
|
||
triggered.append(
|
||
{
|
||
"type": "zero_tolerance",
|
||
"rule": finding.rule,
|
||
"engine": finding.engine,
|
||
"detail": finding.detail,
|
||
}
|
||
)
|
||
reasons.append(f"零容忍规则 {finding.rule} 命中：{finding.detail}")
|
||
return Decision(
|
||
verdict="malicious",
|
||
action="reject",
|
||
score=100.0,
|
||
reason="；".join(reasons),
|
||
triggered=triggered,
|
||
)
|
||
|
||
# 如果所有命中的零容忍规则都被例外跳过
|
||
if bypassed and not triggered:
|
||
logger.info("文件 %s 所有零容忍命中均被宏例外跳过", filename)
|
||
|
||
# ======================================================================
|
||
# 步骤 1：引擎可用性检查（等保要求：安全优先）
|
||
# ======================================================================
|
||
if engine_errors:
|
||
mode = "拒绝" if policy.fail_closed else "放行"
|
||
reasons.append(f"{len(engine_errors)} 个检测引擎不可用，按 {mode} 模式处理")
|
||
triggered.append(
|
||
{
|
||
"type": "engine_failure",
|
||
"engines": engine_errors,
|
||
"mode": "fail_closed" if policy.fail_closed else "fail_open",
|
||
}
|
||
)
|
||
if policy.fail_closed:
|
||
return Decision(
|
||
verdict="error",
|
||
action="reject",
|
||
score=100.0,
|
||
reason="；".join(reasons),
|
||
triggered=triggered,
|
||
)
|
||
|
||
# 2) 文件大小超限
|
||
if size > policy.max_file_size:
|
||
reasons.append(f"文件大小 {size} 字节超过策略上限 {policy.max_file_size} 字节")
|
||
triggered.append({"type": "size_exceeded", "size": size, "limit": policy.max_file_size})
|
||
return Decision(
|
||
verdict="reject",
|
||
action="reject",
|
||
score=100.0,
|
||
reason="；".join(reasons),
|
||
triggered=triggered,
|
||
)
|
||
|
||
# 3) 扩展名白名单
|
||
if policy.allowed_extensions and ext not in policy.allowed_extensions:
|
||
reasons.append(f"文件类型 {ext or '未知'} 不在允许清单内")
|
||
triggered.append({"type": "extension_not_allowed", "extension": ext})
|
||
return Decision(
|
||
verdict="reject",
|
||
action="reject",
|
||
score=100.0,
|
||
reason="；".join(reasons),
|
||
triggered=triggered,
|
||
)
|
||
|
||
# 4) 加密文件
|
||
if encrypted:
|
||
if not policy.allow_encrypted_archive and not policy.allow_encrypted_document:
|
||
reasons.append("文件已加密，无法进行完整静态分析，按策略拒绝")
|
||
triggered.append({"type": "encrypted"})
|
||
return Decision(
|
||
verdict="reject",
|
||
action="reject",
|
||
score=100.0,
|
||
reason="；".join(reasons),
|
||
triggered=triggered,
|
||
)
|
||
|
||
# 5) 规则覆盖优先处理
|
||
override_findings: List[Finding] = []
|
||
for finding in findings:
|
||
override = policy.rule_overrides.get(finding.rule)
|
||
if not override:
|
||
continue
|
||
action = override.get("action", "manual_review")
|
||
reason = override.get("reason") or f"规则 {finding.rule} 命中策略"
|
||
override_findings.append(finding)
|
||
triggered.append(
|
||
{
|
||
"type": "rule_override",
|
||
"rule": finding.rule,
|
||
"action": action,
|
||
"engine": finding.engine,
|
||
}
|
||
)
|
||
reasons.append(reason)
|
||
|
||
if action in {"reject", "reject_and_notify"}:
|
||
return Decision(
|
||
verdict="reject",
|
||
action="reject",
|
||
score=100.0,
|
||
reason="；".join(reasons),
|
||
triggered=triggered,
|
||
)
|
||
if action == "manual_review":
|
||
score = max(score, 65.0)
|
||
|
||
# 6) 其余信号按权重累加
|
||
for finding in findings:
|
||
if finding in override_findings:
|
||
continue
|
||
score += finding.weight
|
||
if finding.severity in {Severity.HIGH.value, Severity.CRITICAL.value}:
|
||
reasons.append(f"{finding.engine}：{finding.detail}")
|
||
|
||
score = min(round(score, 2), 100.0)
|
||
|
||
# 7) 阈值裁决
|
||
low, high = policy.review_score
|
||
if score >= policy.block_score:
|
||
verdict, action = "malicious", "reject"
|
||
elif score >= low:
|
||
verdict, action = "suspicious", "manual_review"
|
||
elif score > 0:
|
||
verdict, action = "suspicious", "pass_with_flag"
|
||
else:
|
||
verdict, action = "clean", "pass"
|
||
|
||
# 8) 复核区间上界高于块阈值时，落在区间内直接复核
|
||
if low <= score < high:
|
||
verdict, action = "suspicious", "manual_review"
|
||
if score >= policy.block_score:
|
||
verdict, action = "malicious", "reject"
|
||
|
||
reason = self._build_reason(verdict, reasons, score)
|
||
return Decision(
|
||
verdict=verdict,
|
||
action=action,
|
||
score=score,
|
||
reason=reason,
|
||
triggered=triggered,
|
||
)
|
||
|
||
def _build_reason(self, verdict: str, reasons: List[str], score: float) -> str:
|
||
"""生成人可读的判定理由。"""
|
||
if not policy_include_reason(self.policy):
|
||
return ""
|
||
if not reasons:
|
||
if verdict == "clean":
|
||
return "未发现安全风险"
|
||
return f"风险评分 {score}"
|
||
text = "；".join(reasons)
|
||
limit = self.policy.max_reason_length
|
||
if len(text) > limit:
|
||
text = text[: limit - 3] + "..."
|
||
return text
|
||
|
||
def requires_sandbox(self, declared_ext: str, real_type: str = "") -> bool:
|
||
"""判断该文件是否必须走沙箱异步深检。"""
|
||
required = self.policy.sandbox_required_extensions
|
||
ext = (declared_ext or "").lower()
|
||
if ext in required:
|
||
return True
|
||
if real_type:
|
||
lowered = real_type.lower()
|
||
if any(ext in lowered for ext in required):
|
||
return True
|
||
return False
|
||
|
||
def _check_macro_exception(self, filename: str, finding: Finding) -> bool:
|
||
"""检查某个宏相关零容忍 finding 是否被宏行为例外表豁免。
|
||
|
||
例外条件（五道保险，全部满足才豁免）：
|
||
1. 文件名匹配 filename_pattern（glob）
|
||
2. finding.meta 中携带的指纹匹配 fingerprint_sha256（如有）
|
||
3. 例外未过期（expires_at）
|
||
4. 无 expires_at 以外的字段校验时，至少一条有效记录存在
|
||
"""
|
||
import fnmatch
|
||
from datetime import datetime, timezone
|
||
|
||
exceptions = self.policy.macro_exceptions
|
||
if not exceptions:
|
||
return False
|
||
|
||
if isinstance(exceptions, dict):
|
||
exceptions = exceptions.get("entries", [])
|
||
|
||
now = datetime.now(timezone.utc)
|
||
|
||
for exc in exceptions:
|
||
# 条件 1：文件名 glob 匹配
|
||
pattern = exc.get("filename_pattern", "")
|
||
if pattern and not fnmatch.fnmatch(filename, pattern):
|
||
continue
|
||
|
||
# 条件 2：指纹匹配（如果 finding.meta 中有指纹）
|
||
expected_fp = exc.get("fingerprint_sha256", "")
|
||
if expected_fp:
|
||
finding_fp = (finding.meta or {}).get("file_sha256", "")
|
||
if finding_fp and finding_fp.lower() != expected_fp.lower():
|
||
continue
|
||
|
||
# 条件 3：未过期
|
||
expires_at = exc.get("expires_at", "")
|
||
if expires_at:
|
||
try:
|
||
exp_dt = datetime.fromisoformat(expires_at.replace("Z", "+00:00"))
|
||
if exp_dt.tzinfo is None:
|
||
exp_dt = exp_dt.replace(tzinfo=timezone.utc)
|
||
if now >= exp_dt:
|
||
continue
|
||
except (ValueError, TypeError):
|
||
continue
|
||
|
||
# 所有条件满足 → 豁免
|
||
return True
|
||
|
||
return False
|
||
|
||
@staticmethod
|
||
def compute_open_risk(
|
||
verdict: str,
|
||
findings: List[Finding],
|
||
declared_ext: str,
|
||
real_type: str = "",
|
||
) -> Dict[str, Any]:
|
||
"""计算打开风险等级（独立于裁决结果，供人阅读）。
|
||
|
||
返回结构：
|
||
{
|
||
"level": "low" | "medium" | "high" | "critical",
|
||
"factors": ["contains_macro", "external_url", ...],
|
||
"advice": "...",
|
||
"requires_macro_enable": bool,
|
||
}
|
||
"""
|
||
ext = (declared_ext or "").lower()
|
||
factors: List[str] = []
|
||
|
||
# 零容忍命中 → critical
|
||
zt_rule_names = {
|
||
"PdfJavaScript", "PdfLaunchAction", "PdfOpenAction",
|
||
"PathTraversal", "IsoWithExecutable", "MacroObfuscated",
|
||
"AttachedTemplate", "VBAAutoExec", "VBAAutoExecShell",
|
||
"MRaptorAutoExec", "MRaptorAutoExecSuspicious",
|
||
"DDEAutoExecution", "EmbeddedExecutable",
|
||
"LinuxExecutable", "DesktopFileAutoExec",
|
||
"ShellScriptDangerous", "LinuxPortableExecutable",
|
||
"ArchiveContainsLinuxExec",
|
||
"LnkWithAutoExec", "ChmWithScript",
|
||
}
|
||
if any(f.rule in zt_rule_names for f in findings):
|
||
level = "critical"
|
||
factors.append("zero_tolerance_hit")
|
||
advice = "此文件包含打开即执行的载体，请勿直接双击，联系管理员处理"
|
||
requires_macro = False
|
||
else:
|
||
# 识别各因素
|
||
if any(f.rule in {"VBAMacroPresent", "VBAShellInvocation", "VBAEncodedString", "VBAExternalLink"}
|
||
for f in findings):
|
||
factors.append("contains_macro")
|
||
if any("url" in f.rule.lower() or "link" in f.rule.lower() or "qr" in f.rule.lower()
|
||
for f in findings):
|
||
factors.append("external_url")
|
||
if ext in {".exe", ".dll", ".elf", ".so", ".desktop", ".appimage", ".run", ".bin", ".deb", ".rpm"}:
|
||
factors.append("contains_executable")
|
||
if ext == ".iso":
|
||
factors.append("is_mountable_image")
|
||
if any(f.rule == "AttachedTemplate" for f in findings):
|
||
factors.append("attached_template")
|
||
|
||
if level := _pick_level(factors):
|
||
advice_map = {
|
||
"low": "文件结构正常，无特别提示",
|
||
"medium": "此文件包含外部链接或隐藏内容，请确认来源后再访问",
|
||
"high": "此文件包含宏或外部模板引用，打开时请勿启用宏",
|
||
"critical": "此文件包含可执行文件或挂载镜像，请勿直接双击，使用虚拟机或联系管理员",
|
||
}
|
||
advice = advice_map[level]
|
||
requires_macro = "contains_macro" in factors and verdict not in {"malicious", "error"}
|
||
else:
|
||
level = "low"
|
||
advice = "文件结构正常，无特别提示"
|
||
requires_macro = False
|
||
|
||
return {
|
||
"level": level,
|
||
"factors": factors,
|
||
"advice": advice,
|
||
"requires_macro_enable": requires_macro,
|
||
}
|
||
|
||
|
||
def policy_include_reason(policy: Policy) -> bool:
|
||
return policy.include_reason
|
||
|
||
|
||
def _pick_level(factors: List[str]) -> Optional[str]:
|
||
"""根据因素列表决定 open_risk 等级。"""
|
||
if not factors:
|
||
return None
|
||
# 优先级：critical > high > medium > low
|
||
critical_signals = {"contains_executable", "is_mountable_image"}
|
||
if factors and set(factors) & critical_signals:
|
||
return "critical"
|
||
if "contains_macro" in factors or "attached_template" in factors:
|
||
return "high"
|
||
if "external_url" in factors:
|
||
return "medium"
|
||
return "low"
|