自动化管道与数据工程

from attackcti import attack_client

# AttackClient 默认从 MITRE CTI 官方源加载指定版本数据
lift = attack_client.AttackClient()

# 检索具体技术对象（返回 attack-pattern SDO）
t1003 = lift.get_technique_by_id("T1003")
print(t1003.name, t1003.x_mitre_version)

# 遍历 Enterprise 全部技术，统计带云平台的条目
enterprise = lift.get_techniques(workspace="enterprise-attack")
cloud_techs = [t for t in enterprise
              if "Cloud" in getattr(t, "x_mitre_platforms", [])]
print("Enterprise 中涉及 Cloud 平台的技术数:", len(cloud_techs))

# 取得某组织及其实战使用的技术（uses 关系）
group = lift.get_group_by_id("G0096")   # 例：Salt Typhoon(v17 新增)
used = lift.get_techniques_used_by_group(group)

import stix2
from stix2 import parse

with open("enterprise-attack.json", "r", encoding="utf-8") as f:
    bundle = parse(f.read())   # 返回 stix2.Bundle

for obj in bundle.objects:
    if obj.type == "attack-pattern":
        # 由 external_references 中的 T 编号定位
        for ref in obj.external_references:
            if ref.source_name == "mitre-attack":
                print(ref.external_id, obj.name)

import json

# 每行: (技术编号, 评分, 证据注释)
mappings = [
    ("T1003", 75, "Sysmon ID10 + EDR; 原子测试 2026-06 通过"),
    ("T1059.001", 50, "PowerShell 日志规则已部署, 未实测"),
    ("T1557.002", 0, "缺网络侧 ARP/NDP 遥测, 已知盲区"),
]

layer = {
    "name": "企业检测覆盖度基线 2026Q3",
    "versions": {"attack": "17", "navigator": "5.x"},
    "domain": "enterprise-attack",
    "description": "L2规则覆盖评分: 0=无 25=L1 50=L2 75=L3 100=L3+自动化",
    "gradient": {"colors": ["#ff6666", "#ffe766", "#8ec843"],
                 "minValue": 0, "maxValue": 100},
    "techniques": [
        {"techniqueID": tid, "score": score, "comment": note}
        for tid, score, note in mappings
    ],
    "filters": {"platforms": ["Windows", "Linux"]},
}

with open("coverage-baseline.layer.json", "w", encoding="utf-8") as f:
    json.dump(layer, f, ensure_ascii=False, indent=2)

import stix2

bundle = stix2.Bundle(
    stix2.Relationship(
        source_ref="intrusion-set--<组织STIX-ID>",
        target_ref="attack-pattern--<技术STIX-ID>",
        relationship_type="uses",
    )
)
print(bundle.serialize())

title: 可疑 LSASS 内存访问 (T1003.001)
id: 9f9b35d6-7c8a-4b2e-9d3a-1c2b3d4e5f60
status: experimental
description: 检测非系统进程以读取权限访问 lsass.exe
references:
  - https://attack.mitre.org/techniques/T1003/001/
tags:
  - attack.credential-access
  - attack.t1003.001
  - attack.t1003
logsource:
  category: process_access
detection:
  selection:
    TargetImage|endswith: '\\lsass.exe'
    GrantedAccess:
      - '0x1010'
      - '0x1410'
  filter:
    SourceImage|endswith:
      - '\\wininit.exe'
      - '\\lsass.exe'
  condition: selection and not filter
falsepositives:
  - 合法安全软件内存扫描
level: high

# 安装 CLI 与目标后端插件：pip install sigma-cli
# 编译为 Splunk SPL（-t 后端，-p 管线）
sigma convert -t splunk -p splunk_windows rules/malicious_lsass_access.yml

# 编译为 Microsoft Sentinel KQL
sigma convert -t kusto -p microsoft365_defender rules/mfa_bombing.yml

# 批量校验规则并输出 ATT&CK 标签审计（自建脚本思路）
sigma check rules/ | grep -o 'attack\.t[0-9.]*' | sort -u

import requests

TAXII_URL = "https://your-tip.example.com/taxii2/collections//objects"
headers = {"Authorization": "Bearer <TOKEN>"}

# 将新增的 STIX Bundle 推送到下游 TIP
resp = requests.post(
    TAXII_URL,
    headers={**headers, "Content-Type": "application/taxii+json;version=2.1"},
    json={"objects": [obj.serialize() for obj in new_objects]},
)
print("TAXII 推送状态:", resp.status_code)