# Copyright (C) 2000-2014 Bastian Kleineidam
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""
Functions used by the WSGI script.
"""
import html
import threading
import locale
import re
import time
import urllib.parse
from . import (
configuration,
strformat,
checker,
director,
get_link_pat,
init_i18n,
url as urlutil,
)
from .decorators import synchronized
# 5 minutes timeout for requests
MAX_REQUEST_SECONDS = 300
# character set encoding for HTML output
HTML_ENCODING = 'utf-8'
[docs]
def application(environ, start_response):
"""WSGI interface: start an URL check."""
# the environment variable CONTENT_LENGTH may be empty or missing
try:
request_body_size = int(environ.get('CONTENT_LENGTH', 0))
except ValueError:
request_body_size = 0
# When the method is POST the query string will be sent
# in the HTTP request body which is passed by the WSGI server
# in the file like wsgi.input environment variable.
if request_body_size > 0:
request_body = environ['wsgi.input'].read(request_body_size)
else:
request_body = environ['wsgi.input'].read()
form = urllib.parse.parse_qs(request_body.decode(HTML_ENCODING))
status = '200 OK'
start_response(status, get_response_headers())
yield from checklink(form=form, env=environ)
_supported_langs = ('de', 'C')
# map language -> locale name
lang_locale = {
'de': 'de_DE',
'C': 'C',
'en': 'en_EN',
}
_is_level = re.compile(r'^(0|1|2|3|-1)$').match
_lock = threading.Lock()
[docs]
class ThreadsafeIO:
"""Thread-safe unicode I/O class."""
def __init__(self):
"""Initialize buffer."""
self.buf = []
self.closed = False
[docs]
@synchronized(_lock)
def write(self, data):
"""Write given unicode data to buffer."""
assert isinstance(data, str)
if self.closed:
raise OSError("Write on closed I/O object")
if data:
self.buf.append(data)
[docs]
@synchronized(_lock)
def get_data(self):
"""Get bufferd unicode data."""
data = "".join(self.buf)
self.buf = []
return data
[docs]
@synchronized(_lock)
def close(self):
"""Reset buffer and close this I/O object."""
self.buf = []
self.closed = True
[docs]
def encode(s):
"""Encode given string in HTML encoding."""
return s.encode(HTML_ENCODING, 'ignore')
[docs]
def checklink(form, env):
"""Validates the CGI form and checks the given links."""
if form is None:
form = {}
try:
checkform(form, env)
except LCFormError as err:
log(env, err)
yield encode(format_error(str(err)))
return
out = ThreadsafeIO()
config = get_configuration(form, out)
url = strformat.stripurl(formvalue(form, "url"))
aggregate = director.get_aggregate(config)
url_data = checker.get_url_from(url, 0, aggregate, extern=(0, 0))
aggregate.urlqueue.put(url_data)
for html_str in start_check(aggregate, out):
yield encode(html_str)
out.close()
[docs]
def start_check(aggregate, out):
"""Start checking in background and write encoded output to out."""
# check in background
t = threading.Thread(target=director.check_urls, args=(aggregate,))
t.start()
# time to wait for new data
sleep_seconds = 2
# current running time
run_seconds = 0
while not aggregate.is_finished():
yield out.get_data()
time.sleep(sleep_seconds)
run_seconds += sleep_seconds
if run_seconds > MAX_REQUEST_SECONDS:
director.abort(aggregate)
break
yield out.get_data()
[docs]
def get_configuration(form, out):
"""Initialize a CGI configuration."""
config = configuration.Configuration()
config["recursionlevel"] = int(formvalue(form, "level"))
config["logger"] = config.logger_new('html', fd=out, encoding=HTML_ENCODING)
config["status"] = False
config["threads"] = 2
if "anchors" in form:
config["enabledplugins"].append("AnchorCheck")
if "errors" not in form:
config["verbose"] = True
# avoid checking of local files or other nasty stuff
pat = "!^%s$" % urlutil.safe_url_pattern
config["externlinks"].append(get_link_pat(pat, strict=True))
config.sanitize()
return config
[docs]
def get_host_name(form):
"""Return host name of given URL."""
return urllib.parse.urlparse(formvalue(form, "url"))[1]
[docs]
def log(env, msg):
"""Log message to WSGI error output."""
logfile = env['wsgi.errors']
logfile.write("%s\n" % msg)
[docs]
def dump(env, form):
"""Log environment and form."""
for var, value in env.items():
log(env, var + "=" + value)
for key in form:
log(env, str(formvalue(form, key)))