403WebShell

403Webshell
Server IP : 13.127.148.211 / Your IP : 216.73.216.149
Web Server : Apache/2.4.41 (Ubuntu)
System : Linux ip-172-31-43-195 5.15.0-1084-aws #91~20.04.1-Ubuntu SMP Fri May 2 06:59:36 UTC 2025 x86_64
User : www-data ( 33)
PHP Version : 7.4.3-4ubuntu2.29
Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare,
MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON
Directory : /lib/python3/dist-packages/twisted/mail/test/
Upload File :
[ Back ]
Current File : /lib/python3/dist-packages/twisted/mail/test/test_pop3client.py
# -*- test-case-name: twisted.mail.test.test_pop3client -*-
# Copyright (c) 2001-2004 Divmod Inc.
# See LICENSE for details.

import sys
import inspect

from zope.interface import directlyProvides

from twisted.internet import reactor, defer, error, protocol, interfaces
from twisted.mail.pop3 import AdvancedPOP3Client as POP3Client
from twisted.mail.pop3 import InsecureAuthenticationDisallowed
from twisted.mail.pop3 import ServerErrorResponse
from twisted.mail.test import pop3testserver
from twisted.protocols import basic, loopback
from twisted.python import log
from twisted.python.compat import intToBytes
from twisted.test.proto_helpers import StringTransport
from twisted.trial import unittest


try:
    from twisted.test.ssl_helpers import ClientTLSContext, ServerTLSContext
except ImportError:
    ClientTLSContext = ServerTLSContext = None



class StringTransportWithConnectionLosing(StringTransport):
    def loseConnection(self):
        self.protocol.connectionLost(error.ConnectionDone())


capCache = {b"TOP": None, b"LOGIN-DELAY": b"180", b"UIDL": None, \
            b"STLS": None, b"USER": None, b"SASL": b"LOGIN"}
def setUp(greet=True):
    p = POP3Client()

    # Skip the CAPA login will issue if it doesn't already have a
    # capability cache
    p._capCache = capCache

    t = StringTransportWithConnectionLosing()
    t.protocol = p
    p.makeConnection(t)

    if greet:
        p.dataReceived(b'+OK Hello!\r\n')

    return p, t



def strip(f):
    return lambda result, f=f: f()



