"""Message Signing Serializer."""fromkombu.serializationimportdisable_insecure_serializersas_disable_insecure_serializersfromkombu.serializationimportregistryfromcelery.exceptionsimportImproperlyConfiguredfrom.serializationimportregister_auth# : need cryptography firstCRYPTOGRAPHY_NOT_INSTALLED="""\You need to install the cryptography library to use the auth serializer.Please install by: $ pip install cryptography"""SECURITY_SETTING_MISSING="""\Sorry, but you have to configure the * security_key * security_certificate, and the * security_cert_storeconfiguration settings to use the auth serializer.Please see the configuration reference for more information."""SETTING_MISSING="""\You have to configure a special task serializerfor signing and verifying tasks: * task_serializer = 'auth'You have to accept only tasks which are serialized with 'auth'.There is no point in signing messages if they are not verified. * accept_content = ['auth']"""__all__=('setup_security',)try:importcryptography# noqaexceptImportError:raiseImproperlyConfigured(CRYPTOGRAPHY_NOT_INSTALLED)
[文档]defsetup_security(allowed_serializers=None,key=None,key_password=None,cert=None,store=None,digest=None,serializer='json',app=None):"""See :meth:`@Celery.setup_security`."""ifappisNone:fromceleryimportcurrent_appapp=current_app._get_current_object()_disable_insecure_serializers(allowed_serializers)# check conf for sane security settingsconf=app.confifconf.task_serializer!='auth'orconf.accept_content!=['auth']:raiseImproperlyConfigured(SETTING_MISSING)key=keyorconf.security_keykey_password=key_passwordorconf.security_key_passwordcert=certorconf.security_certificatestore=storeorconf.security_cert_storedigest=digestorconf.security_digestifnot(keyandcertandstore):raiseImproperlyConfigured(SECURITY_SETTING_MISSING)withopen(key)askf:withopen(cert)ascf:register_auth(kf.read(),key_password,cf.read(),store,digest,serializer)registry._set_default_serializer('auth')