Python poplib 模块，POP3_SSL 实例源码

def checkMailAccount(server,user,password,ssl=False,port=None):
    '''
    ????Mail??????
    '''
    if not port:
        port = 995 if ssl else 110

    try:
        pop3 = poplib.POP3_SSL(server, port) if ssl else poplib.POP3(server, port)

        pop3.user(user)
        auth = pop3.pass_(password)
        pop3.quit()
    except Exception as error:
        #print "[!] chekcing {0} failed, reason:{1}".format(user, str(error))
        return False

    if "+OK" in auth:
        return True
    else:
        return False

def run(self):
        value, user = getword()
        user = user.replace("\n","")
        value = value.replace("\n","")

        try:
            print "-"*12
            print "[+] User:",user,"Password:",value
            pop = poplib.POP3_SSL(server, 995)
            pop.user(user)
            pop.pass_(value)
            print "\t\t\n\nLogin successful:",user, value
            print "\t\tMail:",pop.stat()[0],"emails"
            print "\t\tSize:",pop.stat()[1],"bytes\n\n"
            success.append(user)
            success.append(value)
            success.append(pop.stat()[0])
            success.append(pop.stat()[1])
            pop.quit()
        except (poplib.error_proto), msg: 
            #print "An error occurred:", msg
            pass

def connect(self):
        """Connects to and authenticates with a POP3 mail server"""

        import poplib

        M = None
        try:
            if (self.keyfile and self.certfile) or self.ssl:
                M = poplib.POP3_SSL(self.host, self.port, self.keyfile, self.certfile)
            else:
                M = poplib.POP3(self.host, self.port)

            M.user(self.username)
            M.pass_(self.password)
        except socket.error, err:
            raise
        else:
            return M

def test_context(self):
            ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
            self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
                              self.server.port, keyfile=CERTFILE, context=ctx)
            self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
                              self.server.port, certfile=CERTFILE, context=ctx)
            self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
                              self.server.port, keyfile=CERTFILE,
                              certfile=CERTFILE, context=ctx)

            self.client.quit()
            self.client = poplib.POP3_SSL(self.server.host, self.server.port,
                                          context=ctx)
            self.assertIsInstance(self.client.sock, ssl.SSLSocket)
            self.assertIs(self.client.sock.context, ctx)
            self.assertTrue(self.client.noop().startswith(b'+OK'))

def test_context(self):
            ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
            self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
                              self.server.port, keyfile=CERTFILE, context=ctx)
            self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
                              self.server.port, certfile=CERTFILE, context=ctx)
            self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
                              self.server.port, keyfile=CERTFILE,
                              certfile=CERTFILE, context=ctx)

            self.client.quit()
            self.client = poplib.POP3_SSL(self.server.host, self.server.port,
                                          context=ctx)
            self.assertIsInstance(self.client.sock, ssl.SSLSocket)
            self.assertIs(self.client.sock.context, ctx)
            self.assertTrue(self.client.noop().startswith(b'+OK'))

def get_vcode(self, mail_user, mail_passwd, mail_host='pop.163.com'):
        # ???POP3???:
        server = poplib.POP3_SSL(mail_host)
        # ???????????:
        # server.set_debuglevel(1)
        # ??:??POP3????????:
        # print(server.getwelcome())
        # ????:
        server.user(mail_user)
        server.pass_(mail_passwd)
        # list()?????????:
        resp, mails, octets = server.list()
        index = len(mails)
        # ??????
        resp, lines, octets = server.retr(index)
        msg = '\r\n'.join([one.decode('utf8') for one in lines])
        if 'security' in msg:
            res = re.findall('\s\d{6}\s', msg)[0].strip()
        else:
            res = None
        # ???????????????????:
        # server.dele(index)
        # ????:
        server.quit()
        return res