class POP3ClientLoginTests(unittest.TestCase):
    def testNegativeGreeting(self):
        p, t = setUp(greet=False)
        p.allowInsecureLogin = True
        d = p.login(b"username", b"password")
        p.dataReceived(b'-ERR Offline for maintenance\r\n')
        return self.assertFailure(
            d, ServerErrorResponse).addCallback(
            lambda exc: self.assertEqual(exc.args[0],
                b"Offline for maintenance"))


    def testOkUser(self):
        p, t = setUp()
        d = p.user(b"username")
        self.assertEqual(t.value(), b"USER username\r\n")
        p.dataReceived(b"+OK send password\r\n")
        return d.addCallback(self.assertEqual, b"send password")


    def testBadUser(self):
        p, t = setUp()
        d = p.user(b"username")
        self.assertEqual(t.value(), b"USER username\r\n")
        p.dataReceived(b"-ERR account suspended\r\n")
        return self.assertFailure(
            d, ServerErrorResponse).addCallback(
            lambda exc: self.assertEqual(exc.args[0], b"account suspended"))


    def testOkPass(self):
        p, t = setUp()
        d = p.password(b"password")
        self.assertEqual(t.value(), b"PASS password\r\n")
        p.dataReceived(b"+OK you're in!\r\n")
        return d.addCallback(self.assertEqual, b"you're in!")


    def testBadPass(self):
        p, t = setUp()
        d = p.password(b"password")
        self.assertEqual(t.value(), b"PASS password\r\n")
        p.dataReceived(b"-ERR go away\r\n")
        return self.assertFailure(
            d, ServerErrorResponse).addCallback(
            lambda exc: self.assertEqual(exc.args[0], b"go away"))


    def testOkLogin(self):
        p, t = setUp()
        p.allowInsecureLogin = True
        d = p.login(b"username", b"password")
        self.assertEqual(t.value(), b"USER username\r\n")
        p.dataReceived(b"+OK go ahead\r\n")
        self.assertEqual(t.value(), b"USER username\r\nPASS password\r\n")
        p.dataReceived(b"+OK password accepted\r\n")
        return d.addCallback(self.assertEqual, b"password accepted")


    def testBadPasswordLogin(self):
        p, t = setUp()
        p.allowInsecureLogin = True
        d = p.login(b"username", b"password")
        self.assertEqual(t.value(), b"USER username\r\n")
        p.dataReceived(b"+OK waiting on you\r\n")
        self.assertEqual(t.value(), b"USER username\r\nPASS password\r\n")
        p.dataReceived(b"-ERR bogus login\r\n")
        return self.assertFailure(
            d, ServerErrorResponse).addCallback(
            lambda exc: self.assertEqual(exc.args[0], b"bogus login"))


    def testBadUsernameLogin(self):
        p, t = setUp()
        p.allowInsecureLogin = True
        d = p.login(b"username", b"password")
        self.assertEqual(t.value(), b"USER username\r\n")
        p.dataReceived(b"-ERR bogus login\r\n")
        return self.assertFailure(
            d, ServerErrorResponse).addCallback(
            lambda exc: self.assertEqual(exc.args[0], b"bogus login"))


    def testServerGreeting(self):
        p, t = setUp(greet=False)
        p.dataReceived(b"+OK lalala this has no challenge\r\n")
        self.assertEqual(p.serverChallenge, None)


    def testServerGreetingWithChallenge(self):
        p, t = setUp(greet=False)
        p.dataReceived(b"+OK <here is the challenge>\r\n")
        self.assertEqual(p.serverChallenge, b"<here is the challenge>")


    def testAPOP(self):
        p, t = setUp(greet=False)
        p.dataReceived(b"+OK <challenge string goes here>\r\n")
        d = p.login(b"username", b"password")
        self.assertEqual(t.value(),
            b"APOP username f34f1e464d0d7927607753129cabe39a\r\n")
        p.dataReceived(b"+OK Welcome!\r\n")
        return d.addCallback(self.assertEqual, b"Welcome!")


    def testInsecureLoginRaisesException(self):
        p, t = setUp(greet=False)
        p.dataReceived(b"+OK Howdy\r\n")
        d = p.login(b"username", b"password")
        self.assertFalse(t.value())
        return self.assertFailure(
            d, InsecureAuthenticationDisallowed)


    def testSSLTransportConsideredSecure(self):
        """
        If a server doesn't offer APOP but the transport is secured using
        SSL or TLS, a plaintext login should be allowed, not rejected with
        an InsecureAuthenticationDisallowed exception.
        """
        p, t = setUp(greet=False)
        directlyProvides(t, interfaces.ISSLTransport)
        p.dataReceived(b"+OK Howdy\r\n")
        d = p.login(b"username", b"password")
        self.assertEqual(t.value(), b"USER username\r\n")
        t.clear()
        p.dataReceived(b"+OK\r\n")
        self.assertEqual(t.value(), b"PASS password\r\n")
        p.dataReceived(b"+OK\r\n")
        return d



class ListConsumer:
    def __init__(self):
        self.data = {}


    def consume(self, result):
        (item, value) = result
        self.data.setdefault(item, []).append(value)



class MessageConsumer:
    def __init__(self):
        self.data = []


    def consume(self, line):
        self.data.append(line)



