钓鱼攻击-常规邮箱验证存活

常规邮箱验证存活

目标邮箱搜集

搜集目标邮箱的一些常规途径：

1、空间搜索引擎：

hunter 、 skymem 、email-format.com

2、搜索引擎：Github、招聘网站、社交软件

3、门户网站： 联系/加入我们 ， 文章资讯（项目征集

xlsx,xls,doc等、近期通告）、登陆/注册

4、邮箱用户枚举 ： 并行组合多种协议，如smtp、IMAP、POP3、web接口等

img

验证目标邮箱存活

有些目标邮箱,中途由于各种各样的原因,可能早已被删除，然而这种邮箱对我们是毫无意义的,发信的目的是想尽可能多的精准命中目标,而不单单只是为了发个信而发信,所以,一次精密的钓鱼行动,尽可能把能想到的意外,全部提前避免掉,一击必中。

使用SMTP协议执行电子邮件地址的验证。有几个SMTP命令可以用。比如 EXPN ， VRFY和 RCPT TO 。 现在的邮件系统很少启用EXPN 和 VRFY，所以最常见的验证命令是 RCPT TO，通过这种被动搜集方式,可以撞到很多从其它地方压根都搜集不到的邮箱。

通讯流程：

  1. 查找目标邮件服务器的MX记录。

img

  1. 与目标邮件服务器建立连接。
  2. 发送HELO或EHLO命令。
  3. 发送MAIL FROM（发件人）命令。
  4. 发送RCPT TO（收件人）命令。

img

1
2
3
4
5
6
7
8
9
10
11
12
13
mx记录解析成功:61.xxx.49.98
连接到:61.xxx.49.98成功！
HELO chacuo.net
>220 bizmx11.qq.com MX QQ Mail Server

MAIL FROM:<4554434523@163.com>
>250 bizmx11.qq.com

RCPT TO:<linrr@cecis.com.cn>
>250 Ok

QUIT
>250 Ok

当客户端将令发送到服务器时，服务器将接受该电子邮件或报告错误。

如果服务器识别地址并接受它，它将返回代码250或251。当我们收到此代码时，表示电子邮件地址有效500到599之间的代码 保留用于永久错误，这说明电子邮件地址不存在

Python

img

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
import random
import smtplib
import logging
import requests
import dns.resolver
import queue
import threading
import argparse

queue = queue.Queue()
lock = threading.Lock()

logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s - %(filename)s [line:%(lineno)d] - %(levelname)s: %(message)s')
logger = logging.getLogger()


class start():

    def __init__(self,domain):
        self.api = "66aee57fd0dc7e47a24b8b2ad2bc26931b3d015c"
        self.target = domain
        self.search_mail = "https://api.hunter.io/v2/domain-search?domain={}&api_key={}".format(self.target, self.api)

    def manage_response(self,data):
        emails = []
        try:
            for email in data['data']['emails']:
                print(str(email['value']))
        except Exception:
            print("Could not find any information about that")
            emails = "-"
        return emails

    def send_request(self,url):
        try:
            response = requests.get(url, timeout=5, allow_redirects=True)
            return response.json()
        except Exception:
            return None

    def search(self):
        """
        Main function of this tool
        """
        try:
            print(self.search_mail)
            response = self.send_request(self.search_mail)
            self.manage_response(response)
        except Exception as exception:
            print("Error in main function" + str(exception))


class ThreadUrl(threading.Thread):
    def __init__(self, queue,domain):
        threading.Thread.__init__(self)
        self.queue = queue
        self.target = domain


    def fetch_mx(self,host):
        '''
        解析服务邮箱
        :param host:
        :return:
        '''
        answers = dns.resolver.query(host, 'MX')
        res = [str(rdata.exchange)[:-1] for rdata in answers]
        return res

    def verify_istrue(self,email):
        '''
        :param email:
        :return:
        '''
        email_list = []
        email_obj = {}
        final_res = {}
        if isinstance(email, str) or isinstance(email, bytes):
            email_list.append(email)
        else:
            email_list = email

        for em in email_list:
            name, host = em.split('@')
            if email_obj.get(host):
                email_obj[host].append(em)
            else:
                email_obj[host] = [em]

        for key in email_obj.keys():
            host = random.choice(self.fetch_mx(key))
            s = smtplib.SMTP(host, timeout=10)
            for need_verify in email_obj[key]:
                s.docmd('HELO chacuo.net')
                s.docmd('MAIL FROM:<test@chacuo.net>')
                send_from = s.docmd('RCPT TO:<%s>' % need_verify)
                if send_from[0] == 250 or send_from[0] == 451:
                    final_res[need_verify] = True  # 存在
                elif send_from[0] == 550:
                    final_res[need_verify] = False  # 不存在
                else:
                    final_res[need_verify] = None  # 未知

            s.close()

        return final_res


    def run(self):
        while True:
            mail = self.queue.get()
            final_list = self.verify_istrue(mail + "@" + self.target)
            print(final_list)
            self.queue.task_done()

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Pscan')
    parser.add_argument("-m", "--mode", help='''****************Choose an scan mode****************\n
            ***************verify:   Mailbox survival verification ***************\n
            ****search:   Mailbox collection *****''', required=True)
    parser.add_argument("--host", help="please input host",required=True)
    parser.add_argument("--txt", help="please input user,",default='mails.txt',required=True)
    parser.add_argument("--thread", help="please input your thread", default=20, type=int)
    args = parser.parse_args()
    if args.mode == "verify":
        answers = dns.resolver.query(args.host, 'MX')
        res = [str(rdata.exchange)[:-1] for rdata in answers]
        logger.info('查找结果为：%s' % res)
        for i in range(args.thread):
            t = ThreadUrl(queue,args.host)
            t.setDaemon(True)
            t.start()
        with open(args.txt) as inFile:
            for i in inFile.readlines():
                queue.put(i.strip())
        queue.join()
    elif args.mode == "search":
        hunte = start(domain=args.host)
        hunte.search()