def test_context(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
        self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
                            self.server.port, keyfile=CERTFILE, context=ctx)
        self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
                            self.server.port, certfile=CERTFILE, context=ctx)
        self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
                            self.server.port, keyfile=CERTFILE,
                            certfile=CERTFILE, context=ctx)

        self.client.quit()
        self.client = poplib.POP3_SSL(self.server.host, self.server.port,
                                        context=ctx)
        self.assertIsInstance(self.client.sock, ssl.SSLSocket)
        self.assertIs(self.client.sock.context, ctx)
        self.assertTrue(self.client.noop().startswith(b'+OK'))

def test_authenticate_success(self, mock_pop3_ssl):
        """
        Test ``authenticate`` for a valid user.

        :param mock_pop3_ssl: mock path for `poplib.POP3_SSL``
        :return: None
        """
        request = HttpRequest()
        response = mock_pop3_ssl.return_value
        response.username.return_value = self.user.username
        response.pass_.return_value = b'+OK'

        credentials = {
            'username': 'username',
            'password': 'password',
            'login_server': '202.141.80.10',
            'port': '995'
        }
        auth_backend = WebMailAuthenticationBackend()
        self.assertEqual(auth_backend.authenticate(request, **credentials), self.user)

def test_authenticate_poplib_exception(self, mock_pop3_ssl):
        """
        Test ``authenticate`` for ``poplib.error_proto`` exception.

        :param mock_pop3_ssl: mock path for `poplib.POP3_SSL``
        :return: None
        """
        mock_pop3_ssl.side_effect = poplib.error_proto
        # response.side_effect = poplib.error_proto
        # response.username.return_value = self.user.username
        # response.pass_.return_value = b'+OK'
        request = HttpRequest()
        credentials = {
            'username': 'username',
            'password': 'password',
            'login_server': '202.141.80.10',
            'port': '995'
        }
        auth_backend = WebMailAuthenticationBackend()
        self.assertIsNone(auth_backend.authenticate(request, **credentials))

def test_authenticate_value_type_exception(self, mock_pop3_ssl):
        """
        Test ``authenticate`` for ``ValueError`` or ``TypeError``
        exception.

        :param mock_pop3_ssl: mock path for `poplib.POP3_SSL``
        :return: None
        """
        request = HttpRequest()
        response = mock_pop3_ssl.return_value
        response.username.return_value = 123456
        response.pass_.return_value = 12345

        credentials = {
            'username': 'username',
            'password': 'password',
            'login_server': '202.141.80.10',
            'port': '995'
        }
        auth_backend = WebMailAuthenticationBackend()
        self.assertRaises(TypeError, auth_backend.authenticate, request, **credentials)

def test_context(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
        self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
                            self.server.port, keyfile=CERTFILE, context=ctx)
        self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
                            self.server.port, certfile=CERTFILE, context=ctx)
        self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
                            self.server.port, keyfile=CERTFILE,
                            certfile=CERTFILE, context=ctx)

        self.client.quit()
        self.client = poplib.POP3_SSL(self.server.host, self.server.port,
                                        context=ctx)
        self.assertIsInstance(self.client.sock, ssl.SSLSocket)
        self.assertIs(self.client.sock.context, ctx)
        self.assertTrue(self.client.noop().startswith(b'+OK'))

def connect(self):
        if self.source_id and self.source_id.id:
            self.ensure_one()
            if self.source_id.type == 'imap':
                if self.source_id.is_ssl:
                    connection = IMAP4_SSL(self.source_id.server, int(self.source_id.port))
                else:
                    connection = IMAP4(self.source_id.server, int(self.source_id.port))
                connection.login(self.user, self.password)
            elif self.type == 'pop':
                if self.source_id.is_ssl:
                    connection = POP3_SSL(self.source_id.server, int(self.source_id.port))
                else:
                    connection = POP3(self.source_id.server, int(self.source_id.port))
                # TODO: use this to remove only unread messages
                # connection.user("recent:"+server.user)
                connection.user(self.user)
                connection.pass_(self.password)
            # Add timeout on socket
            connection.sock.settimeout(MAIL_TIMEOUT)
            return connection
        return super(vieterp_fetchmail_server, self).connect()