class POP3ClientListTests(unittest.TestCase):
    def testListSize(self):
        p, t = setUp()
        d = p.listSize()
        self.assertEqual(t.value(), b"LIST\r\n")
        p.dataReceived(b"+OK Here it comes\r\n")
        p.dataReceived(b"1 3\r\n2 2\r\n3 1\r\n.\r\n")
        return d.addCallback(self.assertEqual, [3, 2, 1])


    def testListSizeWithConsumer(self):
        p, t = setUp()
        c = ListConsumer()
        f = c.consume
        d = p.listSize(f)
        self.assertEqual(t.value(), b"LIST\r\n")
        p.dataReceived(b"+OK Here it comes\r\n")
        p.dataReceived(b"1 3\r\n2 2\r\n3 1\r\n")
        self.assertEqual(c.data, {0: [3], 1: [2], 2: [1]})
        p.dataReceived(b"5 3\r\n6 2\r\n7 1\r\n")
        self.assertEqual(c.data, {0: [3], 1: [2], 2: [1], 4: [3], 5: [2],
                                  6: [1]})
        p.dataReceived(b".\r\n")
        return d.addCallback(self.assertIdentical, f)


    def testFailedListSize(self):
        p, t = setUp()
        d = p.listSize()
        self.assertEqual(t.value(), b"LIST\r\n")
        p.dataReceived(b"-ERR Fatal doom server exploded\r\n")
        return self.assertFailure(
            d, ServerErrorResponse).addCallback(
            lambda exc: self.assertEqual(exc.args[0],
                b"Fatal doom server exploded"))


    def testListUID(self):
        p, t = setUp()
        d = p.listUID()
        self.assertEqual(t.value(), b"UIDL\r\n")
        p.dataReceived(b"+OK Here it comes\r\n")
        p.dataReceived(b"1 abc\r\n2 def\r\n3 ghi\r\n.\r\n")
        return d.addCallback(self.assertEqual, [b"abc", b"def", b"ghi"])


    def testListUIDWithConsumer(self):
        p, t = setUp()
        c = ListConsumer()
        f = c.consume
        d = p.listUID(f)
        self.assertEqual(t.value(), b"UIDL\r\n")
        p.dataReceived(b"+OK Here it comes\r\n")
        p.dataReceived(b"1 xyz\r\n2 abc\r\n5 mno\r\n")
        self.assertEqual(c.data, {0: [b"xyz"], 1: [b"abc"], 4: [b"mno"]})
        p.dataReceived(b".\r\n")
        return d.addCallback(self.assertIdentical, f)


    def testFailedListUID(self):
        p, t = setUp()
        d = p.listUID()
        self.assertEqual(t.value(), b"UIDL\r\n")
        p.dataReceived(b"-ERR Fatal doom server exploded\r\n")
        return self.assertFailure(
            d, ServerErrorResponse).addCallback(
            lambda exc: self.assertEqual(exc.args[0],
                b"Fatal doom server exploded"))



