# -*- Mode: Python; test-case-name: flumotion.test.test_pb -*-
# vi:si:et:sw=4:sts=4:ts=4
#
# Flumotion - a streaming media server
# Copyright (C) 2004,2005,2006,2007 Fluendo, S.L. (www.fluendo.com).
# All rights reserved.

# This file may be distributed and/or modified under the terms of
# the GNU General Public License version 2 as published by
# the Free Software Foundation.
# This file is distributed without any warranty; without even the implied
# warranty of merchantability or fitness for a particular purpose.
# See "LICENSE.GPL" in the source distribution for more information.

# Licensees having purchased or holding a valid Flumotion Advanced
# Streaming Server license may use this file in accordance with the
# Flumotion Advanced Streaming Server Commercial License Agreement.
# See "LICENSE.Flumotion" in the source distribution for more information.

# Headers in this file shall remain intact.

import crypt

from twisted.cred import portal
from twisted.internet import defer, reactor
from twisted.python import log as tlog
from twisted.spread import pb as tpb
from twisted.trial import unittest
from zope.interface import implements

from flumotion.common import testsuite
from flumotion.common import keycards, log, errors
from flumotion.component.bouncers import htpasswdcrypt, saltsha256
from flumotion.twisted import checkers, pb
from flumotion.twisted import portal as fportal

from flumotion.test.common import haveTwisted

htpasswdcryptConf = {
    'name': 'testbouncer',
    'plugs': {},
    'properties': {'data': "user:qi1Lftt0GZC0o"}}

saltsha256Conf = {
    'name': 'testbouncer',
    'plugs': {},
    'properties': {
    'data':
    ("user:"
     "iamsalt:"
     "2f826124ada2b2cdf11f4fd427c9ca48de0ed49b41476266d8df08d2cf86120e")}}

### lots of fake objects to have fun with


class FakePortalPlaintext(portal.Portal):
    # a fake portal with a checker that lets username, password in

    def __init__(self):
        checker = checkers.FlexibleCredentialsChecker()
        checker.addUser("username", "password")
        por
tal.Portal.__init__(self, FakeTRealm(), (checker, ))


class FakePortalWrapperPlaintext:
    # a fake wrapper with a checker that lets username, password in

    def __init__(self):
        self.broker = FakeBroker()
        self.portal = FakePortalplaintext()