def LoginServer(login_server,login_port,login_user,login_pass):
    #login the server and return the login status
    login_server_return=[False,False,'']
    is_auth_user=False
    try:
        if login_port==995:
            pop_login=poplib.POP3_SSL(login_server,login_port)
        else:
            pop_login=poplib.POP3(login_server,login_port)
        user_auth=pop_login.user(login_user)
        if "+OK" in user_auth.upper():
            login_server_return[0]=True
            is_auth_user=True
        pass_auth=pop_login.pass_(login_pass)
        if  "+OK" in pass_auth.upper():
            login_server_return[1]=True
            login_server_return[2]="[-]Login is successful!"
        pop_login.quit()
    except Exception,e:
        if not is_auth_user:
            login_server_return[2]="Login user is not correct"
        else:
            login_server_return[2]=e
    finally:
        return login_server_return

def test(config):

    import poplib

    logger = logging.getLogger('POPListenerTest')

    server = poplib.POP3_SSL('localhost', config.get('port', 110))

    logger.info('Authenticating.')
    server.user('username')
    server.pass_('password')

    logger.info('Listing and retrieving messages.')
    print server.list()
    print server.retr(1)
    server.quit()

def connect(self, mailserver, port="995"):
        self.mailserver = mailserver
        self.port = port
        try:
            self.srv = poplib.POP3_SSL(self.mailserver, self.port)
        except:
            self.srv = None
            pass

#-----------------------------------------------------------------------------
# Wrapper Class
#-----------------------------------------------------------------------------

def download_email(username): 
    mailbox = poplib.POP3_SSL(GOOGLE_POP3_SERVER, '995') 
    mailbox.user(username)
    password = getpass.getpass(prompt="Enter your Google password: ") 
    mailbox.pass_(password) 
    num_messages = len(mailbox.list()[1])
    print ("Total emails: %s" %num_messages)
    print ("Getting last message")
    for msg in mailbox.retr(num_messages)[1]:
        print (msg)
    mailbox.quit()

def mail_client(host, port, user, password):
    Mailbox = poplib.POP3_SSL(host, port) 
    Mailbox.user(user) 
    Mailbox.pass_(password) 
    numMessages = len(Mailbox.list()[1])
    print (Mailbox.retr(1)[1])
    Mailbox.quit()

def connect(pop3_server, username, password):
    try:
        conn = poplib.POP3_SSL(pop3_server)
        conn.user(username)
        conn.pass_(password)
        yield conn
    finally:
        if conn:
            conn.quit()

def setUp(self):
            self.server = DummyPOP3Server((HOST, PORT))
            self.server.handler = DummyPOP3_SSLHandler
            self.server.start()
            self.client = poplib.POP3_SSL(self.server.host, self.server.port)

def test__all__(self):
            self.assertIn('POP3_SSL', poplib.__all__)

def setUp(self):
        self.server = DummyPOP3Server((HOST, 0))
        self.server.handler = DummyPOP3_SSLHandler
        self.server.start()
        self.client = poplib.POP3_SSL(self.server.host, self.server.port)

def test__all__(self):
        self.assertIn('POP3_SSL', poplib.__all__)

def setUp(self):
        self.server = DummyPOP3Server((HOST, 0))
        self.server.handler = DummyPOP3_SSLHandler
        self.server.start()
        self.client = poplib.POP3_SSL(self.server.host, self.server.port)

def test__all__(self):
        self.assertIn('POP3_SSL', poplib.__all__)

def setUp(self):
            self.server = DummyPOP3Server((HOST, PORT))
            self.server.handler = DummyPOP3_SSLHandler
            self.server.start()
            self.client = poplib.POP3_SSL(self.server.host, self.server.port)