class POP3ClientMessageTests(unittest.TestCase):
    def testRetrieve(self):
        p, t = setUp()
        d = p.retrieve(7)
        self.assertEqual(t.value(), b"RETR 8\r\n")
        p.dataReceived(b"+OK Message incoming\r\n")
        p.dataReceived(b"La la la here is message text\r\n")
        p.dataReceived(b"..Further message text tra la la\r\n")
        p.dataReceived(b".\r\n")
        return d.addCallback(
            self.assertEqual,
            [b"La la la here is message text",
             b".Further message text tra la la"])


    def testRetrieveWithConsumer(self):
        p, t = setUp()
        c = MessageConsumer()
        f = c.consume
        d = p.retrieve(7, f)
        self.assertEqual(t.value(), b"RETR 8\r\n")
        p.dataReceived(b"+OK Message incoming\r\n")
        p.dataReceived(b"La la la here is message text\r\n")
        p.dataReceived(b"..Further message text\r\n.\r\n")
        return d.addCallback(self._cbTestRetrieveWithConsumer, f, c)


    def _cbTestRetrieveWithConsumer(self, result, f, c):
        self.assertIdentical(result, f)
        self.assertEqual(c.data, [b"La la la here is message text",
                                   b".Further message text"])


    def testPartialRetrieve(self):
        p, t = setUp()
        d = p.retrieve(7, lines=2)
        self.assertEqual(t.value(), b"TOP 8 2\r\n")
        p.dataReceived(b"+OK 2 lines on the way\r\n")
        p.dataReceived(b"Line the first!  Woop\r\n")
        p.dataReceived(b"Line the last!  Bye\r\n")
        p.dataReceived(b".\r\n")
        return d.addCallback(
            self.assertEqual,
            [b"Line the first!  Woop",
             b"Line the last!  Bye"])


    def testPartialRetrieveWithConsumer(self):
        p, t = setUp()
        c = MessageConsumer()
        f = c.consume
        d = p.retrieve(7, f, lines=2)
        self.assertEqual(t.value(), b"TOP 8 2\r\n")
        p.dataReceived(b"+OK 2 lines on the way\r\n")
        p.dataReceived(b"Line the first!  Woop\r\n")
        p.dataReceived(b"Line the last!  Bye\r\n")
        p.dataReceived(b".\r\n")
        return d.addCallback(self._cbTestPartialRetrieveWithConsumer, f, c)


    def _cbTestPartialRetrieveWithConsumer(self, result, f, c):
        self.assertIdentical(result, f)
        self.assertEqual(c.data, [b"Line the first!  Woop",
                                  b"Line the last!  Bye"])


    def testFailedRetrieve(self):
        p, t = setUp()
        d = p.retrieve(0)
        self.assertEqual(t.value(), b"RETR 1\r\n")
        p.dataReceived(b"-ERR Fatal doom server exploded\r\n")
        return self.assertFailure(
            d, ServerErrorResponse).addCallback(
            lambda exc: self.assertEqual(exc.args[0],
                b"Fatal doom server exploded"))


    def test_concurrentRetrieves(self):
        """
        Issue three retrieve calls immediately without waiting for any to
        succeed and make sure they all do succeed eventually.
        """
        p, t = setUp()
        messages = [
            p.retrieve(i).addCallback(
                self.assertEqual,
                [b"First line of " + intToBytes(i + 1) + b".",
                 b"Second line of " + intToBytes(i + 1) + b"."])
            for i
            in range(3)]

        for i in range(1, 4):
            self.assertEqual(t.value(), b"RETR " + intToBytes(i) + b"\r\n")
            t.clear()
            p.dataReceived(b"+OK 2 lines on the way\r\n")
            p.dataReceived(b"First line of " + intToBytes(i) + b".\r\n")
            p.dataReceived(b"Second line of " + intToBytes(i) + b".\r\n")
            self.assertEqual(t.value(), b"")
            p.dataReceived(b".\r\n")

        return defer.DeferredList(messages, fireOnOneErrback=True)



