api/plugins/openapi.py - incubator-warble-server - Git at Google

 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 # 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.

 """
 This is the OpenAPI validator library.
 Validates input using the OpenAPI specification version 3 from
 https://github.com/OAI/OpenAPI-Specification (a simplified version, ahem)
 """

 import yaml
 import json
 import functools
 import operator
 import re

 class OpenAPIException(Exception):
     def __init__(self, message):
         self.message = message

 # Python type names to JSON type names
 py2JSON = {
     'int':      'integer',
     'float':    'float',
     'str':      'string',
     'list':     'array',
     'dict':     'object',
     'bool':     'boolean'
 }

 mcolors = {
     'PUT':      '#fca130',
     'DELETE':   '#f93e3e',
     'GET':      '#61affe',
     'POST':     '#49cc5c',
     'PATCH':    '#d5a37e'
 }

 class OpenAPI():
     def __init__(self, APIFile):
         """ Instantiates an OpenAPI validator given a YAML specification"""
         if APIFile.endswith(".json") or APIFile.endswith(".js"):
             self.API = json.load(open(APIFile))
         else:
             self.API = yaml.load(open(APIFile))

     def validateType(self, field, value, ftype):
         """ Validate a single field value against an expected type """

         # Get type of value, convert to JSON name of type.
         pyType = type(value).__name__
         jsonType = py2JSON[pyType] if pyType in py2JSON else pyType

         # Check if type matches
         if ftype != jsonType:
             raise OpenAPIException("OpenAPI mismatch: Field '%s' was expected to be %s, but was really %s!" % (field, ftype, jsonType))

     def validateSchema(self, pdef, formdata, schema = None):
         """ Validate (sub)parameters against OpenAPI specs """

         # allOf: list of schemas to validate against
         if 'allOf' in pdef:
             for subdef in pdef['allOf']:
                 self.validateSchema(subdef, formdata)

         where = "JSON body"
         # Symbolic link??
         if 'schema' in pdef:
             schema = pdef['schema']['$ref']
         if '$ref' in pdef:
             schema = pdef['$ref']
         if schema:
             # #/foo/bar/baz --> dict['foo']['bar']['baz']
             pdef = functools.reduce(operator.getitem, schema.split('/')[1:], self.API)
             where = "item matching schema %s" % schema

         # Check that all required fields are present

         if 'required' in pdef:
             for field in pdef['required']:
                 if not field in formdata:
                     raise OpenAPIException("OpenAPI mismatch: Missing input field '%s' in %s!" % (field, where))

         # Now check for valid format of input data
         for field in formdata:
             if 'properties' not in pdef or field not in pdef['properties'] :
                 raise OpenAPIException("Unknown input field '%s' in %s!" % (field, where))
             if 'type' not in pdef['properties'][field]:
                 raise OpenAPIException("OpenAPI mismatch: Field '%s' was found in api.yaml, but no format was specified in specs!" % field)
             ftype = pdef['properties'][field]['type']
             self.validateType(field, formdata[field], ftype)

             # Validate sub-arrays
             if ftype == 'array' and 'items' in pdef['properties'][field]:
                 for item in formdata[field]:
                     if '$ref' in pdef['properties'][field]['items']:
                         self.validateSchema(pdef['properties'][field]['items'], item)
                     else:
                         self.validateType(field, formdata[field], pdef['properties'][field]['items']['type'])

             # Validate sub-hashes
             if ftype == 'hash' and 'schema' in pdef['properties'][field]:
                 self.validateSchema(pdef['properties'][field], formdata[field])
     def validateParameters(self, defs, formdata):
         #
         pass

     def validate(self, method = "GET", path = "/foo", formdata = None):
         """ Validate the request method and input data against the OpenAPI specification """

         # Make sure we're not dealing with a dynamic URL.
         # If we find /foo/{key}, we fold that into the form data
         # and process as if it's a json input field for now.
         if not self.API['paths'].get(path):
             for xpath in self.API['paths']:
                 pathRE = re.sub(r"\{(.+?)\}", r"(?P<\1>[^/]+)", xpath)
                 m = re.match(pathRE, path)
                 if m:
                     for k,v  in m.groupdict().items():
                         formdata[k] = v
                     path = xpath
                     break

         if self.API['paths'].get(path):
             defs = self.API['paths'].get(path)
             method = method.lower()
             if method in defs:
                 mdefs = defs[method]
                 if formdata and 'parameters' in mdefs:
                     self.validateParameters(mdefs['parameters'], formdata)
                 elif formdata and 'requestBody' not in mdefs:
                     raise OpenAPIException("OpenAPI mismatch: JSON data is now allowed for this request type")
                 elif 'requestBody' in mdefs and 'content' in mdefs['requestBody']:
                     # SHORTCUT: We only care about JSON input for Warble! Disregard other types
                     if not 'application/json' in mdefs['requestBody']['content']:
                         raise OpenAPIException ("OpenAPI mismatch: API endpoint accepts input, but no application/json definitions found in api.yaml!")
                     jdefs = mdefs['requestBody']['content']['application/json']

                     # Check that required params are here
                     self.validateSchema(jdefs, formdata)

             else:
                 raise OpenAPIException ("OpenAPI mismatch: Method %s is not registered for this API" % method)
         else:
             raise OpenAPIException("OpenAPI mismatch: Unknown API path '%s'!" % path)

     def dumpExamples(self, pdef, array = False):
         schema = None
         if 'schema' in pdef:
             if 'type' in pdef['schema'] and pdef['schema']['type'] == 'array':
                 array = True
                 schema = pdef['schema']['items']['$ref']
             else:
                 schema = pdef['schema']['$ref']
         if '$ref' in pdef:
             schema = pdef['$ref']
         if schema:
             # #/foo/bar/baz --> dict['foo']['bar']['baz']
             pdef = functools.reduce(operator.getitem, schema.split('/')[1:], self.API)
         js = {}
         desc = {}
         if 'properties' in pdef:
             for k, v in pdef['properties'].items():
                 if 'description' in v:
                     desc[k] = [v['type'], v['description']]
                 if 'example' in v:
                     js[k] = v['example']
                 elif 'items' in v:
                     if v['type'] == 'array':
                         js[k], foo = self.dumpExamples(v['items'], True)
                     else:
                         js[k], foo = self.dumpExamples(v['items'])
         return [js if not array else [js], desc]

     def toHTML(self):
         """ Blurps out the specs in a pretty HTML blob """
         print("""
 <!DOCTYPE html>
 <html lang="en">
 <head>
 </head>
 <body>
 """)
         li = "<h3>Overview:</h3><ul style='font-size: 12px; font-family: Open Sans, sans-serif;'>"
         for path, spec in sorted(self.API['paths'].items()):
             for method, mspec in sorted(spec.items()):
                 method = method.upper()
                 summary = mspec.get('summary', 'No summary available')
                 linkname = "%s%s" % (method.lower(), path.replace('/', '-'))
                 li += "<li><a href='#%s'>%s %s</a>: %s</li>\n" % (linkname, method, path, summary)
         li += "</ul>"
         print(li)
         for path, spec in sorted(self.API['paths'].items()):
             for method, mspec in sorted(spec.items()):
                 method = method.upper()
                 summary = mspec.get('summary', 'No summary available')
                 resp = ""
                 inp = ""
                 inpvars = ""
                 linkname = "%s%s" % (method.lower(), path.replace('/', '-'))
                 if 'responses' in mspec:
                     for code, cresp in sorted(mspec['responses'].items()):
                         for ctype, pdef in cresp['content'].items():
                             xjs, desc = self.dumpExamples(pdef)
                             js = json.dumps(xjs, indent = 4)
                             resp += "<div style='float: left; width: 90%%;'><pre style='width: 600px;'><b>%s</b>:\n%s</pre>\n</div>\n" % (code, js)

                 if 'requestBody' in mspec:
                     for ctype, pdef in mspec['requestBody']['content'].items():
                         xjs, desc = self.dumpExamples(pdef)
                         if desc:
                             for k, v in desc.items():
                                 inpvars += "<kbd><b>%s:</b></kbd> (%s) <span style='font-size: 12px; font-family: Open Sans, sans-serif;'>%s</span><br/>\n" % (k, v[0], v[1])
                         js = json.dumps(xjs, indent = 4)
                         inp += "<div style='float: left; width: 90%%;'><h4>Input examples:</h4><blockquote><pre style='width: 600px;'><b>%s</b>:\n%s</pre></blockquote>\n</div>" % (ctype, js)

                 if inpvars:
                     inpvars = "<div style='float: left; width: 90%%;'><blockquote><pre style='width: 600px;'>%s</pre>\n</blockquote></div>" % inpvars


                 print("""
                       <div id="%s" style="margin: 20px; display: flex; box-sizing: border-box; width: 900px; border-radius: 6px; border: 1px solid %s; font-family: sans-serif; background: %s30;">
                         <div style="min-height: 32px;">
                           <!-- method -->

                           <div style="float: left; align-items: center; margin: 4px; border-radius: 5px; text-align: center; padding-top: 4px; height: 20px; width: 100px; color: #FFF; font-weight: bold; background: %s;">%s</div>

                           <!-- path and summary -->
                           <span style="display: flex; padding-top: 6px;"><kbd><strong>%s</strong></kbd></span>
                           <div style="box-sizing: border-box; flex: 1; font-size: 13px; font-family: Open Sans, sans-serif; float: left; padding-top: 6px; margin-left: 20px;">
                           %s</div>
                           <div style="float: left; width: 90%%;display: %s; ">
                             <h4>JSON parameters:</h4>
                             %s
                             <br/>
                             %s
                           </div>
                           <div style="float: left; width: 90%%; ">
                             <h4>Response examples:</h4>
                             <blockquote>%s</blockquote>
                           </div>
                         </div>
                       </div>
                       """ % (linkname, mcolors[method], mcolors[method], mcolors[method], method, path, summary, "block" if inp else "none", inpvars, inp, resp))
                 #print("%s %s: %s" % (method.upper(), path, mspec['summary']))
         print("</body></html>")
	#!/usr/bin/env python3
	# -- coding: utf-8 --
	# 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.

	"""
	This is the OpenAPI validator library.
	Validates input using the OpenAPI specification version 3 from
	https://github.com/OAI/OpenAPI-Specification (a simplified version, ahem)
	"""

	import yaml
	import json
	import functools
	import operator
	import re

	class OpenAPIException(Exception):
	def __init__(self, message):
	self.message = message

	# Python type names to JSON type names
	py2JSON = {
	'int': 'integer',
	'float': 'float',
	'str': 'string',
	'list': 'array',
	'dict': 'object',
	'bool': 'boolean'
	}

	mcolors = {
	'PUT': '#fca130',
	'DELETE': '#f93e3e',
	'GET': '#61affe',
	'POST': '#49cc5c',
	'PATCH': '#d5a37e'
	}

	class OpenAPI():
	def __init__(self, APIFile):
	""" Instantiates an OpenAPI validator given a YAML specification"""
	if APIFile.endswith(".json") or APIFile.endswith(".js"):
	self.API = json.load(open(APIFile))
	else:
	self.API = yaml.load(open(APIFile))

	def validateType(self, field, value, ftype):
	""" Validate a single field value against an expected type """

	# Get type of value, convert to JSON name of type.
	pyType = type(value).__name__
	jsonType = py2JSON[pyType] if pyType in py2JSON else pyType

	# Check if type matches
	if ftype != jsonType:
	raise OpenAPIException("OpenAPI mismatch: Field '%s' was expected to be %s, but was really %s!" % (field, ftype, jsonType))

	def validateSchema(self, pdef, formdata, schema = None):
	""" Validate (sub)parameters against OpenAPI specs """

	# allOf: list of schemas to validate against
	if 'allOf' in pdef:
	for subdef in pdef['allOf']:
	self.validateSchema(subdef, formdata)

	where = "JSON body"
	# Symbolic link??
	if 'schema' in pdef:
	schema = pdef['schema']['$ref']
	if '$ref' in pdef:
	schema = pdef['$ref']
	if schema:
	# #/foo/bar/baz --> dict['foo']['bar']['baz']
	pdef = functools.reduce(operator.getitem, schema.split('/')[1:], self.API)
	where = "item matching schema %s" % schema

	# Check that all required fields are present

	if 'required' in pdef:
	for field in pdef['required']:
	if not field in formdata:
	raise OpenAPIException("OpenAPI mismatch: Missing input field '%s' in %s!" % (field, where))

	# Now check for valid format of input data
	for field in formdata:
	if 'properties' not in pdef or field not in pdef['properties'] :
	raise OpenAPIException("Unknown input field '%s' in %s!" % (field, where))
	if 'type' not in pdef['properties'][field]:
	raise OpenAPIException("OpenAPI mismatch: Field '%s' was found in api.yaml, but no format was specified in specs!" % field)
	ftype = pdef['properties'][field]['type']
	self.validateType(field, formdata[field], ftype)

	# Validate sub-arrays
	if ftype == 'array' and 'items' in pdef['properties'][field]:
	for item in formdata[field]:
	if '$ref' in pdef['properties'][field]['items']:
	self.validateSchema(pdef['properties'][field]['items'], item)
	else:
	self.validateType(field, formdata[field], pdef['properties'][field]['items']['type'])

	# Validate sub-hashes
	if ftype == 'hash' and 'schema' in pdef['properties'][field]:
	self.validateSchema(pdef['properties'][field], formdata[field])
	def validateParameters(self, defs, formdata):
	#
	pass

	def validate(self, method = "GET", path = "/foo", formdata = None):
	""" Validate the request method and input data against the OpenAPI specification """

	# Make sure we're not dealing with a dynamic URL.
	# If we find /foo/{key}, we fold that into the form data
	# and process as if it's a json input field for now.
	if not self.API['paths'].get(path):
	for xpath in self.API['paths']:
	pathRE = re.sub(r"\{(.+?)\}", r"(?P<\1>[^/]+)", xpath)
	m = re.match(pathRE, path)
	if m:
	for k,v in m.groupdict().items():
	formdata[k] = v
	path = xpath
	break

	if self.API['paths'].get(path):
	defs = self.API['paths'].get(path)
	method = method.lower()
	if method in defs:
	mdefs = defs[method]
	if formdata and 'parameters' in mdefs:
	self.validateParameters(mdefs['parameters'], formdata)
	elif formdata and 'requestBody' not in mdefs:
	raise OpenAPIException("OpenAPI mismatch: JSON data is now allowed for this request type")
	elif 'requestBody' in mdefs and 'content' in mdefs['requestBody']:
	# SHORTCUT: We only care about JSON input for Warble! Disregard other types
	if not 'application/json' in mdefs['requestBody']['content']:
	raise OpenAPIException ("OpenAPI mismatch: API endpoint accepts input, but no application/json definitions found in api.yaml!")
	jdefs = mdefs['requestBody']['content']['application/json']

	# Check that required params are here
	self.validateSchema(jdefs, formdata)

	else:
	raise OpenAPIException ("OpenAPI mismatch: Method %s is not registered for this API" % method)
	else:
	raise OpenAPIException("OpenAPI mismatch: Unknown API path '%s'!" % path)

	def dumpExamples(self, pdef, array = False):
	schema = None
	if 'schema' in pdef:
	if 'type' in pdef['schema'] and pdef['schema']['type'] == 'array':
	array = True
	schema = pdef['schema']['items']['$ref']
	else:
	schema = pdef['schema']['$ref']
	if '$ref' in pdef:
	schema = pdef['$ref']
	if schema:
	# #/foo/bar/baz --> dict['foo']['bar']['baz']
	pdef = functools.reduce(operator.getitem, schema.split('/')[1:], self.API)
	js = {}
	desc = {}
	if 'properties' in pdef:
	for k, v in pdef['properties'].items():
	if 'description' in v:
	desc[k] = [v['type'], v['description']]
	if 'example' in v:
	js[k] = v['example']
	elif 'items' in v:
	if v['type'] == 'array':
	js[k], foo = self.dumpExamples(v['items'], True)
	else:
	js[k], foo = self.dumpExamples(v['items'])
	return [js if not array else [js], desc]

	def toHTML(self):
	""" Blurps out the specs in a pretty HTML blob """
	print("""
	<!DOCTYPE html>
	<html lang="en">
	<head>
	</head>
	<body>
	""")
	li = "<h3>Overview:</h3><ul style='font-size: 12px; font-family: Open Sans, sans-serif;'>"
	for path, spec in sorted(self.API['paths'].items()):
	for method, mspec in sorted(spec.items()):
	method = method.upper()
	summary = mspec.get('summary', 'No summary available')
	linkname = "%s%s" % (method.lower(), path.replace('/', '-'))
	li += "<li><a href='#%s'>%s %s</a>: %s</li>\n" % (linkname, method, path, summary)
	li += "</ul>"
	print(li)
	for path, spec in sorted(self.API['paths'].items()):
	for method, mspec in sorted(spec.items()):
	method = method.upper()
	summary = mspec.get('summary', 'No summary available')
	resp = ""
	inp = ""
	inpvars = ""
	linkname = "%s%s" % (method.lower(), path.replace('/', '-'))
	if 'responses' in mspec:
	for code, cresp in sorted(mspec['responses'].items()):
	for ctype, pdef in cresp['content'].items():
	xjs, desc = self.dumpExamples(pdef)
	js = json.dumps(xjs, indent = 4)
	resp += "<div style='float: left; width: 90%%;'><pre style='width: 600px;'><b>%s</b>:\n%s</pre>\n</div>\n" % (code, js)

	if 'requestBody' in mspec:
	for ctype, pdef in mspec['requestBody']['content'].items():
	xjs, desc = self.dumpExamples(pdef)
	if desc:
	for k, v in desc.items():
	inpvars += "<kbd><b>%s:</b></kbd> (%s) <span style='font-size: 12px; font-family: Open Sans, sans-serif;'>%s</span><br/>\n" % (k, v[0], v[1])
	js = json.dumps(xjs, indent = 4)
	inp += "<div style='float: left; width: 90%%;'><h4>Input examples:</h4><blockquote><pre style='width: 600px;'><b>%s</b>:\n%s</pre></blockquote>\n</div>" % (ctype, js)

	if inpvars:
	inpvars = "<div style='float: left; width: 90%%;'><blockquote><pre style='width: 600px;'>%s</pre>\n</blockquote></div>" % inpvars


	print("""
	<div id="%s" style="margin: 20px; display: flex; box-sizing: border-box; width: 900px; border-radius: 6px; border: 1px solid %s; font-family: sans-serif; background: %s30;">
	<div style="min-height: 32px;">
	<!-- method -->

	<div style="float: left; align-items: center; margin: 4px; border-radius: 5px; text-align: center; padding-top: 4px; height: 20px; width: 100px; color: #FFF; font-weight: bold; background: %s;">%s</div>

	<!-- path and summary -->
	<span style="display: flex; padding-top: 6px;"><kbd><strong>%s</strong></kbd></span>
	<div style="box-sizing: border-box; flex: 1; font-size: 13px; font-family: Open Sans, sans-serif; float: left; padding-top: 6px; margin-left: 20px;">
	%s</div>
	<div style="float: left; width: 90%%;display: %s; ">
	<h4>JSON parameters:</h4>
	%s
	<br/>
	%s
	</div>
	<div style="float: left; width: 90%%; ">
	<h4>Response examples:</h4>
	<blockquote>%s</blockquote>
	</div>
	</div>
	</div>
	""" % (linkname, mcolors[method], mcolors[method], mcolors[method], method, path, summary, "block" if inp else "none", inpvars, inp, resp))
	#print("%s %s: %s" % (method.upper(), path, mspec['summary']))
	print("</body></html>")