我有一个简单的Twisted-Klein
服务器,全局启用了HTTP Basic Auth:
from klein import Klein
import attr
from zope.interface import implementer
from twisted.cred.portal import IRealm
from twisted.internet.defer import succeed
from twisted.cred.portal import Portal
from twisted.cred.checkers import FilePasswordDB
from twisted.web.resource import IResource
from twisted.web.guard import HTTPAuthSessionWrapper, BasicCredentialFactory
from werkzeug.datastructures import MultiDict
from bson import json_util
import json
app = Klein()
# health check
@app.route('/health', methods=['GET'])
def health_check(request):
return ''
# dataset query API
@app.route('/query/<path:expression>', methods=['GET'])
def query(request, expression):
response = evaluate_expression(expression)
return response
@implementer(IRealm)
@attr.s
class HTTPAuthRealm(object):
resource = attr.ib()
def requestAvatar(self, avatarId, mind, *interfaces):
return succeed((IResource, self.resource, lambda: None))
def resource():
realm = HTTPAuthRealm(resource=app.resource())
portal = Portal(realm, [FilePasswordDB('./configs/server-auth.db')])
credential_factory = BasicCredentialFactory('Authentication required')
return HTTPAuthSessionWrapper(portal, [credential_factory])
我想仅为特定 API 端点禁用身份验证,例如,在本例中为/health
API 端点禁用身份验证。我已经阅读了文档,但只是无法理解它。
一种方法是仅包装要为其进行身份验证的层次结构部分:
from twisted.web.resource import Resource
class Health(Resource):
# ...
def resource():
realm = HTTPAuthRealm(resource=app.resource())
portal = Portal(realm, [FilePasswordDB('./configs/server-auth.db')])
credential_factory = BasicCredentialFactory('Authentication required')
guarded = HTTPAuthSessionWrapper(portal, [credential_factory])
root = Resource()
root.putChild(b"health", Health())
root.putChild(b"this-stuff-requires-auth", guarded)
return root
用于调度请求的正常资源遍历逻辑将从root
开始。 如果请求是针对/health
(或任何子项),则它将转到root
的health
子项 - 这是此示例中创建的Health
实例。 请注意HTTPAuthSessionWrapper
如何不参与其中。 如果请求是针对/this-stuff-requires-auth
(或任何子级),则遍历确实会通过身份验证包装器,因此需要身份验证。
另一种方法是根据凭据改变您的头像。 在此方案中,您实际上仍对每个人进行身份验证,但您授权匿名用户访问某些层次结构。
from twisted.cred.checkers import ANONYMOUS
@implementer(IRealm)
@attr.s
class HTTPAuthRealm(object):
def requestAvatar(self, avatarId, mind, *interfaces):
avatar = Resource()
avatar.putChild(b"health", Health())
if avatarId is not ANONYMOUS:
avatar.putChild(b"this-stuff-requires-auth", SecretResource())
return succeed((IResource, avatar, lambda: None))
您还需要使用匿名凭据的凭据检查器配置门户:
from twisted.cred.checkers import AllowAnonymousAccess
portal = Portal(
realm, [
FilePasswordDB('./configs/server-auth.db'),
AllowAnonymousAccess(),
],
)
在此方法中,HTTPAuthSessionWrapper
再次成为您的根资源。
匿名请求与ANONYMOUS
头像标识符相关联,HTTPAuthRealm
返回一个IResource
,该只知道匿名用户应该可用的资源。
具有有效用户凭据的请求与不同的头像标识符(通常是其用户名)相关联,HTTPAuthRealm
返回附加了更多子项的IResource
,从而授予更多访问权限。