class POP3ClientMiscTests(unittest.TestCase):
    def testCapability(self):
        p, t = setUp()
        d = p.capabilities(useCache=0)
        self.assertEqual(t.value(), b"CAPA\r\n")
        p.dataReceived(b"+OK Capabilities on the way\r\n")
        p.dataReceived(b"X\r\nY\r\nZ\r\nA 1 2 3\r\nB 1 2\r\nC 1\r\n.\r\n")
        return d.addCallback(
            self.assertEqual,
            {b"X": None, b"Y": None, b"Z": None,
             b"A": [b"1", b"2", b"3"],
             b"B": [b"1", b"2"],
             b"C": [b"1"]})


    def testCapabilityError(self):
        p, t = setUp()
        d = p.capabilities(useCache=0)
        self.assertEqual(t.value(), b"CAPA\r\n")
        p.dataReceived(b"-ERR This server is lame!\r\n")
        return d.addCallback(self.assertEqual, {})


    def testStat(self):
        p, t = setUp()
        d = p.stat()
        self.assertEqual(t.value(), b"STAT\r\n")
        p.dataReceived(b"+OK 1 1212\r\n")
        return d.addCallback(self.assertEqual, (1, 1212))


    def testStatError(self):
        p, t = setUp()
        d = p.stat()
        self.assertEqual(t.value(), b"STAT\r\n")
        p.dataReceived(b"-ERR This server is lame!\r\n")
        return self.assertFailure(
            d, ServerErrorResponse).addCallback(
            lambda exc: self.assertEqual(exc.args[0], b"This server is lame!"))


    def testNoop(self):
        p, t = setUp()
        d = p.noop()
        self.assertEqual(t.value(), b"NOOP\r\n")
        p.dataReceived(b"+OK No-op to you too!\r\n")
        return d.addCallback(self.assertEqual, b"No-op to you too!")


    def testNoopError(self):
        p, t = setUp()
        d = p.noop()
        self.assertEqual(t.value(), b"NOOP\r\n")
        p.dataReceived(b"-ERR This server is lame!\r\n")
        return self.assertFailure(
            d, ServerErrorResponse).addCallback(
            lambda exc: self.assertEqual(exc.args[0], b"This server is lame!"))


    def testRset(self):
        p, t = setUp()
        d = p.reset()
        self.assertEqual(t.value(), b"RSET\r\n")
        p.dataReceived(b"+OK Reset state\r\n")
        return d.addCallback(self.assertEqual, b"Reset state")


    def testRsetError(self):
        p, t = setUp()
        d = p.reset()
        self.assertEqual(t.value(), b"RSET\r\n")
        p.dataReceived(b"-ERR This server is lame!\r\n")
        return self.assertFailure(
            d, ServerErrorResponse).addCallback(
            lambda exc: self.assertEqual(exc.args[0], b"This server is lame!"))


    def testDelete(self):
        p, t = setUp()
        d = p.delete(3)
        self.assertEqual(t.value(), b"DELE 4\r\n")
        p.dataReceived(b"+OK Hasta la vista\r\n")
        return d.addCallback(self.assertEqual, b"Hasta la vista")


    def testDeleteError(self):
        p, t = setUp()
        d = p.delete(3)
        self.assertEqual(t.value(), b"DELE 4\r\n")
        p.dataReceived(b"-ERR Winner is not you.\r\n")
        return self.assertFailure(
            d, ServerErrorResponse).addCallback(
            lambda exc: self.assertEqual(exc.args[0], b"Winner is not you."))



class SimpleClient(POP3Client):
    def __init__(self, deferred, contextFactory=None):
        self.deferred = deferred
        self.allowInsecureLogin = True


    def serverGreeting(self, challenge):
        self.deferred.callback(None)



class POP3HelperMixin:
    serverCTX = None
    clientCTX = None

    def setUp(self):
        d = defer.Deferred()
        self.server = pop3testserver.POP3TestServer(
            contextFactory=self.serverCTX)
        self.client = SimpleClient(d, contextFactory=self.clientCTX)
        self.client.timeout = 30
        self.connected = d


    def tearDown(self):
        del self.server
        del self.client
        del self.connected


    def _cbStopClient(self, ignore):
        self.client.transport.loseConnection()


    def _ebGeneral(self, failure):
        self.client.transport.loseConnection()
        self.server.transport.loseConnection()
        return failure


    def loopback(self):
        return loopback.loopbackTCP(self.server, self.client, noisy=False)



class TLSServerFactory(protocol.ServerFactory):
    class protocol(basic.LineReceiver):
        context = None
        output = []
        def connectionMade(self):
            self.factory.input = []
            self.output = self.output[:]
            for line in self.output.pop(0):
                self.sendLine(line)


        def lineReceived(self, line):
            self.factory.input.append(line)
            [self.sendLine(l) for l in self.output.pop(0)]
            if line == b'STLS':
                self.transport.startTLS(self.context)



