# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
"""
Functions to syntax-validate output content
"""
from os import path
import os
import sys
import logging
import tempfile
import subprocess
import json
import six.moves.urllib.parse
import six.moves.urllib.request
import six.moves.urllib.error
import re
import pkg_resources
import six
import webtest
from webtest import TestApp, TestResponse
from ming.utils import LazyProperty
import requests
from allura.lib import utils
log = logging.getLogger(__name__)
class Config:
"Config to encapsulate flexible/complex test enabled/disabled rules."
_instance = None
def __init__(self):
self.ini_config = None
pass
@classmethod
def instance(cls):
if not cls._instance:
cls._instance = cls()
return cls._instance
@LazyProperty
def test_ini(self):
if not self.ini_config:
from . import controller
import six.moves.configparser
conf = six.moves.configparser.ConfigParser(
{'validate_html5': 'false', 'validate_inlinejs': 'false'})
conf.read(controller.get_config_file())
self.ini_config = conf
return self.ini_config
def validation_enabled(self, val_type):
env_var = os.getenv('ALLURA_VALIDATION')
if env_var == 'all':
return True
elif env_var == 'none':
return False
elif env_var is not None:
return val_type in env_var.split(',')
enabled = self.test_ini.getboolean('validation', 'validate_' + val_type)
return enabled
def report_validation_error(val_name, filename, message):
message = f'{val_name} Validation errors ({filename}):\n{message}\n'
raise AssertionError(message)
def dump_to_file(prefix, contents, suffix=''):
f = tempfile.NamedTemporaryFile('w', prefix=prefix, delete=False, suffix=suffix)
f.write(contents)
f.close()
return f.name
def validate_html(html_or_response):
if hasattr(html_or_response, 'text'):
html = html_or_response.text
else:
html = html_or_response
html = html.lstrip()
if html.startswith('<!DOCTYPE html>'):
return validate_html5(html)
else:
assert False, 'Non-valid HTML: ' + html[:100] + '...'
def validate_json(json_or_response):
if hasattr(json_or_response, 'text'):
j = json_or_response.text
else:
j = json_or_response
try:
obj = json.loads(j)
except Exception as e:
raise AssertionError("Couldn't validate JSON: " + str(e) + ':' + j[:100] + '...')
return obj
def validate_html5(html_or_response):
if hasattr(html_or_response, 'text'):
html = html_or_response.text
else:
html = html_or_response
count = 3
while True:
try:
# TODO switch to http://validator.w3.org/nu/?out=text but it has more validation errors for us to fix
# Docs: https://github.com/validator/validator/wiki/Service-%C2%BB-Input-%C2%BB-POST-body and other pages
resp = requests.post('http://html5.validator.nu/nu/?out=text', # could do out=json
data=html,
headers={'Content-Type': 'text/html; charset=utf-8'},
timeout=5)
resp = resp.text
break
except OSError:
resp = "Couldn't connect to validation service to check the HTML"
count -= 1
if count == 0:
sys.stderr.write('WARNING: ' + resp + '\n')
break
resp = resp.replace('“', '"').replace('”', '"').replace('–', '-')
ignored_errors = [
'Required attributes missing on element "object"',
'Stray end tag "embed".',
'Stray end tag "param".',
r'Bad value .+? for attribute "onclick" on element "input": invalid return',
]
for ignore in ignored_errors:
resp = re.sub('Error: ' + ignore, 'Ignoring: ' + ignore, resp)
if 'Error:' in resp:
fname = dump_to_file('html5-', html, suffix='.html')
message = resp.decode('ascii', 'ignore')
report_validation_error('html5', fname, message)
def validate_html5_chunk(html):
""" When you don't have a html & body tags - this adds it"""
# WebTest doesn't like HTML fragments without doctype,
# so we output them sometimes for fragments, which is hack.
# Unhack it here.
doctype = '<!DOCTYPE html>'
if html.startswith(doctype):
html = html[len(doctype):]
html = '''<!DOCTYPE html>
<html>
<head><title>Not empty</title></head>
<body>
%s
</body></html>''' % html
return validate_html5(html)
def validate_js(html_or_response, within_html=False):
if hasattr(html_or_response, 'text'):
if html_or_response.status_int != 200:
return
text = html_or_response.text
else:
text = html_or_response
fname = dump_to_file('eslint-', text, suffix='.html' if within_html else '.js')
eslintrc = os.path.join(pkg_resources.get_distribution('allura').location, '../.eslintrc-es5')
cmd = ['npm', 'run', 'eslint', '--',
'-c', eslintrc, # since we're in a tmp dir
'--no-ignore', # tmp dirs ignored by default
]
if within_html:
cmd += ['--rule', 'indent: 0'] # inline HTML always has indentation wrong
cmd += ['--plugin', 'html']
cmd += [fname]
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
stdout, stderr = p.communicate()
if p.returncode == 0:
os.unlink(fname)
else:
stdout = stdout.decode('utf8')
report_validation_error('js', fname, stdout)
def validate_page(html_or_response):
if Config.instance().validation_enabled('html5'):
validate_html(html_or_response)
if Config.instance().validation_enabled('inlinejs'):
validate_js(html_or_response, within_html=True)
class AntiSpamTestApp(TestApp):
def post(self, *args, **kwargs) -> TestResponse:
antispam = utils.AntiSpam()
if kwargs.pop('antispam', False):
params = {
'timestamp': antispam.timestamp_text,
'spinner': antispam.spinner_text,
antispam.enc('honey0'): '',
antispam.enc('honey1'): '',
}
for k, v in kwargs['params'].items():
params[antispam.enc(k)] = v
params['_session_id'] = kwargs['params'].get('_session_id') # exclude csrf token from encryption
kwargs['params'] = params
return super().post(*args, **kwargs)
def antispam_field_names(self, form):
"""
:param form: a WebTest form (i.e. from a self.app.get response)
:return: a dict of field names -> antispam encoded field names
"""
timestamp = form['timestamp'].value
spinner = form['spinner'].value
antispam = utils.AntiSpam(timestamp=int(timestamp), spinner=utils.AntiSpam._unwrap(spinner))
names = list(form.fields.keys())
name_mapping = {}
for name in names:
try:
decoded = antispam.dec(name)
except Exception:
decoded = name
name_mapping[decoded] = name
return name_mapping
class PostParamCheckingTestApp(AntiSpamTestApp):
def _validate_params(self, params, method):
if not params:
return
# params can be raw data (json data post, for example)
if isinstance(params, (bytes, (str,))):
return
# params can be a list or a dict
if hasattr(params, 'items'):
params = list(params.items())
for k, v in params:
if not isinstance(k, str):
raise TypeError('%s key %s is %s, not str' %
(method, k, type(k)))
self._validate_val(k, v, method)
def _validate_val(self, k, v, method):
if isinstance(v, (list, tuple)):
for vv in v:
self._validate_val(k, vv, method)
elif not isinstance(v, (str, bytes, webtest.forms.File, webtest.forms.Upload)):
raise TypeError(
'%s key %s has value %s of type %s, not str. ' %
(method, k, v, type(v)))
def get(self, *args, **kwargs) -> TestResponse:
params = None
if 'params' in kwargs:
params = kwargs['params']
elif len(args) > 1:
params = args[1]
self._validate_params(params, 'get')
return super().get(*args, **kwargs)
def post(self, *args, **kwargs) -> TestResponse:
params = None
if 'params' in kwargs:
params = kwargs['params']
elif len(args) > 1:
params = args[1]
self._validate_params(params, 'post')
return super().post(*args, **kwargs)
class ValidatingTestApp(PostParamCheckingTestApp):
# Subclasses may set this to True to skip validation altogether
validate_skip = False
def _validate(self, resp, method, val_params):
"""Perform validation on webapp response. This handles responses of
various types and forms."""
if resp.status_int != 200:
return
content_type = resp.headers['Content-Type']
if content_type.startswith('text/html'):
if val_params['validate_chunk']:
if Config.instance().validation_enabled('html5'):
validate_html5_chunk(resp.text)
else:
validate_page(resp)
elif content_type.split(';', 1)[0] in ('text/plain', 'text/x-python', 'application/octet-stream'):
pass
elif content_type.startswith('application/json'):
validate_json(resp.text)
elif content_type.startswith(('application/x-javascript', 'application/javascript', 'text/javascript')):
validate_js(resp.text)
elif content_type.startswith('application/xml'):
import feedparser
d = feedparser.parse(resp.text)
assert d.bozo == 0, 'Non-wellformed feed'
elif content_type.startswith(('image/', 'application/x-www-form-urlencoded')):
pass
else:
assert False, 'Unexpected output content type: ' + content_type
def _get_validation_params(self, kw):
"Separate validation params from normal TestApp methods params."
params = {}
for k in ('validate_skip', 'validate_chunk'):
params[k] = kw.pop(k, False)
return params, kw
def get(self, *args, **kw) -> TestResponse:
val_params, kw = self._get_validation_params(kw)
resp = super().get(*args, **kw)
if not self.validate_skip and not val_params['validate_skip']:
self._validate(resp, 'get', val_params)
return resp
def post(self, *args, **kw) -> TestResponse:
val_params, kw = self._get_validation_params(kw)
resp = super().post(*args, **kw)
if not self.validate_skip and not val_params['validate_skip']:
self._validate(resp, 'post', val_params)
return resp
def delete(self, *args, **kw) -> TestResponse:
val_params, kw = self._get_validation_params(kw)
resp = super().delete(*args, **kw)
if not self.validate_skip and not val_params['validate_skip']:
self._validate(resp, 'delete', val_params)
return resp
def do_request(self, *args, **kwargs) -> TestResponse:
# middleware should do this already, but be sure that no global c/config/request etc remains between tests
resp = super().do_request(*args, **kwargs)
tgGlobalsRegistry = resp.request.environ['paste.registry']
try:
tgGlobalsRegistry.cleanup()
except IndexError:
# already cleaned up
pass
except Exception:
log.warning('Error cleaning up TG Registry', exc_info=True)
return resp