Python flask.request 模块，path() 实例源码
我们从Python开源项目中，提取了以下49个代码示例，用于说明如何使用flask.request.path()。
def build_sitemap():
from redberry.models import RedPost, RedCategory
from apesmit import Sitemap
sm = Sitemap(changefreq='weekly')
for post in RedPost.all_published():
sm.add(url_for('redberry.show_post', slug=post.slug, _external=True), lastmod=post.updated_at.date())
for category in RedCategory.query.all():
sm.add(url_for('redberry.show_category', category_slug=category.slug, _external=True), lastmod=category.updated_at.date())
with open(os.path.join(REDBERRY_ROOT, 'static', 'redberry', 'sitemap.xml'), 'w') as f:
sm.write(f)
flash("Sitemap created.", 'success')
return redirect(url_for('redberry.home'))
##############
# ADMIN ROUTES
##############
def after_request(response):
response.headers.add('Access-Control-Allow-Methods', 'GET, POST')
response.headers.add('Access-Control-Allow-Credentials', 'true')
response.headers.add('Access-Control-Allow-Headers', 'Content-Type, *')
response.headers.add('Cache-Control', 'no-cache')
response.headers.add('Cache-Control', 'no-store')
if api.auth.is_logged_in():
if 'token' in session:
response.set_cookie('token', session['token'], domain=app.config['SESSION_COOKIE_DOMAIN'])
else:
csrf_token = api.common.token()
session['token'] = csrf_token
response.set_cookie('token', csrf_token, domain=app.config['SESSION_COOKIE_DOMAIN'])
# JB: This is a hack. We need a better solution
if request.path[0:19] != "/api/autogen/serve/":
response.mimetype = 'application/json'
return response
def make_pagination_headers(limit, curpage, total, link_header=True):
"""Return Link Hypermedia Header."""
lastpage = int(math.ceil(1.0 * total / limit) - 1)
headers = {'X-Total-Count': str(total), 'X-Limit': str(limit),
'X-Page-Last': str(lastpage), 'X-Page': str(curpage)}
if not link_header:
return headers
base = "{}?%s".format(request.path)
links = {}
links['first'] = base % urlencode(dict(request.args, **{PAGE_ARG: 0}))
links['last'] = base % urlencode(dict(request.args, **{PAGE_ARG: lastpage}))
if curpage:
links['prev'] = base % urlencode(dict(request.args, **{PAGE_ARG: curpage - 1}))
if curpage < lastpage:
links['next'] = base % urlencode(dict(request.args, **{PAGE_ARG: curpage + 1}))
headers['Link'] = ",".join(['<%s>; rel="%s"' % (v, n) for n, v in links.items()])
return headers
# pylama:ignore=R0201
def iam_sts_credentials(api_version, requested_role, junk=None):
if not _supports_iam(api_version):
return passthrough(request.path)
try:
role_params = roles.get_role_params_from_ip(
request.remote_addr,
requested_role=requested_role
)
except roles.UnexpectedRoleError:
msg = "Role name {0} doesn't match expected role for container"
log.error(msg.format(requested_role))
return '', 404
log.debug('Providing assumed role credentials for {0}'.format(role_params['name']))
assumed_role = roles.get_assumed_role_credentials(
role_params=role_params,
api_version=api_version
)
return jsonify(assumed_role)
def ip_whitelist_add(ip_to_allow, info_record_dict=None):
"""??ip????, ?????"""
if ip_to_allow in single_ip_allowed_set:
return
dbgprint('ip white added', ip_to_allow, 'info:', info_record_dict)
single_ip_allowed_set.add(ip_to_allow)
is_ip_not_in_allow_range.cache_clear()
append_ip_whitelist_file(ip_to_allow)
# dbgprint(single_ip_allowed_set)
try:
with open(zmirror_root(human_ip_verification_whitelist_log), 'a', encoding='utf-8') as fp:
fp.write(datetime.now().strftime('%Y-%m-%d %H:%M:%S') + " " + ip_to_allow
+ " " + str(request.user_agent)
+ " " + repr(info_record_dict) + "\n")
except: # coverage: exclude
errprint('Unable to write log file', os.path.abspath(human_ip_verification_whitelist_log))
traceback.print_exc()
def check_auth(func):
"""
This decorator for routes checks that the user is authorized (or that no login is required).
If they haven't, their intended destination is stored and they're sent to get authorized.
It has to be placed AFTER @app.route() so that it can capture `request.path`.
"""
if 'login' not in conf:
return func
# inspired by <https://flask-login.readthedocs.org/en/latest/_modules/flask_login.html#login_required>
@functools.wraps(func)
def decorated_view(*args, **kwargs):
if current_user.is_anonymous:
print('unauthorized user visited {!r}'.format(request.path))
session['original_destination'] = request.path
return redirect(url_for('get_authorized'))
print('{} visited {!r}'.format(current_user.email, request.path))
assert current_user.email.lower() in conf.login['whitelist'], current_user
return func(*args, **kwargs)
return decorated_view
def after_request(response):
response.headers.add('Access-Control-Allow-Methods', 'GET, POST')
response.headers.add('Access-Control-Allow-Credentials', 'true')
response.headers.add('Access-Control-Allow-Headers', 'Content-Type, *')
response.headers.add('Cache-Control', 'no-cache')
response.headers.add('Cache-Control', 'no-store')
if api.auth.is_logged_in():
if 'token' in session:
response.set_cookie('token', session['token'])
else:
csrf_token = api.common.token()
session['token'] = csrf_token
response.set_cookie('token', csrf_token)
# JB: This is a hack. We need a better solution
if request.path[0:19] != "/api/autogen/serve/":
response.mimetype = 'appication/json'
return response
def convert(content_type, content):
cachekey = request.path
cachevalue = cache.get(cachekey)
if cachevalue is None:
content = decode(content)
content_type = decode(content_type).decode()
if content_type.startswith('image/svg+xml'):
content = __svg2png(content)
content_type = 'image/png'
cache.set(cachekey, (content, content_type), timeout=CACHE_RETENTION)
else:
content = cachevalue[0]
content_type = cachevalue[1]
response = make_response(content)
response.content_type = content_type
response.cache_control.max_age = CACHE_RETENTION
return response
def init_utils(app):
app.jinja_env.filters['unix_time'] = unix_time
app.jinja_env.filters['unix_time_millis'] = unix_time_millis
app.jinja_env.filters['long2ip'] = long2ip
app.jinja_env.globals.update(pages=pages)
app.jinja_env.globals.update(can_register=can_register)
app.jinja_env.globals.update(mailserver=mailserver)
app.jinja_env.globals.update(ctf_name=ctf_name)
@app.context_processor
def inject_user():
if authed():
return dict(session)
return dict()
@app.before_request
def needs_setup():
if request.path == '/setup' or request.path.startswith('/static'):
return
if not is_setup():
return redirect('/setup')
def after_request(response):
response.headers.add('Access-Control-Allow-Methods', 'GET, POST')
response.headers.add('Access-Control-Allow-Credentials', 'true')
response.headers.add('Access-Control-Allow-Headers', 'Content-Type, *')
response.headers.add('Cache-Control', 'no-cache')
response.headers.add('Cache-Control', 'no-store')
if api.auth.is_logged_in():
if 'token' in session:
response.set_cookie('token', session['token'])
else:
csrf_token = api.common.token()
session['token'] = csrf_token
response.set_cookie('token', csrf_token)
# JB: This is a hack. We need a better solution
if request.path[0:19] != "/api/autogen/serve/":
response.mimetype = 'appication/json'
return response
def _atlassian_jwt_post_token(self):
if not getattr(g, 'ac_client', None):
return dict()
args = request.args.copy()
try:
del args['jwt']
except KeyError:
pass
signature = encode_token(
'POST',
request.path + '?' + urlencode(args),
g.ac_client.clientKey,
g.ac_client.sharedSecret)
args['jwt'] = signature
return dict(atlassian_jwt_post_url=request.path + '?' + urlencode(args))
def log_exception(self, exc_info):
self.logger.error("""
Path: %s
HTTP Method: %s
Client IP Address: %s
User Agent: %s
User Platform: %s
User Browser: %s
User Browser Version: %s
GET args: %s
view args: %s
URL: %s
""" % (
request.path,
request.method,
request.remote_addr,
request.user_agent.string,
request.user_agent.platform,
request.user_agent.browser,
request.user_agent.version,
dict(request.args),
request.view_args,
request.url
), exc_info=exc_info)
def log_request(code='-'):
proto = request.environ.get('SERVER_PROTOCOL')
msg = request.method + ' ' + request.path + ' ' + proto
code = str(code)
if code[0] == '1': # 1xx - Informational
msg = color(msg, attrs=['bold'])
if code[0] == '2': # 2xx - Success
msg = color(msg, color='white')
elif code == '304': # 304 - Resource Not Modified
msg = color(msg, color='cyan')
elif code[0] == '3': # 3xx - Redirection
msg = color(msg, color='green')
elif code == '404': # 404 - Resource Not Found
msg = color(msg, color='yellow')
elif code[0] == '4': # 4xx - Client Error
msg = color(msg, color='red', attrs=['bold'])
else: # 5xx, or any other response
msg = color(msg, color='magenta', attrs=['bold'])
logger.info('%s - - [%s] "%s" %s', request.remote_addr, log_date_time_string(), msg, code)
def authenticated(fn):
"""Mark a route as requiring authentication."""
@wraps(fn)
def decorated_function(*args, **kwargs):
if not session.get('is_authenticated'):
return redirect(url_for('login', next=request.url))
if request.path == '/logout':
return fn(*args, **kwargs)
if (not session.get('name') or
not session.get('email') or
not session.get('institution')) and request.path != '/profile':
return redirect(url_for('profile', next=request.url))
return fn(*args, **kwargs)
return decorated_function
def __call__(self, func):
@wraps(func)
def decorator(*args, **kwargs):
response = func(*args, **kwargs)
if not USE_CACHE:
return response
for path in self.routes:
route = '%s,%s,%s' % (path, json.dumps(self.args), None)
Cache.delete(unicode(route))
user = kwargs.get('user', None) if self.clear_for_user else None
Logger.debug(unicode('clear %s from cache' % (route)))
if user:
route = '%s,%s,%s' % (path, json.dumps(self.args), user)
Cache.delete(unicode(route))
Logger.debug(unicode('clear %s from cache' % (route)))
return response
return decorator
def sitemap():
conn,curr = sphinx_conn()
querysql='SELECT info_hash,create_time FROM film order by create_time desc limit 100'
curr.execute(querysql)
rows=curr.fetchall()
sphinx_close(curr,conn)
sitemaplist=[]
for row in rows:
info_hash = row['info_hash']
mtime = datetime.datetime.fromtimestamp(int(row['create_time'])).strftime('%Y-%m-%d')
url = domain+'hash/{}.html'.format(info_hash)
url_xml = '<url><loc>{}</loc><lastmod>{}</lastmod><changefreq>daily</changefreq><priority>0.8</priority></url>'.format(url, mtime)
sitemaplist.append(url_xml)
xml_content = '<?xml version="1.0" encoding="UTF-8"?><urlset>{}</urlset>'.format("".join(x for x in sitemaplist))
with open('static/sitemap.xml', 'wb') as f:
f.write(xml_content)
f.close()
return send_from_directory(app.static_folder, request.path[1:])
def chan_friends(channel):
"""
If you GET this endpoint, go to /api/v1/channel/<channel>/friend
with <channel> replaced for the channel of the friends you want to get
<channel> can either be an int that matches the channel, or a string
that matches the owner's username
"""
# model = request.path.split("/")[-1]
model = "Friend"
if channel.isdigit():
fields = {"channelId": int(channel)}
else:
fields = {"owner": channel.lower()}
packet, code = generate_response(
model,
request.path,
request.method,
request.values,
data=results,
fields=fields
)
return make_response(jsonify(packet), code)
# There was an error!
# if not str(code).startswith("2"):
# return make_response(jsonify(packet), code)
# NOTE: Not needed currently, but this is how you would check
# TODO: Fix this endpoint to remove timing elements (friends are forever)
# TODO: Use Object.update(**changes) instead of Object(**updated_object).save()
def chan_messages(channel):
"""
If you GET this endpoint, go to /api/v1/channel/<channel>/messages
with <channel> replaced for the messages you want to get
"""
model = "Message"
if channel.isdigit():
fields = {"channelId": int(channel)}
else:
fields = {"owner": channel.lower()}
packet, code = generate_response(
model,
request.path,
request.method,
request.values,
fields=fields
)
return make_response(jsonify(packet), code)
def user_quotes(channel):
"""
If you GET this endpoint, go to /api/v1/channel/<channel>/quote
with <channel> replaced for the channel you want to get quotes for
"""
model = "quote"
if channel.isdigit():
fields = {"channelId": int(channel), "deleted": False}
else:
fields = {"owner": channel.lower(), "deleted": False}
packet, code = generate_response(
model,
request.path,
request.method,
request.values,
fields=fields
)
return make_response(jsonify(packet), code)
def user_commands(channel):
"""
If you GET this endpoint, simply go to /api/v1/channel/<channel>/command
with <channel> replaced for the channel you want to get commands for
"""
model = "Command"
if channel.isdigit():
fields = {"channelId": int(channel), "deleted": False}
else:
fields = {"channelName": channel.lower(), "deleted": False}
packet, code = generate_response(
model,
request.path,
request.method,
request.values,
fields=fields
)
return make_response(jsonify(packet), code)
def after_request_log(response):
name = dns_resolve(request.remote_addr)
current_app.logger.warn(u"""[client {ip} {host}] {http} "{method} {path}" {status}
Request: {method} {path}
Version: {http}
Status: {status}
Url: {url}
IP: {ip}
Hostname: {host}
Agent: {agent_platform} | {agent_browser} | {agent_browser_version}
Raw Agent: {agent}
""".format(method=request.method,
path=request.path,
url=request.url,
ip=request.remote_addr,
host=name if name is not None else '?',
agent_platform=request.user_agent.platform,
agent_browser=request.user_agent.browser,
agent_browser_version=request.user_agent.version,
agent=request.user_agent.string,
http=request.environ.get('SERVER_PROTOCOL'),
status=response.status))
return response
def request_start():
content_type = request.headers.get('Accept') or ''
real_ip = request.headers.get('X-Real-Ip') or ''
Log.info(request.path+' '+format_args(request.args)\
+' '+real_ip\
+' '+content_type)
#Test content_type
# if content_type and content_type not in AVAILABLE_CONTENT_TYPES:
# results = {'message' : 'Content-Type not supported',
# 'message_code' : 8
# }
# return {'error' : 'content-type'}
# return self.render(results, status_code = 405)
def cached(key='view{path}'):
""" Sets up a function to have cached results """
def decorator(func):
""" Gets a cached value or calculates one """
@wraps(func)
def decorated_function(*args, **kwargs):
ckey = cache_key(key, *args, **kwargs)
value = cache.get(ckey)
if value is not None:
return value
print("DB HIT: \"{}\"".format(ckey))
value = func(*args, **kwargs)
cache.set(ckey, value)
return value
return decorated_function
return decorator
def post(year, month, day, post_name):
rel_url = request.path[len('/post/'):]
fixed_rel_url = storage.fix_post_relative_url(rel_url)
if rel_url != fixed_rel_url:
return redirect(request.url_root + 'post/' + fixed_rel_url) # it's not the correct relative url, so redirect
post_ = storage.get_post(rel_url, include_draft=False)
if post_ is None:
abort(404)
post_d = post_.to_dict()
del post_d['raw_content']
post_d['content'] = get_parser(post_.format).parse_whole(post_.raw_content)
post_d['content'], post_d['toc'], post_d['toc_html'] = parse_toc(post_d['content'])
post_d['url'] = make_abs_url(post_.unique_key)
post_ = post_d
return custom_render_template(post_['layout'] + '.html', entry=post_)
def entity(entity_id):
entity = load_entity(entity_id, request.auth)
if entity is None:
raise NotFound()
# load possible duplicates via fingerprint expansion
# fingerprints = expand_fingerprints(entity.fingerprints, request.auth)
query = Query(request.args, prefix='duplicates_', path=request.path,
limit=5)
# query.add_facet('countries', 'Countries', country_label)
duplicates = search_duplicates(entity.id, entity.fingerprints, query,
request.auth)
# load links
query = Query(request.args, prefix='links_', path=request.path,
limit=10)
query.add_facet('schemata', 'Types', link_schema_label)
query.add_facet('remote.countries', 'Countries', country)
links = search_links(entity, query, request.auth)
return render_template("entity.html", entity=entity, links=links,
duplicates=duplicates)
def store_form(form):
entry = FormEntry(form_id=g.page.id)
for f in form:
field = Field.query.filter_by(form_id=g.page.id).filter_by(name=f.name).one_or_none()
if field is None:
continue
field_entry = FieldEntry(field_id=field.id)
data = f.data
if field.type == 'file_input':
file_data = request.files[field.name]
filename = '%s-%s-%s.%s' %(field.name, date_stamp(), str(time.time()).replace('.', ''), os.path.splitext(file_data.filename)[-1])
path = os.path.join(app.config['FORM_UPLOADS_PATH'], secure_filename(filename))
file_data.save(path)
data = filename
field_entry.value = data
db.session.add(field_entry)
entry.fields.append(field_entry)
db.session.add(entry)
db.session.commit()
def route():
args = request.json
args = args if args else {}
cfg = args.get('cfg', None)
log_request(
args.get('user', 'unknown'),
args.get('hostname', 'unknown'),
request.remote_addr,
request.method,
request.path,
args)
try:
endpoint = create_endpoint(request.method, cfg, args)
json, status = endpoint.execute()
except AutocertError as ae:
status = 500
json = dict(errors={ae.name: ae.message})
return make_response(jsonify(json), status)
if not json:
raise EmptyJsonError(json)
return make_response(jsonify(json), status)
def _create_etag(path, recursive, lm=None):
if lm is None:
lm = _create_lastmodified(path, recursive)
if lm is None:
return None
hash = hashlib.sha1()
hash.update(str(lm))
hash.update(str(recursive))
if path.endswith("/files") or path.endswith("/files/sdcard"):
# include sd data in etag
hash.update(repr(sorted(printer.get_sd_files(), key=lambda x: x[0])))
return hash.hexdigest()
def get_full_path(ver):
""" L2-L4 Path Query, use onePath for single paths """
onepath = False
depth = '20'
error = version_chk(ver, ['v1.0', 'v1.1', 'v2'])
if error:
return error
if 'onepath' in request.args:
if request.args['onepath'] == "True":
onepath = True
if 'depth' in request.args:
depth = request.args['depth']
try:
return jsonify(upgrade_api(nglib.query.path.get_full_path(request.args['src'], \
request.args['dst'], {"onepath": onepath, "depth": depth}), ver))
except ResultError as e:
return jsonify(errors.json_error(e.expression, e.message))
def get_routed_path(ver):
""" Routed Paths, accepts vrf """
onepath = False
depth = '20'
vrf = 'default'
error = version_chk(ver, ['v1.0', 'v1.1', 'v2'])
if error:
return error
if 'onepath' in request.args:
if request.args['onepath'] == "True":
onepath = True
if 'depth' in request.args:
depth = request.args['depth']
if 'vrf' in request.args:
vrf = request.args['vrf']
try:
return jsonify(upgrade_api(nglib.query.path.get_routed_path(request.args['src'], \
request.args['dst'], {"onepath": onepath, "depth": depth, \
"VRF": vrf}), ver))
except ResultError as e:
return jsonify(errors.json_error(e.expression, e.message))
def render_redberry(template_name, **kwargs):
# For Accelerated Mobile Pages (AMP, ref: https://www.ampproject.org) use templates with a .amp suffix
if '.amp' in request.path:
kwargs['render_amp'] = True
return render_template(template_name, **kwargs)
def login_required(func):
@wraps(func)
def wrapper(*args, **kwargs):
if request.method == 'POST' :
if not is_authenticated():
abort(401)
else:
return func(*args, **kwargs)
else:
if not is_authenticated():
return redirect("/login/" + "?next=" + request.path)
else:
return func(*args, **kwargs)
return wrapper
def image_list(masterip):
data = {
"user": session['username']
}
# path = request.path[:request.path.rfind("/")]
# path = path[:path.rfind("/")+1]
result = dockletRequest.post("/image/list/", data, masterip)
logger.debug("image" + str(type(result)))
return json.dumps(result)
def monitor_request(comid,infotype,masterip):
data = {
"user": session['username']
}
path = request.path[request.path.find("/")+1:]
path = path[path.find("/")+1:]
path = path[path.find("/")+1:]
logger.debug(path + "_____" + masterip)
result = dockletRequest.post("/monitor/"+path, data, masterip)
logger.debug("monitor" + str(type(result)))
return json.dumps(result)
def monitor_user_request(issue,masterip):
data = {
"user": session['username']
}
path = "/monitor/user/" + str(issue) + "/"
logger.debug(path + "_____" + masterip)
result = dockletRequest.post(path, data, masterip)
logger.debug("monitor" + str(type(result)))
return json.dumps(result)
def jupyter_prefix():
path = request.args.get('next')
if path == None:
return redirect('/login/')
return redirect('/login/'+'?next='+path)
def login_required(func):
@wraps(func)
def wrapper(*args, **kwargs):
global G_usermgr
logger.info ("get request, path: %s" % request.path)
token = request.form.get("token", None)
if (token == None):
return json.dumps({'success':'false', 'message':'user or key is null'})
cur_user = G_usermgr.auth_token(token)
if (cur_user == None):
return json.dumps({'success':'false', 'message':'token failed or expired', 'Unauthorized': 'True'})
return func(cur_user, cur_user.username, request.form, *args, **kwargs)
return wrapper
def get_request_information():
"""
Returns a dictionary of contextual information about the user at the time of logging.
Returns:
The dictionary.
"""
information = {}
if has_request_context():
information["request"] = {
"api_endpoint_method": request.method,
"api_endpoint": request.path,
"ip": request.remote_addr,
"platform": request.user_agent.platform,
"browser": request.user_agent.browser,
"browser_version": request.user_agent.version,
"user_agent":request.user_agent.string
}
if api.auth.is_logged_in():
user = api.user.get_user()
team = api.user.get_team()
groups = api.team.get_groups()
information["user"] = {
"username": user["username"],
"email": user["email"],
"team_name": team["team_name"],
"groups": [group["name"] for group in groups]
}
return information
def teacher_session():
if '/teacher/' in request.path:
if 'credentials' not in flask.session:
return flask.redirect(flask.url_for('index'))
elif not flask.session['is_teacher']:
return flask.redirect(flask.url_for('register'))
def student_session():
if '/student/' in request.path:
if 'credentials' not in flask.session:
return flask.redirect(flask.url_for('index'))
elif not flask.session['is_student']:
return flask.redirect(flask.url_for('register'))
# make sure user is authenticated w/ live session on every request
def manage_session():
# want to go through oauth flow for this route specifically
# not get stuck in redirect loop
if request.path == '/oauth/callback':
return
# allow all users to visit the index page without a session
if request.path == '/' or request.path == '/oauth/logout':
return
# validate that user has valid session
# add the google user info into session
if 'credentials' not in flask.session:
flask.session['redirect'] = request.path
return flask.redirect(flask.url_for('oauth2callback'))
def iam_role_info(api_version, junk=None):
if not _supports_iam(api_version):
return passthrough(request.path)
role_params_from_ip = roles.get_role_params_from_ip(request.remote_addr)
if role_params_from_ip['name']:
log.debug('Providing IAM role info for {0}'.format(role_params_from_ip['name']))
return jsonify(roles.get_role_info_from_params(role_params_from_ip))
else:
log.error('Role name not found; returning 404.')
return '', 404
def iam_role_name(api_version):
if not _supports_iam(api_version):
return passthrough(request.path)
role_params_from_ip = roles.get_role_params_from_ip(request.remote_addr)
if role_params_from_ip['name']:
return role_params_from_ip['name']
else:
log.error('Role name not found; returning 404.')
return '', 404
def load_ip_whitelist_file():
"""?????ip???"""
set_buff = set()
if os.path.exists(zmirror_root(human_ip_verification_whitelist_file_path)):
with open(zmirror_root(human_ip_verification_whitelist_file_path), 'r', encoding='utf-8') as fp:
set_buff.add(fp.readline().strip())
return set_buff
def response_cookies_deep_copy():
"""
It's a BAD hack to get RAW cookies headers, but so far, we don't have better way.
We'd go DEEP inside the urllib's private method to get raw headers
raw_headers example:
[('Cache-Control', 'private'),
('Content-Length', '48234'),
('Content-Type', 'text/html; Charset=utf-8'),
('Server', 'Microsoft-IIS/8.5'),
('Set-Cookie','BoardList=BoardID=Show; expires=Mon, 02-May-2016 16:00:00 GMT; path=/'),
('Set-Cookie','aspsky=abcefgh; expires=Sun, 24-Apr-2016 16:00:00 GMT; path=/; HttpOnly'),
('Set-Cookie', 'ASPSESSIONIDSCSSDSSQ=OGKMLAHDHBFDJCDMGBOAGOMJ; path=/'),
('X-Powered-By', 'ASP.NET'),
('Date', 'Tue, 26 Apr 2016 12:32:40 GMT')]
"""
raw_headers = parse.remote_response.raw._original_response.headers._headers
header_cookies_string_list = []
for name, value in raw_headers:
if name.lower() == 'set-cookie':
if my_host_scheme == 'http://':
value = value.replace('Secure;', '')
value = value.replace(';Secure', ';')
value = value.replace('; Secure', ';')
if 'httponly' in value.lower():
if enable_aggressive_cookies_path_rewrite:
# ??cookie path??, ???path???? /
value = regex_cookie_path_rewriter.sub('path=/;', value)
elif enable_aggressive_cookies_path_rewrite is not None:
# ??HttpOnly Cookies?path???url?
# eg(/extdomains/a.foobar.com): path=/verify; -> path=/extdomains/a.foobar.com/verify
if parse.remote_domain not in domain_alias_to_target_set: # do not rewrite main domains
value = regex_cookie_path_rewriter.sub(
'\g<prefix>=/extdomains/' + parse.remote_domain + '\g<path>', value)
header_cookies_string_list.append(value)
return header_cookies_string_list
def assemble_parse():
"""??????URL???????????URL"""
_temp = decode_mirror_url()
parse.remote_domain = _temp['domain'] # type: str
parse.is_https = _temp['is_https'] # type: bool
parse.remote_path = _temp['path'] # type: str
parse.remote_path_query = _temp['path_query'] # type: str
parse.is_external_domain = is_external_domain(parse.remote_domain)
parse.remote_url = assemble_remote_url() # type: str
parse.url_no_scheme = parse.remote_url[parse.remote_url.find('//') + 2:] # type: str
recent_domains[parse.remote_domain] = True # ?????????
dbgprint('after assemble_parse, url:', parse.remote_url, ' path_query:', parse.remote_path_query)
def delete(short_name):
(project, owner, n_tasks,
n_task_runs, overall_progress, last_activity,
n_results) = project_by_shortname(short_name)
title = project_title(project, "Delete")
ensure_authorized_to('read', project)
ensure_authorized_to('delete', project)
pro = pro_features()
project_sanitized, owner_sanitized = sanitize_project_owner(project, owner, current_user)
if request.method == 'GET':
response = dict(template='/projects/delete.html',
title=title,
project=project_sanitized,
owner=owner_sanitized,
n_tasks=n_tasks,
overall_progress=overall_progress,
last_activity=last_activity,
pro_features=pro,
csrf=generate_csrf())
return handle_content_type(response)
######### this block was edited by shruthi
if("directory_names" in project.info.keys()):
for i in project.info["directory_names"]:
if os.path.exists(i):
shutil.rmtree(i)#deleting the actual folder
project_repo.delete(project)
########## end block
auditlogger.add_log_entry(project, None, current_user)
flash(gettext('Project deleted!'), 'success')
return redirect_content_type(url_for('account.profile', name=current_user.name))
def _check_if_redirect_to_password(project):
cookie_exp = current_app.config.get('PASSWD_COOKIE_TIMEOUT')
passwd_mngr = ProjectPasswdManager(CookieHandler(request, signer, cookie_exp))
if passwd_mngr.password_needed(project, get_user_id_or_ip()):
return redirect(url_for('.password_required',
short_name=project.short_name, next=request.path))
def ratelimit(limit, per, send_x_headers=True,
scope_func=lambda: request.remote_addr,
key_func=lambda: request.endpoint,
path=lambda: request.path):
"""
Decorator for limiting the access to a route.
Returns the function if within the limit, otherwise TooManyRequests error
"""
def decorator(f):
@wraps(f)
def rate_limited(*args, **kwargs):
try:
key = 'rate-limit/%s/%s/' % (key_func(), scope_func())
rlimit = RateLimit(key, limit, per, send_x_headers)
g._view_rate_limit = rlimit
#if over_limit is not None and rlimit.over_limit:
if rlimit.over_limit:
raise TooManyRequests
return f(*args, **kwargs)
except Exception as e:
return error.format_exception(e, target=path(),
action=f.__name__)
return update_wrapper(rate_limited, f)
return decorator