我们从Python开源项目中，提取了以下44个代码示例，用于说明如何使用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 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_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 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 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 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
