#!/usr/bin/env python
# Todo:
#
# - How does script_raw check the exitcode? Document this.
# - Default values for input fields.
# - If there are errors in the form, its values are empties.
# - Send responses using self.send_ if possible
# - Complain about no registered callback on startup instead of serving.
import sys
import optparse
import os
import stat
import json
import BaseHTTPServer
from BaseHTTPServer import BaseHTTPRequestHandler
from SocketServer import ThreadingMixIn
import cgi
import re
import datetime
import subprocess
import base64
import tempfile
import hashlib
import urlparse
html_header = '''
{footer}
'''
class ValidationError(BaseException):
pass
class ScriptFormError(BaseException):
pass
class ScriptForm:
"""
'Main' class that orchestrates parsing the Form configurations, hooking up
callbacks and running the webserver.
"""
def __init__(self, config_file, callbacks=None):
self.config_file = config_file
if callbacks:
self.callbacks = callbacks
else:
self.callbacks = {}
self.basepath = os.path.realpath(os.path.dirname(config_file))
def get_form_config(self):
"""
Read and return the form configuration in the form of a FormConfig
instance. If it has already been read, a cached version is returned.
"""
# Cache
if hasattr(self, 'form_config_singleton'):
return self.form_config_singleton
path = self.config_file
config = json.load(file(path, 'r'))
title = config['title']
forms = []
callbacks = self.callbacks
users = None
if 'users' in config:
users = config['users']
for form_name, form in config['forms'].items():
if 'script' in form:
script = os.path.join(self.basepath, form['script'])
else:
script = None
forms.append(
FormDefinition(form_name,
form['title'],
form['description'],
form['fields'],
script,
output=form.get('output', 'escaped'),
submit_title=form.get('submit_title', None),
allowed_users=form.get('allowed_users', None))
)
form_config = FormConfig(
config['title'],
forms,
callbacks,
users
)
self.form_config_singleton = form_config
return form_config
def run(self, listen_addr='0.0.0.0', listen_port=80):
"""
Start the webserver on address `listen_addr` and port `listen_port`.
This call is blocking and will never return unless the user hits
Ctrl-c.
"""
ScriptFormWebApp.scriptform = self
WebSrv(ScriptFormWebApp, listen_addr=listen_addr, listen_port=listen_port)
class FormConfig:
"""
FormConfig is the in-memory representation of a form configuration JSON
file. It holds information (title, users, the form definitions) on the
form configuration being served by this instance of ScriptForm.
"""
def __init__(self, title, forms, callbacks={}, users={}):
self.title = title
self.users = users
self.forms = forms
self.callbacks = callbacks
# Validate scripts
for form_def in self.forms:
if form_def.script:
if not stat.S_IXUSR & os.stat(form_def.script)[stat.ST_MODE]:
raise ScriptFormError("{0} is not executable".format(form_def.script))
else:
if not form_def.name in self.callbacks:
raise ScriptFormError("No script or callback registered for '{0}'".format(form_def.name))
def get_form_def(self, form_name):
"""
Return the form definition for the form with name `form_name`. Returns
an instance of FormDefinition class or raises ValueError if the form
was not found.
"""
for form_def in self.forms:
if form_def.name == form_name:
return form_def
else:
raise ValueError("No such form: {0}".format(form_name))
def callback(self, form_name, form_values, request):
"""
Perform a callback for the form `form_name`. This either calls a script
or a Python callable, depending on how the callback is registered.
`form_values` is a dictionary of validated values as returned by
FormDefinition.validate(). `request` is the request handler context
(ScriptFormWebApp). Scripts and Python callables can use it to send
their responses.
"""
form = self.get_form_def(form_name)
if form.script:
return self.callback_script(form, form_values, request.wfile)
else:
return self.callback_python(form, form_values, request)
def callback_script(self, form, form_values, output_fh=None):
# Pass form values to the script through the environment as strings.
env = os.environ.copy()
for k, v in form_values.items():
env[k] = str(v)
if form.output == 'raw':
p = subprocess.Popen(form.script, shell=True, stdout=output_fh,
stderr=output_fh, env=env)
stdout, stderr = p.communicate(input)
return None
else:
p = subprocess.Popen(form.script, shell=True, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
env=env)
stdout, stderr = p.communicate()
return {
'stdout': stdout,
'stderr': stderr,
'exitcode': p.returncode
}
def callback_python(self, form, form_values, request):
callback = self.callbacks[form.name]
try:
result = callback(form_values, request)
if result:
return {
'stdout': result,
'stderr': '',
'exitcode': 0
}
else:
# Raw output
pass
except Exception,e :
return {
'stdout': '',
'stderr': str(e),
'exitcode': 1
}
class FormDefinition:
"""
FormDefinition holds information about a single form and provides methods
for validation of the form values.
"""
def __init__(self, name, title, description, fields, script=None,
output='escaped', submit_title="Submit",
allowed_users=None):
self.name = name
self.title = title
self.description = description
self.fields = fields
self.script = script
self.output = output
self.submit_title = submit_title
self.allowed_users = allowed_users
def _get_field(self, field_name):
for field in self.fields:
if field['name'] == field_name:
return field
def validate(self, form_values):
"""
Validate all relevant fields for this form against form_values.
"""
errors = {}
values = form_values.copy()
# First make sure all required fields are there
for field in self.fields:
if 'required' in field and \
field['required'] is True and \
field['name'] not in form_values:
errors.setdefault(field['name'], []).append(
"This field is required"
)
# Validate the field values, possible casting them to the correct type.
for field in self.fields:
field_name = field['name']
if field_name == 'form_name':
continue
try:
v = self._field_validate(field_name, form_values)
if v is not None:
values[field_name] = v
except ValidationError, e:
errors.setdefault(field_name, []).append(str(e))
return (errors, values)
def _field_validate(self, field_name, form_values):
"""
Validate a field in this form. This does a dynamic call to a method on
this class in the form 'validate_'.
"""
# Find field definition by iterating through all the fields.
field_def = self._get_field(field_name)
if not field_def:
raise ValidationError("Unknown field: {0}".format(field_name))
field_type = field_def['type']
validate_cb = getattr(self, 'validate_{0}'.format(field_type), None)
return validate_cb(field_def, form_values)
def validate_string(self, field_def, form_values):
value = form_values[field_def['name']]
maxlen = field_def.get('maxlen', None)
minlen = field_def.get('minlen', None)
if minlen is not None and len(value) < minlen:
raise ValidationError("Minimum length is {0}".format(minlen))
if maxlen is not None and len(value) > maxlen:
raise ValidationError("Maximum length is {0}".format(maxlen))
return value
def validate_integer(self, field_def, form_values):
value = form_values[field_def['name']]
max = field_def.get('max', None)
min = field_def.get('min', None)
try:
value = int(value)
except ValueError:
raise ValidationError("Must be an integer number")
if min is not None and value < min:
raise ValidationError("Minimum value is {0}".format(min))
if max is not None and value > max:
raise ValidationError("Maximum value is {0}".format(max))
return int(value)
def validate_float(self, field_def, form_values):
value = form_values[field_def['name']]
max = field_def.get('max', None)
min = field_def.get('min', None)
try:
value = float(value)
except ValueError:
raise ValidationError("Must be an real (float) number")
if min is not None and value < min:
raise ValidationError("Minimum value is {0}".format(min))
if max is not None and value > max:
raise ValidationError("Maximum value is {0}".format(max))
return float(value)
def validate_date(self, field_def, form_values):
value = form_values[field_def['name']]
max = field_def.get('max', None)
min = field_def.get('min', None)
try:
value = datetime.datetime.strptime(value, '%Y-%m-%d').date()
except ValueError:
raise ValidationError("Invalid date, must be in form YYYY-MM-DD")
if min is not None:
if value < datetime.datetime.strptime(min, '%Y-%m-%d').date():
raise ValidationError("Minimum value is {0}".format(min))
if max is not None:
if value > datetime.datetime.strptime(max, '%Y-%m-%d').date():
raise ValidationError("maximum value is {0}".format(max))
return value
def validate_radio(self, field_def, form_values):
value = form_values[field_def['name']]
if not value in [o[0] for o in field_def['options']]:
raise ValidationError(
"Invalid value for radio button: {0}".format(value))
return value
def validate_select(self, field_def, form_values):
value = form_values[field_def['name']]
if not value in [o[0] for o in field_def['options']]:
raise ValidationError(
"Invalid value for dropdown: {0}".format(value))
return value
def validate_text(self, field_def, form_values):
value = form_values[field_def['name']]
minlen = field_def.get('minlen', None)
maxlen = field_def.get('maxlen', None)
if minlen is not None:
if len(value) < minlen:
raise ValidationError("minimum length is {0}".format(minlen))
if maxlen is not None:
if len(value) > maxlen:
raise ValidationError("maximum length is {0}".format(maxlen))
return value
def validate_password(self, field_def, form_values):
value = form_values[field_def['name']]
minlen = field_def.get('minlen', None)
if minlen is not None:
if len(value) < minlen:
raise ValidationError("minimum length is {0}".format(minlen))
return value
def validate_file(self, field_def, form_values):
value = form_values[field_def['name']]
field_name = field_def['name']
upload_fname = form_values['{0}__name'.format(field_name)]
upload_fname_ext = os.path.splitext(upload_fname)[-1].lstrip('.')
extensions = field_def.get('extensions', None)
if extensions is not None and upload_fname_ext not in extensions:
raise ValidationError("Only file types allowed: {0}".format(','.join(extensions)))
return value
class ThreadedHTTPServer(ThreadingMixIn, BaseHTTPServer.HTTPServer):
pass
class WebSrv:
"""
Very basic web server.
"""
def __init__(self, request_handler, listen_addr='', listen_port=80):
httpd = ThreadedHTTPServer((listen_addr, listen_port), request_handler)
httpd.serve_forever()
class WebAppHandler(BaseHTTPRequestHandler):
"""
Basic web server request handler. Handles GET and POST requests. This class
should be extended with methods (starting with 'h_') to handle the actual
requests. If no path is set, it dispatches to the 'index' or 'default'
method.
"""
def do_GET(self):
self._call(*self._parse(self.path))
def do_POST(self):
form_values = cgi.FieldStorage(
fp=self.rfile,
headers=self.headers,
environ={'REQUEST_METHOD': 'POST'})
self._call(self.path.strip('/'), params={'form_values': form_values})
def _parse(self, reqinfo):
url_comp = urlparse.urlsplit(reqinfo)
path = url_comp.path
qs = urlparse.parse_qs(url_comp.query)
# Only return the first value of each query var. E.g. for
# "?foo=1&foo=2" return '1'.
vars = dict( [(k, v[0]) for k, v in qs.items()] )
return (path.strip('/'), vars)
def _call(self, path, params):
"""
Find a method to call on self.app_class based on `path` and call it.
The method that's called is in the form 'h_'. If no path was
given, it will try to call the 'index' method. If no method could be
found but a `default` method exists, it is called. Otherwise 404 is
sent.
Methods should take care of sending proper headers and content
themselves using self.send_response(), self.send_header(),
self.end_header() and by writing to self.wfile.
"""
method_name = 'h_{0}'.format(path)
method_cb = None
try:
if hasattr(self, method_name) and \
callable(getattr(self, method_name)):
method_cb = getattr(self, method_name)
elif path == '' and hasattr(self, 'index'):
method_cb = self.index
elif hasattr(self, 'default'):
method_cb = self.default
else:
self.send_error(404, "Not found")
return
method_cb(**params)
except Exception, e:
self.send_error(500, "Internal server error")
raise
class ScriptFormWebApp(WebAppHandler):
"""
This class is a request handler for WebSrv.
"""
def index(self):
"""
Index handler. If there's only one form defined, render that form.
Otherwise render a list of available forms.
"""
form_config = self.scriptform.get_form_config()
if len(form_config.forms) == 1:
first_form = form_config.forms[0]
return self.h_form(first_form.name)
else:
return self.h_list()
def auth(self):
"""
Verify that the user is authenticated. This is required if the form
definition contains a 'users' field. Returns True if the user is
validated. Otherwise, returns False and sends 401 HTTP back to the
client.
"""
form_config = self.scriptform.get_form_config()
self.username = None
# If a 'users' element was present in the form configuration file, the
# user must be authenticated.
if form_config.users:
authorized = False
auth_header = self.headers.getheader("Authorization")
if auth_header is not None:
auth_realm, auth_unpw = auth_header.split(' ', 1)
username, password = base64.decodestring(auth_unpw).split(":")
pw_hash = hashlib.sha256(password).hexdigest()
# Validate the username and password
if username in form_config.users and \
pw_hash == form_config.users[username]:
self.username = username
authorized = True
if not authorized:
# User is not authenticated. Send authentication request.
self.send_response(401)
self.send_header("WWW-Authenticate", 'Basic realm="Private Area"')
self.end_headers()
return False
return True
def h_list(self):
"""
Render a list of available forms.
"""
form_config = self.scriptform.get_form_config()
if not self.auth():
return
h_form_list = []
for form_def in form_config.forms:
if form_def.allowed_users is not None and \
self.username not in form_def.allowed_users:
continue # User is not allowed to run this form
h_form_list.append('''
'''.format(
title=field['title'],
input=input,
errors=', '.join(errors)
)
)
# Make sure the user is allowed to access this form.
form_def = form_config.get_form_def(form_name)
if form_def.allowed_users is not None and \
self.username not in form_def.allowed_users:
self.send_error(401, "You're not authorized to view this form")
return
html_errors = ''
if errors:
html_errors = '
'
for error in errors:
html_errors += '
{0}
'.format(error)
html_errors += '
'
output = html_form.format(
header=html_header.format(title=form_config.title),
footer=html_footer,
title=form_def.title,
description=form_def.description,
errors=html_errors,
name=form_def.name,
fields=''.join([render_field(f, errors.get(f['name'], [])) for f in form_def.fields]),
submit_title=form_def.submit_title
)
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write(output)
def h_submit(self, form_values):
"""
Handle the submitting of a form by validating the values and then doing
a callback to either a script or a Python function. How the output is
handled depends on settings in the form definition.
"""
form_config = self.scriptform.get_form_config()
if not self.auth():
return
form_name = form_values.getfirst('form_name', None)
form_def = form_config.get_form_def(form_name)
if form_def.allowed_users is not None and \
self.username not in form_def.allowed_users:
self.send_error(401, "You're not authorized to view this form")
return
# Convert FieldStorage to a simple dict, because we're not allowd to
# add items to it. For normal fields, the form field name becomes the
# key and the value becomes the field value. For file upload fields, we
# stream the uploaded file to a temp file and then put the temp file in
# the destination dict. We also add an extra field with the originally
# uploaded file's name.
values = {}
tmp_files = []
for field_name in form_values:
field = form_values[field_name]
if field.filename:
# Field is an uploaded file. Stream it to a temp file
tmpfile = tempfile.mktemp(prefix="scriptform_")
f = file(tmpfile, 'w')
while True:
buf = field.file.read(1024 * 16)
if not buf:
break
f.write(buf)
f.close()
field.file.close()
tmp_files.append(tmpfile) # For later cleanup
values[field_name] = tmpfile
values['{0}__name'.format(field_name)] = field.filename
else:
# Field is a normal form field. Store its value.
values[field_name] = form_values.getfirst(field_name, None)
# Validate the form values
form_errors, form_values = form_def.validate(values)
if not form_errors:
# Call user's callback. If a result is returned, we wrap its output
# in some nice HTML. If no result is returned, the output was raw
# and the callback should have written its own response to the
# self.wfile filehandle.
result = form_config.callback(form_name, form_values, self)
if result:
if result['exitcode'] != 0:
msg = '{0}'.format(cgi.escape(result['stderr']))
else:
if form_def.output == 'escaped':
msg = '
{0}
'.format(cgi.escape(result['stdout']))
else:
msg = result['stdout']
output = html_submit_response.format(
header=html_header.format(title=form_config.title),
footer=html_footer,
title=form_def.title,
form_name=form_def.name,
msg=msg,
)
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write(output)
else:
# Form had errors
self.h_form(form_name, form_errors)
# Clean up uploaded files
for file_name in tmp_files:
if os.path.exists(file_name):
os.unlink(file_name)
def main_generate_pw(parser, options, args):
import getpass
plain_pw = getpass.getpass()
if not plain_pw == getpass.getpass('Repeat password: '):
sys.stderr.write("Passwords do not match.\n")
sys.exit(1)
print hashlib.sha256(plain_pw).hexdigest()
sys.exit(0)
def main_serve(parser, options, args):
if len(args) < 1:
parser.error("Insufficient number of arguments")
sf = ScriptForm(args[0])
sf.run(listen_port=options.port)
if __name__ == "__main__":
usage = [
sys.argv[0] + " [option] ",
" " + sys.argv[0] + " --generate-pw",
]
parser = optparse.OptionParser()
parser.set_usage('\n'.join(usage))
parser.add_option("-g", "--generate-pw", dest="generate_pw", action="store_true", default=False, help="Generate password")
parser.add_option("-p", "--port", dest="port", action="store", type="int", default=80, help="Port to listen on")
(options, args) = parser.parse_args()
if options.generate_pw:
main_generate_pw(parser, options, args)
else:
main_serve(parser, options, args)