Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def convert_to_lines(raw_stacktrace, crash_state_lines, crash_type):
"""Convert an array of string to an array of Line."""
if not raw_stacktrace or not raw_stacktrace.strip():
return []
raw_lines = raw_stacktrace.splitlines()
frames = get_stack_frames(crash_state_lines)
escaped_frames = [jinja2.escape(f) for f in frames]
combined_frames = frames + escaped_frames
# Certain crash types have their own customized frames that are not related to
# the stacktrace. Therefore, we make our best effort to preview stacktrace
# in a reasonable way; we preview around the the top of the stacktrace.
for unique_type in data_types.CRASH_TYPES_WITH_UNIQUE_STATE:
if crash_type.startswith(unique_type):
combined_frames = ['ERROR']
break
lines = []
for index, content in enumerate(raw_lines):
important = _is_line_important(content, combined_frames)
lines.append(Line(index + 1, content, important))
return lines
def nl2br(self, eval_ctx, value):
_paragraph_re = re.compile(r'(?:\r\n|\r|\n){2,}')
result = u'\n\n'.join(u'<p>%s</p>' % p.replace('\n', '<br>\n') for p in _paragraph_re.split(escape(value)))
if eval_ctx.autoescape:
result = Markup(result)
return result
try:
if request.method == 'POST':
data = request.json
if data:
with create_session() as session:
var = models.Variable(key=form, val=json.dumps(data))
session.add(var)
session.commit()
return ""
else:
return self.render(
'airflow/variables/{}.html'.format(form)
)
except:
# prevent XSS
form = escape(form)
return ("Error: form airflow/variables/{}.html "
"not found.").format(form), 404
output = [u'<option class="', jinja2.escape(' '.join(status)), u'" label="" for="" in="" selected="" value="', jinja2.escape(files[0].id), u'">', jinja2.escape(label), u'</option>\n'))
return output
# Has protocol?
middle = fragment[middle_start:middle_end]
if middle.startswith(('http://', 'https://', 'www.')):
unclosed_parens = middle.count('(') - middle.count(')')
# Special case for parentheses (Wikipedia), but not brackets (Slack bridge)
if end and len(end) >= unclosed_parens > 0 and end[:unclosed_parens] == ')' * unclosed_parens:
middle += end[:unclosed_parens]
end = end[unclosed_parens:]
if middle.startswith('www.'):
href = "http://" + middle
else:
href = middle
splitted[i] = "{0}<a href="\'{1}\'">{2}</a>{3}".format(escape(begin), href, escape(middle), escape(end))
else:
splitted[i] = escape(fragment)
return ' '.join(splitted)
if isinstance(value, (list, dict)):
value = dumps_json(value)
elif isinstance(value, bytes):
try:
encoding = detect_encoding(value)
value = value.decode(encoding)
if key == "body":
try:
# request body is in json format
value = json.loads(value)
value = dumps_json(value)
except json.JSONDecodeError:
pass
value = escape(value)
except UnicodeDecodeError:
pass
elif not isinstance(value, (str, bytes, int, float, Iterable)):
# class instance, e.g. MultipartEncoder()
value = repr(value)
elif isinstance(value, RequestsCookieJar):
value = value.get_dict()
request_data[key] = value
markup = u'<a href="{url}">{text}</a>'
q = u' '.join(newquery)
text = u' '.join(newwords)
query_dict = context['request'].GET.copy()
query_dict['q'] = q
if 'page' in query_dict:
query_dict['page'] = 1
items = [(k, v) for k in query_dict for v in query_dict.getlist(k) if v]
query_string = urlencode(items)
url = u'%s?%s' % (reverse('search'), query_string)
return jinja2.Markup(markup.format(url=jinja2.escape(url),
text=text))
def fe(s, *args, **kwargs):
"""Format a safe string with potentially unsafe arguments, then return a
safe string."""
s = six.text_type(s)
args = [jinja2.escape(smart_text(v)) for v in args]
for k in kwargs:
kwargs[k] = jinja2.escape(smart_text(kwargs[k]))
return jinja2.Markup(s.format(*args, **kwargs))
def get_cards(self, job):
doc = OrderedDict(sorted(job.document.items(), key=lambda t: t[0]))
for key in doc:
if key.startswith('_'):
# Don't allow users to edit "private" keys that begin with _
del doc[key]
else:
doc[key] = escape(repr(doc[key]))
return [{'name': self.name, 'content': render_template(
self.template, document=doc, jobid=job._id)}]
def get_field(self, *args, **kw):
# obj is the value getting interpolated into the string.
obj, used_key = super(SafeFormatter, self).get_field(*args, **kw)
return jinja2.escape(obj), used_key