Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def send_redirect(self, new_path, url_parts, urlrewriter):
scheme, netloc, path, query, frag = url_parts
path = new_path
url = urlunsplit((scheme, netloc, path, query, frag))
resp = WbResponse.redir_response(urlrewriter.rewrite(url),
'307 Temporary Redirect')
if self.enable_memento:
resp.status_headers['Link'] = MementoUtils.make_link(url, 'original')
return resp
def options_response(env):
"""Construct WbResponse for OPTIONS based on the WSGI env dictionary
:param dict env: The WSGI environment dictionary
:return: The WBResponse for the options request
:rtype: WbResponse
"""
status_headers = StatusAndHeaders('200 Ok', [
('Content-Type', 'text/plain'),
('Content-Length', '0'),
])
response = WbResponse(status_headers)
response.add_access_control_headers(env=env)
return response
cdx_url += environ.get('QUERY_STRING')
if self.query_limit:
cdx_url += '&' if '?' in cdx_url else '?'
cdx_url += 'limit=' + str(self.query_limit)
try:
res = requests.get(cdx_url, stream=True)
content_type = res.headers.get('Content-Type')
return WbResponse.bin_stream(StreamIter(res.raw),
content_type=content_type)
except Exception as e:
return WbResponse.text_response('Error: ' + str(e), status='400 Bad Request')
def text_stream(stream, content_type='text/plain; charset=utf-8', status='200 OK'):
"""Utility method for constructing a streaming text response.
:param Any stream: The response body stream
:param str content_type: The content-type of the response
:param str status: The HTTP status line
:return: WbResponse that is a text stream
:rtype WbResponse:
"""
if 'charset' not in content_type:
content_type += '; charset=utf-8'
return WbResponse.bin_stream(WbResponse.encode_stream(stream), content_type, status)
inx = referer[1:].find('http')
if not inx:
inx = referer[1:].find('///')
if inx < 0:
return
url = referer[inx + 1:]
host = referer[:inx + 1]
orig_url = environ['PATH_INFO']
if environ.get('QUERY_STRING'):
orig_url += '?' + environ['QUERY_STRING']
full_url = host + urljoin(url, orig_url)
return WbResponse.redir_response(full_url, '307 Redirect')
def format_response(self, response, wb_url, full_prefix, is_timegate, is_proxy):
memento_ts = None
if not isinstance(response, WbResponse):
content_type = 'text/html'
# if not replay outer frame, specify utf-8 charset
if not self.is_framed_replay(wb_url):
content_type += '; charset=utf-8'
else:
memento_ts = wb_url.timestamp
response = WbResponse.text_response(response, content_type=content_type)
if self.enable_memento and response.status_headers.statusline.startswith('200'):
self._add_memento_links(wb_url.url, full_prefix, None, memento_ts,
response.status_headers, is_timegate, is_proxy)
return response
def redir_response(location, status='302 Redirect', headers=None):
"""Utility method for constructing redirection response.
:param str location: The location of the resource redirecting to
:param str status: The HTTP status line
:param list[tuple[str, str]] headers: Additional headers for this response
:return: WbResponse redirection response
:rtype: WbResponse
"""
redir_headers = [('Location', location), ('Content-Length', '0')]
if headers:
redir_headers += headers
return WbResponse(StatusAndHeaders(status, redir_headers))
:return: The WbResponse for serving the home (/) path
:rtype: WbResponse
"""
home_view = BaseInsertView(self.rewriterapp.jinja_env, 'index.html')
fixed_routes = self.warcserver.list_fixed_routes()
dynamic_routes = self.warcserver.list_dynamic_routes()
routes = fixed_routes + dynamic_routes
all_metadata = self.metadata_cache.get_all(dynamic_routes)
content = home_view.render_to_string(environ,
routes=routes,
all_metadata=all_metadata)
return WbResponse.text_response(content, content_type='text/html; charset="utf-8"')