class POP3TLSTests(unittest.TestCase):
    """
    Tests for POP3Client's support for TLS connections.
    """

    def test_startTLS(self):
        """
        POP3Client.startTLS starts a TLS session over its existing TCP
        connection.
        """
        sf = TLSServerFactory()
        sf.protocol.output = [
            [b'+OK'],  # Server greeting
            [b'+OK', b'STLS', b'.'],  # CAPA response
            [b'+OK'],  # STLS response
            [b'+OK', b'.'],  # Second CAPA response
            [b'+OK']  # QUIT response
            ]
        sf.protocol.context = ServerTLSContext()
        port = reactor.listenTCP(0, sf, interface='127.0.0.1')
        self.addCleanup(port.stopListening)
        H = port.getHost().host
        P = port.getHost().port

        connLostDeferred = defer.Deferred()
        cp = SimpleClient(defer.Deferred(), ClientTLSContext())
        def connectionLost(reason):
            SimpleClient.connectionLost(cp, reason)
            connLostDeferred.callback(None)
        cp.connectionLost = connectionLost
        cf = protocol.ClientFactory()
        cf.protocol = lambda: cp

        conn = reactor.connectTCP(H, P, cf)

        def cbConnected(ignored):
            log.msg("Connected to server; starting TLS")
            return cp.startTLS()

        def cbStartedTLS(ignored):
            log.msg("Started TLS; disconnecting")
            return cp.quit()

        def cbDisconnected(ign):
            log.msg("Disconnected; asserting correct input received")
            self.assertEqual(
                sf.input,
                [b'CAPA', b'STLS', b'CAPA', b'QUIT'])

        def cleanup(result):
            log.msg("Asserted correct input; disconnecting "
                    "client and shutting down server")
            conn.disconnect()
            return connLostDeferred

        cp.deferred.addCallback(cbConnected)
        cp.deferred.addCallback(cbStartedTLS)
        cp.deferred.addCallback(cbDisconnected)
        cp.deferred.addBoth(cleanup)

        return cp.deferred



class POP3TimeoutTests(POP3HelperMixin, unittest.TestCase):
    def testTimeout(self):
        def login():
            d = self.client.login('test', 'twisted')
            d.addCallback(loggedIn)
            d.addErrback(timedOut)
            return d

        def loggedIn(result):
            self.fail("Successfully logged in!?  Impossible!")


        def timedOut(failure):
            failure.trap(error.TimeoutError)
            self._cbStopClient(None)

        def quit():
            return self.client.quit()

        self.client.timeout = 0.01

        # Tell the server to not return a response to client.  This
        # will trigger a timeout.
        pop3testserver.TIMEOUT_RESPONSE = True

        methods = [login, quit]
        map(self.connected.addCallback, map(strip, methods))
        self.connected.addCallback(self._cbStopClient)
        self.connected.addErrback(self._ebGeneral)
        return self.loopback()



if ClientTLSContext is None:
    for case in (POP3TLSTests,):
        case.skip = "OpenSSL not present"
elif interfaces.IReactorSSL(reactor, None) is None:
    for case in (POP3TLSTests,):
        case.skip = "Reactor doesn't support SSL"



import twisted.mail.pop3client

class POP3ClientModuleStructureTests(unittest.TestCase):
    """
    Miscellaneous tests more to do with module/package structure than
    anything to do with the POP3 client.
    """
    def test_all(self):
        """
        twisted.mail.pop3client.__all__ should be empty because all classes
        should be imported through twisted.mail.pop3.
        """
        self.assertEqual(twisted.mail.pop3client.__all__, [])


    def test_import(self):
        """
        Every public class in twisted.mail.pop3client should be available as a
        member of twisted.mail.pop3 with the exception of
        twisted.mail.pop3client.POP3Client which should be available as
        twisted.mail.pop3.AdvancedClient.
        """
        publicClasses = [c[0] for c in inspect.getmembers(
                                       sys.modules['twisted.mail.pop3client'],
                                       inspect.isclass)
                         if not c[0][0] == '_']

        for pc in publicClasses:
            if not pc == 'POP3Client':
                self.assertTrue(hasattr(twisted.mail.pop3, pc))
            else:
                self.assertTrue(hasattr(twisted.mail.pop3,
                    'AdvancedPOP3Client'))
Youez - 2016 - github.com/yon3zu
LinuXploit