def clean_headers(status):
"""Remove any headers which should not apply to an error response."""
import cherrypy
response = cherrypy.serving.response
# Remove headers which applied to the original content,
# but do not apply to the error page.
respheaders = response.headers
for key in ['Accept-Ranges', 'Age', 'ETag', 'Location', 'Retry-After',
'Vary', 'Content-Encoding', 'Content-Length', 'Expires',
'Content-Location', 'Content-MD5', 'Last-Modified']:
if key in respheaders:
del respheaders[key]
if status != 416:
# A server sending a response with status code 416 (Requested
# range not satisfiable) SHOULD include a Content-Range field
# with a byte-range-resp-spec of "*". The instance-length
# specifies the current length of the selected resource.
# A response with status code 206 (Partial Content) MUST NOT
# include a Content-Range field with a byte-range- resp-spec of "*".
if 'Content-Range' in respheaders:
del respheaders['Content-Range']
python类response()的实例源码
def _populate_known_types(self):
b = [x for x in vars(builtins).values()
if type(x) is type(str)]
def traverse(obj, namespace):
for name in dir(obj):
# Hack for 3.2's warning about body_params
if name == 'body_params':
continue
vtype = type(getattr(obj, name, None))
if vtype in b:
self.known_config_types[namespace + '.' + name] = vtype
traverse(cherrypy.request, 'request')
traverse(cherrypy.response, 'response')
traverse(cherrypy.server, 'server')
traverse(cherrypy.engine, 'engine')
traverse(cherrypy.log, 'log')
def run_standard_benchmarks():
print("")
print("Client Thread Report (1000 requests, 14 byte response body, "
"%s server threads):" % cherrypy.server.thread_pool)
print_report(thread_report())
print("")
print("Client Thread Report (1000 requests, 14 bytes via staticdir, "
"%s server threads):" % cherrypy.server.thread_pool)
print_report(thread_report("%s/static/index.html" % SCRIPT_NAME))
print("")
print("Size Report (1000 requests, 50 client threads, "
"%s server threads):" % cherrypy.server.thread_pool)
print_report(size_report())
# modpython and other WSGI #
def clean_headers(status):
"""Remove any headers which should not apply to an error response."""
import cherrypy
response = cherrypy.serving.response
# Remove headers which applied to the original content,
# but do not apply to the error page.
respheaders = response.headers
for key in ["Accept-Ranges", "Age", "ETag", "Location", "Retry-After",
"Vary", "Content-Encoding", "Content-Length", "Expires",
"Content-Location", "Content-MD5", "Last-Modified"]:
if key in respheaders:
del respheaders[key]
if status != 416:
# A server sending a response with status code 416 (Requested
# range not satisfiable) SHOULD include a Content-Range field
# with a byte-range-resp-spec of "*". The instance-length
# specifies the current length of the selected resource.
# A response with status code 206 (Partial Content) MUST NOT
# include a Content-Range field with a byte-range- resp-spec of "*".
if "Content-Range" in respheaders:
del respheaders["Content-Range"]
def _be_ie_unfriendly(status):
import cherrypy
response = cherrypy.serving.response
# For some statuses, Internet Explorer 5+ shows "friendly error
# messages" instead of our response.body if the body is smaller
# than a given size. Fix this by returning a body over that size
# (by adding whitespace).
# See http://support.microsoft.com/kb/q218155/
s = _ie_friendly_error_sizes.get(status, 0)
if s:
s += 1
# Since we are issuing an HTTP error status, we assume that
# the entity is short, and we should just collapse it.
content = response.collapse_body()
l = len(content)
if l and l < s:
# IN ADDITION: the response must be written to IE
# in one chunk or it will still get replaced! Bah.
content = content + (ntob(" ") * (s - l))
response.body = content
response.headers['Content-Length'] = str(len(content))
def _populate_known_types(self):
b = [x for x in vars(builtins).values()
if type(x) is type(str)]
def traverse(obj, namespace):
for name in dir(obj):
# Hack for 3.2's warning about body_params
if name == 'body_params':
continue
vtype = type(getattr(obj, name, None))
if vtype in b:
self.known_config_types[namespace + "." + name] = vtype
traverse(cherrypy.request, "request")
traverse(cherrypy.response, "response")
traverse(cherrypy.server, "server")
traverse(cherrypy.engine, "engine")
traverse(cherrypy.log, "log")
def jsonify_error(status, message, traceback, version):
response = cherrypy.response
response.headers['Content-Type'] = 'application/json'
return json.dumps({'status': int(status.split()[0]), 'message': message})
cherrypyserver.py 文件源码
项目：arduino-ciao-meteor-ddp-connector
作者: andrea689
项目源码
文件源码
阅读 20
收藏 0
点赞 0
评论 0
def cleanup_headers(self):
"""
Some clients aren't that smart when it comes to
headers lookup.
"""
response = cherrypy.response
if not response.header_list:
return
headers = response.header_list[:]
for (k, v) in headers:
if k[:7] == 'Sec-Web':
response.header_list.remove((k, v))
response.header_list.append((k.replace('Sec-Websocket', 'Sec-WebSocket'), v))
def cleanup_headers(self):
"""
Some clients aren't that smart when it comes to
headers lookup.
"""
response = cherrypy.response
if not response.header_list:
return
headers = response.header_list[:]
for (k, v) in headers:
if k[:7] == 'Sec-Web':
response.header_list.remove((k, v))
response.header_list.append((k.replace('Sec-Websocket', 'Sec-WebSocket'), v))
def _set_response(body):
# The XML-RPC spec (http://www.xmlrpc.com/spec) says:
# "Unless there's a lower-level error, always return 200 OK."
# Since Python's xmlrpclib interprets a non-200 response
# as a "Protocol Error", we'll just return 200 every time.
response = cherrypy.response
response.status = '200 OK'
response.body = ntob(body, 'utf-8')
response.headers['Content-Type'] = 'text/xml'
response.headers['Content-Length'] = len(body)
def __init__(self, urls, status=None, encoding=None):
import cherrypy
request = cherrypy.serving.request
if isinstance(urls, text_or_bytes):
urls = [urls]
abs_urls = []
for url in urls:
url = tonative(url, encoding or self.encoding)
# Note that urljoin will "do the right thing" whether url is:
# 1. a complete URL with host (e.g. "http://www.example.com/test")
# 2. a URL relative to root (e.g. "/dummy")
# 3. a URL relative to the current path
# Note that any query string in cherrypy.request is discarded.
url = _urljoin(cherrypy.url(), url)
abs_urls.append(url)
self.urls = abs_urls
# RFC 2616 indicates a 301 response code fits our goal; however,
# browser support for 301 is quite messy. Do 302/303 instead. See
# http://www.alanflavell.org.uk/www/post-redirect.html
if status is None:
if request.protocol >= (1, 1):
status = 303
else:
status = 302
else:
status = int(status)
if status < 300 or status > 399:
raise ValueError('status must be between 300 and 399.')
self.status = status
CherryPyException.__init__(self, abs_urls, status)
def set_response(self):
"""Modify cherrypy.response status, headers, and body to represent
self.
CherryPy uses this internally, but you can also use it to create an
HTTPError object and set its output without *raising* the exception.
"""
import cherrypy
response = cherrypy.serving.response
clean_headers(self.code)
# In all cases, finalize will be called after this method,
# so don't bother cleaning up response values here.
response.status = self.status
tb = None
if cherrypy.serving.request.show_tracebacks:
tb = format_exc()
response.headers.pop('Content-Length', None)
content = self.get_error_page(self.status, traceback=tb,
message=self._message)
response.body = content
_be_ie_unfriendly(self.code)
def _set_response(body):
# The XML-RPC spec (http://www.xmlrpc.com/spec) says:
# "Unless there's a lower-level error, always return 200 OK."
# Since Python's xmlrpclib interprets a non-200 response
# as a "Protocol Error", we'll just return 200 every time.
response = cherrypy.response
response.status = '200 OK'
response.body = ntob(body, 'utf-8')
response.headers['Content-Type'] = 'text/xml'
response.headers['Content-Length'] = len(body)
def run(self, method, path, query_string, protocol, headers, rfile):
cherrypy.response.status = "200 OK"
cherrypy.response.header_list = [("Content-Type", 'text/html'),
("Server", "Null CherryPy"),
("Date", httputil.HTTPDate()),
("Content-Length", "0"),
]
cherrypy.response.body = [""]
return cherrypy.response
def __init__(self, urls, status=None, encoding=None):
import cherrypy
request = cherrypy.serving.request
if isinstance(urls, text_or_bytes):
urls = [urls]
abs_urls = []
for url in urls:
url = tonative(url, encoding or self.encoding)
# Note that urljoin will "do the right thing" whether url is:
# 1. a complete URL with host (e.g. "http://www.example.com/test")
# 2. a URL relative to root (e.g. "/dummy")
# 3. a URL relative to the current path
# Note that any query string in cherrypy.request is discarded.
url = _urljoin(cherrypy.url(), url)
abs_urls.append(url)
self.urls = abs_urls
# RFC 2616 indicates a 301 response code fits our goal; however,
# browser support for 301 is quite messy. Do 302/303 instead. See
# http://www.alanflavell.org.uk/www/post-redirect.html
if status is None:
if request.protocol >= (1, 1):
status = 303
else:
status = 302
else:
status = int(status)
if status < 300 or status > 399:
raise ValueError("status must be between 300 and 399.")
self.status = status
CherryPyException.__init__(self, abs_urls, status)
def set_response(self):
"""Modify cherrypy.response status, headers, and body to represent
self.
CherryPy uses this internally, but you can also use it to create an
HTTPError object and set its output without *raising* the exception.
"""
import cherrypy
response = cherrypy.serving.response
clean_headers(self.code)
# In all cases, finalize will be called after this method,
# so don't bother cleaning up response values here.
response.status = self.status
tb = None
if cherrypy.serving.request.show_tracebacks:
tb = format_exc()
response.headers.pop('Content-Length', None)
content = self.get_error_page(self.status, traceback=tb,
message=self._message)
response.body = content
_be_ie_unfriendly(self.code)
def default_error_page(**kwargs):
"""This function is registered as the default error page
for CherryPy errors. This sets the response headers to
be uncacheable, and then returns a HTTP response that is
identical to the default CherryPy message format."""
response = cherrypy.response
for key in ('Cache-Control', 'Pragma'):
if key in response.headers:
del response.headers[key]
return _HTTPErrorTemplate % kwargs
def file_2(self, *tokens):
"""Outputs the contents of the file, named by the SHA hash
name in the request path, directly to the client."""
method = cherrypy.request.method
if method == "HEAD":
try:
fhash = tokens[0]
except IndexError:
fhash = None
try:
fpath = self.repo.file(fhash,
pub=self._get_req_pub())
except srepo.RepositoryFileNotFoundError as e:
raise cherrypy.HTTPError(http_client.NOT_FOUND,
str(e))
except srepo.RepositoryError as e:
# Treat any remaining repository error as a 404,
# but log the error and include the real failure
# information.
cherrypy.log("Request failed: {0}".format(
str(e)))
raise cherrypy.HTTPError(http_client.NOT_FOUND,
str(e))
csize, chashes = misc.compute_compressed_attrs(fhash,
file_path=fpath)
response = cherrypy.response
for i, attr in enumerate(chashes):
response.headers["X-Ipkg-Attr-{0}".format(i)] = \
"{0}={1}".format(attr, chashes[attr])
# set expiration of response to one day
self.__set_response_expires("file", 86400, 86400)
return serve_file(fpath, "application/data")
return self.file_1(*tokens)
def open_0(self, *tokens):
"""Starts a transaction for the package name specified in the
request path. Returns no output."""
request = cherrypy.request
response = cherrypy.response
client_release = request.headers.get("Client-Release", None)
try:
pfmri = tokens[0]
except IndexError:
pfmri = None
# XXX Authentication will be handled by virtue of possessing a
# signed certificate (or a more elaborate system).
if not pfmri:
raise cherrypy.HTTPError(http_client.BAD_REQUEST,
_("A valid package FMRI must be specified."))
try:
pfmri = fmri.PkgFmri(pfmri, client_release)
trans_id = self.repo.open(client_release, pfmri)
except (fmri.FmriError, srepo.RepositoryError) as e:
# Assume a bad request was made. A 404 can't be
# returned here as misc.versioned_urlopen will interpret
# that to mean that the server doesn't support this
# operation.
cherrypy.log("Request failed: {0}".format(e))
raise cherrypy.HTTPError(http_client.BAD_REQUEST, str(e))
if pfmri.publisher and not self._get_req_pub():
self.__map_pub_ops(pfmri.publisher)
# Set response headers before returning.
response.headers["Content-type"] = "text/plain; charset=utf-8"
response.headers["Transaction-ID"] = trans_id
def append_0(self, *tokens):
"""Starts an append transaction for the package name specified
in the request path. Returns no output."""
request = cherrypy.request
response = cherrypy.response
client_release = request.headers.get("Client-Release", None)
try:
pfmri = tokens[0]
except IndexError:
pfmri = None
# XXX Authentication will be handled by virtue of possessing a
# signed certificate (or a more elaborate system).
if not pfmri:
raise cherrypy.HTTPError(http_client.BAD_REQUEST,
_("A valid package FMRI must be specified."))
try:
pfmri = fmri.PkgFmri(pfmri, client_release)
trans_id = self.repo.append(client_release, pfmri)
except (fmri.FmriError, srepo.RepositoryError) as e:
# Assume a bad request was made. A 404 can't be
# returned here as misc.versioned_urlopen will interpret
# that to mean that the server doesn't support this
# operation.
cherrypy.log("Request failed: {0}".format(e))
raise cherrypy.HTTPError(http_client.BAD_REQUEST, str(e))
if pfmri.publisher and not self._get_req_pub():
self.__map_pub_ops(pfmri.publisher)
# Set response headers before returning.
response.headers["Content-type"] = "text/plain; charset=utf-8"
response.headers["Transaction-ID"] = trans_id
def __upload_file(self, *tokens):
"""Adds a file to an in-flight transaction for the Transaction
ID specified in the request path. The content is expected to be
in the request body. Returns no output."""
try:
# cherrypy decoded it, but we actually need it encoded.
trans_id = quote(tokens[0], "")
except IndexError:
raise
trans_id = None
request = cherrypy.request
response = cherrypy.response
size = int(request.headers.get("Content-Length", 0))
if size < 0:
raise cherrypy.HTTPError(http_client.BAD_REQUEST,
_("file/1 must be sent a file."))
data = request.rfile
attrs = dict(
val.split("=", 1)
for hdr, val in request.headers.items()
if hdr.lower().startswith("x-ipkg-setattr")
)
basename = attrs.get("basename", None)
try:
self.repo.add_file(trans_id, data, basename, size)
except srepo.RepositoryError as e:
# Assume a bad request was made. A 404 can't be
# returned here as misc.versioned_urlopen will interpret
# that to mean that the server doesn't support this
# operation.
raise cherrypy.HTTPError(http_client.BAD_REQUEST, str(e))
response.headers["Content-Length"] = "0"
return response.body
def __upload_manifest(self, *tokens):
"""Adds a file to an in-flight transaction for the Transaction
ID specified in the request path. The content is expected to be
in the request body. Returns no output."""
try:
# cherrypy decoded it, but we actually need it encoded.
trans_id = quote(tokens[0], "")
except IndexError:
raise
trans_id = None
request = cherrypy.request
response = cherrypy.response
size = int(request.headers.get("Content-Length", 0))
if size < 0:
raise cherrypy.HTTPError(http_client.BAD_REQUEST,
_("manifest/1 must be sent a file."))
data = request.rfile
try:
self.repo.add_manifest(trans_id, data)
except srepo.RepositoryError as e:
# Assume a bad request was made. A 404 can't be
# returned here as misc.versioned_urlopen will interpret
# that to mean that the server doesn't support this
# operation.
raise cherrypy.HTTPError(http_client.BAD_REQUEST, str(e))
response.headers["Content-Length"] = "0"
return response.body
def index_0(self, *tokens):
"""Provides an administrative interface for search indexing.
Returns no output if successful; otherwise the response body
will contain the failure details.
"""
try:
cmd = tokens[0]
except IndexError:
cmd = ""
# These commands cause the operation requested to be queued
# for later execution. This does mean that if the operation
# fails, the client won't know about it, but this is necessary
# since these are long running operations (are likely to exceed
# connection timeout limits).
try:
if cmd == "refresh":
# Update search indexes.
self.__bgtask.put(self.repo.refresh_index,
pub=self._get_req_pub())
else:
err = "Unknown index subcommand: {0}".format(
cmd)
cherrypy.log(err)
raise cherrypy.HTTPError(http_client.NOT_FOUND, err)
except queue.Full:
raise cherrypy.HTTPError(http_client.SERVICE_UNAVAILABLE,
"Another operation is already in progress; try "
"again later.")
def nasty_before_handler(nasty_depot, maxroll=100):
"""Cherrypy Tool callable which generates various problems prior to a
request. Possible outcomes: retryable HTTP error, short nap."""
# Must be set in _cp_config on associated request handler.
assert nasty_depot
# Adjust nastiness values once per incoming request.
nasty_depot.nasty_housekeeping()
# Just roll the main nasty dice once.
if not nasty_depot.need_nasty(maxroll=maxroll):
return False
while True:
roll = random.randint(0, 10)
if roll == 0:
nasty_depot.nasty_nap()
if random.randint(0, 1) == 1:
# Nap was enough. Let the normal handler run.
return False
if 1 <= roll <= 8:
nasty_depot.nasty_raise_error()
else:
cherrypy.log("NASTY: return bogus or empty response")
response = cherrypy.response
response.body = random.choice(['',
'set this is a bogus action',
'Instead of office chair, '
'package contained bobcat.',
'{"this is a": "fragment of json"}'])
return True
return False
def info(self, *tokens):
"""Use a DepotHTTP to return an info response."""
dh = self.__build_depot_http()
tokens = self.__strip_pub(tokens, dh.repo)
# In Python 3, a WSGI application must return bytes as its output
return misc.force_bytes(dh.info_0(*tokens[3:]))
def p5i(self, *tokens):
"""Use a DepotHTTP to return a p5i response."""
dh = self.__build_depot_http()
tokens = self.__strip_pub(tokens, dh.repo)
headers = cherrypy.response.headers
headers["Content-Type"] = pkg.p5i.MIME_TYPE
return dh.p5i_0(*tokens[3:])
def search_1(self, *tokens, **params):
"""Use a DepotHTTP to return a search/1 response."""
toks = cherrypy.request.path_info.lstrip("/").split("/")
dh = self.__build_depot_http()
toks = self.__strip_pub(tokens, dh.repo)
query_str = "/".join(toks[3:])
return dh.search_1(query_str)
def default_error_page(status=http_client.NOT_FOUND, message="oops",
traceback=None, version=None):
"""This function is registered as the default error page
for CherryPy errors. This sets the response headers to
be uncacheable, and then returns a HTTP response."""
response = cherrypy.response
for key in ('Cache-Control', 'Pragma'):
if key in response.headers:
del response.headers[key]
# Server errors are interesting, so let's log them. In the case
# of an internal server error, we send a 404 to the client. but
# log the full details in the server log.
if (status == http_client.INTERNAL_SERVER_ERROR or
status.startswith("500 ")):
# Convert the error to a 404 to obscure implementation
# from the client, but log the original error to the
# server logs.
error = cherrypy._cperror._HTTPErrorTemplate % \
{"status": http_client.NOT_FOUND,
"message": http_client.responses[http_client.NOT_FOUND],
"traceback": "",
"version": cherrypy.__version__}
print("Path that raised exception was {0}".format(
cherrypy.request.path_info))
print(message)
return error
else:
error = cherrypy._cperror._HTTPErrorTemplate % \
{"status": http_client.NOT_FOUND, "message": message,
"traceback": "", "version": cherrypy.__version__}
return error
def set_response(self):
"""Modify cherrypy.response status, headers, and body to represent
self.
CherryPy uses this internally, but you can also use it to create an
HTTPRedirect object and set its output without *raising* the exception.
"""
import cherrypy
response = cherrypy.serving.response
response.status = status = self.status
if status in (300, 301, 302, 303, 307):
response.headers['Content-Type'] = 'text/html;charset=utf-8'
# "The ... URI SHOULD be given by the Location field
# in the response."
response.headers['Location'] = self.urls[0]
# "Unless the request method was HEAD, the entity of the response
# SHOULD contain a short hypertext note with a hyperlink to the
# new URI(s)."
msg = {
300: 'This resource can be found at ',
301: 'This resource has permanently moved to ',
302: 'This resource resides temporarily at ',
303: 'This resource can be found at ',
307: 'This resource has moved temporarily to ',
}[status]
msg += '<a href=%s>%s</a>.'
msgs = [msg % (saxutils.quoteattr(u), u) for u in self.urls]
response.body = ntob('<br />\n'.join(msgs), 'utf-8')
# Previous code may have set C-L, so we have to reset it
# (allow finalize to set it).
response.headers.pop('Content-Length', None)
elif status == 304:
# Not Modified.
# "The response MUST include the following header fields:
# Date, unless its omission is required by section 14.18.1"
# The "Date" header should have been set in Response.__init__
# "...the response SHOULD NOT include other entity-headers."
for key in ('Allow', 'Content-Encoding', 'Content-Language',
'Content-Length', 'Content-Location', 'Content-MD5',
'Content-Range', 'Content-Type', 'Expires',
'Last-Modified'):
if key in response.headers:
del response.headers[key]
# "The 304 response MUST NOT contain a message-body."
response.body = None
# Previous code may have set C-L, so we have to reset it.
response.headers.pop('Content-Length', None)
elif status == 305:
# Use Proxy.
# self.urls[0] should be the URI of the proxy.
response.headers['Location'] = ntob(self.urls[0], 'utf-8')
response.body = None
# Previous code may have set C-L, so we have to reset it.
response.headers.pop('Content-Length', None)
else:
raise ValueError('The %s status code is unknown.' % status)
def testHookErrors(self):
self.getPage("/demo/?id=1")
# If body is "razdrez", then on_end_request is being called too early.
self.assertBody("A horrorshow lomtick of cherry 3.14159")
# If this fails, then on_end_request isn't being called at all.
time.sleep(0.1)
self.getPage("/demo/ended/1")
self.assertBody("True")
valerr = '\n raise ValueError()\nValueError'
self.getPage("/demo/err?id=3")
# If body is "razdrez", then on_end_request is being called too early.
self.assertErrorPage(502, pattern=valerr)
# If this fails, then on_end_request isn't being called at all.
time.sleep(0.1)
self.getPage("/demo/ended/3")
self.assertBody("True")
# If body is "razdrez", then on_end_request is being called too early.
if (cherrypy.server.protocol_version == "HTTP/1.0" or
getattr(cherrypy.server, "using_apache", False)):
self.getPage("/demo/errinstream?id=5")
# Because this error is raised after the response body has
# started, the status should not change to an error status.
self.assertStatus("200 OK")
self.assertBody("nonconfidential")
else:
# Because this error is raised after the response body has
# started, and because it's chunked output, an error is raised by
# the HTTP client when it encounters incomplete output.
self.assertRaises((ValueError, IncompleteRead), self.getPage,
"/demo/errinstream?id=5")
# If this fails, then on_end_request isn't being called at all.
time.sleep(0.1)
self.getPage("/demo/ended/5")
self.assertBody("True")
# Test the "__call__" technique (compile-time decorator).
self.getPage("/demo/restricted")
self.assertErrorPage(401)
# Test compile-time decorator with kwargs from config.
self.getPage("/demo/userid")
self.assertBody("Welcome!")