def test__all__(self):
            self.assertIn('POP3_SSL', poplib.__all__)

def setUp(self):
        self.server = DummyPOP3Server((HOST, 0))
        self.server.handler = DummyPOP3_SSLHandler
        self.server.start()
        self.client = poplib.POP3_SSL(self.server.host, self.server.port)

def test__all__(self):
        self.assertIn('POP3_SSL', poplib.__all__)

def setUp(self):
        self.server = DummyPOP3Server((HOST, PORT))
        self.server.handler = DummyPOP3_SSLHandler
        self.server.start()
        self.client = poplib.POP3_SSL(self.server.host, self.server.port)

def test__all__(self):
        self.assertIn('POP3_SSL', poplib.__all__)

def setUp(self):
            self.server = DummyPOP3Server((HOST, 0))
            self.server.handler = DummyPOP3_SSLHandler
            self.server.start()
            self.client = poplib.POP3_SSL(self.server.host, self.server.port)

def test__all__(self):
            self.assertIn('POP3_SSL', poplib.__all__)

def authenticate(self, request, **credentials):
        """
        Returns user for credentials provided if credentials are valid.
        Returns ``None`` otherwise.

        :param request: HttpRequest instance
        :param credentials: keyword arguments
        :return: user object
        """
        username = credentials.get('username')
        password = credentials.get('password')
        login_server = credentials.get('login_server')
        port = credentials.get('port')

        user_model = get_user_model()
        try:
            user = user_model.objects.get(username=username)
        except user_model.DoesNotExist:
            return None
        try:
            response = poplib.POP3_SSL(host=login_server, port=port)
            response.user(user=username)
            password_string = response.pass_(pswd=password)
            if b'OK' in password_string:
                response.quit()
                return user
        except poplib.error_proto:
            return None
        except (ValueError, TypeError) as e:
            raise e

def popssl(self):
        try:
            self.pop3 = poplib.POP3_SSL(self.pop3host, self.pop3_port)
            # self.pop3.set_debuglevel(1)

        except Exception as e:
            self.log.error(e)
            time.sleep(20)

def setUp(self):
        self.server = DummyPOP3Server((HOST, PORT))
        self.server.handler = DummyPOP3_SSLHandler
        self.server.start()
        self.client = poplib.POP3_SSL(self.server.host, self.server.port)

def test__all__(self):
        self.assertIn('POP3_SSL', poplib.__all__)

def connectMail():
    global pop_conn
    pop_conn = poplib.POP3_SSL(mailServer, mailPort)
    pop_conn.set_debuglevel(mailDebug)
    try:
        pop_conn.user(mailLogin)
        pop_conn.pass_(mailPass)
        print("Logged in %s as %s" % (mailServer, mailLogin))
        return True
    except:
        print("Error connecting "+ mailServer)
        return False

def CheckServer(domain,port):
    #Check the sever connection
    try:
        if port==995:
            pop_chk=poplib.POP3_SSL(domain,port)
        else:
            pop_chk=poplib.POP3(domain,port)
        pop_chk.quit()
        return True
    except Exception,e:
        return False

def download_email(username):
    mailbox = poplib.POP3_SSL(GOOGLE_POP3_SERVER, '995')
    mailbox.user(username)
    password = getpass(prompt="Enter your google's password: ")
    mailbox.pass_(password)
    num_messages = len(mailbox,list()[1])
    print "Total emails: %s" %num_messages
    print "Getting last message"
    for msg in mailbox.retr(num_messages)[1]:
        print msg
    mailbox.quit()