class FakePortalWrapperCrypt:
    # a fake wrapper with a checker that lets username, crypt(password, iq) in

    def __init__(self):
        self.checker = checkers.CryptChecker()
        cryptPassword = crypt.crypt('password', 'i
q')
        self.checker.addUser("username", cryptPassword)
        self.portal = portal.Portal(FakeTRealm(), (self.checker, ))

# FIXME: using real portal


class FakeBouncerPortal:
    # a fake wrapper implementing BouncerPortal lookalike

    def __init__(self, bouncer):
        self.bouncer = bouncer

    def login(self, keycard, mind, interfaces):
        return self.bouncer.authenticate(keycard)


class FakeAvatar(tpb.Avatar):
    implements(tpb.IPerspective)
    loggedIn = loggedOut = False

    def __init__(self):
        pass

    def logout(self):
        self.loggedOut = True


class FakeTRealm:

    def __init__(self):
        self.avatar = FakeAvatar()

    def requestAvatar(self, avatarId, mind, *ifaces):
        interface = ifaces[0]
        assert interface == tpb.IPerspective, (
            "interface is %r and not IPerspective" % interface)
        self.avatar.loggedIn = True
        # we can return a deferred, or return directly
        return defer.succeed((tpb.IPerspective,
                              self.avatar, self.avatar.logout))


class FakeFRealm(FakeTRealm):

    def requestAvatar(self, avatarId, keycard, mind, *interfaces):
        return FakeTRealm.requestAvatar(self, avatarId, mind, *interfaces)


class FakeMind(tpb.Referenceable):
    pass


class FakeBroker(tpb.Broker):
    pass

# our test for twisted's challenger
# this is done for comparison with our challenger


class TestTwisted_PortalAuthChallenger(testsuite.TestCase):

    def setUp(self):
        root = tpb.IPBRoot(FakePortalPlaintext()).rootObject(
            broker=FakeBroker())
        # PB server creates a challenge and a challenger to send to the client
        self.challenge, self.challenger = root.remote_login('username')

    def testRightPassword(self):
        # client is asked to respond, so generate the response
        response = tpb.respond(self.challenge, 'password')

        self.challenger.remote_respond(response, None)

    def testWrongPassword(self):
        # client is asked to respond, so generate the response
        response = tpb.respond(self.challenge, 'wrong')
        d = self.challenger.remote_respond(response, None)

        def wrongPasswordErrback(wrongpasserror):
            self.assert_(isinstance(wrongpasserror.type(),
                                    errors.NotAuthenticatedError))

        d.addErrback(wrongPasswordErrback)
        return d

### SHINY NEW FPB


class Test_BouncerWrapper(testsuite.TestCase):

    def setUp(self):
        broker = FakeBroker()

        self.bouncer = htpasswdcrypt.HTPasswdCrypt(htpasswdcryptConf)
        self.bouncerPortal = fportal.BouncerPortal(FakeFRealm(), self.bouncer)
        self.wrapper = pb._BouncerWrapper(self.bouncerPortal, broker)

    def tearDown(self):
        self.bouncer.stop()

    def testUACPPOk(self):
        mind = FakeMind()
        keycard = keycards.KeycardUACPP('user', 'test', '127.0.0.1')
        d = self.wrapper.remote_login(keycard, mind,
            'twisted.spread.pb.IPerspective')

        def uacppOkCallback(result):
            self.assert_(isinstance(result, tpb.AsReferenceable))
            return result

        d.addCallback(uacppOkCallback)
        return d

    def testUACPPWrongPassword(self):
        keycard = keycards.KeycardUACPP('user', 'tes', '127.0.0.1')
        d = self.wrapper.remote_login(keycard, "avatarId",
            'twisted.spread.pb.IPerspective')

        def uacppWrongPasswordErrback(wrongpasserror):
            self.assert_(isinstance(wrongpasserror.type(),
                                    errors.NotAuthenticatedError))

        d.addErrback(uacppWrongPasswordErrback)
        return d

    def testUACPCCOk(self):
        # create
        keycard = keycards.KeycardUACPCC('user', '127.0.0.1')

        # send
        d = self.wrapper.remote_login(keycard, None,
            'twisted.spread.pb.IPerspective')

        def uacpccOkCallback(keycard):
            self.assertEquals(keycard.state, keycards.REQUESTING)
            # respond to challenge
            keycard.setPassword('test')
            d = self.wrapper.remote_login(keycard, None,
                'twisted.spread.pb.IPerspective')

            def uacpccOkCallback2(result):
                self.assert_(isinstance(result, tpb.AsReferenceable))
                return result
            d.addCallback(uacpccOkCallback2)
            return d

        d.addCallback(uacpccOkCallback)
        return d

    def testUACPCCWrongUser(self):
        # create
        keycard = keycards.KeycardUACPCC('wronguser', '127.0.0.1')

        # send
        d = self.wrapper.remote_login(keycard, "avatarId",
            'twisted.spread.pb.IPerspective')

        def uacpccWrongUserCallback(keycard):
            self.assertEquals(keycard.state, keycards.REQUESTING)
            # respond to challenge
            keycard.setPassword('test')
            d = self.wrapper.remote_login(keycard, "avatarId",
                'twisted.spread.pb.IPerspective')

            def uacpccWrongUserErrback(failure):
                self.assert_(isinstance(failure.type(),
                    errors.NotAuthenticatedError))
                return True
            d.addErrback(uacpccWrongUserErrback)
            return d

        d.addCallback(uacpccWrongUserCallback)
        return d

    def testUACPCCWrongPassword(self):
        # create
        keycard = keycards.KeycardUACPCC('user', '127.0.0.1')

        # send
        d = self.wrapper.remote_login(keycard, "avatarId",
            'twisted.spread.pb.IPerspective')

        def uacpccWrongPasswordCallback(keycard):
            sel