Namespace Issue

authn.py

# This file is one copy from https://github.com/arskom/spyne/blob/master/examples/authentication/server_soap.py
import sys
import random
import logging

from spyne.model.complex import ComplexModel
from spyne.model.fault import Fault
from spyne.decorator import srpc
from spyne.error import ArgumentError
from spyne.protocol.soap import Soap11
from spyne.interface.wsdl import Wsdl11
from spyne.model.primitive import Mandatory
from spyne.model.primitive import String
from spyne.service import ServiceBase
from spyne.server.wsgi import WsgiApplication
from spyne.application import Application


class PublicKeyError(Fault):
    __type_name__ = 'KeyError'
    __namespace__ = 'spyne.examples.authentication'

    def __init__(self, value):
        Fault.__init__(self,
                faultcode='Client.KeyError',
                faultstring='Value %r not found' % value
            )


class AuthenticationError(Fault):
    __namespace__ = 'spyne.examples.authentication'

    def __init__(self, user_name):
        # TODO: self.transport.http.resp_code = HTTP_401

        Fault.__init__(self,
                faultcode='Client.AuthenticationError',
                faultstring='Invalid authentication request for %r' % user_name
            )


class AuthorizationError(Fault):
    __namespace__ = 'spyne.examples.authentication'

    def __init__(self):
        # TODO: self.transport.http.resp_code = HTTP_401

        Fault.__init__(self,
                faultcode='Client.AuthorizationError',
                faultstring='You are not authozied to access this resource.'
            )


class SpyneDict(dict):
    def __getitem__(self, key):
        try:
            return dict.__getitem__(self, key)
        except KeyError:
            raise PublicKeyError(key)


class Security(ComplexModel):
    __namespace__ = 'http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd'

    session_id = Mandatory.String
    user_name = Mandatory.String



class Preferences(ComplexModel):
    __namespace__ = 'spyne.examples.authentication'

    language = String(max_len=2)
    time_zone = String


user_db = {
    'neo': "xxx",
}

session_db = set()

preferences_db = SpyneDict({
    'neo': Preferences(language='en', time_zone='Underground/Zion'),
    'smith': Preferences(language='xx', time_zone='Matrix/Core'),
})


class AuthenticationService(ServiceBase):
    __tns__ = 'spyne.examples.authentication'

    @srpc(Mandatory.String, Mandatory.String, _returns=String,
                                                    _throws=AuthenticationError)
    def authenticate(user_name, password):
        password_hash = user_db.get(user_name, None)

        if password_hash is None:
           raise AuthenticationError(user_name)

        if 1:
            session_id = (user_name, '%x' % random.randint(1<<128, (1<<132)-1))
            session_db.add(session_id)
        else:
           raise AuthenticationError(user_name)

        return session_id[1]

class UserService(ServiceBase):
    __tns__ = 'spyne.examples.authentication'
    __in_header__ = Security

    @srpc(Mandatory.String, _throws=PublicKeyError, _returns=Preferences)
    def get_preferences(user_name):
        if user_name == 'smith':
            raise AuthorizationError()

        retval = preferences_db[user_name]

        return retval

def _on_method_call(ctx):
    if ctx.in_object is None:
        raise ArgumentError("Security is null")
    if not (ctx.in_header.user_name, ctx.in_header.session_id) in session_db:
        raise AuthenticationError(ctx.in_object.user_name)

UserService.event_manager.add_listener('method_call', _on_method_call)

if __name__=='__main__':
    try:
        from wsgiref.simple_server import make_server
    except ImportError:
      pass

    logging.basicConfig(level=logging.DEBUG)
    logging.getLogger('spyne.protocol.xml').setLevel(logging.DEBUG)
    logging.getLogger('twisted').setLevel(logging.DEBUG)

    application = Application([AuthenticationService,UserService],
        tns='spyne.examples.authentication',
        interface=Wsdl11(),
        in_protocol=Soap11(validator='lxml'),
        out_protocol=Soap11()
    )
    server = make_server('127.0.0.1', 7789, WsgiApplication(application))
    print "wsdl is at: http://localhost:7789/?wsdl"
    server.serve_forever()