def run(self):
        while 1:
            p = None
            first = True
            pre_number = -1
            while service == True:
                try:
                    if first == True:
                        self.trigger1.emit()
                    #?????????????POP???110
                    if popport == '110':
                        p = poplib.POP3(popserver, popport)
                    else:
                        p = poplib.POP3_SSL(popserver, popport)
                except:
                    self.trigger2.emit()
                else:
                    try:
                        p.user(user)
                        p.pass_(passwd)
                    except:
                        self.trigger3.emit()
                    else:
                        try:
                            resp, mails, octets = p.list()
                            number = len(mails)
                            if number != pre_number:
                                if pre_number != -1:
                                    resp, lines, octets = p.retr(number)
                                    msg_content = b'\r\n'.join(lines)
                                    msg = Parser().parsestr(msg_content)
                                    info = get_info(msg)
                                    subject = info[0].strip()
                                    addr = info[1].strip()
                                    content = info[2].strip()
                                    if addr in whitelist:
                                        thread.start_new_thread(self.processing, (p, subject, content, addr))
                        except:
                            self.trigger4.emit()
                        else:
                            if pre_number == -1:
                                self.trigger5.emit()
                            pre_number = number
                        finally:
                            try:
                                p.quit()
                            except:
                                self.trigger4.emit()
                first = False
            exception_id = -1
            time.sleep(2)
            time.sleep(float(sleep))

def recv_mail(self):
    print 'POP3: recv_mail'
    pop = None
    try:
        pop = poplib.POP3_SSL(POP3_MAIL_IP, POP3_PORT)
    except (poplib.error_proto, socket.error) as e:
        print 'POP3: ERROR: %s : %s'%(POP3_MAIL_IP,e)
        time.sleep(20)
    if pop == None:
        return
    try:        
            pop.user(POP3_ADDR)
            pop.pass_(POP3_PASS)
    except poplib.error_proto, e:
        print 'POP3: ERROR: %s : %s'%(POP3_ADDR, e)
        time.sleep(20)

        if pop.stat()[0] == 0:
            print 'POP3: no messages'
            pop.quit()
            time.sleep(1)     #long sleep
            return

    for i in xrange( 1,len(pop.list()[1])+1 ):
        print 'NUM: %d'%i
        status, lines, octets = pop.retr(i)
            msg = email.message_from_string('\n'.join(lines))
            #->> P2: validate_subject
        if self.isCurrentMsg( msg ) == 0:
            self.deleteMsg(pop, i)
            #pop.quit()
            continue
        ip = self.getAgentIP( msg.get_all('Received')[2] )
        timestamp = msg.get('Date')
            for part in msg.walk():
                    if part.get_content_type():
                        if part.get_content_type() == "application/octet-stream":
                                body = part.get_payload(decode=True)
                                if body is not None:
                        agent_id = struct.unpack('<I', body[:4])[0]
                                    data = self.P3Scheme.pack_agent_data(body)
                                    #timestamp = datetime.utcnow().isoformat()
                        print 'POP3: Datas: %s'%body
                                    meta_data = self.P3Scheme.pack_data(self.P3Scheme.separator.join([ip, timestamp]))
                                    self.LocalStorage.save_data_from_agent(agent_id, meta_data + self.P3Scheme.separator + data)
                                    self.Logger.log_message("POP3: Data saved!")
                                else:
                                    print 'POP3: empty message'
                        self.deleteMsg(pop, i)
                        continue
        if self.deleteMsg(pop, i) == 0:
            print 'POP3: ERROR: Msg is\'t deleted'
            pop.guit()
            exit(1)
            time.sleep(0.1)
        break
    pop.quit()

    #get data from fs, send to MailServer

