利用Maltego平台生成攻击溯源图

发表于 | 分类于

前言

       最近没啥事打开kali想玩玩没有用过的工具，就打开了Maltego，很早以前用过但没有深入的研究过，就去官网看看文档，对自己公司收集一些信息，发现画出来的图还不错，其中的关联关系表达较为清晰。刚好看到Maltego支持自定义的Transform,突然灵机一动，最近公司在研究攻击溯源图，那我能否通过自定义的Transform来生成一份攻击溯源图呢？

行动开始

       我们从elasticsearch中抽取告警数据和日志数据来进行分析，所以这里先写好一个需要调用的es查询语句的class,由于Maltego本身是一个社工平台，一般不会由多个元素作为条件去溯源数据，所以本身的框架只支持引入一个上层变量，其他的变量需要手动输入，所以这里写的es查询语句中，我依照现有的数据把src_address字段写死。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
#!/usr/bin/python
# -*- coding: UTF-8 -*-
# developer: Lawliet
# dsl.py

class dsl(object):
    # 定义ES地址
    def es_ip(self):
        es_host_ip = "172.16.100.29"
        return es_host_ip
    # 根据源地址获取告警
    def Dsl_alert_from_srcip(self, src_ip):
        body = {
            "size": 0,
            "query": {
                "bool": {
                    "must": [
                        {
                            "term": {
                                "src_address": src_ip
                            }
                        }
                    ]
                }
            },
            "aggs": {
                "group_by_alarm_name": {
                    "terms": {
                        "field": "alarm_name",
                        "size": 100
                    }
                }
            }
        }
        return body
    # 根据告警获取攻击源地址
    def Dsl_srcip_from_alarm(self, alarm_name):
        body = {
            "size": 0,
            "query": {
                "bool": {
                    "must": [
                        {
                            "term": {
                                "alarm_name": alarm_name
                            }
                        }
                    ]
                }
            },
            "aggs": {
                "group_by_srcip": {
                    "terms": {
                        "field": "src_address",
                        "size": 100
                    }
                }
            }
        }
        return body

    # 根据源地址和目的端口获取目的地址
    def Dsl_dstip_from_srcip_and_dstport(self, dst_port):
        body = {
            "size": 0,
            "query": {
                "bool": {
                    "must": [
                        {
                            "term": {
                                "dst_port": dst_port
                            }
                        },
                        {
                            "term": {
                                "src_address": "172.16.100.21"
                            }
                        }
                    ]
                }
            },
            "aggs": {
                "group_by_dstip": {
                    "terms": {
                        "field": "dst_address",
                        "size": 100
                    }
                }
            }
        }
        return body

    # 根据源地址和告警获取目的端口
    def Dsl_dstport_from_alarm_srcip(self, alarm_name):
        body = {
            "size": 0,
            "query": {
                "bool": {
                    "must": [
                        {
                            "term": {
                                "alarm_name": alarm_name
                            }
                        },
                        {
                            "term": {
                                "src_address": "172.16.100.21"
                            }
                        }
                    ]
                }
            },
            "aggs": {
                "group_by_dstport": {
                    "terms": {
                        "field": "dst_port",
                        "size": 100
                    }
                }
            }
        }
        return body

    # 根据源地址和告警获取域名
    def Dsl_domain_from_alarm_srcip(self, alarm_name):
        body = {
            "size": 0,
            "query": {
                "bool": {
                    "must": [
                        {
                            "term": {
                                "alarm_name": alarm_name
                            }
                        },
                        {
                            "term": {
                                "src_address": "172.16.100.21"
                            }
                        }
                    ]
                }
            },
            "aggs": {
                "group_by_domain": {
                    "terms": {
                        "field": "domain_name",
                        "size": 100
                    }
                }
            }
        }
        return body

    # 根据源地址和告警获取目的地址
    def Dsl_dstip_from_alarm_srcip(self, alarm_name):
        body = {
            "size": 0,
            "query": {
                "bool": {
                    "must": [
                        {
                            "term": {
                                "alarm_name": alarm_name
                            }
                        },
                        {
                            "term": {
                                "src_address": "172.16.100.21"
                            }
                        }
                    ]
                }
            },
            "aggs": {
                "group_by_dstip": {
                    "terms": {
                        "field": "dst_address",
                        "size": 100
                    }
                }
            }
        }
        return body

       当发生了一个告警时，我们想要通过该告警去溯源整个事件发生的过程及影响范围，图例：我们检测到发生“主机扫描特定端口后发起针对该端口的攻击”，通过自定义Transform来查询发生该告警的源地址，Maltego支持任何脚本，因为Maltego直接调用系统命令执行脚本。
image

Transform代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
#!/usr/bin/python
# -*- coding: utf-8 -*-
# Author: Lawliet
# alarm_to_srcip.py

from elasticsearch2 import Elasticsearch
from MaltegoTransform import *
from dsl import *
import sys
reload(sys)
sys.setdefaultencoding('utf-8')


def alarm_to_ip():
    dsl_search = dsl()
    es_host_ip = dsl_search.es_ip()
    es = Elasticsearch([{'host': es_host_ip, 'port': 9200}])
    body = dsl_search.Dsl_srcip_from_alarm(alarm_name=sys.argv[1])
    res = es.search(index='incident_alarm_20181106', body=body, request_timeout=60)
    alarm_list = []
    alarm_count = []
    for aggs in res['aggregations']['group_by_srcip']['buckets']:
        alarm_list.append([aggs['key'], aggs['doc_count']])
    me = MaltegoTransform()
    for i in alarm_list:
        me.addEntity("maltego.IPv4Address", i[0])
    me.returnOutput()
if __name__ == '__main__':
    alarm_to_ip()

       添加自定义Transform操作如下，“Entity Value”指的是上层实例的值作为参数传递进脚本进行查询，filed为自定义参数。
image

这里一共新增了6个我们需要用到的溯源逻辑

  1. 根据告警搜索源地址
  2. 根据源地址搜索告警
  3. 根据源地址和告警搜索域名
  4. 根据源地址和告警搜索目的地址
  5. 根据源地址和告警搜索目的端口
  6. 更具源地址和目的端口搜索目的地址
    image

    效果展示

           通过定义好Transform最终溯源出下图，随着定义的Transform越来越多，能覆盖更多的溯源场景。
    image

    总结

  • 首先Maltego平台的绘图能力强，但是Transform只能引用上一层的“Entity Value”来做过滤条件，导致脚本无法自动产生多个过滤条件，所以无法直接使用Maltego来绘画溯源图，本文只是从Maltego借鉴其优点。
  • Maltego在查询时有incoming和outgoing的概念，借此概念来绘制线条方向，点击实例时，incoming和outgoing线条加粗，可下钻到其他的实例查看相关信息，整个Detail view栏也是不错的功能，可以看到数据类型和该数据的来源。
    image
    image
    image
  • Maltego通过预定义脚本来查询数据并展示结果，如果通过封装的Transform实现对日志的图形化分析，效果会如何呢？