def render(self, request):
"""
Render a given resource. See L{IResource}'s render method.
I delegate to methods of self with the form 'render_METHOD'
where METHOD is the HTTP that was used to make the
request. Examples: render_GET, render_HEAD, render_POST, and
so on. Generally you should implement those methods instead of
overriding this one.
render_METHOD methods are expected to return a string which
will be the rendered page, unless the return value is
twisted.web.server.NOT_DONE_YET, in which case it is this
class's responsibility to write the results to
request.write(data), then call request.finish().
Old code that overrides render() directly is likewise expected
to return a string or NOT_DONE_YET.
"""
if request.code not in coap.requests:
raise error.UnsupportedMethod()
m = getattr(self, 'render_' + coap.requests[request.code], None)
if not m:
raise error.UnallowedMethod()
return m(request)
python类IResource()的实例源码
def test_getChildWithDefaultAuthorized(self):
"""
Resource traversal which encounters an L{HTTPAuthSessionWrapper}
results in an L{IResource} which renders the L{IResource} avatar
retrieved from the portal when the request has a valid I{Authorization}
header.
"""
self.credentialFactories.append(BasicCredentialFactory('example.com'))
request = self.makeRequest([self.childName])
child = self._authorizedBasicLogin(request)
d = request.notifyFinish()
def cbFinished(ignored):
self.assertEqual(request.written, [self.childContent])
d.addCallback(cbFinished)
request.render(child)
return d
def test_renderAuthorized(self):
"""
Resource traversal which terminates at an L{HTTPAuthSessionWrapper}
and includes correct authentication headers results in the
L{IResource} avatar (not one of its children) retrieved from the
portal being rendered.
"""
self.credentialFactories.append(BasicCredentialFactory('example.com'))
# Request it exactly, not any of its children.
request = self.makeRequest([])
child = self._authorizedBasicLogin(request)
d = request.notifyFinish()
def cbFinished(ignored):
self.assertEqual(request.written, [self.avatarContent])
d.addCallback(cbFinished)
request.render(child)
return d
def _authorizedResource(self, request):
"""
Get the L{IResource} which the given request is authorized to receive.
If the proper authorization headers are present, the resource will be
requested from the portal. If not, an anonymous login attempt will be
made.
"""
authheader = request.getHeader(b'authorization')
if not authheader:
return util.DeferredResource(self._login(Anonymous()))
factory, respString = self._selectParseHeader(authheader)
if factory is None:
return UnauthorizedResource(self._credentialFactories)
try:
credentials = factory.decode(respString, request)
except error.LoginFailed:
return UnauthorizedResource(self._credentialFactories)
except:
log.err(None, "Unexpected failure from credentials factory")
return ErrorPage(500, None, None)
else:
return util.DeferredResource(self._login(credentials))
def requestAvatar(self, avatarId, mind, *interfaces):
if resource.IResource not in interfaces:
raise NotImplementedError("no interface")
if avatarId:
return (resource.IResource,
MarkAuthenticatedResource(self.resource, avatarId),
lambda:None)
else:
return resource.IResource, self.nonauthenticated, lambda:None
def __init__(self, portal, callback=None, errback=None):
"""Constructs a UsernamePasswordWrapper around the given portal.
@param portal: A cred portal for your web application. The checkers
associated with this portal must be able to accept username/password
credentials.
@type portal: L{twisted.cred.portal.Portal}
@param callback: Gets called after a successful login attempt.
A resource that redirects to "." will display the avatar resource.
If this parameter isn't provided, defaults to a standard Woven
"Thank You" page.
@type callback: A callable that accepts a Woven
L{model<twisted.web.woven.interfaces.IModel>} and returns a
L{IResource<twisted.web.resource.Resource>}.
@param errback: Gets called after a failed login attempt.
If this parameter is not provided, defaults to a the standard Woven
form error (i.e. The original form on a page of its own, with
errors noted.)
@type errback: A callable that accepts a Woven
L{model<twisted.web.woven.interfaces.IModel>} and returns a
L{IResource<twisted.web.resource.Resource>}.
"""
Resource.__init__(self)
self.portal = portal
self.callback = callback
self.errback = errback
def registerControllerForModel(controller, model):
"""
Registers `controller' as an adapter of `model' for IController, and
optionally registers it for IResource, if it implements it.
@param controller: A class that implements L{interfaces.IController}, usually a
L{Controller} subclass. Optionally it can implement
L{resource.IResource}.
@param model: Any class, but probably a L{twisted.web.woven.model.Model}
subclass.
"""
components.registerAdapter(controller, model, interfaces.IController)
if resource.IResource.implementedBy(controller):
components.registerAdapter(controller, model, resource.IResource)
def getChild(self, path, request):
if path == '': path = 'index'
path = path.replace(".","_")
cm = getattr(self, "wchild_"+path, None)
if cm:
p = cm(request)
if isinstance(p, Deferred):
return util.DeferredResource(p)
adapter = IResource(p, None)
if adapter is not None:
return adapter
# maybe we want direct support for ModelLoader?
# cl = getattr(self, "wload_"+path, None) #???
return Resource.getChild(self, path, request)
def getChild(self, path, request):
"""See twisted.web.Resource.getChild.
"""
self.restat()
if not self.isdir():
return self.childNotFound
if path:
fpath = self.child(path)
else:
fpath = self.childSearchPreauth(*self.indexNames)
if fpath is None:
return self.directoryListing()
if not fpath.exists():
fpath = fpath.siblingExtensionSearch(*self.ignoredExts)
if fpath is None:
return self.childNotFound
if platformType == "win32":
# don't want .RPY to be different than .rpy, since that would allow
# source disclosure.
processor = InsensitiveDict(self.processors).get(fpath.splitext()[1])
else:
processor = self.processors.get(fpath.splitext()[1])
if processor:
return resource.IResource(processor(fpath.path, self.registry))
return self.createSimilarFile(fpath.path)
# methods to allow subclasses to e.g. decrypt files on the fly:
def requestAvatar(self, user, mind, *interfaces):
# the resource is passed on regardless of user
if IResource in interfaces:
return (IResource, self.resource, lambda: None)
raise NotImplementedError()
def requestAvatar(self, avatarId, mind, *interfaces):
if resource.IResource not in interfaces:
raise NotImplementedError("no interface")
if avatarId:
return (resource.IResource,
MarkAuthenticatedResource(self.resource, avatarId),
lambda:None)
else:
return resource.IResource, self.nonauthenticated, lambda:None
def __init__(self, portal, callback=None, errback=None):
"""Constructs a UsernamePasswordWrapper around the given portal.
@param portal: A cred portal for your web application. The checkers
associated with this portal must be able to accept username/password
credentials.
@type portal: L{twisted.cred.portal.Portal}
@param callback: Gets called after a successful login attempt.
A resource that redirects to "." will display the avatar resource.
If this parameter isn't provided, defaults to a standard Woven
"Thank You" page.
@type callback: A callable that accepts a Woven
L{model<twisted.web.woven.interfaces.IModel>} and returns a
L{IResource<twisted.web.resource.Resource>}.
@param errback: Gets called after a failed login attempt.
If this parameter is not provided, defaults to a the standard Woven
form error (i.e. The original form on a page of its own, with
errors noted.)
@type errback: A callable that accepts a Woven
L{model<twisted.web.woven.interfaces.IModel>} and returns a
L{IResource<twisted.web.resource.Resource>}.
"""
Resource.__init__(self)
self.portal = portal
self.callback = callback
self.errback = errback
def getChild(self, path, request):
s = request.getSession()
if s is None:
return request.setupSession()
if path == INIT_PERSPECTIVE:
def loginSuccess(result):
interface, avatarAspect, logout = result
s.setResourceForPortal(avatarAspect, self.portal, logout)
def triggerLogin(username, password, submit=None):
return self.portal.login(
UsernamePassword(username, password),
None,
IResource
).addCallback(
loginSuccess
).addErrback(
self._ebFilter
)
return form.FormProcessor(
newLoginSignature.method(
triggerLogin
),
callback=self.callback,
errback=self.errback
)
elif path == DESTROY_PERSPECTIVE:
s.portalLogout(self.portal)
return Redirect(".")
else:
r = s.resourceForPortal(self.portal)
if r:
## Delegate our getChild to the resource our portal says is the right one.
return getResource(r[0], path, request)
else:
return DeferredResource(
self.portal.login(Anonymous(), None, IResource
).addCallback(
lambda (interface, avatarAspect, logout):
getResource(s.setResourceForPortal(avatarAspect,
self.portal, logout),
path, request)))
def registerControllerForModel(controller, model):
"""
Registers `controller' as an adapter of `model' for IController, and
optionally registers it for IResource, if it implements it.
@param controller: A class that implements L{interfaces.IController}, usually a
L{Controller} subclass. Optionally it can implement
L{resource.IResource}.
@param model: Any class, but probably a L{twisted.web.woven.model.Model}
subclass.
"""
components.registerAdapter(controller, model, interfaces.IController)
if resource.IResource.implementedBy(controller):
components.registerAdapter(controller, model, resource.IResource)
def getChild(self, path, request):
if path == '': path = 'index'
path = path.replace(".","_")
cm = getattr(self, "wchild_"+path, None)
if cm:
p = cm(request)
if isinstance(p, Deferred):
return util.DeferredResource(p)
adapter = IResource(p, None)
if adapter is not None:
return adapter
# maybe we want direct support for ModelLoader?
# cl = getattr(self, "wload_"+path, None) #???
return Resource.getChild(self, path, request)
def getChild(self, path, request):
"""See twisted.web.Resource.getChild.
"""
self.restat()
if not self.isdir():
return self.childNotFound
if path:
fpath = self.child(path)
else:
fpath = self.childSearchPreauth(*self.indexNames)
if fpath is None:
return self.directoryListing()
if not fpath.exists():
fpath = fpath.siblingExtensionSearch(*self.ignoredExts)
if fpath is None:
return self.childNotFound
if platformType == "win32":
# don't want .RPY to be different than .rpy, since that would allow
# source disclosure.
processor = InsensitiveDict(self.processors).get(fpath.splitext()[1])
else:
processor = self.processors.get(fpath.splitext()[1])
if processor:
return resource.IResource(processor(fpath.path, self.registry))
return self.createSimilarFile(fpath.path)
# methods to allow subclasses to e.g. decrypt files on the fly:
def getChildWithDefault(self, name, request):
"""
Reject attempts to retrieve a child resource. All path segments beyond
the one which refers to this resource are handled by the WebSocket
connection.
@type name: C{bytes}
@param name: A single path component from a requested URL.
@type request: L{twisted.web.iweb.IRequest} provider
@param request: The request received.
"""
raise RuntimeError(
"Cannot get IResource children from WebSocketsResource")
def putChild(self, path, child):
"""
Reject attempts to add a child resource to this resource. The
WebSocket connection handles all path segments beneath this resource,
so L{IResource} children can never be found.
@type path: C{bytes}
@param path: A single path component.
@type child: L{IResource} provider
@param child: A resource to put underneat this one.
"""
raise RuntimeError(
"Cannot put IResource children under WebSocketsResource")
def requestAvatar(self, avatarId, mind, *interfaces):
if IResource in interfaces:
return IResource, RootResource(
self.facadeClient, session=avatarId), lambda: None
raise NotImplementedError()
def requestAvatar(self, avatarId, mind, *interfaces):
if IResource in interfaces:
if avatarId == checkers.ANONYMOUS:
return (IResource, TestHTTPUser(self.template, 'anonymous'),
lambda: None)
return (IResource, TestHTTPUser(self.template, avatarId),
lambda: None)
raise NotImplementedError("Only IResource interface is supported")
def requestAvatar(self, avatarId, mind, *interfaces):
if IResource in interfaces:
self.loggedIn += 1
return IResource, self.avatarFactory(avatarId), self.logout
raise NotImplementedError()
def _authorizedBasicLogin(self, request):
"""
Add an I{basic authorization} header to the given request and then
dispatch it, starting from C{self.wrapper} and returning the resulting
L{IResource}.
"""
authorization = b64encode(self.username + b':' + self.password)
request.requestHeaders.addRawHeader(b'authorization',
b'Basic ' + authorization)
return getChildForRequest(self.wrapper, request)
def test_interface(self):
"""
L{UserDirectory} instances provide L{resource.IResource}.
"""
self.assertTrue(verifyObject(resource.IResource, self.directory))
def test_unsupported(self):
"""
A L{WSGIResource} cannot have L{IResource} children. Its
C{getChildWithDefault} and C{putChild} methods raise L{RuntimeError}.
"""
self.assertRaises(
RuntimeError,
self.resource.getChildWithDefault,
b"foo", Request(DummyChannel(), False))
self.assertRaises(
RuntimeError,
self.resource.putChild,
b"foo", Resource())
def getChild(self, path, request):
"""
If this L{File}'s path refers to a directory, return a L{File}
referring to the file named C{path} in that directory.
If C{path} is the empty string, return a L{DirectoryLister} instead.
"""
self.restat(reraise=False)
if not self.isdir():
return self.childNotFound
if path:
try:
fpath = self.child(path)
except filepath.InsecurePath:
return self.childNotFound
else:
fpath = self.childSearchPreauth(*self.indexNames)
if fpath is None:
return self.directoryListing()
if not fpath.exists():
fpath = fpath.siblingExtensionSearch(*self.ignoredExts)
if fpath is None:
return self.childNotFound
if platformType == "win32":
# don't want .RPY to be different than .rpy, since that would allow
# source disclosure.
processor = InsensitiveDict(self.processors).get(fpath.splitext()[1])
else:
processor = self.processors.get(fpath.splitext()[1])
if processor:
return resource.IResource(processor(fpath.path, self.registry))
return self.createSimilarFile(fpath.path)
# methods to allow subclasses to e.g. decrypt files on the fly:
def getChildWithDefault(self, name, request):
"""
Reject attempts to retrieve a child resource. All path segments beyond
the one which refers to this resource are handled by the WSGI
application object.
"""
raise RuntimeError("Cannot get IResource children from WSGIResource")
def putChild(self, path, child):
"""
Reject attempts to add a child resource to this resource. The WSGI
application object handles all path segments beneath this resource, so
L{IResource} children can never be found.
"""
raise RuntimeError("Cannot put IResource children under WSGIResource")
def _login(self, credentials):
"""
Get the L{IResource} avatar for the given credentials.
@return: A L{Deferred} which will be called back with an L{IResource}
avatar or which will errback if authentication fails.
"""
d = self._portal.login(credentials, None, IResource)
d.addCallbacks(self._loginSucceeded, self._loginFailed)
return d
def _loginSucceeded(self, args):
"""
Handle login success by wrapping the resulting L{IResource} avatar
so that the C{logout} callback will be invoked when rendering is
complete.
"""
interface, avatar, logout = args
class ResourceWrapper(proxyForInterface(IResource, 'resource')):
"""
Wrap an L{IResource} so that whenever it or a child of it
completes rendering, the cred logout hook will be invoked.
An assumption is made here that exactly one L{IResource} from
among C{avatar} and all of its children will be rendered. If
more than one is rendered, C{logout} will be invoked multiple
times and probably earlier than desired.
"""
def getChildWithDefault(self, name, request):
"""
Pass through the lookup to the wrapped resource, wrapping
the result in L{ResourceWrapper} to ensure C{logout} is
called when rendering of the child is complete.
"""
return ResourceWrapper(self.resource.getChildWithDefault(name, request))
def render(self, request):
"""
Hook into response generation so that when rendering has
finished completely (with or without error), C{logout} is
called.
"""
request.notifyFinish().addBoth(lambda ign: logout())
return super(ResourceWrapper, self).render(request)
return ResourceWrapper(avatar)
def requestAvatar(self, credentials, mind, *interfaces):
if IResource in interfaces:
return (IResource, self.resource, lambda: None)
raise NotImplementedError()