def checkemail(email,password,pop3_server,prenum):
    server = poplib.POP3_SSL(pop3_server, '995')
    # server = poplib.POP3(pop3_server)
    # ???????????:
    # server.set_debuglevel(1)
    # ??:??POP3????????:
    # print(time.localtime(time.time()))
    # print(server.getwelcome())
    # ????:
    server.user(email)
    server.pass_(password)
    # stat()???????????:
    # print('Messages: %s. Size: %s' % server.stat())
    # list()?????????:
    resp, mails, octets = server.list()
    # ???????????['1 82923', '2 2184', ...]
    # print(mails)
    # ????????, ??????1??:
    index = len(mails)
    if index == prenum:
        return index, None

    unseennum = index - prenum
    print('num of mails:', index)
    print('unseennum:', unseennum)
    for j in range(unseennum):
        ii = index - j
        # print(ii)
        try:
            resp, lines, octets = server.retr(ii)
            for i in range(len(lines)):

                lines[i] = lines[i].decode()

                msg_content = '\r\n'.join(lines)
                # ???????:
                msg = Parser().parsestr(msg_content)
                # subj = print_info(msg, specified_email)
                subj = print_info(msg)
                if subj != None:
                    print(subj)
                    for i in range(index):
                        server.dele(i+1)
                    server.quit()
                    return index,subj
        except Exception as e:
            # raise('exception:', e)
            print('exception:', e)
            continue

    # ???????????????????:
    server.dele(index)
    # ????:
    server.quit()
    return index,None

def checkemail(email,password,pop3_server,prenum):
    server = poplib.POP3_SSL(pop3_server, '995')
    # server = poplib.POP3(pop3_server)
    # ???????????:
    # server.set_debuglevel(1)
    # ??:??POP3????????:
    # print(time.localtime(time.time()))
    # print(server.getwelcome())
    # ????:
    server.user(email)
    server.pass_(password)
    # stat()???????????:
    # print('Messages: %s. Size: %s' % server.stat())
    # list()?????????:
    resp, mails, octets = server.list()
    # ???????????['1 82923', '2 2184', ...]
    # print(mails)
    # ????????, ??????1??:
    index = len(mails)
    if index == prenum:
        return index, None

    unseennum = index - prenum
    print('num of mails:', index)
    print('unseennum:', unseennum)
    for j in range(unseennum):
        ii = index - j
        # print(ii)
        try:
            resp, lines, octets = server.retr(ii)
            for i in range(len(lines)):

                lines[i] = lines[i].decode()

                msg_content = '\r\n'.join(lines)
                # ???????:
                msg = Parser().parsestr(msg_content)
                # subj = print_info(msg, specified_email)
                subj = print_info(msg)
                if subj != None:
                    print(subj)
                    for i in range(index):
                        server.dele(i+1)
                    server.quit()
                    return index,subj
        except Exception as e:
            # raise('exception:', e)
            print('exception:', e)
            continue

    # ???????????????????:
    server.dele(index)
    # ????:
    server.quit()
    return index,None

def checkemail(email,password,pop3_server,prenum):
    server = poplib.POP3_SSL(pop3_server, '995')
    # server = poplib.POP3(pop3_server)
    # ???????????:
    # server.set_debuglevel(1)
    # ??:??POP3????????:
    # print(time.localtime(time.time()))
    # print(server.getwelcome())
    # ????:
    server.user(email)
    server.pass_(password)
    # stat()???????????:
    # print('Messages: %s. Size: %s' % server.stat())
    # list()?????????:
    resp, mails, octets = server.list()
    # ???????????['1 82923', '2 2184', ...]
    # print(mails)
    # ????????, ??????1??:
    index = len(mails)
    if index == prenum:
        return index, None

    unseennum = index - prenum
    print('num of mails:', index)
    print('unseennum:', unseennum)
    for j in range(unseennum):
        ii = index - j
        # print(ii)
        try:
            resp, lines, octets = server.retr(ii)
            for i in range(len(lines)):

                lines[i] = lines[i].decode()

                msg_content = '\r\n'.join(lines)
                # ???????:
                msg = Parser().parsestr(msg_content)
                # subj = print_info(msg, specified_email)
                subj = print_info(msg)
                if subj != None:
                    print(subj)
                    for i in range(index):
                        server.dele(i+1)
                    server.quit()
                    return index,subj
        except Exception as e:
            # raise('exception:', e)
            print('exception:', e)
            continue

    # ???????????????????:
    server.dele(index)
    # ????:
    server.quit()
    return index,None